1. 从if-else泥潭到Agent工作流引擎：我为什么重写流程调度
最近在做一个Java Agent项目，接了意图识别、工具调用、RAG检索、人工确认好几套逻辑，一开始图省事，所有流程调度全用if-else硬写。等需求改到第三版的时候，我终于意识到自己给自己埋了一个巨大的雷：方法里叠了六层判断，每一层还要处理"成功、失败、超时、需要人工介入"这些分支，加一个新环节就得在关键路径上找半天该插到哪个else里，改一个顺序牵一发动全身。
这不是代码质量问题，是流程建模的方式从一开始就选错了。Agent的核心本质是"一个能自主决策、分步执行的程序"，它的每一次执行都像一条流水线：先判断用户意图，再调用工具，然后生成回复，可能中间还要停下来问用户一句"确认执行吗"。这种场景天然适合用流程引擎来描述，而不是用if-else来描述。于是我把原来的调度逻辑全部推翻，用一套非常轻量的流程引擎重做了。
这套引擎的核心就两件事：节点状态轮转和流式输出。节点状态轮转让流程调度变成一张可配置的图，节点自己决定"我完成后下一步该交给谁"，引擎只负责按状态推进；流式输出则让Agent执行的每一个中间结果、每一个工具调用日志、每一条token增量，都能实时推给前端。做完之后最直观的感受是：新增一个Agent节点，只需要写一个节点类然后注册进去，调度逻辑完全不用动，代码里再也没出现过一堆if-else嵌套。
这篇文章就是把我从0到1实现这套流程引擎的全过程写出来，包括核心设计、状态机模型、流式输出方案、完整代码实操以及我踩过的坑。如果你也在用Java搞Agent开发，或者想把自己项目里的硬编码流程改成可编排的引擎，这篇可以直接当参考。
2. 流程引擎整体设计与状态轮转机制
2.1 节点模型与状态机的设计
想把流程交给引擎调度，第一步是定义清楚"节点"长什么样。在我的设计里，一个流程是由多个节点组成的有向图，每个节点做一件独立的事，节点执行完会返回一个结果，结果里包含"下一步要去哪个节点"以及"当前节点的状态"。
节点本身是一个抽象概念，实际业务里它可以是"调用大模型生成意图"、"执行某个工具方法"、"发送一条MQ消息"、"等待用户确认"等等。我最开始犯过一个错误，就是让节点直接持有下一个节点的引用，结果整个流程还是硬编码的。后来改成节点通过返回值声明下一步，引擎从注册表里按名字找下一个节点，这才把节点之间的耦合彻底解开。
节点的生命周期状态我用一个枚举来定义：
public enum NodeStatus { PENDING, // 待执行 RUNNING, // 执行中 SUCCESS, // 执行成功 FAILED, // 执行失败（可重试或终止） SKIPPED, // 跳过（条件不成立） WAITING // 等待外部输入（如人工确认） }这个状态机就是整个流程引擎的底层骨架。节点最初都是PENDING，引擎从Start节点开始跑，拿到节点后先把状态置为RUNNING，执行一段业务代码，根据返回结果把状态改成SUCCESS或FAILED，再依据结果里的targetNodeId找到下一个节点继续跑。这就是所谓的"状态轮转"——流程的推进不是靠代码里的条件分支，而是靠节点"当前状态 + 返回的下一跳"自然流转。
2.2 状态轮转如何替代if-else
很多人一开始不理解：你用状态轮转和用if-else，最终不都是要判断"下一步走哪个分支"吗？区别到底在哪？
我用一个例子说明。假设一个Agent任务要经历：意图识别 -> 调用工具 -> 生成回复 -> 人工确认。用传统if-else写，代码长这样：
if (intent == 查询天气) { String weather = callWeatherTool(city); if (weather != null) { String answer = generateAnswer(weather); if (needManualReview(answer)) { waitingForManual(); } else { send(answer); } } else { handleError(); } } else if (intent == 订机票) { // 又是一大坨 }看着好像还行，但实际上每个if里面还要加各种异常分支、重试分支、日志埋点，一旦扩展到十几个节点，这个方法的行数会迅速膨胀到几百上千行。最要命的是，流程的"形状"被写死在代码里，你想在"工具调用"和"生成回复"之间插一个"工具结果格式化"节点，必须改这段嵌套逻辑。
改用流程引擎后，同样的流程只需要定义一张节点关系映射：
Map<String, List<Transition>> graph = new HashMap<>(); graph.put("intent", List.of(new Transition("TOOL", "intent==tool"))); graph.put("TOOL", List.of(new Transition("GENERATE", "success"))); graph.put("GENERATE", List.of(new Transition("REVIEW", "need_review")));引擎拿到当前节点执行完返回的状态和下一跳目标，去这张表里查消息，走到对应节点。节点之间的关系是数据，不是代码。新增一个节点就是往图里加一个条目，任何时候想改变流程走向，只需要改配置或改节点返回的targetNodeId。if-else判断的是"当前这一刻怎么走"，状态轮转描述的是"整张图怎么走"，后者才能驾驭复杂的Agent编排。
2.3 引擎核心调度器实现
调度器是整个流程引擎的心脏，我把它实现成一个非常精简的循环。核心逻辑就三件事：拿到当前节点、执行、根据结果决定下一个节点。看代码：
public class WorkflowEngine { private final Map<String, Node> nodeRegistry = new ConcurrentHashMap<>(); private final WorkflowContext context = new WorkflowContext(); public void registerNode(Node node) { nodeRegistry.put(node.getName(), node); } public void run(String startNodeName) { String currentNodeName = startNodeName; while (currentNodeName != null && !Thread.currentThread().isInterrupted()) { Node currentNode = nodeRegistry.get(currentNodeName); if (currentNode == null) { throw new IllegalStateException("Node not found: " + currentNodeName); } currentNode.updateStatus(NodeStatus.RUNNING); NodeResult result = currentNode.execute(context); currentNode.updateStatus(result.getStatus()); // 根据状态决定是否重试、跳过或结束 if (result.getStatus() == NodeStatus.FAILED && result.isAllowRetry()) { currentNodeName = currentNodeName; // 原地重试 continue; } if (result.getStatus() == NodeStatus.SKIPPED) { currentNodeName = result.getTargetNodeId(); continue; } if (result.getStatus() == NodeStatus.WAITING) { // 挂起，等待外部信号，这里简化处理为结束 break; } currentNodeName = result.getTargetNodeId(); } } }这个循环写起来简单，难点在于考虑完整。比如"人工确认"节点，它执行完不会立即给出结果，而是要挂起流程，等用户在外部点击确认后再触发引擎继续跑。所以我给NodeStatus加了WAITING，遇到WAITING就先把当前节点挂起，引擎主循环退出，后续通过一个resumeWorkflow方法从挂起节点继续推进。
3. 流式输出：让Agent的每个动作都实时可见
3.1 为什么Agent场景必须用流式输出
Agent和普通后端接口有个巨大的差异：执行时间长且过程信息有展示价值。一次完整的Agent调用可能涉及模型推理、多个工具调用、多轮内部思考，如果像普通接口一样等到全部执行完再一次性返回结果，用户在前端就是白屏等待十几秒甚至几分钟，体验极其糟糕。
我之前试过轮询方案，Agent每执行一步就往数据库里写一条状态，前端每隔两秒查一次。能用，但太蠢：第一，轮询有延迟；第二，数据库被高频写入；第三，执行过程的一些半结构化日志（比如大模型流式吐出来的token）很难通过轮询还原成自然的打字机效果。后来我把方案换成了服务器推送事件（SSE），也就是基于HTTP长连接的服务端流式推送。SSE实现简单、兼容性好、Java生态支持完善，特别适合Agent这种服务和前端是单向数据流的场景。
流式输出解决了两个问题：一是让用户看到Agent"正在做什么"，而不是干等；二是让中间产物（工具调用参数、检索到的资料、推理片段）能被前端实时渲染，比如在界面上展示"当前正在调用天气工具，参数：北京"这样的过程卡片。
3.2 基于Spring WebFlux的流式推送实现
我项目里用的Spring Boot 3，直接依赖了WebFlux的响应式栈。这里我不会把整个响应式编程讲一遍，只说怎么在一个普通接口里实现SSE流式输出。
核心思路是用一个广播源作为事件总线。Agent引擎里的任何节点都能往这个广播源里发消息，HTTP响应通过Flux订阅这个广播源，把消息实时转发给前端。
import org.springframework.http.codec.ServerSentEvent; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; import reactor.core.publisher.Flux; import reactor.core.publisher.Sinks; import java.util.concurrent.ConcurrentHashMap; @RestController public class AgentFlowController { // 广播源：支持多订阅者 private final Sinks.Many<ServerSentEvent<String>> sink = Sinks.many().multicast().onBackpressureBuffer(); private final WorkflowEngine engine; public AgentFlowController(WorkflowEngine engine) { this.engine = engine; } @GetMapping("/execute") public Flux<ServerSentEvent<String>> executeAgent() { // 这里把当前会话的sink实例传给引擎，让引擎在节点执行时向这个sink发送事件 engine.setEventSink(sink); // 异步启动引擎 engine.runAsync("START"); return sink.asFlux(); } }在Node执行过程中，只需要调用sink.tryEmitNext()就能推送一条事件。比如在执行工具调用节点时，就会发一条类似"tool_start"的事件：
sink.tryEmitNext(ServerSentEvent.builder("开始调用天气工具，城市=北京") .event("tool_start") .build());这样前端就能实时收到文本片段，配合CSS就能实现打字机效果。注意WebFlux的Flux是无限流，所以需要确保当Agent流程结束后，要主动向sink发一个"flow_end"事件，然后调用sink.tryEmitComplete()关闭流，否则前端会一直等待。
3.3 节点状态与流式事件如何联动
流程引擎有了流式输出能力之后，我做了个很顺手的扩展：把节点状态变化也作为事件推出去。也就是说，前端不仅能看到Agent输出了什么，还能看到一个实时的流程状态面板，类似"意图识别 -> 识别完成 -> 工具调用中 -> 工具调用成功 -> 生成回复中"的可视化进度条。
我定义了一个FlowEvent类型，包含事件类型、节点名、节点状态、数据载荷：
public class FlowEvent { private String type; // NODE_STATUS, TOKEN, TOOL_CALL, ERROR, FLOW_END private String nodeName; private NodeStatus status; private Object data; // 构造器、getter省略 }节点执行前、执行中、执行后分别emit不同的事件。在节点基类里统一做这件事：
public abstract class Node { protected Sinks.Many<ServerSentEvent<String>> sink; public void setEventSink(Sinks.Many<ServerSentEvent<String>> sink) { this.sink = sink; } public final NodeResult execute(WorkflowContext context) { emitStatus(NodeStatus.RUNNING); try { NodeResult result = doExecute(context); emitStatus(result.getStatus()); return result; } catch (Exception e) { emitError(e.getMessage()); return NodeResult.failure("ERROR", e.getMessage()); } } private void emitStatus(NodeStatus status) { if (sink != null) { sink.tryEmitNext(ServerSentEvent.builder("") .event("node_status") .comment(nodeName + ":" + status) .build()); } } protected abstract NodeResult doExecute(WorkflowContext context); }这里有个细节：ServerSentEvent的comment字段不会显示给前端，但可以用来传递元数据，前端通过解析event和data来更新状态面板。节点状态和数据流通过两个不同的事件类型推送，前端各做各的渲染，互不干扰。
4. 从0到1的实操过程：一个可用的Agent流程引擎
4.1 环境准备与项目结构
实际操作环境建议：
| 组件 | 版本 | 说明 |
|---|---|---|
| JDK | 17+ | 用到了record、sealed等新特性 |
| Spring Boot | 3.2+ | WebFlux支持内置SSE |
| Maven | 3.8+ | 依赖管理 |
| Reactor | 3.6+ | WebFlux自带 |
项目结构很清爽：
src/main/java/com/example/agentflow ├── engine │ ├── WorkflowEngine.java │ ├── WorkflowContext.java │ ├── NodeResult.java │ ├── NodeStatus.java │ └── Node.java ├── node │ ├── StartNode.java │ ├── IntentNode.java │ ├── ToolCallNode.java │ ├── GenerateNode.java │ └── EndNode.java ├── event │ ├── FlowEvent.java │ └── AgentFlowController.java └── Application.java这是一个极其简化的结构，实际项目里我还会把NodeResult和WorkflowContext做更丰富的扩展，比如WorkflowContext里放会话ID、用户ID、traceId，方便追踪整条链路。
4.2 节点与流程定义
我用一个"查询天气并生成回复"的Agent流程来演示。流程是：START -> INTENT -> TOOL_CALL -> GENERATE -> END。如果意图不是查询天气，TOOL_CALL节点会被跳过，直接走向GENERATE。
先定义各个节点的核心逻辑。
StartNode，流程入口，只有一个作用就是决定第一个实际节点：
public class StartNode extends Node { @Override protected NodeResult doExecute(WorkflowContext context) { return NodeResult.success("INTENT"); } }IntentNode，调用大模型(这里简化成硬编码判断)识别意图：
public class IntentNode extends Node { @Override protected NodeResult doExecute(WorkflowContext context) { String userInput = context.get("userInput", String.class); if (userInput.contains("天气")) { context.put("intent", "weather"); return NodeResult.success("TOOL_CALL"); } context.put("intent", "chat"); return NodeResult.success("GENERATE"); } }ToolCallNode，调用天气工具：
public class ToolCallNode extends Node { @Override protected NodeResult doExecute(WorkflowContext context) { String city = context.get("city", String.class); // 这里模拟工具调用，实际项目中会换成HTTP调用或函数方法 String weather = "晴，25度"; context.put("weather", weather); // 推送工具调用事件 if (sink != null) { sink.tryEmitNext(ServerSentEvent.builder("调用天气工具完成：" + weather) .event("tool_result") .build()); } return NodeResult.success("GENERATE"); } }这里注意，所有节点都必须通过NodeResult指定下一步节点名，这是状态轮转的关键约束。如果一个节点没有下一步了，就返回null或者用一个END节点兜底。
GenerateNode，生成回复：
public class GenerateNode extends Node { @Override protected NodeResult doExecute(WorkflowContext context) { String intent = context.get("intent", String.class); if ("weather".equals(intent)) { String weather = context.get("weather", String.class); String answer = "当前城市天气：" + weather; context.put("answer", answer); // 模拟流式输出：逐字推送 for (char c : answer.toCharArray()) { if (sink != null) { sink.tryEmitNext(ServerSentEvent.builder(String.valueOf(c)) .event("token") .build()); } try { Thread.sleep(50); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } else { context.put("answer", "我现在只能查询天气哦"); } return NodeResult.success("END"); } }EndNode什么都不做，返回null让引擎结束：
public class EndNode extends Node { @Override protected NodeResult doExecute(WorkflowContext context) { return NodeResult.end(); } }4.3 核心代码实现
WorkflowContext就是一个简单的Map封装，但为了防止在并发节点执行时数据错乱，我用的是ConcurrentHashMap：
public class WorkflowContext { private final Map<String, Object> data = new ConcurrentHashMap<>(); public void put(String key, Object value) { data.put(key, value); } public <T> T get(String key, Class<T> type) { return type.cast(data.get(key)); } }NodeResult负责承载节点执行结果。我给它加了几个静态工厂方法，让节点代码更简洁：
public class NodeResult { private final String targetNodeId; private final NodeStatus status; private final boolean allowRetry; private NodeResult(String targetNodeId, NodeStatus status, boolean allowRetry) { this.targetNodeId = targetNodeId; this.status = status; this.allowRetry = allowRetry; } public static NodeResult success(String targetNodeId) { return new NodeResult(targetNodeId, NodeStatus.SUCCESS, false); } public static NodeResult failure(String targetNodeId, boolean allowRetry) { return new NodeResult(targetNodeId, NodeStatus.FAILED, allowRetry); } public static NodeResult end() { return new NodeResult(null, NodeStatus.SUCCESS, false); } public static NodeResult skip(String targetNodeId) { return new NodeResult(targetNodeId, NodeStatus.SKIPPED, false); } public static NodeResult waitForExternal(String targetNodeId) { return new NodeResult(targetNodeId, NodeStatus.WAITING, false); } // getter省略 }接着把上面的引擎主循环补全，加入事件sink和异步执行能力。异步执行我用一个简单的线程池封装：
public class WorkflowEngine { private final Map<String, Node> nodeRegistry = new ConcurrentHashMap<>(); private final ExecutorService executor = Executors.newCachedThreadPool(); private Sinks.Many<ServerSentEvent<String>> sink; public void setEventSink(Sinks.Many<ServerSentEvent<String>> sink) { this.sink = sink; nodeRegistry.values().forEach(node -> node.setEventSink(sink)); } public void registerNode(Node node) { node.setEventSink(sink); nodeRegistry.put(node.getName(), node); } public void runAsync(String startNodeName) { executor.submit(() -> run(startNodeName)); } public void run(String startNodeName) { String currentNodeName = startNodeName; int maxStep = 100; int step = 0; while (currentNodeName != null) { if (step++ >= maxStep) { throw new IllegalStateException("流程疑似死循环，超过最大步数"); } Node currentNode = nodeRegistry.get(currentNodeName); if (currentNode == null) { throw new IllegalStateException("未找到节点: " + currentNodeName); } NodeResult result = currentNode.execute(createContext()); NodeStatus status = result.getStatus(); if (status == NodeStatus.FAILED && result.isAllowRetry()) { continue; } if (status == NodeStatus.WAITING) { // 流程挂起，等待外部resume break; } currentNodeName = result.getTargetNodeId(); } if (sink != null) { sink.tryEmitNext(ServerSentEvent.builder("流程结束").event("flow_end").build()); sink.tryEmitComplete(); } } }4.4 运行效果演示
启动应用后，用curl模拟前端发起SSE请求：
curl -N -H "Accept: text/event-stream" http://localhost:8080/execute?userInput=北京天气怎么样响应流会依次出现：
event: node_status data: START : RUNNING event: node_status data: INTENT : RUNNING event: token data: 当 event: token data: 前 event: token data: 城 event: token data: 市 event: node_status data: TOOL_CALL : SUCCESS event: tool_result data: 调用天气工具完成：晴，25度 event: flow_end data: 流程结束前端可以根据事件类型做不同的UI渲染：node_status更新状态面板，token追加到正文，tool_result展示工具卡片。这套联动让Agent执行过程可视化程度非常高，实际体验很接近ChatGPT那种"思考中...然后开始输出答案"的交互。
5. 常见问题与排查技巧实录
5.1 节点重复执行问题
我在实现初期犯过一个非常经典的错：在run循环里，当一个节点执行完并返回SUCCESS后，如果目标节点ID正好还是它自己（比如配错了图），就会导致同一个节点无限循环执行。后来我在引擎里加了最大步数兜底，但这只是"防止死循环"，并没有真正解决"重复执行"的隐患。
真正需要重视的是节点幂等性。Agent流程里经常有"发送通知""写入日志"这类节点，如果在流程恢复或重试时被执行两次，就会产生重复副作用。我的解决方案是在节点基类提供isIdempotent()方法，默认返回false；对于非幂等节点，引擎收到外部resume请求时，不会从当前节点重新执行，而是从它的下一个节点开始。另外，节点执行前在WorkflowContext里记录一个executionId，同一executionId的节点即使被再次触发也会被跳过，用"执行标记"保证单次流程内每个节点最多执行一次。
5.2 SSE流式输出中断和乱序
SSE本身是基于HTTP长连接，如果中间有Nginx代理，默认可能有缓冲和超时问题。我本地调试好好的，一上测试环境就发现前端收不到消息，最后查出来是Nginx没关缓冲：
proxy_buffering off; proxy_cache off; proxy_read_timeout 3600s;这一点是SSE部署的经典坑，遇到"本地能用，走代理不行"的问题，优先检查这行配置。
乱序问题是我在并发节点时候踩的坑。虽然我现在演示的流程是串行的，但Agent实际运行中有些节点可以并行，比如同时调用两个工具。如果两个并行节点同时往sink里emit事件，前端拿到的顺序可能跟期望不一致。我的做法是给FlowEvent加一个递增的sequence字段，前端收到事件后按sequence重排。简单有效，不用引入复杂的分布式消息。
5.3 引擎如何扩展新的节点类型
很多朋友会问：新需求来了，我要加一个"发送邮件"节点怎么办？非常简单：
- 写一个类继承Node，实现doExecute方法；
- 在doExecute里指定它的下一步节点（通常是某个后续节点名）；
- 在启动配置里注册这个节点。
整个过程不需要改动WorkflowEngine和任何既有节点。这就是状态轮转带来的核心收益——新增节点是加法操作，不是修改操作。如果某个新的Agent流程需要不同的编排，我甚至可以直接用配置文件定义节点顺序，引擎启动时读取配置构建注册表。这个我已经在做了，流程定义完全外置，业务上写流程编排就像搭乐高。
5.4 并发场景下的引擎安全
Agent服务不可能只有一个用户，所以引擎必须支持多流程并发。我刚开始把WorkflowContext定义成单例，多个流程同时跑的时候数据互相污染。后来改成每次流程启动时创建独立的Context，并把Context作为NodeResult执行时的参数传递进去。
具体做法是在注册节点时只存节点实例，但节点内部不能持有任何用户态的成员变量。所有数据都放进Context。如果一个节点是有状态的（比如需要缓存某次工具调用的结果），就把缓存数据放到Context里，而不是节点的字段里。这个经验对我来说是血泪教训，最初那个节点里塞了个List存放临时结果，结果两个用户并发请求时数据串得一塌糊涂，排查了一下午。
6. 写在最后：一点实战体会
整套流程引擎做下来，最大的体会是："不用if-else"不是目的，流程可编排才是目的。if-else是图灵完备的，什么流程都能写，但问题在于它不可观测、不可复用、不可动态调整。状态轮转把流程的"形状"从代码中抽离出来，让每个节点只关心自己的事，这是Agent这种复杂场景下更优雅的建模方式。
最后分享一个小技巧：我引擎里加了一个"调试模式"开关，开启后每个节点执行前后的状态变化、目标节点、耗时都会通过流式事件推给前端调试面板。这个功能在联调时帮了大忙——团队成员肉眼就能看到流程走到哪一步、为什么走到这一步，排查问题的速度翻倍。
如果你也在设计Agent工作流，建议先从最精简的引擎开始，只做到节点注册和状态轮转，再逐步加上流式输出、并行节点、人工确认、重试策略。这套架构的扩展方向非常多，但核心模型一旦稳固，后面加功能都是水到渠成的事。
