这段时间在推进公司内部 Agent 平台的工具调用改造，最让我头疼的不是模型怎么选、提示词怎么写，而是调度层怎么把一堆乱七八糟的工具调用组织起来。任务本身不复杂，但工具多了以后，串行执行慢得让人崩溃，全部并发又会被依赖关系和下游限流反复打脸。折腾一圈下来，沉淀出的核心思路就是：把工具调用建模成 DAG，做并行编排。这篇文章就把这次工程实践里的依赖解析、扇出调度、失败传播三个关键环节完整讲一遍，适合正在做 AI Agent 工具编排、任务调度、或者准备自己撸一个 Agent 执行引擎的开发者参考。

1. 为什么工具调用会卡在串行上

1.1 一次真实的多工具调用场景复盘

先还原一个我处理过的典型任务。Agent 接到用户指令：“整理上季度华东区各品类销售数据，输出图表，并发送到部门群”。

这个任务拆解出来大概要调用这么几个工具：

• 查询销售明细数据（工具 A，依赖数据库）
• 查询组织架构与群成员列表（工具 B，依赖通讯录接口）
• 计算同比、环比指标（工具 C，依赖 A 的结果）
• 生成图表图片（工具 D，依赖 C 的结果）
• 发送消息到群（工具 E，依赖 B 的群成员信息和 D 的图片）

如果让 Agent 按顺序一个一个调用，假设每个工具平均耗时 3 到 5 秒，整个链路跑下来要 30 到 40 秒。这还不算每轮调用之间 LLM 重新决策、组织参数的额外时间。用户等 40 秒收到一张报表，体验基本属于灾难级别。

稍微分析一下就能看出来，工具 B 和工具 A 其实完全无关，B 根本不用等 A 的结果。真正有硬依赖关系的只有 A → C → D，以及 B + D → E 这条汇聚路径。如果能让 A 和 B 同时跑，这个任务的理论最小时长就是从 A 到 C 再到 D，最后 E 这一条最长链路的耗时，大概 15 秒左右，比串行快了一倍不止。

这就是我要引入 DAG 并行编排的最直接动机：工具调用之间天然存在依赖关系，而依赖关系之外的节点可以并行，但并行又不能破坏依赖约束。用图的结构把它们组织起来，是唯一能同时满足“快”和“正确”两个目标的做法。

1.2 串行、并发 map、DAG：三种方案的取舍

在决定上 DAG 之前，团队内部先比了一圈方案。最朴素的是纯串行，代码写起来最简单，一个 for 循环挨个执行，每一步的结果都手递手传给下一步。缺点显而易见，没有任何并行度，链路一长延迟就成倍放大，只适合工具数量少、延迟低的场景。

第二种是并发 map，所有工具不分青红皂白一起调。表面上看起来很快，但实际上有依赖关系的工具会因为拿不到上游数据而报错，还得靠 Agent 重新规划、补跑，反而更慢。更麻烦的是，无脑并发会瞬间把外部服务的 QPS 打满，限流问题接踵而至。

相比之下，DAG 编排把工具的调用计划建模成一张有向无环图，节点是一次工具调用实例，边是数据依赖。调度器只需要做两件事：等一个节点的所有上游完成，然后把它放入可执行队列；以及在一个节点完成时，唤醒它下游所有等待的节点。依赖关系维护得清清楚楚，无关节点自动并行，有关节点自动等待。

这里也想提醒一句，DAG 不是银弹。如果任务只有两三个工具、依赖也不复杂，引入 DAG 属于过度设计。我的判断标准很简单：工具调用数大于 5，或者存在两层以上依赖关系，DAG 带来的收益才明显。否则老老实实串行就好，简单系统的首要目标是可维护性，不是花活。

2. 依赖解析：从工具 Schema 到执行图

2.1 工具参数与数据依赖的映射关系

DAG 编排的第一步不是写调度器，而是先把依赖关系解析出来。这里的核心问题很具体：一个工具要执行，它的输入参数从哪里来？需要等待哪些上游工具的输出？

目前主流的 Agent 工具定义方式，无论是 OpenAI 的 function calling，还是 MCP 协议里的工具描述，都会给每个工具声明输入参数的 JSON Schema。这个 Schema 既定义了参数类型，也给了我们解析依赖的关键线索。

我见过三种做法。第一种是在工具定义里显式声明依赖，比如在描述里写清楚“本工具依赖 toolA.output.salesData 字段”。这样最明确，但需要所有工具开发者都遵守规范，推广成本高。

第二种是参数名自动匹配，也就是下游工具的某个参数如果和上游工具输出里的某个字段同名，就默认存在依赖。这种方式实现简单，但字段重名容易产生歧义，比如两个上游工具都返回了total，系统根本不知道该取谁。

第三种是 Schema 路径推导，根据上游输出的 JSON Schema，自动匹配下游参数对应的字段路径，比如sales_tool.output.data.by_region。这种最灵活，也最能扛住复杂场景，因为工具输出的结构本身已经通过 Schema 描述清楚了。

实际工程里我用的是第二种和第三种混合。MCP 工具的定义一般会包含 inputSchema，下游参数名如果能唯一匹配上游某个字段名，就直接建立依赖；匹配不上再走 Schema 路径推导。对于实在无法自动判断的，就在依赖解析阶段把它标记为“外部输入”，留给 Agent 在规划时填充。

下面是一个简化版的工具定义示例，可以看到chart_generator的source_data参数明显指向sales_analyzer的输出字段，依赖关系就由此建立：

{ "tool": "chart_generator", "inputSchema": { "type": "object", "properties": { "source_data": { "type": "object", "description": "来自 sales_analyzer 输出的分析结果" }, "chart_type": { "type": "string" } }, "required": ["source_data", "chart_type"] } }

2.2 建图与拓扑排序的实现要点

依赖关系解析出来后，下一步就是构建执行图。注意一个容易踩坑的细节：图的节点应该是“工具调用实例”，而不是“工具本身”。同一个工具可能在一个任务里被调用多次，输入参数不同，依赖也不同。如果按工具维度建图，两次调用会被错误地合并成一个节点，调度就全乱了。

图构建完成后，第一件事是检测环。有向无环图这个名字已经暗示了，环形依赖是非法状态，必须在运行前拦住。检测环最常用的办法就是拓扑排序，我用的是 Kahn 算法，思路非常简单：统计每个节点的入度，把入度为 0 的节点依次拿掉，拿到最后如果还有剩余节点，说明有环。

from collections import deque def kahn_topological_sort(graph): indegree = {n: 0 for n in graph.nodes} for node in graph.nodes: for neighbor in graph.edges[node]: indegree[neighbor] += 1 queue = deque([n for n in graph.nodes if indegree[n] == 0]) sorted_nodes = [] while queue: node = queue.popleft() sorted_nodes.append(node) for neighbor in graph.edges[node]: indegree[neighbor] -= 1 if indegree[neighbor] == 0: queue.append(neighbor) if len(sorted_nodes) != len(graph.nodes): cyclic = [n for n in graph.nodes if indegree[n] > 0] raise RuntimeError(f"检测到循环依赖: {cyclic}") return sorted_nodes

这里要区分两个概念：拓扑排序序列可以作为执行合法性的校验结果，但真正执行时并没有必要严格按这个顺序串行跑。正确做法是把它当成“依赖校验”，实际执行用事件驱动，节点完成一个，就触发一批可执行的下游节点。拓扑排序在这套体系里的价值是运行前的安全检查，防止带着环的错误计划进入调度器。

2.3 动态依赖：结果驱动的新节点注入

静态依赖解析能覆盖大部分场景，但 Agent 场景里有种棘手情况：某个工具在执行完以后，返回的结果里包含了一个“新的待办事项”，需要后续再调用另一个工具来处理。

举例来说，代码执行工具跑完一段脚本，发现脚本里引用了一个外部数据文件，于是需要额外调用文件下载工具。这个依赖在最初规划时并不存在，是执行过程中动态产生的。

工程上处理动态依赖有两条路。一条是在节点内部做循环，把动态调用收敛在单节点里，由 Agent 自行决定是否需要继续调用新的工具。这样做调度器完全不用感知动态变化，实现成本低，但节点的执行时间不可控，可能会长时间霸占资源。

另一条路是支持 DAG 运行时的节点注入，允许在任务运行中往图里动态添加新节点和边。这个方法灵活性高，但复杂度也会上一个台阶，尤其是已经处于 RUNNING 状态的节点，新注入的节点可能会导致部分依赖关系需要重新计算，调度器的状态管理会变得很脆弱。

从我个人的工程倾向来说，大部分业务场景用第一种就够了。把动态依赖“包住”，让它成为节点内部的一次循环决策，整个执行图保持静态稳定，调度器的复杂度就始终可控。除非你的 Agent 任务里有大量的代码执行、文件处理这类不确定行为，否则不建议轻易上动态注入。

3. 扇出调度：并发放大与流量控制

3.1 扇出与扇入的调度模型

DAG 调度器里最核心的两个原语是扇出（fan-out）和扇入（fan-in）。扇出指的是一个节点执行完成后，同时触发多个下游节点的执行。扇入指的是一个节点需要等待所有上游节点都完成后，才能开始执行。

扇出很好理解，节点 A 完成，它下游的 B、C、D 三个节点同时进入待执行队列。比较容易被忽略的是扇入的判断逻辑。一个节点有多个依赖时，不能简单地在第一个上游完成时就把节点放行，否则下游会用上残缺的数据。

正确做法是给每个节点维护一个“剩余依赖计数”，初始值等于直接前驱的数量。每收到一个上游完成事件，计数减一；当计数减到零时，才把该节点投入可执行队列。这个逻辑听起来简单，但在高并发下要注意原子性问题，这个计数器的增减必须保证线程安全，否则两个上游同时完成时，可能把本应值为 0 的计数减成了负，或者把两个完成事件串成两次放行。

在实际代码里，我倾向于让每个节点的状态通过事件总线串行处理，把所有状态变更都收敛在事件循环里，避免多线程直接改共享计数器。事件驱动带来的额外好处是，调度过程天然就是异步的，不会出现一个节点运行太久导致调度线程被卡住。

3.2 有界并发：信号量与动态任务队列

DAG 能放大并发，但并发不能无限放大。一个任务里有十几个节点，如果所有就绪节点同时执行，假设有多个任务同时跑，外部接口很快就会被打爆。我见过不止一次，DAG 一跑起来，下游数据库连接数瞬间耗尽，取而代之的是大面积的连接超时。

控制并发的核心工具是信号量。调度器维护一个全局信号量，限制整个任务当前最多同时执行的节点数。这个数值怎么定？我一般先看下游系统能接受的并发上限，再看单节点的平均执行时长，然后通过压测微调。比如某个内部数据服务最多支持 20 个并发连接，那全局信号量最多设为 16 到 18，留一点余量给其他任务的调度开销。

除了全局限制，还要针对单个外部服务做独立的限流。两个不同任务可能都调用了同一个搜索服务，各自内部的并发上限都是 8，但叠加起来对下游就是 16 路并发。所以高级一点的方案是给每个下游服务维护独立的信号量或令牌桶，调用前同时申请“全局资源”和“下游资源”，两者都拿到才真正执行。

这个做法相当于把并发控制从单任务视野提升到了全系统视野，两个任务的扇出相互叠加时，下游依然稳得住。代价是代码复杂度增加，但 Agent 平台一旦要接入大量外部工具，这个成本必须付。

3.3 调度器实现的核心逻辑

下面给一个简化版的 DAG 调度器核心代码，完整展示了扇出扇入的执行逻辑。这段代码基于 asyncio，核心数据结构是每个节点的依赖计数和就绪队列。

import asyncio class DAGNode: def __init__(self, node_id, fn, deps=()): self.id = node_id self.fn = fn self.deps = list(deps) self.remaining_deps = len(deps) self.downstream = [] self.result = None self.status = "PENDING" class DagScheduler: def __init__(self, nodes, semaphore_limit=8): self.nodes = {n.id: n for n in nodes} for n in nodes: for dep_id in n.deps: self.nodes[dep_id].downstream.append(n.id) self.semaphore = asyncio.Semaphore(semaphore_limit) self.ready_queue = asyncio.Queue() async def run(self): for n in self.nodes.values(): if n.remaining_deps == 0: await self.ready_queue.put(n.id) while not self.ready_queue.empty(): node_id = await self.ready_queue.get() node = self.nodes[node_id] if node.status != "PENDING": continue async with self.semaphore: node.status = "RUNNING" try: node.result = await node.fn() node.status = "SUCCEEDED" except Exception as e: node.status = "FAILED" node.result = e self._propagate(node) def _propagate(self, node): for child_id in node.downstream: child = self.nodes[child_id] child.remaining_deps -= 1 if child.remaining_deps == 0: self.ready_queue.put_nowait(child_id)

这段代码有几个细节值得说明。第一，节点的fn是异步函数，也就是实际的工具调用，调度器完全不关心工具内部是 HTTP 请求还是本地计算。第二，节点的成功和失败结果统一存到result字段里，下游节点在启动前可以根据上游状态决定自己要不要消费这个结果。第三，信号量包裹的是“执行阶段”而不是“等待阶段”，这是关键。如果信号量把整个节点生命周期都包住，大量已就绪节点会被信号量阻塞，导致调度器吞吐骤降。

实际线上版本还会多一点东西：节点级超时、重试封装、结果缓存、进度上报。但核心骨架就是上面这个样子，先把骨架跑通，再往里面加功能，思路会清晰很多。

4. 失败传播：Agent 场景下的降级与补偿

4.1 失败分类与传播策略

DAG 执行过程中，节点大概率会失败。外部工具调用从来不是可靠的，超时、限流、参数错误、服务端 5xx，各种异常每天都能遇到。设计失败处理的第一步，是给失败分类。

我把失败分成三类。第一类是瞬时错误，比如网络抖动导致的超时、限流返回的 429，这类错误值得重试。第二类是稳定错误，比如鉴权失败、参数格式非法、工具不存在，重试一百次结果都一样，必须立即失败。第三类是软失败，工具调用本身返回成功，但返回的数据质量不正常，比如字段缺失、结果为空、置信度低于阈值，这种最隐蔽，需要业务侧自定义校验逻辑才能发现。

分好类以后，要确定失败传播策略。Agent 场景里我主要在两个策略之间做选择。FailFast 指的是任一节点失败，整个 DAG 立即取消，所有未完成的节点标记为 SKIPPED，已经完成的节点结果作废。这种策略适合强一致性的任务，比如支付流程、订单处理、写操作链路。

FailSafe 指的是失败节点不影响其他节点继续执行，失败结果传给下游，下游可以选择降级处理或者跳过，最终任务还是能给到一个结果，只是会带上警告信息。这种策略适合分析、报表、内容生成类任务。我个人的倾向是，Agent 应用默认 FailSafe，因为 Agent 的价值恰恰在于能在一个不完美的执行结果之上继续组织信息，给出一个“至少有点用”的答案，而不是遇到一个工具失败就整个任务崩掉。

4.2 超时、重试与幂等控制

超时设置是失败传播的第一道防线。我给每个工具调用设置了两个超时：单次调用超时和整个 DAG 总超时。单次调用超时根据工具的典型响应时间设定，比如查询类工具给 10 秒，代码执行类工具给 30 秒。整个 DAG 总超时通常设为链路中各节点单次超时之和的 60% 到 70%，如果超过这个时间还没完成，大概率是调度出了问题，直接按失败处理。

重试不是所有失败节点都适用的。瞬时错误可以重试，稳定错误绝对不能重试。读写操作也得分清，查询类工具重试基本安全，写操作类工具重试前必须确认是否具备幂等条件。

这里有一个我在实践中踩过的深坑：外部工具可能已经执行成功，但响应在网络上丢失了，超时被触发后调度器重试，导致同一个操作被执行了两次。要解决这个问题，必须在调用工具时携带幂等键，比如 requestId，外部系统根据幂等键判断是否已经处理过相同请求。调用方也必须把幂等键记录在节点状态里，重试时复用同一个键，不能每次生成新的。

重试策略我采用指数退避加随机抖动，第一次重试等 1 秒，第二次 2 秒，第三次 4 秒，每次加上不超过 200 毫秒的随机偏移。这么设计是为了防止 N 个失败节点同时重试，在同一时刻对下游发起一波排队请求。

4.3 部分成功结果的聚合与兜底

FailSafe 策略下，DAG 结束后往往不是全红也不是全绿，而是中间灰蒙蒙一片。比如报表生成任务里，销售数据查询成功了，但图表生成工具因为临时故障失败了，最后 Agent 该怎么向用户交代？

我的做法是，调度器输出的不只是一个最终结果，而是一棵完整的结果树。每个节点的状态、结果值、错误信息、耗时时长全部保留。Agent 拿到这棵树以后，把关键信息组装成上下文，交给 LLM 生成最终答复。如果某个节点失败了，LLM 可以明确告诉用户：图表生成失败了，原因是服务超时，销售数据是完整的，我可以先用表格形式展示。

这就是 Agent 场景和传统数据处理流水线最大的区别。失败信息不是简单地抛给上层调用方，而是作为语境的一部分进入模型生成环节。调度的职责是尽量把任务做完，模型的能力体现在把做得不完美的结果讲清楚。两者要配合好，调度器就必须把失败细节原样暴露出来，而不是吞掉。

兜底方面，我会给每个重要节点配置一个降级策略。比如数据查询失败时，可以降级为读取上一周期的缓存数据；图表生成失败时，降级为输出 CSV 附件。降级结果会标记为 WARNING 状态，这样最终结果里能看到来源，不至于让用户误以为降级数据是新鲜数据。

5. 实战中的五个坑与排查实录

5.1 循环依赖与环检测

第一个坑是循环依赖。之前说过建图后要立刻做环检测，但实际开发里还是会在运行时遇到环，问题往往出在依赖解析阶段没有处理好自引用。

最常见的场景是 LLM 生成的工具调用计划里出现了 A 依赖 B、B 又依赖 A 的情况。这通常不是业务真的需要循环，而是模型把参数占位符写错了，导致下游工具错误地等待了一个实际上还没执行的节点。

处理经验是：对话模型生成的计划在进入调度器之前，必须先做一次严格校验，环检测不过的图直接返回给 Agent，让它重新规划，而不是试图在调度器内部解决环。调度器只负责执行合法的 DAG，不负责修正模型的错误。拓扑排序检测出环后，我会把环上的节点 ID 完整打印出来，输出到日志里，这个信息对定位模型规划错误特别有用。

5.2 任务泄漏与悬挂状态

第二个坑是任务悬挂。节点执行完成了，但调度器一直收不到完成事件，导致所有依赖它的下游节点永远停在等待状态。

导致悬挂的原因五花八门。最常见的是回调丢失，比如工具调用通过消息队列异步完成，但 ack 失败导致回调消息被重复消费或者丢弃。还有一种是节点执行过程中出现异常，代码没有走到 complete 流程，直接返回了，调度器对它的印象就停留在 RUNNING。

我最后是加了一个 watchdog 机制解决的。调度器启动一个定时任务，周期性检查所有 RUNNING 状态的节点，任何一个节点的执行时长超过设定阈值，就强制标记为超时失败，并按失败传播策略处理。这个机制不一定能挽回已经丢失的回调，但至少能保证整个 DAG 不会因为一个节点的异常而永久卡死。

5.3 并发穿透与下游限流

第三个坑是并发控制失效。前面提到过信号量控制并发，但信号量只在单个调度器实例内部有效。如果你部署了多个调度器副本，或者多个 Agent 任务共用同一个外部服务，下游仍然可能被打穿。

我见过一个现场：内部某个搜索服务的 QPS 上限是 100，监控面板显示调用方单任务并发 8，看起来很安全。但同一时间有 20 个 Agent 任务在跑，实际打过去的 QPS 是 160，直接触发限流，大量请求返回 429。

解决办法是把限流器从调度器里独立出去，部署成集中式服务，每个下游目标一个令牌桶，全局共享。调度器在调用外部工具前向限流服务申请配额，拿到了才真正发起请求。虽然多了一次网络调用，但稳定性提升立竿见影，再也没有因为并发穿透把下游打挂过。

5.4 状态一致性：从执行图到结果树

第四个坑是执行图的状态被中途修改。刚开始我也是直接在图节点上记录执行结果，后来发现重试机制会导致同一个节点的结果被覆盖，而下游节点可能早就读取了旧结果。

正确的做法是执行图和结果树分离。执行图是静态计划，节点状态、依赖关系在运行前就固定下来，运行中不修改。结果树是执行后的数据，每个节点的成功结果、失败信息、重试次数、执行耗时都单独维护。下游依赖通过结果树查找上游数据，而不是从图节点上取。

状态机设计也很关键。我把节点状态机定义为 PENDING、READY、RUNNING、SUCCEEDED、FAILED、SKIPPED、TIMED_OUT 七种。每次状态转移都要带时间戳记录到日志里，这样出问题时，可以完整还原每个节点的生命周期，定位到底是调度延迟还是执行延迟。

5.5 排查工具与观测手段

最后一个建议是给 DAG 编排链路做足够的观测。我先说一个最实用的工具：可视化。把 DAG 的执行计划渲染成一张图，环在哪里、哪个节点扇出最多、哪个节点迟迟不完成，一眼就能看出来。前端用 dagre 或者简单的 Graphviz 都可以，关键是这张图要能展示实时状态，而不是只能看静态计划。

日志和指标方面，我给每个 DAG 实例生成一个 traceId，节点每次状态变化都打印日志，包含节点 ID、traceId、时间戳、当前依赖计数。指标上重点统计两类耗时：每个节点的等待时间和执行时间。等待时间异常长，说明调度器阻塞或者上游依赖卡住；执行时间异常长，说明工具本身有问题。

这些观测手段看似繁琐，但在排查线上问题时能省下大量时间。Agent 场景的失败本来就千奇百怪，没有可观测性就像蒙着眼睛修车，只能靠猜。

6. 关于这套编排方案的个人体会

最后说几点反直觉的经验。

第一，别为了并行而并行。DAG 并行编排解决的是“有依赖关系的工具调用提速”问题，不是所有调用都适合并行。工具间没有依赖、且串行本来就能在 3 秒内完成的任务，强行上 DAG 只会增加调度开销和一个不必要出错的环节。我在项目里定了个规矩：少于 5 个节点、链路里没有明显延迟瓶颈的任务，直接用简单串行执行器，不引入 DAG。

第二，调度器永远是系统里最需要“保守设计”的部分。外部工具调用可以出各种幺蛾子，但调度器自己不能跟着一起摆烂。该捕获的异常要捕获，该兜底的超时机制要兜底，该记的日志要记全。我把这次改造的调度器当作一个公共服务来设计，而不是嵌在某个 Agent 里的临时脚本，心态不同，代码质量完全不同。

第三，失败传播的策略一定要和业务场景对齐。我之前踩过的一个坑是，全站统一用了 FailFast，结果内容生成类任务只要有一个数据源失败，整个报告就出不来，用户反馈比失败本身还严重。后来改成 FailSafe 加显式地失败标注，产品体验立刻好转。策略没有绝对地对错，只有和场景是否匹配。

这套 DAG 编排方案目前已经在我这边的 Agent 平台上稳定跑了一段时间，收益最明显的是凡是有多工具协同的任务，延迟普遍下降了 60% 以上，而且工具调用失败时用户的感知从“整个任务失败”变成了“局部数据缺失，但结果仍然可用”。如果你也在做 Agent 工具调度，建议先从一个只有五六个节点的真实任务开始，把依赖解析、扇出调度、失败传播这三条主链路跑通，再加高级功能。DAG 本身不复杂，复杂的是把异常场景一个个处理干净。