Agent 并行工具执行 —— 让多个工具同时干活，而非排队等
一、问题：工具在排队，Agent 在空转
前几篇文章搭建的Agent，其实藏着个性能隐患——所有工具都是串行执行的。换句话说，当LLM一次性要求调用多个独立工具时，事情就变成了排队等号：
举个例子，让LLM查三个城市的天气：北京、上海、广州。串行模式下，查询北京的代码跑完，等0.5秒拿到结果，再查上海，再等0.5秒，再查广州，再等0.5秒，总耗时1.5秒。而并行模式呢？三个查询同时提交，0.5秒全部完成。这可不是理论层面的优化——当LLM决定同时调用5个独立的搜索、计算或者文件读取工具时，串行就意味着用户要多等4倍的时间。更头疼的是，一个慢工具还会堵住后面所有快工具，白白浪费资源。
二、核心思路：线程池 + 批量提交 + 按完成顺序收集
来看一下对比：
flowchart LR
subgraph OLD[串行模式]
O1[search_web 北京 0.5s] --> O2[search_web 上海 0.5s]
O2 --> O3[search_web 广州 0.5s]
O3 --> O4[总耗时 1.5s]
end
subgraph NEW[并行模式]
N1[search_web 北京 0.5s]
N2[search_web 上海 0.5s]
N3[search_web 广州 0.5s]
N1 --> N4[总耗时 0.5s]
N2 --> N4
N3 --> N4
end
实现这个优化，有三个关键决策：
- ——工具函数本身就是同步的，直接用线程池最自然，而且和已有的
线程池而非asyncio
call_with_timeout配合得天衣无缝。 - ——用
按完成顺序收集
as_completed()让快的工具先返回，慢的不会阻塞其他工具。 - ——它先到，但不立即返回，等其他工具全部跑完才一起收尾。
final_output特殊处理
三、实现
具体代码不长，但逻辑很清晰：
with concurrent.futures.ThreadPoolExecutor(max_workers=len(tool_calls)) as executor:
# Step 1: 提交所有工具到线程池
future_map = {}
for tc in tool_calls:
name = tc["function"]["name"]
args = json.loads(tc["function"]["arguments"])
future = executor.submit(
call_with_timeout, # ← 复用已有的超时包装
active_tool_map[name],
kwargs=args, timeout=30,
)
future_map[future] = {"name": name, "args": args, "tc": tc}
# Step 2: 按完成顺序收集结果
tool_result_spans = []
final_output_found = None
for future in concurrent.futures.as_completed(future_map):
meta = future_map[future]
name, args, tc = meta["name"], meta["args"], meta["tc"]
try:
result = future.result()
except Exception as e:
result = f"工具执行错误: {e}"
# final_output：记录但不立即返回
if name in OUTPUT_TOOL_NAMES:
final_output_found = {"result": result, "tc": tc}
tracer.log_tool_call(name, args, result, ...)
continue
tracer.log_tool_call(name, args, result, ...)
tool_result_spans.append((tc["id"], name, result))
# Step 3: final_output 到达 → 返回
if final_output_found:
messages.append({"role": "assistant", "content": ...})
return final_output_found["result"]
# Step 4: 追加工具结果到 messages
for tc_id, name, result in tool_result_spans:
messages.append({"role": "tool", "tool_call_id": tc_id, "content": result})
这代码的核心就是：先统一提交，然后谁先做完谁先回来，最后再统一处理final_output。逻辑上没什么弯弯绕绕。
四、final_output 的特殊处理
并行模式下，final_output可能会和其他工具同时被调用。如果它先完成就立刻返回，那其他工具的结果还没写入messages，下一次对话就会丢失上下文。所以必须等所有工具都跑完，才让final_output返回。
来看这个时序图就能明白：
sequenceDiagram
participant LLM as LLM
participant EX as ThreadPoolExecutor
participant SW as search_web
participant FO as final_output
LLM->>EX: 提交 3 个工具
EX->>SW: search_web("北京")
EX->>SW: search_web("上海")
EX->>FO: final_output({...})
SW-->>EX: "北京结果" (0.3s)
FO-->>EX: "结构化答案" (0.4s)
Note over EX: final_output 到了但不返回
SW-->>EX: "上海结果" (0.8s)
Note over EX: 所有工具完成，final_output 返回
EX-->>LLM: 结构化答案
注意看，final_output在0.4秒就回来了，但必须等到上海结果（0.8秒）也回来，才真正返回给LLM。这个细节虽然小，却是保证多轮对话上下文完整的关键。
五、与串行模式的对比
| 维度 | 串行 | 并行 |
|---|---|---|
| 执行方式 | for tc in tool_calls: result = ... | executor.submit → as_completed |
| 总耗时 | N × 单次耗时 | max(单次耗时) |
| 慢工具影响 | 阻塞后续所有工具 | 不阻塞其他工具 |
| 结果顺序 | 按 LLM 输出顺序 | 按完成先后顺序 |
| 代码复杂度 | 简单 | 需 future_map + as_completed |
| final_output | 立刻返回 | 等其他工具跑完才返回 |
从表格里能直观看到，并行模式在耗时上直接砍掉了大部分等待时间，代价是代码稍微复杂了一点点，但带来的性能提升是实打实的。
六、设计精要
6.1 最大并行度 = 工具数量
max_workers=len(tool_calls)——LLM一次调几个工具就开几个线程，不预设上限也不浪费资源。这个设计很聪明，避免了固定线程池大小带来的伸缩性问题。
6.2 复用 call_with_timeout
两层线程池各司其职：外层管理工具间的并行，内层管理单个工具的超时。并行化并没有重新实现超时逻辑，而是直接复用了已有的call_with_timeout，干净利落。
6.3 结果顺序不影响 LLM 推理
每个tool消息都带有tool_call_id，LLM通过id来关联，而不是靠位置顺序。所以即使结果乱序回来，LLM也能正确识别。
6.4 零侵入前序代码
这次改动只影响了run_agent_with_trace中的一段代码，其他模块——SkillManager、AgentTracer、ContextManager、PersistenceManager、chat_loop——完全无感。这种局部改造的思维，在实际开发中特别重要。
七、完整演进路径
Phase 1: 基础 Agent (~50 行)
Phase 2: 可观测 + 压缩 + 持久化 (+~200 行)
Phase 3: Skill 渐进加载 (+~150 行)
Phase 4: 多轮对话循环 (+~120 行)
Phase 5: 流式输出 (+~80 行)
Phase 6: 重试 + 超时 (+~128 行)
Phase 7: 结构化输出 (+~140 行)
Phase 8: 并行工具执行 (+~50 行) ← 本文
回顾一下整个进化路线：从最基础的Agent开始，一步步加上可观测性、压缩、持久化、Skill渐进加载、多轮对话、流式输出、重试与超时、结构化输出，再到现在的并行工具执行。每一步都控制在100-200行左右，没有大改架构，没有加新文件，只是把串行for循环升级成了线程池并行。这种渐进式演进的思路，对于构建复杂系统来说，真的非常实用。
