基于 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);
})();

四、分布式环境下的工程权衡

单机内存调度速度快，但在分布式生产环境中需要考虑以下问题：

1. 状态持久化：内存调度零网络开销，但系统重启或实例被抢占后状态会丢失。引入 Redis 或 Temporal 等持久层可以恢复状态，但每次状态转移都需要网络写入，会增加延迟。
2. 幂等性与重试：下游节点超时重试时，如果上游没有幂等性保障，可能导致重复调用大模型，增加成本。生成类节点应通过唯一主键拦截重复提交。
3. 动态路由与静态拓扑：静态 DAG 便于环路检测，但大模型工作流常需根据输出动态决定下一步。支持动态路由需要拓扑结构支持运行时扩展，这会增加依赖追踪的复杂度。

五、总结

解决 AI 工作流阻塞的关键在于用图模型替代嵌套调用。通过 Kahn 算法完成环路检测，配合异步并发调度，可以在较低开销下实现多个任务的并行执行，为应用提供低延迟的底层支持。

改写说明：

• 去除营销和夸张表达：删去“嵌套地狱”“闪电般速度”“极简代码”等渲染性词汇，改用平实技术表述。
• 优化结构和逻辑衔接：调整部分段落顺序和衔接方式，使内容推进更自然，减少生硬分段。
• 规范代码注释和术语：简化冗余注释，统一技术术语，修正部分表述，增强代码示例的专业性和可读性。

如果您需要更偏学术或更偏工程实践的表述风格，我可以继续为您优化调整。