Agent 任务编排多步骤推理的依赖关系如何建模一、Agent 不是单次调用是多步骤推理的连锁反应普通 API 调用是一次输入一次输出。Agent 不同一个用户问题可能触发工具调用、子 Agent 调用、检索增强和结果合成每个步骤都可能依赖于前面步骤的输出。当步骤从 3 个增加到 15 个时依赖关系会变得极其复杂。典型的错误模式Agent 提前返回了部分结果但其实隐含的依赖链还没完成两个独立步骤被错误地假设为有依赖导致不必要的串行等待某个工具调用失败后依赖它的所有下游步骤都静默丢失。这些问题在开发环境很难复现因为测试 Case 通常只有 2-3 个步骤无法暴露长链推理的依赖管理缺陷。依赖建模的核心挑战有三个一、依赖关系不是静态的而是在推理过程中动态生成的——Agent 决定调用哪个工具后才形成下一步的依赖二、部分依赖是软依赖如果搜索结果不够好可以跳过摘要步骤直接返回原搜索片段而传统工作流引擎只支持硬依赖三、失败传播需要细粒度控制——某个步骤失败后哪些下游应该停止、哪些可以继续。二、基于 DAG 条件边的动态依赖建模将 Agent 的多步骤推理表示为动态有向无环图DAG节点是推理步骤边代表依赖关系。与传统 DAG 不同这里的边有两种类型硬依赖必须等前置步骤完成和软依赖前置步骤结果不佳时可以跳过。DAG 中的菱形节点代表条件分支它们本身也是推理步骤的产物。Agent 执行到检索步骤后由检索结果的质量评估决定走 RAG 增强推理还是降级到通用推理。这种动态决策无法在 DAG 初始化时预定义需要在运行时由 Agent 注入新的边。三、Agent DAG 执行引擎的 Go 实现以下实现支持动态添加节点和条件边package agentdag import ( context fmt sync ) // DependencyType 依赖类型 type DependencyType int const ( HardDep DependencyType iota // 硬依赖必须成功 SoftDep // 软依赖可降级 ) // NodeStatus 节点执行状态 type NodeStatus int const ( StatusPending NodeStatus iota StatusRunning StatusSuccess StatusFailed StatusSkipped // 软依赖前置失败时跳过 ) // DAGNode 推理步骤节点 type DAGNode struct { ID string Execute func(ctx context.Context, input map[string]interface{}) (interface{}, error) HardDeps []string // 硬依赖节点 ID 列表 SoftDeps []string // 软依赖节点 ID 列表 Meta map[string]string } // DAG 有向无环图 type DAG struct { mu sync.Mutex nodes map[string]*DAGNode status map[string]NodeStatus output map[string]interface{} // 节点 ID - 执行输出 } // NewDAG 创建空 DAG func NewDAG() *DAG { return DAG{ nodes: make(map[string]*DAGNode), status: make(map[string]NodeStatus), output: make(map[string]interface{}), } } // AddNode 静态添加节点 func (d *DAG) AddNode(node *DAGNode) error { d.mu.Lock() defer d.mu.Unlock() if _, exists : d.nodes[node.ID]; exists { return fmt.Errorf(节点 %s 已存在, node.ID) } d.nodes[node.ID] node d.status[node.ID] StatusPending return nil } // AddDynamicNode Agent 运行时动态添加节点和依赖 func (d *DAG) AddDynamicNode(node *DAGNode, addHardDeps, addSoftDeps []string) error { node.HardDeps append(node.HardDeps, addHardDeps...) node.SoftDeps append(node.SoftDeps, addSoftDeps...) return d.AddNode(node) } // Execute 拓扑排序 并行执行 DAG func (d *DAG) Execute(ctx context.Context) error { // 检测环 if err : d.detectCycle(); err ! nil { return err } for { ready : d.getReadyNodes() if len(ready) 0 { break } var wg sync.WaitGroup errCh : make(chan error, len(ready)) for _, nodeID : range ready { wg.Add(1) go func(nid string) { defer wg.Done() if err : d.executeNode(ctx, nid); err ! nil { errCh - err } }(nodeID) } wg.Wait() close(errCh) // 收集所有并行执行的错误 var errs []error for err : range errCh { errs append(errs, err) } // 硬依赖失败传播失败给下游 d.propagateFailure() if len(errs) 0 { return fmt.Errorf(本轮执行中 %d 个节点失败, len(errs)) } } d.checkCompletion() return nil } // getReadyNodes 获取所有依赖已满足的就绪节点 func (d *DAG) getReadyNodes() []string { d.mu.Lock() defer d.mu.Unlock() var ready []string for id, node : range d.nodes { if d.status[id] ! StatusPending { continue } // 检查硬依赖所有 HardDeps 必须成功 hardSatisfied : true for _, depID : range node.HardDeps { if d.status[depID] ! StatusSuccess { hardSatisfied false break } } if !hardSatisfied { continue } // 检查软依赖SoftDeps 成功或失败都可以失败时后续会处理降级 softReady : true for _, depID : range node.SoftDeps { status : d.status[depID] if status StatusPending || status StatusRunning { softReady false break } } if !softReady { continue } ready append(ready, id) } return ready } // executeNode 执行单个推理节点 func (d *DAG) executeNode(ctx context.Context, nodeID string) error { d.mu.Lock() node, exists : d.nodes[nodeID] if !exists { d.mu.Unlock() return fmt.Errorf(节点 %s 不存在, nodeID) } d.status[nodeID] StatusRunning d.mu.Unlock() // 构建输入收集所有依赖节点的输出 input : make(map[string]interface{}) d.mu.Lock() for _, depID : range node.HardDeps { input[depID] d.output[depID] } for _, depID : range node.SoftDeps { status : d.status[depID] if status StatusSuccess { input[depID] d.output[depID] } else if status StatusFailed { input[depID] nil // 降级标记 } } d.mu.Unlock() output, err : node.Execute(ctx, input) d.mu.Lock() defer d.mu.Unlock() if err ! nil { d.status[nodeID] StatusFailed return fmt.Errorf(节点 %s 执行失败: %w, nodeID, err) } d.status[nodeID] StatusSuccess d.output[nodeID] output return nil } // propagateFailure 将失败传播到硬依赖的下游节点 func (d *DAG) propagateFailure() { d.mu.Lock() defer d.mu.Unlock() changed : true for changed { changed false for id, node : range d.nodes { if d.status[id] ! StatusPending { continue } for _, depID : range node.HardDeps { if d.status[depID] StatusFailed { d.status[id] StatusFailed changed true break } } } } } // checkCompletion 检查是否所有节点都已完成 func (d *DAG) checkCompletion() error { d.mu.Lock() defer d.mu.Unlock() var pending, failed []string for id, status : range d.status { switch status { case StatusPending: pending append(pending, id) case StatusFailed: failed append(failed, id) } } if len(pending) 0 { return fmt.Errorf(存在未完成节点: %v, pending) } if len(failed) 0 { return fmt.Errorf(存在失败节点: %v, failed) } return nil } // detectCycle 检测 DAG 中是否存在环 func (d *DAG) detectCycle() error { visited : make(map[string]bool) recStack : make(map[string]bool) var dfs func(string) error dfs func(nodeID string) error { visited[nodeID] true recStack[nodeID] true node, ok : d.nodes[nodeID] if !ok { recStack[nodeID] false return nil } allDeps : append([]string{}, node.HardDeps...) allDeps append(allDeps, node.SoftDeps...) for _, dep : range allDeps { if !visited[dep] { if err : dfs(dep); err ! nil { return err } } else if recStack[dep] { return fmt.Errorf(检测到环: %s - %s, nodeID, dep) } } recStack[nodeID] false return nil } for id : range d.nodes { if !visited[id] { if err : dfs(id); err ! nil { return err } } } return nil }使用示例func main() { dag : NewDAG() // 静态节点 dag.AddNode(DAGNode{ ID: intent_parse, Execute: func(ctx context.Context, input map[string]interface{}) (interface{}, error) { return search_intent, nil }, }) dag.AddNode(DAGNode{ ID: vector_search, HardDeps: []string{intent_parse}, Execute: func(ctx context.Context, input map[string]interface{}) (interface{}, error) { return []string{doc1, doc2}, nil }, }) // Agent 运行时动态添加节点 dag.AddDynamicNode(DAGNode{ ID: quality_check, Execute: func(ctx context.Context, input map[string]interface{}) (interface{}, error) { docs : input[vector_search].([]string) if len(docs) 2 { return low_quality, nil } return high_quality, nil }, }, []string{vector_search}, nil) dag.AddDynamicNode(DAGNode{ ID: rag_reasoning, Execute: func(ctx context.Context, input map[string]interface{}) (interface{}, error) { return 基于检索结果的推理答案, nil }, }, nil, []string{quality_check}) ctx : context.Background() if err : dag.Execute(ctx); err ! nil { panic(err) } }四、动态 DAG 的边界与约束软依赖的降级策略软依赖节点的失败不应导致整个流程失败但需要明确的降级路径。在上面的实现中执行器会检查软依赖的状态失败时传递 nil 而非输出。节点 Execute 函数需要能够处理 nil 输入——这是软依赖场景下每个节点必须实现的防御逻辑。DAG 的循环风险Agent 推理本身就可能是迭代的——工具调用结果不够理想时需要重新检索。DAG 中检测到环时不应该直接报错而应该支持有限循环设置最大迭代次数超过阈值才终止。上例中的检测仅做安全防护生产环境需要增强为带计数器的环限制器。执行顺序的非确定性由于 DAG 中同层节点并行执行而工具调用可能有副作用如写入外部系统需要在节点设计中明确保证幂等性。如果一个工具调用是写操作它必须能在重试时产生相同的结果。五、总结Agent 多步骤推理的依赖建模核心在于区分硬依赖和软依赖支持运行时动态添加节点。DAG 拓扑排序 并行执行 条件分支的组合提供了足够的灵活性同时保持了执行的可预测性。实现中最容易被忽略的点是软依赖的降级处理——每个节点都必须能够处理来自软依赖前置节点的 nil 输入这是保障系统鲁棒性的关键设计。