概述

工具调用不是“模型执行函数”，而是一次消息协议 + 图路由 + 本地执行

很多人第一次看 LangChain Tool 调用时，会误以为：

LLM 直接调用了 Python 函数

实际上不是。

LLM 不会进入你的 Python 进程，也不会真的执行 search_weather()、calculator()、query_database() 这些函数。

真正发生的是：

Python 函数
  |
  v
@tool 包装成 BaseTool / StructuredTool
  |
  v
create_agent() 把工具 schema 绑定给模型
  |
  v
模型返回 AIMessage.tool_calls
  |
  v
LangGraph 条件边路由到 ToolNode
  |
  v
ToolNode 在本地执行 Python 函数
  |
  v
执行结果变成 ToolMessage
  |
  v
ToolMessage 回到 messages，模型继续推理

所以工具调用的本质是三层协作：

层级	负责什么	典型源码
Tool 定义层	把函数变成带 name、description、args schema 的工具对象	langchain_core/tools.py
Agent 编排层	绑定工具 schema、构建图节点、决定是否进入工具节点	langchain/agents/factory.py
Tool 执行层	解析 tool call、执行本地工具、生成 ToolMessage	langgraph/prebuilt/tool_node.py

模型只负责“提出要调用哪个工具以及传什么参数”，真正执行工具的是 LangGraph 图里的 ToolNode。

先看一个最小例子：从函数到 Agent

先写一个最小工具：

from langchain.agents import create_agent
from langchain.tools import tool


@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"{city} is sunny, 28C"


agent = create_agent(
    model="openai:gpt-5.4-mini",
    tools=[get_weather],
    system_prompt="你是一个会查询天气的助手。",
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "北京今天天气怎么样？"}]}
)

从业务代码看，只是把函数放进 tools=[get_weather]。

但源码里，这个函数会经历下面这条链路：

get_weather(city: str) -> str
  |
  | @tool
  v
StructuredTool(
  name="get_weather",
  description="Get current weather for a city.",
  args_schema=...
)
  |
  | create_agent(tools=[...])
  v
ToolNode([get_weather])
  |
  | model.bind_tools(final_tools)
  v
模型知道有一个 get_weather 工具，参数 city 是字符串
  |
  | AIMessage(tool_calls=[...])
  v
ToolNode 执行 get_weather(city="北京")
  |
  v
ToolMessage(content="北京 is sunny, 28C", tool_call_id="...")

这一篇就沿着这条链路往源码里追。

第一层：@tool 到底包装出了什么

@tool 的入口在：

langchain_core/tools.py

核心函数大致从 977 行开始：

def tool(
    *args: Union[str, Callable, Runnable],
    return_direct: bool = False,
    args_schema: Optional[Type[BaseModel]] = None,
    infer_schema: bool = True,
) -> Callable:
    """Make tools out of functions, can be used with or without arguments."""

这几个参数非常关键：

参数	作用
*args	支持 @tool、@tool("name")、把 Runnable 转工具
return_direct	工具执行后是否直接结束 Agent 循环
args_schema	手动指定 Pydantic 参数模型
infer_schema	是否从函数签名自动推断参数 schema

如果我们写：

@tool
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

@tool 会进入内部的 _make_with_name() 和 _make_tool()。

简化后的源码逻辑是：

def _make_tool(dec_func):
    if isinstance(dec_func, Runnable):
        # Runnable 转工具
        schema = runnable.input_schema
        func = invoke_wrapper
        coroutine = ainvoke_wrapper
    elif inspect.iscoroutinefunction(dec_func):
        # async 函数
        coroutine = dec_func
        func = None
        schema = args_schema
    else:
        # 普通同步函数
        coroutine = None
        func = dec_func
        schema = args_schema

    if infer_schema or args_schema is not None:
        return StructuredTool.from_function(
            func,
            coroutine,
            name=tool_name,
            description=description,
            return_direct=return_direct,
            args_schema=schema,
            infer_schema=infer_schema,
        )

    return Tool(...)

这说明 @tool 包装时主要做两件事：

• 识别函数类型：同步函数、异步函数、Runnable 分别处理。
• 选择工具类型：默认走 StructuredTool.from_function()，只有关闭 schema 推断时才退回简单 Tool。

为什么默认是 StructuredTool？

因为现代 function calling / tool calling 模型需要知道参数结构。例如：

{
  "name": "add",
  "description": "Add two numbers.",
  "parameters": {
    "type": "object",
    "properties": {
      "a": {"type": "integer"},
      "b": {"type": "integer"}
    },
    "required": ["a", "b"]
  }
}

没有结构化参数 schema，模型就只能猜字符串，很难稳定地产生可执行参数。

@tool 的核心不是装饰器语法，而是把 Python 函数转换成一个有名称、描述和参数 schema 的 BaseTool。

第二层：StructuredTool.from_function() 如何推断 name、description 和 args_schema

继续看：

langchain_core/tools.py
StructuredTool.from_function()
约 901-974 行

关键源码逻辑可以简化成：

if func is not None:
    source_function = func
elif coroutine is not None:
    source_function = coroutine
else:
    raise ValueError("Function and/or coroutine must be provided")

name = name or source_function.__name__
description_ = description or source_function.__doc__

if description_ is None and args_schema:
    description_ = args_schema.__doc__

if description_ is None:
    raise ValueError(
        "Function must have a docstring if description not provided."
    )

if _args_schema is None and infer_schema:
    _args_schema = create_schema_from_function(name, source_function)

return cls(
    name=name,
    func=func,
    coroutine=coroutine,
    args_schema=_args_schema,
    description=description_,
    return_direct=return_direct,
    **kwargs,
)

这里有三个容易踩坑的点。

第一，工具名默认来自函数名。

@tool
def get_weather(city: str) -> str:
    """Get current weather."""
    ...

工具名就是：

get_weather

如果你想改名，需要写：

@tool("weather_search")
def get_weather(city: str) -> str:
    """Get current weather."""
    ...

第二，工具描述默认来自 docstring。

模型选择工具时，最重要的文本信号就是工具描述。

下面这种写法不推荐：

@tool
def search(query: str) -> str:
    return "..."

因为没有 docstring 时，源码会直接抛错：

Function must have a docstring if description not provided.

更好的写法：

@tool
def search(query: str) -> str:
    """Search public web pages by a natural language query."""
    return "..."

第三，参数 schema 默认从函数签名推断。

例如：

@tool
def query_order(order_id: str, include_items: bool = False) -> str:
    """Query order detail by order id."""
    ...

会推断出类似：

order_id: string, required
include_items: boolean, optional, default=False

所以工具函数签名不要写得太随意。下面这种写法会让 schema 变得很差：

@tool
def query_order(payload: dict) -> str:
    """Query order."""
    ...

模型只能知道有一个 payload，却不知道里面应该有哪些字段。

更推荐：

from pydantic import BaseModel, Field


class QueryOrderArgs(BaseModel):
    order_id: str = Field(description="The order id, for example O202607070001")
    include_items: bool = Field(default=False, description="Whether to include item list")


@tool(args_schema=QueryOrderArgs)
def query_order(order_id: str, include_items: bool = False) -> str:
    """Query order detail by order id."""
    ...

StructuredTool.from_function() 把函数名、docstring、类型注解转成模型可理解的工具协议。

第三层：create_agent() 如何处理 tools=[...]

@tool 只是把函数包装成工具对象。

真正让 Agent 使用这些工具，是 create_agent() 做的。

源码入口：

langchain/agents/factory.py
create_agent()

在工具准备阶段，关键逻辑约在 1028-1056 行：

# Setup tools
tool_node: ToolNode | None = None

# Extract built-in provider tools (dict format) and regular tools (BaseTool/callables)
built_in_tools = [t for t in tools if isinstance(t, dict)]
regular_tools = [t for t in tools if not isinstance(t, dict)]

# Tools that require client-side execution (must be in ToolNode)
available_tools = middleware_tools + regular_tools

# Create ToolNode if we have client-side tools OR if middleware defines wrap_tool_call
tool_node = (
    ToolNode(
        tools=available_tools,
        wrap_tool_call=wrap_tool_call_wrapper,
        awrap_tool_call=awrap_tool_call_wrapper,
    )
    if available_tools or wrap_tool_call_wrapper or awrap_tool_call_wrapper
    else None
)

if tool_node:
    default_tools = list(tool_node.tools_by_name.values()) + built_in_tools
else:
    default_tools = list(built_in_tools)

这里可以看到 LangChain 把工具分成两类：

工具类型	例子	谁执行
客户端工具	Python 函数、@tool、BaseTool	本地 ToolNode 执行
Provider 内置工具	某些模型供应商支持的内置搜索、代码解释器等 dict 工具	模型供应商侧执行

这也是为什么源码里有：

built_in_tools = [t for t in tools if isinstance(t, dict)]
regular_tools = [t for t in tools if not isinstance(t, dict)]

普通 Python 工具必须进入 ToolNode。

原因很简单：

模型不会执行你的 Python 函数
LangGraph 才能在本地执行你的 Python 函数

所以只要有客户端工具，就会创建：

ToolNode(tools=available_tools, ...)

并从 tool_node.tools_by_name.values() 取出转换后的 BaseTool，作为默认可用工具。

create_agent() 会把 Python 工具交给 ToolNode 管理，把工具 schema 交给模型理解。

第四层：model.bind_tools() 把工具 schema 绑定给模型

工具在本地准备好以后，还要告诉模型：

你可以调用哪些工具？
每个工具叫什么？
每个工具需要什么参数？

这个动作在 factory.py 约 1334-1378 行。

无结构化输出时，核心逻辑是：

if final_tools:
    return (
        request.model.bind_tools(
            final_tools,
            tool_choice=request.tool_choice,
            **request.model_settings,
        ),
        None,
    )

return request.model.bind(**request.model_settings), None

也就是说：

BaseTool / StructuredTool
  |
  v
model.bind_tools(final_tools)
  |
  v
供应商模型适配层把工具转成 OpenAI / Anthropic / Google 等各自的 tool schema

这一层有个关键认知：

bind_tools() 不是执行工具，而是把工具说明绑定到模型请求上。

模型收到工具 schema 后，可能会返回普通自然语言，也可能返回工具调用请求。

普通自然语言类似：

AIMessage(content="北京今天晴。")

工具调用请求类似：

AIMessage(
    content="",
    tool_calls=[
        {
            "name": "get_weather",
            "args": {"city": "北京"},
            "id": "call_xxx",
            "type": "tool_call",
        }
    ],
)

这就是下一步路由的依据。

bind_tools() 只负责让模型“知道工具”，不负责执行工具。

第五层：模型返回 AIMessage.tool_calls 后，图如何决定进入 tools 节点

create_agent() 底层会构建 LangGraph 图。

在有 ToolNode 的情况下，源码会添加两条关键条件边。

位置：

langchain/agents/factory.py
约 1594-1641 行

简化后：

graph.add_conditional_edges(
    "tools",
    RunnableCallable(
        _make_tools_to_model_edge(...),
        trace=False,
    ),
    tools_to_model_destinations,
)

graph.add_conditional_edges(
    loop_exit_node,
    RunnableCallable(
        _make_model_to_tools_edge(...),
        trace=False,
    ),
    model_to_tools_destinations,
)

这两条边分别负责：

条件边	方向	作用
_make_model_to_tools_edge()	model -> tools / end / model	看模型是否产生未执行的 tool calls
_make_tools_to_model_edge()	tools -> model / end	工具执行后是否回到模型

画成图：

START
  |
  v
model
  |
  | AIMessage.tool_calls 为空
  v
END

model
  |
  | AIMessage.tool_calls 非空
  v
tools
  |
  | ToolMessage 写回 messages
  v
model

这就是 ReAct 类 Agent 最核心的循环：

模型思考 -> 请求工具 -> 工具执行 -> 模型观察结果 -> 继续思考

Agent 是否调用工具，不是 if else 写在业务代码里，而是由 LangGraph 条件边根据 AIMessage.tool_calls 路由。

第六层：_make_model_to_tools_edge() 如何找出待执行工具

继续看：

langchain/agents/factory.py
_make_model_to_tools_edge()
约 1815-1865 行

关键逻辑可以简化成：

last_ai_message, tool_messages = _fetch_last_ai_and_tool_messages(state["messages"])

if last_ai_message is None:
    return end_destination

tool_message_ids = [m.tool_call_id for m in tool_messages]

if len(last_ai_message.tool_calls) == 0:
    return end_destination

pending_tool_calls = [
    c
    for c in last_ai_message.tool_calls
    if c["id"] not in tool_message_ids
    and c["name"] not in structured_output_tools
]

if pending_tool_calls:
    return [Send("tools", [tool_call]) for tool_call in pending_tool_calls]

if "structured_response" in state:
    return end_destination

return model_destination

这里的核心是 pending_tool_calls。

什么叫 pending？

AIMessage 里有 tool_call
但 messages 里还没有对应 tool_call_id 的 ToolMessage

也就是说：

AIMessage.tool_calls = [
    {"id": "call_1", "name": "get_weather", "args": {"city": "北京"}}
]

ToolMessage.tool_call_id = "call_1"

只有当 ToolMessage.tool_call_id 和 AIMessage.tool_calls[*].id 对上，框架才认为这个工具调用已经处理完。

如果没对上，就会构造：

Send("tools", [tool_call])

把这个工具调用发送给 "tools" 节点。

这也是 Debug 工具调用时非常重要的一条规则：

每一个 AIMessage.tool_calls 里的 id，都必须最终对应一个 ToolMessage.tool_call_id。

否则模型会一直处在“我请求了工具，但没有收到工具结果”的状态。

_make_model_to_tools_edge() 不是简单判断有没有 tool_calls，而是判断有没有尚未被 ToolMessage 响应的 pending tool calls。

第七层：ToolNode 初始化时如何管理工具

ToolNode 在：

langgraph/prebuilt/tool_node.py

初始化逻辑约在 758-791 行：

def __init__(
    self,
    tools: Sequence[BaseTool | Callable],
    name: str = "tools",
    tags: list[str] | None = None,
    handle_tool_errors: bool | str | Callable[..., str] | tuple[type[Exception], ...] = ...,
    messages_key: str = "messages",
    wrap_tool_call: ToolCallWrapper | None = None,
    awrap_tool_call: AsyncToolCallWrapper | None = None,
) -> None:
    super().__init__(self._func, self._afunc, name=name, tags=tags, trace=False)
    self._tools_by_name: dict[str, BaseTool] = {}
    self._injected_args: dict[str, _InjectedArgs] = {}
    self._handle_tool_errors = handle_tool_errors
    self._messages_key = messages_key
    self._wrap_tool_call = wrap_tool_call
    self._awrap_tool_call = awrap_tool_call

    for tool in tools:
        if not isinstance(tool, BaseTool):
            tool_ = create_tool(tool)
        else:
            tool_ = tool
        self._tools_by_name[tool_.name] = tool_
        self._injected_args[tool_.name] = _get_all_injected_args(tool_)

这段源码说明了几件事。

第一，ToolNode 内部按工具名建索引。

self._tools_by_name[tool_.name] = tool_

所以模型返回：

{"name": "get_weather", "args": {"city": "北京"}}

ToolNode 就能通过：

self.tools_by_name.get("get_weather")

找到真正要执行的 BaseTool。

第二，普通 callable 会再次被转换成工具。

if not isinstance(tool, BaseTool):
    tool_ = create_tool(tool)

这意味着你传入 create_agent(tools=[普通函数]) 时，框架仍然会尝试把它变成工具对象。

但实际项目里我仍然建议显式使用 @tool：

@tool
def query_order(order_id: str) -> str:
    """Query order detail."""
    ...

这样 name、description、schema 更明确，也更容易 debug。

第三，注入参数只在 ToolNode 执行阶段处理。

self._injected_args[tool_.name] = _get_all_injected_args(tool_)

例如 InjectedState、InjectedStore、ToolRuntime 这类参数，不应该暴露给模型填写，而是在执行工具前由 ToolNode 注入。

ToolNode 是工具执行注册表，它用工具名把模型的 tool call 映射到本地 BaseTool。

第八层：ToolNode 如何从输入中解析 tool calls

ToolNode 的输入不止一种形态。

它既可以接收完整 state：

{"messages": [HumanMessage(...), AIMessage(tool_calls=[...])]}

也可以接收从条件边 Send("tools", [tool_call]) 传来的 tool call 列表。

源码位置：

langgraph/prebuilt/tool_node.py
_parse_input()
约 1224-1266 行

简化逻辑：

if isinstance(input, list):
    if isinstance(input[-1], dict) and input[-1].get("type") == "tool_call":
        input_type = "tool_calls"
        tool_calls = input
        return tool_calls, input_type
    input_type = "list"
    messages = input
elif isinstance(input, dict) and input.get("__type") == "tool_call_with_context":
    input_type = "tool_calls"
    return [input_with_ctx["tool_call"]], input_type
elif isinstance(input, dict) and (messages := input.get(self._messages_key, [])):
    input_type = "dict"
elif messages := getattr(input, self._messages_key, []):
    input_type = "dict"
else:
    raise ValueError("No message found in input")

latest_ai_message = next(
    m for m in reversed(messages) if isinstance(m, AIMessage)
)

tool_calls = list(latest_ai_message.tool_calls)
return tool_calls, input_type

这段代码说明：

• 如果输入本身就是 tool call 列表，直接执行。
• 如果输入是 state，就从 messages 里倒序找最近的 AIMessage。
• 找到 AIMessage 后，取它的 tool_calls。
• 如果没有消息或没有 AIMessage，直接抛错。

所以当你手写 LangGraph 工具节点时，最常见的错误是：

No message found in input
No AIMessage found in input

本质原因通常是 state schema 不符合 ToolNode 预期。

默认情况下，ToolNode 期待消息字段叫：

messages

如果你的状态字段叫 chat_history，就要传：

ToolNode(tools=[...], messages_key="chat_history")

ToolNode 执行前第一件事，就是从输入 state 或 Send payload 中取出标准化的 ToolCall 列表。

第九层：同步和异步工具如何并发执行

解析出 tool calls 后，ToolNode 会进入 _func() 或 _afunc()。

源码位置：

langgraph/prebuilt/tool_node.py
约 793-860 行

同步版本简化为：

tool_calls, input_type = self._parse_input(input)
config_list = get_config_list(config, len(tool_calls))

tool_runtimes = []
for call, cfg in zip(tool_calls, config_list, strict=False):
    state = self._extract_state(input, cfg)
    tool_runtime = ToolRuntime(
        state=state,
        tool_call_id=call["id"],
        config=cfg,
        context=runtime.context,
        store=runtime.store,
        stream_writer=runtime.stream_writer,
        tools=list(self.tools_by_name.values()),
        execution_info=runtime.execution_info,
        server_info=runtime.server_info,
    )
    tool_runtimes.append(tool_runtime)

with get_executor_for_config(config) as executor:
    outputs = list(
        executor.map(self._run_one, tool_calls, input_types, tool_runtimes)
    )

return self._combine_tool_outputs(outputs, input_type)

异步版本则使用：

outputs = await asyncio.gather(*coros)

这说明如果模型一次返回多个 tool call：

AIMessage(
    tool_calls=[
        {"id": "call_1", "name": "get_weather", "args": {"city": "北京"}},
        {"id": "call_2", "name": "get_weather", "args": {"city": "上海"}},
    ]
)

ToolNode 不是简单串行执行，而是会按配置并发执行。

这里还创建了一个非常重要的对象：

ToolRuntime(...)

它把下面这些执行上下文封装起来：

字段	作用
state	当前图状态
tool_call_id	当前工具调用 id
config	Runnable 配置
context	运行上下文
store	持久化存储
stream_writer	流式输出写入器
tools	当前可用工具列表

这就是为什么工具函数可以使用 InjectedState、InjectedStore 或 ToolRuntime 获取运行时信息。

ToolNode 会把多个 tool calls 标准化成多个执行任务，并为每个任务构造独立的 ToolRuntime。

第十层：单个工具调用如何真正执行

单个工具调用的执行入口是：

langgraph/prebuilt/tool_node.py
_run_one()
约 1014-1067 行

核心逻辑：

tool = self.tools_by_name.get(call["name"])

tool_request = ToolCallRequest(
    tool_call=call,
    tool=tool,
    state=tool_runtime.state,
    runtime=tool_runtime,
)

if self._wrap_tool_call is None:
    return self._execute_tool_sync(tool_request, input_type, config)

def execute(req: ToolCallRequest) -> ToolMessage | Command:
    return self._execute_tool_sync(req, input_type, config)

return self._wrap_tool_call(tool_request, execute)

这段代码有两个重点。

第一，工具是按 name 查找的。

如果模型返回：

{"name": "weather", "args": {"city": "北京"}}

但你注册的工具叫：

get_weather

就找不到工具。

所以不要随意改工具名，也不要给模型模糊的工具描述。

第二，wrap_tool_call 是包裹单次工具执行的中间件。

如果你在 Middleware 里定义了 wrap_tool_call，它不是包住整个 Agent，而是包住这里的：

execute(req)

这意味着它适合做：

• 工具调用日志
• 参数审计
• 重试
• 缓存
• 超时保护
• 人工审批
• 特定工具降级

例如伪代码：

def wrap_tool_call(request, handler):
    print("tool start:", request.tool_call["name"])
    try:
        return handler(request)
    finally:
        print("tool end:", request.tool_call["id"])

真正执行工具的是 _execute_tool_sync()。

源码位置：

langgraph/prebuilt/tool_node.py
约 927-1012 行

关键逻辑：

call = request.tool_call
tool = request.tool

if tool is None:
    if invalid_tool_message := self._validate_tool_call(call):
        return invalid_tool_message
    raise TypeError(...)

injected_call = self._inject_tool_args(call, request.runtime, tool)
call_args = {**injected_call, "type": "tool_call"}

try:
    response = tool.invoke(call_args, config)
    return self._normalize_tool_response(response, request.tool_call, input_type)
except Exception as e:
    if not self._handle_tool_errors or not isinstance(e, handled_types):
        raise
    content = _handle_tool_error(e, flag=self._handle_tool_errors)
    return ToolMessage(
        content=content,
        name=call["name"],
        tool_call_id=call["id"],
        status="error",
    )

这才是工具执行的关键点：

response = tool.invoke(call_args, config)

前面的 @tool、bind_tools()、AIMessage.tool_calls 都是在准备协议和路由。

真正调用 Python 函数，就是这里。

ToolNode._execute_tool_sync() 才是工具从“模型请求”变成“本地函数执行”的地方。

第十一层：工具参数是怎么被注入和校验的

注意这一行：

injected_call = self._inject_tool_args(call, request.runtime, tool)
call_args = {**injected_call, "type": "tool_call"}

原始 tool call 可能是：

{
    "name": "query_order",
    "args": {"order_id": "O202607070001"},
    "id": "call_1",
    "type": "tool_call",
}

如果工具函数声明了注入参数：

from typing import Annotated
from langchain.tools import InjectedState, tool


@tool
def query_order(
    order_id: str,
    state: Annotated[dict, InjectedState],
) -> str:
    """Query order detail by order id."""
    user_id = state["user_id"]
    ...

模型只需要提供：

{"order_id": "O202607070001"}

state 不应该由模型提供。

ToolNode 在执行阶段根据 _injected_args 自动把 state 注入进去。

这解决了一个安全问题：

模型可以决定调用什么工具、传什么业务参数
但不应该伪造系统 state、store、runtime

如果参数校验失败，源码会捕获 ValidationError，并包装成 ToolInvocationError：

except ValidationError as exc:
    injected = self._injected_args.get(call["name"])
    filtered_errors = _filter_validation_errors(exc, injected)
    raise ToolInvocationError(
        call["name"], exc, call["args"], filtered_errors
    ) from exc

这里还会过滤注入参数相关错误，避免把内部注入字段暴露成模型应该填写的字段。

模型只负责普通业务参数，状态、存储、运行时上下文由 ToolNode 在本地注入。

第十二层：工具返回值为什么必须变成 ToolMessage

工具执行完成后，ToolNode 要把结果放回消息历史。

因为下一轮模型要“观察”工具结果。

模型不是读取 Python 返回值，而是读取消息列表里的：

ToolMessage(
    content="北京 is sunny, 28C",
    name="get_weather",
    tool_call_id="call_1",
)

其中最重要的是：

tool_call_id

它必须对应前面：

AIMessage.tool_calls[0]["id"]

否则模型协议不闭合。

ToolNode 对工具返回值会做规范化。

源码位置：

langgraph/prebuilt/tool_node.py
_normalize_tool_response()
约 1437-1453 行

简化逻辑：

if isinstance(response, Command):
    return self._validate_tool_command(response, tool_call, input_type)

if isinstance(response, ToolMessage):
    response.content = msg_content_output(response.content)
    return response

if isinstance(response, list):
    if all(isinstance(r, (Command, ToolMessage)) for r in response):
        return self._validate_tool_command_list(response, tool_call, input_type)
    raise TypeError(...)

raise TypeError(...)

你可能会问：

普通工具返回字符串怎么办？

对于通过 @tool 创建的 BaseTool，tool.invoke() 本身会把普通返回值包装成工具协议需要的结果。ToolNode 收到的通常已经是 ToolMessage 或可被规范化的结构。

如果你手写底层工具或返回复杂结构，就要特别注意返回值类型。

错误示例：

@tool
def broken_tool(x: int) -> dict:
    """Return raw dict."""
    return {"value": x}

在一些场景下，原始 dict 可能不是你想要的消息内容。更稳妥的方式是返回字符串或明确序列化：

import json


@tool
def stable_tool(x: int) -> str:
    """Return serialized result."""
    return json.dumps({"value": x}, ensure_ascii=False)

工具最终不是把 Python 对象直接塞给模型，而是要形成带 tool_call_id 的 ToolMessage。

第十三层：多个工具结果如何合并回 state

多个工具执行完成后，ToolNode 会调用：

_combine_tool_outputs()
约 862-887 行

核心逻辑：

if not any(isinstance(output, Command) for output in flat_outputs):
    return (
        flat_outputs
        if input_type == "list"
        else {self._messages_key: flat_outputs}
    )

如果输入是普通 state dict，返回结果就是：

{"messages": [ToolMessage(...), ToolMessage(...)]}

然后 LangGraph 的 message reducer 会把这些 ToolMessage 合并回历史消息。

合并后的消息大概是：

HumanMessage: 北京今天天气怎么样？
AIMessage: tool_calls=[{"id": "call_1", "name": "get_weather", ...}]
ToolMessage: content="北京 is sunny, 28C", tool_call_id="call_1"

然后图从 "tools" 条件边回到 "model"。

_make_tools_to_model_edge() 的默认逻辑约在 1895-1928 行：

client_side_tool_calls = [
    c for c in last_ai_message.tool_calls if c["name"] in tool_node.tools_by_name
]

if client_side_tool_calls and all(
    tool_node.tools_by_name[c["name"]].return_direct
    for c in client_side_tool_calls
):
    return end_destination

if any(t.name in structured_output_tools for t in tool_messages):
    return end_destination

return model_destination

默认情况下：

工具执行完 -> 回到模型 -> 模型基于 ToolMessage 生成最终回答

只有在这些情况下可能直接结束：

• 工具设置了 return_direct=True。
• 执行的是结构化输出相关工具。
• Middleware 或图逻辑显式跳转。

ToolNode 返回的是 state update，LangGraph 把 ToolMessage 合并回 messages，再让模型读取工具结果。

全链路时序图：一个函数调用的完整旅程

把前面的源码串起来，就是下面这张图：

对应源码链路：

@tool
  -> StructuredTool.from_function()
  -> create_agent()
  -> ToolNode(tools=...)
  -> model.bind_tools(final_tools)
  -> AIMessage.tool_calls
  -> _make_model_to_tools_edge()
  -> Send("tools", [tool_call])
  -> ToolNode._parse_input()
  -> ToolNode._run_one()
  -> ToolNode._execute_tool_sync()
  -> tool.invoke(call_args, config)
  -> ToolMessage
  -> _combine_tool_outputs()
  -> model

工具调用全链路可以拆成“定义工具、绑定工具、模型请求、图路由、本地执行、消息回灌”六个动作。

Debug 视角：工具调用最常见的 10 类问题

源码看懂以后，排查问题会快很多。

问题一：@tool 报函数必须有 docstring

典型报错：

Function must have a docstring if description not provided.

原因：

description_ = description or source_function.__doc__

修复：

@tool
def search(query: str) -> str:
    """Search documents by query."""
    ...

不要写空泛描述：

"""Search."""

应该写清楚工具适用场景：

"""Search internal product documents by a natural language query."""

问题二：模型总是不调用工具

优先检查四件事：

• 工具是否真的传入了 create_agent(tools=[...])。
• 工具描述是否足够清楚。
• 当前模型是否支持 tool calling。
• model.bind_tools(final_tools) 是否被走到。

从源码看，只有 final_tools 非空时才会执行：

request.model.bind_tools(final_tools, ...)

如果工具没有进入 final_tools，模型根本不知道工具存在。

问题三：模型调用了不存在的工具

典型表现：

Tool xxx is not registered with ToolNode

或返回 error ToolMessage。

原因在：

requested_tool = call["name"]
if requested_tool not in self.tools_by_name:
    ...

排查：

print(tool.name)

确认模型返回的：

call["name"]

是否和 ToolNode.tools_by_name 里的 key 完全一致。

问题四：参数校验失败

典型原因：

• 模型给了错误类型。
• 函数类型注解太模糊。
• args_schema 和函数签名不一致。
• 必填字段没有描述清楚。

建议用 Pydantic schema 明确字段：

class SearchArgs(BaseModel):
    query: str = Field(description="Natural language search query")
    top_k: int = Field(default=5, ge=1, le=20, description="Number of results")

问题五：工具执行异常后 Agent 直接失败

看 ToolNode 的 handle_tool_errors。

源码里：

if not self._handle_tool_errors or not isinstance(e, handled_types):
    raise

return ToolMessage(..., status="error")

如果开启错误处理，异常会变成 error ToolMessage，模型还能继续解释或重试。

如果关闭错误处理，异常会直接抛出。

问题六：工具执行了，但模型像没收到结果

检查：

AIMessage.tool_calls[*].id
ToolMessage.tool_call_id

这两个必须对上。

源码里就是靠它判断 pending tool calls：

if c["id"] not in tool_message_ids:
    pending_tool_calls.append(c)

如果你手写 ToolMessage，一定要带正确的 tool_call_id。

问题七：手写 LangGraph 时 ToolNode 找不到 messages

默认字段是：

messages_key="messages"

如果你的 state 是：

class State(TypedDict):
    chat_history: list

需要：

ToolNode(tools=[...], messages_key="chat_history")

否则 _parse_input() 会抛：

No message found in input

问题八：多个工具调用顺序和预期不一致

ToolNode 支持并发执行多个 tool calls。

同步路径用：

executor.map(...)

异步路径用：

asyncio.gather(...)

所以工具之间不要依赖隐式执行顺序。

如果工具 B 必须依赖工具 A 的结果，应该让模型分两轮调用，或在一个工具内部完成这个事务。

问题九：工具返回复杂对象导致模型读不懂

建议工具返回字符串或标准内容块。

对于业务对象：

return {
    "order_id": order_id,
    "status": "PAID",
}

更稳妥：

return json.dumps(
    {"order_id": order_id, "status": "PAID"},
    ensure_ascii=False,
)

尤其是在跨供应商模型时，字符串化结果更容易保持一致。

问题十：把内部状态暴露给模型填写

不要让模型填写 user_id、tenant_id、auth_token 这类字段。

错误设计：

@tool
def query_order(user_id: str, order_id: str) -> str:
    """Query user order."""
    ...

更好的设计：

@tool
def query_order(
    order_id: str,
    state: Annotated[dict, InjectedState],
) -> str:
    """Query current user's order by order id."""
    user_id = state["user_id"]
    ...

业务身份从 state 注入，而不是让模型生成。

Debug 工具调用时，不要只盯着函数本身，要同时检查工具 schema、模型返回的 tool call、图路由和 ToolMessage 回灌。

生产建议：如何写更稳定的 LangChain 工具

建议一：工具名用动词 + 业务对象

推荐：

search_documents
query_order
calculate_refund
send_notification

不推荐：

run
handler
tool1
search
api

工具名越具体，模型越不容易选错。

建议二：docstring 写“什么时候用”，不是只写“做什么”

一般描述：

"""Search documents."""

更好的描述：

"""Search internal product documents when the user asks about product usage, limits, pricing, or troubleshooting."""

模型选择工具时，最需要的是适用边界。

建议三：用 Pydantic Field 描述参数

class RefundArgs(BaseModel):
    order_id: str = Field(description="Order id, such as O202607070001")
    reason: str = Field(description="Refund reason provided by the user")
    amount: float = Field(gt=0, description="Refund amount in CNY")

比裸函数签名更稳定：

def refund(order_id: str, reason: str, amount: float) -> str:
    ...

建议四：工具内部做二次校验

不要因为模型传了 schema 合法参数，就默认业务安全。

@tool(args_schema=RefundArgs)
def create_refund(order_id: str, reason: str, amount: float) -> str:
    """Create a refund request after validating order status."""
    order = load_order(order_id)
    if order.status != "PAID":
        return "Cannot refund: order is not paid."
    if amount > order.remaining_refundable_amount:
        return "Cannot refund: amount exceeds refundable balance."
    ...

LLM 负责推理，不负责最终业务安全。

建议五：有副作用的工具要配合人工确认

这些工具要谨慎：

• 删除数据
• 发送消息
• 创建订单
• 发起退款
• 修改权限
• 调用外部支付

可以用 Middleware 的 wrap_tool_call 或 Human-in-the-loop 机制，在真正执行前拦截审批。

建议六：工具结果返回“可继续推理的信息”

不要只返回：

success

更好：

Refund request R202607070001 has been created for order O202607070001. Status: pending approval.

模型需要根据工具结果继续回答用户。

一个好工具不仅要能执行，还要让模型知道什么时候用、怎么传参、执行后如何继续推理。

手写 ToolNode 时的最小图结构

虽然 create_agent() 已经帮我们封装了大部分逻辑，但理解手写图也很有价值。

一个最小工具调用图大概是：

from typing import Annotated, TypedDict

from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition


class State(TypedDict):
    messages: Annotated[list, add_messages]


@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"{city} is sunny, 28C"


model = init_chat_model("openai:gpt-5.4-mini", temperature=0)
model_with_tools = model.bind_tools([get_weather])


def call_model(state: State):
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response]}


graph_builder = StateGraph(State)
graph_builder.add_node("model", call_model)
graph_builder.add_node("tools", ToolNode([get_weather]))

graph_builder.add_edge(START, "model")
graph_builder.add_conditional_edges("model", tools_condition)
graph_builder.add_edge("tools", "model")

graph = graph_builder.compile()

这段代码对应的逻辑就是：

model -> 如果有 tool_calls 就去 tools，否则 END
tools -> 执行工具并写入 ToolMessage -> 回 model

create_agent() 做的事情比这个复杂得多：

• 支持 Middleware。
• 支持结构化输出。
• 支持 provider 内置工具。
• 支持 checkpointer。
• 支持动态模型配置。
• 支持 return_direct。
• 支持并发工具调用。

但最小骨架就是这一张图。

create_agent() 是生产级封装，手写 StateGraph + ToolNode + tools_condition 是理解工具调用机制的最好方式。

总结

如果以后你要 debug LangChain 工具调用，建议按下面 6 个断点看：

断点	看什么	源码位置
@tool 后	工具名、描述、参数 schema 是否正确	langchain_core/tools.py::tool()
StructuredTool.from_function()	docstring、函数签名、Pydantic schema 是否被正确推断	langchain_core/tools.py
create_agent() 工具准备	工具是否进入 ToolNode 和 default_tools	langchain/agents/factory.py
model.bind_tools()	模型请求是否真的绑定了工具 schema	langchain/agents/factory.py
_make_model_to_tools_edge()	是否产生 pending tool calls	langchain/agents/factory.py
ToolNode._execute_tool_sync()	工具是否被找到、参数是否校验通过、结果是否变成 ToolMessage	langgraph/prebuilt/tool_node.py

最后再用一张最短链路图收尾：

Python function
  -> @tool
  -> StructuredTool
  -> create_agent(tools=[...])
  -> ToolNode.tools_by_name
  -> model.bind_tools()
  -> AIMessage.tool_calls
  -> Send("tools", [tool_call])
  -> tool.invoke()
  -> ToolMessage
  -> model final answer

真正掌握这条链路以后，再遇到工具调用问题，就不需要靠猜。

你只要问四个问题：

• 模型知道这个工具吗？
• 模型生成的 tool call 对吗？
• ToolNode 找得到这个工具吗？
• 工具结果有没有变成匹配 tool_call_id 的 ToolMessage？

这四个问题能覆盖绝大多数 LangChain Tool 调用故障。

Tool 调用的稳定性，来自清晰的工具 schema、正确的图路由、可靠的本地执行，以及闭合的消息协议。