并行工具调用：Agent提速的并行术

可视化演示

展示并行工具调用三大机制：DAG依赖图识别可并行工具、串行vs并行时间对比、以及异常隔离与部分失败容错处理策略。

并行基础：总时延从“求和”到“取最大”

餐厅里如果只有一位厨师，晚上八点同时进来三桌客人，他只能一桌一桌做。这在信息技术里叫“串行”——一次只做一件事。坏处显而易见：总耗时等于每道菜耗时之和。Agent 也一样：一个请求要依次调用“天气查询→航班查询→酒店比价→门票预订”四个工具时，一次问答的时延就是四段网络往返的简单叠加，用户等得心焦。

主厨很快发现，凉菜和烤肉互不干扰，完全可以同时做。放到 Agent 里，工具调用之间也分两种：有依赖的（后一个需要前一个的输出）和无依赖的（各问各的）。无依赖的一组调用，可以组成一个“批次”同时发出。这就是并行工具调用的出发点：把等待时间重叠起来。

串行总时延是累加，并行总时延是取最大值。前提是这些调用彼此独立、互不读取对方的中间结果。实际收益受“最慢的那个工具”限制——就像一桌菜里烤鸭要四十分钟，凉菜再快也得等烤鸭上桌才算上齐。

回到餐厅：主厨拿到订单的第一步，不是埋头做菜，而是先判断哪些菜能“同时下锅”。能同时下锅的越多，整桌上齐越快；但只要有一道炖四十分钟的硬菜，整桌就快不过它。并行术的第一课，就是认清“总和”与“最慢”之间的天壤之别。

依赖分析：先做哪道菜，心里要有谱

不是所有菜都能同时做。熬高汤的汤锅台要先于借用高汤的热炒台开工；烤肉要先腌制两小时。主厨心里那张“先后关系图”，就是 Agent 的依赖图。工具 A 的输出是工具 B 的输入时，B 必须等 A 完成，强行并行只会让 B 拿到空参数而报错。

依赖图用有向边表达“谁等谁”：节点是工具，边表示数据流向。有了这张图，还要做拓扑排序——给每个工具排一个合法执行顺序，保证被依赖方永远排在依赖方前面。排序之后，同一层的工具互不依赖，天然适合放进同一个并行批次。

排序过程中一旦发现环路——A 需要 B、B 又需要 A——就说明任务描述自相矛盾，没有合法执行顺序。工程上的处理是直接报错或人工介入拆环，而不是让它进入“互相死等”的空转。分层分组的技巧在于：层号越小越先执行，同一层并行。

主厨心里没有这张“先后图”，就会把没腌好的肉塞进烤箱。Agent 心里没有依赖图，就会把需要航班号的余票查询和航班查询一起发出去，然后收到一堆“参数缺失”的报错。依赖分析，是并行术的地基。

并行识别：找出能同时下锅的菜

高汤没熬好，热炒台就闲着吗？不——主厨会先把不需要高汤的菜安排给热炒台。把工具分成“现在就能做”和“必须等别人”两组，这个筛选动作，就是并行识别。

识别的方法很直观：检查工具两两之间有没有依赖路径。所谓路径，不光是“直接依赖”，还包括“间接依赖”——A 的输出喂给 B、B 的输出喂给 C，那么 A 和 C 之间就存在一条可达路径，C 不能和 A 并行。只有两个工具双向都不可达，它们才是真正意义上的无依赖，可以安全地放进同一批次。

工程上，这一步通常在调用前做静态预分析：把整个工具调用图扫一遍，标记出所有无依赖对。参数独立性是另一个必要条件——即使两个工具在依赖图上没连边，如果它们读写同一个共享状态（比如同时写同一个用户缓存），也要小心并发冲突。预分析的结果就是一张“可并行矩阵”，运行时直接查表分组，省去每次调度的重复计算。

主厨扫一眼菜单，就能报出“这三道可以同时做，那两道得等汤好”。Agent 的并行识别做得也是同样的事：在动手前把“能同时下锅的菜”圈出来，剩下的任务自然沉淀到后面的批次。识别得越准，时间重叠得越多。

并发控制：灶台有限，排队等候

厨房里灶台只有六个。同时开工的任务超过六个，帮厨互相撞车、煤气不够、出菜口堆满盘子。主厨定下的“最多同时开 N 个灶”，就是并发度上限。Agent 调用外部 API 同样有隐形的灶台：QPS 限制、连接池大小、下游服务的负载能力。

实现并发控制最常用的工具是信号量（Semaphore）：一个计数器记录当前正在执行的工具数，达到上限时，新任务进入等待队列，等某个工具完成后让出一个“灶位”再放行。排队策略可以简单粗暴（先来先服务），也可以带优先级（关键路径上的工具插队）。

资源预算同样重要：内存、网络连接、第三方额度都是有限的。总预算除以单次调用开销，就能估算出一个安全的并发上限——宁可在队列里多等 50 毫秒，也不要一次性把下游打挂。并发度为 k 时，总耗时存在一个理论下界，它由“最慢工具”和“总工作量均摊”共同决定。

主厨不会因为客人催单就把二十道菜同时丢进六个灶台——那只会得到一片焦糊和更大的混乱。Agent 也一样：并发不是越多越好，而是“在资源允许的范围内，让每一个灶台都不闲着”。排队不是退步，而是对整体稳定性的保护。

结果合并：出菜口按桌号摆盘

各工位完工时间不同——烤箱的烤肉先出炉，凉菜反而后做好？出菜口不能谁先到谁上桌，必须按桌号、按菜单顺序摆盘。Agent 的并行调用也一样：先返回的未必是第一个请求的结果，必须按原始序号重新排列，否则就会把“机票价格”填进“天气温度”的位置。

这就是乱序完成问题。并行发出四个请求后，它们各自走各自的网络路径，完成顺序完全是“谁先到家谁先敲门”。正确处理方式是：给每个工具调用挂一个序号或请求 ID，返回时按 ID 放进对应的格子，而不是按到达时间的先后追加进数组。

合并阶段还要做去重与聚合：多个工具可能返回重复数据（比如两次检索命中同一篇文章），先按主键去重；然后按业务规则聚合排序，比如把机票、酒店、景点按总预算排成一张比价表。最终交付给调用方的结果，必须保持与发出调用时一致的顺序，这叫顺序一致性——让上层代码感觉不到并行的存在，只享受并行的速度。

主厨在出菜口做的事，从来不是“谁先出锅谁上桌”，而是“每一桌的菜按菜单顺序齐活”。结果合并就是 Agent 的出菜口：外面看起来秩序井然，里面早已千帆竞发。

异常隔离：一道菜失败，不连坐整桌

蒸锅台的火候没控制好，一条鱼蒸老了。主厨不会把整桌菜都倒掉——他会给这桌客人补一道菜，其他菜照常上。厨房里任何工位的失误，都不能连坐整桌。这就是异常隔离：并行调用中，单个工具失败不应该拖垮整个批次。

现实中最常见的故障是慢和挂：某个第三方 API 迟迟不返回，把整个批次的完成时刻无限拉长。对策是超时熔断——为每个工具设定等待上限 \(\tau\)，超时就掐断这次调用并标记失败；连续失败多次后直接熔断，短时间内不再请求该工具，防止它继续拖累整体。降级补偿则是失败后的兜底：用缓存数据、默认值或占位文案填上失败的位置。

部分失败是并行场景的常态。四个调用成功三个，我们就返回三个真实结果加一个降级结果，而不是把整批结果全部作废。比起串行模式“一个出错全链报废”，并行加隔离让系统有了优雅降级的能力——用户至少拿到大部分答案，而不是一句冷冰冰的“服务错误”。

主厨的哲学是：一道菜失误是局部事故，整桌餐是整体承诺。异常隔离让 Agent 在部分工具失灵时依然能“把菜上齐”——用降级替代失败，用超时替代无限等待。这种韧性，正是生产环境里并行调用比串行调用更难也更有价值的地方。

工程实践：把并行术跑进生产

晚高峰的餐厅，主厨把这套“同时开工、出菜口统一摆盘、单菜失败不连坐”的流程跑顺之后，翻台率翻了一倍。实时 Agent 也一样：把并行术嵌进架构，让用户产生“秒回”的体验。我们来盘点三个最典型的落地场景。

第一个是实时问答。语音助手被问到“明天去上海出差要带伞吗？顺便订最早航班”，串行查询日历、天气、航班三个工具要 3 秒以上，用户早就打断重问了；并行发出后 1.2 秒内全部返回，合并成一句话回答。第二个是多源数据聚合查询：旅行规划 Agent 拿到“杭州 3 日游”请求，同时拉取航班、酒店、景点、天气四个渠道，按“总预算”统一排序输出比价表。第三个是并行信息检索（RAG 多路召回）：同步检索 Elasticsearch、向量库、Wiki API 三路，去重后交给 LLM 生成答案；某一检索源超时自动降级为两路结果，整体体验不受影响。

衡量这套并行术的价值，最直接的指标是加速比——改造前的串行耗时除以改造后的并行耗时。工程上的经验值通常在 2 到 5 倍之间，具体取决于无依赖工具的数量和耗时分布。别忘了配合可观测性：把每个工具的耗时、失败率、排队时间都记下来，持续优化“最慢的那道菜”。

餐厅的晚高峰不会消失，但主厨的调度术让每一桌都能按时上菜。Agent 的请求洪峰也不会消失，但并行工具调用让每一次问答都快人一步。把这张“并行术”的流程图挂在架构图上，你的 Agent 就有了从“能用”到“好用”的底气。

AI 应用场景

Python 演示

import timefrom concurrent.futures import ThreadPoolExecutor, as_completed  def fetch_weather(city):    time.sleep(0.5)  # 模拟 0.5s 网络耗时    return "sunny 25C" def fetch_flight(city):    time.sleep(1.2)  # 模拟 1.2s 网络耗时    return "flight avg 1280 CNY" def fetch_hotel(city):    time.sleep(0.8)  # 模拟 0.8s 网络耗时    return "hotel min 320 CNY" def fetch_sight(city):    time.sleep(0.4)  # 模拟 0.4s 网络耗时    return "sights: West Lake, Lingyin" TOOLS = {    "weather": fetch_weather,    "flight": fetch_flight,    "hotel": fetch_hotel,    "sight": fetch_sight,}  def call_sequential(city, tool_names):    results = {}    t0 = time.time()    # 串行：一个接一个调用，总耗时是各工具耗时之和    for name in tool_names:        results[name] = TOOLS[name](city)    return results, time.time() - t0  def call_parallel(city, tool_names, max_workers=4):    results = {}    t0 = time.time()    # 线程池：最多 max_workers 个工具同时执行（对应并发度 k）    with ThreadPoolExecutor(max_workers=max_workers) as pool:        # 同时提交所有无依赖的工具调用，future 上挂上工具名        futures = {pool.submit(TOOLS[name], city): name for name in tool_names}        # 谁先完成谁先被处理，完成顺序与提交顺序无关        for fut in as_completed(futures):            name = futures[fut]            try:                results[name] = fut.result()            except Exception as exc:                # 异常隔离：单个失败只记录该工具，不影响其他结果                results[name] = "failed: " + str(exc)    return results, time.time() - t0  def merge_by_order(results, tool_names):    # 结果合并：按原始调用顺序重排，而不是按完成顺序    parts = []    for name in tool_names:        parts.append(name + "=" + results[name])    return " | ".join(parts)  if __name__ == "__main__":    city = "Hangzhou"    tool_names = ["weather", "flight", "hotel", "sight"]    seq_res, seq_cost = call_sequential(city, tool_names)    par_res, par_cost = call_parallel(city, tool_names)    print("sequential:", merge_by_order(seq_res, tool_names))    print("parallel:  ", merge_by_order(par_res, tool_names))    print("seq cost %.2fs vs par cost %.2fs" % (seq_cost, par_cost))    print("speedup: %.2fx" % (seq_cost / par_cost))

1. with ThreadPoolExecutor(max_workers=max_workers) as pool:创建“灶台”线程池，同时最多跑 max_workers 个工具，这就是并发度上限。
2. futures = {pool.submit(TOOLS[name], city): name for name in tool_names}把四个无依赖的工具一次性提交进线程池，每个 future 挂上自己的工具名，对应并行识别后的“同批派遣”。
3. for fut in as_completed(futures):谁先完成谁先被收割，直观体现“乱序完成”——完成顺序与提交顺序无关。
4. results[name] = fut.result()按 future 上挂的工具名把结果放进对应格子，天然实现“按名归位”。

常见误区

面试题