LangGraph 系列 6：Qwen3 流式工具调用与联网工具实战

上一篇已经介绍了 Tool 的几种定义方式，并使用 bind_tools() 手动完成了一次工具调用循环。非流式调用比较容易理解：模型一次返回完整的 AIMessage，应用从 tool_calls 中读取工具名称和参数，再执行对应函数。

实际聊天界面通常希望模型边生成边显示内容。工具调用进入流式模式后，返回的不再是一个完整的 AIMessage，而是一组 AIMessageChunk。工具名称和 JSON 参数也可能被拆成多个 ToolCallChunk，需要先合并，再读取完整工具调用。

此外，Tool 不一定只处理内存中的固定数据。它也可以访问数据库、文件或互联网接口。本篇先观察本地 Qwen3 的流式工具调用结构，再使用 BaseTool 调用真实的 Open-Meteo 城市查询接口，完成一次联网 Agent 调用。

1. 为什么 Tool 定义正确仍然可能调用失败

使用 @tool、StructuredTool 或 BaseTool 定义工具，只解决了应用程序这一侧的问题。要让模型稳定返回标准工具调用，还需要下面几层共同配合：

1. 应用把消息和 Tool Schema 交给模型服务。
2. Chat Template 把消息、工具说明和生成提示转换成模型能够理解的格式。
3. 模型按照约定生成工具名称和参数。
4. 推理服务的工具调用解析器识别模型输出，并转换成 OpenAI 兼容的 tool_calls。
5. LangChain 把接口响应转换成 AIMessage 或 AIMessageChunk。

下图将这条链路拆成客户端请求、推理服务处理和客户端消息合并三层。这样可以看到，工具调用解析器与 AIMessageChunk 合并发生在不同位置，解决的也不是同一个问题。

推理服务解析器负责把模型生成的 Token 转换成接口中的 delta.tool_calls；LangChain 再把流式响应转换成 AIMessageChunk。LangGraph Server 可以把这些 Chunk 转发给 messages-tuple 订阅者，同时把合并后的完整 AIMessage 写入 Graph 状态。因此，“服务端解析”“流式事件”和“状态中的完整消息”不能视为同一个步骤。

这几层中任意一层不匹配，都可能出现下面的现象：

• 模型把工具调用当作普通文字输出。
• 非流式调用成功，流式调用解析失败。
• 工具名存在，但参数 JSON 不完整。
• LangChain 收到内容，却没有得到 tool_calls。

因此，工具调用失败时不能只检查 Python 函数。还要检查模型、Chat Template、推理服务版本和工具解析器是否互相兼容。

2. 创建并启动 LangGraph 项目

本篇不再把示例写成直接执行的 Python 脚本，而是把四个示例分别导出为 Graph，再由 LangGraph Server 加载。运行时通过 Studio 或 LangGraph API 提交消息，Python 文件本身只负责定义 Agent。

客户端继续使用 LangGraph 系列 2 创建的 .venv_langgraph。模型服务复用 LangGraph 系列 3 创建的 .venv_tool_server，两个环境分别管理，避免模型推理依赖影响 Agent 项目。

2.1 客户端环境

在 llm_learning 根目录执行：

cd /path/to/llm-learning

source .venv_langgraph/bin/activate
uv pip install \
  -r langgraph/p06_streaming_and_online_tools/requirements.txt

客户端依赖由 requirements.txt 和 pyproject.toml 统一声明。

2.2 注册四个 Agent

p06_streaming_and_online_tools 目录现在是一个独立 LangGraph 项目：

p06_streaming_and_online_tools/
├── 01_stream_tool_call_chunks.py
├── 02_stream_agent_tool_loop.py
├── 03_city_search_base_tool.py
├── 04_agent_with_city_search.py
├── .env.example
├── langgraph.json
├── pyproject.toml
└── requirements.txt

四个 Python 文件都导出名为 graph 的编译图。langgraph.json 使用不同的 Agent ID 注册这些入口：

{
  "$schema": "https://langgra.ph/schema.json",
  "dependencies": ["."],
  "graphs": {
    "stream_tool_call_agent": "./01_stream_tool_call_chunks.py:graph",
    "calculator_agent": "./02_stream_agent_tool_loop.py:graph",
    "city_search_tool_agent": "./03_city_search_base_tool.py:graph",
    "city_search_agent": "./04_agent_with_city_search.py:graph"
  },
  "env": ".env",
  "python_version": "3.12"
}

四个 Agent 的职责如下：

Agent ID	用途	是否依赖本地 Qwen3
stream_tool_call_agent	只生成工具调用，观察流式参数片段	是
calculator_agent	自动完成计算器工具循环	是
city_search_tool_agent	直接执行 BaseTool，隔离测试 Open-Meteo	否
city_search_agent	让 Qwen3 选择并执行城市查询工具	是

模型地址通过环境变量提供。第一次运行时复制示例文件：

cd langgraph/p06_streaming_and_online_tools
cp .env.example .env

.env.example 的内容如下：

QWEN_MODEL=Qwen3-14B-AWQ-4bit-MLX
QWEN_BASE_URL=http://127.0.0.1:18080/v1
QWEN_API_KEY=not-needed

2.3 模型服务环境

如果没有按 LangGraph 系列 3 创建模型服务环境，先创建 Python 3.12 虚拟环境；已经存在则直接激活：

python3.12 -m venv \
  --prompt llm_learning_tool_server \
  .venv_tool_server

source .venv_tool_server/bin/activate
python -m pip install "pip==26.0.1"

模型服务的直接依赖写在 requirements-server.in。

项目已经生成包含传递依赖的锁文件。安装时直接使用锁文件，避免重新解析出不同的传递依赖：

uv pip sync \
  langgraph/p06_streaming_and_online_tools/requirements-server-lock.txt

2.4 启动 Qwen3 服务

确保当前目录是 llm_learning，并且根目录下的 model 指向已经下载的 Qwen3 模型：

source .venv_tool_server/bin/activate

"$VIRTUAL_ENV/bin/python" -m mlx_lm server \
  --model Qwen3-14B-AWQ-4bit-MLX \
  --host 127.0.0.1 \
  --port 18080 \
  --prompt-cache-size 0 \
  --chat-template-args '{"enable_thinking": false}'

这个终端需要保持运行。另开一个终端执行健康检查：

curl --noproxy '*' --fail --silent \
  http://127.0.0.1:18080/v1/models

/v1/models 返回模型列表后，再启动 LangGraph Server。这样可以先排除服务未启动、端口错误和模型路径错误。

2.5 启动 LangGraph Server

另开终端，进入本篇项目目录启动开发服务器：

cd /path/to/llm-learning/langgraph/p06_streaming_and_online_tools
source ../../.venv_langgraph/bin/activate
langgraph dev

LangGraph Server 默认监听 127.0.0.1:2024。启动日志中应出现四条 importing graph 记录，随后可以在 Studio 中选择 Agent，也可以向 /runs/stream 提交无状态运行。

3. values 与 messages-tuple 流模式

直接调用本地 Graph 时，invoke() 会等待整张图执行完成，返回包含完整消息的最终状态；通过 LangGraph API 提交时，stream_mode 决定客户端观察哪一层输出。

• values 返回每一步完成后的完整 Graph 状态，最终 AIMessage 中的 tool_calls 已经是完整工具调用。
• messages-tuple 返回模型和工具节点产生的消息流，每个事件包含消息块与执行元数据，适合观察 AIMessageChunk、ToolCallChunk 和 ToolMessage。

通过 API 请求时可以同时指定两种模式：

{
  "stream_mode": ["messages-tuple", "values"]
}

这里有四个容易混淆的对象：

对象	含义
AIMessage	Graph 状态中保存的完整模型消息
AIMessageChunk	messages-tuple 流返回的一块模型消息
tool_calls	完整 AIMessage 中已经解析完成的工具调用列表
tool_call_chunks	AIMessageChunk 中的工具名称或参数片段

工具参数可能一次完整返回，也可能被拆成多个字符串片段。流式客户端不能假设每个 Chunk 都包含合法、完整的 JSON；需要完整参数时，应读取最终状态中的 tool_calls，或在客户端合并所有 AIMessageChunk。

4. 观察 Qwen3 的 ToolCallChunk

第一个 Agent 只调用一次绑定了计算器的模型，不执行工具。这样既能从 messages-tuple 事件观察 tool_call_chunks，也能从 values 事件读取完整的 tool_calls。

langgraph/p06_streaming_and_online_tools/01_stream_tool_call_chunks.py：

"""只生成工具调用请求的 LangGraph Agent，用于观察流式 tool_call_chunks。"""

import os
from typing import Literal

import httpx
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, MessagesState, StateGraph
from pydantic import BaseModel, Field


class CalculatorInput(BaseModel):
    """计算器工具的输入参数。"""

    a: float = Field(description="第一个数字")
    b: float = Field(description="第二个数字")
    operation: Literal["multiply"] = Field(description="运算类型，只能是 multiply")

@tool(args_schema=CalculatorInput)
def calculator(a: float, b: float, operation: str) -> str:
    """计算两个数字的乘积。"""

    return str(a * b)

# LangGraph Server 会长期持有 Graph，因此模型客户端也在模块加载时创建。
model = ChatOpenAI(
    model=os.getenv("QWEN_MODEL", "Qwen3-14B-AWQ-4bit-MLX"),
    base_url=os.getenv("QWEN_BASE_URL", "http://127.0.0.1:18080/v1"),
    api_key=os.getenv("QWEN_API_KEY", "not-needed"),
    temperature=0,
    max_tokens=128,
    # 同步和异步客户端都禁用系统代理，确保本机模型请求直连。
    http_client=httpx.Client(trust_env=False),
    http_async_client=httpx.AsyncClient(trust_env=False),
)
model_with_tools = model.bind_tools([calculator])


async def call_model(state: MessagesState) -> dict[str, list]:
    """调用一次模型，让模型生成工具名称和参数，但不执行工具。"""

    response = await model_with_tools.ainvoke(state["messages"])
    return {"messages": [response]}

# 这个 Graph 故意只包含模型节点。
# 通过 LangGraph API 使用 messages-tuple 流模式提交时，可以观察工具参数分块；
# 最终状态中的 AIMessage.tool_calls 则保存合并后的完整工具调用。
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", END)
graph = builder.compile()

代码不再写死用户问题，也不在模块导入时开始执行。LangGraph Server 加载 graph 后，使用下面的请求提交消息：

curl --no-buffer --request POST \
  --url http://127.0.0.1:2024/runs/stream \
  --header 'Content-Type: application/json' \
  --data '{
    "assistant_id": "stream_tool_call_agent",
    "input": {
      "messages": [
        {
          "role": "user",
          "content": "请使用计算器工具计算 18 乘以 7。"
        }
      ]
    },
    "stream_mode": ["messages-tuple", "values"]
  }'

本机实际流中的关键字段如下：

messages 事件：
tool_call_chunks = [
  {
    "name": "calculator",
    "args": "{\"a\": 18, \"b\": 7, \"operation\": \"multiply\"}",
    "index": 0,
    "type": "tool_call_chunk"
  }
]

最终 values 事件：
tool_calls = [
  {
    "name": "calculator",
    "args": {"a": 18, "b": 7, "operation": "multiply"},
    "type": "tool_call"
  }
]

这次 MLX-LM 在一个 Chunk 中返回了完整工具名称和 JSON 参数。更换模型、推理服务或网络传输方式后，参数仍可能被拆成多个 Chunk，客户端不能把这一次输出当作固定规律。

4.1 Graph 状态与流式事件的区别

call_model() 使用 ainvoke() 返回完整 AIMessage，因此写入 MessagesState 的是已经合并好的 tool_calls。与此同时，LangGraph Server 可以通过模型回调把生成过程中的 AIMessageChunk 发送给 messages-tuple 订阅者。

因此，观察生成过程时读取 messages-tuple 中的 tool_call_chunks；真正执行工具或保存运行状态时读取完整 AIMessage.tool_calls。不要对中间 args 片段立即调用 json.loads()。

5. 使用 Agent 自动完成流式工具循环

上一个 Graph 只观察模型生成的工具调用，没有执行工具。create_agent() 会自动完成下面的循环：

1. 调用模型并获得工具请求。
2. 根据工具名执行对应 Tool。
3. 生成与调用 ID 对应的 ToolMessage。
4. 再次调用模型生成最终回答。

第二个文件只定义 Agent，不再在文件内部调用 stream()。提交和流式订阅由 LangGraph Server 负责。

langgraph/p06_streaming_and_online_tools/02_stream_agent_tool_loop.py：

"""可由 LangGraph Server 提交运行的计算器 Agent。"""

import os
from typing import Literal

import httpx
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field


class CalculatorInput(BaseModel):
    """计算器工具的输入参数。"""

    a: float = Field(description="第一个数字")
    b: float = Field(description="第二个数字")
    operation: Literal["multiply"] = Field(description="运算类型，只能是 multiply")


@tool(args_schema=CalculatorInput)
def calculator(a: float, b: float, operation: str) -> str:
    """计算两个数字的乘积。"""

    return str(a * b)


model = ChatOpenAI(
    model=os.getenv("QWEN_MODEL", "Qwen3-14B-AWQ-4bit-MLX"),
    base_url=os.getenv("QWEN_BASE_URL", "http://127.0.0.1:18080/v1"),
    api_key=os.getenv("QWEN_API_KEY", "not-needed"),
    temperature=0,
    max_tokens=128,
    http_client=httpx.Client(trust_env=False),
    http_async_client=httpx.AsyncClient(trust_env=False),
)


# create_agent() 返回编译后的 LangGraph，可直接注册到 langgraph.json。
# 提交运行后，Graph 会自动完成“模型请求工具 → 执行工具 → 模型回答”的循环。
graph = create_agent(
    model=model,
    tools=[calculator],
    system_prompt="需要计算时必须使用计算器工具，然后用中文简洁回答。",
)

向 calculator_agent 提交同样的问题：

curl --no-buffer --request POST \
  --url http://127.0.0.1:2024/runs/stream \
  --header 'Content-Type: application/json' \
  --data '{
    "assistant_id": "calculator_agent",
    "input": {
      "messages": [
        {
          "role": "user",
          "content": "请计算 18 乘以 7。"
        }
      ]
    },
    "stream_mode": "messages-tuple"
  }'

本机实际流中的关键消息依次为：

AIMessageChunk.tool_call_chunks：
calculator({"a": 18, "b": 7, "operation": "multiply"})

ToolMessage.content：
126.0

最终 AIMessageChunk.content：
18 乘以 7 的结果是 126。

5.1 messages-tuple 模式返回什么

通过 Python SDK 订阅时，每个 messages 事件包含两个值：

message, metadata = chunk.data

• message 是序列化后的 AIMessageChunk、ToolMessage 等消息。
• metadata 描述这条消息来自 Graph 的哪个执行步骤。

curl 会把这两个值作为 SSE data 输出。完整元数据通常包含运行 ID、Graph 节点名等动态字段，文章只保留与工具调用相关的关键内容。

5.2 工具流和文本流不是同一件事

流式 Agent 中至少存在两类输出：

1. 工具调用结构流：模型逐步生成工具名称和参数。
2. 自然语言 Token 流：工具执行完成后，模型逐步生成最终回答。

工具调用不是最终答案。create_agent() 构建的 Graph 会等待工具执行，得到 ToolMessage，再让模型根据结果回答用户；远程客户端只负责提交和消费运行流。

6. 推理服务为什么需要工具调用解析器

模型最终生成的仍然是 Token。推理服务需要识别其中的工具名称、参数和调用边界，再把它们转换成 OpenAI 兼容接口的结构。

以流式响应为例，接口通常在 delta 中逐步返回内容。LangChain 收到后再转换成 ToolCallChunk。因此，解析器位于推理服务层，不属于 Tool，也不是 LangChain 中的 BaseTool。

6.1 vLLM 中的两个参数

vLLM 的自动工具调用通常涉及：

--enable-auto-tool-choice
--tool-call-parser <解析器名称>

• –enable-auto-tool-choice 允许模型自行判断是否调用工具。
• –tool-call-parser 指定如何从模型输出中提取工具调用。

Hermes 是一种工具调用格式解析器，不是 Tool，也不会替应用执行函数。模型、Chat Template 和解析器格式不一致时，可能出现非流式可用、流式失败或工具调用变成普通文本的情况。

6.2 SGLang 中的配置思路

处理 Qwen3 兼容问题时，可以使用类似下面的参数组合：

--tool-call-parser qwen25 \
--reasoning-parser qwen3

两个解析器负责不同内容：

• qwen25 工具解析器识别工具调用结构。
• qwen3 推理解析器识别 Qwen3 的推理内容。

这些参数属于 SGLang GPU 服务的兼容配置，不是本文 M1 Max 的启动命令。本篇真实运行使用的是 MLX-LM，不能把上面的命令写成 Apple Silicon 实测结果。

也不能把特定服务版本中的解析问题概括成“Qwen3 不支持工具调用”。本篇已经使用 Qwen3、MLX-LM 和 LangChain 真实获得 ToolCallChunk、ToolMessage 和最终回答。

7. 使用 BaseTool 调用真实城市接口

前面的计算器只处理本地数字。下面定义一个 CitySearchTool，调用 Open-Meteo Geocoding API 查询城市所属国家、省级行政区、经纬度和时区。

这个接口不需要 API Key。示例只查询位置资料，不查询实时天气，输出不会随着天气变化。为了让 BaseTool 测试也通过 LangGraph 提交，本节再用单节点 StateGraph 包装工具；它不依赖 Qwen3。

langgraph/p06_streaming_and_online_tools/03_city_search_base_tool.py：

"""直接执行 BaseTool 的 LangGraph Agent，不依赖本地大模型。"""

import httpx
from langchain_core.messages import AIMessage
from langchain_core.tools import BaseTool
from langgraph.graph import END, START, MessagesState, StateGraph
from pydantic import BaseModel, Field

class CitySearchInput(BaseModel):
    """城市查询工具的输入参数。"""

    city: str = Field(description="需要查询的城市中文名或英文名")

class CitySearchTool(BaseTool):
    """调用 Open-Meteo Geocoding API 的城市查询工具。"""

    name: str = "city_search"
    description: str = "查询城市所属国家、省级行政区、经纬度和时区"
    args_schema: type[BaseModel] = CitySearchInput

    @staticmethod
    def _format_result(city: str, payload: dict) -> str:
        """把接口响应整理为稳定文本。"""

        results = payload.get("results", [])
        if not results:
            return f"没有找到城市：{city}"

        city_info = results[0]
        return (
            f"城市：{city_info.get('name', '')}\n"
            f"国家：{city_info.get('country', '')}\n"
            f"省级行政区：{city_info.get('admin1', '')}\n"
            f"纬度：{city_info.get('latitude', '')}\n"
            f"经度：{city_info.get('longitude', '')}\n"
            f"时区：{city_info.get('timezone', '')}"
        )

    def _run(self, city: str) -> str:
        """同步查询城市信息。"""
        try:
            with httpx.Client(trust_env=False, timeout=15.0) as client:
                response = client.get(
                    "https://geocoding-api.open-meteo.com/v1/search",
                    params={
                        "name": city,
                        "count": 1,
                        "language": "zh",
                        "format": "json",
                    },
                )
                response.raise_for_status()
        except httpx.HTTPError as error:
            return f"城市信息查询失败：{error}"

        return self._format_result(city, response.json())

    async def _arun(self, city: str) -> str:
        """异步查询城市信息，供 LangGraph Server 调用。"""

        try:
            async with httpx.AsyncClient(trust_env=False, timeout=15.0) as client:
                response = await client.get(
                    "https://geocoding-api.open-meteo.com/v1/search",
                    params={
                        "name": city,
                        "count": 1,
                        "language": "zh",
                        "format": "json",
                    },
                )
                response.raise_for_status()
        except httpx.HTTPError as error:
            return f"城市信息查询失败：{error}"

        return self._format_result(city, response.json())


city_search = CitySearchTool()


async def call_city_search(state: MessagesState) -> dict[str, list[AIMessage]]:
    """把最后一条用户消息作为城市名，直接执行 BaseTool。"""

    if not state["messages"]:
        raise ValueError("必须提交一条包含城市名的用户消息")

    content = state["messages"][-1].content
    if not isinstance(content, str) or not content.strip():
        raise ValueError("最后一条消息必须是非空城市名称")

    result = await city_search.ainvoke({"city": content.strip()})
    return {"messages": [AIMessage(content=str(result))]}


# 这个 Agent 用于隔离测试 BaseTool：输入消息只写城市名，例如“杭州”。
builder = StateGraph(MessagesState)
builder.add_node("city_search", call_city_search)
builder.add_edge(START, "city_search")
builder.add_edge("city_search", END)
graph = builder.compile()

提交时，最后一条用户消息只填写城市名：

curl --no-buffer --request POST \
  --url http://127.0.0.1:2024/runs/stream \
  --header 'Content-Type: application/json' \
  --data '{
    "assistant_id": "city_search_tool_agent",
    "input": {
      "messages": [
        {
          "role": "user",
          "content": "杭州"
        }
      ]
    },
    "stream_mode": "values"
  }'

最终 values 事件中的 AIMessage 内容如下：

城市：深圳
国家：中国
省级行政区：广东
纬度：22.54554
经度：114.0683
时区：Asia/Shanghai

7.1 为什么工具内部仍然要处理异常

Graph 也不会替应用处理网络超时、HTTP 错误和空结果。联网 Tool 至少需要考虑：

• 设置合理的请求超时。
• 对非 2xx 响应调用 raise_for_status()。
• 捕获 HTTP 异常。
• 检查接口是否返回结果。
• 把错误整理成 Agent 可以继续处理的文本。

LangGraph Server 通过异步接口运行，所以 CitySearchTool 同时实现 _run() 和 _arun()。异步路径使用 httpx.AsyncClient，不会用同步网络请求阻塞服务器事件循环。

示例使用 trust_env=False，避免系统代理或 VPN 影响 HTTP 客户端。本机 Qwen3 请求使用它是为了确保 127.0.0.1 不经过代理；Open-Meteo 请求使用它则表示本示例明确不读取环境代理配置。

8. 将联网工具交给 Qwen3 Agent

最后把 CitySearchTool 交给 Agent。模型负责理解问题和生成 city=”杭州”，Agent 负责执行工具，工具负责访问 Open-Meteo，模型再把结果整理成中文回答。

下图按照真实调用顺序展示两次模型请求。Open-Meteo 的 JSON 响应会先返回 CitySearchTool，Tool 整理成字符串后交给 Agent，再由 Agent 创建与工具调用 ID 匹配的 ToolMessage。

Open-Meteo 不认识 LangChain 消息，也不会直接生成 ToolMessage。这个消息对象属于 Agent 工具循环，用来把本地 Tool 的执行结果交回模型。

langgraph/p06_streaming_and_online_tools/04_agent_with_city_search.py：

"""可由 LangGraph Server 提交运行的联网城市查询 Agent。"""

import os

import httpx
from langchain.agents import create_agent
from langchain_core.tools import BaseTool
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field


class CitySearchInput(BaseModel):
    """城市查询工具的输入参数。"""

    city: str = Field(description="需要查询的城市中文名或英文名")


class CitySearchTool(BaseTool):
    """调用 Open-Meteo Geocoding API 的城市查询工具。"""

    name: str = "city_search"
    description: str = "查询城市所属国家、省级行政区、经纬度和时区"
    args_schema: type[BaseModel] = CitySearchInput

    @staticmethod
    def _format_result(city: str, payload: dict) -> str:
        """把接口响应整理为适合模型继续处理的文本。"""

        results = payload.get("results", [])
        if not results:
            return f"没有找到城市：{city}"

        city_info = results[0]
        return (
            f"城市：{city_info.get('name', '')}\n"
            f"国家：{city_info.get('country', '')}\n"
            f"省级行政区：{city_info.get('admin1', '')}\n"
            f"纬度：{city_info.get('latitude', '')}\n"
            f"经度：{city_info.get('longitude', '')}\n"
            f"时区：{city_info.get('timezone', '')}"
        )

    def _run(self, city: str) -> str:
        """同步查询城市信息。"""

        try:
            with httpx.Client(trust_env=False, timeout=15.0) as client:
                response = client.get(
                    "https://geocoding-api.open-meteo.com/v1/search",
                    params={
                        "name": city,
                        "count": 1,
                        "language": "zh",
                        "format": "json",
                    },
                )
                response.raise_for_status()
        except httpx.HTTPError as error:
            return f"城市信息查询失败：{error}"

        return self._format_result(city, response.json())

    async def _arun(self, city: str) -> str:
        """异步查询城市信息，避免阻塞 Agent 运行循环。"""

        try:
            async with httpx.AsyncClient(trust_env=False, timeout=15.0) as client:
                response = await client.get(
                    "https://geocoding-api.open-meteo.com/v1/search",
                    params={
                        "name": city,
                        "count": 1,
                        "language": "zh",
                        "format": "json",
                    },
                )
                response.raise_for_status()
        except httpx.HTTPError as error:
            return f"城市信息查询失败：{error}"

        return self._format_result(city, response.json())


city_search = CitySearchTool()
model = ChatOpenAI(
    model=os.getenv("QWEN_MODEL", "Qwen3-14B-AWQ-4bit-MLX"),
    base_url=os.getenv("QWEN_BASE_URL", "http://127.0.0.1:18080/v1"),
    api_key=os.getenv("QWEN_API_KEY", "not-needed"),
    temperature=0,
    max_tokens=256,
    http_client=httpx.Client(trust_env=False),
    http_async_client=httpx.AsyncClient(trust_env=False),
)


graph = create_agent(
    model=model,
    tools=[city_search],
    system_prompt="涉及城市信息时必须使用城市查询工具，然后根据工具结果用中文回答。",
)

向 city_search_agent 提交自然语言问题：

curl --no-buffer --request POST \
  --url http://127.0.0.1:2024/runs/stream \
  --header 'Content-Type: application/json' \
  --data '{
    "assistant_id": "city_search_agent",
    "input": {
      "messages": [
        {
          "role": "user",
          "content": "请查询杭州所属的省级行政区、经纬度和时区。"
        }
      ]
    },
    "stream_mode": "messages-tuple"
  }'

本机实际流中的关键消息如下：

AIMessageChunk.tool_call_chunks：
city_search({"city": "杭州"})

ToolMessage.content：
城市：杭州
国家：中国
省级行政区：浙江
纬度：30.29365
经度：120.16142
时区：Asia/Shanghai

最终 AIMessageChunk.content：
杭州所属的省级行政区是浙江，经纬度为纬度30.29365，
经度120.16142，时区为Asia/Shanghai。

模型回答的具体措辞可能变化，但工具名称、参数结构、城市接口返回字段和消息执行顺序应保持一致。

8.1 谁真正访问了互联网

这条调用链中各组件的职责如下：

1. Qwen3 选择 city_search 并生成城市参数。
2. LangGraph Agent 根据工具名执行 CitySearchTool。
3. LangGraph Server 通过 CitySearchTool._arun() 和 httpx.AsyncClient 请求 Open-Meteo。
4. Agent 把返回值包装成 ToolMessage。
5. Qwen3 根据 ToolMessage 生成最终回答。

模型没有直接获得任意互联网访问权限。网络地址、超时、错误处理和返回字段都由 Tool 代码控制。

9. 小结

Tool 定义解决的是“应用有哪些能力、参数是什么”；Chat Template 和推理服务解析器解决的是“模型如何表达工具调用、接口如何识别它”；LangChain 与 LangGraph 负责把结构转换成消息，并组织工具执行循环。

本篇把四个示例都注册成了 LangGraph Agent。Python 文件只定义 graph，LangGraph Server 负责加载项目，Studio 或 /runs/stream 负责提交输入。values 适合读取完整 Graph 状态，messages-tuple 可以同时观察工具请求、ToolMessage 和最终自然语言 Token。

流式工具名称和参数可能位于多个 ToolCallChunk 中，不能直接把任意中间片段当作完整 JSON。需要执行工具时读取状态中的 AIMessage.tool_calls；需要展示生成过程时订阅 messages-tuple。

联网 Tool 本质上仍然是受应用控制的 Python 代码。模型只选择工具和生成参数，真实 HTTP 请求由 BaseTool 执行。CitySearchTool 同时提供同步和异步实现，使直接调用与 LangGraph Server 运行都能复用同一套工具逻辑。

到这里，从 Python 函数、Runnable 和 BaseTool 创建本地工具，到非流式调用、流式工具调用、推理服务解析器和联网工具，这组内容已经完整闭环。下一篇开始介绍智能体运行时上下文，区分 configurable 与项目使用的 Runtime Context。