Function Calling 编排引擎设计：DAG + 条件分支的工作流定义-Seo优化-塔城地区网站建设公司

Function Calling 编排引擎设计：DAG + 条件分支的工作流定义

一、LLM 的自由调用是一把双刃剑

让 LLM 自由决定调用哪些工具、以什么顺序调用，在简单场景下工作得很好。但当任务变得复杂时，问题就暴露了：

三个工具应该按 A→B→C 的顺序执行，但 LLM 有时调成 A→C→B，结果不理想
工具 B 的输入依赖工具 A 的输出，LLM 有时会把参数搞混
缺少错误重试和超时控制，一个工具挂了整个任务就卡死

答案是引入一个编排引擎——不让 LLM 自由决策调用顺序，而是用 DAG（有向无环图）预先定义工作流，LLM 只负责在执行时填充参数。

二、编排引擎的架构

编排引擎的架构主要划分为工作流定义、核心引擎与运行时管理三个层次，各层职责如下：

工作流定义：基于 DAG（有向无环图）构建，节点代表工具调用，边代表依赖关系。支持普通节点（执行单个工具）、条件节点（if/else 分支）、并行节点（同时执行多个工具）以及 LLM 节点（需要动态决策时让 LLM 介入）。
核心引擎：包含 DAG 解析器（加载工作流定义）、调度器（按拓扑顺序执行）、执行器（管理单节点生命周期）以及状态管理（上下文传递 + 断点恢复）。
运行时管理：针对执行中的工作流，提供失败重试（指数退避 × 3）、超时控制（单节点 30s 上限）以及部分回滚（已执行步骤的补偿）。

这种分层设计确保了工作流的定义与执行解耦，同时增强了系统的稳定性与可控性。

三、Go 实现编排引擎

工作流定义

package workflow import ( "context" "encoding/json" --- "fmt" "time" ) // NodeType 节点类型 type NodeType string const ( NodeTypeTool NodeType = "tool" // 工具调用 NodeTypeCondition NodeType = "condition" // 条件分支 NodeTypeParallel NodeType = "parallel" // 并行执行 NodeTypeLLM NodeType = "llm" // LLM 推理 ) // NodeStatus 节点执行状态 type NodeStatus string const ( StatusPending NodeStatus = "pending" StatusRunning NodeStatus = "running" StatusCompleted NodeStatus = "completed" StatusFailed NodeStatus = "failed" StatusSkipped NodeStatus = "skipped" ) // WorkflowNode 工作流节点 type WorkflowNode struct { ID string // 节点唯一 ID Type NodeType // 节点类型 ToolName string // 工具名（type=tool 时） Config json.RawMessage // 节点配置 DependsOn []string // 依赖的前置节点 ID RetryConfig *RetryConfig // 重试配置 Timeout time.Duration // 超时时间 } // RetryConfig 重试配置 type RetryConfig struct { MaxRetries int // 最大重试次数 Backoff time.Duration // 基础退避时间 MaxBackoff time.Duration // 最大退避时间 } // Workflow 工作流定义 type Workflow struct { Name string Version string Description string Nodes []WorkflowNode } // WorkflowContext 工作流执行上下文 type WorkflowContext struct { WorkflowID string Variables map[string]interface{} // 全局变量 NodeOutputs map[string]interface{} // 节点输出（nodeID → output） StartTime time.Time }

编排引擎核心

// OrchestrationEngine 编排引擎 type OrchestrationEngine struct { toolRegistry ToolRegistry // 工具注册表 maxParallel int // 最大并行节点数 } // ToolRegistry 工具注册表接口 type ToolRegistry interface { Execute(ctx context.Context, toolName string, params map[string]interface{}) (interface{}, error) } // ExecutionResult 工作流执行结果 type ExecutionResult struct { WorkflowID string Status string NodeResults map[string]NodeResult TotalTime time.Duration Error error } type NodeResult struct { NodeID string Status NodeStatus Output interface{} Error string Duration time.Duration Retries int } // Execute 执行工作流 func (oe *OrchestrationEngine) Execute(ctx context.Context, wf *Workflow, input map[string]interface{}) (*ExecutionResult, error) { // 初始化执行上下文 wfCtx := &WorkflowContext{ WorkflowID: fmt.Sprintf("wf_%d", time.Now().UnixNano()), Variables: input, NodeOutputs: make(map[string]interface{}), StartTime: time.Now(), } // 构建 DAG + 计算执行顺序 executor := oe.buildDAG(wf) // 按拓扑顺序执行 for _, batch := range executor.ExecutionOrder() { if err := oe.executeBatch(ctx, wfCtx, batch); err != nil { return oe.buildErrorResult(wfCtx, err), err } } return oe.buildSuccessResult(wfCtx), nil }

DAG 构建与拓扑排序

// DAGExecutor DAG 执行器 type DAGExecutor struct { nodes map[string]*WorkflowNode edges map[string][]string // node → dependents inDegree map[string]int ordered [][]string // 分层执行顺序 } func (oe *OrchestrationEngine) buildDAG(wf *Workflow) *DAGExecutor { executor := &DAGExecutor{ nodes: make(map[string]*WorkflowNode), edges: make(map[string][]string), inDegree: make(map[string]int), } // 注册所有节点 for i := range wf.Nodes { node := &wf.Nodes[i] executor.nodes[node.ID] = node executor.inDegree[node.ID] = 0 } // 建立依赖关系 for _, node := range wf.Nodes { for _, depID := range node.DependsOn { executor.edges[depID] = append(executor.edges[depID], node.ID) executor.inDegree[node.ID]++ } } // Kahn 算法分层排序 var queue []string for id, degree := range executor.inDegree { if degree == 0 { queue = append(queue, id) } } visited := 0 for len(queue) > 0 { var currentBatch []string var nextQueue []string for _, id := range queue { currentBatch = append(currentBatch, id) visited++ // 移除该节点的出边 for _, dep := range executor.edges[id] { executor.inDegree[dep]-- if executor.inDegree[dep] == 0 { nextQueue = append(nextQueue, dep) } } } executor.ordered = append(executor.ordered, currentBatch) queue = nextQueue } // 检测循环依赖 if visited < len(executor.nodes) { fmt.Printf("[WARN] DAG 包含循环依赖，已访问 %d/%d 节点\n", visited, len(executor.nodes)) } return executor } // ExecutionOrder 返回分层执行顺序 func (de *DAGExecutor) ExecutionOrder() [][]string { return de.ordered }

条件分支和并行执行

// executeBatch 执行一批节点（同层可并行） func (oe *OrchestrationEngine) executeBatch( ctx context.Context, wfCtx *WorkflowContext, batch []string, ) error { // 单节点场景：顺序执行 if len(batch) == 1 { return oe.executeNode(ctx, wfCtx, batch[0]) } // 多节点场景：并行执行 errCh := make(chan error, len(batch)) for _, nodeID := range batch { nid := nodeID go func() { errCh <- oe.executeNode(ctx, wfCtx, nid) }() } // 等待所有并行节点完成 var firstErr error for i := 0; i < len(batch); i++ { if err := <-errCh; err != nil && firstErr == nil { firstErr = err } } return firstErr } // executeNode 执行单个节点 func (oe *OrchestrationEngine) executeNode( ctx context.Context, wfCtx *WorkflowContext, nodeID string, ) error { node, exists := oe.getNode(nodeID) if !exists { return fmt.Errorf("节点 %s 不存在", nodeID) } // 设置超时 timeout := node.Timeout if timeout == 0 { timeout = 30 * time.Second } ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel() // 解析节点输入参数（可从上游节点输出中引用） params, err := oe.resolveParams(wfCtx, node) if err != nil { return fmt.Errorf("解析参数失败: %w", err) } switch node.Type { case NodeTypeTool: return oe.executeToolNode(ctx, wfCtx, node, params) case NodeTypeCondition: return oe.executeConditionNode(ctx, wfCtx, node, params) case NodeTypeParallel: return oe.executeParallelNode(ctx, wfCtx, node, params) case NodeTypeLLM: return oe.executeLLMNode(ctx, wfCtx, node, params) default: return fmt.Errorf("未知节点类型: %s", node.Type) } } // executeConditionNode 条件分支节点 func (oe *OrchestrationEngine) executeConditionNode( ctx context.Context, wfCtx *WorkflowContext, node *WorkflowNode, params map[string]interface{}, ) error { // 解析条件表达式 var condConfig struct { Expression string `json:"expression"` // "${output.order_count} > 10" TrueBranch []string `json:"true_branch"` // 条件为真时执行的节点 FalseBranch []string `json:"false_branch"` // 条件为假时执行的节点 } if err := json.Unmarshal(node.Config, &condConfig); err != nil { return err } // 评估条件（简化版：使用模板替换） result, err := oe.evaluateExpression(wfCtx, condConfig.Expression) if err != nil { return err } // 选择分支并执行 var branch []string if result { branch = condConfig.TrueBranch } else { branch = condConfig.FalseBranch } for _, nid := range branch { if err := oe.executeNode(ctx, wfCtx, nid); err != nil { return err } } return nil }

参数解析——支持跨节点引用

// resolveParams 解析节点参数，支持从上游节点输出中引用 // 语法: "${node_id.output_field}" → 从指定节点的输出中取值 func (oe *OrchestrationEngine) resolveParams( wfCtx *WorkflowContext, node *WorkflowNode, ) (map[string]interface{}, error) { var rawParams map[string]interface{} if len(node.Config) > 0 { if err := json.Unmarshal(node.Config, &rawParams); err != nil { return nil, err } } resolved := make(map[string]interface{}) for key, value := range rawParams { strVal, ok := value.(string) if !ok { resolved[key] = value continue } // 检测是否为引用语法: ${node_id.field} if strings.HasPrefix(strVal, "${") && strings.HasSuffix(strVal, "}") { ref := strVal[2 : len(strVal)-1] parts := strings.SplitN(ref, ".", 2) if len(parts) != 2 { return nil, fmt.Errorf("无效的引用语法: %s", strVal) } nodeID, field := parts[0], parts[1] output, exists := wfCtx.NodeOutputs[nodeID] if !exists { return nil, fmt.Errorf("引用的节点 %s 尚未执行", nodeID) } // 从节点输出中提取字段值 fieldValue, err := oe.extractField(output, field) if err != nil { return nil, fmt.Errorf("提取字段失败: %w", err) } resolved[key] = fieldValue } else { resolved[key] = value } } return resolved, nil }