《AI Agent 工程实战》8.1 引擎内并发：并行工具与并行子 Agent

8.1 引擎内并发：并行工具与并行子 Agent

第 7 章把 run 交给后台 worker 之后，单个 run 内部还是「一步接一步」地跑。但 ResearchCopilot 的一次研究很少是线性的：用户问「对比 A 和 B 两个方案的取舍」，Agent 往往要同时查 A、查 B、查两者的评测数据；一个复杂问题还会被拆成「找证据」「查反例」「算指标」几个子任务。这些工作天然可以重叠——串着跑就是在浪费用户的时间。

本节把 ResearchCopilot 推进到「一轮内真并发」：把卷一 4.3 的线程池调度升级成 asyncio.TaskGroup 版本，演示并行工具与并行子 Agent 两种形状，用带 asyncio.sleep 的假工具实测串行与并行的墙钟差，并划清哪些环节能并发、哪些必须串行。

8.1.1 为什么 Agent 天生并发

卷一 4.3 已经讲过：模型一轮返回的 ModelResponse.message.tool_calls 是一个列表，里面每一项都是独立的 ToolCall。当 stop_reason == "tool_use" 时，列表里有几项就该并发跑几项。当时用的是 concurrent.futures.ThreadPoolExecutor——工具是同步函数（requests、sqlite3）时它最省事。

但到了卷二，工具链已经全面异步化：检索走 httpx.AsyncClient，缓存走 redis.asyncio，数据库走 psycopg 的异步连接。同步工具用线程池，异步工具就该用 asyncio 原生并发。混用会带来「在线程里再开事件循环」的麻烦，所以卷二统一把调度层切到 asyncio。

切过来之后，并发原语也从「线程池 + future」换成了 asyncio.TaskGroup。选它的理由卷一的 14.2 已经论证过：它把「一组任务必须同生共死」写进了语言机制，作用域退出前保证组内任务全部结束，一个任务失败就取消其余，不会有孤儿任务，每个任务的 finally 都会执行。这一节直接沿用这个结论，不再重复论证。

三个容易混淆的词，先摆清楚，后面所有讨论都按这个口径：

严格说，asyncio 单线程里的「并发」是协作式的：任务只在 await 处让出控制权，同一时刻仍只有一个协程在跑。但在 IO 密集的 Agent 场景里，瓶颈是「等网络 / 等数据库」，协作式并发足以让这些等待重叠，收益和真并行几乎一样——这正是它比线程池更适合本卷的原因。

8.1.2 用 TaskGroup 重写并行工具调用

先看调度函数。它接收一轮的所有 ToolCall，并发执行，把每个结果回填到调用列表的对应位置：

import asyncio

async def run_parallel(calls, exec_tool):
    """并发执行一轮工具调用，结果按 calls 的顺序返回。"""
    results: list[str | None] = [None] * len(calls)   # 按调用顺序预分配槽位

    async def one(i, tc):
        results[i] = await exec_tool(tc)              # 回填到「调用位置」而非「完成位置」

    async with asyncio.TaskGroup() as tg:
        for i, tc in enumerate(calls):
            tg.create_task(one(i, tc))
    return results

这里最关键的是回填策略：results 是一个长度等于调用数的列表，每个任务拿到自己的下标 i，把结果写回 results[i]。这样无论哪个工具先跑完、哪个后跑完，最终返回的列表永远与 calls 的顺序一一对应。

反过来，如果写成「谁先跑完就先 append」，就会踩到并发最经典的坑：完成顺序 ≠ 提交顺序。慢的那个工具结果会被排到最后，模型下一轮看到的「请求 X → 结果 Y」就对不上号了。卷一 4.3 用 futs[tc.id].result() 按 id 取回也是同一个道理，只是这里用下标更直接。

回填完之后，工具结果还要变成 role="tool" 的消息挂回对话，tool_call_id 指回对应的请求：

for tc, out in zip(calls, results):
    messages.append(Message(role="tool", content=out, tool_call_id=tc.id))

zip(calls, results) 能正确对齐的前提，正是上一步「按顺序回填」——两者是配套的。顺序一旦错位，zip 会把 A 的结果挂到 B 的 tool_call_id 上，模型就会拿到张冠李戴的证据。

把 run_parallel 接回主循环，就是一个能端到端的版本。下面用假模型脚本发出一轮三个 tool_call，再让它收尾：

async def run(model, user_input):
    messages = [("user", user_input)]
    for step in range(8):
        text, calls = await model.complete(messages)
        if not calls:
            return text                      # 没有 tool_call 就收敛
        outs = await run_parallel(calls, exec_tool)
        for c, o in zip(calls, outs):
            messages.append(("tool", c.id, o))
    return "(max steps)"

真跑轨迹（三个工具耗时 0.4s / 0.5s / 0.3s）：

  [assistant] calls=c1,c2,c3
  [tools] 0.50s -> ['result(search_corpus)', 'result(query_pg)', 'result(fetch_url)']
  [final] 三个来源都已检索完毕  总耗时 0.50s

轨迹里 calls=c1,c2,c3 是一次 assistant 消息发出的三个调用，[tools] 0.50s 说明这三个调用是并发跑的（串行会是 1.2s），返回列表的顺序仍是 c1,c2,c3。最后 assistant 收到三条 tool 消息后给出终答，整轮 0.50s 结束。

8.1.3 真跑：串行与并行的墙钟差

用一个带 asyncio.sleep 的假工具来量。三个工具的耗时分别是 0.4s、0.5s、0.3s，模拟「查语料」「查 Postgres」「抓网页」：

import time

async def slow_tool(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return name

async def run_serial(calls):
    t0 = time.perf_counter()
    out = [await slow_tool(n, d) for n, d in calls]
    return out, time.perf_counter() - t0

真跑（Python 3.14.6，time.perf_counter 计时）：

串行: ['search_corpus', 'query_pg', 'fetch_url'] 1.20s
并行: ['search_corpus', 'query_pg', 'fetch_url'] 0.50s
加速比: 2.4x  顺序一致: True

读这份数字：三个工具的耗时之和是 0.4 + 0.5 + 0.3 = 1.2s，串行实测 1.20s，完全吻合；并行实测 0.50s，等于其中最慢的那个（0.5s）——因为它们完全重叠了。加速比 2.4x 不到 3x，是因为三个工具耗时不等，并行的下界由最慢的那个决定，而不是平均值。这正是并发的收益上限：总耗时 = max(单个耗时)，而不是 sum(单个耗时)。

「顺序一致: True」是 8.1.2 那个回填策略的直接验证——并行跑完，结果顺序仍与调用顺序一致。如果这里打印出 False，就说明有人写了「按完成顺序 append」。

8.1.4 与卷一 4.3 的差异

同样是「一轮多工具并发」，卷一的线程池版和本节的 TaskGroup 版有几处本质区别，值得对照：

一句话概括：线程池解决「同步代码怎么并发」，TaskGroup 解决「异步代码怎么并发且不出孤儿」。卷二既然已经把工具链异步化，调度层就跟着切到 TaskGroup，这样第 8.3 节的超时与取消才有统一的落点。

8.1.5 并行子 Agent

比并行工具更高一层的是并行子 Agent。当一个复杂问题能被拆成互不依赖的子任务时，让每个子任务各自跑一条完整的 Agent 循环（多跳检索、多轮工具），再把它们的结论汇总，往往比「一个 Agent 串着做」快得多。

子 Agent 之间共享同一套调度形状，只是每个子任务内部可能有好几跳：

async def sub_agent(task: str, hops: int) -> str:
    t0 = time.perf_counter()
    for _ in range(hops):                 # 每个子 Agent 内部是多跳
        await asyncio.sleep(0.3)
    return f"{task} (hops={hops}, {time.perf_counter() - t0:.2f}s)"

async def fan_out(subtasks):
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(sub_agent(name, hops)) for name, hops in subtasks]
    return [t.result() for t in tasks]     # 按提交顺序取结果

三个子 Agent 分别是 2 跳、3 跳、1 跳（各跳 0.3s），真跑：

子Agent: ['A (hops=2, 0.60s)', 'B (hops=3, 0.91s)', 'C (hops=1, 0.30s)'] 总 0.91s

串行的话总耗时会是 0.60 + 0.91 + 0.30 = 1.81s；并行实测 0.91s，正好等于最长的那个子 Agent。子 Agent 并行的语义很清晰：总耗时 = 最慢子任务，其余子任务的时间被完全藏在了它后面。

注意 tg.create_task 返回的 Task 被收进 tasks 列表，退出 async with 后所有任务都已结束，此时 t.result() 立刻可用，不会阻塞。这是 TaskGroup 的一个便利：作用域边界就是天然的「全部完成」同步点。

汇总阶段通常把各子 Agent 的结论拼成一段材料，交给「主编」Agent 生成最终报告：

def merge_subresults(subtasks, results):
    parts = [f"### 子任务：{name}\n{out}" for (name, _), out in zip(subtasks, results)]
    return "\n\n".join(parts)             # 按提交顺序拼接，结果稳定可复现

用 zip(subtasks, results) 而不是「遍历结果列表」，同样是保证顺序——结果列表的顺序由 tasks 列表决定，而 tasks 是按 subtasks 顺序建的，所以两边天然对齐。可复现性对 Agent 尤其重要：同样的输入要产出同样的拼接顺序，否则第 12 章的评测集就没法做确定性断言。

这里有个必须想清楚的问题：子 Agent 各自会写自己的对话历史，如果它们共享一个可变的消息列表或工具结果缓存，并发写就会互相踩踏。所以并行的子 Agent 必须各自持有独立的状态，只在最后汇总时合并结论——这是下面那张表里「共享可变状态」一行的由来。

8.1.6 失败传播：一个工具失败会怎样

并发跑多个工具时，最该提前想清楚的是其中一个失败了会怎样。TaskGroup 的行为是确定的：一个任务抛出异常（非 CancelledError），组会立刻取消其余还在跑的任务，然后等它们退出作用域，最后把异常包成 ExceptionGroup 抛出。用一个「跑 5s 的慢工具」加一个「0.1s 后抛 403」的工具验证：

async def boom():
    await asyncio.sleep(0.1)
    raise RuntimeError("tool 403")

async def run3():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(slow_tool("ok_tool", 5.0, log))   # 还在跑
        tg.create_task(boom())                            # 0.1s 后失败

真跑输出：

[3] 组内异常 RuntimeError: tool 403
    轨迹 ['ok_tool: 启动', 'ok_tool: 被取消']

读这两行：ok_tool 已经启动（说明并发是真的），但 boom 抛异常后它被立即取消（轨迹里出现「被取消」而不是「完成」）——它没有傻等 5 秒。这正是「快速失败」：确定的错误要尽早止损，而不是让整轮陪跑。

对 Agent 来说这个行为要小心使用。不可重试的错误（权限 403、参数非法）让整组取消是对的；但单个工具的偶发错误不该炸掉整轮——卷一的约定是工具执行出错由 ToolRegistry.call 包成 {"error": ...} JSON 回灌，不抛异常，所以正常路径下 TaskGroup 不会因为工具报错而取消同伴。真正会触发取消的是「调度层自己抛的异常」（比如结果回填时越界），那是 bug，就该快速失败。

8.1.7 并发安全边界表

并发不是「能并就并」。哪些环节可以并发、哪些必须串行，取决于它们之间有没有数据依赖、有没有共享的可变状态、是不是幂等：

这张表可以浓缩成一条默认策略：只读的、独立的、各自持有状态的，才并发；写操作、有依赖的、共享可变状态的，一律串行或加锁。卷一的 minagent 默认「只对只读工具并行」，卷二把这个默认沿用下来，并在需要并发写的地方用 asyncio.Lock 或原子操作兜底。

值得强调的是最后一行「同一租户的配额扣减」。第 6.3 节的 token 配额是「先读当前用量、加一、写回」——这个读-改-写如果并发执行就会丢更新。它不该用「并发」去优化，而该用 Redis 的原子命令（INCR）或数据库的行锁来做。并发只解决「等待重叠」，不解决「竞争条件」——把两者混为一谈是线上事故的常见源头。

8.1.8 小结与验收

本节把 run 内部的调度从串行升级成了真并发，两条形状都验证过：

• 并行工具：一轮多个 ToolCall 用 TaskGroup 并发，结果按调用顺序回填，实测加速比 2.4x。
• 并行子 Agent：多个子任务各自跑多跳循环再汇总，总耗时等于最慢子任务（0.91s vs 串行 1.81s）。
• 失败传播：一个任务抛异常即取消其余，验证了「快速失败」。
• 安全边界：只读、独立、各持状态的才并发；写、依赖、共享可变状态的一律串行。

验收清单：

• 一轮多工具的结果顺序与 calls 一致（不是完成顺序）
• 并行的工具之间没有共享可变状态
• 有数据依赖的调用被拆成了两轮，而不是塞进同一个 TaskGroup
• 并发写的地方有 asyncio.Lock 或原子操作，而不是靠「应该不会同时」的侥幸

真实环境 / 真实模型下会怎样：本节的工具是 asyncio.sleep 假装的，重叠是确定性的。真实工具的耗时抖动很大——一个检索可能 0.2s 也可能 3s，并行总耗时由最慢的那个决定，所以「并行一定快」并不成立：如果最慢的工具占了整轮 90% 的时间，并行其余几个收益有限。真实模型还会在一轮里发出十几个工具调用，无限制并发会直接打爆下游（模型的并发限制、数据库连接池、第三方 API 的 QPS），这正是下一节「背压与资源池」要处理的问题。此外，真实子 Agent 之间若共享一个 Redis 缓存或同一个租户的配额，并发写就会引入本节表里那些「不能」的场景，需要在实现层显式加锁。

阅读导航：上一节：7.3 定时与延迟任务：周期语料更新与日报 · 下一节：8.2 背压与资源池：信号量限并发 。