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。最后执行 gathered + chunk 的是客户端代码，它把多个 Chunk 累加成完整的 tool_calls。因此，“服务端解析”和“客户端合并”不能视为同一个步骤。

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

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

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

2. 准备两个 Python 环境

客户端代码继续使用 LangGraph 系列 2 创建的 .venv_langgraph。模型服务复用 LangGraph 系列 3 创建的 .venv_tool_server，并按本篇锁文件刷新间接依赖，避免改变前面已经使用的主环境。

2.1 客户端环境

在 llm_learning 根目录执行：

cd source/_posts/llm_learning

source .venv_langgraph/bin/activate
python -m pip install \
  -r langgraph/p06_streaming_and_online_tools/requirements.txt
python -m pip check

客户端沿用下面这些已经锁定的主要版本：

langchain==1.2.13
langchain-core==1.2.22
langchain-openai==1.1.12
langgraph==1.1.3
openai==2.30.0
httpx==0.28.1

2.2 模型服务环境

如果没有按 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：

mlx-lm==0.31.1
mlx==0.31.0
transformers==5.3.0

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

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

安装并检查环境：

python -m pip install \
  -r langgraph/p06_streaming_and_online_tools/requirements-server-lock.txt
python -m pip check

实际检查结果：

No broken requirements found.

2.3 启动 Qwen3 服务

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

source .venv_tool_server/bin/activate

"$VIRTUAL_ENV/bin/python" -m mlx_lm server \
  --model model \
  --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 返回模型列表后，再运行后面的脚本。这样可以先排除服务未启动、端口错误和模型路径错误。

3. invoke() 与 stream() 的区别

invoke() 等待模型完成整次生成，然后返回一个完整消息：

message = model_with_tools.invoke(messages)
print(type(message).__name__)  # AIMessage

工具调用已经整理在：

message.tool_calls

stream() 返回一个迭代器，每次产生一小块消息：

for chunk in model_with_tools.stream(messages):
    print(type(chunk).__name__)  # AIMessageChunk

工具调用的流式片段位于：

chunk.tool_call_chunks

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

对象	含义
AIMessage	一次非流式模型调用的完整消息
AIMessageChunk	流式模型调用返回的一块消息
tool_calls	已经解析完成的工具调用列表
tool_call_chunks	流式返回的工具名称或参数片段

工具参数可能一次完整返回，也可能被拆成多个字符串片段。应用不能假设每个 Chunk 都包含合法、完整的 JSON。

4. 观察 Qwen3 的 ToolCallChunk

第一个脚本把计算器绑定到本地 Qwen3，打印每个工具调用块，再使用 + 合并全部 AIMessageChunk。

langgraph/p06_streaming_and_online_tools/01_stream_tool_call_chunks.py：

"""观察 Qwen3 流式返回的工具调用块，并把消息块合并成完整工具调用。"""

from typing import Literal

import httpx
from langchain_core.messages import AIMessageChunk, HumanMessage
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)


# trust_env=False 避免系统代理或 VPN 转发本机请求。
with httpx.Client(trust_env=False) as http_client:
    model = ChatOpenAI(
        model="default_model",
        base_url="http://127.0.0.1:18080/v1",
        api_key="not-needed",
        temperature=0,
        max_tokens=128,
        http_client=http_client,
    )
    model_with_tools = model.bind_tools([calculator])

    gathered: AIMessageChunk | None = None
    chunk_number = 0

    # stream() 返回 AIMessageChunk；工具参数可能被拆成一个或多个块。
    for chunk in model_with_tools.stream(
        [HumanMessage(content="请使用计算器工具计算 18 乘以 7。")]
    ):
        gathered = chunk if gathered is None else gathered + chunk

        for tool_chunk in chunk.tool_call_chunks:
            chunk_number += 1
            print(f"工具调用块 {chunk_number}：")
            print("  工具名：", tool_chunk.get("name"))
            print("  参数片段：", tool_chunk.get("args"))

    if gathered is None or not gathered.tool_calls:
        raise RuntimeError("Qwen3 没有生成完整的工具调用")

    # 消息块相加后，LangChain 会把参数片段解析为完整字典。
    complete_call = gathered.tool_calls[0]
    print("合并后的工具名：", complete_call["name"])
    print("合并后的参数：", complete_call["args"])

运行：

source .venv_langgraph/bin/activate
python \
  langgraph/p06_streaming_and_online_tools/01_stream_tool_call_chunks.py

本机实际输出：

工具调用块 1：
  工具名： calculator
  参数片段： {"a": 18, "b": 7, "operation": "multiply"}
合并后的工具名： calculator
合并后的参数： {'a': 18, 'b': 7, 'operation': 'multiply'}

这次 MLX-LM 在一个 Chunk 中返回了完整工具名称和 JSON 参数，但代码仍然执行了消息合并。更换模型、推理服务或网络传输方式后，参数可能被拆成多个 Chunk，不能根据这一次结果删掉合并逻辑。

4.1 为什么使用 chunk 相加

下面这行代码不是字符串拼接：

gathered = chunk if gathered is None else gathered + chunk

AIMessageChunk.add() 会合并文本、工具调用片段和其他消息字段。合并完成后，gathered.tool_calls 才是适合应用继续处理的完整工具调用。

如果直接对某个中间 Chunk 的 args 调用 json.loads()，参数恰好被拆开时就会产生 JSON 解析错误。

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

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

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

第二个脚本使用 stream_mode=”messages” 观察整个过程。

langgraph/p06_streaming_and_online_tools/02_stream_agent_tool_loop.py：

"""使用 LangGraph Agent 流式观察工具请求、工具结果和最终回答。"""

from typing import Literal

import httpx
from langchain.agents import create_agent
from langchain_core.messages import AIMessageChunk, ToolMessage
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)


with httpx.Client(trust_env=False) as http_client:
    model = ChatOpenAI(
        model="default_model",
        base_url="http://127.0.0.1:18080/v1",
        api_key="not-needed",
        temperature=0,
        max_tokens=128,
        http_client=http_client,
    )
    agent = create_agent(
        model=model,
        tools=[calculator],
        system_prompt="需要计算时必须使用计算器工具，然后用中文简洁回答。",
    )

    tool_finished = False
    final_answer = ""

    # messages 模式每次返回“消息块 + 执行元数据”。
    for message, metadata in agent.stream(
        {"messages": [{"role": "user", "content": "请计算 18 乘以 7。"}]},
        stream_mode="messages",
    ):
        if isinstance(message, AIMessageChunk) and message.tool_calls:
            tool_call = message.tool_calls[0]
            print("模型选择的工具：", tool_call["name"])
            print("模型生成的参数：", tool_call["args"])
        elif isinstance(message, ToolMessage):
            tool_finished = True
            print("工具执行结果：", message.content)
        elif tool_finished and isinstance(message, AIMessageChunk) and message.content:
            # 工具执行完成后收到的内容，才是最终自然语言回答。
            text = str(message.content)
            final_answer += text

    print("Qwen3 最终回答：", final_answer)

运行：

python \
  langgraph/p06_streaming_and_online_tools/02_stream_agent_tool_loop.py

本机实际输出：

模型选择的工具： calculator
模型生成的参数： {'a': 18, 'b': 7, 'operation': 'multiply'}
工具执行结果： 126.0
Qwen3 最终回答： 18 乘以 7 的结果是 126。

5.1 messages 模式返回什么

agent.stream(…, stream_mode=”messages”) 每次返回两个值：

message, metadata

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

本篇只根据消息类型识别工具调用、工具结果和最终回答，不打印整段元数据。完整元数据通常包含运行 ID 等动态字段，直接输出既不利于阅读，也不适合作为稳定的文章结果。

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

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

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

工具调用不是最终答案。应用必须等待工具执行，得到 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。示例只查询位置资料，不查询实时天气，输出不会随着天气变化。

langgraph/p06_streaming_and_online_tools/03_city_search_base_tool.py：

"""继承 BaseTool，通过 Open-Meteo 公共接口查询真实城市信息。"""

import httpx
from langchain_core.tools import BaseTool
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

    def _run(self, city: str) -> str:
        """发送 HTTP 请求，并把第一条匹配结果整理为稳定文本。"""

        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}"

        results = response.json().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', '')}"
        )


city_search = CitySearchTool()

print("查询杭州：")
print(city_search.invoke({"city": "杭州"}))

print("\n查询不存在的城市：")
print(city_search.invoke({"city": "不存在的城市xyz12345"}))

运行：

python \
  langgraph/p06_streaming_and_online_tools/03_city_search_base_tool.py

本机实际输出：

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

查询不存在的城市：
没有找到城市：不存在的城市xyz12345

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

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

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

示例使用 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：

"""把真实联网城市查询工具交给本地 Qwen3 Agent 自动选择和执行。"""

import httpx
from langchain.agents import create_agent
from langchain_core.messages import AIMessageChunk, ToolMessage
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

    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}"

        results = response.json().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', '')}"
        )


city_search = CitySearchTool()

with httpx.Client(trust_env=False) as http_client:
    model = ChatOpenAI(
        model="default_model",
        base_url="http://127.0.0.1:18080/v1",
        api_key="not-needed",
        temperature=0,
        max_tokens=256,
        http_client=http_client,
    )
    agent = create_agent(
        model=model,
        tools=[city_search],
        system_prompt="涉及城市信息时必须使用城市查询工具，然后根据工具结果用中文回答。",
    )

    tool_finished = False
    final_answer = ""

    for message, metadata in agent.stream(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "请查询杭州属于哪个省级行政区，并给出经纬度和时区。",
                }
            ]
        },
        stream_mode="messages",
    ):
        if isinstance(message, AIMessageChunk) and message.tool_calls:
            tool_call = message.tool_calls[0]
            print("模型选择的工具：", tool_call["name"])
            print("模型生成的参数：", tool_call["args"])
        elif isinstance(message, ToolMessage):
            tool_finished = True
            print("工具返回的信息：")
            print(message.content)
        elif tool_finished and isinstance(message, AIMessageChunk) and message.content:
            final_answer += str(message.content)

    print("Qwen3 最终回答：", final_answer)

运行：

python \
  langgraph/p06_streaming_and_online_tools/04_agent_with_city_search.py

本机一次实际输出如下：

模型选择的工具： city_search
模型生成的参数： {'city': '杭州'}
工具返回的信息：
城市：杭州
国家：中国
省级行政区：浙江
纬度：30.29365
经度：120.16142
时区：Asia/Shanghai
Qwen3 最终回答： 杭州属于中国的浙江省，其经纬度为纬度30.29365，经度120.16142，时区为Asia/Shanghai。

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

8.1 谁真正访问了互联网

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

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

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

9. 常见问题

9.1 连接本地模型时报 502 或连接失败

先检查模型服务：

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

如果健康接口失败，检查虚拟环境、model 路径和 18080 端口。不要先修改 LangChain Tool。

9.2 模型没有返回 tool_calls

依次检查：

• Tool 名称、描述和参数 Schema 是否清楚。
• 提示词是否真的需要使用工具。
• 模型与 Chat Template 是否支持工具调用。
• 推理服务是否启用了正确的工具解析器。
• 先用 invoke() 验证非流式调用，再检查 stream()。

9.3 参数 JSON 解析失败

不要假设单个 ToolCallChunk 包含完整 JSON。先合并所有 AIMessageChunk，再从合并结果的 tool_calls 读取参数。

9.4 ToolMessage 与工具调用不匹配

每个工具结果都必须保留对应的 tool_call_id。使用 Agent 时会自动处理；手写工具循环时，应该把完整 tool_call 传给 Tool：

tool_message = selected_tool.invoke(tool_call)

9.5 只看到工具调用，看不到最终回答

工具执行后还需要把 ToolMessage 交回模型。Agent 会自动发起第二次模型调用；手动循环则需要自己追加消息并再次调用模型。

9.6 外部接口超时或没有结果

联网 Tool 必须设置超时、捕获 httpx.HTTPError 并处理空列表。不要让一次网络失败直接中断整个 Agent。

9.7 VPN 或代理影响本机请求

本机 OpenAI 兼容接口可以使用：

httpx.Client(trust_env=False)

命令行健康检查可以使用：

curl --noproxy '*'

这两种方式都能避免环境代理转发 127.0.0.1 请求。

10. 小结

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

非流式调用返回完整 AIMessage.tool_calls。流式调用返回 AIMessageChunk，工具名称和参数可能位于多个 ToolCallChunk 中，因此必须正确合并消息块。Agent 的 messages 流还可以同时观察工具请求、ToolMessage 和最终自然语言 Token。

联网 Tool 本质上仍然是受应用控制的 Python 代码。模型只选择工具和生成参数，真实 HTTP 请求由 BaseTool 执行。本篇已经完成从本地计算器到真实城市接口的完整验证，也补齐了 Qwen3 工具解析和流式调用这一部分内容。

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

知识点	关键对象或方法	作用	本篇实测结果
非流式工具调用	invoke()、AIMessage.tool_calls	一次得到完整工具调用	作为流式排错基准
流式工具调用	stream()、AIMessageChunk	分块接收模型输出	本地 Qwen3 正常返回消息块
工具调用片段	ToolCallChunk	保存流式工具名和参数片段	本次在一个 Chunk 中返回完整参数
消息合并	gathered + chunk	合并文本和工具调用片段	合并后得到标准 tool_calls 字典
Agent 消息流	stream_mode=”messages”	观察工具请求、工具结果和最终 Token	依次得到 calculator、126.0 和最终回答
工具解析器	推理服务参数	把模型输出转换成 API 工具结构	MLX-LM 主线运行成功；vLLM/SGLang 作为兼容说明
联网 BaseTool	CitySearchTool._run()	受控调用外部 HTTP 接口	Open-Meteo 返回杭州位置资料
联网 Agent	create_agent()	自动选择、执行工具并组织回答	Qwen3 选择 city_search 并完成中文回答