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 // 工具名typetool 时 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 }四、边界分析与 Trade-offsDAG 编排 vs LLM 自由调用DAG 编排确定性高适合标准化流程审批流、数据处理流水线LLM 自由调用灵活性高适合探索性任务客服对话、搜索查询建议混合DAG 做骨架关键路径上插入 LLM 节点做动态决策条件分支的复杂度控制分支层级超过 3 层时建议拆分为子工作流过于复杂的条件逻辑应该在外部 DSL如 JSON/YAML中定义断点恢复长时间运行的工作流需要支持中断恢复。节点执行完成后立即持久化状态重启后从最后完成的节点重新开始。并行节点的错误处理默认策略一个并行节点失败等待其他并行节点完成后再返回错误可选策略快速失败——一个失败立即取消其他并行节点五、总结Function Calling 编排引擎的设计要点DAG 定义工作流——节点 边明确执行顺序拓扑排序调度——同层并行层间串行参数引用机制——节点间通过${node_id.field}传递数据条件分支——if/else 逻辑在编排层面控制LLM 节点——在需要动态决策处插入 LLM 推理编排引擎的核心价值将 LLM 从顺序决策者变为参数填充者大幅降低非确定性行为的风险。
