AI 响应流式处理从全量等待到 Server-Sent Events 的体验重构一、全量响应模式的体验断崖用户向 AI 助手提出一个复杂问题帮我分析这份 20 页合同的潜在风险点。非流式模式下用户点击发送后看到三个旋转的加载点持续 18 秒后整段回复一次性展示。这 18 秒内的用户行为数据显示30% 的用户在 10 秒后切换到了其他标签页12% 的用户完全关闭了页面。核心问题不在于 18 秒太长而在于 18 秒内用户完全不知道 AI 在做什么。没有任何进度反馈的等待期会触发用户的放弃心理——与其说用户不能等待不如说用户不能在没有信号的情况下等待。流式响应的价值不是减少总耗时而是将等待从不确定的黑箱变为可见的渐进过程。当用户看到逐字生成的文本时即使总时长不变感知等待时间降低约 40%。二、SSE 流式传输的协议设计流式响应的核心技术是 Server-Sent EventsSSE对比 WebSocket 更轻量、单向推送、浏览器原生支持自动重连sequenceDiagram participant Browser as 浏览器 participant API as API 网关 participant LLM as 大模型服务 Browser-API: POST /api/chat (stream: true) API-LLM: 发起流式请求 LLM--API: data: {token: 根据}br/逐 Token 返回 API--Browser: data: {delta: 根据} LLM--API: data: {token: 合同} API--Browser: data: {delta: 合同} LLM--API: data: {token: 第} API--Browser: data: {delta: 第} Note over Browser: 文本实时追加到界面br/用户看到逐字生成 LLM--API: data: [DONE] API--Browser: data: [DONE] Browser-Browser: 关闭连接标记完成关键设计在于API 网关在浏览器和大模型之间充当流式中转。浏览器端使用 EventSource API 或 fetch ReadableStream 接收 SSE 事件每收到一个 delta 就追加到渲染缓冲区。三、前端流式消费的实现// 流式 AI 对话的完整实现 // 设计意图使用 fetch ReadableStream 消费 SSE 流 // 支持中途取消、错误恢复和断线重连 interface StreamMessage { id: string; role: assistant; content: string; // 是否已完成收到 [DONE] 信号 isComplete: boolean; } class StreamingChatClient { private abortController: AbortController | null null; async streamChat( messages: { role: string; content: string }[], onDelta: (text: string) void, onComplete: () void, onError: (error: Error) void, ): Promisevoid { // 取消之前的请求——防止快速连续发送导致多个流并存 this.abortController?.abort(); this.abortController new AbortController(); try { const response await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ messages, stream: true }), signal: this.abortController.signal, }); if (!response.ok) { throw new Error(API 请求失败: ${response.status}); } // 检查响应头是否为流式类型 const contentType response.headers.get(content-type); if (!contentType?.includes(text/event-stream)) { throw new Error(服务端未返回流式响应); } const reader response.body?.getReader(); if (!reader) { throw new Error(无法获取响应流); } const decoder new TextDecoder(); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; // 解码二进制数据为文本 buffer decoder.decode(value, { stream: true }); // SSE 事件以双换行分隔 const events buffer.split(\n\n); // 最后一个 event 可能不完整保留到下次循环 buffer events.pop() || ; for (const event of events) { if (!event.trim()) continue; // 检查是否为 [DONE] 信号 if (event.trim() data: [DONE]) { onComplete(); return; } // 解析 data: 前缀后的 JSON if (event.startsWith(data: )) { try { const json JSON.parse(event.slice(6)); if (json.delta) { onDelta(json.delta); } // 处理错误事件——服务端可能在流中间发送错误 if (json.error) { onError(new Error(json.error)); return; } } catch (parseError) { // JSON 解析失败——跳过当前事件继续处理后续 console.warn(SSE 事件解析失败:, event.slice(0, 50)); } } } } // 读取结束但未收到 [DONE]——视为异常终止 onError(new Error(流式响应异常终止)); } catch (error) { if ((error as Error).name AbortError) { // 用户主动取消——不算错误 return; } onError(error as Error); } } // 停止当前流 stop(): void { this.abortController?.abort(); } } // React Hook——将流式客户端集成到组件 function useStreamingChat() { const [messages, setMessages] useStateStreamMessage[]([]); const [isStreaming, setIsStreaming] useState(false); const clientRef useRef(new StreamingChatClient()); const sendMessage useCallback(async (content: string) { // 创建一条空的 assistant 消息——内容将逐字填充 const assistantMsg: StreamMessage { id: crypto.randomUUID(), role: assistant, content: , isComplete: false, }; setMessages(prev [...prev, assistantMsg]); setIsStreaming(true); await clientRef.current.streamChat( [...messages, { role: user, content }], // onDelta——追加文本到当前 assistant 消息 (delta) { setMessages(prev { const updated [...prev]; const lastMsg updated[updated.length - 1]; if (lastMsg lastMsg.role assistant) { updated[updated.length - 1] { ...lastMsg, content: lastMsg.content delta, }; } return updated; }); }, // onComplete——标记消息为已完成 () { setMessages(prev { const updated [...prev]; const lastMsg updated[updated.length - 1]; if (lastMsg) { updated[updated.length - 1] { ...lastMsg, isComplete: true, }; } return updated; }); setIsStreaming(false); }, // onError——在消息末尾追加错误提示 (error) { setMessages(prev { const updated [...prev]; const lastMsg updated[updated.length - 1]; if (lastMsg) { updated[updated.length - 1] { ...lastMsg, content: lastMsg.content \n\n[回复生成失败请重试], isComplete: true, }; } return updated; }); setIsStreaming(false); }, ); }, [messages]); return { messages, isStreaming, sendMessage }; }三个关键设计AbortController 控制请求取消——用户发送新消息时中止旧流防止多个流并发SSE 解析的容错——数据格式错误时跳过当前事件继续处理不因一个脏数据中断整个流错误降级——流中断时在消息末尾追加错误提示而非完全丢失已生成的内容。四、流式响应的性能权衡服务端连接数压力。SSE 是长连接1000 个并发用户意味着 1000 个 TCP 连接。对于 API 网关和反向代理需要调整连接超时和缓冲区大小配置。Nginx 的默认proxy_read_timeout是 60s超过这个时间连接被关闭会导致用户看到流中断提示。流式 Token 渲染的 DOM 开销。每次收到 delta 都更新 DOM 意味着生成 1000 Token 的回复需要 1000 次 DOM 操作。React 的批量更新可以合并在同一帧内的多次 setState但跨帧的更新仍会产生实际 DOM 操作。可以使用节流策略每 50ms 或每积累 5 个 Token 才触发一次状态更新。移动端的电池消耗。持续的 SSE 连接保持网络活跃对移动端电池有影响。当用户切换到后台时应暂停流的渲染连接保持但不更新 DOM切回前台时补上累积的内容。五、总结AI 响应流式处理的优化方向SSE 协议实现 Token 级渐进渲染核心价值是消除无反馈等待前端 ReadableStream 消费 SSE 流支持取消、错误恢复和 JSON 解析容错性能优化——Token 渲染节流、移动端后台暂停 DOM 更新、Nginx 连接超时调优。落地建议后端使用text/event-stream的 Content-Type 实现 SSE 推送前端基于 fetch ReadableStream 消费流避免过时的 EventSource API默认开启流式但对stream: false参数的请求保留非流式回退监控流中断率——如果超过 5% 的连接异常终止排查网关超时配置。