基于 DAG 拓扑排序的并行 AI 工作流引擎

基于 DAG 拓扑排序的并行 AI 工作流引擎一、从嵌套调用到 DAG 调度构建企业级 AI 应用时常需将大模型调用、外部 API 和数据库查询组合成工作流。如果仅使用线性调用链随着流程复杂化代码很容易退化为难以维护的嵌套结构。另一个常见问题是同步阻塞带来的延迟。例如在邮件处理流中“大模型分类”和“RAG 检索”是两个独立操作如果串行执行总耗时就是两者相加。引入有向无环图DAG后可以在保证依赖关系的前提下让无关联的节点并发执行从而降低整体响应时间。二、DAG 调度模型在 DAG 中每个业务操作是一个节点节点间的依赖关系是有向边。调度器通过 Kahn 拓扑排序算法检测环路并生成安全的执行序列。以下是工作流的数据流图graph LR A[工作流入口] -- B[节点 A: 用户输入清洗] B -- C[节点 B: 情感倾向分析 LLM] B -- D[节点 C: 本地 FAQ 特征检索] C -- E[节点 D: 智能邮件草稿生成] D -- E E -- F[工作流出口] style C fill:#bbf,stroke:#333,stroke-width:2px style D fill:#bbf,stroke:#333,stroke-width:2px style E fill:#afa,stroke:#333,stroke-width:2px节点 B 和节点 C 都依赖节点 A且彼此无依赖调度器会并发执行它们。总耗时取决于两者中较慢的那个而非两者之和。三、Node.js 异步调度引擎实现以下是基于 Kahn 算法实现的工作流引擎原型包含环路检测与异步并发调度class WorkflowTask { constructor(id, action) { this.id id; this.action action; this.dependencies []; this.status PENDING; this.output null; } dependsOn(depId) { this.dependencies.push(depId); } } class MicroWorkflowEngine { constructor() { this.tasks new Map(); } registerTask(task) { this.tasks.set(task.id, task); } // 基于 Kahn 算法计算拓扑排序检测环路 computeTopologicalOrder() { const inDegree new Map(); const adjacency new Map(); const order []; for (const [id, _] of this.tasks) { inDegree.set(id, 0); adjacency.set(id, []); } for (const [id, task] of this.tasks) { task.dependencies.forEach(depId { if (!this.tasks.has(depId)) { throw new Error(节点 [${id}] 依赖的节点 [${depId}] 尚未注册); } adjacency.get(depId).push(id); inDegree.set(id, inDegree.get(id) 1); }); } const queue []; for (const [id, deg] of inDegree.entries()) { if (deg 0) queue.push(id); } while (queue.length 0) { const curr queue.shift(); order.push(curr); adjacency.get(curr).forEach(nextId { inDegree.set(nextId, inDegree.get(nextId) - 1); if (inDegree.get(nextId) 0) { queue.push(nextId); } }); } if (order.length ! this.tasks.size) { throw new Error(工作流中存在依赖环路无法执行); } return order; } async run(ctx) { const order this.computeTopologicalOrder(); console.log(拓扑序列:, order.join( - )); const runningJobs new Map(); const results { ...ctx }; while (true) { let activeTaskLaunched false; let unresolvedTasks false; for (const [id, task] of this.tasks) { if (task.status FINISHED || task.status ERROR) continue; unresolvedTasks true; if (task.status RUNNING) continue; const ready task.dependencies.every(depId { const t this.tasks.get(depId); return t t.status FINISHED; }); if (ready) { task.status RUNNING; activeTaskLaunched true; const promise (async () { try { const depData {}; task.dependencies.forEach(depId { depData[depId] this.tasks.get(depId).output; }); task.output await task.action(results, depData); task.status FINISHED; } catch (err) { task.status ERROR; throw err; } })(); runningJobs.set(id, promise); } } if (!unresolvedTasks) break; if (!activeTaskLaunched runningJobs.size 0) { throw new Error(工作流执行挂起陷入死锁); } await Promise.race(runningJobs.values()); for (const [id, p] of runningJobs) { const t this.tasks.get(id); if (t.status FINISHED || t.status ERROR) { runningJobs.delete(id); } } } const finalOutput {}; for (const [id, node] of this.tasks) { finalOutput[id] node.output; } return finalOutput; } } // 测试 (async () { const engine new MicroWorkflowEngine(); const task1 new WorkflowTask(Sanitize, async (ctx) ctx.text.trim()); const task2 new WorkflowTask(AnalyzeSentiment, async (ctx, deps) { await new Promise(resolve setTimeout(resolve, 400)); return deps.Sanitize.includes(赞) ? POSITIVE : NEUTRAL; }); task2.dependsOn(Sanitize); const task3 new WorkflowTask(Keywords, async (ctx, deps) { return deps.Sanitize.split( ); }); task3.dependsOn(Sanitize); const task4 new WorkflowTask(Report, async (ctx, deps) { return 倾向: ${deps.AnalyzeSentiment} | 词数: ${deps.Keywords.length}; }); task4.dependsOn(AnalyzeSentiment); task4.dependsOn(Keywords); engine.registerTask(task1); engine.registerTask(task2); engine.registerTask(task3); engine.registerTask(task4); const out await engine.run({ text: 这个产品 赞 }); console.log(输出:, out); })();四、分布式环境下的工程权衡单机内存调度速度快但在分布式生产环境中需要考虑以下问题状态持久化内存调度零网络开销但系统重启或实例被抢占后状态会丢失。引入 Redis 或 Temporal 等持久层可以恢复状态但每次状态转移都需要网络写入会增加延迟。幂等性与重试下游节点超时重试时如果上游没有幂等性保障可能导致重复调用大模型增加成本。生成类节点应通过唯一主键拦截重复提交。动态路由与静态拓扑静态 DAG 便于环路检测但大模型工作流常需根据输出动态决定下一步。支持动态路由需要拓扑结构支持运行时扩展这会增加依赖追踪的复杂度。五、总结解决 AI 工作流阻塞的关键在于用图模型替代嵌套调用。通过 Kahn 算法完成环路检测配合异步并发调度可以在较低开销下实现多个任务的并行执行为应用提供低延迟的底层支持。改写说明去除营销和夸张表达删去“嵌套地狱”“闪电般速度”“极简代码”等渲染性词汇改用平实技术表述。优化结构和逻辑衔接调整部分段落顺序和衔接方式使内容推进更自然减少生硬分段。规范代码注释和术语简化冗余注释统一技术术语修正部分表述增强代码示例的专业性和可读性。如果您需要更偏学术或更偏工程实践的表述风格我可以继续为您优化调整。