LLM 流式输出实现：SSE 与 WebSocket
为什么需要流式输出 LLM 生成完整回答可能需要 5-15 秒。用户盯着空白屏幕等待体验极差。流式输出让用户像看打字机一样实时看到生成内容，首 token 延迟降到 200-500ms，体感速度提升 10 倍。 SSE vs WebSocket 对比 维度 SSE (Server-Sent Events) WebSocket 协议 HTTP/1.1 或 HTTP/2 独立协议(ws://) 方向 服务器到客户端(单向) 双向 自动重连 浏览器内置 需手动实现 代理兼容性 好(就是HTTP) 差(需特殊配置) 适合场景 LLM 流式输出(推荐) 实时对话/多Agent通信 实现复杂度 低 中 结论：LLM 流式输出首选 SSE，需要双向通信才用 WebSocket。 SSE 后端实现（FastAPI） from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import httpx, json, asyncio app = FastAPI() LLM_API_URL = "https://api.openai.com/v1/chat/completions" @app.post("/api/chat/stream") async def chat_stream(request: Request): body = await request.json() async def event_generator(): headers = { "Authorization": f"Bearer {body['api_key']}", "Content-Type": "application/json" } payload = { "model": body["model"], "messages": body["messages"], "stream": True, "max_tokens": 4096 } try: async with httpx.AsyncClient(timeout=120) as client: async with client.stream("POST", LLM_API_URL, headers=headers, json=payload) as resp: async for line in resp.aiter_lines(): if not line.startswith("data: "): continue data = line[6:] if data.strip() == "[DONE]": yield "data: " + json.dumps({"type": "done"}) + "\n\n" break chunk = json.loads(data) delta = chunk["choices"][0]["delta"] if "content" in delta: token = delta["content"] yield "data: " + json.dumps({"type": "token", "content": token}) + "\n\n" except httpx.ReadTimeout: yield "data: " + json.dumps({"type": "error", "msg": "LLM timeout"}) + "\n\n" except Exception as e: yield "data: " + json.dumps({"type": "error", "msg": str(e)}) + "\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive"} ) SSE 前端实现 class SSEClient { constructor(url) { this.url = url; this.controller = null; } async stream(messages, onToken, onDone, onError) { this.controller = new AbortController(); try { const resp = await fetch(this.url, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ messages }), signal: this.controller.signal }); const reader = resp.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'); buffer = lines.pop(); for (const line of lines) { if (!line.startsWith('data: ')) continue; const data = JSON.parse(line.slice(6)); if (data.type === 'token') onToken(data.content); else if (data.type === 'done') onDone(); else if (data.type === 'error') onError(data.msg); } } } catch (e) { if (e.name !== 'AbortError') onError(e.message); } } abort() { this.controller?.abort(); } } // 使用 const client = new SSEClient('/api/chat/stream'); let fullText = ''; await client.stream( [{ role: 'user', content: '解释 RAG 架构' }], (token) => { fullText += token; renderMarkdown(fullText); }, () => console.log('完成'), (err) => console.error('错误:', err) ); WebSocket 方案 需要双向通信（用户中途打断、实时修正）时使用： ...