LLM 流式响应实现指南：SSE vs WebSocket vs gRPC
为什么需要流式响应 LLM 生成一个完整回答可能需要 5-15 秒。如果等全部生成再返回，用户体验极差。流式响应让用户在第一个 token 生成时就能看到输出，体感延迟从"等待 10 秒"变成"等待 0.5 秒"。 非流式: [████████████████████████████] 10s → 一次性返回 流式: [█][█][█][█][█][█][█][█][█][█] 0.5s 首字 + 持续输出 三种流式协议对比 特性 SSE WebSocket gRPC Stream 方向 服务端→客户端 双向 双向 底层 HTTP/1.1 或 HTTP/2 HTTP 升级 HTTP/2 浏览器支持 原生 EventSource 原生 WebSocket 需要 gRPC-Web 自动重连 内置 需手动实现 需手动实现 代理友好 非常好 一般 差（需 HTTP/2） 适用场景 LLM 流式输出 实时对话 + 语音 内部微服务 结论：面向用户的 LLM 流式输出，SSE 是首选。需要双向交互（如语音对话）用 WebSocket。内部服务间通信用 gRPC。 SSE 实现 后端：FastAPI + OpenAI Streaming from fastapi import FastAPI from fastapi.responses import StreamingResponse from openai import AsyncOpenAI import json import asyncio app = FastAPI() client = AsyncOpenAI() @app.post("/api/chat/stream") async def chat_stream(request: dict): async def event_generator(): try: stream = await client.chat.completions.create( model="gpt-4o-mini", messages=request["messages"], stream=True, temperature=0.7, ) async for chunk in stream: if chunk.choices[0].delta.content: data = json.dumps({ "content": chunk.choices[0].delta.content, "role": "assistant" }) yield f"data: {data}\n\n" yield f"data: {json.dumps({'done': True})}\n\n" except Exception as e: yield f"data: {json.dumps({'error': str(e)})}\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", # nginx: 禁用缓冲 } ) 前端：EventSource + ReadableStream // 方案1: 使用 fetch + ReadableStream (支持 POST) async function streamChat(messages: Message[]) { const response = await fetch('/api/chat/stream', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ messages }), }); 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 }); const lines = buffer.split('\n\n'); buffer = lines.pop() || ''; for (const line of lines) { if (!line.startsWith('data: ')) continue; const data = JSON.parse(line.slice(6)); if (data.done) { console.log('Stream completed'); break; } if (data.error) { console.error('Stream error:', data.error); break; } // 追加到 UI appendToUI(data.content); } } } // 方案2: 使用 EventSource (仅支持 GET) const evtSource = new EventSource('/api/chat/stream?prompt=hello'); evtSource.onmessage = (e) => { const data = JSON.parse(e.data); if (data.done) evtSource.close(); else appendToUI(data.content); }; evtSource.onerror = (e) => { console.error('SSE error, will auto-reconnect'); }; SSE 断线重连 SSE 原生支持自动重连，但需要服务端配合发送 retry 和 id 字段： ...
