TL;DR：AI 应用需要实时展示生成结果（打字机效果）。方案有两个：WebSocket 和 Server-Sent Events（SSE）。本文用真实案例对比两者的适用场景，并给出 FastAPI + 前端的完整实现代码。

1. 先说结论

✅ WebSocket 适用

• AI Agent 多轮对话（双向通信）
• 需要服务器主动推送的场景
• 高频数据交换（游戏、协作工具）

✅ SSE 适用

• AI 流式输出（打字机效果）
• 服务器单向推送（通知、日志）
• 实现简单，无需 WebSocket 握手

2. 为什么 AI 应用选 SSE 而不是 WebSocket

2.1 AI 流式输出的原理

大模型生成回答是一个「流」的过程：

流式输出的数据格式

# 服务器逐块返回（每个 chunk 是一个 token） data: {"content": "今"} data: {"content": "天"} data: {"content": "天"} data: {"content": "气"} data: {"content": "很"} data: {"content": "好"} data: {"content": "。"} [DONE]

SSE 天然适合这种场景——每个 `data:` 就是一个数据块，浏览器收到后直接渲染到页面上，实现「打字机效果」。

2.2 用 SSE 实现 AI 流式输出

Python - FastAPI SSE 后端

from fastapi import FastAPI from fastapi.responses import StreamingResponse import openai import asyncio import json app = FastAPI() client = openai.OpenAI( api_key="your-api-key", base_url="https://api.siliconflow.cn/v1" ) async def stream_openai_response(prompt: str): """将 OpenAI 的流式响应转为 SSE 格式""" stream = client.chat.completions.create( model="deepseek-ai/DeepSeek-V3", messages=[{"role": "user", "content": prompt}], stream=True ) for chunk in stream: if chunk.choices[0].delta.content: content = chunk.choices[0].delta.content # SSE 格式：data: {"content": "xxx"}\n\n yield f"data: {json.dumps({'content': content})}\n\n" # 发送结束信号 yield "data: [DONE]\n\n" @app.get("/stream-chat") async def stream_chat(prompt: str): """SSE 流式聊天接口""" return StreamingResponse( stream_openai_response(prompt), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no" # 关闭 Nginx 缓冲 } )

JavaScript - 前端 SSE 接收

async function streamChat(prompt) { const response = await fetch( `http://localhost:8000/stream-chat?prompt=${encodeURIComponent(prompt)}` ); const reader = response.body.getReader(); const decoder = new TextDecoder(); let answer = ""; while (true) { const { done, value } = await reader.read(); if (done) break; const chunk = decoder.decode(value); const lines = chunk.split('\n'); for (const line of lines) { if (line.startsWith('data: ')) { const data = line.slice(6); if (data === '[DONE]') { console.log('完成！'); return; } const parsed = JSON.parse(data); answer += parsed.content; // 打字机效果：实时更新页面 document.getElementById('output').textContent = answer; } } } } // 使用示例 streamChat('解释一下什么是 REST API');

3. 用 WebSocket 实现 AI Agent 对话

SSE 只能单向推送。如果你要做 AI Agent——用户可以随时打断、发送新消息、调用工具——需要双向通信，用 WebSocket。

Python - FastAPI WebSocket 服务

from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse import openai import json app = FastAPI() client = openai.OpenAI(api_key="your-api-key", base_url="https://api.siliconflow.cn/v1") class ConnectionManager: def __init__(self): self.active_connections: list[WebSocket] = [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) manager = ConnectionManager() @app.websocket("/ws/chat") async def websocket_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: # 接收用户消息 data = await websocket.receive_text() message = json.loads(data) # 处理不同类型的消息 if message["type"] == "chat": # 流式回复 stream = client.chat.completions.create( model="deepseek-ai/DeepSeek-V3", messages=[{"role": "user", "content": message["content"]}], stream=True ) for chunk in stream: if chunk.choices[0].delta.content: await websocket.send_json({ "type": "token", "content": chunk.choices[0].delta.content }) # 发送结束信号 await websocket.send_json({"type": "done"}) elif message["type"] == "tool_call": # 处理工具调用（如搜索、计算等） result = await handle_tool_call(message) await websocket.send_json({ "type": "tool_result", "result": result }) except WebSocketDisconnect: manager.disconnect(websocket)

JavaScript - 前端 WebSocket 客户端

class AIChatAgent { constructor(wsUrl) { this.ws = new WebSocket(wsUrl); this.onToken = null; this.onToolResult = null; this.ws.onmessage = (event) => { const msg = JSON.parse(event.data); if (msg.type === 'token') { this.onToken && this.onToken(msg.content); } else if (msg.type === 'done') { console.log('生成完成'); } else if (msg.type === 'tool_result') { this.onToolResult && this.onToolResult(msg.result); } }; } sendMessage(content) { this.ws.send(JSON.stringify({ type: 'chat', content })); } callTool(toolName, params) { this.ws.send(JSON.stringify({ type: 'tool_call', tool: toolName, params })); } } // 使用示例 const agent = new AIChatAgent('ws://localhost:8000/ws/chat'); let answer = ''; agent.onToken = (token) => { answer += token; document.getElementById('output').textContent = answer; }; agent.sendMessage('帮我查一下今天的天气');

4. 实战场景对比

场景 1：AI 写作助手（打字机效果）

用户输入一段文字，AI 实时续写。

• 推荐方案：SSE
• 单向通信足够了（用户发一次，AI 流式输出）
• 实现简单，浏览器自动重连

实测延迟：SSE 平均延迟 80ms，WebSocket 平均延迟 75ms——对于打字机效果，两者在体感上没有区别，但 SSE 实现更简单。

场景 2：AI 客服机器人（多轮对话）

• 推荐方案：WebSocket
• 用户可以随时打断 AI（发送新问题）
• 需要双向通信
• AI 可能调用工具（搜索、查数据库）需要主动推送结果

场景 3：AI 代码生成（带进度展示）

• 推荐方案：WebSocket
• AI 生成代码时，需要展示「正在写 X 模块」
• 写完后需要自动运行测试
• 测试结果需要主动推送给前端

5. 混合方案：SSE + WebSocket 并存

实际项目中，两个方案可以并存——WebSocket 用于 Agent 主对话，SSE 用于辅助任务（如日志推送、进度更新）。

FastAPI 路由设计示例

# 主对话：用 WebSocket（双向） @app.websocket("/ws/agent") async def agent_ws(websocket: WebSocket): """AI Agent 主对话入口""" pass # 辅助任务：用 SSE（单向） @app.get("/sse/logs/{task_id}") async def task_logs(task_id: str): """任务日志流（SSE）""" async def log_generator(): while True: log = await get_task_log(task_id) yield f"data: {json.dumps(log)}\n\n" await asyncio.sleep(1) return StreamingResponse(log_generator(), media_type="text/event-stream") @app.get("/sse/progress/{task_id}") async def task_progress(task_id: str): """任务进度流（SSE）""" async def progress_generator(): while True: progress = await get_task_progress(task_id) yield f"data: {json.dumps(progress)}\n\n" await asyncio.sleep(0.5) return StreamingResponse(progress_generator(), media_type="text/event-stream")

6. 总结

选型口诀：「单向用 SSE，双向用 WebSocket，简单查询用 REST。」AI 写作助手 / 进度展示 → SSE；AI Agent / 实时协作 → WebSocket。

如果对你有帮助，欢迎在评论区聊聊你的实时通信方案选型经验。