1. 先说结论为什么 Agent 场景必须上“推”而不是“拉”做 Agent 应用做久了你会发现一个很尴尬的现实Agent 根本不是一个“请求-响应”模型而是一个“思考-调用-再思考-再调用-最终回答”的连续过程。一次用户提问背后可能是两次大模型推理中间还夹杂着查询数据库、调外部 API、读取文件等工具动作整个链路跑下来快则两三秒慢则几十秒甚至更久。如果在这里用传统 HTTP 接口去等结果前端会遇到两个非常头疼的问题。第一是超时前端 fetch 的默认超时通常也就是几十秒Agent 推理一旦慢一点用户那边直接看到网络错误但实际上后端还在跑这种体验谁遇到谁知道。第二是体验差大模型生成答案本身是流式的用户习惯了一个字一个字往外蹦的“思考感”结果你憋了半天一次性丢给他一整段心理落差非常大。轮询虽然能解决一部分问题但代价也不小。最常见的是前端每隔一两秒去拉一次任务状态任务没结束就继续拉这种做法在并发量上来之后非常浪费资源。而且轮询有一个天然劣势它拿不到中间的“增量”每次拉回来的都是全量状态想要展示 Agent 的实时思考过程得把原先后端存的整段 JSON 反复拉下来网络开销和前端解析成本都很高。我这次的做法是直接放弃“拉”全部改成“推”。服务端用 SSEServer-Sent Events把 Agent 的每一个步骤包括思考内容、工具调用、工具返回结果、最终回答全部实时推给前端Redis 在中间承担任务状态存储和分布式协调的职责。这套方案做下来前端代码简单了后端也不再有那种“任务卡住没人知道”的焦虑整体结构清晰很多。这套东西适合谁如果你正在做 AI Agent、MCP、工作流引擎或者你在后端对接大模型需要流式返回再或者你已经在用轮询方式拉任务状态想换掉那这篇内容应该能帮你少走不少弯路。2. 方案选型SSE 凭什么比 WebSocket 和轮询更适合2.1 三种方案的直观对比先看一下三种方案放在一起是什么效果。轮询其实就是定时请求实现简单但延迟不可避免就算间隔已经压到一秒也比真正的实时慢半拍。而且这个方案在 Agent 场景特别别扭因为你不知道任务到底要跑多久间隔大了用户体验差间隔小了服务端压力大属于怎么做都不太对。WebSocket 是很多人一上来就选的方案功能确实强双向通信性能也优秀。但它有两个问题。第一是复杂度前端要处理连接建立、心跳、断线重连、消息协议后端要维护长连接的会话状态如果只是为了让服务端往客户端单向推消息这有点用牛刀杀鸡。第二是基建成本如果项目里有 Nginx 或网关还得额外确认它们对 WebSocket 的支持和超时配置是否没问题有些内部网关对 WS 协议支持得并不好排查起来很费劲。SSE 正好卡在两者之间。它是基于 HTTP 的单向流式推送方案服务端往客户端推数据非常自然客户端用浏览器原生的 EventSource 接口就能接不需要任何额外的客户端库。它还自带断线重连机制连接断了浏览器会自动重连重连时还能通过Last-Event-ID把断点告诉服务端让服务端把错过的消息补回来。这些东西 WebSocket 默认是不提供的自己实现要花不少功夫。2.2 SSE 在这套架构里的准确位置但也要说清楚SSE 不是万能的。它是单向的不适合做双向实时交互比如协同编辑、实时聊天这类需求还是 WebSocket 更合适。但在 Agent 场景里交互模式基本是固定的前端发起请求Agent 持续往回报进度前端偶尔需要中断一次生成这种时候用一个单独的 DELETE 或 POST 接口处理就行根本不依赖双向通道。所以我的判断是Agent 的流式展示场景SSE 是性价比最高的方案。技术上它能达到实时效果代码量比 WebSocket 小一个数量级前端接入成本低而且天然兼容 HTTP 协议栈网关和负载均衡器基本不用额外调整。再配合 Redis前面提到的多实例问题也能解开。后端服务部署多个节点时客户端连接的 SSE 通道只存在于它连上的那一台机器上如果 Agent 任务被另一个节点执行了怎么把事件投递给这个客户端做法是让所有节点订阅 Redis 的主题频道执行节点把事件发布到 Redis持有 SSE 连接的节点收到消息后转发给对应的前端。这样既解耦了任务调度和连接管理也实现了横向扩容。3. 整体架构与 Redis 的三个关键角色3.1 一个生产级 Agent 中枢长什么样我画一条链路你就明白了。前端打开页面发起一个对话请求这个请求先进入网关和后端服务后端把请求转给 Agent 编排引擎。引擎开始工作后每一次大模型推理的输出、每一次工具调用的中间结果都以事件的形式推给前端。Redis 在这个过程里不是一个可有可无的缓存它是整个机制的底盘。整套架构的核心参与者是这么几个组件职责接入层接收用户请求校验会话参数建立 SSE 连接Agent 编排引擎负责思考循环决定什么时候调工具、什么时候输出最终答案事件通道SSE 长连接负责把 Agent 的每一步推给前端Redis 状态存储存会话上下文、任务事件轨迹、工具执行结果缓存、并发控制锁Redis Pub/Sub多节点间的事件转发通道解决“连接在哪台机器上”的问题这个架构的好处是每个环节都是可替换的。SSE 连接挂了可以重连Redis 里的状态可以重建任务Agent 引擎可以根据事件轨迹恢复上下文。不会因为一个实例重启就把用户的整个对话过程丢掉。3.2 角色一会话状态与事件轨迹的存储底座Agent 运行过程中会产生大量中间状态。比如大模型思考了哪几步、调用了哪个工具、工具返回了什么结果、当前处于什么阶段这些如果只放在内存里服务重启就全没了而且多实例环境下也没法共享。我用 Redis 存了几类数据。第一个是会话上下文每次大模型推理后都会把最新的上下文记录写进去用 Hash 结构存一个 field 放原始用户消息一个 field 放累计的推理历史一个 field 放最新的工具执行结果。第二个是事件轨迹每个请求产生的事件都追加写入 Redis 的 List 结构这个 List 相当于一个操作流水账前端断线重连后可以按序号从 Redis 里拉取错过的内容。还有一类容易被忽略的是工具执行结果的短期缓存。Agent 可能会多次调用同一个语言模型而某些外部 API 对短时间内的重复请求是有限制的我在 Redis 里给每个工具调用设置一个去重窗口同一个参数的调用在一定时间内直接命中缓存返回结果既节省了外部资源也让整个 Agent 的响应速度更快了。3.3 角色二分布式环境下的节点间事件总线单节点跑 SSE 不会遇到什么问题但生产环境不可能只有一个实例。这时候就会出现一个特别常见的情况用户 A 的 SSE 连接建立在后端实例 1 上Agent 的任务却被负载均衡转发到了实例 2实例 2 跑完了任务事件不知道该怎么推给用户。这个问题最直接的解法是 Redis Pub/Sub。所有后端实例启动时都去订阅同一个主题比如说agent-events。任何一个节点执行 Agent 任务时把产生的事件发布到这个主题上所有订阅了这个主题的节点都会收到但只有持有对应 SSE 连接的那个节点会把事件真正转发给前端。听起来很简单但这里有几个细节要注意。第一Pub/Sub 的消息不持久化节点在消息发布后再启动就收不到之前发布出来的内容所以事件轨迹的持久化还是要落到 Redis 的 List 或者 Stream 数据结构里Pub/Sub 只承担实时推送这一层。第二Pub/Sub 在 Redis 里是广播模式业务量大的时候所有节点都会收到消息需要自己在消息体里带一个路由标识节点收到后判断这个任务是不是自己处理的不是就直接忽略。第三如果团队的技术栈支持也可以考虑用 Redis Stream 里带的 Consumer Group 做点对点分发在需要可靠投递的场景会更稳。3.4 角色三并发控制与幂等屏障Agent 场景里有几种并发问题特别典型。同一个用户反复点击提交导致同一个会话被多个任务同时操作消息顺序乱掉同一个 Agent 在多个节点上同时运行导致工具被重复调用。这些问题的通用解法就是分布式锁。我用 Redis 的SET key value NX PX原子命令来实现互斥。用户发起请求时先尝试获取会话锁拿到锁就继续处理拿不到就直接返回“已有任务在处理中”。锁的 value 用一个全局唯一的请求 ID释放锁时只删除自己持有 ID 的锁避免误删别人刚设置的锁。这套逻辑加上过期时间和自动续期基本能覆盖日常的并发场景。4. 基于 Spring Boot 3 的 SSE 核心链路实现4.1 SseEmitter 的正确打开方式Spring Boot 3 里做 SSE 用到的是SseEmitter使用起来不复杂但有几个细节必须注意。先看基本用法。RestController RequestMapping(/agent) public class AgentSseController { private final AgentTaskService taskService; public AgentSseController(AgentTaskService taskService) { this.taskService taskService; } GetMapping(value /stream/{sessionId}, produces MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter stream(PathVariable String sessionId, RequestHeader(value HttpHeaders.LAST_EVENT_ID_HEADER, required false) String lastEventId) { SseEmitter emitter new SseEmitter(0L); // 把连接注册到连接管理器供后续事件推送使用 SseConnectionManager.register(sessionId, emitter, lastEventId); return emitter; } }第一是SseEmitter的构造参数。如果不传参默认超时时间是 30 秒也就是说 30 秒内没有往连接写数据连接就会被自动关闭。Agent 工具调用时间往往不可控我建议要么传0L表示永不超时要么传一个合理的大值比如 15 分钟之后靠心跳机制来维持连接活跃。第二是响应头。Spring 已经帮你处理了Content-Type: text/event-stream问题不大。真正需要关心的是连接建立之后务必要在拦截器或过滤器里关闭缓存。因为 Nginx 或一些代理层会给响应加缓冲数据会堆积在缓冲区里用户长时间看不到输出这个问题后面详细说。第三是连接生命周期管理。每个 SSE 连接都应该注册到全局的 Map 里key 是 sessionIdvalue 是对应的 SseEmitter 对象。连接完成或超时时要清理掉这个映射防止内存泄漏。4.2 事件协议设计让前端拿到结构化流SSE 本身只是一层推送通道推送的内容才是核心。我设计了一套简单的事件协议用 JSON 表达。事件类型说明示例字段thinking大模型的思考过程content: 思考内容片段tool_call准备调用某个工具tool: 工具名arguments: 参数tool_result工具执行完成tool: 工具名result: 执行结果message最终回复内容content: 最终输出片段status阶段状态变化status: RUNNING/SUCCEEDED/FAILED实际推送的时候用SseEmitter.event()方法把事件名字和事件数据都发出去。前端可以根据事件名做分发处理而不是统一切一个 JSON 再去判断类型。// 推送思考过程 emitter.send(SseEmitter.event() .name(thinking) .data(Map.of( sessionId, sessionId, content, 用户想了解天气情况我需要调用天气服务, timestamp, System.currentTimeMillis() ))); // 推送工具调用 emitter.send(SseEmitter.event() .name(tool_call) .data(Map.of( sessionId, sessionId, tool, WeatherApi, arguments, Map.of(city, 北京), timestamp, System.currentTimeMillis() ))); // 推送工具结果 emitter.send(SseEmitter.event() .name(tool_result) .data(Map.of( sessionId, sessionId, tool, WeatherApi, result, 晴25度北风2级, timestamp, System.currentTimeMillis() )));设计事件协议时最关键的一点是不要让前端去理解业务细节。前端只需要知道“有一个思考事件来了”“有一个工具调用事件来了”然后渲染成 UI 即可具体内容是什么、后续要做什么那是 Agent 引擎逻辑的事。4.3 工具调用编排与流式推送的完整时序Agent 的核心是工具调用循环。大模型给出一个推理结果我们判断这个结果里有没有工具调用意图有就去执行工具把结果回传给大模型再推理一次直到大模型输出最终答案。这套链路里SSE 推送不是事后加上去的装饰而是整个编排的组成部分。每一步都产生事件每一个事件都实时推送。我把核心逻辑简化成一个伪代码流程你可以直观感受一下。public void executeAgentFlow(String sessionId, String userMessage) { // 事件总线初始化 EventPublisher publisher eventPublisher(sessionId); // 获取会话锁防重入 boolean locked sessionLock.tryLock(sessionId, 5, TimeUnit.SECONDS); if (!locked) { publisher.send(status, BUSY); return; } // 组装上下文包括从 Redis 恢复的历史消息 AgentContext context contextRepository.load(sessionId); try { int maxLoop 5; for (int i 0; i maxLoop; i) { publisher.send(status, THINKING); // 调用大模型流式取推理结果片段 String reasoning llmService.streamReasoning(context, chunk - publisher.send(thinking, chunk)); // 判断是否有工具调用意向 OptionalToolCall toolCall ToolCallExtractor.parse(reasoning); if (toolCall.isEmpty()) { // 没有工具调用说明拿到了最终答案 String finalAnswer llmService.streamFinalAnswer(context, chunk - publisher.send(message, chunk)); publisher.send(status, COMPLETED); break; } // 执行工具调用 publisher.send(tool_call, Map.of(tool, toolCall.get().getName(), arguments, toolCall.get().getArguments())); String toolResult toolExecutor.execute(toolCall.get()); context.appendToolResult(toolCall.get(), toolResult); // 工具执行结果先推给前端再把结果喂给大模型 publisher.send(tool_result, Map.of(tool, toolCall.get().getName(), result, toolResult)); } // 把生成的上下文和事件轨迹写入 Redis contextRepository.save(sessionId, context); } finally { sessionLock.release(sessionId); connectionManager.complete(sessionId); } }这里面比较见功夫的是思考事件的处理。大模型的推理结果本身可能是长文本如果你等到整段推理完了再一次性推送那和轮询没有本质区别。我的做法是在streamReasoning的回调里逐片段推送大模型输出一个 token 块就立刻通过 SSE 发给前端这样用户能看到思考内容一个词一个词地蹦出来体验和大模型正常打字输出保持一致。4.4 Redis 数据结构选型与代码示例Redis 在这个架构里存的东西不少每种数据用什么样的结构我列一下我的实践。数据结构说明会话上下文Hash存原始消息、推理历史、工具结果事件轨迹List每次推送前先RPUSH一份断线重连用LRANGE拉取会话锁StringSET NX PX实现互斥工具结果缓存String带过期时间短期去重心跳String每次心跳更新 TTL用于连接存活检测会话上下文的读写逻辑是这样// 保存上下文 redisTemplate.opsForHash().putAll(agent:context: sessionId, Map.of( userMessage, userMessage, history, JsonUtils.toJson(historyList), toolResults, JsonUtils.toJson(toolResultList) )); redisTemplate.expire(agent:context: sessionId, Duration.ofHours(24)); // 追加事件轨迹 redisTemplate.opsForList().rightPushAll(agent:events: sessionId, event1Json, event2Json, event3Json); // 只保留最近 500 条防止无限增长 redisTemplate.opsForList().trim(agent:events: sessionId, -500, -1);Redis 的 Key 设计建议带前缀和业务标识比如agent:context:{sessionId}、agent:events:{sessionId}这样在 Redis 里一眼能看出属于哪个模块排查问题也更方便。另外要给会话上下文设置过期时间避免用户不再回访导致内存持续占用。5. 生产环境必须踩过去的五个坑5.1 SseEmitter 超时与异步线程池这个坑我是在压测时才发现的。Tomcat 默认对响应有超时限制如果长时间没有向 SseEmitter 写入数据连接会被容器自动关闭。你以为设置了构造参数就不会超时了但容器层面可能还要单独调整。# Spring Boot 3 配置示例 spring.mvc.async.request-timeout600000 server.tomcat.connection-timeout600000另一个隐蔽的问题是线程管理。SseEmitter 的异步发送操作会占用 Tomcat 的工作线程如果你使用Async注解跑 Agent 任务务必配置一个独立的线程池别让 Agent 的长任务和普通 HTTP 请求争抢线程。我自己给了 8~16 个线程专门跑 Agent 任务避免相互影响。5.2 Nginx 和网关会把你的流吃掉SSE 最大的坑之一其实不在应用层而在代理层。Nginx 默认会开启响应缓冲数据会在缓冲区里攒够一整包才会发给客户端这在 SSE 场景下是致命的——用户会感觉前端一直没反应但后端明明已经在推数据了。解决办法是关闭缓冲并调大超时location /agent/stream/ { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_read_timeout 300s; proxy_send_timeout 300s; proxy_set_header Connection ; chunked_transfer_encoding off; }5.3 断线重连不是前端的事后端要配合做浏览器 EventSource 在连接断开后会自动重连这个机制是好的但默认情况下重连不会带任何断点信息。如果 Agent 任务已经执行到中间步骤重连后的新连接从零开始前端拿不到任何数据。解决办法是在服务端读取请求头里的Last-Event-ID参数。这个 ID 是前端从上次收到的 SSE 消息里保存下来的序号服务端收到重连请求时从 Redis 的事件轨迹 List 里拉取这个序号之后的所有事件一次性补齐。// 重连时补发遗漏事件 ListString missingEvents redisTemplate.opsForList() .range(agent:events: sessionId, lastEventId 1, -1); for (String eventJson : missingEvents) { emitter.send(SseEmitter.event().data(eventJson)); }这样做之后即使网络抖动导致连接断开几次用户看到的内容依然是连续的。5.4 Redis 分布式锁的续期与释放分布式锁的操作本身很简单但生产环境会遇到一个细节问题任务执行时间超过了锁的过期时间锁自动过期了另一个请求趁虚而入导致同一个会话被两个任务同时处理。我在实现时采用了一个简单的续期机制获取锁时设置一个后台定时任务每隔三分之一过期时长就续期一次任务完成时取消定时任务并释放锁。也可以用 Redisson 的RLock它自带的看门狗机制就是干这个的不过要用对方式别在无锁模式下创建实例就行。// Redisson 写法 RLock lock redissonClient.getLock(agent:session: sessionId); boolean locked lock.tryLock(0, 30, TimeUnit.SECONDS); // 业务逻辑 lock.unlock();5.5 长连接的监控与清理SSE 连接数量多了以后连接管理本身会成为一个问题。我在 SseConnectionManager 里做了一套简单的监控机制每个连接注册时记录时间戳同时开启心跳检查后端每 30 秒向连接写一个注释行如果连接已死写入时会抛异常捕获到就清理连接。// 心跳线程任务 Scheduled(fixedDelay 15000) public void heartbeatCheck() { for (SseEmitter emitter : connectionManager.getAllConnections()) { try { emitter.send(SseEmitter.event().comment(heartbeat)); } catch (IOException e) { connectionManager.remove(emitter); log.warn(移除失效 SSE 连接: {}, emitter); } } }清理动作做对了就不会出现那种“用户已经关了页面但服务端仍然保存着连接导致线程和内存一直占用”的问题。6. 从这套架构延伸出去的几个想法这套基于 Spring Boot 3 SSE Redis 的 Agent 中枢其实只是抓住了问题的主干。我还在它上面扩展过几个东西各有各的坑但也各有各的收益。第一个是任务状态机的完整化。现在事件轨迹里已经有了一套状态流转的雏形从 CREATE 到 THINKING、TOOL_CALLING、TOOL_RESULT、COMPLETED、FAILED下一步可以把它明确设计成一个状态机配合 Redis 的过期键和延迟队列就能做出任务超时自动补偿、失败自动重试的能力。第二个是前端渲染的多态化。现在前端拿到的是结构化的事件流每一种事件都对应一种 UI 组件。思考事件可以做成打字机效果的文本框工具调用事件可以做成一个信息卡片展示工具名和参数工具结果事件可以折叠成一个可展开的调试面板。这些如果在一开始接口设计阶段没留好余地后面补起来真的很痛苦。第三个是权限与审计。Agent 工具调用会接触到业务数据所以工具执行日志要留全。Redis 的事件轨迹虽然方便但保留时间有限长期审计建议异步转存到数据库或对象存储。我在后面的版本里是把事件轨迹通过消息队列同步一份到数据仓库Redis 里的数据只负责最近几小时的重放需求。一个比较新的方向是拥抱 MCPModel Context Protocol。我发现基于这套事件协议天然适合对接 MCP ServerMCP 的消息本质上就是工具调用和结果返回的结构化流转和 Agent 事件模型非常契合。后续想让系统支持多种模型供应商时用 MCP 作为工具接入层会比现在手写每一个工具适配器清晰得多。踩了几个坑之后我最深的感受是Agent 应用的难点不在调用大模型的 API也不在把工具写出来而在于怎么把整个运行过程以结构化、可追踪、可恢复的方式组织起来。SSE 和 Redis 给的正是这套组织能力。前者把过程实时呈现给用户让 Agent 的“思考感”有了实体形态后者把过程沉淀下来让分布式环境下多实例协作成为可能。两者结合起来才是一个真正能上生产的 Agent 中枢而不是一个只能跑通 Demo 的玩具。
