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

实现这个优化，有三个关键决策：

1. 线程池而非asyncio
   ——工具函数本身就是同步的，直接用线程池最自然，而且和已有的call_with_timeout配合得天衣无缝。

2. 按完成顺序收集
   ——用as_completed()让快的工具先返回，慢的不会阻塞其他工具。

3. 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。这个细节虽然小，却是保证多轮对话上下文完整的关键。

五、与串行模式的对比

从表格里能直观看到，并行模式在耗时上直接砍掉了大部分等待时间，代价是代码稍微复杂了一点点，但带来的性能提升是实打实的。

六、设计精要

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循环升级成了线程池并行。这种渐进式演进的思路，对于构建复杂系统来说，真的非常实用。