尧图网站建设 
不知道你有没有在这种场景里崩溃过一个Agent流程先调用大模型理解用户意图再根据意图决定要不要调某个工具工具返回之后再让大模型做总结最后把结果推给浏览器。需求方说“很简单的逻辑嘛”你点点头打开代码一开始还挺清爽后边就变成了十几个if-else嵌套再后来你开始在里面塞状态变量、补异常分支、做重试代码越来越长新同事看一眼就想跑路。我最近在做AI Agent的时候也被这种写法折磨过一轮。后来实在忍不了用Java从0到1自己撸了一套流程引擎把Agent工作流里那些“先做A、再看结果做B、C失败就重试、D拿到全部数据后边生成边推给前端”的场景全部收进节点状态轮转和流式输出里。做完之后再看原来的if-else代码我只想说一句真的太Low了。这篇文章就把整个思路和实现细节完整拆开。适合两种人看一种是被Agent流程控制搞得头皮发麻的Java后端另一种是想自己设计流程引擎、又不知道从哪里下手的同学。我会把抽象模型、节点状态机、流转表、Flux流式输出的链路、主线程调度和真实踩坑全部写出来代码可以直接抄。1. 为什么把Agent流程塞进状态机if-else写法到底哪里不对劲1.1 一个五步Agent流程的if-else灾难先还原一下典型的Agent业务代码。最开始可能是这样// 第一步调用大模型理解意图 String intent llm.chat(userInput); if (search.equals(intent)) { // 第二步调用搜索工具 ListString results searchTool.query(userInput); if (results.isEmpty()) { // 第三步没有结果怎么办直接返回兜底文案 return 没查到; } // 第四步用大模型总结搜索结果 String answer llm.summarize(results); // 第五步返回给前端 return answer; } else if (translate.equals(intent)) { ... } else { ... }这段代码在流程只有三四步的时候还能忍。等你要加“重试策略”“超时兜底”“中间结果上报”“失败后人工介入”“某个节点要并发执行”这些需求的时候if-else就会开始失控。我见过最夸张的版本一个方法里写了二十多个状态变量每个if分支里还套着三个try-catch根本没法改。这里面最核心的问题不是代码多而是流程的状态被散落在局部变量和if分支里。你没法在任何一个时刻回答“当前这个流程跑到哪一步了”“下一步有哪些分支可选”“失败之后该退回到哪”因为没有一张全局的状态表只有一团纠缠的逻辑。1.2 Agent工作流的三个刚性需求Agent和普通接口最大的区别在于它的运行时间远超普通HTTP请求。用户提一个问题大模型生成可能要几十秒中间还要调工具、看结果、再生成。这个时候会冒出三个绕不开的需求异步长任务不可能让HTTP请求一直挂着等所有节点跑完需要让流程“跑起来之后可查询、可等待、可推送”。动态分支大模型的输出是概率性的同一个问题可能今天走搜索分支明天走总结分支。分支决策必须由运行时的节点结果决定而不是写死在代码里。流式输出与可观测用户等一个完整的LLM回答等不了几十秒必须让生成的Token像打字机一样持续流到前端。同时平台运营要能看到每个节点现在是什么状态、卡在哪一步。这三个需求用if-else基本不可能优雅实现。你需要一个真正意义上的流程引擎。1.3 流程引擎本质状态机加事件驱动用一句话概括就是把流程拆成节点节点之间有明确的连边关系每个节点都有自己的生命周期状态引擎通过事件驱动节点状态轮转直到整个流程终结。状态机保证“什么状态能转成什么状态”是有约束的不会出现从PENDING直接跳到COMPLETED这种非法情况。事件驱动保证“某个节点完成”这个消息能被引擎捕获自动触发下一个节点的调度。这就是Agent工作流引擎的地基。2. 引擎的第一块地基节点模型、上下文与状态机的抽象设计2.1 节点(Node)一个“动作加决策”的抽象第一步把流程里每一步都抽象成节点。我不喜欢把节点设计成纯粹的“动作”因为Agent流程里很多节点的输出会影响下一步走哪条分支。所以我的节点定义是这样的public interface Node { String getId(); NodeType getType(); void execute(NodeContext ctx, StreamSink sink); }NodeType先分成三类ACTION普通动作节点比如“调用LLM”“查数据库”“调搜索工具”。它执行完就结束输出会写进Context。CONDITION条件分支节点。它不做什么重活只是读取Context里的数据根据规则决定流程接下来走哪条边。END流程终点。一个节点对应一个类每个Agent流程就是一组Node实例和边的组合。以后要调整流程不需要改一堆if-else只要重新组装节点和边就行。2.2 边(Edge)连接节点让流程走起来节点只负责“干活”怎么串联是Edge的事。我用的边设计比较简单每条边有来源节点、目标节点、优先级和一个条件表达式public class Edge { private final String fromNodeId; private final String toNodeId; private final BiPredicateNodeContext, NodeResult condition; private final int order; }引擎在某个节点完成之后会读取这个节点所有出边按order排序后依次用condition判断命中第一条条件满足的边就把流程推到下一个节点。这样条件判断本身就是可配置的新增一条分支完全不需要动引擎代码。2.3 上下文(Context)一块贯穿全流程的“黑板”Agent流程里所有节点之间要共享数据。大模型第一步产出的意图第二步要根据意图查工具第三步要根据工具结果做总结。这些数据不能靠方法参数一层层传所以需要一个贯穿整个流程的Context。我的Context本质是一个线程安全的Map但做了一些封装public class NodeContext { private final ConcurrentHashMapString, Object data new ConcurrentHashMap(); private final String flowId; private final MapString, Object metadata; public T T get(String key) { ... } public void set(String key, Object val) { ... } public T T getOrThrow(String key, ClassT type) { ... } }之所以叫“黑板模式”是因为每个节点都可以在上面写信息、读信息不需要知道信息是谁写的。这让节点之间的耦合度降到最低。注意所有读写要走ConcurrentHashMap因为后面并发调用工具时多个线程可能同时读写Context。2.4 节点状态和流程状态两个维度不能混这里特别容易搞混。我一开始也把节点状态和流程状态混在一个枚举里结果引擎根本没法判断“是某个节点失败还是整个流程失败”。实际设计必须拆开节点状态PENDING、READY、RUNNING、SUCCESS、FAILED、SKIPPED、TIMEOUT流程状态ACTIVE、SUSPENDED、COMPLETED、FAILED节点状态是流程状态的基础。一个节点失败流程到底算FAILED还是走重试分支由引擎策略决定但节点本身的状态必须是FAILED不能直接跳到COMPLETED。2.5 为什么显式建模比硬编码灵活显式建模最大的好处是流程本身变为了数据。节点、边、状态都是可以实例化、可持久化、可视化的对象。你可以把当前流程定义序列化成JSON存库可以在前端画一个流程图展示当前卡在哪个节点可以让运营在不改代码的前提下调整分支条件还可以在流程失败后从某个节点恢复重跑。这三样if-else任何一样都做不到。3. 节点状态轮转的真正实现枚举状态机加流转表不是一堆switch3.1 用枚举而不是一个int字段很多老代码喜欢用int常量表示状态0是初始1是运行中2是成功。这种写法最大的问题是魔法数字满天飞而且编译器完全不会帮你检查状态是否合法。Java里最自然的选择就是枚举。public enum NodeStatus { PENDING, // 刚创建还没进调度队列 READY, // 前置条件满足可以执行 RUNNING, // 正在执行中 SUCCESS, // 执行成功 FAILED, // 执行失败 SKIPPED, // 被跳过比如条件分支走了另一边 TIMEOUT // 执行超时 }枚举相比int的好处不只是可读性它还能天然承担状态机的核心逻辑。后面我会介绍如何用EnumMap做流转表。3.2 流转表所有合法流转全部收进一张表这是整个引擎最核心的思想把“当前状态能转到哪些状态”显式声明成一张表。代码里不再出现任何“if (status RUNNING next SUCCESS)”这种判断而是查表。import java.util.EnumMap; import java.util.EnumSet; import java.util.Map; import java.util.Set; public class NodeStateMachine { private static final MapNodeStatus, SetNodeStatus TRANSITIONS new EnumMap(NodeStatus.class); static { // PENDING只能变成READY或SKIPPED TRANSITIONS.put(NodeStatus.PENDING, EnumSet.of(NodeStatus.READY, NodeStatus.SKIPPED)); // READY可以开始执行也可以被跳过 TRANSITIONS.put(NodeStatus.READY, EnumSet.of(NodeStatus.RUNNING, NodeStatus.SKIPPED)); // RUNNING只允许三种终局成功、失败、超时 TRANSITIONS.put(NodeStatus.RUNNING, EnumSet.of(NodeStatus.SUCCESS, NodeStatus.FAILED, NodeStatus.TIMEOUT)); // 终态不再允许流转 TRANSITIONS.put(NodeStatus.SUCCESS, EnumSet.noneOf(NodeStatus.class)); TRANSITIONS.put(NodeStatus.FAILED, EnumSet.noneOf(NodeStatus.class)); TRANSITIONS.put(NodeStatus.SKIPPED, EnumSet.noneOf(NodeStatus.class)); TRANSITIONS.put(NodeStatus.TIMEOUT, EnumSet.noneOf(NodeStatus.class)); } public static boolean canTransition(NodeStatus current, NodeStatus next) { SetNodeStatus allowed TRANSITIONS.get(current); if (allowed null) { return false; } return allowed.contains(next); } public static void checkTransition(NodeStatus current, NodeStatus next) { if (!canTransition(current, next)) { throw new IllegalStateException( 非法状态流转: current - next); } } }这种“表驱动”的思路本质上是把状态流转规则从代码里抽离出来变成一张可读性极强的配置表。它的好处有三个非法流转在运行前就能暴露。假设有人把节点状态从PENDING直接改成SUCCESS引擎会立刻抛异常而不是带着坏状态继续跑。流程规则变更时只需要改表里的枚举组合。比如需求方说“超时之后允许重试”你就在TIMEOUT对应的集合里加上READY。可以写单元测试对整张表做全量验证。后面我会专门讲这个测试技巧。3.3 状态变更钩子状态一动回调就触发只有流转表还不够。引擎需要在状态变化的瞬间做一些副操作比如更新数据库、记录日志、给前端推送状态。这就需要在状态机里加钩子。public class NodeStateMachine { private static final MapNodeStatus, ListConsumerNodeContext AFTER_CHANGE_HOOKS new EnumMap(NodeStatus.class); public static void registerAfterHook(NodeStatus status, ConsumerNodeContext hook) { AFTER_CHANGE_HOOKS.computeIfAbsent(status, k - new ArrayList()).add(hook); } public static void fireAfterHook(NodeStatus newStatus, NodeContext ctx) { ListConsumerNodeContext hooks AFTER_CHANGE_HOOKS.get(newStatus); if (hooks ! null) { for (ConsumerNodeContext hook : hooks) { hook.accept(ctx); } } } }实际使用中我会把状态变更封装成一个方法调用方只传“目标状态”方法内部先查表校验再改节点对象的状态最后触发钩子。钩子里最常见的就是写日志和发WebSocket事件。3.4 节点状态机和流程状态机联动现在进入关键环节节点状态怎么带动流程状态。引擎主循环的逻辑很简单当一个节点进入SUCCESS或SKIPPED状态时引擎就去节点对应的出边集合里找下一个节点如果下一个节点存在把它从PENDING变成READY再投递到执行队列如果已经没有出边了说明流程到头把流程状态从ACTIVE变成COMPLETED。这里有一个容易踩的坑节点状态变成SUCCESS之后并不代表整个流程就该继续。比如条件分支节点变成SUCCESS只表示“条件判断完成”真正的下一步在它选中的那条边上。所以节点状态机的驱动逻辑和流程状态机的驱动逻辑必须分开写先由节点状态触发边选择再由边选择触发下一个节点的调度。3.5 启动时全表自校验让非法状态流转无处可藏这里分享我自己的一个习惯也是这篇文章里最想让你记住的一招。在Spring Boot项目启动时或者引擎初始化时写一个方法遍历所有节点状态对流转表做全量自校验确保每个状态都能通过最终走向终态没有死循环没有孤立状态。public class StateMachineValidator { public void validate() { for (NodeStatus status : NodeStatus.values()) { SetNodeStatus visited new HashSet(); assertReachableTerminal(status, visited); } } private void assertReachableTerminal(NodeStatus current, SetNodeStatus visited) { if (!visited.add(current)) { throw new IllegalStateException(出现了环形流转当前状态: current); } SetNodeStatus allowed TRANSITIONS.get(current); if (allowed null || allowed.isEmpty()) { return; // 终态 } for (NodeStatus next : allowed) { assertReachableTerminal(next, visited); } } }这个方法看起来简单但真的能救命。我曾经漏配了READY到SKIPPED的流转结果条件分支不满足时流程直接卡死用户界面永远停在“分析中……”。有了这个自校验这类问题在启动阶段就会被发现而不是等到线上用户来骂。4. 流式输出的那点事事件流模型与Flux实现细节4.1 为什么要流式输出如果做完所有节点把最终答案一次性返回Agent流程的用户体验会非常糟糕。大模型生成一个500字的回答可能耗时20秒用户在浏览器里盯着空白页20秒不知道是挂了还是在思考。流式输出就是解决这个问题LLM每产出一个Token就立刻推到前端用户看着文字一个字一个字蹦出来体感上会快非常多。在Agent流程引擎里流式输出比单次LLM调用更复杂因为中间可能穿插多个节点。比如先输出“正在搜索资料……”这个事件再输出“搜索到3条结果”再开始流式输出总结文本。这些事件需要一个统一的事件流模型来承载。4.2 选型SSE、WebSocket还是WebFlux我把选择分成两层内部引擎层用Reactor Flux处理节点产生的事件流因为Flow从一个节点到另一个节点的传递用Flux极其自然。对外HTTP层用SSEServer-Sent Events把Flux映射给前端浏览器。SSE比WebSocket更适合这个场景原因是Agent流的消息方向是单向的服务端往客户端推客户端几乎不需要往回推送控制指令。SSE天然支持重连前端用EventSource就能监听不需要引入WebSocket的状态管理。如果后续确实需要双向交互再在网关层把SSE升级为WebSocket也不难。4.3 事件流模型从Sink到Flux我在引擎里定义了一个极简的流式接口叫Sink专门用来承载节点产生的事件。public interface StreamSink { void onNext(Chunk chunk); void onComplete(); void onError(Throwable t); }Chunk是流事件的最小单位本质是一个携带类型和数据的对象public class Chunk { private final ChunkType type; // TEXT, STATUS, NODE_BEGIN, NODE_END private final String nodeId; private final String content; // 还有时间戳、token总数等字段这里省略 }每个节点在执行过程中都可以向StreamSink发射多个Chunk。比如LLM调用节点每收到一个Token片段就调用一次sink.onNext(new Chunk(TEXT, nodeId, token))。普通动作节点可以只发射一个STATUS类型的Chunk告诉前端“我现在开始执行了”。引擎内部用Reactor的Sinks类把多个节点的Sink统一成一个Flux这样入口方法的返回值就是Flux 既支持背压又方便后续做SSE映射。4.4 用Sinks.Many实现流式管道直接上代码。引擎的入口方法大概是这个形态public FluxChunk startFlow(FlowDefinition flowDef, MapString, Object initData) { Sinks.ManyChunk sink Sinks.many().unicast().onBackpressureBuffer(); FluxChunk flux sink.asFlux(); // 异步启动整个流程把事件写入sink scheduler.execute(() - runFlow(flowDef, initData, sink)); return flux; }runFlow在执行过程中把收到的Chunk挨个写入sink直到流程结束后调用sink.tryEmitComplete()。如果某个节点抛出异常调用sink.tryEmitError(ex)。注意我用的Sinks.many().unicast()表示只能订阅一次这正好适合一个HTTP请求对应一个SSE流。如果同一个流程想推送给多个订阅者那就得换成replay()或multicast()不过实际项目中一个请求对应一个订阅者的场景占绝大多数。4.5 Token如何从LLM节点流到浏览器把整条链路串起来看一个LLM调用节点从发出请求到前端展示大概经过这么几步用户请求进入引擎引擎为这个流程创建一个唯一的flowId初始化NodeContext。第一个节点是ACTION节点内部调用LLM SDK。现在的LLM SDK基本都支持流式接口返回的是Flux 。节点拿到Flux 后每来一个Delta就通过StreamSink发射一个Text类型的Chunk。引擎把这些Chunk实时写入内部Sinks从而驱动外层Flux 。Controller层把Flux 映射成Server-Sent Events直接response.flush()。浏览器EventSource收到消息把文字追加到页面上。这一步的关键在于你在节点代码里不能把LLM SDK的流“攒起来”等结束再发而必须把LLM的Flux直接映射到StreamSink保持整条流是实时贯通的。5. 主循环、并发与超时执行引擎的最后一公里5.1 主循环取事件、查流转表、调度节点有了前面两章的设计引擎的主循环反而不复杂。它本质上是一个不断处理事件的循环private void runFlow(FlowDefinition flowDef, NodeContext ctx, StreamSink sink) { // 找到起始节点 Node currentNode flowDef.getStartNode(); transitionNode(currentNode, NodeStatus.READY); while (!currentNode.getType().isEnd()) { transitionNode(currentNode, NodeStatus.RUNNING); NodeResult result doExecute(currentNode, ctx, sink); Edge nextEdge flowDef.findNextEdge(currentNode, ctx, result); if (nextEdge null) { sink.onError(new IllegalStateException(没有可用流转边: currentNode.getId())); return; } Node nextNode flowDef.getNode(nextEdge.getToNodeId()); if (nextNode null) { sink.onError(new IllegalStateException(目标节点不存在: nextEdge.getToNodeId())); return; } // 当前节点进入SUCCESS下一个节点进入READY transitionNode(currentNode, NodeStatus.SUCCESS); transitionNode(nextNode, NodeStatus.READY); currentNode nextNode; } transitionNode(currentNode, NodeStatus.SUCCESS); sink.onComplete(); }注意这里的doExecute不是直接同步调用而是把任务提交给线程池执行并携带超时控制。后面马上讲。5.2 并发模型单线程事件循环加线程池执行节点我建议的并发模型是这样的不共享状态变量的多线程粗放并发而是给整个流程配一个调度线程池节点执行走工作线程池。具体来说每个流程实例由引擎的主循环调度线程负责推进状态、查边、决定下一个节点。节点本身的耗时代码比如调LLM、调工具提交到独立的ExecutorService执行。主循环用Future.get(timeout)等待节点结果确保不会因为某个节点卡死而导致整个流程无法推进。不同流程实例之间天然并行因为每个实例都自己持有Context互不干扰。这个模型看起来简单但非常实用。它既避免了全局Context被多线程竞争又能利用工作线程池处理耗时任务。5.3 超时处理Future加线程池是标配超时是Agent流程里一定要处理的。LLM接口偶发变慢是常态某个搜索工具挂掉更是不可避免。我用ScheduledExecutorService加Future实现超时控制private NodeResult executeWithTimeout(Node node, NodeContext ctx, ExecutorService workerPool, long timeout, TimeUnit unit) throws Exception { FutureNodeResult future workerPool.submit( () - node.execute(ctx, sink) ); try { return future.get(timeout, unit); } catch (TimeoutException e) { future.cancel(true); transitionNode(node, NodeStatus.TIMEOUT); throw new NodeTimeoutException(node.getId(), timeout); } }超时之后节点状态变成TIMEOUT流转表里允许从TIMEOUT转回READY这就给重试留了口子。5.4 重试机制不能无脑重试必须考虑幂等有了超时和TIMEOUT状态重试就顺理成章了。我通常会在Edge或Node上配置重试策略三要素是最大重试次数、重试间隔、幂等性要求。重试间隔可以用简单的指数退避。幂等性这块要特别小心后面踩坑部分我会讲一个真实案例这里先提醒一句调LLM接口重试是要花钱的必须要用requestId做幂等键或者明确业务上允许“三次请求都可能产生Token”。5.5 持久化与崩溃恢复如果Agent流程运行时间很长比如一个自动写研报的Agent可能要跑几分钟甚至更久中途服务重启就会把流程丢光。所以引擎必须支持状态持久化。实际方案不复杂节点每次状态变更时把最新的NodeStatus和Context快照写入数据库。引擎启动时提供一个“未完成任务恢复”方法扫描状态不是终态的流程实例从最后一个成功的节点继续执行。Context里必须放进一个可序列化版本这样重启后还能继续读取之前节点产出的数据。我是用一张简单的flow_instance表加一张node_state表实现的字段包括flow_id、node_id、status、context_json、create_time、update_time。handling重启恢复时把状态机流转表的校验逻辑再做一遍确保不会从非法状态继续跑。6. 真实Agent场景接入与踩坑记录LLM调用、工具节点、分支判断6.1 一个自动写研报的Agent实战案例纸上谈兵没意思直接看一个能跑通的真实链路。假设我们要做一个“行业研报助手”用户输入一个行业名称引擎自动执行节点A调用LLM对大模型说“请提取该行业的核心关键词并输出JSON”。节点B条件分支解析节点A输出的JSON如果关键词为空直接走EndNode结束否则走下一步。节点C调用数据查询工具根据关键词查询最近一周的研报数据。节点D调用LLM把查询结果整理成流式研报。关键代码长这样public class ExtractKeywordsNode implements Node { Override public void execute(NodeContext ctx, StreamSink sink) { String industry ctx.getOrThrow(industryName, String.class); sink.onNext(new Chunk(ChunkType.STATUS, getId(), 正在提取行业关键词...)); FluxDelta deltaFlux llmClient.streamCompletion( 提取 industry 的核心关键词输出JSON数组 ); StringBuilder sb new StringBuilder(); deltaFlux.doOnNext(delta - { sb.append(delta.getContent()); // 关键词提取阶段不需要实时流给前端只攒结果 }).blockLast(); ctx.set(keywords, parseKeywords(sb.toString())); sink.onNext(new Chunk(ChunkType.STATUS, getId(), 关键词提取完成)); } }节点B是条件分支节点它读取Context里的keywords字段判断是否为空然后通过condition返回true或false引擎据此找到对应的边。6.2 工具调用节点让Agent具备行动能力工具调用节点在实际项目里最复杂。搜索、查库、调外部API每个工具的参数格式都不一样响应时间也不一样。我的做法是给所有工具做一个统一包装public interface ToolExecutor { String getToolName(); ToolResult execute(MapString, Object args, NodeContext ctx, StreamSink sink); }工具节点执行时先发一个“开始调用工具”的Chunk然后把参数传给ToolExecutor拿到结果之后写回Context。如果工具本身支持流式比如命令行的实时日志输出同样可以映射到StreamSink。这里要提一点工具调用必须考虑鉴权和参数校验不能把用户输入原样拼进SQL或Shell命令里。Agent流程引擎本身不替你做安全校验但如果你的Agent会调用外部工具这是必须自己补上的防线。6.3 分支节点让流程真正“智能”Agent最迷人的地方是流程可以自己决定下一步。LLM节点输出一段结构化JSON条件分支节点解析JSON后决定走哪条边。我用了一个很轻量的实现方式条件分支节点的condition直接写成一个Java lambda从Context读取LLM产出这样不需要解析器灵活度反而最高。Edge searchEdge Edge.builder() .fromNodeId(extractKeywords) .toNodeId(searchReports) .condition((ctx, result) - { ListString keywords ctx.get(keywords); return keywords ! null !keywords.isEmpty(); }) .order(1) .build(); Edge endEdge Edge.builder() .fromNodeId(extractKeywords) .toNodeId(end) .condition((ctx, result) - ctx.get(keywords) null) .order(2) .build();这种写法的好处是分支条件和业务流程本身彻底解耦要调整分支逻辑只需要改Edge的condition节点代码完全不用动。6.4 踩坑记录这些坑我替你踩过了把我在开发过程中遇到的三个大坑和解决思路整理成表格比直接背结论更有用。坑现象根因解决流式输出背压没有控制用户量大时后端内存暴涨最终OOMSinks创建时用了onBackpressureBuffer但没有限制缓冲区大小LLM产生的Token太多时全部积压在内存里改用onBackpressureBuffer(1000)并给前端SSE配置合理的超时时间量再大就上Kafka或Redis Stream做削峰状态轮转过早节点执行完成但流式事件还没有发射完前端数据缺失节点代码里先调了sink.onComplete()然后才调用工具持久化结果严格约定onComplete必须是节点执行的所有业务逻辑结束之后才能调用放在execute方法最后一行超时重试导致LLM重复扣费一个请求因超时重试了三次LLM产生了三份费用重试前没有幂等控制每次重试都发全新请求在NodeContext里记录外层requestIdLLM SDK加上请求级别的幂等键重试次数最多一次且必须确认上一次请求确实没有进入生成阶段这些坑在文档里基本不会写明属于实际跑业务才会遇到的教训。特别是LLM重复扣费那个建议所有做Agent的人提前把幂等键设计好不然上生产环境第二天账单会很难看。6.5 测试与可视化状态机表驱动带来的附加红利最后聊几句测试。因为用了流转表状态机的测试覆盖率可以做到几乎百分百。我写了一个测试类遍历所有NodeStatus把每个合法流转都执行一遍断言不会抛异常再把所有非法流转也遍历一遍断言必定抛异常。这两组测试跑完状态机的正确性就有了基本保障。流程可视化也因为这个设计变得顺理成章。把完整流程图定义序列化成JSON用前端组件渲染出来再通过WebSocket推送每个节点的实时状态前端就能画出一张“动态走光的流程图”。这个功能在给业务方演示的时候效果极好也方便排查线上问题——一眼就能看到流程卡在哪个节点。尾声我把这套引擎拆完之后的几点体会整篇文章到这里该讲的原理和代码都讲完了。最后说点个人实际操作的感受。第一流程引擎本身不需要做得特别复杂把节点、边、Context、状态机、流式输出这五个核心要素做好大部分Agent场景就都够用了。过度设计才是最大的坑比如一开始就给每个节点设计复杂的审批流程、分布式事务只会让项目卡死在初期。第二状态机流转表真的比if-else稳定太多。它不是“换一种写法”而是“换一种建模方式”。一旦你开始用表驱动思维看问题后面写订单状态、审批流、任务调度都会顺手很多。第三如果你也是用Java写Agent建议把Node接口、StreamSink和NodeStateMachine这三部分作为最小核心先跑通然后再慢慢加边、加超时、加持久化。我写第一版只用了不到两天但后面拆坑、调整并发模型用了一周。多跑几个真实流程再回来看设计你会比看十篇文章都理解得深。最后再分享一个小技巧不管你的流程引擎用不用这套设计一定要在启动时给状态表做一次全量自校验。这是我踩过“流程卡死无响应”的坑之后加上的救命线强烈建议你也加一条。 Java实现Agent流程引擎：状态机与Flux流式输出实战
不知道你有没有在这种场景里崩溃过一个Agent流程先调用大模型理解用户意图再根据意图决定要不要调某个工具工具返回之后再让大模型做总结最后把结果推给浏览器。需求方说“很简单的逻辑嘛”你点点头打开代码一开始还挺清爽后边就变成了十几个if-else嵌套再后来你开始在里面塞状态变量、补异常分支、做重试代码越来越长新同事看一眼就想跑路。我最近在做AI Agent的时候也被这种写法折磨过一轮。后来实在忍不了用Java从0到1自己撸了一套流程引擎把Agent工作流里那些“先做A、再看结果做B、C失败就重试、D拿到全部数据后边生成边推给前端”的场景全部收进节点状态轮转和流式输出里。做完之后再看原来的if-else代码我只想说一句真的太Low了。这篇文章就把整个思路和实现细节完整拆开。适合两种人看一种是被Agent流程控制搞得头皮发麻的Java后端另一种是想自己设计流程引擎、又不知道从哪里下手的同学。我会把抽象模型、节点状态机、流转表、Flux流式输出的链路、主线程调度和真实踩坑全部写出来代码可以直接抄。1. 为什么把Agent流程塞进状态机if-else写法到底哪里不对劲1.1 一个五步Agent流程的if-else灾难先还原一下典型的Agent业务代码。最开始可能是这样// 第一步调用大模型理解意图 String intent llm.chat(userInput); if (search.equals(intent)) { // 第二步调用搜索工具 ListString results searchTool.query(userInput); if (results.isEmpty()) { // 第三步没有结果怎么办直接返回兜底文案 return 没查到; } // 第四步用大模型总结搜索结果 String answer llm.summarize(results); // 第五步返回给前端 return answer; } else if (translate.equals(intent)) { ... } else { ... }这段代码在流程只有三四步的时候还能忍。等你要加“重试策略”“超时兜底”“中间结果上报”“失败后人工介入”“某个节点要并发执行”这些需求的时候if-else就会开始失控。我见过最夸张的版本一个方法里写了二十多个状态变量每个if分支里还套着三个try-catch根本没法改。这里面最核心的问题不是代码多而是流程的状态被散落在局部变量和if分支里。你没法在任何一个时刻回答“当前这个流程跑到哪一步了”“下一步有哪些分支可选”“失败之后该退回到哪”因为没有一张全局的状态表只有一团纠缠的逻辑。1.2 Agent工作流的三个刚性需求Agent和普通接口最大的区别在于它的运行时间远超普通HTTP请求。用户提一个问题大模型生成可能要几十秒中间还要调工具、看结果、再生成。这个时候会冒出三个绕不开的需求异步长任务不可能让HTTP请求一直挂着等所有节点跑完需要让流程“跑起来之后可查询、可等待、可推送”。动态分支大模型的输出是概率性的同一个问题可能今天走搜索分支明天走总结分支。分支决策必须由运行时的节点结果决定而不是写死在代码里。流式输出与可观测用户等一个完整的LLM回答等不了几十秒必须让生成的Token像打字机一样持续流到前端。同时平台运营要能看到每个节点现在是什么状态、卡在哪一步。这三个需求用if-else基本不可能优雅实现。你需要一个真正意义上的流程引擎。1.3 流程引擎本质状态机加事件驱动用一句话概括就是把流程拆成节点节点之间有明确的连边关系每个节点都有自己的生命周期状态引擎通过事件驱动节点状态轮转直到整个流程终结。状态机保证“什么状态能转成什么状态”是有约束的不会出现从PENDING直接跳到COMPLETED这种非法情况。事件驱动保证“某个节点完成”这个消息能被引擎捕获自动触发下一个节点的调度。这就是Agent工作流引擎的地基。2. 引擎的第一块地基节点模型、上下文与状态机的抽象设计2.1 节点(Node)一个“动作加决策”的抽象第一步把流程里每一步都抽象成节点。我不喜欢把节点设计成纯粹的“动作”因为Agent流程里很多节点的输出会影响下一步走哪条分支。所以我的节点定义是这样的public interface Node { String getId(); NodeType getType(); void execute(NodeContext ctx, StreamSink sink); }NodeType先分成三类ACTION普通动作节点比如“调用LLM”“查数据库”“调搜索工具”。它执行完就结束输出会写进Context。CONDITION条件分支节点。它不做什么重活只是读取Context里的数据根据规则决定流程接下来走哪条边。END流程终点。一个节点对应一个类每个Agent流程就是一组Node实例和边的组合。以后要调整流程不需要改一堆if-else只要重新组装节点和边就行。2.2 边(Edge)连接节点让流程走起来节点只负责“干活”怎么串联是Edge的事。我用的边设计比较简单每条边有来源节点、目标节点、优先级和一个条件表达式public class Edge { private final String fromNodeId; private final String toNodeId; private final BiPredicateNodeContext, NodeResult condition; private final int order; }引擎在某个节点完成之后会读取这个节点所有出边按order排序后依次用condition判断命中第一条条件满足的边就把流程推到下一个节点。这样条件判断本身就是可配置的新增一条分支完全不需要动引擎代码。2.3 上下文(Context)一块贯穿全流程的“黑板”Agent流程里所有节点之间要共享数据。大模型第一步产出的意图第二步要根据意图查工具第三步要根据工具结果做总结。这些数据不能靠方法参数一层层传所以需要一个贯穿整个流程的Context。我的Context本质是一个线程安全的Map但做了一些封装public class NodeContext { private final ConcurrentHashMapString, Object data new ConcurrentHashMap(); private final String flowId; private final MapString, Object metadata; public T T get(String key) { ... } public void set(String key, Object val) { ... } public T T getOrThrow(String key, ClassT type) { ... } }之所以叫“黑板模式”是因为每个节点都可以在上面写信息、读信息不需要知道信息是谁写的。这让节点之间的耦合度降到最低。注意所有读写要走ConcurrentHashMap因为后面并发调用工具时多个线程可能同时读写Context。2.4 节点状态和流程状态两个维度不能混这里特别容易搞混。我一开始也把节点状态和流程状态混在一个枚举里结果引擎根本没法判断“是某个节点失败还是整个流程失败”。实际设计必须拆开节点状态PENDING、READY、RUNNING、SUCCESS、FAILED、SKIPPED、TIMEOUT流程状态ACTIVE、SUSPENDED、COMPLETED、FAILED节点状态是流程状态的基础。一个节点失败流程到底算FAILED还是走重试分支由引擎策略决定但节点本身的状态必须是FAILED不能直接跳到COMPLETED。2.5 为什么显式建模比硬编码灵活显式建模最大的好处是流程本身变为了数据。节点、边、状态都是可以实例化、可持久化、可视化的对象。你可以把当前流程定义序列化成JSON存库可以在前端画一个流程图展示当前卡在哪个节点可以让运营在不改代码的前提下调整分支条件还可以在流程失败后从某个节点恢复重跑。这三样if-else任何一样都做不到。3. 节点状态轮转的真正实现枚举状态机加流转表不是一堆switch3.1 用枚举而不是一个int字段很多老代码喜欢用int常量表示状态0是初始1是运行中2是成功。这种写法最大的问题是魔法数字满天飞而且编译器完全不会帮你检查状态是否合法。Java里最自然的选择就是枚举。public enum NodeStatus { PENDING, // 刚创建还没进调度队列 READY, // 前置条件满足可以执行 RUNNING, // 正在执行中 SUCCESS, // 执行成功 FAILED, // 执行失败 SKIPPED, // 被跳过比如条件分支走了另一边 TIMEOUT // 执行超时 }枚举相比int的好处不只是可读性它还能天然承担状态机的核心逻辑。后面我会介绍如何用EnumMap做流转表。3.2 流转表所有合法流转全部收进一张表这是整个引擎最核心的思想把“当前状态能转到哪些状态”显式声明成一张表。代码里不再出现任何“if (status RUNNING next SUCCESS)”这种判断而是查表。import java.util.EnumMap; import java.util.EnumSet; import java.util.Map; import java.util.Set; public class NodeStateMachine { private static final MapNodeStatus, SetNodeStatus TRANSITIONS new EnumMap(NodeStatus.class); static { // PENDING只能变成READY或SKIPPED TRANSITIONS.put(NodeStatus.PENDING, EnumSet.of(NodeStatus.READY, NodeStatus.SKIPPED)); // READY可以开始执行也可以被跳过 TRANSITIONS.put(NodeStatus.READY, EnumSet.of(NodeStatus.RUNNING, NodeStatus.SKIPPED)); // RUNNING只允许三种终局成功、失败、超时 TRANSITIONS.put(NodeStatus.RUNNING, EnumSet.of(NodeStatus.SUCCESS, NodeStatus.FAILED, NodeStatus.TIMEOUT)); // 终态不再允许流转 TRANSITIONS.put(NodeStatus.SUCCESS, EnumSet.noneOf(NodeStatus.class)); TRANSITIONS.put(NodeStatus.FAILED, EnumSet.noneOf(NodeStatus.class)); TRANSITIONS.put(NodeStatus.SKIPPED, EnumSet.noneOf(NodeStatus.class)); TRANSITIONS.put(NodeStatus.TIMEOUT, EnumSet.noneOf(NodeStatus.class)); } public static boolean canTransition(NodeStatus current, NodeStatus next) { SetNodeStatus allowed TRANSITIONS.get(current); if (allowed null) { return false; } return allowed.contains(next); } public static void checkTransition(NodeStatus current, NodeStatus next) { if (!canTransition(current, next)) { throw new IllegalStateException( 非法状态流转: current - next); } } }这种“表驱动”的思路本质上是把状态流转规则从代码里抽离出来变成一张可读性极强的配置表。它的好处有三个非法流转在运行前就能暴露。假设有人把节点状态从PENDING直接改成SUCCESS引擎会立刻抛异常而不是带着坏状态继续跑。流程规则变更时只需要改表里的枚举组合。比如需求方说“超时之后允许重试”你就在TIMEOUT对应的集合里加上READY。可以写单元测试对整张表做全量验证。后面我会专门讲这个测试技巧。3.3 状态变更钩子状态一动回调就触发只有流转表还不够。引擎需要在状态变化的瞬间做一些副操作比如更新数据库、记录日志、给前端推送状态。这就需要在状态机里加钩子。public class NodeStateMachine { private static final MapNodeStatus, ListConsumerNodeContext AFTER_CHANGE_HOOKS new EnumMap(NodeStatus.class); public static void registerAfterHook(NodeStatus status, ConsumerNodeContext hook) { AFTER_CHANGE_HOOKS.computeIfAbsent(status, k - new ArrayList()).add(hook); } public static void fireAfterHook(NodeStatus newStatus, NodeContext ctx) { ListConsumerNodeContext hooks AFTER_CHANGE_HOOKS.get(newStatus); if (hooks ! null) { for (ConsumerNodeContext hook : hooks) { hook.accept(ctx); } } } }实际使用中我会把状态变更封装成一个方法调用方只传“目标状态”方法内部先查表校验再改节点对象的状态最后触发钩子。钩子里最常见的就是写日志和发WebSocket事件。3.4 节点状态机和流程状态机联动现在进入关键环节节点状态怎么带动流程状态。引擎主循环的逻辑很简单当一个节点进入SUCCESS或SKIPPED状态时引擎就去节点对应的出边集合里找下一个节点如果下一个节点存在把它从PENDING变成READY再投递到执行队列如果已经没有出边了说明流程到头把流程状态从ACTIVE变成COMPLETED。这里有一个容易踩的坑节点状态变成SUCCESS之后并不代表整个流程就该继续。比如条件分支节点变成SUCCESS只表示“条件判断完成”真正的下一步在它选中的那条边上。所以节点状态机的驱动逻辑和流程状态机的驱动逻辑必须分开写先由节点状态触发边选择再由边选择触发下一个节点的调度。3.5 启动时全表自校验让非法状态流转无处可藏这里分享我自己的一个习惯也是这篇文章里最想让你记住的一招。在Spring Boot项目启动时或者引擎初始化时写一个方法遍历所有节点状态对流转表做全量自校验确保每个状态都能通过最终走向终态没有死循环没有孤立状态。public class StateMachineValidator { public void validate() { for (NodeStatus status : NodeStatus.values()) { SetNodeStatus visited new HashSet(); assertReachableTerminal(status, visited); } } private void assertReachableTerminal(NodeStatus current, SetNodeStatus visited) { if (!visited.add(current)) { throw new IllegalStateException(出现了环形流转当前状态: current); } SetNodeStatus allowed TRANSITIONS.get(current); if (allowed null || allowed.isEmpty()) { return; // 终态 } for (NodeStatus next : allowed) { assertReachableTerminal(next, visited); } } }这个方法看起来简单但真的能救命。我曾经漏配了READY到SKIPPED的流转结果条件分支不满足时流程直接卡死用户界面永远停在“分析中……”。有了这个自校验这类问题在启动阶段就会被发现而不是等到线上用户来骂。4. 流式输出的那点事事件流模型与Flux实现细节4.1 为什么要流式输出如果做完所有节点把最终答案一次性返回Agent流程的用户体验会非常糟糕。大模型生成一个500字的回答可能耗时20秒用户在浏览器里盯着空白页20秒不知道是挂了还是在思考。流式输出就是解决这个问题LLM每产出一个Token就立刻推到前端用户看着文字一个字一个字蹦出来体感上会快非常多。在Agent流程引擎里流式输出比单次LLM调用更复杂因为中间可能穿插多个节点。比如先输出“正在搜索资料……”这个事件再输出“搜索到3条结果”再开始流式输出总结文本。这些事件需要一个统一的事件流模型来承载。4.2 选型SSE、WebSocket还是WebFlux我把选择分成两层内部引擎层用Reactor Flux处理节点产生的事件流因为Flow从一个节点到另一个节点的传递用Flux极其自然。对外HTTP层用SSEServer-Sent Events把Flux映射给前端浏览器。SSE比WebSocket更适合这个场景原因是Agent流的消息方向是单向的服务端往客户端推客户端几乎不需要往回推送控制指令。SSE天然支持重连前端用EventSource就能监听不需要引入WebSocket的状态管理。如果后续确实需要双向交互再在网关层把SSE升级为WebSocket也不难。4.3 事件流模型从Sink到Flux我在引擎里定义了一个极简的流式接口叫Sink专门用来承载节点产生的事件。public interface StreamSink { void onNext(Chunk chunk); void onComplete(); void onError(Throwable t); }Chunk是流事件的最小单位本质是一个携带类型和数据的对象public class Chunk { private final ChunkType type; // TEXT, STATUS, NODE_BEGIN, NODE_END private final String nodeId; private final String content; // 还有时间戳、token总数等字段这里省略 }每个节点在执行过程中都可以向StreamSink发射多个Chunk。比如LLM调用节点每收到一个Token片段就调用一次sink.onNext(new Chunk(TEXT, nodeId, token))。普通动作节点可以只发射一个STATUS类型的Chunk告诉前端“我现在开始执行了”。引擎内部用Reactor的Sinks类把多个节点的Sink统一成一个Flux这样入口方法的返回值就是Flux 既支持背压又方便后续做SSE映射。4.4 用Sinks.Many实现流式管道直接上代码。引擎的入口方法大概是这个形态public FluxChunk startFlow(FlowDefinition flowDef, MapString, Object initData) { Sinks.ManyChunk sink Sinks.many().unicast().onBackpressureBuffer(); FluxChunk flux sink.asFlux(); // 异步启动整个流程把事件写入sink scheduler.execute(() - runFlow(flowDef, initData, sink)); return flux; }runFlow在执行过程中把收到的Chunk挨个写入sink直到流程结束后调用sink.tryEmitComplete()。如果某个节点抛出异常调用sink.tryEmitError(ex)。注意我用的Sinks.many().unicast()表示只能订阅一次这正好适合一个HTTP请求对应一个SSE流。如果同一个流程想推送给多个订阅者那就得换成replay()或multicast()不过实际项目中一个请求对应一个订阅者的场景占绝大多数。4.5 Token如何从LLM节点流到浏览器把整条链路串起来看一个LLM调用节点从发出请求到前端展示大概经过这么几步用户请求进入引擎引擎为这个流程创建一个唯一的flowId初始化NodeContext。第一个节点是ACTION节点内部调用LLM SDK。现在的LLM SDK基本都支持流式接口返回的是Flux 。节点拿到Flux 后每来一个Delta就通过StreamSink发射一个Text类型的Chunk。引擎把这些Chunk实时写入内部Sinks从而驱动外层Flux 。Controller层把Flux 映射成Server-Sent Events直接response.flush()。浏览器EventSource收到消息把文字追加到页面上。这一步的关键在于你在节点代码里不能把LLM SDK的流“攒起来”等结束再发而必须把LLM的Flux直接映射到StreamSink保持整条流是实时贯通的。5. 主循环、并发与超时执行引擎的最后一公里5.1 主循环取事件、查流转表、调度节点有了前面两章的设计引擎的主循环反而不复杂。它本质上是一个不断处理事件的循环private void runFlow(FlowDefinition flowDef, NodeContext ctx, StreamSink sink) { // 找到起始节点 Node currentNode flowDef.getStartNode(); transitionNode(currentNode, NodeStatus.READY); while (!currentNode.getType().isEnd()) { transitionNode(currentNode, NodeStatus.RUNNING); NodeResult result doExecute(currentNode, ctx, sink); Edge nextEdge flowDef.findNextEdge(currentNode, ctx, result); if (nextEdge null) { sink.onError(new IllegalStateException(没有可用流转边: currentNode.getId())); return; } Node nextNode flowDef.getNode(nextEdge.getToNodeId()); if (nextNode null) { sink.onError(new IllegalStateException(目标节点不存在: nextEdge.getToNodeId())); return; } // 当前节点进入SUCCESS下一个节点进入READY transitionNode(currentNode, NodeStatus.SUCCESS); transitionNode(nextNode, NodeStatus.READY); currentNode nextNode; } transitionNode(currentNode, NodeStatus.SUCCESS); sink.onComplete(); }注意这里的doExecute不是直接同步调用而是把任务提交给线程池执行并携带超时控制。后面马上讲。5.2 并发模型单线程事件循环加线程池执行节点我建议的并发模型是这样的不共享状态变量的多线程粗放并发而是给整个流程配一个调度线程池节点执行走工作线程池。具体来说每个流程实例由引擎的主循环调度线程负责推进状态、查边、决定下一个节点。节点本身的耗时代码比如调LLM、调工具提交到独立的ExecutorService执行。主循环用Future.get(timeout)等待节点结果确保不会因为某个节点卡死而导致整个流程无法推进。不同流程实例之间天然并行因为每个实例都自己持有Context互不干扰。这个模型看起来简单但非常实用。它既避免了全局Context被多线程竞争又能利用工作线程池处理耗时任务。5.3 超时处理Future加线程池是标配超时是Agent流程里一定要处理的。LLM接口偶发变慢是常态某个搜索工具挂掉更是不可避免。我用ScheduledExecutorService加Future实现超时控制private NodeResult executeWithTimeout(Node node, NodeContext ctx, ExecutorService workerPool, long timeout, TimeUnit unit) throws Exception { FutureNodeResult future workerPool.submit( () - node.execute(ctx, sink) ); try { return future.get(timeout, unit); } catch (TimeoutException e) { future.cancel(true); transitionNode(node, NodeStatus.TIMEOUT); throw new NodeTimeoutException(node.getId(), timeout); } }超时之后节点状态变成TIMEOUT流转表里允许从TIMEOUT转回READY这就给重试留了口子。5.4 重试机制不能无脑重试必须考虑幂等有了超时和TIMEOUT状态重试就顺理成章了。我通常会在Edge或Node上配置重试策略三要素是最大重试次数、重试间隔、幂等性要求。重试间隔可以用简单的指数退避。幂等性这块要特别小心后面踩坑部分我会讲一个真实案例这里先提醒一句调LLM接口重试是要花钱的必须要用requestId做幂等键或者明确业务上允许“三次请求都可能产生Token”。5.5 持久化与崩溃恢复如果Agent流程运行时间很长比如一个自动写研报的Agent可能要跑几分钟甚至更久中途服务重启就会把流程丢光。所以引擎必须支持状态持久化。实际方案不复杂节点每次状态变更时把最新的NodeStatus和Context快照写入数据库。引擎启动时提供一个“未完成任务恢复”方法扫描状态不是终态的流程实例从最后一个成功的节点继续执行。Context里必须放进一个可序列化版本这样重启后还能继续读取之前节点产出的数据。我是用一张简单的flow_instance表加一张node_state表实现的字段包括flow_id、node_id、status、context_json、create_time、update_time。handling重启恢复时把状态机流转表的校验逻辑再做一遍确保不会从非法状态继续跑。6. 真实Agent场景接入与踩坑记录LLM调用、工具节点、分支判断6.1 一个自动写研报的Agent实战案例纸上谈兵没意思直接看一个能跑通的真实链路。假设我们要做一个“行业研报助手”用户输入一个行业名称引擎自动执行节点A调用LLM对大模型说“请提取该行业的核心关键词并输出JSON”。节点B条件分支解析节点A输出的JSON如果关键词为空直接走EndNode结束否则走下一步。节点C调用数据查询工具根据关键词查询最近一周的研报数据。节点D调用LLM把查询结果整理成流式研报。关键代码长这样public class ExtractKeywordsNode implements Node { Override public void execute(NodeContext ctx, StreamSink sink) { String industry ctx.getOrThrow(industryName, String.class); sink.onNext(new Chunk(ChunkType.STATUS, getId(), 正在提取行业关键词...)); FluxDelta deltaFlux llmClient.streamCompletion( 提取 industry 的核心关键词输出JSON数组 ); StringBuilder sb new StringBuilder(); deltaFlux.doOnNext(delta - { sb.append(delta.getContent()); // 关键词提取阶段不需要实时流给前端只攒结果 }).blockLast(); ctx.set(keywords, parseKeywords(sb.toString())); sink.onNext(new Chunk(ChunkType.STATUS, getId(), 关键词提取完成)); } }节点B是条件分支节点它读取Context里的keywords字段判断是否为空然后通过condition返回true或false引擎据此找到对应的边。6.2 工具调用节点让Agent具备行动能力工具调用节点在实际项目里最复杂。搜索、查库、调外部API每个工具的参数格式都不一样响应时间也不一样。我的做法是给所有工具做一个统一包装public interface ToolExecutor { String getToolName(); ToolResult execute(MapString, Object args, NodeContext ctx, StreamSink sink); }工具节点执行时先发一个“开始调用工具”的Chunk然后把参数传给ToolExecutor拿到结果之后写回Context。如果工具本身支持流式比如命令行的实时日志输出同样可以映射到StreamSink。这里要提一点工具调用必须考虑鉴权和参数校验不能把用户输入原样拼进SQL或Shell命令里。Agent流程引擎本身不替你做安全校验但如果你的Agent会调用外部工具这是必须自己补上的防线。6.3 分支节点让流程真正“智能”Agent最迷人的地方是流程可以自己决定下一步。LLM节点输出一段结构化JSON条件分支节点解析JSON后决定走哪条边。我用了一个很轻量的实现方式条件分支节点的condition直接写成一个Java lambda从Context读取LLM产出这样不需要解析器灵活度反而最高。Edge searchEdge Edge.builder() .fromNodeId(extractKeywords) .toNodeId(searchReports) .condition((ctx, result) - { ListString keywords ctx.get(keywords); return keywords ! null !keywords.isEmpty(); }) .order(1) .build(); Edge endEdge Edge.builder() .fromNodeId(extractKeywords) .toNodeId(end) .condition((ctx, result) - ctx.get(keywords) null) .order(2) .build();这种写法的好处是分支条件和业务流程本身彻底解耦要调整分支逻辑只需要改Edge的condition节点代码完全不用动。6.4 踩坑记录这些坑我替你踩过了把我在开发过程中遇到的三个大坑和解决思路整理成表格比直接背结论更有用。坑现象根因解决流式输出背压没有控制用户量大时后端内存暴涨最终OOMSinks创建时用了onBackpressureBuffer但没有限制缓冲区大小LLM产生的Token太多时全部积压在内存里改用onBackpressureBuffer(1000)并给前端SSE配置合理的超时时间量再大就上Kafka或Redis Stream做削峰状态轮转过早节点执行完成但流式事件还没有发射完前端数据缺失节点代码里先调了sink.onComplete()然后才调用工具持久化结果严格约定onComplete必须是节点执行的所有业务逻辑结束之后才能调用放在execute方法最后一行超时重试导致LLM重复扣费一个请求因超时重试了三次LLM产生了三份费用重试前没有幂等控制每次重试都发全新请求在NodeContext里记录外层requestIdLLM SDK加上请求级别的幂等键重试次数最多一次且必须确认上一次请求确实没有进入生成阶段这些坑在文档里基本不会写明属于实际跑业务才会遇到的教训。特别是LLM重复扣费那个建议所有做Agent的人提前把幂等键设计好不然上生产环境第二天账单会很难看。6.5 测试与可视化状态机表驱动带来的附加红利最后聊几句测试。因为用了流转表状态机的测试覆盖率可以做到几乎百分百。我写了一个测试类遍历所有NodeStatus把每个合法流转都执行一遍断言不会抛异常再把所有非法流转也遍历一遍断言必定抛异常。这两组测试跑完状态机的正确性就有了基本保障。流程可视化也因为这个设计变得顺理成章。把完整流程图定义序列化成JSON用前端组件渲染出来再通过WebSocket推送每个节点的实时状态前端就能画出一张“动态走光的流程图”。这个功能在给业务方演示的时候效果极好也方便排查线上问题——一眼就能看到流程卡在哪个节点。尾声我把这套引擎拆完之后的几点体会整篇文章到这里该讲的原理和代码都讲完了。最后说点个人实际操作的感受。第一流程引擎本身不需要做得特别复杂把节点、边、Context、状态机、流式输出这五个核心要素做好大部分Agent场景就都够用了。过度设计才是最大的坑比如一开始就给每个节点设计复杂的审批流程、分布式事务只会让项目卡死在初期。第二状态机流转表真的比if-else稳定太多。它不是“换一种写法”而是“换一种建模方式”。一旦你开始用表驱动思维看问题后面写订单状态、审批流、任务调度都会顺手很多。第三如果你也是用Java写Agent建议把Node接口、StreamSink和NodeStateMachine这三部分作为最小核心先跑通然后再慢慢加边、加超时、加持久化。我写第一版只用了不到两天但后面拆坑、调整并发模型用了一周。多跑几个真实流程再回来看设计你会比看十篇文章都理解得深。最后再分享一个小技巧不管你的流程引擎用不用这套设计一定要在启动时给状态表做一次全量自校验。这是我踩过“流程卡死无响应”的坑之后加上的救命线强烈建议你也加一条。