企业官网建设流程全解析

做 Agent 流式输出，最怕的不是模型笨，而是连接断。前期在一个真实项目（后面我统一叫它“模拟项目X”）里做多轮对话 Agent，一开始图省事用同步接口加前端轮询，结果线上天天有人反馈“转圈圈好几秒没反应”“看到一半突然断掉”“刷新以后上下文对不上”。后来痛定思痛，从轮询改成 SSE，引入 Redis 做事件中枢，才算把这条链路真正盘活。

这篇文章把完整思路写下来：为什么轮询会造成假等待和真断流，SSE 协议到底解决了什么，Spring Boot 3 怎么把 SSE 落地，Redis 又如何解决多实例部署下的流式推送一致性问题，以及在工具调用场景下事件模型应该怎么设计。内容不止有代码，还会讲清楚每个方案的适用边界和我在生产环境里踩过的坑，适合正在做 Agent 服务化、想把“思考过程”和“工具调用过程”实时呈现给前端的同学参考。

1. 为什么同步接口总觉得卡：轮询、长轮询与流式等待的本质区别

1.1 同步 HTTP 响应模型承载不了 Agent 的“思考过程”

大多数初版 Agent 服务都会走同一个套路：前端 POST 一个用户问题，后端同步调用大模型接口，拿到完整回答后 JSON 一次性返回。这个模型在普通问答里没什么问题，但一旦落到 Agent 场景，体验立刻变得很糟。

Agent 的完整执行链路往往不是“一次推理出结果”，而是“多轮推理 + 多轮工具调用”。模型可能先思考要不要搜索、然后调用搜索工具、等待工具返回、再基于工具结果继续推理。这整个过程耗时可能从几秒到几十秒不等。同步接口只能在整个链路跑完之后把最终结果返回给前端，这段时间用户面对的就是一个无反馈的加载态。

可能有人会说，那就让前端轮询任务状态接口，每 500ms 问一次“好了没有”。这在短任务场景勉强能用，但一旦 Agent 链路变长，轮询会带来两个明显问题：第一，前端无法感知中间状态，用户只知道“在跑”，不知道“在思考”还是“在调工具”；第二，后端要为每次轮询做一次任务状态查询，QPS 直接翻几倍。

1.2 轮询机制下的“假活”与断流隐患

我见过很多系统上线初期轮询效果看着还行，等到并发一上来，问题就集中爆发了。首先是数据库压力，每 500ms 轮询一次意味着每个会话每秒钟至少查询两次任务状态，如果在线会话有几千个，状态表的读压力就非常难看。

其次是网关超时。同步接口往往需要等 Agent 完整跑完才返回，如果一次 Agent 链路超过 60 秒，网关层的 read timeout 会直接把连接掐断。前端收到超时错误后再重新发起轮询逻辑，又会导致任务被重复触发或者状态错乱。

更隐蔽的问题在于“假活”：任务其实已经因为工具调用异常中止了，但状态字段没及时更新，前端还在无限轮询，用户看到的就是一个永远转圈的页面。这个体验对生产系统是致命的，用户不会认为“是模型太慢”，只会认为“你的系统挂了”。

1.3 Agent 思维链交互天然需要“流”而不是“快照”

还有个更根本的问题：Agent 的思考过程和工具调用过程，本质上是一串连续事件，而不是一个静态结果。模型在一次推理中会输出多段中间思考，之后可能会串行或并行发起多个工具调用，每个工具调用又有自己的开始、结束、成功、失败几个阶段。

这种事件流模型，用轮询快照来表达是天生错位的。快照只能展示“当前状态”，表达不了“中间发生过什么”。而用户真正想要的，是看到一个 Agent 在实时地“思考 → 调工具 → 看结果 → 继续推理”的完整过程。这也就是业界常说的 AI 原生交互体验。要做到这一点，必须有一个服务端向客户端主动推送事件流的通道，SSE 就是最合适的协议之一。

2. 为什么选 SSE 而不是 WebSocket：协议特性与 Spring Boot 3 支持细节

2.1 SSE 与 WebSocket 的本质对比

SSE（Server-Sent Events）和 WebSocket 经常被放在一起比较，但它们的定位完全不同。

对 Agent 流式输出场景来说，核心诉求就是“服务端把事件流推给客户端”，几乎不需要客户端向服务端发实时消息——用户的输入仍然是一个普通的 POST 请求。所以 SSE 是更契合的方案，不需要引入 WebSocket 的复杂度与网关兼容性成本。

2.2 Spring Boot 3 中 SSE 的落地形态

Spring Boot 3 的 MVC 栈内置了SseEmitter，使用方式和传统的异步返回值非常接近。简单说，Controller 方法返回一个 SseEmitter 实例，Spring 会把 HTTP 响应升级为text/event-stream流式响应，之后业务代码可以随时往这个 emitter 里写入事件。

@RestController @RequestMapping("/agent") public class AgentController { private final AgentStreamService streamService; public AgentController(AgentStreamService streamService) { this.streamService = streamService; } @PostMapping(value = "/chat", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter chat(@RequestBody ChatRequest request) { SseEmitter emitter = new SseEmitter(0L); // 0L 表示不超时 streamService.runAgentTask(request, emitter); return emitter; } }

这里produces = MediaType.TEXT_EVENT_STREAM_VALUE是关键，Spring 会根据这个值生成符合规范的 SSE 响应头。前端使用原生EventSource或者fetch的ReadableStream都能直接消费。

2.3 前端基础接入

原生EventSource非常简单，但有一个限制：它只支持 GET 请求。如果你需要 POST 携带用户输入，就不能直接用EventSource，需要改用fetch配合ReadableStream解析。

const response = await fetch('/api/agent/chat', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ message: '帮我查一下明天的天气', sessionId: 's_001' }), }); const reader = response.body.getReader(); const decoder = new TextDecoder('utf-8'); let buffer = ''; while (true) { const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); const events = buffer.split('\n\n'); buffer = events.pop() || ''; for (const rawEvent of events) { const dataLines = rawEvent .split('\n') .filter(line => line.startsWith('data: ')) .map(line => line.slice(6)); if (dataLines.length) { const payload = JSON.parse(dataLines.join('\n')); handleAgentEvent(payload); } } }

前端只需要关心两类东西：事件里event字段标识的事件类型，以及data字段里的 JSON 载荷。注意 SSE 消息以空行分隔，所以解析时要用\n\n做分段。

3. Redis 在流式链路中的角色：为什么多实例部署必须引入事件中枢

3.1 单实例没问题，一上负载均衡就翻车

很多人刚上手 SSE 时只在本地跑 Demo，一切正常，但一部署到多实例环境就发现：用户请求被负载均衡分到了 A 实例，Agent 任务却在 B 实例上异步执行，A 实例完全不知道事件已经产生，SSE 连接就卡死等到超时。

问题的根源在于：SSE 连接是由具体的某个服务实例持有的，但异步任务的事件是在另一个实例上产生的。没有事件中枢，两个实例之间无法完成事件的投递。

这时 Redis 的价值就体现出来了。每个实例启动时往 Redis 注册自己的实例标识，当需要推送事件到某个会话时，先根据会话与实例的映射关系找到目标实例，再通过 Redis 的事件通道把消息投递给目标实例，由目标实例上的投递组件写入对应的 SseEmitter。

3.2 用 Pub/Sub 还是 Stream：生产场景的取舍

Redis 提供两种消息模型：Pub/Sub 和 Stream。很多文章推荐 Pub/Sub，因为使用简单、实时性好。但 Pub/Sub 有一个致命缺陷：消息不持久化。如果目标实例临时断开连接，消息直接就丢了。Agent 运行到关键节点时事件丢失，前端就会断流，用户不知道 Agent 进行到哪一步。

Stream 则不同。Stream 支持消费者组、消息持久化和 ACK 机制。即使目标实例暂时不可用，消息也会保存在 Stream 里，实例恢复后可以继续消费。对生产环境来说，我更倾向于使用Redis Stream + Consumer Group，它的可靠性和可回溯性明显更强。

spring: data: redis: host: 127.0.0.1 port: 6379 timeout: 3000ms

3.3 从 Stream 到 SseEmitter 的桥接组件

实现思想很直接：Agent 任务在任意实例上运行，运行过程产生的每一步事件全部写入 Redis Stream；每个实例启动一个后台线程，消费自己负责的会话事件，然后转发给本地持有的 SseEmitter。

@Component public class SseEventBridge implements InitializingBean { private static final String STREAM_KEY = "agent:stream:events"; private final StringRedisTemplate redisTemplate; private final Map<String, SseEmitter> emitterRegistry = new ConcurrentHashMap<>(); public SseEventBridge(StringRedisTemplate redisTemplate) { this.redisTemplate = redisTemplate; } @Override public void afterPropertiesSet() { // 后台消费线程 Executors.newSingleThreadExecutor().execute(this::consumeEvents); } public void register(String sessionId, SseEmitter emitter) { emitterRegistry.put(sessionId, emitter); emitter.onCompletion(() -> emitterRegistry.remove(sessionId)); emitter.onTimeout(() -> emitterRegistry.remove(sessionId)); emitter.onError(e -> emitterRegistry.remove(sessionId)); } public void publish(String sessionId, AgentEvent event) { String payload = JsonUtils.toJson(event); redisTemplate.opsForStream().add(STREAM_KEY, Map.of(sessionId, payload)); // key 为 sessionId } private void consumeEvents() { while (!Thread.currentThread().isInterrupted()) { try { List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream().read(STREAM_KEY, StreamOffset.fromStart(STREAM_KEY)); for (MapRecord<String, Object, Object> record : records) { record.getValue().forEach((sessionKey, eventData) -> { String sessionId = sessionKey.toString(); SseEmitter emitter = emitterRegistry.get(sessionId); if (emitter != null) { try { emitter.send(SseEmitter.event() .name("agent") .data(eventData.toString())); } catch (IOException e) { emitterRegistry.remove(sessionId); emitter.completeWithError(e); } } }); } } catch (Exception e) { // 消费线程降级等待 } } } }

这段代码是简化的核心骨架，实际生产里还要加上消费者组XREADGROUP、ACK 确认、批量读取和空等待设置，否则每轮循环都会空转打满 CPU。用BLPOP思路类比 Stream 的空等待，可以参考StreamReadOptions里的block参数，让读取阻塞指定毫秒数。

3.4 Redis 同时承担会话状态存储

除了事件中转，Redis 还能顺手承担会话上下文存储。Agent 的多轮对话需要携带历史消息，如果全部放在内存，实例重启后上下文就丢了。实践中我会把会话上下文、运行状态、工具执行记录统一放到 Redis Hash 里，用session:{id}作为 key，这样不仅能跨实例共享状态，排查问题时还能直接通过 RedisInsight 查看某次对话到底停在哪个环节。

4. Agent 事件模型设计：让“思考”和“工具调用”都能被前端理解

4.1 统一事件类型定义

SSE 通道只解决了“怎么传”的问题，真正决定体验的是“传什么”。在模拟项目X里，我把 Agent 运行过程抽象成四类核心事件：

• thinking：模型输出的中间思考文本片段
• tool_call：Agent 发起了一个工具调用请求
• tool_result：工具执行完成并返回结果
• answer：最终回答文本片段
• error：链路中出现异常
• done：整条链路结束

每类事件统一一个 JSON 结构：

{ "event": "tool_call", "sessionId": "s_001", "data": { "toolName": "web_search", "toolCallId": "call_123", "args": { "query": "明天的上海天气" }, "ts": 1710000000000 } }

前端收到tool_call事件时，可以立刻在界面上渲染一个“正在调用工具 web_search”的卡片，展示参数内容。这比单纯显示“思考中”要直观得多。

4.2 模型流式输出如何转成事件

大模型接口返回的流式内容是增量文本，比如 OpenAI 兼容接口的chat.completion.chunk。每拿到一个 chunk，直接把它包装成thinking或answer事件推给前端，就能形成打字机效果。

不同之处在于是思考阶段还是最终回答阶段。我的做法是让 Agent 执行器维护一个phase状态：

• 首次推理阶段：chunk 文本进入thinking事件
• 触发工具调用且返回结果后：后续 chunk 进入answer事件

注意思考文本可能包含模型内部推理，不一定适合展示给所有用户。如果想保守一点，可以在thinking事件里做脱敏或只传摘要。对于内部知识助手类产品，完整展示思考过程反而容易暴露提示词细节，这块需要根据业务自己权衡。

4.3 工具调用的串行与并行处理

工具调用在 Agent 链路里可能是串行执行多个工具，也可能是一次性并行发起多个调用。事件模型应该支持tool_call携带一个sequence序号或者group标识，方便前端把并行调用显示成并列卡片。

当工具调用结束后，tool_result事件里的内容可能很大。比如网页抓取工具返回全文，几万字符一次塞给前端既不友好又占带宽。实践做法是只把工具结果的前 N 个字符作为摘要推给前端，完整内容放到 Redis 里按需触达。这一步能显著降低 SSE 推送的压力。

5. 生产环境下的断流治理：超时、重连、背压与网关协同

5.1 SseEmitter 的超时配置

默认的 SseEmitter 有超时时间，超过设定时间没有任何写入，连接会自动关闭。对于 Agent 长任务，我建议构造时传入0L表示不超时，但要配套心跳机制，否则代理网关可能会因为长时间没有数据传输而断开连接。

心跳逻辑很简单：单独起一个定时任务，每 15 秒向所有活跃 emitter 发送一个ping事件。Spring 的SseEmitter.event().name("ping")就是干这个用的。前端收到ping就忽略或更新时间戳，连接则保持活性。

@Scheduled(fixedRate = 15000) public void heartbeat() { emitterRegistry.values().forEach(emitter -> { try { emitter.send(SseEmitter.event().name("ping").data("{}")); } catch (Exception ignored) { emitterRegistry.values().remove(emitter); } }); }

5.2 客户端断连与 Redis 中的残留事件清理

客户端断连后，SseEmitter 会触发onCompletion或onError回调，此时要立刻把当前 session 从注册表中移除。如果不移除，后续事件还会尝试写入已经失效的 emitter，产生无意义异常。

更麻烦的是已经写进 Redis Stream 但还没被消费的事件。由于消费者是逐个实例在跑，A 实例持有的 emitter 断开后，B 实例补位时处理残留事件需要知道当前会话由谁接管。简单方案：在onError回调里，额外往 Redis 写一条session:terminated事件，消费端读到该事件就把后续事件丢弃。

5.3 网关层与 Nginx 的缓冲关闭

Nginx 默认会缓冲后端响应，对 SSE 这种逐条推送的流式响应会带来几秒延迟甚至断流。生产环境一定要关闭代理缓冲：

location /api/agent/chat { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Connection ""; proxy_buffering off; proxy_cache off; proxy_read_timeout 300s; chunked_transfer_encoding off; }

proxy_buffering off是关键，同时proxy_read_timeout调大，避免长时间无数据时 Nginx 主动断开连接。如果前端直接走 CDN 或其他云产品，也需要确认它们具有 SSE 透传能力，否则流式响应会被砍断。

5.4 背压：当模型吐得比网络快时怎么办

正常情况模型生成速度远低于带宽，但在某些高并发集群场景，多个 Agent 任务同时往同一实例推送事件，可能导致瞬间事件堆积。在模拟项目X的压测里，我们用单实例同时挂 200 个 SSE 连接，模型输出瞬时峰值时，内网传输没问题，但外网出口会成为瓶颈。

处理思路是对SseEmitter.send()的结果做失败重试，同时用信号量限制单实例最大活跃 emitter 数量。超出阈值时，新的会话请求直接返回“繁忙”状态码，避免雪崩。

6. Agent 任务执行器设计：流式产出背后的异步执行模型

6.1 从请求线程到后台任务线程的切换

Controller 返回 SseEmitter 后，请求线程会被释放。不能让请求线程阻塞等待 Agent 执行完毕，必须把任务提交到独立的线程池，让任务执行器把事件依次传给 Redis Stream，再由桥接组件转发给 emitter。

@Service public class AgentStreamService { private final ThreadPoolTaskExecutor agentExecutor; private final SseEventBridge bridge; public AgentStreamService(ThreadPoolTaskExecutor agentExecutor, SseEventBridge bridge) { this.agentExecutor = agentExecutor; this.bridge = bridge; } public void runAgentTask(ChatRequest request, SseEmitter emitter) { // 先注册会话到桥组件，再提交到后台线程 bridge.register(request.getSessionId(), emitter); agentExecutor.execute(() -> { try { bridge.publish(request.getSessionId(), AgentEvent.of("thinking", "开始解析用户意图...")); AgentResult result = runAgentLoop(request); bridge.publish(request.getSessionId(), AgentEvent.of("done", result)); } catch (Exception e) { bridge.publish(request.getSessionId(), AgentEvent.of("error", e.getMessage())); } finally { emitter.complete(); } }); } }

线程池的参数要单独设计，不能用Executors.newCachedThreadPool()这种无界线程池，并发一高直接把内存打爆。用 Spring 的ThreadPoolTaskExecutor，核心线程数、最大线程数、队列容量按服务资源设定，同时拒绝策略设为CallerRunsPolicy或者直接返回错误。

6.2 Agent 主循环与工具调用时机的衔接

Agent 主循环可以简洁地概括为一个 while 循环：

private AgentResult runAgentLoop(ChatRequest request) { List<ChatMessage> messages = buildMessages(request); int maxSteps = 5; for (int step = 0; step < maxSteps; step++) { StreamChunk response = modelClient.streamChat(messages); StringBuilder thinking = new StringBuilder(); ToolCallRequest toolCall = null; for (Chunk chunk : response.getChunks()) { if (chunk.getToolCalls() != null && !chunk.getToolCalls().isEmpty()) { toolCall = chunk.getToolCalls().get(0); } else { thinking.append(chunk.getContent()); bridge.publish(request.getSessionId(), AgentEvent.thinking(chunk.getContent())); } } if (toolCall != null) { bridge.publish(request.getSessionId(), AgentEvent.toolCall(toolCall)); String result = executeTool(toolCall); bridge.publish(request.getSessionId(), AgentEvent.toolResult(toolCall, result)); messages.add(resultAsMessage(toolCall, result)); continue; } bridge.publish(request.getSessionId(), AgentEvent.answer(thinking.toString())); return AgentResult.success(thinking.toString()); } return AgentResult.error("Agent steps 超出限制"); }

这段代码是模拟项目X里的核心主循环逻辑。注意流式接口的 chunk 里，工具调用参数往往是增量拼出来的，需要在循环里累积参数片段，不能每块直接解析，否则 JSON 必定残缺。

6.3 工具执行隔离与超时保护

工具执行不能信任外部接口的耗时。每个工具调用都应该包一层带超时的 Future，超过比如 30 秒直接中断，并向流中推送tool_result事件，内容标明超时失败。

public String executeTool(ToolCallRequest call) { ExecutorService toolExecutor = Executors.newSingleThreadExecutor(); try { Future<String> future = toolExecutor.submit(() -> { return toolRegistry.invoke(call.getToolName(), call.getArgs()); }); return future.get(30, TimeUnit.SECONDS); } catch (TimeoutException e) { return "{\"error\": \"tool timeout\"}"; } finally { toolExecutor.shutdownNow(); } }

工具结果返回后塞回 messages 给模型继续推理这一步特别关键。如果漏掉这一步，模型第二次推理时看不到工具返回结果，就会重复发起同样的工具调用，形成死循环。在早期版本上，这个问题反复出现，后来靠日志和链路追踪定位到是消息构造时少了 role 和 content 的格式映射。

7. 横向扩展与多租户隔离：把单实例 Demo 升级成真正可用服务

7.1 多实例会话与实例的映射策略

前面提到的emitterRegistry是实例内存态，多实例时需要保证“会话 S 的事件，只会被持有该会话 emitter 的实例消费”。最简单的映射方式是：建立 SSE 连接时，把 sessionId 与实例ID写入 Redis Hash，例如agent:session:里记录sessionId -> instanceId。

Agent 执行器在写入事件前，不直接写公共 Stream，而是先查询会话所在实例，将事件写入该实例专属 Stream 或带上分区标识。每个实例只消费自己专属 Stream，这样避免同一事件被多个实例抢到但只有一个能推送，另一个拿走却用不上的问题。

7.2 消费线程的可靠性与故障恢复

消费者线程如果异常退出，事件就无法投递。我在afterPropertiesSet里启动的不是裸线程，而是经过 Spring 管理的TaskScheduler，并加了线程内异常捕获和日志告警。配合 Redis Stream 的 ACK 机制，消息即使被投递但 emitter 已经失效，ACK 仍然完成，保证消息不会反复投递。真正需要重试的是工具调用错误的事件，那应该在 Agent 内部重试，而不是在消息投递层重试。

7.3 租户隔离与鉴权透传

如果这个中枢要服务多个业务方，事件模型里必须带上租户标识。Redis 的 Stream key 可以按租户拆分，例如agent:stream:{tenantId}。SSE 接口的鉴权信息通过请求头进入，事件推送到前端时要把鉴权信息剥离，只保留业务数据，避免内部 token 泄露到浏览器。

我这边的习惯是：把租户 ID 和会话 ID 作为 Redis key 的前缀组合，所有查询、清理、监控脚本都基于这个前缀展开。等到要排查生产问题时，一条命令就能按租户把活跃会话捞出来。

8. 监控与排障：事件延迟、丢失与断流的定位方法

8.1 链路追踪事件流

SSE 流式链路比较隐蔽，问题往往发生在“Agent 执行器 → Redis → 桥接线程 → SseEmitter”这条链路的某个环节。没有追踪字段很难快速定位。我在每个事件里都加了traceId和timestamp，同时在 Redis Stream 里记录事件写入时间，在桥接组件里记录事件投递时间，两边一对比就能看出卡在哪段。

如果 Redis 投递到桥接组件有延迟，看消费者线程是否堆积；如果桥接组件到前端有延迟，看那是网络带宽问题还是 emitter 写入异常。

8.2 Redis 队列积压监控

在模拟项目X上线后，我添加了一个定时监控，每分钟检查 Stream 队列长度。正常情况下队列长度接近 0，因为事件被实时消费。如果队列长度持续增长且消费者 lag 居高不下，基本可以确定消费者的桥接线程卡死或者某个会话的 emitter 长期未移除，事件一直在空转堆积。

XLEN agent:stream:events XINFO GROUPS agent:stream:events

两条命令就能定位问题。XINFO GROUPS里会显示pending和lag两个关键字段，pending 高说明消费了但没 ACK，lag 高说明生产速度远超消费速度。

8.3 前端断流的常见诱因

不是所有断流都是后端问题。根据统计，大量断流发生在移动网络切换 Wi-Fi、手机锁屏、浏览器标签页休眠等场景。移动端浏览器在锁屏后 WebSocket 和 SSE 连接都可能被系统冻结。

解决方案是从产品层让“断线续传”具备真实可用性：前端保存最近的 contextId，重连时通过 POST 传递 contextId，后端从 Redis 拉取最后 N 个事件做补偿推流。这个功能我在模拟项目X的第三个迭代才补上，补上之后用户反馈里“看到一半断了”的占比明显下降。

8.4 容量评估基础公式

单实例所能支撑的 SSE 连接数，往往不是后端内存决定的，而是连接本身的文件描述符和线程管理开销。每个 SseEmitter 对应一个连接，每个连接至少占据一个 fd。生产容器默认的ulimit -n如果只有 1024，撑个几百路 SSE 就到瓶颈了。

容量规划时按“单实例 1000 条活跃 SSE 连接”做初步预算，每连接按照心跳 15 秒一条、业务事件平均每秒 2 条计算，单实例面对的业务消息吞吐其实不高。真正高开销在模型返回的 chunk 文本大小，一个完整思考事件可能几千字符，换算成流量要单独估算。

9. 从模拟项目X到通用方案：这套架构的边界与可扩展方向

9.1 什么场景没必要用这套架构

SSE + Redis Stream 的组合不是银弹。如果只是做简单的聊天问答、需求里完全没有工具调用和中间思考展示，同步接口就足够了。引入 Redis 和消费线程的复杂度反而让简单的链路变得难以排查。

另外，如果业务客户端主要是移动 App 且网络环境不稳定，SSE 在弱网下的恢复体验不如 WebSocket 稳定。这种场景可以从长连接选型上重新评估，或者配合客户端主动拉流补偿。

9.2 事件协议的兼容扩展空间

目前事件协议里有thinking、tool_call、tool_result、answer、error、done六类。实际业务扩展时，可以增加cleanup、escalation、human_handoff这类高级事件。比如某些客服场景，Agent 判断自己无法回答时需要转人工，这时推送一个handoff事件，前端可以直接弹人工客服按钮。

事件结构只要坚持event + data + ts的通用形态，后续不管加什么语义，前端都能无痛接入。这是我比较推荐保留的最小事件协议，不要为了省流量把事件语义压得太扁，否则前端解析成本会无限上升。

9.3 迁移到其他消息中间件的可能性

Redis Stream 在单集群内可靠性和吞吐都能满足中小规模场景。但如果未来会话量到达百万级、或者需要离线的流式日志回放，可以考虑升级到专业的消息队列。关键点是不要把业务逻辑和 Redis 的 API 强耦合，可以在桥接组件里封装一层EventPublisher接口，Redis 只是其中一个实现。

我之所以在模拟项目X里直接用了 Redis，是因为这套系统在评估时运维侧 Redis 已经是成熟组件，不用新增消息队列的运维负担。这一点对很多小团队来说是非常重要的决策因素。

10. 最后分享两个实操细节

第一个细节是，SseEmitter 在send之后并不是立刻把数据推到直达前端，中间还隔着 TCP 缓冲区和服务端进程的 IO 调度。如果你在本地测试时感觉“有延迟”，先别怀疑 Redis，先看是不是send频率太低或者前端没在onmessage里做即时处理。SSE 是流式协议，数据会一段段到达，前端解析代码写得不好仍然会攒成一块渲染。

第二个细节是，Redis Stream 的XREADGROUP消费循环里，如果 block 时间设置太长，比如 10 秒，那某个新会话注册之后，需要等待最多 10 秒才能收到自己的事件。线上最直观的表现就是“请求进来了，但页面 10 秒后才看到第一个字符”。早期我把 block 设为 5000ms，就遇到这个奇怪的延迟现象。测试时用短连接不明显，并发稍稍一大，问题立刻暴露。后来统一改成 200ms 轮询间隔加 block 500ms，延迟就从秒级降到了可感知范围内。

Agent 流式输出的核心不是某一个单独的技术点，而是把协议、中间件、任务执行、事件模型、网关策略串成一条完整的链路。只要这条链路的事件被清晰定义、可靠传递、可观测可恢复，用户的体感就会从“转圈等待”变成“看着 AI 一步步工作”。希望这篇实践复盘能帮你少走一些我绕过的弯路。