把Agent的一连串步骤用if-else串起来是我见过最劝退的写法。你可能觉得流程不复杂无非就是“先调用大模型、判断结果、再决定下一步”但一旦节点超过五六个你就要维护一张隐形的状态网下一步该走哪、失败怎么回退、上下文传什么、第几个token该推给前端。我上个月做一个简历筛选Agent项目一开始图省事就用if-else硬串结果差点把自己绕晕。后来干脆用Java从0到1写了套轻量级Agent工作流流程引擎把每个节点当状态机来轮转再把AI的输出走流式推给页面。跑通之后回过头看之前的写法确实太Low了。这篇就把我实现这套引擎的思路、代码主干、踩过的坑一次讲完。适合Java后端、想搞Agent工作流编排、又不想引入Camunda这种重框架的同学参考。核心围绕几个词展开节点状态轮转、流程引擎、流式输出也顺便聊聊为什么自研这么轻的东西才是最优解。1. 为什么流程引擎是Agent编排的正解而不是if-else堆叠1.1 用if-else编排Agent到底有多痛苦说白了Agent工作流就是把多个“智能步骤”串在一起。比如一个简历筛选Agent可能要经历“读取简历 - 大模型抽取结构化信息 - 关键词初筛 - 大模型综合评估 - 输出结果”这五步。用if-else写的初版长这样public String run(Resume resume) { String extracted llm.extract(resume.text()); if (extracted.contains(Java) extracted.contains(3年)) { boolean keywordPass matchKeyword(extracted); if (keywordPass) { String evaluated llm.evaluate(extracted); if (evaluated.contains(通过)) { return recommend: evaluated; } else { return reject: evaluated; } } } return reject: need manual review; }看起来还能看但真实项目根本不是这样。每个节点会有重试、超时、失败处理有些节点要并行跑比如同时查两张大模型有些节点要跳过用户还会在中途查看当前进度。等这些需求叠上来if-else就开始失控了。最要命的是判断条件散落在各个分支里流程状态根本无迹可寻线上出了问题只能靠日志猜。1.2 流程引擎带来的三个核心改变我实现完引擎之后回头总结发现它真正解决的问题是这三个第一声明式流程定义。每个节点只需要声明自己是谁、干什么、下一跳是谁流程的结构一目了然而不是把控制流散落在代码里。第二状态可追踪。每个节点都有固定的状态机PENDING、RUNNING、SUCCESS、FAILED、SKIPPED。不管流程走到哪里只需要看节点状态就知道卡在哪、要不要重试、能不能恢复。第三天然适合流式输出。因为引擎节点是逐个触发的每个节点的产出都可以通过事件总线推给前端用户能看到Agent“思考到哪一步”了而不是傻等一个最终响应。1.3 自研轻量引擎还是引入Camunda/Flowable这个取舍我纠结过。Camunda、Flowable这类BPMN引擎确实成熟有可视化建模、有持久化、有审批人机制。但用下来有两个问题它们面向的是“人审批流程”节点类型偏人工任务Agent节点这种“要动态调LLM、要出流式token”的节点很难直接映射过去。BPMN的XML定义和部署模型太重量级了写Agent项目时我更希望用Java代码定义流程然后立即能跑。因此我选择了自研主题很明确轻量、状态机驱动、事件总线抛流式结果。整体不到两千行代码维护成本完全可控。2. 节点模型设计工作流的基础抽象长什么样2.1 抽象Node接口入参、执行逻辑、出参一切流程引擎本质上都是“定义节点、连接节点、驱动节点”。所以第一步就是把节点抽象出来。我设计的Node接口如下public interface Node { String getId(); void execute(NodeContext context); }NodeContext是贯穿整个流程的上下文对象存放原始输入、中间产物、节点状态、输出事件。在引擎内部每个节点执行的时候都会更新自己的状态然后告知调度器去拉取next节点。2.2 AgentNode把LLM调用封装成一个普通节点Agent工作流最核心的节点是“AI节点”。它不关心大模型是OpenAI格式、阿里云还是自部署的只关心你给它一个FunctionString, FluxString输入是prompt或上游结果输出是一段文本但这个文本是流式的。public class AgentNode extends BaseNode { private final FunctionString, FluxString llmExecutor; Override protected void doExecute(NodeContext context) { String prompt context.getInput(this.getId()); FluxString stream llmExecutor.apply(prompt); String fullText stream.collectList().block().stream().collect(Collectors.joining()); context.setOutput(this.getId(), fullText); } }注意这里的FluxString我用的响应式流来表示LLM的增量token。哪怕最终同步汇总成完整文本这个设计也让流式输出有了入口。2.3 条件节点与汇聚节点分支逻辑不再靠if-else流程里不止有串行节点还有条件判断。传统做法是if-else这里我把它收敛成ConditionNode和EndNode两种特殊节点。ConditionNode不生产业务输出它只干一件事根据上游结果从MapString, String里选出下一个节点的id。如果只有两个分支就配置default和pass两个出口。EndNode则是流程的终止点引擎看到它就知道整个工作流跑完了。2.4 为什么用组合而不是多层继承来扩展节点很多初学Java的人看到“节点”第一反应是建一个庞大的继承树AgentNode继承BaseNodeConditionNode继承BaseNode。但实际扩展多了就会发现继承树很容易僵化。比如突然要一个“带重试的Agent节点”是再建一层RetryAgentNode还是给BaseNode加重试参数我选的是组合路线。BaseNode只负责状态机和链式调度具体业务能力通过“策略”注入。比如重试功能用一个RetryDecorator包装节点并行功能用ParallelNode组合多个子节点。这样每个节点职责单一扩展点越多、越灵活反而越好维护。3. 节点状态轮转把流程推进从if-else里解放出来3.1 为什么状态机比布尔flag靠谱工作流最怕的状态问题是“这个节点到底跑没跑过”。如果用布尔flag你需要记running、done、failed三个字段再来个重试字段数量直接爆炸。状态机的好处是节点在所有时刻只有一个状态且状态的流转是受控的、可预测的。我定义了这么一组枚举public enum NodeState { PENDING, RUNNING, SUCCESS, FAILED, SKIPPED }每个节点初始是PENDING被调度器选中后变成RUNNING正常执行完是SUCCESS抛异常是FAILED。条件不满足被跳过时是SKIPPED。3.2 状态轮转的触发机制事件驱动而不是手动调用既然叫“轮转”就不能是在节点内部手动改状态。我引入了一个简单的事件机制。每个节点执行完之后触发一个事件事件携带nodeId和执行结果事件总线再根据结果决定下一个状态和下一批节点public class NodeEvent { private final String nodeId; private final NodeState targetState; private final String nextNodeId; }节点不直接调用其他节点它只负责发出事件。调度器订阅这个事件找到目标节点驱动它进入RUNNING状态。这样做的核心价值是解耦如果你想加“失败重试”或者“异步补偿”只需要新增一个事件处理器不需要改动节点的执行逻辑。3.3 状态轮转表用Map定义节点间的关系把流程看成有向图每条边都是一次状态轮转的触发条件。我用一个MapString, ListString来存储每个节点可能的next节点// key: 当前节点id, value: 可到达的下一批节点id MapString, ListString transitionTable new HashMap();当ConditionNode执行结束时它往事件里写入nextNodeId passNodeId调度器去查transitionTable确认这个跳转合法然后把passNode从PENDING轮转为RUNNING。这样即使在运行时报错“非法节点跳转”也能第一时间发现而不是一脸懵。3.4 Agent场景下特殊处理失败重试与挂起恢复LLM调用有个特点不稳定。网络超时、限流、返回JSON格式错误都是家常便饭。所以Agent节点的状态轮转必须支持FAILED - PENDING - RUNNING这样一个重试闭环。我是这么做的在事件处理器里如果收到FAILED事件先检查该节点剩余重试次数大于零就重置为PENDING并挂起一小段时间再重新RUNNING如果重试耗尽流程整体进入挂起状态等待人工介入。这个设计在面试场景讲出来也非常加分因为它体现的不只是CRUD而是对真实分布式系统容错的理解。4. 流式输出让Agent的每一段思考都被看到4.1 为什么Agent工作流需要流式输出用户等待一个Agent工作流完成如果前端一直转圈很容易以为挂了。流式输出不只是体验问题更是一个工程问题流程节点每改一次状态、大模型每吐出一个token前端都要即时感知。流的颗粒度可以分两层节点级别的流当前跑到哪个节点了比如“正在读取简历 - 正在抽取字段 - 正在评估”。token级别的流每个Agent节点内部的LLM响应像ChatGPT一样一个字一个字蹦出来。4.2 SseEmitter接入比WebSocket轻得多Java后端做流式输出最省事的方案是Spring MVC自带的SseEmitter。它基于HTTP长连接前端用EventSource就能接不需要WebSocket那套握手协议。我把它设计成了全局的SseEmitterRegistryComponent public class SseEmitterRegistry { private final ConcurrentHashMapString, SseEmitter emitters new ConcurrentHashMap(); public void add(String clientId, SseEmitter emitter) { ... } public void send(String clientId, String eventName, Object data) { ... } }引擎里每次节点状态变化都往这个Registry推一条事件。前端页面就能实时看到“节点3执行失败正在重试”这样的进度。4.3 把LLM的StreamResponse接进节点事件流光有节点级状态流还不够LLM的token流也不能断。我是这样衔接的AgentNode执行时先把大模型的FluxString订阅起来。每个token到达后不直接等所有token收集完而是立刻通过注册表把增量推给前端。同时把增量拼接进StringBuilder保证最终上下文完整。llmExecutor.apply(prompt) .doOnNext(token - { emitterRegistry.send(clientId, agent-token, token); fullText.append(token); }) .collectList() .block();有一个细节要注意如果给前端的SSE连接是主线程而token流是响应式线程两者之间跨线程访问SseEmitter有并发问题。我统一通过ExecutorService提交推送操作避免SseEmitter在线程池中被多个线程同时写。4.4 流式输出与状态轮转如何协作这里容易踩坑。很多人在实现流式输出时只顾着把token推给前端却忘了更新节点状态。正确的顺序应该是节点进入RUNNING - 推节点开始事件 - 逐个推token - 节点进入SUCCESS - 推节点结束事件。我把状态轮转和流式输出放在同一个事件链上。简单说就是状态机每次变化都会发出一个NodeEvent这个事件有两个消费者一个是推进流程的调度器一个是推送前端的SSE处理器。逻辑就变成调度器推进流程SSE处理器反馈进度千万不要在节点内部手动推前端不然代码会非常乱。5. 完整落地一个Agent简历筛选工作流的真实跑通过程5.1 需求简历筛选Agent的流程定义我拿最开始的简历筛选项目作为典型案例。流程一共四步读取简历文本。Agent抽取结构化字段姓名、年限、技能列表。条件判断是否匹配Java岗位。Agent写评估结论。用引擎定义出来是这样的Workflow wf Workflow.create(resume-check) .node(new StartNode(start)) .node(new AgentNode(extract, extractPrompt, llmExecutor)) .node(new ConditionNode(match, extract, Map.of(pass, evaluate, default, end))) .node(new AgentNode(evaluate, evaluatePrompt, llmExecutor)) .node(new EndNode(end)) .link(start, extract) .link(extract, match) .link(match-pass, evaluate) .link(match-default, end) .link(evaluate, end);没有一行if-else流程长什么样代码就长什么样这就是声明式的威力。5.2 引擎驱动与上下文传递的实现要点NodeContext在引擎里怎么流转决定了流程是清晰还是混乱。我的做法是上下文只有两份数据入口和一个产物池。public class NodeContext { private final MapString, String inputs; // 每个节点的入参 private final MapString, String outputs; // 每个节点的产物 }ConditionNode拿到outputs里extract节点的结果做关键词判断evaluate节点同样从outputs里取extract的结果拼prompt。节点之间不会直接互相调用所有数据都通过上下文传递这样你随时可以dump出任何一次运行的全量中间状态排查问题时特别有用。5.3 实测踩坑状态不一致、SSE断连、JSON丢失跑通只用了半天踩坑却用了整整两天。这里分享三个我印象最深的坑。第一个坑是状态覆盖丢失。最初我把状态直接存在节点对象里但流程定义是单例的多个请求同时跑一个节点同时被两个线程改状态直接乱套。解决方式是状态不放在节点实例上而是放在NodeContext里每个请求都有自己的状态副本。第二个坑是SSE断连后还在推token。用户刷新页面旧Emitter没关闭引擎还在往里写数据控制台刷一堆IOException。后面加了心跳检测onCompletion和onTimeout回调统一移出注册表才算干净。第三个坑是大模型返回JSON里有换行符直接导致抽取字段解析失败。这个问题和引擎无关但Agent节点的鲁棒性必须考虑。我在AgentNode外面套了个SchemaFixDecorator专门处理JSON中缺失引号、多余逗号之类的常见格式问题。5.4 实测效果和if-else版本对比同一个简历筛选流程我对比了改造成引擎前后的开发体验对比项if-else版本流程引擎版本流程结构散落在各行代码里声明式一目了然新增一个步骤要找到插入点改嵌套加一个节点、加一条link失败重试手写循环事件处理器统一处理前端流式进度没有节点状态和token实时推送定位问题靠日志断点直接看各节点状态和上下文开发效率提升是肉眼可见的。最爽的是产品提新需求时我只需要在流程定义里加减节点不需要动核心引擎。6. 引擎还能往哪里扩展我的下一步计划6.1 并行节点与聚合结果现在的引擎是单线串行但真实Agent场景经常要并行。比如筛选简历时可能同时调用“技能抽取”和“经验年限校验”两个小Agent最后汇总结果。我准备加一个ParallelNode它会在事件总线中同时触发多个子节点并维护一个计数器所有子节点都SUCCESS之后再轮转到下一个节点。这个设计尤其适合“AI怎么扛并发”这类场景的讨论。6.2 流程持久化与可视化回放如果流程做到一半宕机重启后怎么恢复答案是把NodeContext序列化存进数据库。下一步我计划给每个流程实例增加快照能力每轮转一次状态就记录一条流水。这样不仅能恢复还能在页面上回放每个节点的状态变化过程做故障复盘非常方便。6.3 让引擎适配多Agent协作当前节点里的LLM执行器是单体的。后续如果接入多Agent框架比如让“管理员Agent”动态派发任务给“写代码Agent”和“写文档Agent”核心引擎不需要大改只需要扩一个AgentGroupNode它内部可以动态创建子上下文、子流程。状态轮转和流式输出的机制还是那一套这个扩展点是自研引擎最让我舒服的地方。最后再分享一个小技巧这套引擎的核心代码量不大但设计时一定要把“事件驱动”这层想透。很多人自研流程引擎写着写着又回到“节点A直接调用节点B”的老路最后变成披着引擎皮的if-else。只要节点只发事件、调度器只决定轮转方向、SSE处理器只负责推送这三个角色各管一段整个系统的复杂度就能控得住。我现在已经在好几个Agent项目里复用这套引擎了每次改流程都只需要动流程定义那一层真正体会到了什么叫“再也不用if-else写流程了”。