1. 先说结论：为什么 Agent 场景必须上“推”而不是“拉”
做 Agent 应用做久了，你会发现一个很尴尬的现实：Agent 根本不是一个“请求-响应”模型，而是一个“思考-调用-再思考-再调用-最终回答”的连续过程。一次用户提问，背后可能是两次大模型推理，中间还夹杂着查询数据库、调外部 API、读取文件等工具动作，整个链路跑下来，快则两三秒，慢则几十秒甚至更久。
如果在这里用传统 HTTP 接口去等结果，前端会遇到两个非常头疼的问题。第一是超时，前端 fetch 的默认超时通常也就是几十秒，Agent 推理一旦慢一点，用户那边直接看到网络错误，但实际上后端还在跑，这种体验谁遇到谁知道。第二是体验差，大模型生成答案本身是流式的，用户习惯了一个字一个字往外蹦的“思考感”，结果你憋了半天一次性丢给他一整段，心理落差非常大。
轮询虽然能解决一部分问题，但代价也不小。最常见的是前端每隔一两秒去拉一次任务状态，任务没结束就继续拉，这种做法在并发量上来之后非常浪费资源。而且轮询有一个天然劣势：它拿不到中间的“增量”，每次拉回来的都是全量状态，想要展示 Agent 的实时思考过程，得把原先后端存的整段 JSON 反复拉下来，网络开销和前端解析成本都很高。
我这次的做法是直接放弃“拉”，全部改成“推”。服务端用 SSE（Server-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 是 sessionId，value 是对应的 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)); // 判断是否有工具调用意向 Optional<ToolCall> 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拉取 |
| 会话锁 | String | SET 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-timeout=600000 server.tomcat.connection-timeout=600000另一个隐蔽的问题是线程管理。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 里拉取这个序号之后的所有事件，一次性补齐。
// 重连时补发遗漏事件 List<String> 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 里的数据只负责最近几小时的重放需求。
一个比较新的方向是拥抱 MCP（Model Context Protocol）。我发现基于这套事件协议，天然适合对接 MCP Server：MCP 的消息本质上就是工具调用和结果返回的结构化流转，和 Agent 事件模型非常契合。后续想让系统支持多种模型供应商时，用 MCP 作为工具接入层，会比现在手写每一个工具适配器清晰得多。
踩了几个坑之后，我最深的感受是：Agent 应用的难点不在调用大模型的 API，也不在把工具写出来，而在于怎么把整个运行过程以结构化、可追踪、可恢复的方式组织起来。SSE 和 Redis 给的正是这套组织能力。前者把过程实时呈现给用户，让 Agent 的“思考感”有了实体形态；后者把过程沉淀下来，让分布式环境下多实例协作成为可能。两者结合起来，才是一个真正能上生产的 Agent 中枢，而不是一个只能跑通 Demo 的玩具。
