Reborn 的技术博客

Agent 并行工具执行 —— 让多个工具同时干活

2026-06-23·AI, Agent, Concurrency, ThreadPool

核心增量：run_agent_with_trace 中工具执行段改写（~50 行净增）

一、问题：工具在排队，Agent 在空转

多个工具是串行执行的。当 LLM 一次性要求调用多个独立工具时：

串行执行（之前）
  search_web("北京") → 等 0.5s → 结果
  search_web("上海") → 等 0.5s → 结果
  search_web("广州") → 等 0.5s → 结果
  总耗时: 1.5s

并行执行（本篇）
  search_web("北京") ─┐
  search_web("上海") ─┼─ 同时跑 → 0.5s 全部完成
  search_web("广州") ─┘
  总耗时: 0.5s

一个慢工具（如超时中的 HTTP 请求）会阻塞后面所有快工具。

二、核心思路：线程池 + 批量提交 + 按完成顺序收集

三个关键决策：

  1. 线程池而非 asyncio —— 工具函数是同步的，线程池对同步阻塞 IO 最自然，且与已有的 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 = {"name": name, "result": result, "tc": tc}
            continue

        tool_result_spans.append((tc["id"], name, result))

    # Step 3: 如果 final_output 出现了，立即返回
    if final_output_found:
        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 先于搜索完成，立即返回会导致搜索的 tool 消息未被记录，下一次对话的上下文会丢失这些调用历史。

解决方案：final_output 完成后记录但不立即返回——等 as_completed 循环自然结束，确保所有并发工具的结果都被记录，再统一返回。

五、与串行模式的对比

| 维度 | 串行 | 并行 | | --- | --- | --- | | 执行方式 | for tc: result = ... | executor.submit → as_completed | | 总耗时 | N × 单次耗时 | max(单次耗时) | | 慢工具影响 | 阻塞后续所有 | 不阻塞其他（超时仍生效） | | 结果顺序 | 按 LLM 输出顺序 | 按完成先后 | | 代码复杂度 | 简单 | 需要 future_map + as_completed | | final_output | 立刻返回 | 等其他工具跑完 |

六、设计精要

最大并行度 = 工具数量

max_workers=len(tool_calls) 不创建多余线程，不预设上限，也不浪费资源。

复用 call_with_timeout

外层 ThreadPoolExecutor 管理工具间并行，内层 call_with_timeout 管理单个工具的超时，两层各司其职。

结果顺序不影响 LLM 推理

每个 tool 消息都带 tool_call_id，LLM 通过 id 关联而非位置。

零侵入前序代码

SkillManager、AgentTracer、ContextManager、PersistenceManager 完全不受影响——它们不知道工具是串行还是并行。

#AI#Agent#Concurrency#ThreadPool