Sim 基于 DAG 的 AI 智能体工作流编排引擎技术解析

3. 技术难点挖掘

1. 动态 DAG 的构建与执行：如何将用户定义的循环、并行等高级结构，正确无误地编译为静态 DAG，并确保执行时的语义正确性（如循环的迭代、并发的扇入扇出）。
2. 有状态工作流的暂停与精确恢复：如何在任意节点（特别是并行分支中）暂停，并序列化所有中间状态（变量、未完成的边、待执行节点队列），保证恢复后执行结果的一致性。
3. 工具调用的安全性与性能：如何安全地执行用户自定义代码（Function Block）、动态发现和调用外部工具（MCP），同时管理连接、超时和错误，并避免对 LLM 请求造成性能瓶颈。
4. 混合执行模式的支持：需要同时支持非流式响应、流式响应（Streaming），并在浏览器和服务器两种环境下都能稳定工作，对网络通信和错误处理要求高。

4. 详细设计图

4.1 核心架构图 (Component Diagram)

4.2 核心链路序列图：工作流执行 (Sequence Diagram)

AI ProviderAgentBlockHandlerNodeOrchestratorDAGExecutionEngine 用户 AI ProviderAgentBlockHandlerNodeOrchestratorDAGExecutionEngine 用户 alt[需要调用工具]loop[处理就绪队列]run(workflowId)initializeQueue()get ready nodes[node1]executeNode(node)execute(ctx, block, inputs)formatTools(), buildMessages()携带工具的 LLM 请求响应 (可能含工具调用)executeFunction(toolCall) 工具执行结果发送工具结果最终响应 BlockOutputhandleNodeCompletion(output)edgeManager.processOutgoingEdges()获取新的就绪节点[node2, node3...]ExecutionResult

4.3 核心类图 (Class Diagram)

构建 持有 持有 委托执行 路由到

typeScript
DAGBuilder
-pathConstructor
-loopConstructor
-nodeConstructor
-edgeConstructor
+build(SerializedWorkflow, triggerBlockId) : DAG

DAG
+nodes: Map<string, DAGNode>
+loopConfigs: Map<string, SerializedLoop>

DAGNode
+id: string
+block: SerializedBlock
+incomingEdges: Set<string>
+outgoingEdges: Map<string, DAGEdge>
+metadata: NodeMetadata

ExecutionEngine
-readyQueue: string[]
-executing: Set<Promise>
-context: ExecutionContext
-dag: DAG
+run(triggerBlockId) : Promise<ExecutionResult>
-processQueue()
-executeNodeAsync(nodeId)
-handleNodeCompletion()
-buildPausedResult()

ExecutionContext
+executedBlocks: Set<string>
+blockLogs: any[]
+metadata: ExecutionMetadata
+workflowVariables: Map
+pendingDynamicNodes: string[]

NodeExecutionOrchestrator
+executeNode(ctx, nodeId) : Promise<NodeResult>

AgentBlockHandler
+canHandle(block) : boolean
+execute(ctx, block, inputs) : Promise<BlockOutput>
-formatTools(tools)
-buildMessages(inputs)
-executeProviderRequest()

5. 核心函数解析

5.1 DAGBuilder.build() - DAG 编译核心

这是将用户定义的静态工作流编译成可执行 DAG 的核心过程，体现了'构造'而非'解释'的设计思想。

// apps/sim/executor/dag/builder.ts - 简化伪代码
export class DAGBuilder {
  build(workflow: SerializedWorkflow, triggerBlockId?: string, savedIncomingEdges?: Record<string, string[]>): DAG {
    const dag: DAG = { 
      nodes: new Map(), 
      loopConfigs: new Map(), 
      parallelConfigs: new Map() 
    };
    // 阶段 1：初始化配置
    this.initializeConfigs(workflow, dag);
    // 提取循环、并行配置
    // 阶段 2：路径分析
    const reachableBlocks = this.pathConstructor.execute(workflow, triggerBlockId);
    // 关键：基于触发点或起点，计算工作流中实际可达的所有区块，实现'按需执行'。
    // 阶段 3：循环结构展开
    this.loopConstructor.execute(dag, reachableBlocks);
    // 关键：将逻辑上的循环节点，在 DAG 中展开为实际的节点和边结构，为执行引擎创造明确的路径。
    // 阶段 4：节点实例化与元数据标记
    const { blocksInLoops, blocksInParallels, pauseTriggerMapping } = this.nodeConstructor.execute(workflow, dag, reachableBlocks);
    // 关键：为每个 SerializedBlock 创建对应的 DAGNode，并标记其元数据（如是否在循环内、是否为恢复触发点等）。
    // 阶段 5：边构建
    this.edgeConstructor.execute(workflow, dag, blocksInParallels, blocksInLoops, reachableBlocks, pauseTriggerMapping);
    // 关键：根据区块间的连接关系，创建 DAGEdge，并可能根据循环/并行上下文调整边的源和目标。
    // 阶段 6：状态恢复（如从快照恢复）
    if(savedIncomingEdges){
      for(const[nodeId, incomingEdgeArray]of Object.entries(savedIncomingEdges)){
        const node = dag.nodes.get(nodeId);
        if(node){ node.incomingEdges = new Set(incomingEdgeArray); // 恢复节点的依赖边状态}
      }
    }
    return dag;
  }
}

技术要点：这是一个多阶段的编译器前端。pathConstructor 确保 DAG 只包含必要节点；loopConstructor 和 edgeConstructor 协同工作，将高级控制流语义'降低'为基本 DAG 结构；最后对 savedIncomingEdges 的处理，是支持状态恢复的关键，它直接修改了 DAG 节点的运行时依赖状态。

5.2 ExecutionEngine.run() - 执行调度核心

这是运行时的心脏，一个基于队列的异步调度器，负责驱动整个 DAG 的执行。

// apps/sim/executor/execution/engine.ts - 简化伪代码
export class ExecutionEngine {
  async run(triggerBlockId?: string): Promise<ExecutionResult> {
    const startTime = Date.now();
    try {
      // 1. 队列初始化：可能从快照恢复，或从触发点/起点开始
      this.initializeQueue(triggerBlockId);
      // 2. 主调度循环：只要有待处理工作就继续
      while(this.hasWork()){
        // 检查 readyQueue 或 executing
        await this.processQueue(); // 处理就绪节点，并等待任意一个执行完成
      }
      // 3. 收尾：等待所有异步执行彻底结束
      await this.waitForAllExecutions();
      // 4. 构建成功结果
      return { success: true, output: this.finalOutput, ...};
    } catch(error) {
      // 5. 错误处理：包装错误信息，保留执行上下文
      const executionResult: ExecutionResult = { success: false, ...};
      if(error && typeof error === 'object'){
        (error as any).executionResult = executionResult; // 将结果附加到原始错误上，便于调试
      }
      throw error; // 重新抛出，由上层捕获
    }
  }
  private async processQueue(): Promise<void> {
    // 批量执行所有当前就绪的节点（实现隐式并行）
    while(this.readyQueue.length > 0){
      const nodeId = this.dequeue();
      const promise = this.executeNodeAsync(nodeId); // 异步执行，不等待
      this.trackExecution(promise); // 加入执行跟踪集合
    }
    // 关键：等待至少一个正在执行的任务完成，以释放资源并可能产生新的就绪节点
    if(this.executing.size > 0){
      await this.waitForAnyExecution(); // 使用 Promise.race
    }
  }
  private async executeNodeAsync(nodeId: string): Promise<void> {
    const wasAlreadyExecuted = this.context.executedBlocks.has(nodeId);
    const result = await this.nodeOrchestrator.executeNode(this.context, nodeId);
    if(!wasAlreadyExecuted){
      // 关键：节点完成后的回调，需要加锁（queueLock）确保状态更新原子性
      await this.withQueueLock(async()=>{
        await this.handleNodeCompletion(nodeId, result.output, result.isFinalOutput);
      });
    }
  }
  private async handleNodeCompletion(nodeId: string, output: NormalizedBlockOutput, isFinalOutput: boolean): Promise<void> {
    const node = this.dag.nodes.get(nodeId);
    // 关键点 1: 检查暂停
    if(output._pauseMetadata){
      this.pausedBlocks.set(pauseMetadata.contextId, output._pauseMetadata);
      this.context.metadata.status = 'paused';
      return; // 不再处理下游，工作流进入暂停状态
    }
    // 关键点 2: 处理节点输出（如变量赋值）
    await this.nodeOrchestrator.handleNodeCompletion(this.context, nodeId, output);
    // 关键点 3: 边处理，激活下游
    const readyNodes = this.edgeManager.processOutgoingEdges(node, output, false);
    this.addMultipleToQueue(readyNodes); // 下游节点进入就绪队列
    // 关键点 4: 处理动态节点（如并行展开新增的节点）
    if(this.context.pendingDynamicNodes?.length > 0){
      this.addMultipleToQueue(this.context.pendingDynamicNodes);
      this.context.pendingDynamicNodes = [];
    }
  }
}

技术要点：

• 队列驱动：核心模型是持续消耗 readyQueue，并通过 edgeManager 的生产者 - 消费者模型。
• 并发控制：通过 executing Set 和 Promise.race/Promise.all 管理并发度，实现自然的并行执行。
• 原子性更新：withQueueLock 确保在处理节点完成、更新边状态、向队列添加新节点这一系列操作时，不会产生竞态条件。
• 暂停机制：在 handleNodeCompletion 中优先检查 _pauseMetadata，一旦发现即停止推进，并构建暂停结果。这是一种'协作式'的中断。

5.3 AgentBlockHandler 中的工具格式化

展示了 Sim 如何将多样化的工具抽象为 LLM 可用的统一接口，这是其扩展性的关键。

// apps/sim/executor/handlers/agent/agent-handler.ts - 节选
private async formatTools(ctx: ExecutionContext, inputTools: ToolInput[]): Promise<any[]> {
  // ... 过滤 usageControl 等 ...
  const mcpTools: ToolInput[] = [];
  const otherTools: ToolInput[] = [];
  // 1. 分类工具
  for(const tool of filtered){
    if(tool.type === 'mcp'){ mcpTools.push(tool); }
    else{ otherTools.push(tool); }
  }
  // 2. 并行处理非 MCP 工具（自定义工具、内置工具转换）
  const otherResults = await Promise.all(otherTools.map(async(tool)=>{
    if(tool.type === 'custom-tool'){ return await this.createCustomTool(ctx, tool); // 创建可执行函数 }
    return this.transformBlockTool(ctx, tool); // 转换内置工具（如 API 块）
  }));
  // 3. 批量处理 MCP 工具（性能优化）
  const mcpResults = await this.processMcpToolsBatched(ctx, mcpTools);
  // 4. 合并并返回
  return [...otherResults,...mcpResults].filter(tool => tool != null);
}

private async createCustomTool(ctx: ExecutionContext, tool: ToolInput): Promise<any> {
  // ... 获取 schema 和 code ...
  const toolId = `${AGENT.CUSTOM_TOOL_PREFIX}${title}`;
  return {
    id: toolId, 
    name: schema.function.name, 
    description: schema.function.description || '', 
    parameters: filteredSchema, // 过滤掉已由 params 提供的参数 
    params: userProvidedParams, // 预设参数 
    usageControl: tool.usageControl || 'auto', // 关键：定义执行函数，在 LLM 调用时被执行
    executeFunction: async(callParams: Record<string,any>)=>{
      const mergedParams = mergeToolParameters(userProvidedParams, callParams);
      const result = await executeTool('function_execute', {
        code, // 用户定义的函数代码...
        mergedParams, 
        timeout: tool.timeout, 
        envVars: ctx.environmentVariables || {}, 
        workflowVariables: ctx.workflowVariables || {},
      }, false, false, ctx);
      if(!result.success){ throw new Error(result.error); }
      return result.output;
    }
  };
}

技术要点：

• 统一包装：无论底层是代码片段、HTTP API 还是 MCP 协议，最终都包装成具有 executeFunction 方法的对象。
• 参数合并：支持在工具配置时提供预设参数（params），并与 LLM 调用时产生的参数（callParams）安全合并。
• 上下文注入：执行时自动注入工作流变量、环境变量等上下文（ctx），使工具能感知工作流状态。
• MCP 优化：processMcpToolsBatched 体现了对性能的考虑，通过按服务器分组发现工具，减少连接开销。

总结

Sim 项目的技术核心在于一个精心设计的 DAG 编译与执行引擎。它成功地将可视化编程的易用性与底层执行引擎的严谨性（状态管理、并发控制、错误处理）结合起来。其架构清晰，通过构造器模式编译工作流，通过队列和 Promise 管理异步执行，通过统一的处理器接口集成多样化功能。代码中体现的'快照恢复'、'工具抽象'和'混合执行模式支持'是应对实际生产环境复杂性的关键设计，使其超越了简单的原型工具范畴，具备了支撑企业级 AI 应用的能力。

同类方案中，相较于 LangChain 等代码库，Sim 提供了更低门槛的可视化界面和开箱即用的状态管理；相较于 n8n 等通用自动化工具，它在 AI 智能体、工具集成和提示工程方面做了更深度的原生集成。其主要的考量点可能在于应对超大规模、高并发工作流时的性能优化，以及更复杂嵌套控制流的场景支持。