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 输入，这是保障系统鲁棒性的关键设计。
 agent和组件)