Agent 编排中的依赖图解析DAG 节点的前置条件与重试链一、Agent A 要等 Agent B 的结果但 B 在等 C 的结果C 超时了——整个编排死锁Agent 编排的核心难点不是单个 Agent 的执行是多 Agent 之间的依赖管理。一个典型的复杂任务用户请求分析竞品 A 的财报并生成对比报告。这个任务需要拆成多个子任务数据收集 Agent → 数据分析 Agent → 报告生成 Agent → 图表生成 Agent。后两个 Agent 都依赖前两个 Agent 的输出但彼此之间不依赖——可以并行执行。这就形成了有向无环图DAG。DAG 的每个节点是一个 Agent 的执行单元边表示依赖关系。编排引擎需要做三件事解析 DAG 确定执行顺序拓扑排序、检测循环依赖这图不能有环、处理节点失败的重试和降级。当前大部分 Agent 框架的编排是线性链——Agent A → Agent B → Agent C串联执行。这在简单场景够用但复杂的多 Agent 场景需要真正的 DAG 执行引擎。线性链的致命问题是B 和 C 都依赖 A 的输出——但 B 和 C 之间互相不依赖它们完全可以并行执行。线性链把并行变成了串行浪费了大量时间。二、底层机制与原理剖析DAG 解析的四个关键组件拓扑排序Topological Sort给定一个有向无环图确定所有节点的执行顺序。标准 Kahn 算法每次选择入度为 0 的节点执行执行完成后将其从图中移除它的后继节点入度 -1。入度为 0 的节点之间没有依赖关系——这批节点可以并行执行。并行执行批次Wave拓扑排序的结果可以被分组为波次——同一波次内的节点没有相互依赖可以并行执行。第一批入度为 0→ 第二批第一批完成后入度变为 0 的节点→ …… 每批之间是串行的等上一批全部完成批内是并行的。依赖数据传递节点 B 的执行需要节点 A 的输出数据。编排引擎在 A 完成时将 A 的结果传递给 B 的输入。数据结构通常是 JSON——A 的输出作为 B 的context的一部分。重试和降级DAG 中任何一个节点失败所有依赖于它的下游节点都会阻塞。必须处理失败重试指数退避、降级使用默认值或备用方案、跳过非关键节点可以跳过。三、生产级代码实现 Agent DAG 编排引擎 核心能力 1. DAG 解析邻接表 拓扑排序 2. 并行波次执行 3. 节点失败的重试和降级 4. 依赖数据传递和上下文合并 import asyncio import logging from typing import Dict, List, Optional, Set, Any, Callable from dataclasses import dataclass, field from enum import Enum from collections import deque import time logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class NodeStatus(Enum): PENDING pending # 等待前置依赖 RUNNING running # 正在执行 COMPLETED completed # 执行成功 FAILED failed # 执行失败 SKIPPED skipped # 被跳过前置节点失败导致 class DAGError(Exception): DAG 解析错误 pass dataclass class DAGNode: DAG 节点定义 node_id: str name: str # 执行函数接收合并后的上下文返回结果字典 executor: Callable # 前置节点列表 dependencies: List[str] field(default_factorylist) # 重试配置 max_retries: int 2 retry_delay_seconds: float 1.0 # 指数退避的基数 # 降级配置 fallback: Optional[Callable] None # 降级函数失败时调用 # 是否关键节点——关键节点失败会导致整个 DAG 终止 critical: bool True # 执行超时 timeout_seconds: float 120 dataclass class NodeResult: 节点执行结果 node_id: str status: NodeStatus output: Optional[Dict[str, Any]] None error: Optional[str] None retries: int 0 elapsed_ms: int 0 class DAGOrchestrator: DAG 编排引擎 核心算法 1. 从邻接表构建 DAG同时检测循环依赖 2. Kahn 拓扑排序生成执行波次 3. 逐波次并行执行节点 def __init__(self, nodes: List[DAGNode]): self.nodes {n.node_id: n for n in nodes} self._validate() # 邻接表node_id → [后继 node_id 列表] self.successors: Dict[str, List[str]] {} # 入度计数 self.in_degree: Dict[str, int] {} self._build_graph() def _validate(self): 验证 DAG 完整性 all_ids set(self.nodes.keys()) for node in self.nodes.values(): for dep in node.dependencies: if dep not in all_ids: raise DAGError( fNode {node.node_id} depends on unknown node {dep} ) # 检查每个依赖节点是否在 nodes 中已有定义 # 如果有外部依赖不在本 DAG 中的节点需要由外部先执行传入结果 def _build_graph(self): 构建邻接表 入度表 for node_id in self.nodes: self.successors[node_id] [] self.in_degree[node_id] 0 for node_id, node in self.nodes.items(): for dep in node.dependencies: if dep in self.successors: self.successors[dep].append(node_id) self.in_degree[node_id] 1 def detect_cycle(self) - bool: 检测 DAG 中是否存在环 使用 DFS 三色标记法 white (0): 未访问 gray (1): 正在访问在当前递归栈中 black (2): 已完成访问 如果遇到 gray 节点 → 存在环 WHITE, GRAY, BLACK 0, 1, 2 color {node_id: WHITE for node_id in self.nodes} def dfs(node_id: str) - bool: color[node_id] GRAY for successor in self.successors.get(node_id, []): if color[successor] GRAY: logger.error(Cycle detected: %s → %s, node_id, successor) return True if color[successor] WHITE: if dfs(successor): return True color[node_id] BLACK return False for node_id in self.nodes: if color[node_id] WHITE: if dfs(node_id): return True return False def get_execution_waves(self) - List[List[str]]: Kahn 算法生成执行波次 返回[[wave1_node_ids], [wave2_node_ids], ...] 同一 wave 内的节点没有依赖关系可以并行执行 if self.detect_cycle(): raise DAGError(DAG contains a cycle — cannot generate execution plan) in_degree dict(self.in_degree) waves [] queue deque() # 第一批入度为 0 的节点 for node_id, degree in in_degree.items(): if degree 0: queue.append(node_id) while queue: wave [] # 当前 wave 的所有节点入度为 0 for _ in range(len(queue)): node_id queue.popleft() wave.append(node_id) # 更新后继节点的入度 for successor in self.successors.get(node_id, []): in_degree[successor] - 1 if in_degree[successor] 0: queue.append(successor) waves.append(wave) # 检查是否有未执行的节点图中可能有孤立或不连通的部分 total_nodes sum(len(w) for w in waves) if total_nodes ! len(self.nodes): missing set(self.nodes.keys()) - set( nid for w in waves for nid in w ) raise DAGError(fUnreachable nodes (possible cycle): {missing}) logger.info(Execution plan: %d waves for %d nodes, len(waves), len(self.nodes)) for i, wave in enumerate(waves): logger.info( Wave %d: %s, i 1, wave) return waves async def execute(self, initial_context: Dict[str, Any] None ) - Dict[str, NodeResult]: 执行整个 DAG 参数: initial_context: 初始上下文外部传入的预填充数据 返回: {node_id: NodeResult} 每个节点的执行结果 initial_context initial_context or {} # 1. 生成执行计划 waves self.get_execution_waves() # 2. 存储每个节点的输出用于传递给下游节点 node_outputs: Dict[str, Any] {} results: Dict[str, NodeResult] {} # 3. 逐波次执行 for wave_idx, wave in enumerate(waves): logger.info( Wave %d/%d: %d nodes , wave_idx 1, len(waves), len(wave)) # 过滤掉因上游失败而被跳过的节点 active_nodes [ nid for nid in wave if results.get(nid, NodeResult(nid, NodeStatus.PENDING)).status ! NodeStatus.SKIPPED ] if not active_nodes: continue # 构建每个节点的上下文 tasks [] for node_id in active_nodes: node self.nodes[node_id] context self._merge_context(node, node_outputs, initial_context) tasks.append(self._execute_node_with_retry(node, context)) # 并行执行当前波次的所有节点 wave_results await asyncio.gather(*tasks, return_exceptionsTrue) # 4. 处理结果 has_critical_failure False for node_id, result in zip(active_nodes, wave_results): if isinstance(result, Exception): result NodeResult( node_idnode_id, statusNodeStatus.FAILED, errorstr(result), ) results[node_id] result node_outputs[node_id] result.output or {} if result.status NodeStatus.COMPLETED: logger.info( ✓ %s (%dms), node_id, result.elapsed_ms) else: node self.nodes[node_id] logger.error( ✗ %s: %s, node_id, result.error) if node.critical: has_critical_failure True # 关键节点失败 → 标记所有下游节点为 SKIPPED if has_critical_failure: self._skip_downstream(wave, results) return results def _merge_context(self, node: DAGNode, node_outputs: Dict[str, Any], initial: Dict[str, Any]) - Dict[str, Any]: 合并节点的执行上下文 上下文来源 1. initial_context外部传入权重最低 2. 所有前置依赖的输出按依赖顺序叠加后覆盖前 merged dict(initial) for dep_id in node.dependencies: if dep_id in node_outputs: merged.update(node_outputs[dep_id]) # 添加节点自身元信息 merged[_node_id] node.node_id merged[_dependencies] node.dependencies return merged async def _execute_node_with_retry(self, node: DAGNode, context: dict) - NodeResult: 执行单个节点带重试和降级 last_error None start_time time.time() for attempt in range(node.max_retries 1): try: # 执行带超时 output await asyncio.wait_for( node.executor(context), timeoutnode.timeout_seconds, ) elapsed int((time.time() - start_time) * 1000) return NodeResult( node_idnode.node_id, statusNodeStatus.COMPLETED, outputoutput if isinstance(output, dict) else {result: output}, retriesattempt, elapsed_mselapsed, ) except asyncio.TimeoutError: last_error fTimeout after {node.timeout_seconds}s logger.warning(Node %s: attempt %d/%d — %s, node.node_id, attempt 1, node.max_retries 1, last_error) except Exception as e: last_error str(e) logger.warning(Node %s: attempt %d/%d — %s, node.node_id, attempt 1, node.max_retries 1, last_error) # 指数退避 if attempt node.max_retries: delay node.retry_delay_seconds * (2 ** attempt) await asyncio.sleep(delay) # 所有重试失败 → 尝试降级 if node.fallback: try: logger.info(Node %s: attempting fallback, node.node_id) fallback_result node.fallback(context) elapsed int((time.time() - start_time) * 1000) return NodeResult( node_idnode.node_id, statusNodeStatus.COMPLETED, # 降级成功视作完成 outputfallback_result, retriesnode.max_retries 1, elapsed_mselapsed, ) except Exception as e: last_error fFallback also failed: {e} elapsed int((time.time() - start_time) * 1000) return NodeResult( node_idnode.node_id, statusNodeStatus.FAILED, errorlast_error, retriesnode.max_retries 1, elapsed_mselapsed, ) def _skip_downstream(self, current_wave: List[str], results: Dict[str, NodeResult]): 标记所有尚未执行的下游节点为 SKIPPED failed_ids [ nid for nid in current_wave if results.get(nid) and results[nid].status NodeStatus.FAILED ] # BFS 遍历下游 visited set(failed_ids) queue deque(failed_ids) while queue: node_id queue.popleft() for successor in self.successors.get(node_id, []): if successor not in visited: visited.add(successor) queue.append(successor) results[successor] NodeResult( node_idsuccessor, statusNodeStatus.SKIPPED, errorfUpstream node {node_id} failed, ) # --------------------------------------------------------------------------- # 使用示例 # --------------------------------------------------------------------------- async def main(): 示例财报分析的多 Agent DAG 编排 # 定义节点 async def collect_data(ctx): await asyncio.sleep(1) return {financial_data: Q2 营收 50 亿, competitors: [A, B]} async def analyze_data(ctx): data ctx.get(financial_data, ) await asyncio.sleep(1.5) return {analysis: f分析结果: {data}} async def search_web(ctx): competitors ctx.get(competitors, []) await asyncio.sleep(2) return {web_results: f搜索到的竞品信息: {competitors}} async def generate_report(ctx): analysis ctx.get(analysis, ) web ctx.get(web_results, ) await asyncio.sleep(1) return {report: f报告: {analysis} {web}} async def generate_chart(ctx): data ctx.get(financial_data, ) await asyncio.sleep(0.5) return {chart: f图表: {data}} nodes [ DAGNode(collect_data, 数据收集, collect_data), DAGNode(analyze_data, 数据分析, analyze_data, dependencies[collect_data]), DAGNode(search_web, 网络搜索, search_web, dependencies[collect_data]), DAGNode(generate_report, 报告生成, generate_report, dependencies[analyze_data, search_web]), DAGNode(generate_chart, 图表生成, generate_chart, dependencies[collect_data]), ] orchestrator DAGOrchestrator(nodes) results await orchestrator.execute({user_query: 分析竞品 A 的财报}) print(\n DAG 执行结果 ) for node_id, result in results.items(): status_icon ✓ if result.status NodeStatus.COMPLETED else ✗ print(f {status_icon} {node_id}: {result.status.value} ({result.elapsed_ms}ms)) if __name__ __main__: asyncio.run(main())四、边界分析与架构权衡DAG 中的循环依赖实际场景中可能出现伪循环依赖——A 需要 B 的部分结果B 需要 A 的部分结果。这本质上是A 和 B 不能完全独立解决方案合并为一个节点如果它们强耦合或者通过外部存储/消息队列解耦不纯的 DAG 模型节点超时对 DAG 的影响单个节点超时可能导致整个 wave 的其余节点等它超时后才继续——降低了并行效率解决方案为每个节点设定独立的超时超时后如果非关键节点则直接进入降级不等其他节点什么场景不该用 DAG 编排节点线性依赖A → B → C——DAG 的复杂度不值得实时性要求极高的场景——DAG 的 wave 调度引入了额外的批处理开销Agent 数量 3——并行执行几乎没有收益直接串行更简单五、总结Agent DAG 编排的核心是将多 Agent 执行拆成有向无环图通过拓扑排序生成并行执行波次最大化并行度。Kahn 算法将 DAG 分波次调度——同一波次内的节点并行执行、波次间串行等待。关键是每个节点的失败处理重试指数退避、降级备用方案、跳下游关键节点失败后标记所有下游为 SKIPPED。DAG 编排的价值在 Agent 数量 5 且存在复杂依赖关系时才体现——线性链的场景用 DAG 反而增加了复杂度。