_{🧠 记忆锚点：只需服务端持续推送选 SSE；双方频繁实时互动选 WebSocket。}

💡 答案要点

两种协议核心对比：

维度	SSE（Server-Sent Events）	WebSocket
通信方向	单向（服务端 → 客户端）	双向（全双工）
协议	HTTP/1.1	WS（基于 TCP 升级）
连接开销	低（复用 HTTP）	中（需握手升级）
浏览器支持	原生支持，自动重连	原生支持，需手动重连
负载均衡	友好（标准 HTTP）	需要粘性会话（sticky session）
适用场景	LLM Token 流式输出	实时双向交互、语音对话

选型决策树：

需要客户端实时向服务端发送消息（非HTTP请求）？
├── 是 → WebSocket
│         场景：语音助手、用户打断Agent、实时协作
└── 否
    需要服务端持续推送数据流？
    ├── 是 → SSE
    │         场景：LLM Token 流式输出、进度推送
    └── 否 → 普通 HTTP 轮询即可

SSE 实现（LLM 流式输出最佳选择）：

展开 Python 代码示例（30 行）

# FastAPI SSE 示例
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import openai

app = FastAPI()

@app.post("/chat/stream")
async def chat_stream(query: str):
    async def generate():
        client = openai.AsyncOpenAI()
        stream = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": query}],
            stream=True
        )
        async for chunk in stream:
            if chunk.choices[0].delta.content:
                # SSE 格式：data: xxx\n\n
                yield f"data: {chunk.choices[0].delta.content}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no"  # 禁用 Nginx 缓冲
        }
    )

WebSocket 实现（需要双向交互）：

# FastAPI WebSocket 示例（支持用户打断 Agent）
from fastapi import WebSocket

@app.websocket("/agent/ws")
async def agent_websocket(websocket: WebSocket):
    await websocket.accept()
    agent_task = None

    while True:
        # 同时监听用户消息和 Agent 输出
        data = await websocket.receive_text()
        msg = json.loads(data)

        if msg["type"] == "user_message":
            # 取消当前 Agent 任务（用户打断）
            if agent_task and not agent_task.done():
                agent_task.cancel()

            # 启动新 Agent 任务
            agent_task = asyncio.create_task(
                run_agent_and_stream(msg["content"], websocket)
            )

        elif msg["type"] == "interrupt":
            # 用户主动打断
            if agent_task:
                agent_task.cancel()
            await websocket.send_json({"type": "interrupted"})

面试话术：

"选 SSE 还是 WebSocket 看交互模式。LLM Token 流式输出用 SSE 最合适——单向推送、HTTP 友好、负载均衡无压力，Claude、ChatGPT 都用 SSE。但 Agent 场景需要用户'打断'或实时发工具执行结果时，就要 WebSocket——全双工，支持服务端推 + 客户端随时插话。我的原则：能用 SSE 就用 SSE，只有真正需要双向实时通信才上 WebSocket，因为 WebSocket 的粘性会话会让运维复杂很多。"

---