LangGraph 系列 12：异步并发工具调用与 ToolNode

langgraph

发布日期: 2026-06-01

文章字数: 2.2k

阅读时长: 8 分

阅读次数:

在搭建完整 Workflow 之前，需要先解决一个更基础的问题：模型一次返回多个 Tool Call 时，工具应该怎样执行？

如果两个工具互不依赖，顺序等待会浪费时间。更合理的方式是同时启动两个异步任务，等待它们全部完成后，再把两个结果作为 ToolMessage 返回模型。

本文先手动实现一个最小异步工具节点，再使用 LangGraph 内置的 ToolNode 简化代码。

1. 同步、异步和并发的区别

同步执行表示当前函数没有结束时，调用方继续等待。异步函数使用 async def 定义，在等待网络、磁盘等 I/O 时，可以把执行机会交给其他任务。

异步不等于自动并发。下面的代码虽然调用了异步函数，但仍然是顺序执行：

weather = await lookup_weather("杭州")
attraction = await lookup_attraction("杭州")

只有同时创建多个任务，才能让等待过程重叠：

asyncio.gather() 会同时调度这两个协程，但返回列表仍按照传入协程的顺序排列。因此，输出顺序稳定并不表示两个工具也是按这个顺序完成的。

results = await asyncio.gather(
    lookup_weather("杭州"),
    lookup_attraction("杭州"),
)

多个 Tool Call 的异步并发执行

示例使用两个各等待一秒的函数。运行：

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

真实输出：

顺序执行结果： ['杭州：晴，25 摄氏度', '杭州：西湖']
顺序执行耗时：2.00 秒
并发执行结果： ['杭州：晴，25 摄氏度', '杭州：西湖']
并发执行耗时：1.00 秒

并发没有让单个工具执行得更快，而是让两个工具的等待时间重叠。只有互不依赖的工具才能这样执行。如果第二个工具需要第一个工具的输出，就必须保持顺序。

2. 模型如何表示多个工具调用

工具调用模型不会直接执行 Python 函数。它返回一条 AIMessage，并在 tool_calls 中描述想调用的工具：

AIMessage(
    content="",
    tool_calls=[
        {
            "name": "lookup_weather",
            "args": {"city": "杭州"},
            "id": "call-weather",
            "type": "tool_call",
        },
        {
            "name": "lookup_attraction",
            "args": {"city": "杭州"},
            "id": "call-attraction",
            "type": "tool_call",
        },
    ],
)

每个 Tool Call 都有三个关键字段：

name：工具名称。
args：调用参数。
id：本次调用的唯一标识。

工具执行完成后，应用需要返回匹配 ID 的 ToolMessage。模型依靠 tool_call_id 判断每个结果属于哪个请求。

3. 手动实现异步工具节点

自定义节点需要完成下面几件事：

检查最后一条消息是不是 AIMessage。
读取 tool_calls。
根据名称找到已经注册的 Tool。
使用 .ainvoke() 执行工具。
构造与调用 ID 对应的 ToolMessage。
使用 asyncio.gather() 等待所有工具。

核心实现如下：

class CustomAsyncToolNode:
    """读取最后一条 AIMessage，并发执行其中的所有工具请求。"""

    def __init__(self, tools: list[BaseTool]) -> None:
        # 先建立“工具名 → 工具对象”的字典，执行时便可按名称查找。
        self.tools_by_name = {current_tool.name: current_tool for current_tool in tools}

    async def __call__(self, state: dict[str, Any]) -> dict[str, list[ToolMessage]]:
        # 第 1 步：读取并检查最后一条消息。
        messages = state.get("messages", [])
        if not messages or not isinstance(messages[-1], AIMessage):
            raise ValueError("最后一条消息必须是 AIMessage")

        tool_calls = messages[-1].tool_calls
        if not tool_calls:
            raise ValueError("AIMessage 中没有工具调用")

        # 第 2 步：定义“执行一个工具调用”的异步函数。
        async def invoke_one(tool_call: dict[str, Any]) -> ToolMessage:
            tool_name = tool_call["name"]
            if tool_name not in self.tools_by_name:
                raise ValueError(f"未注册的工具：{tool_name}")

            # ainvoke() 执行真正的异步工具。
            result = await self.tools_by_name[tool_name].ainvoke(tool_call["args"])
            return ToolMessage(
                content=str(result),
                name=tool_name,
                tool_call_id=tool_call["id"],
            )

        # 第 3 步：多个 Tool Call 互不依赖，因此可以同时执行。
        outputs = await asyncio.gather(*(invoke_one(call) for call in tool_calls))
        return {"messages": list(outputs)}

运行完整脚本：

python langgraph/p12_async_tool_execution/02_custom_async_tool_node.py

真实输出：

lookup_weather call-weather 杭州：晴，25 摄氏度
lookup_attraction call-attraction 杭州：西湖
错误检查： 未注册的工具：missing_tool

工具名称校验非常重要。不能直接相信模型返回的名称，更不能让模型通过名称调用任意 Python 函数。

这个自定义实现还有一个明确边界：只要其中一个工具抛出异常，asyncio.gather() 默认就会把异常继续向上抛出。实际项目需要根据业务要求决定是终止整批调用，还是把单个错误转换成对应的 ToolMessage。

4. 使用内置 ToolNode

LangGraph 已经提供 ToolNode。它会读取 Tool Call、并行执行互不依赖的工具，并把普通工具结果转换成 ToolMessage。错误处理可以配置，但不能简单理解为所有异常都会被自动吞掉。

自定义工具节点与内置 ToolNode

结合上一节实现的 CustomAsyncToolNode，两种方式可以从下面几个方面比较：

对比项	CustomAsyncToolNode（自定义）	ToolNode（内置）
工具调用解析	手动读取 AIMessage.tool_calls，根据 name 查找工具并传入 args	自动解析 Tool Call，并在已经注册的工具中完成名称和参数匹配
同步与异步执行	自己决定调用 .invoke() 还是 .ainvoke()；多个调用需要手动使用 asyncio.gather() 等方式并发	同时支持同步和异步工具，并内置多个 Tool Call 的并行执行
结果与状态更新	手动构造 ToolMessage、关联 tool_call_id，再返回 {“messages”: …}	自动把普通工具结果转换成 ToolMessage，并以 State 局部更新的形式返回
错误处理	手动校验工具名称、捕获异常，并决定抛出还是转换成错误消息	内置工具名称与参数校验；可以通过 handle_tool_errors 配置异常处理策略
LangGraph 集成	需要自己注册节点、添加边并编写路由函数	可以直接作为 StateGraph 节点，并配合 tools_condition 建立标准工具循环
并发与资源控制	并发、超时、重试和限流策略都由业务代码实现	提供通用并发调度；特殊的顺序、限流和重试要求仍需额外实现
适用场景	需要审批、审计、参数改写或特殊调度规则的工具流程	大多数标准的模型工具调用流程

这里有两个容易混淆的地方：

ToolNode 返回的是 State 的局部更新，最终仍由 LangGraph 按 State Schema 和 Reducer 合并，并不是直接修改原 State。
tools_condition 是与 ToolNode 配合使用的预构建路由函数，不是 ToolNode 内部自动创建的边。

在下面的状态输入方式中，ToolNode 放进 StateGraph 后，由 LangGraph Runtime 提供执行配置：

# 第 1 步：把 ToolNode 放入最小图中，由 LangGraph Runtime 提供执行配置。
builder = StateGraph(MessagesState)
builder.add_node("tools", ToolNode([lookup_weather, lookup_attraction]))
builder.add_edge(START, "tools")
builder.add_edge("tools", END)
graph = builder.compile()

# 第 2 步：把包含 Tool Call 的 AIMessage 交给图。
result = await graph.ainvoke({"messages": [ai_message]})
for message in result["messages"]:
    if isinstance(message, ToolMessage):
        print(message.name, message.tool_call_id, message.content)

如果只传入消息字典，直接对这个版本的 ToolNode 调用 ainvoke()，会出现：

ValueError: Missing required config key 'N/A' for 'tools'.

这不是工具参数缺失，而是节点脱离了 LangGraph Runtime。放入最小图后运行成功：

lookup_weather call-weather 杭州：晴，25 摄氏度
lookup_attraction call-attraction 杭州：西湖

还需要区分两类错误：

参数缺失或不符合 Tool Schema 时，默认会返回 status=”error” 的 ToolMessage。
工具函数内部抛出的运行时异常默认会继续上抛；只有设置 handle_tool_errors=True 或提供自定义处理策略时，才会转换成错误 ToolMessage。

因此，ToolNode 提供的是可配置的错误处理机制，而不是无条件忽略所有工具异常。

5. 使用 tools_condition 路由

执行工具之前还需要判断最后一条模型消息是否包含 Tool Call。tools_condition 已经实现了这段判断：

# tools_condition 只检查 State 中的最后一条消息。
route_with_tool = tools_condition({"messages": [message_with_tool]})
route_without_tool = tools_condition({"messages": [message_without_tool]})

print("包含工具调用：", route_with_tool)
print("不包含工具调用：", route_without_tool)

真实输出：

包含工具调用： tools
不包含工具调用： __end__

在完整图中，可以直接这样连接：

builder.add_conditional_edges("chatbot", tools_condition)
builder.add_edge("tools", "chatbot")

模型提出工具请求时进入 tools，没有工具请求时结束。工具执行后重新回到模型，让模型读取 ToolMessage 并决定继续调用工具还是生成最终回答。

6. 自定义工具节点

普通工具循环优先使用 ToolNode，因为它已经处理了大量边界情况。下面这些场景才需要自定义：

只允许特定工具并发，其他工具必须顺序执行。
需要统一记录每个工具的业务审计日志。
不同工具使用不同的重试和超时策略。
执行高风险工具前需要人工审批。
需要修改 Tool Call 参数后再执行。

后续人工介入案例会再次使用这种细粒度控制能力。

hnbian

https://www.hnbian.cn/posts/a15d8adf.html

本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 hnbian !

langgraph tool asyncio

LangGraph 系列 13：使用 Bing 与 12306 MCP 构建 Workflow

2026-06-04 langgraph

langgraph workflow mcp

LangGraph 系列 11：使用评估器优化器循环改进模型输出

2026-05-28 langgraph

langgraph structured-output evaluator