輪轉(zhuǎn)替代if-else：構(gòu)建Agent流程引擎及SSE流式輸出實(shí)戰(zhàn))
好久沒寫工程向的分享了。前面一個(gè)項(xiàng)目里我負(fù)責(zé)做一個(gè)智能客服Agent一開始圖省事所有流程控制全用if-else堆如果意圖不明就追問如果知識(shí)庫沒命中就兜底如果LLM調(diào)用失敗就重試……兩周之后代碼已經(jīng)沒法看了加一個(gè)節(jié)點(diǎn)要?jiǎng)尤奶幍胤綔y(cè)試用例寫了一堆還是漏。后來我花了一個(gè)周末把流程控制層整個(gè)重寫成基于狀態(tài)輪轉(zhuǎn)的流程引擎——Java實(shí)現(xiàn)節(jié)點(diǎn)抽象狀態(tài)顯式定義流轉(zhuǎn)用路由表驅(qū)動(dòng)執(zhí)行過程用SSE流式輸出到前端。核心代碼大概不到300行if-else基本被消滅了。這篇文章就圍繞這個(gè)設(shè)計(jì)展開講講節(jié)點(diǎn)狀態(tài)輪轉(zhuǎn)和流式輸出到底怎么落地順便把踩過的坑一并整理出來適合正在做Agent開發(fā)、AI應(yīng)用編排、或者被業(yè)務(wù)分支邏輯折磨的Java工程師參考。1. 為什么Agent工作流不能靠if-else硬寫1.1 當(dāng)Agent流程開始復(fù)雜if-else的賬就算不清了Agent的工作流和普通后端接口最大的不同在于它的執(zhí)行路徑幾乎不受你控制。一個(gè)典型Agent要經(jīng)歷意圖識(shí)別、槽位補(bǔ)齊、知識(shí)庫檢索、工具調(diào)用、上下文記憶更新、答案生成這幾個(gè)階段而每個(gè)階段都可能產(chǎn)生分支意圖置信度低要追問信息不足要澄清檢索結(jié)果為空要走兜底工具執(zhí)行超時(shí)要重試。把這些分支全用if-else寫本質(zhì)上就是在用“順序代碼”去描述一張“有向圖”。問題在前端接口還能忍一旦節(jié)點(diǎn)數(shù)超過五個(gè)代碼就會(huì)變成這個(gè)樣子外層if判斷結(jié)果類型內(nèi)層if判斷狀態(tài)碼再內(nèi)層if判斷是否需要調(diào)用另一個(gè)方法。每新增一個(gè)節(jié)點(diǎn)你要找到所有上游節(jié)點(diǎn)的出口邏輯一處處補(bǔ)判斷。更難受的是調(diào)試——線上用戶觸發(fā)了某條冷門分支你根本不知道當(dāng)時(shí)走到了哪一步因?yàn)闆]有狀態(tài)記錄。我見過不少團(tuán)隊(duì)在Agent項(xiàng)目里維護(hù)一段超過500行的switch-case每加一個(gè)工具調(diào)用就往里塞一個(gè)case分支。這本質(zhì)上和if-else沒有區(qū)別只是換了個(gè)寫法。流程控制一旦退化成這種形式它就不再是“可編排”的了而是“寫死”的。1.2 狀態(tài)機(jī)思想從路牌到流程圖如果退一步看問題Agent的執(zhí)行過程本質(zhì)上就是一個(gè)狀態(tài)機(jī)每個(gè)節(jié)點(diǎn)是一個(gè)狀態(tài)節(jié)點(diǎn)的執(zhí)行結(jié)果決定下一個(gè)狀態(tài)節(jié)點(diǎn)在流轉(zhuǎn)中會(huì)產(chǎn)生輸出事件。用狀態(tài)機(jī)的思路去建模比用代碼堆邏輯要自然得多。我打個(gè)比方你開車去一個(gè)陌生的地方靠的不是在每個(gè)路口背誦“如果看到A就左轉(zhuǎn)看到B就右轉(zhuǎn)否則直行”這種if-else口訣而是一張路線圖每個(gè)路口告訴你當(dāng)前在哪個(gè)位置根據(jù)當(dāng)前位置決定下一段怎么走。狀態(tài)輪轉(zhuǎn)就是這張路線圖節(jié)點(diǎn)就是路口的指示牌執(zhí)行結(jié)果就是你要做的選擇。流程引擎的核心價(jià)值在這里體現(xiàn)出來了它把“業(yè)務(wù)邏輯”和“流程控制”徹底拆開。業(yè)務(wù)邏輯寫在節(jié)點(diǎn)內(nèi)部流程控制交給引擎的路由機(jī)制。以后哪怕要調(diào)整整個(gè)Agent的執(zhí)行順序你也不需要改業(yè)務(wù)代碼在路由表里重新編排節(jié)點(diǎn)ID就可以。2. 流程引擎的整體設(shè)計(jì)先抽象再實(shí)現(xiàn)2.1 核心抽象Node節(jié)點(diǎn)和Workflow工作流設(shè)計(jì)一個(gè)流程引擎最重要的不是先寫代碼而是定義好抽象。我這邊核心抽象只有兩個(gè)Node節(jié)點(diǎn)和Workflow工作流。Node是流程中的最小執(zhí)行單元。它負(fù)責(zé)做三件事聲明自己的ID、執(zhí)行具體的業(yè)務(wù)邏輯、明確執(zhí)行完成之后下一步去哪個(gè)節(jié)點(diǎn)。業(yè)務(wù)邏輯五花八門沒關(guān)系我把它統(tǒng)一收斂到一個(gè)抽象方法里返回執(zhí)行狀態(tài)。狀態(tài)決定了路由方向。Workflow則是節(jié)點(diǎn)的集合和入口配置。它知道整個(gè)流程從哪個(gè)節(jié)點(diǎn)開始手里有一張注冊(cè)表存著所有節(jié)點(diǎn)的ID到實(shí)例的映射。引擎執(zhí)行的時(shí)候只需要拿到入口節(jié)點(diǎn)ID然后不停查詢當(dāng)前節(jié)點(diǎn)、執(zhí)行當(dāng)前節(jié)點(diǎn)、根據(jù)返回狀態(tài)找下一個(gè)節(jié)點(diǎn)循環(huán)往復(fù)直到走到終止?fàn)顟B(tài)。這里有一個(gè)很重要的設(shè)計(jì)取舍節(jié)點(diǎn)之間互相不直接依賴。A節(jié)點(diǎn)執(zhí)行完后它不需要知道B節(jié)點(diǎn)是什么只需要在路由表里聲明“狀態(tài)SUCCESS時(shí)去node_detail_query這個(gè)ID”。這種解耦帶來的直接好處是節(jié)點(diǎn)可以獨(dú)立測(cè)試可以隨意替換也可以在不同工作流里復(fù)用。2.2 狀態(tài)輪轉(zhuǎn)如何替代if-else傳統(tǒng)寫法里控制權(quán)在“調(diào)用方”手里。一個(gè)方法調(diào)用另一個(gè)方法根據(jù)返回值決定下一步層層嵌套。狀態(tài)輪轉(zhuǎn)的設(shè)計(jì)里控制權(quán)被上交給“引擎”手里。每個(gè)節(jié)點(diǎn)執(zhí)行完成后會(huì)返回一個(gè)NodeStatus引擎拿到這個(gè)狀態(tài)去查當(dāng)前節(jié)點(diǎn)的路由表拿到目標(biāo)節(jié)點(diǎn)ID接著執(zhí)行下一個(gè)節(jié)點(diǎn)。也就是說分支邏輯從“代碼的調(diào)用關(guān)系”變成了“數(shù)據(jù)表里的映射關(guān)系”。if-else被徹底從代碼里剝離你看到的是清晰的路由表SUCCESS → 下一步做什么FAILED → 是重試還是走兜底NEED_INPUT → 是否要跳轉(zhuǎn)到追問節(jié)點(diǎn)有人會(huì)說這不就是把if-else搬了個(gè)位置嗎其實(shí)區(qū)別很大。if-else是硬編碼流程一旦寫錯(cuò)或者要調(diào)整必須改代碼重新發(fā)布路由表是結(jié)構(gòu)化數(shù)據(jù)可以配置化、可視化、動(dòng)態(tài)修改。而且對(duì)于流程引擎來說路由表天然適合做遍歷檢查——你可以啟動(dòng)時(shí)掃描所有節(jié)點(diǎn)的路由目標(biāo)檢測(cè)有沒有指向不存在的節(jié)點(diǎn)ID有沒有死循環(huán)。這些能力是普通if-else寫法根本給不了的。2.3 為什么不用Activiti、Flowable這些現(xiàn)成引擎肯定有人要問Java生態(tài)里不是有Activiti、Flowable、Camunda這些成熟的工作流引擎嗎為什么還要自己寫我自己評(píng)估過這些引擎強(qiáng)在“人參與審批”的場(chǎng)景任務(wù)分配、會(huì)簽、或簽、超時(shí)提醒。但Agent場(chǎng)景完全是另一回事節(jié)點(diǎn)執(zhí)行的是LLM推理調(diào)用、工具API調(diào)用、向量檢索不是人工填表審批。引入Activiti這種重型BPM框架光是把流程定義文件和數(shù)據(jù)模型接進(jìn)來就要花不少功夫再加上它們本身那套部署方式、歷史表結(jié)構(gòu)對(duì)一個(gè)小團(tuán)隊(duì)來說負(fù)擔(dān)很重。更關(guān)鍵的一點(diǎn)是流式輸出。Agent執(zhí)行過程需要實(shí)時(shí)把中間狀態(tài)推到前端——用戶能看著“正在理解意圖→正在檢索知識(shí)庫→正在生成回答”這是現(xiàn)代Agent應(yīng)用的基本體驗(yàn)要求。Activiti這類引擎的監(jiān)聽機(jī)制是為系統(tǒng)內(nèi)部事件設(shè)計(jì)的要接到SSE推送還得自己搭一層橋梁。與其繞一圈適配不如直接自己寫一個(gè)輕量引擎保留最核心的節(jié)點(diǎn)路由能力把事件推送機(jī)制做成一等公民。實(shí)測(cè)下來核心引擎不到300行后面所有業(yè)務(wù)節(jié)點(diǎn)都在復(fù)用這一套骨架。3. 核心實(shí)現(xiàn)節(jié)點(diǎn)狀態(tài)輪轉(zhuǎn)引擎代碼拆解3.1 狀態(tài)枚舉與節(jié)點(diǎn)抽象類先看最基本的定義狀態(tài)枚舉我一開始只設(shè)計(jì)了5個(gè)狀態(tài)后來加了TIMEOUT和TERMINATED別嫌多實(shí)際寫業(yè)務(wù)的時(shí)候你會(huì)發(fā)現(xiàn)每個(gè)狀態(tài)都有對(duì)應(yīng)場(chǎng)景。public enum NodeStatus { PENDING(待執(zhí)行), RUNNING(執(zhí)行中), SUCCESS(成功), FAILED(失敗), SKIPPED(跳過), TIMEOUT(超時(shí)), TERMINATED(終止); private final String desc; NodeStatus(String desc) { this.desc desc; } public String getDesc() { return desc; } }然后是節(jié)點(diǎn)的抽象基類。核心就兩個(gè)東西一個(gè)execute方法用于執(zhí)行業(yè)務(wù)并返回狀態(tài)一個(gè)路由表用于指明不同狀態(tài)分別去哪個(gè)節(jié)點(diǎn)。public abstract class BaseNodeT { private final String id; private final String name; private final MapNodeStatus, String routeTable new EnumMap(NodeStatus.class); public BaseNode(String id, String name) { this.id id; this.name name; } public String getId() { return id; } public String getName() { return name; } public abstract NodeStatus execute(NodeContextT context); protected void on(NodeStatus status, String targetNodeId) { routeTable.put(status, targetNodeId); } public String route(NodeStatus status) { return routeTable.get(status); } }設(shè)計(jì)要點(diǎn)在route方法執(zhí)行完一個(gè)節(jié)點(diǎn)引擎把返回狀態(tài)交給路由表查詢找到的就是下一站節(jié)點(diǎn)ID。如果某個(gè)狀態(tài)沒有配置路由說明這個(gè)狀態(tài)就是終態(tài)引擎會(huì)終止循環(huán)。這里我建議一個(gè)實(shí)際優(yōu)化在做驗(yàn)證時(shí)遍歷注冊(cè)表里的所有節(jié)點(diǎn)檢查每個(gè)節(jié)點(diǎn)的路由目標(biāo)是否存在于注冊(cè)表中。這個(gè)檢查我寫在啟動(dòng)階段一旦發(fā)現(xiàn)懸空引用直接報(bào)錯(cuò)比運(yùn)行時(shí)走到一半才發(fā)現(xiàn)要好得多。3.2 執(zhí)行上下文讓所有節(jié)點(diǎn)共享數(shù)據(jù)和狀態(tài)節(jié)點(diǎn)之間光靠參數(shù)傳遞是不夠的Agent執(zhí)行過程中會(huì)產(chǎn)生大量上下文數(shù)據(jù)比如用戶的原始問題、意圖識(shí)別的結(jié)果、知識(shí)庫檢索出來的片段、歷史對(duì)話記錄這些數(shù)據(jù)需要在不同節(jié)點(diǎn)之間流轉(zhuǎn)。為此我設(shè)計(jì)了一個(gè)NodeContextpublic class NodeContextT { private final String traceId; private final T payload; private final MapString, Object variables new ConcurrentHashMap(); private String currentNodeId; private NodeStatus currentNodeStatus; public NodeContext(String traceId, T payload) { this.traceId traceId; this.payload payload; } public void setVariable(String key, Object value) { variables.put(key, value); } SuppressWarnings(unchecked) public V V getVariable(String key) { return (V) variables.get(key); } public String getTraceId() { return traceId; } public T getPayload() { return payload; } public String getCurrentNodeId() { return currentNodeId; } public void markNode(String nodeId, NodeStatus status) { this.currentNodeId nodeId; this.currentNodeStatus status; } }traceId是每次Agent請(qǐng)求的唯一標(biāo)識(shí)貫穿整個(gè)執(zhí)行鏈路既用于日志追蹤也用于流式輸出時(shí)關(guān)聯(lián)事件。variables是節(jié)點(diǎn)間的數(shù)據(jù)交換層A節(jié)點(diǎn)把意圖識(shí)別的結(jié)果放進(jìn)去B節(jié)點(diǎn)直接取不需要方法簽名級(jí)別的耦合。實(shí)際項(xiàng)目中這個(gè)上下文還可以擴(kuò)展出上下文窗口管理、令牌消耗統(tǒng)計(jì)、重試次數(shù)記錄等能力。它本質(zhì)上是Agent的“工作記憶”記得在哪一步停過、拿過什么結(jié)果、下一步還缺什么信息。3.3 引擎主循環(huán)用路由表驅(qū)動(dòng)節(jié)點(diǎn)跳轉(zhuǎn)引擎是整個(gè)流程的中樞它負(fù)責(zé)調(diào)度節(jié)點(diǎn)、處理狀態(tài)輪轉(zhuǎn)、廣播事件。我把它設(shè)計(jì)成可復(fù)用的通用組件public class FlowEngineT { private final MapString, BaseNodeT registry new HashMap(); private final ExecutorService executorService; private final ListWorkflowListener listeners new CopyOnWriteArrayList(); public FlowEngine(ExecutorService executorService) { this.executorService executorService; } public void registerNode(BaseNodeT node) { registry.put(node.getId(), node); } public void addListener(WorkflowListener listener) { listeners.add(listener); } public void start(String entryNodeId, T payload, String traceId) { executorService.submit(() - execute(entryNodeId, payload, traceId)); } private void execute(String entryNodeId, T payload, String traceId) { NodeContextT context new NodeContext(traceId, payload); String currentNodeId entryNodeId; try { while (currentNodeId ! null) { BaseNodeT node registry.get(currentNodeId); if (node null) { throw new IllegalStateException(節(jié)點(diǎn)不存在: currentNodeId); } context.markNode(currentNodeId, NodeStatus.RUNNING); emitEvent(NodeEvent.started(context, System.currentTimeMillis())); NodeStatus status node.execute(context); context.markNode(currentNodeId, status); emitEvent(NodeEvent.finished(context, System.currentTimeMillis())); currentNodeId node.route(status); } emitEvent(NodeEvent.completed(context, System.currentTimeMillis())); } catch (Exception e) { context.setVariable(lastError, e.getMessage()); emitEvent(NodeEvent.failed(context, e, System.currentTimeMillis())); } } private void emitEvent(NodeEvent event) { for (WorkflowListener listener : listeners) { listener.onEvent(event); } } }主循環(huán)邏輯很簡潔根據(jù)當(dāng)前節(jié)點(diǎn)ID拿到節(jié)點(diǎn)實(shí)例執(zhí)行取狀態(tài)查路由得到下一個(gè)節(jié)點(diǎn)ID。沒有判斷流程邏輯的if-else所有分支都收斂在節(jié)點(diǎn)的路由表里。這里我想強(qiáng)調(diào)一個(gè)容易忽略的點(diǎn)開始執(zhí)行用的start方法是異步的把任務(wù)丟進(jìn)線程池立即返回。為什么因?yàn)楣ぷ髁饕浜狭魇捷敵鋈绻划惒交疕TTP響應(yīng)線程會(huì)被阻塞前端的SSE流根本沒機(jī)會(huì)實(shí)時(shí)收到事件。異步化之后整個(gè)執(zhí)行變成事件驅(qū)動(dòng)的推送模型。4. 流式輸出讓Agent執(zhí)行過程實(shí)時(shí)可見4.1 全量返回為什么會(huì)讓用戶焦慮Agent場(chǎng)景和普通接口最大的體驗(yàn)差異在于普通接口用戶等一兩秒拿到結(jié)果沒問題但Agent執(zhí)行一個(gè)復(fù)雜任務(wù)可能耗時(shí)幾十秒甚至幾分鐘——比如多輪工具調(diào)用再加長文本生成。如果用戶點(diǎn)完按鈕之后頁面一直轉(zhuǎn)圈沒有任何過程反饋體驗(yàn)是災(zāi)難性的。流式輸出解決的就是這個(gè)問題。它的本質(zhì)是把Agent執(zhí)行過程中產(chǎn)生的每一個(gè)階段性事件從服務(wù)端實(shí)時(shí)推送到客戶端。注意是事件流不只是最終答案。用戶能看到“正在判斷意圖”→“正在檢索知識(shí)庫”→“正在調(diào)用外部工具”→“正在生成回答”每一條狀態(tài)變化都實(shí)時(shí)渲染到頁面上。技術(shù)選型上我用了SSEServer-Sent Events而不是WebSocket。原因有兩點(diǎn)Agent執(zhí)行狀態(tài)是服務(wù)端到客戶端的單向數(shù)據(jù)流不需要客戶端頻繁上行消息SSE基于HTTP協(xié)議不需要額外的握手協(xié)議前端的EventSource對(duì)象可以直接消費(fèi)實(shí)現(xiàn)成本低得多。4.2 基于SseEmitter的事件推送實(shí)現(xiàn)Spring Boot生態(tài)里SseEmitter是原生的SSE支持用起來很直接。我把它和流程引擎的事件監(jiān)聽器接在一起執(zhí)行過程中每產(chǎn)生一個(gè)事件就通過emitter推送給前端RestController public class AgentController { private final FlowEngineAgentRequest flowEngine; private final ExecutorService executorService Executors.newCachedThreadPool(); public AgentController(FlowEngineAgentRequest flowEngine) { this.flowEngine flowEngine; } GetMapping(value /agent/run, produces MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter run(RequestParam String question) { String traceId UUID.randomUUID().toString().substring(0, 8); SseEmitter emitter new SseEmitter(60_000L); WorkflowListener listener event - { try { MapString, Object data new HashMap(); data.put(traceId, event.getTraceId()); data.put(nodeId, event.getContext().getCurrentNodeId()); data.put(status, event.getContext().getCurrentNodeStatus().name()); data.put(type, event.getType().name()); if (event.getTargetNodeId() ! null) { data.put(next, event.getTargetNodeId()); } emitter.send(SseEmitter.event().name(agent-event).data(data)); } catch (IOException e) { emitter.completeWithError(e); } }; flowEngine.addListener(listener); emitter.onCompletion(() - flowEngine.removeListener(listener)); emitter.onTimeout(() - flowEngine.removeListener(listener)); flowEngine.start(ENTRY_NODE_ID, new AgentRequest(question), traceId); return emitter; } }這里有一個(gè)非常關(guān)鍵的設(shè)計(jì)SseEmitter推送給前端的不只是最終結(jié)果而是執(zhí)行過程中的每一次狀態(tài)輪轉(zhuǎn)。節(jié)點(diǎn)開始執(zhí)行推一個(gè)started事件節(jié)點(diǎn)結(jié)束推一個(gè)finished事件事件里帶上當(dāng)前節(jié)點(diǎn)ID和狀態(tài)。前端拿到這些事件之后就可以在頁面上逐個(gè)渲染狀態(tài)變化。用戶看到的是“流程在往前走”而不是一個(gè)死等loading。數(shù)據(jù)事件里還帶了next字段如果有這個(gè)字段說明流程還要繼續(xù)如果沒有說明這個(gè)節(jié)點(diǎn)是終態(tài)。4.3 線程池、事件監(jiān)聽器與斷連處理流式輸出本身不復(fù)雜復(fù)雜在工程細(xì)節(jié)。第一個(gè)細(xì)節(jié)是線程隔離。Agent工作流的執(zhí)行必須放在獨(dú)立的線程池里不能占著Tomcat的請(qǐng)求線程。如果直接在請(qǐng)求線程里跑工作流SseEmitter就永遠(yuǎn)來不及發(fā)送數(shù)據(jù)因?yàn)轫憫?yīng)還沒返回。我用了獨(dú)立的線程池請(qǐng)求線程只負(fù)責(zé)創(chuàng)建emitter和注冊(cè)監(jiān)聽器真正的工作流執(zhí)行在另一個(gè)線程里異步跑。第二個(gè)細(xì)節(jié)是監(jiān)聽器的生命周期管理。每個(gè)SSE連接對(duì)應(yīng)一個(gè)監(jiān)聽器實(shí)例連接關(guān)閉或者超時(shí)之后監(jiān)聽器必須從引擎里移除。否則emitter已經(jīng)關(guān)閉了工作流還繼續(xù)往里面send數(shù)據(jù)會(huì)拋IOException。我在onCompletion和onTimeout回調(diào)里都調(diào)用了removeListener確保連接斷開后不再有推送動(dòng)作。第三個(gè)細(xì)節(jié)是超時(shí)設(shè)置。SseEmitter的構(gòu)造參數(shù)是超時(shí)毫秒數(shù)我設(shè)置了60秒。如果Agent在這個(gè)時(shí)間內(nèi)還沒跑完連接會(huì)斷掉用戶需要重新發(fā)起請(qǐng)求。后續(xù)我改成了更合理的方案用定時(shí)器在超時(shí)前刷新一下連接或者把超時(shí)設(shè)成一個(gè)比較長的值配合心跳機(jī)制保持連接活性。對(duì)于生產(chǎn)環(huán)境建議單獨(dú)評(píng)估每個(gè)Agent流程的最大耗時(shí)再?zèng)Q定超時(shí)時(shí)長。5. 實(shí)操案例從0搭建一個(gè)智能客服Agent5.1 節(jié)點(diǎn)設(shè)計(jì)與路由編排理論講了一堆最終要落到一個(gè)能跑通的例子。我以智能客服Agent為例設(shè)計(jì)了5個(gè)業(yè)務(wù)節(jié)點(diǎn)intent_node意圖識(shí)別節(jié)點(diǎn)。調(diào)用LLM判斷用戶意圖返回四個(gè)狀態(tài)BUY想買產(chǎn)品、AFTER_SALE售后咨詢、OTHER閑聊、UNKNOWN無法識(shí)別。clarify_node追問節(jié)點(diǎn)。當(dāng)意圖識(shí)別返回UNKNOWN時(shí)追問用戶具體需求然后重新路由回意圖識(shí)別節(jié)點(diǎn)。knowledge_node知識(shí)庫檢索節(jié)點(diǎn)。根據(jù)意圖和用戶問題做向量檢索從知識(shí)庫中匹配相關(guān)內(nèi)容。如果檢索到的內(nèi)容相似度低于閾值返回FAILED。fallback_node兜底節(jié)點(diǎn)。知識(shí)庫檢索不到內(nèi)容時(shí)給出一個(gè)預(yù)設(shè)話術(shù)引導(dǎo)用戶轉(zhuǎn)接人工。answer_node答案生成節(jié)點(diǎn)。把檢索結(jié)果和歷史上下文拼進(jìn)Prompt調(diào)用LLM生成最終回答。節(jié)點(diǎn)定義好了路由表就是整個(gè)Agent的“劇本”。我專門用一個(gè)配置類把路由關(guān)系集中管理Configuration public class AgentWorkflowConfig { public static final String ENTRY_NODE intent_node; Bean public FlowEngineAgentRequest agentFlowEngine() { FlowEngineAgentRequest engine new FlowEngine(Executors.newFixedThreadPool(8)); BaseNodeAgentRequest intentNode new BaseNodeAgentRequest(intent_node, 意圖識(shí)別) { Override public NodeStatus execute(NodeContextAgentRequest context) { IntentResult intent llmService.recognizeIntent(context.getPayload().getQuestion()); context.setVariable(intent, intent); if (intent.confidence() 0.6) { return NodeStatus.NEED_INPUT; } return NodeStatus.SUCCESS; } }; intentNode.on(NodeStatus.SUCCESS, knowledge_node); intentNode.on(NodeStatus.NEED_INPUT, clarify_node); intentNode.on(NodeStatus.FAILED, fallback_node); BaseNodeAgentRequest clarifyNode new BaseNodeAgentRequest(clarify_node, 追問) { Override public NodeStatus execute(NodeContextAgentRequest context) { return confirmAnswer(context) ? NodeStatus.SUCCESS : NodeStatus.FAILED; } }; clarifyNode.on(NodeStatus.SUCCESS, intent_node); clarifyNode.on(NodeStatus.FAILED, fallback_node); // knowledge_node、fallback_node、answer_node 類似按流程編排 engine.registerNode(intentNode); engine.registerNode(clarifyNode); engine.registerNode(knowledgeNode); engine.registerNode(fallbackNode); engine.registerNode(answerNode); return engine; } }這個(gè)設(shè)計(jì)的好處你寫幾個(gè)節(jié)點(diǎn)就能體會(huì)出來。比如我想在答案生成之前加一個(gè)“敏感詞檢查”節(jié)點(diǎn)只需要新建一個(gè)節(jié)點(diǎn)類然后把a(bǔ)nswer_node的上游路由改一下其他節(jié)點(diǎn)的代碼一個(gè)字都不用動(dòng)。5.2 注冊(cè)節(jié)點(diǎn)并啟動(dòng)工作流節(jié)點(diǎn)配置完成后啟動(dòng)一次Agent工作流只需要一行代碼flowEngine.start(AgentWorkflowConfig.ENTRY_NODE, new AgentRequest(question), traceId);引擎會(huì)自動(dòng)從intent_node開始一路輪轉(zhuǎn)下去。每個(gè)節(jié)點(diǎn)的執(zhí)行結(jié)果都會(huì)觸發(fā)狀態(tài)事件這些事件經(jīng)過監(jiān)聽器、SseEmitter最終變成前端頁面上的實(shí)時(shí)狀態(tài)更新。我在實(shí)際部署中還加了一個(gè)內(nèi)容審核節(jié)點(diǎn)放在最后專門檢查LLM生成結(jié)果是否合規(guī)合規(guī)度低就重新生成一次最多重試兩次。這些控制都通過路由表表達(dá)answer_node的下一步根據(jù)審核結(jié)果分別指向結(jié)束節(jié)點(diǎn)或者重試節(jié)點(diǎn)。流程的復(fù)雜度上來了但代碼結(jié)構(gòu)一直是同一套。5.3 前端收到的事件流長什么樣配套的前端邏輯也很簡單。用瀏覽器原生的EventSource監(jiān)聽事件每次收到消息就更新頁面上的步驟狀態(tài)const eventSource new EventSource(/agent/run?question encodeURIComponent(question)); eventSource.addEventListener(agent-event, (event) { const data JSON.parse(event.data); renderNodeStatus(data.nodeId, data.status); if (data.next) { appendNode(data.next, 等待執(zhí)行); } });實(shí)際在瀏覽器里看到的執(zhí)行過程是這樣的init請(qǐng)求發(fā)出去后頁面先顯示“意圖識(shí)別·執(zhí)行中”1秒后變成“意圖識(shí)別·成功”同時(shí)追加“知識(shí)庫檢索·執(zhí)行中”檢索完成后變成“知識(shí)庫檢索·成功”追加“答案生成·執(zhí)行中”最后收到一個(gè)沒有next字段的事件流程結(jié)束。整個(gè)過程是連續(xù)滾動(dòng)的。這個(gè)體驗(yàn)比一個(gè)轉(zhuǎn)圈loading強(qiáng)太多了。用戶能清楚感知到Agent正在一步步處理問題而不是卡住了。6. 常見問題速查與踩坑記錄6.1 節(jié)點(diǎn)異常與失敗重試怎么處理第一版引擎設(shè)計(jì)里有個(gè)問題節(jié)點(diǎn)拋出異常就直接把整個(gè)工作流終止了。這在生產(chǎn)環(huán)境不行LLM調(diào)用經(jīng)常因?yàn)榫W(wǎng)絡(luò)抖動(dòng)或者Token限制失敗一次失敗就終止整條流程用戶體驗(yàn)很糟糕。我后來做了三層兜底。第一層節(jié)點(diǎn)內(nèi)部自己捕獲短期異常像LLM超時(shí)這種內(nèi)部重試一次再?zèng)Q定返回什么狀態(tài)。第二層路由表提供FAILED和TIMEOUT的流轉(zhuǎn)配置失敗時(shí)可以跳轉(zhuǎn)到專門的重試節(jié)點(diǎn)而不是直接終結(jié)。第三層引擎捕獲未處理異常后廣播failed事件前端收到事件后可以提示用戶重試同時(shí)記錄traceId供排查。注意一個(gè)比較容易踩的坑節(jié)點(diǎn)執(zhí)行失敗后上下文里可能已經(jīng)寫入了一部分臟數(shù)據(jù)。重試之前必須考慮這些數(shù)據(jù)的覆蓋問題。我一般建議變量寫入采用“覆蓋式”而不是“追加式”重試節(jié)點(diǎn)開始前把上一輪的中間變量清掉防止舊數(shù)據(jù)污染新結(jié)果。6.2 并發(fā)場(chǎng)景下的上下文隔離因?yàn)楣ぷ髁魇钱惒綀?zhí)行的同一個(gè)引擎實(shí)例會(huì)同時(shí)運(yùn)行多個(gè)Agent請(qǐng)求。這時(shí)候最怕的就是節(jié)點(diǎn)里不小心把數(shù)據(jù)寫進(jìn)了共享內(nèi)存。每個(gè)請(qǐng)求的NodeContext是按traceId隔離的節(jié)點(diǎn)操作variables里的變量時(shí)永遠(yuǎn)只操作自己上下文的數(shù)據(jù)這是最基本的原則。但還有一個(gè)隱性問題線程池里的線程被不同請(qǐng)求復(fù)用如果節(jié)點(diǎn)實(shí)現(xiàn)里用了ThreadLocal保存上下文一個(gè)請(qǐng)求結(jié)束后ThreadLocal沒有清理下一個(gè)請(qǐng)求復(fù)用這個(gè)線程時(shí)就會(huì)讀到上一個(gè)請(qǐng)求的殘留數(shù)據(jù)。我踩過這個(gè)坑排查了很久最后強(qiáng)制規(guī)定節(jié)點(diǎn)內(nèi)禁止使用ThreadLocal必須通過NodeContext傳遞數(shù)據(jù)。如果你的節(jié)點(diǎn)實(shí)在繞不開ThreadLocal請(qǐng)?jiān)趂inally塊里徹底remove。同時(shí)建議給線程池設(shè)置一個(gè)明確的拒絕策略。Agent并發(fā)上來之后線程池排滿任務(wù)新請(qǐng)求不能無限等待。我用的是AbortPolicy配合前端提示“當(dāng)前服務(wù)繁忙”比把請(qǐng)求都堆在內(nèi)存里等超時(shí)要好。6.3 我用這個(gè)方案后的一些體會(huì)最后說點(diǎn)個(gè)人的體會(huì)算不上結(jié)論更多是經(jīng)驗(yàn)。狀態(tài)輪轉(zhuǎn)這套東西最大的收益不是代碼變短了而是思考方式變了。以前設(shè)計(jì)Agent流程我腦子里是一段線性代碼的執(zhí)行順序現(xiàn)在設(shè)計(jì)Agent流程我腦子里就是一張有向圖——節(jié)點(diǎn)是圖的頂點(diǎn)路由是圖的邊所有流程控制都體現(xiàn)在路由關(guān)系上。這個(gè)轉(zhuǎn)變讓流程變得可以觀測(cè)。引擎里的每一個(gè)事件都記錄了節(jié)點(diǎn)ID、狀態(tài)、時(shí)間戳我可以輕松把這套事件流接到日志系統(tǒng)或者監(jiān)控面板上。用戶問“為什么我這個(gè)問題走了兜底”我查一下traceId對(duì)應(yīng)的事件流立刻就能復(fù)現(xiàn)完整路徑。這在if-else時(shí)代是做不到的。如果有朋友也想在項(xiàng)目里落地這套方案我的建議很簡單不要一開始就追求大而全先把節(jié)點(diǎn)抽象、路由表、異步事件這三樣做出來跑通一個(gè)最簡單的三節(jié)點(diǎn)流程然后逐步往里面加節(jié)點(diǎn)、加狀態(tài)、加監(jiān)聽器。等你真的把Agent流程跑起來再回頭看那些糾纏在一起的if-else大概率會(huì)和我一樣再也不想回去了。 狀態(tài)輪轉(zhuǎn)替代if-else：構(gòu)建Agent流程引擎及SSE流式輸出實(shí)戰(zhàn)
好久沒寫工程向的分享了。前面一個(gè)項(xiàng)目里我負(fù)責(zé)做一個(gè)智能客服Agent一開始圖省事所有流程控制全用if-else堆如果意圖不明就追問如果知識(shí)庫沒命中就兜底如果LLM調(diào)用失敗就重試……兩周之后代碼已經(jīng)沒法看了加一個(gè)節(jié)點(diǎn)要?jiǎng)尤奶幍胤綔y(cè)試用例寫了一堆還是漏。后來我花了一個(gè)周末把流程控制層整個(gè)重寫成基于狀態(tài)輪轉(zhuǎn)的流程引擎——Java實(shí)現(xiàn)節(jié)點(diǎn)抽象狀態(tài)顯式定義流轉(zhuǎn)用路由表驅(qū)動(dòng)執(zhí)行過程用SSE流式輸出到前端。核心代碼大概不到300行if-else基本被消滅了。這篇文章就圍繞這個(gè)設(shè)計(jì)展開講講節(jié)點(diǎn)狀態(tài)輪轉(zhuǎn)和流式輸出到底怎么落地順便把踩過的坑一并整理出來適合正在做Agent開發(fā)、AI應(yīng)用編排、或者被業(yè)務(wù)分支邏輯折磨的Java工程師參考。1. 為什么Agent工作流不能靠if-else硬寫1.1 當(dāng)Agent流程開始復(fù)雜if-else的賬就算不清了Agent的工作流和普通后端接口最大的不同在于它的執(zhí)行路徑幾乎不受你控制。一個(gè)典型Agent要經(jīng)歷意圖識(shí)別、槽位補(bǔ)齊、知識(shí)庫檢索、工具調(diào)用、上下文記憶更新、答案生成這幾個(gè)階段而每個(gè)階段都可能產(chǎn)生分支意圖置信度低要追問信息不足要澄清檢索結(jié)果為空要走兜底工具執(zhí)行超時(shí)要重試。把這些分支全用if-else寫本質(zhì)上就是在用“順序代碼”去描述一張“有向圖”。問題在前端接口還能忍一旦節(jié)點(diǎn)數(shù)超過五個(gè)代碼就會(huì)變成這個(gè)樣子外層if判斷結(jié)果類型內(nèi)層if判斷狀態(tài)碼再內(nèi)層if判斷是否需要調(diào)用另一個(gè)方法。每新增一個(gè)節(jié)點(diǎn)你要找到所有上游節(jié)點(diǎn)的出口邏輯一處處補(bǔ)判斷。更難受的是調(diào)試——線上用戶觸發(fā)了某條冷門分支你根本不知道當(dāng)時(shí)走到了哪一步因?yàn)闆]有狀態(tài)記錄。我見過不少團(tuán)隊(duì)在Agent項(xiàng)目里維護(hù)一段超過500行的switch-case每加一個(gè)工具調(diào)用就往里塞一個(gè)case分支。這本質(zhì)上和if-else沒有區(qū)別只是換了個(gè)寫法。流程控制一旦退化成這種形式它就不再是“可編排”的了而是“寫死”的。1.2 狀態(tài)機(jī)思想從路牌到流程圖如果退一步看問題Agent的執(zhí)行過程本質(zhì)上就是一個(gè)狀態(tài)機(jī)每個(gè)節(jié)點(diǎn)是一個(gè)狀態(tài)節(jié)點(diǎn)的執(zhí)行結(jié)果決定下一個(gè)狀態(tài)節(jié)點(diǎn)在流轉(zhuǎn)中會(huì)產(chǎn)生輸出事件。用狀態(tài)機(jī)的思路去建模比用代碼堆邏輯要自然得多。我打個(gè)比方你開車去一個(gè)陌生的地方靠的不是在每個(gè)路口背誦“如果看到A就左轉(zhuǎn)看到B就右轉(zhuǎn)否則直行”這種if-else口訣而是一張路線圖每個(gè)路口告訴你當(dāng)前在哪個(gè)位置根據(jù)當(dāng)前位置決定下一段怎么走。狀態(tài)輪轉(zhuǎn)就是這張路線圖節(jié)點(diǎn)就是路口的指示牌執(zhí)行結(jié)果就是你要做的選擇。流程引擎的核心價(jià)值在這里體現(xiàn)出來了它把“業(yè)務(wù)邏輯”和“流程控制”徹底拆開。業(yè)務(wù)邏輯寫在節(jié)點(diǎn)內(nèi)部流程控制交給引擎的路由機(jī)制。以后哪怕要調(diào)整整個(gè)Agent的執(zhí)行順序你也不需要改業(yè)務(wù)代碼在路由表里重新編排節(jié)點(diǎn)ID就可以。2. 流程引擎的整體設(shè)計(jì)先抽象再實(shí)現(xiàn)2.1 核心抽象Node節(jié)點(diǎn)和Workflow工作流設(shè)計(jì)一個(gè)流程引擎最重要的不是先寫代碼而是定義好抽象。我這邊核心抽象只有兩個(gè)Node節(jié)點(diǎn)和Workflow工作流。Node是流程中的最小執(zhí)行單元。它負(fù)責(zé)做三件事聲明自己的ID、執(zhí)行具體的業(yè)務(wù)邏輯、明確執(zhí)行完成之后下一步去哪個(gè)節(jié)點(diǎn)。業(yè)務(wù)邏輯五花八門沒關(guān)系我把它統(tǒng)一收斂到一個(gè)抽象方法里返回執(zhí)行狀態(tài)。狀態(tài)決定了路由方向。Workflow則是節(jié)點(diǎn)的集合和入口配置。它知道整個(gè)流程從哪個(gè)節(jié)點(diǎn)開始手里有一張注冊(cè)表存著所有節(jié)點(diǎn)的ID到實(shí)例的映射。引擎執(zhí)行的時(shí)候只需要拿到入口節(jié)點(diǎn)ID然后不停查詢當(dāng)前節(jié)點(diǎn)、執(zhí)行當(dāng)前節(jié)點(diǎn)、根據(jù)返回狀態(tài)找下一個(gè)節(jié)點(diǎn)循環(huán)往復(fù)直到走到終止?fàn)顟B(tài)。這里有一個(gè)很重要的設(shè)計(jì)取舍節(jié)點(diǎn)之間互相不直接依賴。A節(jié)點(diǎn)執(zhí)行完后它不需要知道B節(jié)點(diǎn)是什么只需要在路由表里聲明“狀態(tài)SUCCESS時(shí)去node_detail_query這個(gè)ID”。這種解耦帶來的直接好處是節(jié)點(diǎn)可以獨(dú)立測(cè)試可以隨意替換也可以在不同工作流里復(fù)用。2.2 狀態(tài)輪轉(zhuǎn)如何替代if-else傳統(tǒng)寫法里控制權(quán)在“調(diào)用方”手里。一個(gè)方法調(diào)用另一個(gè)方法根據(jù)返回值決定下一步層層嵌套。狀態(tài)輪轉(zhuǎn)的設(shè)計(jì)里控制權(quán)被上交給“引擎”手里。每個(gè)節(jié)點(diǎn)執(zhí)行完成后會(huì)返回一個(gè)NodeStatus引擎拿到這個(gè)狀態(tài)去查當(dāng)前節(jié)點(diǎn)的路由表拿到目標(biāo)節(jié)點(diǎn)ID接著執(zhí)行下一個(gè)節(jié)點(diǎn)。也就是說分支邏輯從“代碼的調(diào)用關(guān)系”變成了“數(shù)據(jù)表里的映射關(guān)系”。if-else被徹底從代碼里剝離你看到的是清晰的路由表SUCCESS → 下一步做什么FAILED → 是重試還是走兜底NEED_INPUT → 是否要跳轉(zhuǎn)到追問節(jié)點(diǎn)有人會(huì)說這不就是把if-else搬了個(gè)位置嗎其實(shí)區(qū)別很大。if-else是硬編碼流程一旦寫錯(cuò)或者要調(diào)整必須改代碼重新發(fā)布路由表是結(jié)構(gòu)化數(shù)據(jù)可以配置化、可視化、動(dòng)態(tài)修改。而且對(duì)于流程引擎來說路由表天然適合做遍歷檢查——你可以啟動(dòng)時(shí)掃描所有節(jié)點(diǎn)的路由目標(biāo)檢測(cè)有沒有指向不存在的節(jié)點(diǎn)ID有沒有死循環(huán)。這些能力是普通if-else寫法根本給不了的。2.3 為什么不用Activiti、Flowable這些現(xiàn)成引擎肯定有人要問Java生態(tài)里不是有Activiti、Flowable、Camunda這些成熟的工作流引擎嗎為什么還要自己寫我自己評(píng)估過這些引擎強(qiáng)在“人參與審批”的場(chǎng)景任務(wù)分配、會(huì)簽、或簽、超時(shí)提醒。但Agent場(chǎng)景完全是另一回事節(jié)點(diǎn)執(zhí)行的是LLM推理調(diào)用、工具API調(diào)用、向量檢索不是人工填表審批。引入Activiti這種重型BPM框架光是把流程定義文件和數(shù)據(jù)模型接進(jìn)來就要花不少功夫再加上它們本身那套部署方式、歷史表結(jié)構(gòu)對(duì)一個(gè)小團(tuán)隊(duì)來說負(fù)擔(dān)很重。更關(guān)鍵的一點(diǎn)是流式輸出。Agent執(zhí)行過程需要實(shí)時(shí)把中間狀態(tài)推到前端——用戶能看著“正在理解意圖→正在檢索知識(shí)庫→正在生成回答”這是現(xiàn)代Agent應(yīng)用的基本體驗(yàn)要求。Activiti這類引擎的監(jiān)聽機(jī)制是為系統(tǒng)內(nèi)部事件設(shè)計(jì)的要接到SSE推送還得自己搭一層橋梁。與其繞一圈適配不如直接自己寫一個(gè)輕量引擎保留最核心的節(jié)點(diǎn)路由能力把事件推送機(jī)制做成一等公民。實(shí)測(cè)下來核心引擎不到300行后面所有業(yè)務(wù)節(jié)點(diǎn)都在復(fù)用這一套骨架。3. 核心實(shí)現(xiàn)節(jié)點(diǎn)狀態(tài)輪轉(zhuǎn)引擎代碼拆解3.1 狀態(tài)枚舉與節(jié)點(diǎn)抽象類先看最基本的定義狀態(tài)枚舉我一開始只設(shè)計(jì)了5個(gè)狀態(tài)后來加了TIMEOUT和TERMINATED別嫌多實(shí)際寫業(yè)務(wù)的時(shí)候你會(huì)發(fā)現(xiàn)每個(gè)狀態(tài)都有對(duì)應(yīng)場(chǎng)景。public enum NodeStatus { PENDING(待執(zhí)行), RUNNING(執(zhí)行中), SUCCESS(成功), FAILED(失敗), SKIPPED(跳過), TIMEOUT(超時(shí)), TERMINATED(終止); private final String desc; NodeStatus(String desc) { this.desc desc; } public String getDesc() { return desc; } }然后是節(jié)點(diǎn)的抽象基類。核心就兩個(gè)東西一個(gè)execute方法用于執(zhí)行業(yè)務(wù)并返回狀態(tài)一個(gè)路由表用于指明不同狀態(tài)分別去哪個(gè)節(jié)點(diǎn)。public abstract class BaseNodeT { private final String id; private final String name; private final MapNodeStatus, String routeTable new EnumMap(NodeStatus.class); public BaseNode(String id, String name) { this.id id; this.name name; } public String getId() { return id; } public String getName() { return name; } public abstract NodeStatus execute(NodeContextT context); protected void on(NodeStatus status, String targetNodeId) { routeTable.put(status, targetNodeId); } public String route(NodeStatus status) { return routeTable.get(status); } }設(shè)計(jì)要點(diǎn)在route方法執(zhí)行完一個(gè)節(jié)點(diǎn)引擎把返回狀態(tài)交給路由表查詢找到的就是下一站節(jié)點(diǎn)ID。如果某個(gè)狀態(tài)沒有配置路由說明這個(gè)狀態(tài)就是終態(tài)引擎會(huì)終止循環(huán)。這里我建議一個(gè)實(shí)際優(yōu)化在做驗(yàn)證時(shí)遍歷注冊(cè)表里的所有節(jié)點(diǎn)檢查每個(gè)節(jié)點(diǎn)的路由目標(biāo)是否存在于注冊(cè)表中。這個(gè)檢查我寫在啟動(dòng)階段一旦發(fā)現(xiàn)懸空引用直接報(bào)錯(cuò)比運(yùn)行時(shí)走到一半才發(fā)現(xiàn)要好得多。3.2 執(zhí)行上下文讓所有節(jié)點(diǎn)共享數(shù)據(jù)和狀態(tài)節(jié)點(diǎn)之間光靠參數(shù)傳遞是不夠的Agent執(zhí)行過程中會(huì)產(chǎn)生大量上下文數(shù)據(jù)比如用戶的原始問題、意圖識(shí)別的結(jié)果、知識(shí)庫檢索出來的片段、歷史對(duì)話記錄這些數(shù)據(jù)需要在不同節(jié)點(diǎn)之間流轉(zhuǎn)。為此我設(shè)計(jì)了一個(gè)NodeContextpublic class NodeContextT { private final String traceId; private final T payload; private final MapString, Object variables new ConcurrentHashMap(); private String currentNodeId; private NodeStatus currentNodeStatus; public NodeContext(String traceId, T payload) { this.traceId traceId; this.payload payload; } public void setVariable(String key, Object value) { variables.put(key, value); } SuppressWarnings(unchecked) public V V getVariable(String key) { return (V) variables.get(key); } public String getTraceId() { return traceId; } public T getPayload() { return payload; } public String getCurrentNodeId() { return currentNodeId; } public void markNode(String nodeId, NodeStatus status) { this.currentNodeId nodeId; this.currentNodeStatus status; } }traceId是每次Agent請(qǐng)求的唯一標(biāo)識(shí)貫穿整個(gè)執(zhí)行鏈路既用于日志追蹤也用于流式輸出時(shí)關(guān)聯(lián)事件。variables是節(jié)點(diǎn)間的數(shù)據(jù)交換層A節(jié)點(diǎn)把意圖識(shí)別的結(jié)果放進(jìn)去B節(jié)點(diǎn)直接取不需要方法簽名級(jí)別的耦合。實(shí)際項(xiàng)目中這個(gè)上下文還可以擴(kuò)展出上下文窗口管理、令牌消耗統(tǒng)計(jì)、重試次數(shù)記錄等能力。它本質(zhì)上是Agent的“工作記憶”記得在哪一步停過、拿過什么結(jié)果、下一步還缺什么信息。3.3 引擎主循環(huán)用路由表驅(qū)動(dòng)節(jié)點(diǎn)跳轉(zhuǎn)引擎是整個(gè)流程的中樞它負(fù)責(zé)調(diào)度節(jié)點(diǎn)、處理狀態(tài)輪轉(zhuǎn)、廣播事件。我把它設(shè)計(jì)成可復(fù)用的通用組件public class FlowEngineT { private final MapString, BaseNodeT registry new HashMap(); private final ExecutorService executorService; private final ListWorkflowListener listeners new CopyOnWriteArrayList(); public FlowEngine(ExecutorService executorService) { this.executorService executorService; } public void registerNode(BaseNodeT node) { registry.put(node.getId(), node); } public void addListener(WorkflowListener listener) { listeners.add(listener); } public void start(String entryNodeId, T payload, String traceId) { executorService.submit(() - execute(entryNodeId, payload, traceId)); } private void execute(String entryNodeId, T payload, String traceId) { NodeContextT context new NodeContext(traceId, payload); String currentNodeId entryNodeId; try { while (currentNodeId ! null) { BaseNodeT node registry.get(currentNodeId); if (node null) { throw new IllegalStateException(節(jié)點(diǎn)不存在: currentNodeId); } context.markNode(currentNodeId, NodeStatus.RUNNING); emitEvent(NodeEvent.started(context, System.currentTimeMillis())); NodeStatus status node.execute(context); context.markNode(currentNodeId, status); emitEvent(NodeEvent.finished(context, System.currentTimeMillis())); currentNodeId node.route(status); } emitEvent(NodeEvent.completed(context, System.currentTimeMillis())); } catch (Exception e) { context.setVariable(lastError, e.getMessage()); emitEvent(NodeEvent.failed(context, e, System.currentTimeMillis())); } } private void emitEvent(NodeEvent event) { for (WorkflowListener listener : listeners) { listener.onEvent(event); } } }主循環(huán)邏輯很簡潔根據(jù)當(dāng)前節(jié)點(diǎn)ID拿到節(jié)點(diǎn)實(shí)例執(zhí)行取狀態(tài)查路由得到下一個(gè)節(jié)點(diǎn)ID。沒有判斷流程邏輯的if-else所有分支都收斂在節(jié)點(diǎn)的路由表里。這里我想強(qiáng)調(diào)一個(gè)容易忽略的點(diǎn)開始執(zhí)行用的start方法是異步的把任務(wù)丟進(jìn)線程池立即返回。為什么因?yàn)楣ぷ髁饕浜狭魇捷敵鋈绻划惒交疕TTP響應(yīng)線程會(huì)被阻塞前端的SSE流根本沒機(jī)會(huì)實(shí)時(shí)收到事件。異步化之后整個(gè)執(zhí)行變成事件驅(qū)動(dòng)的推送模型。4. 流式輸出讓Agent執(zhí)行過程實(shí)時(shí)可見4.1 全量返回為什么會(huì)讓用戶焦慮Agent場(chǎng)景和普通接口最大的體驗(yàn)差異在于普通接口用戶等一兩秒拿到結(jié)果沒問題但Agent執(zhí)行一個(gè)復(fù)雜任務(wù)可能耗時(shí)幾十秒甚至幾分鐘——比如多輪工具調(diào)用再加長文本生成。如果用戶點(diǎn)完按鈕之后頁面一直轉(zhuǎn)圈沒有任何過程反饋體驗(yàn)是災(zāi)難性的。流式輸出解決的就是這個(gè)問題。它的本質(zhì)是把Agent執(zhí)行過程中產(chǎn)生的每一個(gè)階段性事件從服務(wù)端實(shí)時(shí)推送到客戶端。注意是事件流不只是最終答案。用戶能看到“正在判斷意圖”→“正在檢索知識(shí)庫”→“正在調(diào)用外部工具”→“正在生成回答”每一條狀態(tài)變化都實(shí)時(shí)渲染到頁面上。技術(shù)選型上我用了SSEServer-Sent Events而不是WebSocket。原因有兩點(diǎn)Agent執(zhí)行狀態(tài)是服務(wù)端到客戶端的單向數(shù)據(jù)流不需要客戶端頻繁上行消息SSE基于HTTP協(xié)議不需要額外的握手協(xié)議前端的EventSource對(duì)象可以直接消費(fèi)實(shí)現(xiàn)成本低得多。4.2 基于SseEmitter的事件推送實(shí)現(xiàn)Spring Boot生態(tài)里SseEmitter是原生的SSE支持用起來很直接。我把它和流程引擎的事件監(jiān)聽器接在一起執(zhí)行過程中每產(chǎn)生一個(gè)事件就通過emitter推送給前端RestController public class AgentController { private final FlowEngineAgentRequest flowEngine; private final ExecutorService executorService Executors.newCachedThreadPool(); public AgentController(FlowEngineAgentRequest flowEngine) { this.flowEngine flowEngine; } GetMapping(value /agent/run, produces MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter run(RequestParam String question) { String traceId UUID.randomUUID().toString().substring(0, 8); SseEmitter emitter new SseEmitter(60_000L); WorkflowListener listener event - { try { MapString, Object data new HashMap(); data.put(traceId, event.getTraceId()); data.put(nodeId, event.getContext().getCurrentNodeId()); data.put(status, event.getContext().getCurrentNodeStatus().name()); data.put(type, event.getType().name()); if (event.getTargetNodeId() ! null) { data.put(next, event.getTargetNodeId()); } emitter.send(SseEmitter.event().name(agent-event).data(data)); } catch (IOException e) { emitter.completeWithError(e); } }; flowEngine.addListener(listener); emitter.onCompletion(() - flowEngine.removeListener(listener)); emitter.onTimeout(() - flowEngine.removeListener(listener)); flowEngine.start(ENTRY_NODE_ID, new AgentRequest(question), traceId); return emitter; } }這里有一個(gè)非常關(guān)鍵的設(shè)計(jì)SseEmitter推送給前端的不只是最終結(jié)果而是執(zhí)行過程中的每一次狀態(tài)輪轉(zhuǎn)。節(jié)點(diǎn)開始執(zhí)行推一個(gè)started事件節(jié)點(diǎn)結(jié)束推一個(gè)finished事件事件里帶上當(dāng)前節(jié)點(diǎn)ID和狀態(tài)。前端拿到這些事件之后就可以在頁面上逐個(gè)渲染狀態(tài)變化。用戶看到的是“流程在往前走”而不是一個(gè)死等loading。數(shù)據(jù)事件里還帶了next字段如果有這個(gè)字段說明流程還要繼續(xù)如果沒有說明這個(gè)節(jié)點(diǎn)是終態(tài)。4.3 線程池、事件監(jiān)聽器與斷連處理流式輸出本身不復(fù)雜復(fù)雜在工程細(xì)節(jié)。第一個(gè)細(xì)節(jié)是線程隔離。Agent工作流的執(zhí)行必須放在獨(dú)立的線程池里不能占著Tomcat的請(qǐng)求線程。如果直接在請(qǐng)求線程里跑工作流SseEmitter就永遠(yuǎn)來不及發(fā)送數(shù)據(jù)因?yàn)轫憫?yīng)還沒返回。我用了獨(dú)立的線程池請(qǐng)求線程只負(fù)責(zé)創(chuàng)建emitter和注冊(cè)監(jiān)聽器真正的工作流執(zhí)行在另一個(gè)線程里異步跑。第二個(gè)細(xì)節(jié)是監(jiān)聽器的生命周期管理。每個(gè)SSE連接對(duì)應(yīng)一個(gè)監(jiān)聽器實(shí)例連接關(guān)閉或者超時(shí)之后監(jiān)聽器必須從引擎里移除。否則emitter已經(jīng)關(guān)閉了工作流還繼續(xù)往里面send數(shù)據(jù)會(huì)拋IOException。我在onCompletion和onTimeout回調(diào)里都調(diào)用了removeListener確保連接斷開后不再有推送動(dòng)作。第三個(gè)細(xì)節(jié)是超時(shí)設(shè)置。SseEmitter的構(gòu)造參數(shù)是超時(shí)毫秒數(shù)我設(shè)置了60秒。如果Agent在這個(gè)時(shí)間內(nèi)還沒跑完連接會(huì)斷掉用戶需要重新發(fā)起請(qǐng)求。后續(xù)我改成了更合理的方案用定時(shí)器在超時(shí)前刷新一下連接或者把超時(shí)設(shè)成一個(gè)比較長的值配合心跳機(jī)制保持連接活性。對(duì)于生產(chǎn)環(huán)境建議單獨(dú)評(píng)估每個(gè)Agent流程的最大耗時(shí)再?zèng)Q定超時(shí)時(shí)長。5. 實(shí)操案例從0搭建一個(gè)智能客服Agent5.1 節(jié)點(diǎn)設(shè)計(jì)與路由編排理論講了一堆最終要落到一個(gè)能跑通的例子。我以智能客服Agent為例設(shè)計(jì)了5個(gè)業(yè)務(wù)節(jié)點(diǎn)intent_node意圖識(shí)別節(jié)點(diǎn)。調(diào)用LLM判斷用戶意圖返回四個(gè)狀態(tài)BUY想買產(chǎn)品、AFTER_SALE售后咨詢、OTHER閑聊、UNKNOWN無法識(shí)別。clarify_node追問節(jié)點(diǎn)。當(dāng)意圖識(shí)別返回UNKNOWN時(shí)追問用戶具體需求然后重新路由回意圖識(shí)別節(jié)點(diǎn)。knowledge_node知識(shí)庫檢索節(jié)點(diǎn)。根據(jù)意圖和用戶問題做向量檢索從知識(shí)庫中匹配相關(guān)內(nèi)容。如果檢索到的內(nèi)容相似度低于閾值返回FAILED。fallback_node兜底節(jié)點(diǎn)。知識(shí)庫檢索不到內(nèi)容時(shí)給出一個(gè)預(yù)設(shè)話術(shù)引導(dǎo)用戶轉(zhuǎn)接人工。answer_node答案生成節(jié)點(diǎn)。把檢索結(jié)果和歷史上下文拼進(jìn)Prompt調(diào)用LLM生成最終回答。節(jié)點(diǎn)定義好了路由表就是整個(gè)Agent的“劇本”。我專門用一個(gè)配置類把路由關(guān)系集中管理Configuration public class AgentWorkflowConfig { public static final String ENTRY_NODE intent_node; Bean public FlowEngineAgentRequest agentFlowEngine() { FlowEngineAgentRequest engine new FlowEngine(Executors.newFixedThreadPool(8)); BaseNodeAgentRequest intentNode new BaseNodeAgentRequest(intent_node, 意圖識(shí)別) { Override public NodeStatus execute(NodeContextAgentRequest context) { IntentResult intent llmService.recognizeIntent(context.getPayload().getQuestion()); context.setVariable(intent, intent); if (intent.confidence() 0.6) { return NodeStatus.NEED_INPUT; } return NodeStatus.SUCCESS; } }; intentNode.on(NodeStatus.SUCCESS, knowledge_node); intentNode.on(NodeStatus.NEED_INPUT, clarify_node); intentNode.on(NodeStatus.FAILED, fallback_node); BaseNodeAgentRequest clarifyNode new BaseNodeAgentRequest(clarify_node, 追問) { Override public NodeStatus execute(NodeContextAgentRequest context) { return confirmAnswer(context) ? NodeStatus.SUCCESS : NodeStatus.FAILED; } }; clarifyNode.on(NodeStatus.SUCCESS, intent_node); clarifyNode.on(NodeStatus.FAILED, fallback_node); // knowledge_node、fallback_node、answer_node 類似按流程編排 engine.registerNode(intentNode); engine.registerNode(clarifyNode); engine.registerNode(knowledgeNode); engine.registerNode(fallbackNode); engine.registerNode(answerNode); return engine; } }這個(gè)設(shè)計(jì)的好處你寫幾個(gè)節(jié)點(diǎn)就能體會(huì)出來。比如我想在答案生成之前加一個(gè)“敏感詞檢查”節(jié)點(diǎn)只需要新建一個(gè)節(jié)點(diǎn)類然后把a(bǔ)nswer_node的上游路由改一下其他節(jié)點(diǎn)的代碼一個(gè)字都不用動(dòng)。5.2 注冊(cè)節(jié)點(diǎn)并啟動(dòng)工作流節(jié)點(diǎn)配置完成后啟動(dòng)一次Agent工作流只需要一行代碼flowEngine.start(AgentWorkflowConfig.ENTRY_NODE, new AgentRequest(question), traceId);引擎會(huì)自動(dòng)從intent_node開始一路輪轉(zhuǎn)下去。每個(gè)節(jié)點(diǎn)的執(zhí)行結(jié)果都會(huì)觸發(fā)狀態(tài)事件這些事件經(jīng)過監(jiān)聽器、SseEmitter最終變成前端頁面上的實(shí)時(shí)狀態(tài)更新。我在實(shí)際部署中還加了一個(gè)內(nèi)容審核節(jié)點(diǎn)放在最后專門檢查LLM生成結(jié)果是否合規(guī)合規(guī)度低就重新生成一次最多重試兩次。這些控制都通過路由表表達(dá)answer_node的下一步根據(jù)審核結(jié)果分別指向結(jié)束節(jié)點(diǎn)或者重試節(jié)點(diǎn)。流程的復(fù)雜度上來了但代碼結(jié)構(gòu)一直是同一套。5.3 前端收到的事件流長什么樣配套的前端邏輯也很簡單。用瀏覽器原生的EventSource監(jiān)聽事件每次收到消息就更新頁面上的步驟狀態(tài)const eventSource new EventSource(/agent/run?question encodeURIComponent(question)); eventSource.addEventListener(agent-event, (event) { const data JSON.parse(event.data); renderNodeStatus(data.nodeId, data.status); if (data.next) { appendNode(data.next, 等待執(zhí)行); } });實(shí)際在瀏覽器里看到的執(zhí)行過程是這樣的init請(qǐng)求發(fā)出去后頁面先顯示“意圖識(shí)別·執(zhí)行中”1秒后變成“意圖識(shí)別·成功”同時(shí)追加“知識(shí)庫檢索·執(zhí)行中”檢索完成后變成“知識(shí)庫檢索·成功”追加“答案生成·執(zhí)行中”最后收到一個(gè)沒有next字段的事件流程結(jié)束。整個(gè)過程是連續(xù)滾動(dòng)的。這個(gè)體驗(yàn)比一個(gè)轉(zhuǎn)圈loading強(qiáng)太多了。用戶能清楚感知到Agent正在一步步處理問題而不是卡住了。6. 常見問題速查與踩坑記錄6.1 節(jié)點(diǎn)異常與失敗重試怎么處理第一版引擎設(shè)計(jì)里有個(gè)問題節(jié)點(diǎn)拋出異常就直接把整個(gè)工作流終止了。這在生產(chǎn)環(huán)境不行LLM調(diào)用經(jīng)常因?yàn)榫W(wǎng)絡(luò)抖動(dòng)或者Token限制失敗一次失敗就終止整條流程用戶體驗(yàn)很糟糕。我后來做了三層兜底。第一層節(jié)點(diǎn)內(nèi)部自己捕獲短期異常像LLM超時(shí)這種內(nèi)部重試一次再?zèng)Q定返回什么狀態(tài)。第二層路由表提供FAILED和TIMEOUT的流轉(zhuǎn)配置失敗時(shí)可以跳轉(zhuǎn)到專門的重試節(jié)點(diǎn)而不是直接終結(jié)。第三層引擎捕獲未處理異常后廣播failed事件前端收到事件后可以提示用戶重試同時(shí)記錄traceId供排查。注意一個(gè)比較容易踩的坑節(jié)點(diǎn)執(zhí)行失敗后上下文里可能已經(jīng)寫入了一部分臟數(shù)據(jù)。重試之前必須考慮這些數(shù)據(jù)的覆蓋問題。我一般建議變量寫入采用“覆蓋式”而不是“追加式”重試節(jié)點(diǎn)開始前把上一輪的中間變量清掉防止舊數(shù)據(jù)污染新結(jié)果。6.2 并發(fā)場(chǎng)景下的上下文隔離因?yàn)楣ぷ髁魇钱惒綀?zhí)行的同一個(gè)引擎實(shí)例會(huì)同時(shí)運(yùn)行多個(gè)Agent請(qǐng)求。這時(shí)候最怕的就是節(jié)點(diǎn)里不小心把數(shù)據(jù)寫進(jìn)了共享內(nèi)存。每個(gè)請(qǐng)求的NodeContext是按traceId隔離的節(jié)點(diǎn)操作variables里的變量時(shí)永遠(yuǎn)只操作自己上下文的數(shù)據(jù)這是最基本的原則。但還有一個(gè)隱性問題線程池里的線程被不同請(qǐng)求復(fù)用如果節(jié)點(diǎn)實(shí)現(xiàn)里用了ThreadLocal保存上下文一個(gè)請(qǐng)求結(jié)束后ThreadLocal沒有清理下一個(gè)請(qǐng)求復(fù)用這個(gè)線程時(shí)就會(huì)讀到上一個(gè)請(qǐng)求的殘留數(shù)據(jù)。我踩過這個(gè)坑排查了很久最后強(qiáng)制規(guī)定節(jié)點(diǎn)內(nèi)禁止使用ThreadLocal必須通過NodeContext傳遞數(shù)據(jù)。如果你的節(jié)點(diǎn)實(shí)在繞不開ThreadLocal請(qǐng)?jiān)趂inally塊里徹底remove。同時(shí)建議給線程池設(shè)置一個(gè)明確的拒絕策略。Agent并發(fā)上來之后線程池排滿任務(wù)新請(qǐng)求不能無限等待。我用的是AbortPolicy配合前端提示“當(dāng)前服務(wù)繁忙”比把請(qǐng)求都堆在內(nèi)存里等超時(shí)要好。6.3 我用這個(gè)方案后的一些體會(huì)最后說點(diǎn)個(gè)人的體會(huì)算不上結(jié)論更多是經(jīng)驗(yàn)。狀態(tài)輪轉(zhuǎn)這套東西最大的收益不是代碼變短了而是思考方式變了。以前設(shè)計(jì)Agent流程我腦子里是一段線性代碼的執(zhí)行順序現(xiàn)在設(shè)計(jì)Agent流程我腦子里就是一張有向圖——節(jié)點(diǎn)是圖的頂點(diǎn)路由是圖的邊所有流程控制都體現(xiàn)在路由關(guān)系上。這個(gè)轉(zhuǎn)變讓流程變得可以觀測(cè)。引擎里的每一個(gè)事件都記錄了節(jié)點(diǎn)ID、狀態(tài)、時(shí)間戳我可以輕松把這套事件流接到日志系統(tǒng)或者監(jiān)控面板上。用戶問“為什么我這個(gè)問題走了兜底”我查一下traceId對(duì)應(yīng)的事件流立刻就能復(fù)現(xiàn)完整路徑。這在if-else時(shí)代是做不到的。如果有朋友也想在項(xiàng)目里落地這套方案我的建議很簡單不要一開始就追求大而全先把節(jié)點(diǎn)抽象、路由表、異步事件這三樣做出來跑通一個(gè)最簡單的三節(jié)點(diǎn)流程然后逐步往里面加節(jié)點(diǎn)、加狀態(tài)、加監(jiān)聽器。等你真的把Agent流程跑起來再回頭看那些糾纏在一起的if-else大概率會(huì)和我一樣再也不想回去了。