《AI Agent 开发入门》4.3 并行工具调用与结果合并

4.3 并行工具调用与结果合并

前三节里，minagent 一次只处理一个工具调用。但真实模型经常在一轮里同时发出好几个 tool_call——比如"查一下北京和上海的天气"会变成两个并行的调用请求。如果注册表只会一个接一个地跑，这些本该重叠的等待就被串成了长链。

本节把 minagent 推进到 minagent/tools.py 的调度层：让 ToolRegistry 支持一轮内多个工具并发执行，并把结果按 tool_call_id 对齐合并回 messages。

4.3.1 一轮发出多个 tool_call

回顾第 1 章的消息协议：一条 assistant 消息的 tool_calls 是一个列表。模型在一轮里返回的 ModelResponse.message.tool_calls 可以有多项，每一项都是一个独立的 ToolCall。第 3 章的循环当时只处理了第一项——这一节把它补全。

先看"一次一个"的串行写法，它直观但慢：

def run_serial(reg, calls):
    out = []
    for tc in calls:
        out.append((tc.id, reg.call(tc)))
    return out

三个工具各睡 0.4 秒，串行就是 1.2 秒。它们之间没有依赖——查 A 和查 B 互不相干，等待完全可以重叠。

循环里怎么知道"这一轮要不要并发"？看 ModelResponse.stop_reason：当它是 "tool_use" 时，说明模型在请求工具；此时 message.tool_calls 里有几项，就并发跑几项。这是个纯机械的判断，不需要任何语义理解：

resp = model.complete(messages, reg.schemas())
if resp.stop_reason == "tool_use":
    calls = resp.message.tool_calls     # 可能有多项

哪些工具适合并行、哪些不能，取决于它们之间的关系：

卷一的默认策略很简单：只对只读工具并行。写操作、有依赖的调用一律串行。

4.3.2 用 ThreadPoolExecutor 并发执行

工具大多是 IO 密集（网络、磁盘、数据库），线程池是最省事的并发手段。把每个 ToolCall 提交成一个 future，再按原顺序取回结果：

from concurrent.futures import ThreadPoolExecutor

def run_parallel(reg, calls):
    with ThreadPoolExecutor(max_workers=len(calls)) as pool:
        futs = {tc.id: pool.submit(reg.call, tc) for tc in calls}
    # 按原顺序取回，保证与 tool_call_id 对应
    return [(tc.id, futs[tc.id].result()) for tc in calls]

注意两个细节：futs 用 tool_call_id 做键，这样取回结果时天然和请求一一对应；max_workers 设成调用数，让它们真正同时开跑（默认值在 Python 3.14 里是 min(32, CPU+4)，多数情况够用，但显式指定更清楚）。

一个慢工具的桩函数，sleep 模拟真实耗时：

import time

def slow_fetch(url: str) -> str:
    time.sleep(0.4)
    return f"fetched {url}"

calls = [ToolCall(f"c{i}", "slow_fetch", {"url": f"https://ex.com/{i}"})
         for i in range(1, 4)]

真跑两种写法，实测耗时（Python 3.14.6）：

=== 串行 ===
  c1: fetched https://ex.com/1
  c2: fetched https://ex.com/2
  c3: fetched https://ex.com/3
串行耗时: 1.21s

=== 并行 ===
  c1: fetched https://ex.com/1
  c2: fetched https://ex.com/2
  c3: fetched https://ex.com/3
并行耗时: 0.41s
加速比: 2.9x
顺序与 tool_call_id 一致: True

三个 0.4 秒的工具，串行 1.21 秒，并行 0.41 秒，加速比接近 3——正好等于调用数，因为它们完全重叠。如果工具之间有依赖（B 要用 A 的输出），就不能这样并行，得拆成两轮。

并发还有一个好处：一个工具失败不会拖垮其他。因为每个 reg.call 内部已经把异常包成了 {"error": ...}，future 本身不会抛，取结果时自然各归各位：

=== 一个失败不影响其他 ===
  c1: fetched https://ex.com/1
  c2: {"error": "ValueError: boom"}
  c3: fetched https://ex.com/2

c2 抛了异常，c1、c3 照常返回。三个结果一起回灌给模型，模型看到"有一个失败"就能决定是重试、换工具还是放弃。这正是 4.2 节"call() 永不抛异常"约定的回报——如果工具层把异常直接抛出来，ThreadPoolExecutor 会在取结果时把整个循环炸掉。

4.3.3 asyncio.gather 的等价写法

如果工具本身就是 async 函数，用 asyncio.gather 更自然，也省掉线程开销：

import asyncio

async def slow_fetch_async(url: str) -> str:
    await asyncio.sleep(0.4)
    return f"fetched {url}"

async def main() -> None:
    calls = [ToolCall(f"c{i}", "slow_fetch", {"url": f"https://ex.com/{i}"})
             for i in range(1, 4)]
    results = await asyncio.gather(*(slow_fetch_async(**c.arguments) for c in calls))
    print(results)

实测同样在 0.40 秒左右完成。两种写法的取舍：

卷一默认用线程池，因为它不需要把同步工具改写成异步。等到第 7 章引入流式事件、第 14 章做超时取消时，才会切到 asyncio 主线。

如果工具是同步的、又想走 asyncio 主线，可以用 loop.run_in_executor 把它丢进线程池再 await：

async def call_async(reg, tc):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, reg.call, tc)

两种方案不是对立的——asyncio 内部本来就用线程池跑阻塞调用。真正的分水岭是工具的实现形态：库是同步的就用线程池，库是异步的（httpx.AsyncClient）就用 gather。混用会带来"在线程里再开事件循环"的麻烦，所以一个项目最好选定一种。

4.3.4 合并：按 tool_call_id 对齐，而不是按完成顺序

并发最大的坑是完成顺序 ≠ 提交顺序。如果 c2 先跑完、c1 后跑完，按完成顺序 append 会把结果错位挂到别的调用上。正确做法是始终按原始 calls 的顺序取结果，用 tool_call_id 保证对齐：

results = run_parallel(reg, calls)
for tc in calls:
    msg = Message(role="tool", content=dict(results)[tc.id],
                  tool_call_id=tc.id)
    messages.append(msg)

每一轮的工具结果都变成一条 role="tool" 的消息，tool_call_id 指回对应的请求。模型下一轮看到这组消息，就能把"我请求了三个 URL"和"三个结果"对上。run_parallel 内部用 futs[tc.id].result() 按 calls 顺序取，正是为了这个对齐——.result() 会阻塞到该 future 完成，但因为我们最后统一取，总耗时仍约等于最慢的那个工具。

可以把对齐逻辑抽成一个小函数，用 tool_call_id 建索引，无论结果以什么顺序回来都能还原成正确顺序：

def merge_tool_messages(calls, results):
    by_id = dict(results)
    return [Message(role="tool", content=by_id[tc.id], tool_call_id=tc.id)
            for tc in calls]

即使结果列表是 [("c2", ...), ("c1", ...)]（完成顺序颠倒），输出仍按 calls 的顺序给出 c1、c2——因为我们是按 tc.id 查，而不是按位置取。

把并行调度接回第 3 章的 run() 循环，就是一个能跑的端到端版本。下面用假模型脚本发出两个 tool_call，再让它收尾：

class FakeModel:
    def __init__(self, script): self.script = list(script)
    def complete(self, messages, tools):
        item = self.script.pop(0)
        if isinstance(item, str):
            return ModelResponse(Message(role="assistant", content=item))
        return ModelResponse(Message(role="assistant", tool_calls=item),
                             stop_reason="tool_use")

script = [[ToolCall("c1", "slow_fetch", {"url": "https://ex.com/1"}),
           ToolCall("c2", "slow_fetch", {"url": "https://ex.com/2"})],
          "两个页面都抓到了"]
msgs = run(FakeModel(script), reg, "抓这两个页面")

实测轨迹（总耗时 0.41 秒，两个 0.4 秒的工具并行）：

[user] 抓这两个页面
[assistant] calls=c1,c2 
[tool] -> c1 fetched https://ex.com/1
[tool] -> c2 fetched https://ex.com/2
[assistant] 两个页面都抓到了

轨迹清晰展示了这一轮的形状：assistant 一次发出 c1、c2 两个调用，两条 tool 消息各带自己的 tool_call_id 回灌，最后 assistant 给出终答。这段轨迹可以被第 16 章的确定性测试直接断言——这也是卷一用假模型的原因：并发这种"看起来对"的东西，只有把轨迹打印出来才能确认它真的对。

4.3.5 并发度不是越大越好

max_workers=len(calls) 在教学里没问题，但真实场景下模型可能一轮发出十几个调用，同时打向同一个第三方接口就会触发限流。给线程池设一个固定上限，多余的调用自动排队：

import time

def work(i: int) -> int:
    time.sleep(0.2)
    return i

with ThreadPoolExecutor(max_workers=2) as pool:   # 最多 2 个并发
    results = list(pool.map(work, range(6)))

实测：6 个各睡 0.2 秒的任务，并发度 2 时耗时 0.61 秒，正好是 6 / 2 × 0.2——说明上限确实在起作用，任务被分批执行。并发度、限流、重试退避是同一类工程问题，卷一先给出"能设上限"这个最小能力，策略化的部分留到卷二。

到这里，工具层三节就闭环了：4.1 定义形状与分发，4.2 守住参数与执行边界，4.3 把一轮内的多个调用压进并发。minagent 从此有了"手"，而且是一双能同时伸出去、又不会互相踩到的手。

真实模型下会怎样：假模型要并行调用，得我们在脚本里手写一个含多个 ToolCall 的列表；真实 Claude 会在一次响应里自然地发出多个 tool_use 块，尤其是当提示里出现"分别"“各自"“对比"这类词时。真实场景的并发还要考虑两个变量：一是工具是否幂等——并行的两个写操作可能互相踩踏，卷一先只并行只读工具；二是API 的并发限制——同一轮里十个工具并发打第三方接口，可能触发限流，需要给线程池设上限而不是无脑 max_workers=len(calls)。这些生产化问题留到卷二。

真实模型下会怎样：假模型要并行调用，得我们在脚本里手写一个含多个 ToolCall 的列表；真实 Claude 会在一次响应里自然地发出多个 tool_use 块，尤其是当提示里出现"分别"“各自"“对比"这类词时。真实场景的并发还要考虑两个变量：一是工具是否幂等——并行的两个写操作可能互相踩踏，卷一先只并行只读工具；二是API 的并发限制——同一轮里十个工具并发打第三方接口，可能触发限流，需要给线程池设上限而不是无脑 max_workers=len(calls)。这些生产化问题留到卷二。

真实模型下会怎样：假模型要并行调用，得我们在脚本里手写一个含多个 ToolCall 的列表；真实 Claude 会在一次响应里自然地发出多个 tool_use 块，尤其是当提示里出现"分别"“各自"“对比"这类词时。真实场景的并发还要考虑两个变量：一是工具是否幂等——并行的两个写操作可能互相踩踏，卷一先只并行只读工具；二是API 的并发限制——同一轮里十个工具并发打第三方接口，可能触发限流，需要给线程池设上限而不是无脑 max_workers=len(calls)。这些生产化问题留到卷二。

阅读导航：上一节：4.2 参数校验与执行隔离 · 下一节：5.1 ReAct：把「想—做—看」写进循环 。