就是 ToolNode。它继承自 RunnableCallable（tool_node.py:622），所以能当图节点用。核心入口是 _func（同步）/_afunc（异步）。整体职责：输入含 tool_calls 的消息/状态 → 执行工具 → 输出 ToolMessage 列表。

# tool_node.py:622（类头，裁剪 docstring）
class ToolNode(RunnableCallable):
    """A node that runs the tools called in the last AIMessage."""
    # 关键内部状态（在 __init__ 里建，见 L02）：
    #   self._tools_by_name : dict[str, BaseTool]   工具名 → 工具
    #   self._handle_tool_errors : 错误处理策略
    #   self._injected_args : 每个工具需要注入哪些特殊参数

tool_node.py:743，构造时做两件准备：

# tool_node.py:743
def __init__(self, tools, *, name="tools", tags=None,
             handle_tool_errors=_default_handle_tool_errors,
             messages_key="messages", wrap_tool_call=None, awrap_tool_call=None):
    super().__init__(self._func, self._afunc, name=name, tags=tags, trace=False)
    self._tools_by_name = {}
    self._injected_args = {}
    self._handle_tool_errors = handle_tool_errors        # 错误策略（L06）
    self._messages_key = messages_key
    self._wrap_tool_call = wrap_tool_call                # 拦截器（重试/缓存）
    for tool in tools:
        if not isinstance(tool, BaseTool):
            tool_ = create_tool(tool)                    # 普通函数 → 包成 BaseTool
        else:
            tool_ = tool
        self._tools_by_name[tool_.name] = tool_          # ★ 建 名字→工具 索引
        self._injected_args[tool_.name] = _get_all_injected_args(tool_)  # 预扫注入参数

节点被调用时，输入可能是消息列表、状态 dict、或 Send 传来的特殊结构。tool_node.py:1224 统一解析：

# tool_node.py:1224
def _parse_input(self, input) -> tuple[list[ToolCall], Literal["list","dict","tool_calls"]]:
    if isinstance(input, list):
        if isinstance(input[-1], dict) and input[-1].get("type") == "tool_call":
            return cast(list[ToolCall], input), "tool_calls"   # 直接给了 tool_call 列表
        input_type = "list"; messages = input
    elif isinstance(input, dict) and input.get("__type") == "tool_call_with_context":
        input_with_ctx = cast(ToolCallWithContext, input)       # ★ D53 v2 的 Send 载荷
        return [input_with_ctx["tool_call"]], "tool_calls"
    elif isinstance(input, dict) and (messages := input.get(self._messages_key, [])):
        input_type = "dict"
    elif messages := getattr(input, self._messages_key, []):
        input_type = "dict"                                     # dataclass 状态
    else:
        raise ValueError("No message found in input")
    try:
        latest_ai_message = next(m for m in reversed(messages) if isinstance(m, AIMessage))
    except StopIteration:
        raise ValueError("No AIMessage found in input")
    return list(latest_ai_message.tool_calls), input_type

主流程 tool_node.py:793：

# tool_node.py:793
def _func(self, input, config, runtime):
    tool_calls, input_type = self._parse_input(input)          # L03
    config_list = get_config_list(config, len(tool_calls))
    tool_runtimes = []
    for call, cfg in zip(tool_calls, config_list):
        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()), ...)
        tool_runtimes.append(tool_runtime)
    input_types = [input_type] * len(tool_calls)
    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)     # L07 合并

单个工具的执行 + 兜错在 tool_node.py:922：

# tool_node.py:922（裁剪）
def _execute_tool_sync(self, request, input_type, config):
    call = request.tool_call; tool = request.tool
    if tool is None:                                            # 模型点了不存在的工具
        if invalid_tool_message := self._validate_tool_call(call):
            return invalid_tool_message                         # 返回"没这个工具"的 ToolMessage
    injected_call = self._inject_tool_args(call, request.runtime, tool)   # 注入 state/store
    call_args = {**injected_call, "type": "tool_call"}
    try:
        try:
            response = tool.invoke(call_args, config)           # ★ 真正执行工具
        except ValidationError as exc:                          # 参数校验失败
            filtered_errors = _filter_validation_errors(exc, self._injected_args.get(call["name"]))
            raise ToolInvocationError(call["name"], exc, call["args"], filtered_errors) from exc
        return self._normalize_tool_response(response, request.tool_call, input_type)
    except GraphBubbleUp:                                       # interrupt() 等特殊信号
        raise                                                   # ← 必须原样抛，不能吞
    except Exception as e:
        handled_types = ...                                     # 根据配置定"哪些异常要接住"
        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")      # ★ 错误也变消息

到底怎么把异常变文本？看 _handle_tool_error（tool_node.py:394）和默认策略（tool_node.py:383）：

# tool_node.py:383  默认策略
def _default_handle_tool_errors(e: Exception) -> str:
    if isinstance(e, ToolInvocationError):    # 只兜"参数校验类"错误
        return e.message
    raise e                                   # 其它错误照样抛（不隐藏真 bug）

# tool_node.py:394  按配置类型生成错误文本
def _handle_tool_error(e, *, flag) -> str:
    if isinstance(flag, (bool, tuple)) or (isinstance(flag, type) and issubclass(flag, Exception)):
        content = TOOL_CALL_ERROR_TEMPLATE.format(error=repr(e))   # 用默认模板
    elif isinstance(flag, str):
        content = flag                        # 用你给的固定字符串
    elif callable(flag):
        content = flag(e)                     # 用你给的函数生成
    else:
        raise ValueError("Got unexpected type of `handle_tool_error`...")
    return content
# TOOL_CALL_ERROR_TEMPLATE = "Error: {error}\n Please fix your mistakes."  (:111)

并行跑完，N 个结果要合并成节点返回值。tool_node.py:862（裁剪）：

# tool_node.py:862
def _combine_tool_outputs(self, outputs, input_type):
    # 有工具返回了 list（多条消息）→ 先摊平
    if any(isinstance(o, list) for o in outputs):
        flat_outputs = []
        for o in outputs:
            flat_outputs.extend(o) if isinstance(o, list) else flat_outputs.append(o)
    else:
        flat_outputs = outputs
    # 常见情况：没有 Command，直接按输入形态返回
    if not any(isinstance(o, Command) for o in flat_outputs):
        return flat_outputs if input_type == "list" else {self._messages_key: flat_outputs}
    # 复杂情况：有工具返回 Command（要跳转/改状态），逐个处理并合并 parent goto
    ...

✋ 10 分钟动手

# 1. 三段核心逐行读
sed -n '793,826p'  libs/prebuilt/langgraph/prebuilt/tool_node.py   # _func 并行
sed -n '922,1013p' libs/prebuilt/langgraph/prebuilt/tool_node.py   # 执行+兜错
sed -n '383,393p'  libs/prebuilt/langgraph/prebuilt/tool_node.py   # 默认错误策略

# 2. 亲手看"工具报错变消息"
python - <<'PY'
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool
from langchain_core.messages import AIMessage
@tool
def div(a:int,b:int)->float:
    "除法"
    return a/b
node = ToolNode([div], handle_tool_errors=True)   # 接住一切错误
ai = AIMessage(content="", tool_calls=[{"name":"div","args":{"a":1,"b":0},"id":"1","type":"tool_call"}])
out = node.invoke({"messages":[ai]})
print(out["messages"][0].content, out["messages"][0].status)  # 看到 Error: ... error
PY