这段时间在推进公司内部 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_limit8): 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 本身不复杂复杂的是把异常场景一个个处理干净。
这段时间在推进公司内部 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_limit8): 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 本身不复杂复杂的是把异常场景一个个处理干净。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
