TL;DRAI 应用需要实时展示生成结果打字机效果。方案有两个WebSocket 和 Server-Sent EventsSSE。本文用真实案例对比两者的适用场景并给出 FastAPI 前端的完整实现代码。1. 先说结论✅ WebSocket 适用AI Agent 多轮对话双向通信需要服务器主动推送的场景高频数据交换游戏、协作工具✅ SSE 适用AI 流式输出打字机效果服务器单向推送通知、日志实现简单无需 WebSocket 握手对比维度WebSocketSSE通信方向全双工双向半双工单向协议复杂度需要握手、心跳基于 HTTP简单断线重连需要手动处理浏览器自动重连兼容性所有浏览器IE 不支持AI 流式输出需要额外处理✅ 原生支持适用 AI 场景Agent 对话、工具调用流式回答、进度展示实现难度⭐⭐⭐⭐⭐⭐2. 为什么 AI 应用选 SSE 而不是 WebSocket2.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_keyyour-api-key, base_urlhttps://api.siliconflow.cn/v1 ) async def stream_openai_response(prompt: str): 将 OpenAI 的流式响应转为 SSE 格式 stream client.chat.completions.create( modeldeepseek-ai/DeepSeek-V3, messages[{role: user, content: prompt}], streamTrue ) for chunk in stream: if chunk.choices[0].delta.content: content chunk.choices[0].delta.content # SSE 格式data: {content: xxx}\n\n yield fdata: {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_typetext/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_keyyour-api-key, base_urlhttps://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( modeldeepseek-ai/DeepSeek-V3, messages[{role: user, content: message[content]}], streamTrue ) 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. 实战场景对比场景 1AI 写作助手打字机效果用户输入一段文字AI 实时续写。推荐方案SSE单向通信足够了用户发一次AI 流式输出实现简单浏览器自动重连实测延迟SSE 平均延迟 80msWebSocket 平均延迟 75ms——对于打字机效果两者在体感上没有区别但 SSE 实现更简单。场景 2AI 客服机器人多轮对话推荐方案WebSocket用户可以随时打断 AI发送新问题需要双向通信AI 可能调用工具搜索、查数据库需要主动推送结果场景 3AI 代码生成带进度展示推荐方案WebSocketAI 生成代码时需要展示「正在写 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 fdata: {json.dumps(log)}\n\n await asyncio.sleep(1) return StreamingResponse(log_generator(), media_typetext/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 fdata: {json.dumps(progress)}\n\n await asyncio.sleep(0.5) return StreamingResponse(progress_generator(), media_typetext/event-stream)6. 总结场景推荐方案理由AI 流式回答打字机效果✅ SSE单向够用实现简单AI Agent 多轮对话✅ WebSocket需要双向通信 工具调用任务进度 / 日志推送✅ SSE单向推送浏览器自动重连代码协作多人同时编辑✅ WebSocket高频双向同步简单 AI 查询问一句答一句✅ REST API不需要实时通信选型口诀「单向用 SSE双向用 WebSocket简单查询用 REST。」AI 写作助手 / 进度展示 → SSEAI Agent / 实时协作 → WebSocket。如果对你有帮助欢迎在评论区聊聊你的实时通信方案选型经验。