目录
Hello，大家好，我是 1M55！👋
继续更新我们的「Agent 面试高频手撕代码」系列！
前面我们已经手撕过 ReAct、Function Calling、Sub-Agent、Agent Skills、MCP Client、RAG 和混合检索。
今天来点简单但实用的。
假设用户提出一个问题：
帮我同时查询北京和上海的天气。
模型在同一轮生成了两个 tool_calls。
如果我们仍然像之前那样，用一个 for 循环依次执行工具，两个请求就只能串行等待。
但它们之间明明没有依赖关系。
既然两个工具可以同时执行，为什么非要一个接一个地等？
所以今天我们来手撕一个 Agent 并行工具执行器。
不依赖 LangChain / LangGraph，核心只用 Python 的 asyncio。
一、Agent 为什么需要并行工具调用？
先看最简单的串行执行：
for call in tool_calls:
result = await dispatch_tool(call)这里假设 dispatch_tool() 是异步函数。
虽然写了 await，但它仍然是串行执行。
因为每一次循环都会等待当前工具完成，然后才开始下一个工具。
假设：
-
工具 A 耗时 2 秒。
-
工具 B 耗时 3 秒。
那么串行执行的总耗时大约是：
工具 A：████ 2s
工具 B： ██████ 3s
总耗时：约 5s如果两个工具之间没有依赖关系，我们可以同时启动它们：
工具 A：████ 2s
工具 B：██████ 3s
总耗时：约 3s这就是并行工具调用的价值。
严格来说，Python asyncio 在这里实现的是异步并发，特别适合网络请求、数据库访问等 I/O 密集型任务，并不意味着两个 Python 函数一定同时占用不同 CPU 核心。
二、核心知识点：asyncio.gather()
先看一段最简单的代码：
import asyncio
async def tool_a():
await asyncio.sleep(2)
return "工具 A 完成"
async def tool_b():
await asyncio.sleep(3)
return "工具 B 完成"
async def main():
results = await asyncio.gather(
tool_a(),
tool_b()
)
print(results)
asyncio.run(main())这里的 asyncio.sleep() 只是用来模拟工具等待外部服务的耗时，不代表真实天气 API。
重点看：
results = await asyncio.gather(
tool_a(),
tool_b()
)asyncio.gather() 会让两个协程并发运行，并在它们完成后收集结果。
这里有一个非常重要的细节：
gather() 返回结果的顺序，与传入任务的顺序一致，不取决于哪个任务先完成。
例如，虽然工具 A 先完成，工具 B 后完成，最终结果仍然是：
[
"工具 A 完成",
"工具 B 完成"
]这一点对 Agent 很重要，因为接下来还需要把每个工具结果对应到原来的 tool_call_id。
三、手撕 Agent 并行工具执行器
正式进入今天的核心代码。
面试官给出题目：
LLM 在一轮响应中返回多个 tool_calls，请使用 Python 并发执行这些工具，并按照 Function Calling 协议回填结果；其中一个工具执行失败，不能导致其他工具结果丢失。
我们依然沿用第二篇学习过的 tool_calls 数据结构。
3.1 工具注册表
首先假设已经有两个支持异步调用的工具：
async def get_weather(city):
# 真实开发中，这里可以调用异步天气 API
...
async def get_stock(product_id):
# 真实开发中，这里可以调用异步库存 API
...
TOOLS = {
"get_weather": get_weather,
"get_stock": get_stock
}工具注册表没有变化。
它仍然负责根据模型生成的工具名称找到对应 Python 函数。
只是这次，我们要求工具是异步函数。
3.2 实现单个工具分发器
这部分与第二篇 Function Calling 非常接近：
import json
async def dispatch_tool(call):
name = call["function"]["name"]
try:
args = json.loads(
call["function"]["arguments"]
)
if not isinstance(args, dict):
raise ValueError("工具参数必须是 JSON 对象")
func = TOOLS.get(name)
if func is None:
raise ValueError(f"未知工具：{name}")
result = await func(**args)
return {"result": result}
except Exception as e:
return {"error": str(e)}这里重点看两行：
func = TOOLS.get(name)根据工具名称查找函数。
result = await func(**args)异步执行对应的工具。
另外，我们使用 try/except 捕获工具错误。
这样单个工具执行失败，也能返回错误结果，而不是直接让整个工具调度过程终止。
这里捕获的是普通运行异常；取消任务等控制流程需要在更完整的执行器中单独处理。
3.3 并行执行多个工具
真正需要手撕的新逻辑来了：
import asyncio
async def run_tools(tool_calls):
tasks = [
dispatch_tool(call)
for call in tool_calls
]
results = await asyncio.gather(*tasks)
tool_messages = []
for call, result in zip(tool_calls, results):
tool_messages.append({
"role": "tool",
"tool_call_id": call["id"],
"content": json.dumps(
result,
ensure_ascii=False
)
})
return tool_messages整个并行调度的核心就是：
results = await asyncio.gather(*tasks)它把原本串行执行的多个异步工具调用，改成并发执行。
后面的代码只负责将执行结果重新包装成模型需要的消息格式。
3.4 为什么可以使用 zip()？
注意这里：
for call, result in zip(tool_calls, results):我们可以这样写，是因为 asyncio.gather() 保证结果顺序与输入顺序一致。
例如：
tool_calls：
call_001 -> get_weather
call_002 -> get_stock
results：
result_001
result_002即使 get_stock 比 get_weather 更早完成，results 中的顺序仍然与提交顺序对应。
因此可以安全地通过 zip() 配对。
最终生成：
[
{
"role": "tool",
"tool_call_id": "call_001",
"content": "..."
},
{
"role": "tool",
"tool_call_id": "call_002",
"content": "..."
}
]这就是多工具执行结果的回填过程。
四、如果其中一个工具失败怎么办？
这也是面试官可能追问的地方。
假设模型同时请求了三个工具：
工具 A：执行成功
工具 B：执行失败
工具 C：执行成功我们不希望因为工具 B 抛异常，就导致工具 A、C 的结果无法回填。
前面的 dispatch_tool() 已经捕获了工具执行过程中的普通异常：
try:
result = await func(**args)
return {"result": result}
except Exception as e:
return {"error": str(e)}因此工具 B 的错误会被转换成普通结果：
{
"error": "工具执行失败"
}另外两个工具仍然可以返回正常结果。
最终三条工具结果都能够生成对应的 tool 消息。
这里的原则是：单个工具失败，不应该自动导致其他独立工具的结果丢失。
但需要注意，并不是所有工具都适合并发执行。
比如两个工具同时修改同一份文件，或者后一个工具依赖前一个工具的输出，就不能简单地全部放进 gather()。
五、如何接入真实 Agent Loop？
这一部分很简单。
假设我们已经有一个 Agent Loop，模型返回：
reply = response.choices[0].message如果包含多个工具调用：
if reply.tool_calls:
...我们只需要将它们转换成前面定义的字典格式：
calls = [
{
"id": call.id,
"function": {
"name": call.function.name,
"arguments": call.function.arguments
}
}
for call in reply.tool_calls
]然后并发执行：
tool_messages = await run_tools(calls)最后回填：
messages.append(
reply.model_dump(exclude_none=True)
)
messages.extend(tool_messages)这里仍然遵守 Chat Completions 的工具调用消息协议：
先保存 assistant 发起的工具调用消息，再追加所有对应的 tool 结果消息。
模型接收到执行结果后，就可以继续生成最终回答。
到这里，我们就完成了从串行工具调用到异步并发工具调用的升级。
六、面试官可能会怎么追问？
Q1：为什么使用 asyncio.gather()？
因为多个独立的 I/O 工具调用可以并发等待。
相比串行执行，并发通常可以减少整体等待时间。
Q2：asyncio.gather() 会保证结果顺序吗？
会。返回结果的顺序与传入协程或任务的顺序一致，与实际完成顺序无关。
因此可以使用 zip(tool_calls, results) 将结果对应回原始调用。
Q3：asyncio 适合所有工具吗？
不是。
它主要适合异步 I/O。
如果工具内部执行大量 CPU 密集型计算，仅仅使用 asyncio 不一定能提升性能。
如果是同步阻塞 I/O，可以考虑通过 asyncio.to_thread() 在线程中执行，或者直接使用支持异步的客户端库。
Q4：某个工具执行失败怎么办？
可以在单工具执行器中捕获普通异常，将错误转换为工具结果，避免影响其他独立工具的结果回填。
但超时、取消任务和敏感操作失败还需要更明确的处理策略。
Q5：是不是所有 tool_calls 都应该并发执行？
不是。
只有彼此独立、允许同时执行的工具才适合并发。
涉及共享状态、执行顺序或有副作用的工具，需要先分析依赖与并发安全性。
Q6：为什么要保留 tool_call_id？
因为一轮模型响应可能包含多个工具调用。
每条工具结果必须与原始调用建立对应关系，模型才能知道哪个结果属于哪个请求。
七、总结
这一篇没有引入复杂框架，而是对第二篇 Function Calling 做了一个很实用的升级。
串行工具调用：
for call in tool_calls:
result = await dispatch_tool(call)并发工具调用：
results = await asyncio.gather(
*(dispatch_tool(call) for call in tool_calls)
)核心的区别就在于：独立的异步工具不再依次等待，而是并发执行。
同时，我们还保留了三个工程上重要的能力：
-
通过
tool_call_id正确关联执行结果。 -
单个工具失败时，不丢失其他独立工具的结果。
-
按照模型工具调用协议，将结果交还给 Agent。
回顾整个系列：
01 · ReAct：循环决策。
02 · Function Calling：工具调用。
03 · Sub-Agent：任务委派。
04 · Agent Skills：技能加载。
05 · MCP Client：外部工具接入。
06 · RAG：检索增强生成。
07 · Hybrid Search：混合检索。
08 · Parallel Tool Calling：并行工具调用。
我是 1M55，我们下篇见！
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/2501_94225300/article/details/169364731
