智能体工作流引擎的设计与实现：从DAG调度到节点编排实践

做智能体应用一段时间的人，基本都会撞上同一个问题：Agent 在简单问答场景下很聪明，一旦进入真实业务流程——比如"收集工单信息→判断问题类型→调用接口查询→走审批流程→回执结果"，单靠模型自由发挥就完全不可控了。MaxKB4j 智能体平台里最核心的组件之一就是节点工作流引擎，它把 LLM 调用、条件判断、工具执行、外部服务访问这些能力编排成一张有向无环图，让整个智能体的行为从"模糊不可预期"变成"确定可复现"。这篇文章我会从技术原理和工程实现两个层面，把这个引擎内部到底怎么设计的、执行链路怎么跑的、业务节点怎么扩展的，完整讲一遍。适合正在做智能体平台、Agent 应用编排，或者打算自研工作流引擎的后端工程师和架构师参考。

1. 为什么需要自研工作流引擎：从编排痛点说起

1.1 单轮智能体演进到复杂编排的必然过程

早期的智能体实现方式很简单：一个Prompt，一次LLM调用，让模型自己决定调什么工具、按什么顺序调，也就是现在常说的 ReAct 循环。这种模式在 Demo 阶段跑得很流畅，因为场景足够窄，工具数量少，模型选错的概率低，而且就算选错了，改一版 Prompt 就能糊弄过去。

一旦进入生产环境，问题就冒出来了。我自己踩过几个非常典型的坑：

一是 Token 消耗不可控。一个工单流程如果完全靠模型自由决策，它可能会反复重试、反复读同一个知识库文档、输出大段没有实际意义的中间推理，单次任务的 Token 消耗能比确定性问题高出 3~5 倍。

二是无法稳定复现。同一套 Prompt 在模型版本更新之后，行为可能直接变化。比如原来能稳定调用某个 API，升级之后模型开始"自作主张"改变参数顺序，导致下游系统解析失败。

三是外部系统状态难以对齐。业务系统是强状态的：一个审批节点要么通过、要么驳回，不存在"也许通过"的状态。但模型生成本质上是概率的，它没办法保证每一次都输出合法的流程控制信号。

工作流引擎解决的就是"把流程确定性交给代码、把内容生成能力交给模型"这个问题。在这个思路下，智能体不再是一个大循环，而是拆成一个个清晰的节点：信息收集节点、意图判断节点、业务查询节点、人工确认节点、结果生成节点。节点之间的流转规则由引擎控制，每个节点内部可以调用 LLM、可以调普通代码、可以访问外部服务。

1.2 与现成开源方案对比后，为什么还是选择自研

提到智能体工作流，业内最常被拿来比较的就是 Dify 这类开源智能体平台。Dify 的工作流编排界面做得非常成熟，可视化拖拽、节点类型丰富、社区生态也很好。那为什么我们做 MaxKB4j 的时候还是要自己写一套引擎，而不是直接把 Dify 嵌进来？

核心原因有三个。

第一是部署形态的差异。Dify 是一个完整的平台，前后端、数据库、任务队列一套全带上。但 MaxKB4j 的定位是作为一个 SDK/基础组件嵌入到我们自己已有的业务系统里，而不是重新部署一套独立的平台。平台级方案和组件级方案在集成成本上完全不是一个量级。

第二是节点扩展机制的约束。Dify 的工作流节点是平台内置好的，你想加一个"私有协议对接"节点，要么改 Dify 源码，要么通过它暴露的插件机制，但插件机制的边界是平台定的，灵活度有限。我们的业务场景里有大量存量系统需要对接，协议五花八门，如果节点扩展权不在自己手里，后面每个新对接都要走平台升级流程，太被动。

第三是执行控制粒度的差异。作为平台，Dify 对单个工作流实例的控制比较粗粒度：运行、停止、看日志。但我们希望引擎能提供更细的执行控制能力，比如节点级别的超时设置、条件分支的动态解析、运行轨迹的分层记录，这些需求在没有源码控制权的前提下很难做得彻底。

简单总结一下我们的取舍逻辑：

• Dify 适合"我要快速搭建一套独立可用的智能体应用"
• 自研引擎适合"我已经有一个业务系统，需要把工作流能力作为模块嵌进去"
• 前者快，后者灵活，没有谁更好，只有谁更匹配

1.3 引擎需要具备的四个底层能力

在动手设计之前，我先把引擎必须要有的能力列了出来。这四个能力缺一个，引擎都撑不起真实业务。

第一个是图的描述与校验能力。工作流本质上就是一张有向无环图（DAG），节点是图的顶点，连线是有向边。引擎必须能解析图定义、检查环的存在、识别孤立节点、判断边的连接是否合法。图都描述不清楚，执行阶段必然出乱子。

第二个是节点的注册与生命周期管理能力。每个节点不是硬编码在引擎里的，而是通过注册机制动态加入。引擎只负责调度，节点负责具体执行。注册机制做得够好，新的业务节点才能低成本接入。

第三个是数据流转与作用域隔离能力。上游节点的输出怎么传给下游节点、全局变量和局部变量怎么区分、多个并行分支之间数据会不会互相污染，这些都要有明确的约定。这一步往往最容易被忽略，但恰恰是后期问题最多的。

第四个是运行状态的可观察性能力。工作流跑起来之后，到底是哪个节点在执行、执行了多久、成功还是失败、失败在哪一步、当时的输入输出是什么，全部要能查询。如果没有可观察性，生产环境出了问题就只能靠猜。

这四个能力对应到工程上，分别就是：图模型模块、节点注册中心、上下文数据总线、执行追踪模块。下面几节我会挨个展开讲。

2. 引擎的宏观架构：DAG 驱动的分层设计

2.1 定义层、执行层、运行时的三层职责划分

MaxKB4j 引擎在代码结构上分了三个清晰的层级，每一层只干自己的事，依赖方向是单向的：定义层被执行层依赖，执行层被运行时依赖，反向不引用。

定义层负责描述"工作流长什么样"。一个工作流定义就是一份 JSON 结构，包含节点列表、连线列表、变量定义、全局配置。这一层是纯数据，不含任何执行逻辑。好处是工作流定义可以被序列化到数据库、可以被管理后台编辑、可以被版本管理，也能方便地导出和导入。

执行层负责解析定义、构建内存图对象、进行调度。它拿到定义层的数据后，会做三件事：第一，把 JSON 反序列化为对象模型；第二，校验图的合法性；第三，构建就绪队列和执行上下文。执行层不关心节点内部具体做了什么，只关心节点的生命周期状态。

运行时层负责具体执行节点代码、记录执行日志、触发事件回调、维护执行实例的状态。比如一个"调用LLM"节点，定义层只知道它的类型是 llm、参数是模型名称和提示词模板；执行层把它放进调度队列；真正调 LLM API、处理返回结果的代码，是运行时层的节点执行器。

这个分层让每一层都能独立测试。我们实际开发中最常用的手段就是：定义层用 JSON Schema 校验，执行层用假节点做调度单测，运行时层针对每个具体节点做集成测试。三层面向不同维度的验证，问题定位会快很多。

2.2 节点模型：定义与执行分离的可扩展抽象

节点的设计是整个引擎的灵魂。MaxKB4j 的节点模型把"节点长什么样"和"节点干什么事"拆成了两个对象：NodeDefinition 和 NodeExecutor。

NodeDefinition 是节点的静态描述，包含基本信息。它被用来做工作流的可视化展示、参数表单渲染、JSON Schema 校验。NodeExecutor 是节点的运行时实现，包含真正的执行逻辑。

我用一个简化版的 Java 接口来说明这个设计：

// 节点的静态定义 public class NodeDefinition { private String type; // 节点类型，如 "llm"、"http"、"condition" private String name; // 节点展示名称 private boolean isStart; // 是否为开始节点 private boolean isEnd; // 是否为结束节点 private List<FieldSchema> inputs; // 输入参数的 JSON Schema } // 节点的运行时执行器 public interface NodeExecutor { NodeResult execute(NodeContext context) throws Exception; }

NodeContext 里有引擎注入的各类依赖，包括：

• 当前工作流实例 ID、节点实例 ID
• 全局变量查询接口
• 上游节点的输出结果
• 日志输出接口
• 事件发布器

这样做最大的优势是扩展性极强。新增一个节点类型，只需要实现 NodeExecutor 接口，然后在注册中心登记类型名称，引擎就能识别并调度它。业务系统对接什么新协议，就加什么新节点，完全不用动引擎主流程代码。

2.3 图数据结构的落地：邻接表与入度控制

图的存储我斟酌过两种方式：邻接矩阵和邻接表。工作流图的节点数一般不会很多，几十个就算大型流程了，所以理论上两者都行。但实际我选择了邻接表，因为工作流图天然是稀疏的，绝大多数顶点只连接了少数几个邻居，用矩阵浪费空间不说，遍历也不直观。

定义层里的连线结构很简单：

{ "edges": [ { "source": "node_start", "target": "node_collect_info" }, { "source": "node_collect_info", "target": "node_condition" } ] }

引擎解析时会把边转成两类数据结构：

• outputEdges：记录每个节点的所有出边，用于执行完一个节点后找下一批候选节点
• inDegree：记录每个节点的入度数量，用于调度判活

有了入度信息，调度逻辑就很清晰了：入度为 0 的节点就是可以立即执行的节点。节点执行完成之后，把它所有出边对应的下游节点入度减 1，如果减到 0，就把下游节点加入就绪队列。

图校验方面，最重要的就是检测环。实现上我用的是拓扑排序：能完成拓扑排序的图是无环的，反之就有环，直接拒绝执行。这个校验必须在引擎启动加载工作流定义时就做，不能等到运行时。试想一个环在工作流里意味着什么——A 节点等 B，B 节点等 C，C 节点又等 A，整个执行会直接卡死，而且这种问题在界面上还特别难发现。

3. 节点生命周期与执行器机制

3.1 节点的五状态状态机与容错转移

每个节点实例在引擎里都会经历状态变化。MaxKB4j 的节点实例有五个状态：PENDING、RUNNING、SUCCEEDED、FAILED、SKIPPED。

节点刚被调度器选中但还没开始执行时，是 PENDING。执行器开始跑 execute 方法的那一刻，变成 RUNNING。执行方法正常返回，变成 SUCCEEDED。抛异常或者超时，变成 FAILED。条件分支判断为不满足的路径上的节点，被标记为 SKIPPED，不执行任何逻辑。

状态转移规则如下：

这个状态机看起来简单，但工程上有个细节很关键：FAILED 之后节点要不要自动重试。我做了可配置的策略，因为并不是所有失败都值得重试。网络抖动类错误可以重试 2~3 次，带退避间隔；参数错误、权限错误这类确定性错误，重试一万次也没用，直接标记 FAILED 并触发工作流失败回调。

3.2 上下文传递与变量作用域约定

工作流引擎的上下文设计，直接影响到节点的编写复杂度和排错难度。MaxKB4j 在上下文设计上遵循一个原则：全局变量只读，局部变量显式引用。

全局变量放在一个 ImmutableMap 里，引擎初始化工作流的时候注入进去。业务系统触发工作流时传入的初始参数、上一个节点的全部输出，都会被固化在全局变量中。

局部变量则是通过变量引用字符串来实现的。比如 A 节点输出了一个字段叫 orderId，下游 B 节点想读取，就配置${nodes.node_a.orderId}。引擎在执行 B 节点前，会去解析这个引用，从 A 节点的输出里取值再传入 B 节点的输入。如果引用路径不存在，就直接报错，绝不静默返回 null。

为什么不做隐式的"所有上游输出都注入所有下游节点"这种设计？因为副作用太大。一是并行分支场景下，两个分支的节点输出谁先谁后没有确定性，下游不知道拿谁的值；二是大量数据堆积在上下文里，单个节点处理不过来时内存会爆。显式引用虽然写起来啰嗦一点，但可预测、可定位、可审计。

3.3 分支、循环与并行：三种控制流的实现

控制流是工作流引擎和普通任务队列最大的区别。MaxKB4j 的引擎实现了三种控制流模式。

条件分支是第一种。典型场景是"如果用户输入的订单号以 10 开头走退款流程，否则走人工客服"。实现方式是：分支节点的输出连接了多条出边，每条边上配置一个条件表达式。节点执行完成后，引擎对每条出边的表达式求值，值为 true 的边被激活，下游节点入度减 1；值为 false 的边不激活，下游节点直接标记为 SKIPPED，并把这种情况也记录到日志里。

并行是第二种。一条出边同时连接多个下游节点时，这些节点理论上可以同时执行。实现上我用了一个线程池，每个下游节点提交为一个独立任务。等所有并行节点都完成之后，汇合节点再继续往下走。汇合时要注意的点是：分支里某些节点可能是 SKIPPED 状态，汇合逻辑应该只等待真正执行了的节点，避免死等不存在的执行结果。

循环是第三种。虽然主图是 DAG，不允许出现环，但真实业务里"循环调用某个 API 直到拿到最终结果"这种需求大量存在。我的做法是提供一个专门的"循环节点"，它内部维护一个 MAX_ITERATIONS 上限，在节点内部做循环逻辑，而整个图上依然是一个节点。这样既保证了 DAG 约束不会被破坏，又满足了实际需求。

4. 核心实现：从定义到运行的完整链路

4.1 工作流定义的 JSON Schema 设计

工作流定义是引擎的输入，它的 Schema 直接决定了系统的表达能力和易用性。我提供一份精简后的核心结构：

{ "workflowId": "wf_001", "title": "售后退款处理流程", "variables": { "input": { "orderId": "string", "userId": "string" } }, "nodes": [ { "id": "node_start", "type": "start", "name": "开始" }, { "id": "node_cond", "type": "condition", "name": "是否退款订单", "config": { "expression": "payload.refundable == true" } }, { "id": "node_http", "type": "http", "name": "调用退款接口", "config": { "method": "POST", "url": "${env.refund_service_url}/api/refund", "body": { "orderId": "${nodes.node_start.output.orderId}" } } } ], "edges": [ { "source": "node_start", "target": "node_cond" }, { "source": "node_cond", "target": "node_http" } ] }

Schema 设计上有三个细节值得说一下。

一是 config 字段用了 OpenAPI 风格的对象结构而不是传统的扁平 key-value，这样每个节点的参数都带上了类型信息，引擎可以统一做法在校验阶段检查类型对不对。

二是变量表达式我用了${}模板语法。在执行器拿到 config 值的时候，引擎会先把所有包含${}的字符串做一次插值解析，再传给节点。这样节点执行器自己不需要关心变量从哪来，它拿到的已经是最终值，实现大大简化。

三是这段 JSON 里没有直接体现"分支不满足时走到哪"，因为分支路径是通过 edge 上的条件表达式表达的，而不是靠节点里配置 target 字段。这种设计让图画起来更像实际流程图，可读性好很多。

4.2 调度引擎的核心实现思路

调度是整个引擎最核心的部分。我贴一段简化后的调度器核心逻辑：

public class WorkflowScheduler { private final ExecutorService executor; private final Map<String, WorkflowNode> nodeMap; private final Map<String, Integer> inDegree = new ConcurrentHashMap<>(); private final Queue<String> readyQueue = new ConcurrentLinkedQueue<>(); public void schedule(WorkflowGraph graph) { // 1. 初始化入度表 for (String nodeId : graph.getNodeIds()) { inDegree.put(nodeId, graph.getInDegree(nodeId)); if (inDegree.get(nodeId) == 0) { readyQueue.offer(nodeId); } } // 2. 循环消费就绪队列 while (!readyQueue.isEmpty()) { String nodeId = readyQueue.poll(); WorkflowNode node = nodeMap.get(nodeId); executor.submit(() -> runNode(node)); } } private void runNode(WorkflowNode node) { try { NodeResult result = node.getExecutor().execute(node.getContext()); markSuccess(node.getId(), result); } catch (Exception e) { markFailed(node.getId(), e); return; } // 3. 节点成功后，更新下游入度 for (WorkflowEdge edge : node.getOutputEdges()) { int remain = inDegree.computeIfPresent(edge.getTarget(), (k, v) -> v - 1); if (remain == 0) { readyQueue.offer(edge.getTarget()); } } } }

这里有一个非常容易踩坑的点：并行场景下的入度减少必须做原子操作。如果两个上游节点同时执行完成，同时去更新同一个下游节点的入度，不用原子类的话就会丢更新，导致入度错误地停留在 1，下游节点永远等不到调度。我在第一版实现里就踩过这个坑，Debug 了整整一天才定位到，后来全部换成了 computeIfPresent 这个原子方法。

线程池的参数也值得单独说。LLM 调用、HTTP 请求这类节点全都是 IO 密集型的，线程池核心线程数如果设置成 CPU 核数，IO 等待时整个调度就会卡住。我最后设置为核心线程数 16、最大线程数 64、队列容量 512，核心线程空闲回收时间 60 秒。这个参数是根据我们平均工作流节点数 15~20 个、单个节点平均耗时 1~3 秒测算出来的，能够支撑约 10 个工作流同时并行执行而不互相阻塞。

4.3 条件分支中的表达式求值

表达式引擎的选择我斟酌了很久。最初考虑直接用现成的 Spring Expression Language（SpEL），功能足够强大，但问题也出在强大上——SpEL 支持调用任意 Java 方法，一旦工作流定义是可以被业务人员编辑的，表达式就变成了一个远程代码执行漏洞。

后来我改成了自研的轻量表达式解析器，只支持有限的语法子集：

• 字面量：字符串、数字、布尔值、null
• 变量引用：如payload.refundable、nodes.node_http.output.code
• 比较运算：==、!=、>、<、>=、<=
• 逻辑运算：&&、||、!
• 可选的一元取反

这个解析器的实现并不复杂，核心就是分词 + 递归下降解析 + 求值。分词器把表达式拆成 Token，解析器按优先级构建抽象语法树，求值时从上下文取变量值做运算。

实际落地时有个性能优化点：一个工作流可能被触发成千上万次，如果每次执行都重新解析一遍表达式，开销完全没必要。我引入了表达式缓存，把解析好的 AST 缓存在 ConcurrentHashMap 里，key 是表达式字符串本身。实测下来，表达式求值的平均耗时从最初的 3 毫秒降到了 0.1 毫秒以内。

5. 扩展实践：如何接入一个新的业务节点

5.1 实际步骤：从需求到实现一个"发送短信"节点

光讲原理不够，我拿一个实际例子走一遍完整接入流程。假设现在业务方要加一个节点：发送短信验证码。

第一步，实现 NodeExecutor。短信服务是一个已有的内部 RPC，我只需要把参数取出来传进去：

public class SmsSendExecutor implements NodeExecutor { private final SmsClient smsClient; @Override public NodeResult execute(NodeContext context) { String phone = context.getConfig("phone"); String content = context.getConfig("content"); String bizId = smsClient.send(phone, content); return NodeResult.success(Map.of("bizId", bizId)); } }

第二步，注册节点。我在配置文件里登记节点类型 sms_send，同时提供输入参数 Schema：

{ "type": "sms_send", "name": "发送短信", "inputs": [ { "field": "phone", "type": "string", "required": true }, { "field": "content", "type": "string", "required": true } ] }

注册完成之后，工作流编辑器里就能看到这个节点，业务人员可以直接拖拽使用。整个过程不需要改引擎任何代码，这就是节点注册机制的价值。

5.2 可视化执行轨迹：可观测性的实际落地

工作流引擎如果只输出一堆日志，生产环境根本没法用。我专门做了执行轨迹记录模块，每次执行都会留下三条层级的数据。

流程级别的记录：整个工作流的开始时间、结束时间、耗时、最终状态。节点级别的记录：每个节点的开始时间、结束时间、状态、出参摘要。事件级别的记录：节点内部打点输出的关键事件，比如 LLM 调用的请求参数和返回结果。

这套设计在执行链路里的效果非常直观。工作流出问题的时候，我只要打开执行记录列表，找到那一条失败记录，放大看是哪个节点挂了，再看它当时的入参出参，基本几分钟内就能定位问题。如果哪天模型返回格式变了导致解析失败，执行记录里能清楚地看到 LLM 节点的输出内容和解析错误日志，不需要去后端系统里翻原始请求。

5.3 超时、并发与资源保护的工程取舍

最后一个实操主题是工程的自我保护。工作流引擎不像普通接口，一个流程可能跑几分钟，如果中间有节点是恶意慢调用，整个链路都会卡住。

我做的第一个保护是节点级超时。每个节点在执行前会计算一个截止时间，执行器线程用 CompletableFuture 包装，到了截止时间还没返回就直接取消并标记 FAILED。HTTP 节点默认超时 30 秒，LLM 节点默认 60 秒，这些都是根据实际服务的 P99 延迟来标定的。

第二个保护是图规模限制。单个工作流的节点数量上限设为 100 个，边的数量上限 200 条。业务方反馈过"我的流程需要 150 个节点怎么办"，我的回复是拆成多个子流程，用子流程节点串联。因为图一旦超过这个规模，可视化界面排布和引擎调度的复杂度都会急剧上升，收益完全覆盖不了成本。

第三个保护是并发工作流实例的数量限制。每个工作流定义都有各自的并发上限，默认是 20 个同时执行。超过上限的触发请求直接返回"当前排队中"，由业务侧决定是排队等还是提示用户稍后再试。这个限流措施防止了某个高频业务把整个引擎线程池打满。

就我个人的实际体验来说，做工作流引擎难度最大的反而不是调度算法本身，而是给能力边界做减法。节点的语义粒度定得太粗，节点之间逻辑耦合严重；定得太细，一个简单流程要画十几二十个节点，业务方根本不想用。最终我找到的相对平衡点是：一个节点应该完成一个"有明确产出、可独立验收"的动作，比如调用一个 API、生成一段文本、读一次知识库，而不是把多个动作硬捏在一个节点里。这样画流程图的人在界面上看到的内容，和实际执行时发生的事情才是一致的，排查问题的时候也才能对着流程一路顺下来。