从零构建轻量级 DAG 编排引擎：处理大模型复杂工作流的实战

一、为什么简单的链式调用不够用

在真实业务里，单靠一个 Prompt 很难处理复杂的流程。开发者通常会把多个 LLM 调用、API 请求和数据清洗步骤串在一起。但一旦逻辑变复杂，这种线性调用就会出问题。

代码里开始出现大量的 if-else 嵌套，处理异步等待和节点依赖变得非常麻烦。这时候，有向无环图（DAG）是一个更稳妥的选择。它不仅能理清任务顺序，还能让没有依赖关系的节点并行执行，减少等待时间。

二、DAG 调度逻辑与拓扑排序

工作流里的每个步骤就是一个“节点”，节点间的依赖关系就是“边”。要执行这个图，核心是先做环路检测，再通过拓扑排序确定执行顺序。

下面的图展示了一个简单的流程：先清洗输入，然后并行做情感分类和关键词提取，最后汇总生成报告。

graph LR
    Start([启动]) --> NodeA[清洗输入]
    NodeA --> NodeB[情感分类]
    NodeA --> NodeC[关键词提取]
    NodeB --> NodeD[生成报告]
    NodeC --> NodeD
    NodeD --> End([结束])

    style NodeB fill:#bbf,stroke:#333,stroke-width:2px
    style NodeD fill:#bfb,stroke:#333,stroke-width:2px

节点 B 和 C 都依赖 A，但它们之间没关系，所以引擎会让它们同时跑。

三、Node.js 轻量级实现

这是一个基于 JavaScript 的简单 DAG 引擎。它实现了拓扑排序来检查依赖，并支持异步并发执行。

class WorkflowNode {
    constructor(id, taskFunction) {
        this.id = id;
        this.taskFunction = taskFunction;
        this.dependencies = [];
        this.status = 'PENDING';
        this.result = null;
    }

    addDependency(nodeId) {
        this.dependencies.push(nodeId);
    }
}

class DagEngine {
    constructor() {
        this.nodes = new Map();
    }

    registerNode(node) {
        this.nodes.set(node.id, node);
    }

    // 拓扑排序：检查环路并决定顺序
    resolveExecutionOrder() {
        const inDegree = new Map();
        const adjList = new Map();
        const order = [];

        for (const [id, node] of this.nodes) {
            inDegree.set(id, 0);
            adjList.set(id, []);
        }

        for (const [id, node] of this.nodes) {
            for (const depId of node.dependencies) {
                if (!this.nodes.has(depId)) {
                    throw new Error(`节点 ${id} 依赖的 ${depId} 未注册`);
                }
                adjList.get(depId).push(id);
                inDegree.set(id, inDegree.get(id) + 1);
            }
        }

        const queue = [];
        for (const [id, degree] of inDegree) {
            if (degree === 0) queue.push(id);
        }

        while (queue.length > 0) {
            const currId = queue.shift();
            order.push(currId);

            for (const nextId of adjList.get(currId)) {
                inDegree.set(nextId, inDegree.get(nextId) - 1);
                if (inDegree.get(nextId) === 0) {
                    queue.push(nextId);
                }
            }
        }

        if (order.length !== this.nodes.size) {
            throw new Error("检测到循环依赖，无法执行");
        }

        return order;
    }

    // 并发执行
    async executeWorkflow(inputContext) {
        const completedResults = { ...inputContext };
        const runningPromises = new Map();

        while (true) {
            let hasPending = false;
            let progressed = false;

            for (const [id, node] of this.nodes) {
                if (node.status === 'COMPLETED' || node.status === 'FAILED') continue;
                hasPending = true;
                if (node.status === 'RUNNING') continue;

                // 检查依赖是否都完成了
                const allDepsMet = node.dependencies.every(depId => {
                    const depNode = this.nodes.get(depId);
                    return depNode && depNode.status === 'COMPLETED';
                });

                if (allDepsMet) {
                    node.status = 'RUNNING';
                    progressed = true;

                    const promise = (async () => {
                        try {
                            const depData = {};
                            node.dependencies.forEach(depId => {
                                depData[depId] = this.nodes.get(depId).result;
                            });
                            node.result = await node.taskFunction(completedResults, depData);
                            node.status = 'COMPLETED';
                        } catch (error) {
                            node.status = 'FAILED';
                            throw error;
                        }
                    })();
                    runningPromises.set(id, promise);
                }
            }

            if (!hasPending) break;
            if (!progressed && runningPromises.size === 0) {
                throw new Error("死锁：没有节点能继续执行");
            }

            await Promise.race(runningPromises.values());
            
            for (const [id, promise] of runningPromises) {
                const node = this.nodes.get(id);
                if (node.status === 'COMPLETED' || node.status === 'FAILED') {
                    runningPromises.delete(id);
                }
            }
        }

        const finalOutput = {};
        for (const [id, node] of this.nodes) {
            finalOutput[id] = node.result;
        }
        return finalOutput;
    }
}

// 测试运行
(async () => {
    const engine = new DagEngine();

    const nodeA = new WorkflowNode('CleanInput', async (context) => {
        return context.rawText.trim().replace(/[<>]/g, '');
    });

    const nodeB = new WorkflowNode('LlmClassify', async (context, depData) => {
        const text = depData.CleanInput;
        await new Promise(resolve => setTimeout(resolve, 500)); // 模拟 API 延迟
        return text.includes("好") ? "POSITIVE" : "NEGATIVE";
    });
    nodeB.addDependency('CleanInput');

    const nodeC = new WorkflowNode('ExtractKeywords', async (context, depData) => {
        const text = depData.CleanInput;
        return text.split(' ').filter(word => word.length > 1);
    });
    nodeC.addDependency('CleanInput');

    const nodeD = new WorkflowNode('GenerateReport', async (context, depData) => {
        const sentiment = depData.LlmClassify;
        const keywords = depData.ExtractKeywords;
        return `情感: ${sentiment}, 关键词: [${keywords.join(', ')}]`;
    });
    nodeD.addDependency('LlmClassify');
    nodeD.addDependency('ExtractKeywords');

    engine.registerNode(nodeA);
    engine.registerNode(nodeB);
    engine.registerNode(nodeC);
    engine.registerNode(nodeD);

    const order = engine.resolveExecutionOrder();
    console.log("执行顺序:", order.join(' -> '));

    const result = await engine.executeWorkflow({ rawText: " 这个产品设计得非常 好，解决了我的痛点。 " });
    console.log("结果:", result);
})();

四、生产环境需要考虑的几个问题

上面的代码适合本地或简单场景，如果要上生产环境，还得考虑下面几点：

1. 内存 vs 持久化
内存里的调度很快，但服务器一挂，中间结果就没了。如果工作流跑了几分钟才失败，重头再来很浪费。生产环境通常要用 Redis 或像 Temporal 这样的状态机来存状态，但这会增加网络延迟。

2. 重试策略与成本
大模型 API 经常超时或限流，加重试机制是必须的。但要注意，如果上游节点因为超时一直重试，可能会在短时间内消耗大量 Token。给每个节点设置重试上限和超时时间是必要的。

3. 静态图 vs 动态分支
DAG 在运行前就定好了结构，容易校验。但 LLM 的输出是动态的，有时候需要根据结果决定下一步走哪条路。如果要支持这种动态分支，图的拓扑结构得在运行时变，这会大大增加调试难度。

五、小结

做智能工作流，核心是把杂乱的调用拆成清晰的节点和依赖。用拓扑排序处理并发，不需要复杂的框架，也能让多个模型任务协同工作。对于小团队来说，这种轻量级的方案既能控制成本，也能保证流程跑得通。