LangGraph 系列 5：本地 Tool 的定义与调用

langgraph

发布日期: 2026-04-08

文章字数: 6.9k

阅读时长: 28 分

阅读次数:

前面创建天气 Agent 时，模型能够判断什么时候查询天气，并把城市名称交给 get_weather()。这个过程看起来像是模型执行了 Python 函数，实际并不是这样。

大模型只负责理解问题、选择工具并生成参数。真正查找工具、调用 Python 函数、处理返回值的仍然是应用程序。LangChain 的 Tool 负责在两者之间建立一份清晰的接口约定。

本篇不使用 Agent 自动循环，而是先简要介绍实现本地 Tool 的三种方式，再分别通过完整代码重点讲解每一种方式，最后使用本地 Qwen3 手动完成一次完整的工具调用。

1. 为什么模型需要工具

大模型擅长理解和生成文本，但只依靠模型自身，仍然无法稳定完成下面这些工作：

查询实时天气、库存或订单状态。
执行精确计算。
读取数据库或本地文件。
调用企业内部接口。
对外部系统执行写入操作。

Tool 可以把这些能力包装成模型能够理解的结构。以计算器为例，应用需要告诉模型：

工具名称：calculator
用途：计算两个数字的乘积
参数：a、b、operation

模型看到的不是 Python 函数源码，而是由名称、描述和参数组成的 JSON Schema。模型可以据此生成工具调用请求，应用程序再执行对应函数。

2. Tool 和普通函数有什么区别

普通 Python 函数已经可以执行计算：

def plain_add(a: int, b: int) -> int:
    return a + b

但是模型并不知道这个函数叫什么、有什么用途、需要哪些参数。LangChain Tool 在函数外面增加了模型调用所需的元数据，并提供统一的 invoke()、ainvoke() 接口。

一个 Tool 最重要的四个属性如下：

属性	作用
name	工具的唯一名称，模型生成工具调用时会使用它
description	说明工具能做什么，帮助模型判断是否应该调用
args_schema	定义参数名称、类型、必填项、说明和校验规则
return_direct	在支持该语义的 Agent 执行循环中，工具执行后是否直接结束循环并返回结果

return_direct 不会改变普通 tool.invoke() 的返回方式。它主要供 Agent 执行器判断工具运行后是否还要继续调用模型。

3. 实现本地 Tool 的三种方式

LangChain 中的本地 Tool 可以通过下面三种方式实现：

包装 Python 函数或协程：使用 @tool 或 StructuredTool.from_function() 把现有函数包装成 Tool。这是最常用的方式，适合无状态或逻辑清晰的功能。
把 Runnable 转换成 Tool：已有 Runnable 调用链时，使用 as_tool() 直接复用现有逻辑。
继承 BaseTool 自定义实现：自己声明工具元数据、参数模型和执行方法，适合封装客户端、连接、配置、状态或复杂生命周期。

下图从左到右分成三层：原始能力、实现方式和统一的 Tool 对象。无论使用哪种方式，最终的 Tool 都需要提供名称、用途、参数 Schema 和可执行实现。

实现本地 Tool 的三种方式

Pydantic、Annotated 和 Docstring 用来补充 args_schema，JSON Schema 是发送给模型的工具说明，它们都不是新的 Tool 实现方式。真正可执行的函数或协程仍然保留在应用程序中。

MCP 是后续独立的工具接入方式，不属于本篇讨论的三种本地 Tool 实现方式。下面开始分别介绍具体实现。

4. 包装 Python 函数或协程实现 Tool

Python 函数或协程是实现本地 Tool 最直接、最常用的方式。简单场景可以使用 @tool 装饰器自动生成名称、说明和参数 Schema；需要分别指定同步与异步实现时，可以使用 StructuredTool.from_function()。

4.1 使用 @tool 包装 Python 函数

@tool 会读取函数名、类型注解和说明，创建一个可以统一调用的 LangChain Tool。

langgraph/p05_local_tools/01_tool_basics.py 的完整代码如下：

"""对比普通 Python 函数与 LangChain Tool，并查看工具的核心属性。"""

import json

from langchain_core.tools import tool
from langchain_core.utils.function_calling import convert_to_openai_tool


def plain_add(a: int, b: int) -> int:
    """计算两个整数之和。"""

    return a + b


@tool("add_numbers", description="计算两个整数之和。")
def add_numbers(a: int, b: int) -> int:
    """计算两个整数之和。"""

    return a + b


print("普通函数类型：", type(plain_add).__name__)
print("普通函数具有 invoke：", hasattr(plain_add, "invoke"))
print("工具类型：", type(add_numbers).__name__)
print("工具名称：", add_numbers.name)
print("工具描述：", add_numbers.description)
print("return_direct：", add_numbers.return_direct)

# args_schema 是由函数参数类型自动生成的 Pydantic 模型。
print("参数 Schema：")
print(json.dumps(add_numbers.args_schema.model_json_schema(), ensure_ascii=False, indent=2))

# convert_to_openai_tool() 可以查看发送给 OpenAI 兼容模型的工具结构。
print("OpenAI 工具 Schema：")
print(json.dumps(convert_to_openai_tool(add_numbers), ensure_ascii=False, indent=2))

# 传入普通参数字典时，invoke() 返回函数本身的执行结果。
result = add_numbers.invoke({"a": 3, "b": 5})
print("字典调用返回类型：", type(result).__name__)
print("字典调用结果：", result)

# 传入完整 tool_call 时，invoke() 会返回包含调用 ID 的 ToolMessage。
tool_call = {
    "name": "add_numbers",
    "args": {"a": 3, "b": 5},
    "id": "call_demo",
    "type": "tool_call",
}
tool_message = add_numbers.invoke(tool_call)
print("tool_call 调用返回类型：", type(tool_message).__name__)
print("ToolMessage 内容：", tool_message.content)
print("ToolMessage 名称：", tool_message.name)
print("ToolMessage 调用 ID：", tool_message.tool_call_id)

运行：

cd /path/to/llm-learning
source .venv_langgraph/bin/activate
python langgraph/p05_local_tools/01_tool_basics.py

输出：

普通函数类型： function
普通函数具有 invoke： False
工具类型： StructuredTool
工具名称： add_numbers
工具描述： 计算两个整数之和。
return_direct： False
字典调用返回类型： int
字典调用结果： 8
tool_call 调用返回类型： ToolMessage
ToolMessage 内容： 8
ToolMessage 名称： add_numbers
ToolMessage 调用 ID： call_demo

普通函数没有 invoke()。经过 @tool 包装后，得到的是 StructuredTool 对象，并且自动拥有参数 Schema。

4.2 模型实际看到的 JSON Schema

convert_to_openai_tool() 可以把 Tool 转换成 OpenAI 兼容模型能够接收的格式。上面的代码实际生成：

{
  "type": "function",
  "function": {
    "name": "add_numbers",
    "description": "计算两个整数之和。",
    "parameters": {
      "properties": {
        "a": {
          "type": "integer"
        },
        "b": {
          "type": "integer"
        }
      },
      "required": [
        "a",
        "b"
      ],
      "type": "object"
    }
  }
}

这份结构告诉模型：工具名是 add_numbers，需要两个整数参数，并且两个参数都是必填项。它没有把函数源码或执行权限交给模型。

4.3 两种 invoke() 输入的区别

传入普通参数字典时：

add_numbers.invoke({"a": 3, "b": 5})

返回函数的原始结果 8。

传入完整的 tool_call 时：

add_numbers.invoke(
    {
        "name": "add_numbers",
        "args": {"a": 3, "b": 5},
        "id": "call_demo",
        "type": "tool_call",
    }
)

返回 ToolMessage。它除了保存结果，还保留 tool_call_id，使模型能够知道这个结果对应哪一次工具请求。后面的 Qwen3 示例会使用这种调用方式。

4.4 使用 Pydantic 定义参数 Schema

简单函数可以依靠类型注解自动生成 Schema。当参数较多、需要枚举值或希望提供更清楚的字段说明时，可以显式定义 Pydantic 模型。

langgraph/p05_local_tools/02_pydantic_args_schema.py：

"""使用 Pydantic 描述工具参数，并验证必填字段和枚举值。"""

import json
from typing import Literal

from langchain_core.tools import tool
from pydantic import BaseModel, Field, ValidationError

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

    a: float = Field(description="第一个数字")
    b: float = Field(description="第二个数字")
    operation: Literal["add", "subtract", "multiply", "divide"] = Field(
        description="运算类型：加、减、乘、除"
    )

@tool(args_schema=CalculatorInput)
def calculator(a: float, b: float, operation: str) -> str:
    """根据指定运算类型计算两个数字。"""

    # 逐个判断四种允许的运算。Pydantic 会提前拒绝其他字符串。
    if operation == "add":
        result = a + b
    elif operation == "subtract":
        result = a - b
    elif operation == "multiply":
        result = a * b
    else:  # operation 只能是 divide。
        if b == 0:
            return "除数不能为 0"
        result = a / b

    return str(result)


# 第 1 步：查看 Pydantic 为模型生成的 JSON Schema。
print("参数 Schema：")
print(json.dumps(calculator.args_schema.model_json_schema(), ensure_ascii=False, indent=2))
print("正确参数结果：", calculator.invoke({"a": 12, "b": 4, "operation": "divide"}))

# 第 2 步：传入不在 Literal 范围内的 operation，观察枚举校验错误。
try:
    calculator.invoke({"a": 12, "b": 4, "operation": "power"})
except ValidationError as error:
    first_error = error.errors()[0]
    print("枚举校验错误位置：", first_error["loc"])
    print("枚举校验错误类型：", first_error["type"])

# 第 3 步：故意漏掉必填参数 b，观察缺少字段的错误。
try:
    calculator.invoke({"a": 12, "operation": "add"})
except ValidationError as error:
    first_error = error.errors()[0]
    print("缺少参数错误位置：", first_error["loc"])
    print("缺少参数错误类型：", first_error["type"])

输出：

正确参数结果： 3.0

枚举校验错误位置： ('operation',)
枚举校验错误类型： literal_error

缺少参数错误位置： ('b',)
缺少参数错误类型： missing

Literal 把 operation 限制为四个固定值。模型能够在 Schema 中看到这些候选值，应用执行工具前也会进行真实校验。Schema 不能保证模型永远生成正确参数，但可以阻止错误参数直接进入函数。

4.5 使用 Annotated 和 Docstring 补充参数说明

除了 Pydantic，还可以使用 Annotated 或 Google 风格 Docstring 为参数增加说明。

langgraph/p05_local_tools/03_annotated_and_docstring.py：

"""演示 Annotated 参数说明和 Google 风格 Docstring 解析。"""

import json
from typing import Annotated

from langchain_core.tools import tool


@tool
def add_integers(
    a: Annotated[int, "第一个整数"],
    b: Annotated[int, "第二个整数"],
) -> str:
    """计算两个整数之和。"""

    return str(a + b)


@tool(parse_docstring=True)
def greet(name: str, title: str = "朋友") -> str:
    """生成一条简单问候语。

    Args:
        name: 被问候人的姓名。
        title: 对被问候人的称呼。
    """

    return f"{title}{name}，你好！"


# 第 1 步：Annotated 中的说明会进入工具参数 Schema。
print("Annotated 参数 Schema：")
print(json.dumps(add_integers.args_schema.model_json_schema(), ensure_ascii=False, indent=2))
print("Annotated 工具结果：", add_integers.invoke({"a": 4, "b": 6}))

# 第 2 步：parse_docstring=True 会读取 Args 部分的参数说明。
print("Docstring 参数 Schema：")
print(json.dumps(greet.args_schema.model_json_schema(), ensure_ascii=False, indent=2))
print("Docstring 工具结果：", greet.invoke({"name": "小明"}))

# 第 3 步：演示 Docstring 参数名写错时的校验错误。
try:
    @tool(parse_docstring=True)
    def invalid_greet(name: str) -> str:
        """生成问候语。

        Args:
            username: 被问候人的姓名。
        """

        return f"你好，{name}" # 这里故意写错参数名，观察错误信息。

except ValueError as error:
    print("无效 Docstring 错误类型：", type(error).__name__)
    print("无效 Docstring 错误：", str(error))

实际输出：

Annotated 工具结果： 10
Docstring 工具结果： 朋友小明，你好！
无效 Docstring 错误类型： ValueError
无效 Docstring 错误： Arg username in docstring not found in function signature.

启用 parse_docstring=True 后，Args: 中的参数名称必须和函数签名一致。示例故意把 name 写成 username，工具在创建阶段就抛出了 ValueError，而不是等到模型调用时才失败。

这些写法都属于第一种实现方式，只是参数说明方式不同，可以按复杂程度选择：

参数很少时，使用类型注解和 Annotated。
已经有规范 Google 风格 Docstring 时，使用 parse_docstring=True。
需要枚举、默认值、字段约束和独立复用时，使用 Pydantic。

4.6 使用 StructuredTool 包装函数和异步协程

@tool 适合直接装饰函数。如果需要显式指定同步函数、异步协程、名称和描述，可以使用 StructuredTool.from_function()。这仍然属于包装 Python 函数或协程的实现方式。

langgraph/p05_local_tools/04_structured_tool.py：

"""使用 StructuredTool 为同一个工具配置同步函数和异步协程。"""

import asyncio

from langchain_core.tools import StructuredTool

def add_sync(a: float, b: float) -> str:
    """同步计算两个数字之和。"""

    return f"同步结果：{a + b}"

async def add_async(a: float, b: float) -> str:
    """异步计算两个数字之和。"""
    # sleep(0) 主动让出一次执行权，用于演示真实协程。
    await asyncio.sleep(0)
    return f"异步结果：{a + b}"

# 第 1 步：把同步函数和异步函数注册为同一个工具。
calculator_tool = StructuredTool.from_function(
    func=add_sync,
    coroutine=add_async,
    name="structured_calculator",
    description="计算两个数字之和。",
)

# 第 2 步：invoke() 走同步函数，ainvoke() 走异步函数。
print("工具类型：", type(calculator_tool).__name__)
print("同步调用：", calculator_tool.invoke({"a": 2, "b": 3}))
print("异步调用：", asyncio.run(calculator_tool.ainvoke({"a": 2, "b": 3})))

实际输出：

工具类型： StructuredTool
同步调用： 同步结果：5.0
异步调用： 异步结果：5.0

invoke() 调用同步函数，ainvoke() 调用异步协程。对于网络请求、数据库访问等 I/O 操作，可以提供真正的异步实现，避免在异步程序中阻塞事件循环。

5. 把 Runnable 转换成 Tool

如果已有一段 Runnable 逻辑，不必再复制成新函数，可以使用 as_tool() 转换。

langgraph/p05_local_tools/05_runnable_as_tool.py：

"""把一个简单 Runnable 转换为 LangChain Tool。"""

import json

from langchain_core.runnables import RunnableLambda
from pydantic import BaseModel, Field


class GreetingInput(BaseModel):
    """问候语 Runnable 的输入参数。"""

    name: str = Field(description="需要问候的姓名")


# RunnableLambda 是最简单的 Runnable，输入和输出逻辑保持直观。
greeting_runnable = RunnableLambda(
    lambda values: f"{values['name']}，欢迎使用 LangChain！"
)

# as_tool() 当前仍会提示 Beta API 警告。
greeting_tool = greeting_runnable.as_tool(
    args_schema=GreetingInput,
    name="generate_greeting",
    description="为指定姓名生成一条欢迎语。",
)

print("工具类型：", type(greeting_tool).__name__)
print("工具名称：", greeting_tool.name)
print("参数 Schema：")
print(json.dumps(greeting_tool.args_schema.model_json_schema(), ensure_ascii=False, indent=2))
print("调用结果：", greeting_tool.invoke({"name": "小明"}))

实际运行时先出现 Beta 警告，然后正常得到结果：

LangChainBetaWarning: This API is in beta and may change in the future.
工具类型： StructuredTool
工具名称： generate_greeting
调用结果： 小明，欢迎使用 LangChain！

Beta 表示 API 仍处于测试状态，不表示不能使用。用于长期维护的项目时，应在升级依赖后重新测试。

6. 继承 BaseTool 自定义 Tool

BaseTool 是工具的基础抽象。继承它需要自己声明名称、描述、参数模型，并实现 _run()。

langgraph/p05_local_tools/06_base_tool.py：

"""继承 BaseTool，创建一个只查询本地固定书单的工具。"""

from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field


BOOKS = [
    "Python 编程入门",
    "杭州旅行指南",
    "中国古诗词精选",
]


class BookSearchInput(BaseModel):
    """本地书籍搜索工具的输入参数。"""

    keyword: str = Field(description="书名中需要包含的关键词")

class LocalBookSearchTool(BaseTool):
    """在内存中的固定书单里搜索书名。"""

    name: str = "search_local_books"
    description: str = "根据关键词在本地固定书单中查找书名。"
    args_schema: type[BaseModel] = BookSearchInput

    def _run(self, keyword: str) -> str:
        """执行同步书名搜索。"""

        # 使用普通 for 循环，清楚展示“逐本检查并保存匹配项”的过程。
        matches = []
        for book in BOOKS:
            if keyword in book:
                matches.append(book)

        if not matches:
            return "没有找到匹配的书籍"
        return "、".join(matches)


# 创建实例后，便可以像其他 LangChain Tool 一样调用 invoke()。
book_search = LocalBookSearchTool()

print("工具类型：", type(book_search).__name__)
print("工具名称：", book_search.name)
print("搜索‘杭州’：", book_search.invoke({"keyword": "杭州"}))
print("搜索‘音乐’：", book_search.invoke({"keyword": "音乐"}))

实际输出：

工具类型： LocalBookSearchTool
工具名称： search_local_books
搜索‘杭州’： 杭州旅行指南
搜索‘音乐’： 没有找到匹配的书籍

这个示例只查询内存中的固定书单，不访问网络。对于简单、无状态的函数，继承 BaseTool 会显得过重；当工具需要持有客户端、连接池、配置或复杂生命周期时，这种方式更容易组织代码。

7. 使用本地 Qwen3 完成工具调用

前面的示例都是应用直接执行 Tool。接下来把计算器 Schema 绑定到本地 Qwen3，让模型自己决定工具名称和参数。

7.1 启动本地模型服务

在 llm_learning 根目录使用 LangGraph 系列 3 创建的模型服务环境启动 MLX-LM：

cd /path/to/llm-learning
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 \
  --chat-template-args '{"enable_thinking": false}'

另开终端执行健康检查：

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

确认返回模型列表后再运行代码。否则连接失败、502 或模型加载错误容易被误判成 Tool 的问题。

7.2 完整代码

langgraph/p05_local_tools/07_bind_tools_with_qwen3.py：

"""把本地计算器绑定到 Qwen3，并手动完成一次工具调用循环。"""

from typing import Literal

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


# trust_env=False 避免系统代理或 VPN 转发本机请求。
with httpx.Client(trust_env=False) as http_client:
    model = ChatOpenAI(
        model="Qwen3-14B-AWQ-4bit-MLX",
        base_url="http://127.0.0.1:18080/v1",
        api_key="not-needed",
        temperature=0,
        max_tokens=128,
        http_client=http_client,
    )

    # 工具列表；新增工具时只需 append，执行阶段按 name 自动分发。
    tools = [calculator]
    tools_by_name = {current_tool.name: current_tool for current_tool in tools}

    # 第 1 步：把工具的名称、说明和参数 Schema 发送给模型。
    model_with_tools = model.bind_tools(tools)
    messages = [HumanMessage(content="请使用计算器工具计算 18 乘以 7。")]

    resp = model_with_tools.invoke(messages)
    print(f"resp: {resp.content}")

    tool_call = resp.tool_calls[0]
    print(f"tool_call: {tool_call}")

    # 按 tool_call["name"] 查找工具。
    tool_result = tools_by_name[tool_call["name"]].invoke(tool_call["args"])
    
    print(f"tool_result: {tool_result}")
    # 当前 MLX-LM 返回的 tool call id 可能是 None。
    # 给调用补一个明确 ID，使 ToolMessage 能与模型请求一一对应。
    tool_call_id = tool_call["id"] or "local_tool_call"
    print(f"tool_call_id: {tool_call_id}")

    tool_call["id"] = tool_call_id

    tool_message = ToolMessage(
        content=str(tool_result),
        tool_call_id=tool_call_id,
        name=tool_call["name"],
    )
    print("工具消息类型：", type(tool_message).__name__)
    print("工具执行结果：", tool_message.content)
    messages.append(tool_message)

    # 第 3 步：把工具结果交回模型，由模型组织最终回答。
    final_message = model_with_tools.invoke(messages)
    print("最终回答：", final_message.content)

运行：

source .venv_langgraph/bin/activate
python langgraph/p05_local_tools/07_bind_tools_with_qwen3.py

一次真实运行结果：

模型选择的工具： calculator
模型生成的参数： {'a': 18, 'b': 7, 'operation': 'multiply'}
tool_call_id: 123a4b70-2931-4042-be1a-e1501af95154
工具消息类型： ToolMessage
工具执行结果： 126.0
最终回答： 18 乘以 7 的结果是 126。

模型回答可能有轻微措辞变化，但工具名称、参数结构和计算结果应保持正确。

下图按照真实消息顺序展开这次调用。第一次模型请求只得到 tool_calls，中间由应用执行计算器；第二次模型请求必须同时包含用户消息、原始 AIMessage 和匹配调用 ID 的 ToolMessage。

本地 Qwen3 手动工具调用时序图

8. 手动工具调用循环是怎样工作的

这段代码包含两个模型调用，中间夹着一次本地函数调用。

8.1 bind_tools() 只绑定工具说明

model_with_tools = model.bind_tools([calculator])

bind_tools() 把计算器的 JSON Schema 附加到模型请求中。它不会提前执行工具，也不会给模型 Python 运行权限。

8.2 第一次模型调用生成 tool_calls

ai_message = model_with_tools.invoke(messages)

Qwen3 返回 AIMessage，其中的 tool_calls 包含工具名称、参数和调用 ID。此时还没有计算出结果。

8.3 应用程序执行本地工具

#calculator，所以可以直接执行它，不需要额外的工具分发表。
tool_result = calculator.invoke(tool_call["args"])
tool_call_id = tool_call["id"] or "local_tool_call"
tool_call["id"] = tool_call_id
tool_message = ToolMessage(
    content=str(tool_result),
    tool_call_id=tool_call_id,
    name=tool_call["name"],
)

应用根据工具名找到 calculator，使用 args 执行 Python 函数，再把结果封装为 ToolMessage。本次 MLX-LM 返回的调用 ID 可能是 None，因此代码先补充本地 ID，确保发回模型的工具结果可以与原工具调用对应。这一步才真正计算了 18 × 7。

实际项目还应检查模型返回的工具名是否在允许列表中，并对可能产生副作用的工具增加权限、超时和人工确认。

8.4 第二次模型调用组织最终回答

final_message = model_with_tools.invoke(messages)

消息列表中已经包含：

用户问题。
模型生成的工具调用请求。
应用返回的 ToolMessage。

Qwen3 根据工具结果生成用户能够直接阅读的最终回答。这就是一次最小但完整的工具调用循环。Agent 会自动完成类似循环，而本篇手动展开是为了看清每一步的职责。

9. Qwen3 工具调用的基础检查

工具定义正确，并不代表任意模型服务都能正确返回 tool_calls。排查时应按下面的顺序进行。

9.1 先检查模型服务

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

模型服务没有启动时，先解决端口、模型路径和虚拟环境问题，不要先修改工具代码。

9.2 检查模型是否真的返回 tool_calls

print(ai_message.tool_calls)

如果列表为空，常见原因包括：

模型或服务端不支持 OpenAI 兼容的工具调用格式。
工具描述不清楚，模型认为不需要调用工具。
参数 Schema 过于复杂或字段说明不足。
提示词没有明确要求使用工具。

本篇使用的本地 Qwen3 和 MLX-LM 服务已经真实返回标准 tool_calls，不需要切换到其他推理框架。

9.3 先使用非流式调用排查

工具调用可能被分散到多个流式消息块中。初次调试时使用普通 invoke() 更容易观察完整的 AIMessage.tool_calls。确认非流式调用正确后，再处理 AIMessageChunk、消息块合并和推理服务解析器；这些内容放在下一篇完整展开。

10. 绑定多工具并动态调用

代码:

"""定义多个 BaseTool 子类，让 Qwen3 根据问题自动选择并调用合适工具。

与 07 的区别：
  07 只有 1 个 calculator，手动完成单轮工具循环；
  08 有 3 个工具，按 tool_call["name"] 动态分发，并测试 3 个不同问题。

运行前需启动本地 Qwen3（18080 端口），与 07 相同。
"""

from typing import Literal

import httpx
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_core.tools import BaseTool
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field


# ---------------------------------------------------------------------------
# 工具 1：计算器
# ---------------------------------------------------------------------------
class CalculatorInput(BaseModel):
    """计算器输入。"""

    a: float = Field(description="第一个数字")
    b: float = Field(description="第二个数字")
    operation: Literal["add", "multiply"] = Field(
        description="运算类型：add 加法，multiply 乘法"
    )


class CalculatorTool(BaseTool):
    """对两个数字做加法或乘法。"""

    name: str = "calculator"
    description: str = "对两个数字做加法或乘法，返回字符串形式的结果。"
    args_schema: type[BaseModel] = CalculatorInput

    def _run(self, a: float, b: float, operation: str) -> str:
        if operation == "add":
            return str(a + b)
        return str(a * b)


# ---------------------------------------------------------------------------
# 工具 2：模拟天气查询
# ---------------------------------------------------------------------------
class WeatherInput(BaseModel):
    """天气查询输入。"""

    city: str = Field(description="要查询的城市名称，例如杭州、北京")


class WeatherTool(BaseTool):
    """查询指定城市的模拟天气（本地固定数据，不访问真实 API）。"""

    name: str = "get_weather"
    description: str = "查询指定城市今天的天气和气温。"
    args_schema: type[BaseModel] = WeatherInput

    _WEATHER_DATA = {
        "杭州": "晴朗，28 摄氏度",
        "北京": "多云，22 摄氏度",
        "上海": "小雨，19 摄氏度",
    }

    def _run(self, city: str) -> str:
        for known_city, weather in self._WEATHER_DATA.items():
            if known_city in city:
                return f"{known_city}今天{weather}。"
        return f"暂无 {city} 的天气数据，目前支持：杭州、北京、上海。"


# ---------------------------------------------------------------------------
# 工具 3：本地书单搜索
# ---------------------------------------------------------------------------
class BookSearchInput(BaseModel):
    """书单搜索输入。"""

    keyword: str = Field(description="书名中需要包含的关键词")


class BookSearchTool(BaseTool):
    """在本地固定书单中按关键词搜索书名。"""

    name: str = "search_books"
    description: str = "根据关键词在本地书单中查找匹配的书名。"
    args_schema: type[BaseModel] = BookSearchInput

    _BOOKS = [
        "Python 编程入门",
        "杭州旅行指南",
        "LangGraph 实战教程",
    ]

    def _run(self, keyword: str) -> str:
        matches = [book for book in self._BOOKS if keyword in book]
        if not matches:
            return "没有找到匹配的书籍"
        return "、".join(matches)


def run_tool_loop(
    model_with_tools,
    tools_by_name: dict,
    question: str,
) -> None:
    """执行一轮完整的工具调用循环：模型决策 → 执行工具 → 生成最终回答。"""
    messages: list = [HumanMessage(content=question)]

    print(f"\n{'=' * 60}")
    print(f"用户问题：{question}")

    # 第 1 步：模型决定调用哪个（或哪些）工具。
    response: AIMessage = model_with_tools.invoke(messages)
    print(f"模型文本回复：{response.content!r}")
    print(f"工具调用请求：{response.tool_calls}")

    if not response.tool_calls:
        print("（模型未调用任何工具，直接回答）")
        return

    # 第 2 步：按 name 动态分发，支持一次返回多个 tool_calls。
    messages.append(response)
    for index, tool_call in enumerate(response.tool_calls):
        tool_name = tool_call["name"]
        tool_call_id = tool_call["id"] or f"local_tool_call_{index}"
        print(f"\n  → 执行工具 [{tool_name}]，参数：{tool_call['args']}")

        tool_result = tools_by_name[tool_name].invoke(tool_call["args"])
        print(f"  → 工具返回：{tool_result}")

        messages.append(
            ToolMessage(
                content=str(tool_result),
                tool_call_id=tool_call_id,
                name=tool_name,
            )
        )

    # 第 3 步：把工具结果交回模型，组织自然语言回答。
    final_message = model_with_tools.invoke(messages)
    print(f"\n最终回答：{final_message.content}")


# --- 初始化模型与工具 ---
tools = [CalculatorTool(), WeatherTool(), BookSearchTool()]
tools_by_name = {current_tool.name: current_tool for current_tool in tools}

with httpx.Client(trust_env=False) as http_client:
    model = ChatOpenAI(
        model="Qwen3-14B-AWQ-4bit-MLX",
        base_url="http://127.0.0.1:18080/v1",
        api_key="not-needed",
        temperature=0,
        max_tokens=256,
        http_client=http_client,
    )
    model_with_tools = model.bind_tools(tools)

    print("已注册工具：", [tool.name for tool in tools])

    # 三个问题分别应触发不同工具，便于观察模型的选择行为。
    run_tool_loop(model_with_tools, tools_by_name, "请用计算器算 12 乘以 9。")
    run_tool_loop(model_with_tools, tools_by_name, "杭州今天天气怎么样？")
    run_tool_loop(model_with_tools, tools_by_name, "帮我找书名里包含 Python 的书。")

结果:

已注册工具： ['calculator', 'get_weather', 'search_books']

============================================================
用户问题：请用计算器算 12 乘以 9。
模型文本回复：''
工具调用请求：[{'name': 'calculator', 'args': {'a': 12, 'b': 9, 'operation': 'multiply'}, 'id': '4570aac7-8644-40d6-b16b-47eca0d5ed3d', 'type': 'tool_call'}]

  → 执行工具 [calculator]，参数：{'a': 12, 'b': 9, 'operation': 'multiply'}
  → 工具返回：108.0

最终回答：12 乘以 9 的结果是 108.0。

============================================================
用户问题：杭州今天天气怎么样？
模型文本回复：''
工具调用请求：[{'name': 'get_weather', 'args': {'city': '杭州'}, 'id': '01efc72b-9f05-4e67-b0c8-cacc2728afa3', 'type': 'tool_call'}]

  → 执行工具 [get_weather]，参数：{'city': '杭州'}
  → 工具返回：杭州今天晴朗，28 摄氏度。

最终回答：杭州今天的天气晴朗，气温为28摄氏度。

============================================================
用户问题：帮我找书名里包含 Python 的书。
模型文本回复：''
工具调用请求：[{'name': 'search_books', 'args': {'keyword': 'Python'}, 'id': 'c46594d1-fb78-4ae0-b71f-9da8d3c730c0', 'type': 'tool_call'}]

  → 执行工具 [search_books]，参数：{'keyword': 'Python'}
  → 工具返回：Python 编程入门

最终回答：找到一本包含 "Python" 的书：《Python 编程入门》。

11. 小结

Tool 是模型和应用能力之间的结构化接口。模型依据 Schema 选择工具和生成参数，应用程序负责校验、执行和回传结果。

LangChain 实现本地 Tool 主要有三种方式：包装 Python 函数或协程、把 Runnable 转换成 Tool、继承 BaseTool 自定义实现。对于大多数简单函数，使用 @tool 已经足够；需要明确同步和异步实现时使用 StructuredTool.from_function()；已有 Runnable 可以通过 as_tool() 转换；需要封装状态或复杂生命周期时再继承 BaseTool。

本篇最后的 Qwen3 示例没有依赖 Agent，手动展示了 AIMessage.tool_calls -> Tool 执行 -> ToolMessage -> 最终 AIMessage 的完整过程。理解这个循环后，再使用 Agent 自动执行工具时，每一层的职责会更清楚。

实现方式	具体写法	参数 Schema	适用场景	本篇实测结果
Python 函数或协程	直接交给 Agent、@tool、StructuredTool.from_function()	类型注解、Pydantic、Annotated 或 Docstring	最常用，适合无状态或逻辑清晰的函数	同步和异步函数均执行成功
Runnable	Runnable.as_tool()	建议显式提供 Pydantic Schema	复用已有 Runnable 调用链	执行成功，但会给出 Beta 警告
BaseTool 子类	继承 BaseTool，实现 _run() / _arun()	显式定义	封装状态、客户端或复杂生命周期	本地书单查询成功