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

前面已经使用 LangGraph 编写了固定流程、条件路由和评估器循环。接下来要实现一个可以连续调用工具的智能小秘书。在搭建完整 Workflow 之前，需要先解决一个更基础的问题：模型一次返回多个 Tool Call 时，工具应该怎样执行？

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

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

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

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

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

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

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

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

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

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

cd source/_posts/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. 手动实现异步工具节点

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

1. 检查最后一条消息是不是 AIMessage。
2. 读取 tool_calls。
3. 根据名称找到已经注册的 Tool。
4. 使用 .ainvoke() 执行工具。
5. 构造与调用 ID 对应的 ToolMessage。
6. 使用 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]]:
        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 中没有工具调用")

        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}")

            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"],
            )

        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。错误处理可以配置，但不能简单理解为所有异常都会被自动吞掉。

本文固定使用 langgraph==1.1.3。在本文使用的状态输入方式中，ToolNode 放进 StateGraph 后，由 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()

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 已经实现了这段判断：

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 参数后再执行。

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

7. 总结

知识点	作用	本文验证结果
async def	定义可以等待 I/O 的异步函数	两个演示工具均可异步调用
await	等待一个异步任务完成	连续 await 仍然顺序执行
asyncio.gather()	同时等待多个独立任务	耗时由约 2 秒降到约 1 秒
AIMessage.tool_calls	保存模型提出的工具请求	一条消息包含两个 Tool Call
ToolMessage	把工具结果返回模型	每条结果保留对应调用 ID
自定义工具节点	完全控制校验、并发和异常逻辑	未注册工具会被明确拒绝
ToolNode	并行执行工具并生成 ToolMessage	参数错误可返回错误消息，运行时异常可配置处理
tools_condition	根据最后一条消息选择工具或结束	返回 tools 或 end

至此，模型、工具节点和路由函数已经准备完成。下一篇会把 Bing 与 12306 两个 MCP Server 接入同一张图，构建完整的智能小秘书 Workflow。