如何正确地调用外部工具（Tool Calling），并将工具执行的结果（ToolMessage）更新回系统状态中？...

知识点图片

在 LangGraph 中，正确调用外部工具并将结果更新到系统状态（State）的核心机制是：大模型生成带有 tool_calls 的 AIMessage -> 路由到工具节点 -> 工具节点执行 Python 函数并生成 ToolMessage -> 将 ToolMessage 追加到状态中 -> 返回大模型继续推理。

LangGraph 提供了非常优雅的内置模块来简化这个过程，同时你也可以通过自定义节点完全控制它。

下面我将分步骤详细讲解，并提供完整的可运行代码示例。

核心原理解析

在包含工具调用的 State 中，最关键的是消息列表（Messages List）的流转：

Agent 节点：调用绑定了工具的 LLM。如果 LLM 决定使用工具，它会返回一个 AIMessage，里面包含 tool_calls 属性（包含工具名、参数和唯一的 tool_call_id）。
状态更新：这个 AIMessage 被追加到 State 的 messages 列表中。
工具节点（Tool Node）：提取上一步的 tool_calls，实际执行你的 Python 函数。
生成 ToolMessage：将函数的执行结果包装成 ToolMessage。注意：ToolMessage 必须包含对应的 tool_call_id，这样大模型才知道这个结果对应的是哪个工具调用请求。
状态更新与循环：ToolMessage 被追加到 State 中，流程切回 Agent 节点，LLM 结合 ToolMessage 给出最终回答。

方法一：使用内置的 `ToolNode`（强烈推荐，最简单）

LangGraph 的 langgraph.prebuilt 模块提供了 ToolNode 和 tools_condition，极大地简化了工具调用的代码。

1. 定义状态和工具

python

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from langchain_core.tools import tool

# 1. 定义系统状态 (使用 add_messages reducer 来自动拼接消息)
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 2. 定义外部工具
@tool
def get_weather(location: str) -> str:
    """获取指定城市的天气。"""
    # 实际应用中这里是调用天气 API
    if "北京" in location:
        return "晴天，25度"
    return "未知天气"

tools = [get_weather]

2. 构建图 (Graph)

python

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode, tools_condition

# 3. 初始化大模型并绑定工具
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
llm_with_tools = llm.bind_tools(tools)

# 4. 定义 Agent 节点
def agent_node(state: State):
    # 将当前的所有消息传给大模型
    response = llm_with_tools.invoke(state["messages"])
    # 返回的内容会被 add_messages 自动追加到 state["messages"] 中
    return {"messages": [response]}

# 5. 构建图
builder = StateGraph(State)

# 添加节点
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode(tools)) # 直接使用内置的 ToolNode

# 添加边 (路由逻辑)
builder.add_edge(START, "agent")

# 条件边：如果 agent 返回了 tool_calls，去 "tools" 节点；否则去 END
builder.add_conditional_edges(
    "agent",
    tools_condition, # 内置的条件判断逻辑
)

# 工具执行完毕后，必须回到 agent 节点让 LLM 总结
builder.add_edge("tools", "agent")

# 编译成可运行的图
graph = builder.compile()

3. 运行测试

python

from langchain_core.messages import HumanMessage

initial_state = {"messages": [HumanMessage(content="北京的天气怎么样？")]}

for event in graph.stream(initial_state, stream_mode="values"):
    last_message = event["messages"][-1]
    last_message.pretty_print()

方法二：自定义工具节点（了解底层机制）

如果你不想使用内置的 ToolNode（例如你需要自定义错误处理、日志记录，或者非阻塞异步执行），你可以自己写一个工具节点函数。

关键点在于：你必须手动构建 ToolMessage，并且保证 tool_call_id 一致。

自定义 Tool Node 的实现：

python

import json
from langchain_core.messages import ToolMessage

# 将工具列表转换为字典，方便按名称查找
tool_mapping = {tool.name: tool for tool in tools}

def custom_tool_node(state: State):
    """自定义执行工具并更新状态的节点"""
    # 1. 获取最后一条消息 (肯定是一个带有 tool_calls 的 AIMessage)
    last_message = state["messages"][-1]
    
    tool_messages = []
    
    # 2. 遍历大模型请求的所有工具调用
    for tool_call in last_message.tool_calls:
        # 提取工具名、参数和 ID
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]
        tool_call_id = tool_call["id"]
        
        # 3. 执行对应的 Python 函数
        if tool_name in tool_mapping:
            tool_func = tool_mapping[tool_name]
            try:
                # 调用工具
                result = tool_func.invoke(tool_args)
            except Exception as e:
                result = f"Error: {str(e)}"
        else:
            result = f"Error: Tool {tool_name} not found."
            
        # 4. 构造 ToolMessage (这是将结果存回 State 的关键)
        tool_message = ToolMessage(
            content=str(result),         # 工具执行结果 (必须是字符串格式)
            name=tool_name,              # 工具名称
            tool_call_id=tool_call_id    # ★★★ 必须严格匹配 AIMessage 中的 id ★★★
        )
        tool_messages.append(tool_message)
        
    # 5. 返回更新的状态 (add_messages 会将其追加到列表中)
    return {"messages": tool_messages}

# 在构建图时，将内置的 ToolNode 替换为你自定义的：
# builder.add_node("tools", custom_tool_node)

总结与避坑指南

add_messages Reducer 是必须的：在定义 State 时，messages 字段必须使用 add_messages 注解。这样你每次从节点返回 {"messages": [new_msg]} 时，LangGraph 才会将其追加到历史记录中，而不是覆盖原列表。
tool_call_id 不可或缺：大模型 API（如 OpenAI）要求 AIMessage 中的请求 ID 和后续提交的 ToolMessage 的 ID 必须一一对应。如果对应不上，API 会抛出错误。这也是为什么推荐直接使用内置 ToolNode 的原因，它在底层自动处理了这些映射关系。
闭环（Loop）：工具节点执行完毕后，边（Edge）必须连回 Agent 节点（builder.add_edge("tools", "agent")），让大模型看到 ToolMessage 的内容，从而生成最终给用户的自然语言回答。

播面