第 8 章工具调用与 ToolNode

8.1 Function Calling / Tool Use 协议

Function Calling（函数调用，亦称 Tool Use） 是大模型与外部工具交互的标准化协议。OpenAI 于 2023 年 6 月在 gpt-3.5-turbo-0613 与 gpt-4-0613 版本中首次推出该能力，随后 Anthropic、Google、国产模型厂商陆续在各自 API 中对齐或兼容该协议，使之成为当前 Agent 工具调用的行业事实标准。协议的核心约定是：模型在响应中不直接返回自然语言文本作为最终答案，而是返回一个结构化的 tool_calls 字段，声明希望调用的工具名称、参数对象、调用 ID；应用侧接收后按名查找工具并执行，将结果以 ToolMessage（工具消息）形式回传模型；模型看到结果后决定是再次调用工具还是生成最终自然语言回复。该协议相比早期的 ReAct prompt 解析（从模型自然语言输出中正则提取 <action> 标签）具备可靠性高、参数结构化、支持并行、可审计的优势。

tool_calls 字段结构（LangChain 标准消息格式）：

python

from langchain_core.messages import AIMessage, ToolMessage

# 模型发起工具调用
ai_msg = AIMessage(
    content="",  # 发起工具调用时 content 通常为空，也可以同时带思考文本
    tool_calls=[
        {
            "id": "call_abc123",          # 调用唯一 ID，由模型生成，用于与 ToolMessage.tool_call_id 配对
            "name": "get_weather",        # 工具名，必须与注册工具的 name 精确匹配（大小写敏感）
            "args": {"city": "北京"},      # 参数 JSON 对象，必须符合工具的 JSON Schema
        },
        {
            "id": "call_def456",
            "name": "get_exchange_rate",
            "args": {"base": "USD", "target": "CNY"},
        },
    ],
)

# 应用侧返回工具结果
tool_msg = ToolMessage(
    content="北京当前晴，25℃",
    tool_call_id="call_abc123",  # 必须与 AIMessage.tool_calls[i].id 一一对应
    name="get_weather",
)

协议的关键约束有三点：第一，tool_calls 是列表而非单个对象，一次响应可包含多个工具调用，应用侧应并行执行；第二，id 字段由模型生成，ToolMessage.tool_call_id 必须精确回填该值，否则模型无法将结果与调用对应，可能出现"看不到工具结果、重复调用"的错误；第三，args 是已解析好的 Python dict（LangChain 反序列化后），应用侧直接按键取值，无需再做 JSON 解析。

多模型适配差异：

模型提供商	原生工具调用字段	多工具并行	备注
OpenAI（gpt-4o/gpt-4o-mini/o1/o3/o4-mini）	`tool_calls`（id/name/args）	支持	事实标准，`tool_choice` 参数控制强制/可选/指定工具
Anthropic（Claude 3.5 Sonnet、Claude 3.7 Sonnet/Opus）	`tool_use` content block（id/name/input）	支持	LangChain ChatAnthropic 自动转换为标准 tool_calls 格式
Google Gemini（1.5 Pro/Flash、2.0 Flash）	`function_call` 字段	支持（部分模型）	参数以结构化 dict 返回，LangChain 适配层转换
阿里通义千问（Qwen2.5、Qwen3 系列）	`tool_calls`（对齐 OpenAI 格式）	支持（Qwen2.5-72B/Qwen3-235B 等）	开源版本需使用支持 tool_calls 的 ChatModel 封装
深度求索（DeepSeek-V3/R1）	`tool_calls`（对齐 OpenAI 格式）	支持	DeepSeek-V3 工具调用稳定性较好
智谱 GLM-4	`tool_calls`	早期版本单工具，新版本并行	需使用最新 langchain-zhipu 适配包

对于不支持原生 tool_calls 的模型（如某些小模型、早期开源模型），需退回到 Prompt-based 工具调用：在 prompt 中约定输出格式（如 XML 标签 <tool><name>...</name><args>...</args></tool>），应用侧用正则或 XML 解析提取工具调用信息。该方式可靠性显著低于原生 tool_calls，参数错误率高，生产环境应优先选择原生支持的模型。

8.2 @tool 装饰器与 StructuredTool

LangChain 提供两种将 Python 可调用对象转换为 BaseTool（工具基类）的方式。

@tool 装饰器（langchain_core.tools.tool）是最简洁的工具声明方式：应用于普通 Python 函数，自动将函数名作为工具名（name），函数 docstring 作为工具描述（description），函数参数的类型注解自动转换为工具参数的 JSON Schema（args_schema）。

python

from langchain_core.tools import tool

@tool
def get_weather(city: str, unit: str = "celsius") -> str:
    """查询指定城市的当前天气。

    Args:
        city: 要查询的城市名称，支持中文与英文。
        unit: 温度单位，可选 "celsius"（摄氏度，默认）或 "fahrenheit"（华氏度）。
    """
    # 实际场景调用第三方天气 API（如和风天气、OpenWeatherMap）
    return f"{city} 当前晴，25{ '℃' if unit == 'celsius' else '℉' }，湿度 40%，微风"

# 检查装饰器生成的元数据
print(get_weather.name)         # "get_weather"
print(get_weather.description)  # 自动从 docstring 提取
print(get_weather.args)         # 自动从类型注解生成 JSON Schema
# get_weather.invoke({"city": "北京"})  # 直接调用

使用 @tool 时需要注意：docstring 的 Args/Parameters 段是模型选择工具、填写参数的核心依据，缺失或模糊会直接导致工具调用失败；未加类型注解的参数会被识别为 anyOf，模型可能填入任意类型值；默认参数（如 unit）在 schema 中标记为非必填，模型可省略。

StructuredTool（langchain_core.tools.StructuredTool）是显式构造方式，适合需要精细控制工具元数据、工具逻辑来自类方法、或需要对参数做严格 Pydantic 校验的场景。args_schema 必须是 Pydantic BaseModel 子类，通过 Field(description=...) 为每个参数提供给模型看的说明。

python

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

class WeatherInput(BaseModel):
    city: str = Field(description="要查询的城市名称，支持中英文")
    unit: str = Field(default="celsius", description="温度单位：celsius 或 fahrenheit")

def _get_weather_impl(city: str, unit: str) -> str:
    return f"{city} 当前晴，25{ '℃' if unit == 'celsius' else '℉' }"

weather_tool = StructuredTool.from_function(
    func=_get_weather_impl,
    name="get_weather",
    description="查询指定城市的当前天气信息，包括温度、湿度、风力",
    args_schema=WeatherInput,
    response_format="content",  # 或 "content_and_artifact" 返回 (文本, 结构化产物) 元组
)

response_format="content_and_artifact" 时工具函数返回 (content_str, artifact_obj) 元组：content 作为 ToolMessage.content 回传给模型；artifact 为结构化对象，通过 ToolMessage.artifact 字段暴露给下游节点处理（如存入 State 或做后续计算），不会让模型直接看到，适合"模型可读摘要+应用可消费结构"的场景。

8.3 ToolNode 详解

ToolNode（langgraph.prebuilt.ToolNode）是 LangGraph 预构建的标准化工具执行节点，封装了"读取模型 tool_calls→按名匹配工具→解析参数→并发执行→结果封装为 ToolMessage→追加到 messages"的完整流程，是构建 Agent 的核心积木。使用者无需手动遍历 tool_calls 或维护调用 ID 映射。

构造函数参数：

参数	类型	默认值	说明
tools	Sequence[BaseTool \| Callable]	必填	工具列表；若传入普通 Python 函数会自动通过 `create_tool` 转为 StructuredTool
name	str	`"tools"`	节点名，默认 "tools"，tools_condition 默认路由目标名与之对应；若改名则需在条件路由里同步修改
handle_tool_errors	bool / str / Callable / tuple[type[Exception], ...]	`True`	异常处理策略，详见下文
messages_key	str	`"messages"`	State 中消息列表字段名；若 State 使用非 "messages" 字段存储消息则需显式指定
tags	list[str] / None	None	LangSmith trace 标签
wrap_tool_call / awrap_tool_call	Callable / None	None	同步/异步工具调用包装器，可用于统一注入日志、超时、审计、幂等键等横切逻辑

执行流程按以下步骤进行（对应 ToolNode._func 与 _afunc 的源码顺序）：

从 state[messages_key] 取最后一条消息（必须是 AIMessage，否则直接返回空增量）；
遍历 AIMessage.tool_calls 列表；若列表为空则返回空 dict（无消息追加）；
对每个 tool_call，按 name 在 _tools_by_name 字典（构造时构建）中查找对应工具；找不到工具则构造一条 content 为错误文本、status="error" 的 ToolMessage；
解析 args（已是 dict），提取被 InjectedState/InjectedStore/InjectedToolArg 标记的参数由运行时自动注入，其余参数来自 LLM 传入；参数校验由 Pydantic 自动执行，校验失败进入错误处理；
对所有工具调用使用 asyncio.gather 并发执行（ainvoke 路径）；同步 invoke 路径使用线程池包装以避免阻塞；
若工具执行抛异常，根据 handle_tool_errors 策略处理（见下表），异常被转换为 status="error" 的 ToolMessage；
将每个工具结果（正常或异常）封装为 ToolMessage(content=..., tool_call_id=call_id, name=tool_name, status=status)，组成列表作为 {messages_key: [...]} 增量返回。

并行工具调用机制：一个 AIMessage 可以包含多个 tool_calls（例如用户问"北京天气和美元汇率"，模型同时调用 get_weather 与 get_exchange_rate），ToolNode 会把所有调用提交到 asyncio.gather 并发执行，总耗时为最慢工具的耗时而非串行求和。所有结果作为同一超步的写入提交到 messages Channel，下一轮超步 call_model 节点一次看到全部工具结果。

handle_tool_errors 策略对比：

取值	行为	适用场景
`True`（默认）	捕获所有 Exception，将异常文本字符串作为 ToolMessage.content 返回，status="error"	生产对话场景，模型能看到错误信息并自我纠错（换参数重试或友好回复）
`False`	异常直接向上抛出，整图在当前超步中断	单元测试期快速暴露 bug；关键工具必须成功的场景（失败则整体失败）
字符串模板（如 `"Tool error: {e}"`）	使用模板格式化异常文本作为 ToolMessage 内容	需要向模型传递特定格式错误提示
可调用对象 `f(e: Exception) -> str`	调用自定义处理函数，返回字符串作为错误内容	按异常类型分级处理（鉴权错误提示用户、超时错误建议重试）
异常类型元组（如 `(ConnectionError, TimeoutError)`）	仅捕获指定类型异常，其余异常原样抛出	精准控制容错范围，未预期异常立即失败

8.4 tools_condition 路由函数与 bind_tools

tools_condition（langgraph.prebuilt.tools_condition）是 LangGraph 预置的条件路由函数，逻辑简洁而常用：检查 state["messages"] 的最后一条消息——若该消息是 AIMessage 且包含非空 tool_calls，则路由到字符串 "tools"（即 ToolNode 的默认 name）；否则路由到 END。这是 ReAct 循环中"模型→工具/结束"分支的标准实现。若 ToolNode 使用了自定义 name（如 name="tool_executor"），则需在 add_conditional_edges 的 path_map 中把 "tools" 映射到自定义名，或使用自定义路由函数替代。

bind_tools 是 LangChain BaseChatModel 的标准方法（model.bind_tools(tools, **kwargs)），其作用是将工具列表的 JSON Schema 注入到模型请求的 tools 参数中，让模型感知工具存在并知道如何调用。bind_tools 返回一个新的 Runnable 实例（原模型不被修改），调用该实例时模型会在适当时机输出 tool_calls。

关键参数 tool_choice（部分模型支持，如 OpenAI）控制工具调用行为：

tool_choice 值	行为
`"auto"`（默认）	模型自主判断是否需要调用工具、调用哪些工具
`"any"` / `"required"`	强制模型必须至少调用一个工具（适合需要强制触发工具的节点）
具体工具名字符串（如 `"get_weather"`）	强制模型调用指定工具，不调用其他工具也不直接回复
`"none"` / `None`	不允许调用工具，模型只输出文本（相当于未 bind_tools 的原始模型）

8.5 完整 LLM↔Tools 循环（手写 ReAct，不使用 create_react_agent）

以下示例完整展示如何使用 StateGraph、ToolNode、tools_condition 从零构建一个标准 ReAct 循环（即 create_react_agent 的等价内部实现），不依赖预构建工厂函数，便于理解每个组件的职责。

python

# langgraph 1.2.x
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.tools import tool
from langchain_core.messages import AIMessage
from langchain_openai import ChatOpenAI

# 1) 定义工具
@tool
def get_weather(city: str) -> str:
    """查询指定城市的当前天气。

    Args:
        city: 要查询的城市名称，支持中英文。
    """
    # 实际场景调用天气 API
    return f"{city} 当前晴，气温 25℃，湿度 40%，东北风 2 级"

tools = [get_weather]

# 2) 绑定工具到模型
# 必须执行 bind_tools 否则模型不会输出 tool_calls
llm_with_tools = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)

# 3) 定义 State（使用 Annotated + add_messages 实现消息追加合并）
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 4) 定义模型调用节点：接收 State.messages，调用绑定了工具的模型，返回增量
def call_model(state: State) -> dict:
    ai_message = llm_with_tools.invoke(state["messages"])
    return {"messages": [ai_message]}

# 5) 构建图：call_model → tools_condition →（有 tool_calls）tools → call_model（循环）
#                                            └→（无 tool_calls）END
builder = StateGraph(State)
builder.add_node("call_model", call_model)
builder.add_node("tools", ToolNode(tools, handle_tool_errors=True))
builder.add_edge(START, "call_model")
builder.add_conditional_edges(
    "call_model",
    tools_condition,  # 内置路由函数
    {"tools": "tools", END: END},  # path_map
)
# 关键回边：工具执行完必须回到 call_model 让模型看到结果并决定下一步
builder.add_edge("tools", "call_model")

graph = builder.compile()

# 6) 运行
result = graph.invoke(
    {"messages": [{"role": "user", "content": "请查询北京今天的天气，并告诉我是否需要带伞"}]},
)
for msg in result["messages"]:
    role = msg.type
    if msg.type == "ai" and msg.tool_calls:
        calls = [f"{tc['name']}({tc['args']})" for tc in msg.tool_calls]
        print(f"[AI] → 调用工具：{', '.join(calls)}")
    elif msg.type == "tool":
        print(f"[Tool:{msg.name}] → {msg.content[:60]}")
    else:
        print(f"[{role}] → {msg.content[:80]}")

执行路径示例：START→call_model（模型判断需要查天气，输出 tool_calls=[get_weather(北京)]）→tools_condition 路由到 "tools"→ToolNode 并发执行 get_weather→返回 ToolMessage→tools→call_model 回边回到 call_model（模型看到天气结果，判断不需要再调工具，生成最终文字回答）→tools_condition 路由到 END→返回终态。

8.6 工具权限安全与 InjectedState/InjectedStore

工具本质是模型可调用的代码，安全性问题不可忽视。以下类型的工具在开放给模型直接调用前必须做严格防护：执行任意代码（exec、eval、subprocess、shell 命令）、写入数据库/文件系统/对象存储、对外发送 HTTP 请求、发邮件/短信/通知、涉及资金转账或权限变更。防护手段包括：

沙箱隔离：代码执行工具放入容器沙箱（如 gVisor、Firecracker、Docker 容器），限制网络、CPU、内存、文件系统访问；
人工审批（HITL）：对写入类、支付类、权限类工具，在 ToolNode 前插入 interrupt 审批节点，人类确认后 resume 才继续执行；
权限白名单：数据库工具只暴露 SELECT 或限定表的写操作，文件工具限制可访问目录；
参数校验：通过 Pydantic args_schema 严格限制参数范围（如金额上限、操作类型枚举）；
审计日志：每次工具调用记录 user_id、tool_name、args、结果、耗时，落盘到独立审计系统。

InjectedState / InjectedStore 是 LangGraph 提供的工具函数参数注入机制，用于让工具函数能够读取当前图 State 或 Store 实例，但这些被注入的参数不会出现在工具的 JSON Schema 中——LLM 既看不到也无法为它们传值，完全由运行时在调用工具前自动填充。

python

from typing import Annotated, Any
from langgraph.prebuilt import InjectedState, InjectedStore
from langgraph.store.base import BaseStore
from langchain_core.tools import tool

@tool
def lookup_user_preference(
    feature: str,                                   # 暴露给模型的业务参数
    state: Annotated[dict, InjectedState],          # 运行时自动注入当前 State
    store: Annotated[BaseStore, InjectedStore],     # 运行时自动注入 Store
) -> str:
    """查询当前用户对指定功能的偏好设置。

    Args:
        feature: 功能名称，如 "language"、"response_style"。
    """
    user_id = state.get("user_id")
    item = store.get(("preferences", user_id), key=feature)
    if item is None:
        return f"用户未设置 {feature} 偏好"
    return f"用户 {user_id} 的 {feature} 偏好：{item.value}"

使用注入机制时需要注意：注入参数必须通过 Annotated[Type, InjectedXxx] 形式声明；工具被 ToolNode 调用时注入才生效，直接 tool.invoke(...) 手动调用不会触发注入；InjectedState 注入的是只读状态快照，工具不应在其中做 in-place 修改。

本章小结

Function Calling/Tool Use 是大模型与外部工具交互的标准化协议，以 AIMessage.tool_calls 发起、ToolMessage 回填结果、tool_call_id 做一一配对。@tool 装饰器与 StructuredTool 是两种工具声明方式：前者简洁（装饰普通函数即可），后者适合需要 Pydantic 严格校验与元数据精细控制的场景。ToolNode 是 LangGraph 预构建的标准化工具执行节点，负责读取 tool_calls、按名匹配、并发执行（asyncio.gather）、异常封装、结果回写；handle_tool_errors 参数控制异常传播策略。tools_condition 实现"有工具调用→tools 节点、无→END"的标准路由；bind_tools 将工具 schema 注入模型让其感知工具存在。手写 ReAct 循环的最小图结构为：call_model→tools_condition→tools/END、tools→call_model 回边。工具权限安全通过沙箱隔离、人工审批（interrupt）、权限白名单、参数校验、审计日志五层防护；InjectedState/InjectedStore 让工具隐式读取运行时资源而不暴露给 LLM。

常见错误

忘记 bind_tools 导致模型不输出 tool_calls。现象：图进入 call_model 后模型始终只回复自然语言文本，tools_condition 永远路由到 END，工具永远不会被调用。原因：llm 实例未调用 .bind_tools(tools)，模型在请求中看不到工具 schema，自然不会生成 tool_calls。修复：在构图前对模型实例调用 llm.bind_tools(tools)，将绑定后的新实例传入 call_model 节点。
tool docstring 或 Field description 不清晰导致模型选错工具。现象：存在多个语义相近的工具时，模型频繁选错工具或填错参数。原因：docstring 只写了一句话功能描述，未说明"什么时候用、什么时候不用"，参数说明缺失。修复：在 docstring 首句清晰描述工具功能边界，Args 段为每个参数说明语义与取值范围，Pydantic Field(description=...) 补充参数约束。
tool_calls 的 id 与 ToolMessage.tool_call_id 不匹配。现象：工具执行完回到 call_model 后，模型似乎没看到结果，又重复发起相同的工具调用，形成死循环。原因：手写 ToolMessage 时 tool_call_id 字段写错、遗漏或多工具调用时 id 错位。修复：让 ToolNode 自动生成 ToolMessage，不要在业务代码中手动构造 ToolMessage；必须手动构造时严格按 tool_calls 顺序一一对应 id。
handle_tool_errors=False 导致生产环境整图崩溃。现象：某个工具偶发超时或第三方 API 返回 500，整图中断，用户看到 500 错误。原因：关闭错误捕获后工具异常直接冒泡到 Pregel 运行时，导致本次 invoke 抛异常终止。修复：生产环境保持 handle_tool_errors=True（默认值），让模型看到错误文本后选择重试或友好回复；对必须成功的关键工具配合节点级 RetryPolicy 做自动重试；监控工具异常率指标。
工具内部修改全局 State 或外部状态。现象：工具函数中直接修改 state 字典的可变字段，但变更在下一轮 superstep 不生效或偶发生效。原因：InjectedState 注入的 state 是只读快照，in-place 修改不经过 Channel reducer，运行时无感知。修复：工具返回字符串或结构化结果（通过 ToolMessage 回传），需要持久化的变更统一由下游节点读取 ToolMessage 后通过返回值写入 State 或通过 Store 手动 put。

生产建议

工具列表按业务场景分组，单次 bind_tools 的工具数控制在 15 个以内；超过 20 个工具时应引入路由节点先做子集选择，或采用检索式工具选择（对工具描述做 embedding 检索，每轮只注入 Top-K 相关工具），避免 prompt 膨胀导致模型选择准确率下降；
所有外部 IO 工具（HTTP、数据库、文件）优先实现 async 版本，ToolNode 在 ainvoke/astream 路径下会自动通过 asyncio.gather 并发执行，显著降低并行工具调用总耗时；
对写入类、支付类、权限类、代码执行类工具强制在 ToolNode 前插入 interrupt 审批节点，审批内容包含工具名、参数摘要、发起用户信息，人类通过/拒绝后 resume 不同值驱动不同分支；
为每个工具编写独立单元测试，重点覆盖三类场景：参数正确时返回预期结果；参数缺失/类型错误时 Pydantic 校验失败；第三方依赖超时/报错时 handle_tool_errors 能正确封装；
工具命名统一使用动宾结构 snake_case（如 get_weather、create_order、search_docs），禁止使用模糊名字（如 do_it、helper、process）；
handle_tool_errors 在开发期可设为 False 快速暴露问题，上线前切回 True 并配合监控告警（工具错误率 >5% 触发告警）；
工具描述、参数描述统一使用中文或团队约定语言，与模型主要交互语言保持一致，避免中英混杂导致模型理解偏差。

C 语言

语言点

深度

新标准

面试整理

第 8 章 工具调用与 ToolNode ​

8.1 Function Calling / Tool Use 协议 ​

8.2 @tool 装饰器与 StructuredTool ​

8.3 ToolNode 详解 ​

8.4 tools_condition 路由函数与 bind_tools ​

8.5 完整 LLM↔Tools 循环（手写 ReAct，不使用 create_react_agent） ​

8.6 工具权限安全与 InjectedState/InjectedStore ​

本章小结 ​

常见错误 ​

生产建议 ​

第 8 章工具调用与 ToolNode

8.1 Function Calling / Tool Use 协议

8.2 @tool 装饰器与 StructuredTool

8.3 ToolNode 详解

8.4 tools_condition 路由函数与 bind_tools

8.5 完整 LLM↔Tools 循环（手写 ReAct，不使用 create_react_agent）

8.6 工具权限安全与 InjectedState/InjectedStore

本章小结

常见错误

生产建议