
流式响应开发指南:SSE、WebSocket 与 AI 对话体验优化
为什么必须流式 AI 生成一个完整回答可能需要 5-30 秒。如果等全部生成完再返回,用户体验极差。流式输出让用户看到"AI 正在打字",将感知等待时间从 15 秒降到 0.5 秒。 SSE vs WebSocket 维度 SSE WebSocket 协议 HTTP 独立协议 方向 服务端→客户端(单向) 双向 重连 自动 需手动实现 兼容性 所有现代浏览器 所有现代浏览器 复杂度 低 中 适用场景 AI 回复流式输出 实时协作、语音对话 结论:AI 对话场景用 SSE 足够,语音/多模态交互用 WebSocket。 SSE 实现 服务端(Python FastAPI) from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio app = FastAPI() @app.post("/api/chat") async def chat(request: dict): async def stream(): # 调用 LLM 流式 API async for chunk in llm.stream( model="gpt-4o", messages=request["messages"] ): # SSE 格式:data: <content>\n\n content = chunk.choices[0].delta.content if content: yield f"data: {json.dumps({'content': content})}\n\n" # 发送结束标记 yield f"data: {json.dumps({'done': True})}\n\n" return StreamingResponse( stream(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", # Nginx 不缓冲 } ) 客户端(JavaScript) class AIChat { async sendMessage(text) { const response = await fetch('/api/chat', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ messages: [...this.history, { role: 'user', content: text }] }) }); const reader = response.body.getReader(); 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 lines = buffer.split('\n\n'); buffer = lines.pop(); // 保留不完整的行 for (const line of lines) { if (line.startsWith('data: ')) { const data = JSON.parse(line.slice(6)); if (data.done) { this.onComplete(); } else { this.onToken(data.content); } } } } } onToken(content) { document.getElementById('response').textContent += content; } } 进阶:流式 + 工具调用 当 Agent 需要调用工具时,流式输出变得复杂: ...


