前一篇聊到一个我很认同的方向复杂 Agent 不能只靠聊天窗口展示状态。这篇不继续讲概念直接做一个能跑的最小版本。目标很简单Spring Boot后端 →持续产生Agent事件 →SSE推给浏览器 →React Flow实时更新节点 →右侧显示Artifact、预算和待审批事项不是为了做一个漂亮 Demo而是验证一个设计Agent UI 应该是运行状态的投影而不是把日志重新排版。最终要看到什么假设 Agent 正在执行“升级 Java 25”任务。前端节点Assess DONE Plan DONE Modify RUNNING Test PENDING Review PENDING Deploy BLOCKED事件进来以后节点颜色和标签实时更新。右侧同时显示Changed files: 6 Tests: 128 passed / 3 failed Model calls: 14 Cost: $2.31 Approval: deploy-canary用户不需要翻聊天记录。技术栈前端使用当前 React Flow 包npminstallxyflow/react后端用 Spring WebFlux 的ServerSentEvent。为什么这里选 SSE不选 WebSocket因为 Run Canvas 的主要实时方向是Server → Browser用户的审批、取消、重试仍然可以走普通 HTTP。SSE 足够简单。如果后面需要高频双向协作再换 WebSocket。先定义事件不要先写 UI这是最重要的一步。publicenumAgentRunEventType{RUN_CREATED,PLAN_CREATED,STEP_STARTED,STEP_COMPLETED,STEP_FAILED,ARTIFACT_CREATED,APPROVAL_REQUESTED,APPROVAL_RESOLVED,BUDGET_UPDATED,RUN_COMPLETED,RUN_FAILED}事件结构publicrecordAgentRunEvent(StringeventId,StringrunId,longsequence,AgentRunEventTypetype,StringstepId,JsonNodepayload,InstantoccurredAt){}sequence很重要。不要只用时间戳排序。多个 Worker 的时钟不一定严格一致。事件应该先持久化再推送错误设计Agent执行 →直接SSE推前端用户刷新页面以后前面的状态全没了。正确设计Agent执行 →Event Store →Projection →SSE页面第一次打开GET /runs/{runId}拿当前 Projection。然后GET /runs/{runId}/events订阅增量。Projection 模型publicrecordAgentRunView(StringrunId,Stringtitle,RunStatusstatus,longlastSequence,ListStepViewsteps,ListArtifactViewartifacts,BudgetViewbudget,ListApprovalViewapprovals){}SteppublicrecordStepView(StringstepId,Stringname,StepStatusstatus,ListStringdependencies,Stringsummary,InstantstartedAt,InstantcompletedAt){}Event Store 最小表createtableagent_run_event(event_idvarchar(128)primarykey,run_idvarchar(128)notnull,sequencebigintnotnull,event_typevarchar(64)notnull,step_idvarchar(128),payload jsonbnotnull,occurred_at timestamptznotnull,unique(run_id,sequence));createindexidx_run_event_run_seqonagent_run_event(run_id,sequence);生产系统中Sequence 应由数据库、Run Actor 或专门序列服务生成不能用System.currentTimeMillis()凑。Spring Boot 的 SSE EndpointRestControllerRequestMapping(/api/runs)publicclassAgentRunStreamController{privatefinalAgentRunEventStreamstream;publicAgentRunStreamController(AgentRunEventStreamstream){this.streamstream;}GetMapping(value/{runId}/events,producesMediaType.TEXT_EVENT_STREAM_VALUE)publicFluxServerSentEventAgentRunEventevents(PathVariableStringrunId,RequestHeader(valueLast-Event-ID,requiredfalse)StringlastEventId){returnstream.subscribe(runId,lastEventId).map(event-ServerSentEvent.builder(event).id(event.eventId()).event(event.type().name()).build());}}不要只用一个内存Sinks.ManyDemo 最容易这么写Sinks.ManyAgentRunEventsinkSinks.many().multicast().onBackpressureBuffer();单机演示没问题。但生产会遇到用户连到Pod A Agent事件产生在Pod B所以事件事实最好先进入数据库/Kafka/Redis Stream再由所有 Web 节点消费。一个简单的 Stream ServiceServicepublicclassAgentRunEventStream{privatefinalEventRepositoryrepository;privatefinalLiveEventBusliveEventBus;publicFluxAgentRunEventsubscribe(StringrunId,StringlastEventId){MonoLongstartSequencerepository.resolveSequence(runId,lastEventId).defaultIfEmpty(0L);returnstartSequence.flatMapMany(seq-{FluxAgentRunEventhistoryrepository.findAfter(runId,seq);FluxAgentRunEventliveliveEventBus.events(runId);returnFlux.concat(history,live).distinct(AgentRunEvent::eventId);});}}重点是先补历史 再接实时还要用 Event ID 去重。SSE 断线恢复浏览器 EventSource 会自动重连。服务端要让客户端能够告诉你我最后收到哪个EventSSE 原生有Last-Event-ID语义。所以每个事件都设置.id(event.eventId())重连后只补缺失事件。还需要 Heartbeat如果几分钟没有 Agent 事件代理或网关可能断开长连接。可以每 15 秒发一个 heartbeatFluxServerSentEventAgentRunEventheartbeatFlux.interval(Duration.ofSeconds(15)).map(i-ServerSentEvent.AgentRunEventbuilder().comment(heartbeat).build());与事件流合并。前端安装 React Flownpminstallxyflow/react样式import xyflow/react/dist/style.css;如果使用 Tailwind 4 和最新 React Flow UI样式组织方式可以按当前文档放到全局 CSS。前端事件类型exporttypeRunEvent{eventId:string;runId:string;sequence:number;type:string;stepId?:string;payload:Recordstring,unknown;occurredAt:string;};Projection 不要完全依赖后端每次传整棵图后端可以传事件{type:STEP_COMPLETED,stepId:inspect-build,sequence:18}前端 Reducer 应用增量。functionreduceRun(state:RunView,event:RunEvent):RunView{switch(event.type){caseSTEP_STARTED:returnupdateStep(state,event.stepId!,{status:RUNNING});caseSTEP_COMPLETED:returnupdateStep(state,event.stepId!,{status:DONE});caseSTEP_FAILED:returnupdateStep(state,event.stepId!,{status:FAILED});default:returnstate;}}React Flow 节点转换importtype{Node,Edge}fromxyflow/react;exportfunctiontoFlow(run:RunView):{nodes:Node[];edges:Edge[]}{constnodes:Node[]run.steps.map((step,index)({id:step.stepId,position:{x:index*240,y:120,},data:{label:step.name,status:step.status,summary:step.summary,},type:agentStep,}));constedges:Edge[][];for(conststepofrun.steps){for(constdepofstep.dependencies){edges.push({id:${dep}-${step.stepId},source:dep,target:step.stepId,});}}return{nodes,edges};}自定义节点function AgentStepNode({ data }: NodeProps) { return ( div className{step step-${data.status}} div classNamestep-title {data.label} /div div classNamestep-status {data.status} /div {data.summary ( div classNamestep-summary {data.summary} /div )} /div ); }不要只靠颜色表达状态。同时显示文字PENDING RUNNING DONE FAILED BLOCKED WAITING_APPROVALEventSource Hook浏览器原生 EventSource 不能方便地设置任意 Header。如果认证依赖 Cookie可以直接用useEffect((){constsourcenewEventSource(/api/runs/${runId}/events,{withCredentials:true});source.onmessage(message){consteventJSON.parse(message.data);dispatch(event);};source.onerror(){console.warn(run stream disconnected);};return()source.close();},[runId]);如果你使用 Bearer Token通常需要 fetch streaming、同源 Session 或专门的 SSE Client而不是把 Token 塞进 URL。页面第一次加载怎么做不要等 SSE 从头重放所有事件。constinitialawaitfetch(/api/runs/${runId}).then(rr.json());setRun(initial);这个接口返回当前 Snapshot。然后 SSE 只接lastSequence之后的事件Snapshot Event 是比较实用的组合只有 Event一个运行三天的Agent 打开页面要重放几十万条事件只有 Snapshot不知道历史发生了什么组合Snapshot负责快速打开 Event负责历史与增量审批不要通过 SSE 回传SSE 是服务端到客户端。用户审批走普通接口POST /api/runs/{runId}/approvals/{approvalId}Body{decision:APPROVE,expectedRunVersion:29}服务端检查用户权限 Approval状态 Run Version Token 过期时间成功后产生APPROVAL_RESOLVEDSSE 再把新状态推回来。右侧 Artifact 面板不要把 Artifact 正文全部塞进 Node。Node 只显示6 files changed点击后右侧打开pom.xml Dockerfile .github/workflows/build.ymlArtifact 后端模型publicrecordArtifactView(StringartifactId,Stringtype,Stringtitle,StringcontentHash,StringpreviewUrl,InstantcreatedAt){}Budget 也应该是一等公民页面顶部Tokens 184K / 300K Cost $2.31 / $5.00 Model calls 14 / 25 Tool calls 28 / 50事件{type:BUDGET_UPDATED,payload:{cost:2.31,tokens:184302}}这样用户会第一次真正意识到Agent执行是有预算的前端不要显示完整 Chain-of-ThoughtRun Canvas 需要展示的是步骤结果证据ToolArtifact错误决策。不是模型内部私有推理。例如Reason: 3个Auth测试失败均与Refresh Token并发有关。就够了。状态图和聊天怎么联动用户在 Chat 里说先不要改Dockerfile。后端不是只把它追加成消息。应该产生CONSTRAINT_ADDED PLAN_UPDATED STEP_CANCELLEDCanvas 立即看到 Plan 变化。这才叫共享状态。一个真正需要处理的坑乱序事件如果多 Worker 并发发布sequence 21 sequence 23 sequence 22前端不能按到达顺序应用。最简单的办法是服务端保证 Run 内 Sequence 有序后再发。否则前端要维护 Bufferif(event.sequencelastSequence1){apply(event);}else{buffer(event);refetchSnapshot();}我更倾向服务端处理。第二个坑页面开了几个小时Run 可能已经归档。服务端发送RUN_COMPLETED前端收到终态以后source.close();不要让数千个完成任务继续占着 SSE 连接。第三个坑Event Store 无限增长保留策略可以是近期Run完整事件 长期压缩事件关键Audit 事故Run永久保留每 N 个事件写一个 Snapshot。推荐的接口集合POST /api/runs GET /api/runs/{runId} GET /api/runs/{runId}/events GET /api/runs/{runId}/artifacts POST /api/runs/{runId}/cancel POST /api/runs/{runId}/approvals/{id} POST /api/runs/{runId}/retry/{stepId}前端基本就够用了。测试别只测页面能动至少覆盖SSE断线重连 Last-Event-ID补发 重复事件去重 乱序 Snapshot与Event一致 多用户权限 审批版本冲突 Run完成关闭连接 Artifact越权 Budget实时更新我会怎么继续扩展这个 Demo第一阶段先做Run Step SSE React Flow第二阶段加Artifact Approval Budget第三阶段再加Trace Diff Replay Multi-Agent不要第一天就做一个“AI IDE”。最后这套实现真正有价值的地方不是 React Flow。React Flow 只是把节点画出来。真正的基础是结构化Run State 不可变Event Snapshot 审批 Artifact一旦这些后端语义存在你可以把前端换成React Flow看板时间线表格IDE 插件。都没关系。Agent 产品真正该摆脱的是所有状态都藏在聊天字符串里当执行过程变成结构化状态以后人才能真正看得懂、插得上手也才敢让 Agent 跑更长的任务。
