API流式输出实现：SSE与WebSocket详解

用过 ChatGPT 的人都会注意到一个细节：回答不是一次性出现的，而是一个字一个字地“吐”出来。这种体验远比“等 10 秒后一次性显示全部内容”要好——用户能立刻看到模型在工作，感知等待时间大幅缩短。这种“边生成边返回”的机制，就是流式输出（Streaming）。

在 LLM 应用中，流式输出几乎成了标配。但实现流式有两种主流方案：SSE（Server-Sent Events） 和 WebSocket。它们各有适用场景，选错了会让架构变得复杂。本文会从原理讲到代码，帮你彻底搞懂这两种方案，并演示如何用 EnlyAI 统一 API 搭建一个流式聊天接口。

一、为什么需要流式输出

先理解一个关键概念：首字延迟（Time To First Token, TTFT）。它衡量的是用户发出请求后，看到第一个字符需要等多久。

传统非流式调用的工作流程是：用户请求 → 模型生成完整回复 → 一次性返回。如果模型要生成 800 字，可能需要 8 秒，用户在这 8 秒里只能盯着 loading 转圈。而流式调用是：用户请求 → 模型每生成一个 token 就立即推送 → 用户边看边等。虽然总耗时差不多，但首字延迟从 8 秒降到了 0.3 秒，体验天差地别。

流式输出对 LLM 应用尤其重要，原因有三：

• LLM 生成是自回归的：每个 token 都依赖前一个，天然适合“边算边吐”。
• 长回复很常见：一篇 2000 字的文章，非流式要等 20 秒，用户会以为卡死了。
• 可随时中断：用户觉得方向不对可以随时停止，省 token。

二、SSE：服务器发送事件

SSE（Server-Sent Events）是基于 HTTP 的单向流式协议。服务器通过一个持久的 HTTP 连接，持续向客户端推送数据。它是 HTML5 标准的一部分，浏览器原生支持。

SSE 的核心特点：

• 单向通信：只能服务器 → 客户端，客户端不能通过同一连接发数据。
• 基于 HTTP：使用普通 HTTP 协议，走 80/443 端口，对基础设施友好。
• 自动重连：浏览器断线后会自动重连，并带上 Last-Event-ID 头。
• 文本协议：数据格式是纯文本，以 data: 开头，两个换行表示一条消息结束。

一个典型的 SSE 响应长这样：

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

data: {"token": "你"}

data: {"token": "好"}

data: {"token": "，"}

data: [DONE]

每条消息以 data: 开头，后面跟 JSON 或纯文本，两个换行符分隔。客户端用 EventSource API 接收：

// 浏览器原生 SSE 客户端
const source = new EventSource("/api/chat?msg=你好");

source.onmessage = (event) => {
    if (event.data === "[DONE]") {
        source.close();
        return;
    }
    const data = JSON.parse(event.data);
    console.log(data.token);  // 逐字输出
};

source.onerror = () => {
    source.close();
};

三、WebSocket：双向通信

WebSocket 是一种全双工通信协议，建立在 TCP 之上。连接建立后，客户端和服务器可以随时互相发消息，不受请求-响应模式限制。

WebSocket 的核心特点：

• 双向通信：客户端和服务器都能主动推送消息。
• 独立协议：握手后从 HTTP 升级为 WebSocket 协议（ws:// 或 wss://）。
• 低开销：握手后帧头很小，适合高频消息。
• 需手动管理重连：断线后不会自动恢复，要自己写重连逻辑。

WebSocket 适合需要双向实时交互的场景，比如在线协作、多人游戏、实时聊天室。但对于“客户端发一个问题、服务器流式返回答案”这种单向场景，WebSocket 有点大材小用。

四、SSE vs WebSocket：如何选择

这是开发者最常问的问题。结论是：对于 LLM 流式响应，绝大多数情况选 SSE。下面这张表帮你理清两者的差异：

维度	SSE	WebSocket
通信方向	服务器 → 客户端（单向）	双向
底层协议	HTTP	TCP（握手时借用 HTTP）
浏览器支持	原生 EventSource API	原生 WebSocket API
自动重连	是	否（需手动实现）
代理/CDN 兼容	好（就是 HTTP）	需特殊配置
适合 LLM 流式	✅ 非常适合	可以，但偏重

OpenAI 的官方 API 用的就是 SSE——你在请求里加 "stream": true，返回的就是 text/event-stream 格式。EnlyAI 完全兼容这个格式，所以接入零成本。

选型建议：如果你的场景是“用户提问 → 模型流式回答”，用 SSE。如果是“多人实时协作编辑”或“语音对话”，才考虑 WebSocket。

五、用 Python 调用流式 API

先看最基础的：用 OpenAI SDK 调用 EnlyAI 的流式接口。EnlyAI 完全兼容 OpenAI API 格式，base URL 是 https://enlyai.com/v1。

from openai import OpenAI

client = OpenAI(
    api_key="sk-your-enlyai-key",
    base_url="https://enlyai.com/v1"
)

stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "讲一个 300 字的科幻小故事"}],
    stream=True
)

for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)

对应的 cURL 命令，可以直接在终端测试流式效果：

curl https://enlyai.com/v1/chat/completions \
  -H "Authorization: Bearer sk-your-enlyai-key" \
  -H "Content-Type: application/json" \
  -N \
  -d '{
    "model": "gpt-4o-mini",
    "messages": [{"role": "user", "content": "讲一个 300 字的科幻小故事"}],
    "stream": true
  }'

-N 参数禁用 curl 的缓冲，让你能实时看到流式输出。你会看到类似这样的数据一块一块地返回：

data: {"id":"chatcmpl-xxx","choices":[{"delta":{"content":"在"},"index":0}]}

data: {"id":"chatcmpl-xxx","choices":[{"delta":{"content":"一"},"index":0}]}

data: {"id":"chatcmpl-xxx","choices":[{"delta":{"content":"个"},"index":0}]}

data: [DONE]

六、用 FastAPI 搭建 SSE 流式接口

实际项目中，你通常需要在自己的后端包一层，把 LLM 的流式输出转发给前端。FastAPI 对 SSE 有很好的支持，下面是一个完整示例：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from openai import OpenAI
import json

app = FastAPI()
client = OpenAI(
    api_key="sk-your-enlyai-key",
    base_url="https://enlyai.com/v1"
)

class ChatRequest(BaseModel):
    message: str
    model: str = "gpt-4o-mini"

def sse_format(data: str) -> str:
    """把数据包装成 SSE 格式"""
    return f"data: {data}\n\n"

def stream_generator(message: str, model: str):
    """调用 EnlyAI 流式接口并转发"""
    stream = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": message}],
        stream=True
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            yield sse_format(json.dumps({"token": delta}))
    yield sse_format("[DONE]")

@app.post("/api/chat")
async def chat(req: ChatRequest):
    return StreamingResponse(
        stream_generator(req.message, req.model),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",  # 禁用 Nginx 缓冲
        }
    )

启动服务后，前端用 EventSource 或 fetch 就能消费这个接口。注意 X-Accel-Buffering: no 这个 header 很关键——如果后端前面有 Nginx，不加这个会导致 Nginx 把流式数据缓冲成一整块再转发，流式效果就没了。

七、前端消费 SSE 流

浏览器原生的 EventSource 只支持 GET 请求，而我们的接口是 POST。这时可以用 fetch 配合 ReadableStream 手动解析 SSE：

async function streamChat(message) {
    const response = await fetch("/api/chat", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ message })
    });

    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 = line.slice(6);
            if (data === "[DONE]") return;
            const { token } = JSON.parse(data);
            console.log(token);  // 逐字输出到页面
        }
    }
}

这段代码的核心是：用 ReadableStream 读取响应体，按 \n\n 分割成完整的 SSE 事件，再解析每个事件的 data 字段。

八、WebSocket 实现示例

虽然 LLM 流式更推荐 SSE，但如果你确实需要双向通信（比如语音对话、实时协作），WebSocket 也能做。下面用 FastAPI 的 WebSocket 支持演示：

from fastapi import FastAPI, WebSocket
from openai import OpenAI
import json

app = FastAPI()
client = OpenAI(
    api_key="sk-your-enlyai-key",
    base_url="https://enlyai.com/v1"
)

@app.websocket("/ws/chat")
async def websocket_chat(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            # 接收客户端消息
            data = await ws.receive_json()
            message = data.get("message", "")

            # 调用 EnlyAI 流式接口
            stream = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": message}],
                stream=True
            )
            for chunk in stream:
                delta = chunk.choices[0].delta.content
                if delta:
                    await ws.send_json({"token": delta})
            await ws.send_json({"done": True})
    except Exception:
        pass
    finally:
        await ws.close()

WebSocket 的优势在这个场景体现得很明显：连接建立后，客户端可以连续发多条消息，不用每次都重新建连。但代价是你要自己处理心跳、重连、消息序号等细节。

九、生产环境注意事项

把流式接口推向生产，有几个坑必须提前知道：

1. 禁用反向代理缓冲：Nginx 默认会缓冲响应体，必须在 location 里加 proxy_buffering off;，或在响应头加 X-Accel-Buffering: no，否则流式会变成“一次性返回”。
2. 设置合理的超时：LLM 生成长文本可能需要 30-60 秒，Nginx 的 proxy_read_timeout 默认 60 秒，长回复可能超时，建议调到 120 秒以上。
3. 处理客户端断开：用户关掉页面后，后端要能感知并停止调用 LLM，否则会白白烧 token。FastAPI 里可以用 asyncio.CancelledError 捕获。
4. 用异步提升并发：同步 SDK 会阻塞事件循环，高并发场景务必用 AsyncOpenAI 和 async for。
5. 前端做断线重连：SSE 虽然自动重连，但重连后要能接着上次的断点继续，可以用 Last-Event-ID 实现。

十、用 EnlyAI 统一 API 实现流式

前面所有代码都用了 gpt-4o-mini，但 EnlyAI 的价值在于一套接口、多种模型。你只需要改 model 字段，就能在 GPT、Claude、Gemini 之间切换，流式格式完全一致：

from openai import OpenAI

client = OpenAI(
    api_key="sk-your-enlyai-key",
    base_url="https://enlyai.com/v1"
)

# 同一套流式代码，只改 model 即可切换
for model in ["gpt-4o-mini", "claude-3-5-sonnet", "gemini-2.0-flash"]:
    print(f"\n=== {model} ===")
    stream = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": "用一句话介绍自己"}],
        stream=True
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            print(delta, end="", flush=True)

统一 API 的好处在流式场景下更加明显：

• 故障转移：某个模型流式服务波动时，改一个字符串就能切到备用模型，前端代码不用动。
• 统一计费：所有模型的流式调用汇总在一个账单，token 消耗一目了然。
• 一致的开发体验：不用为每个模型学一套流式 SDK，OpenAI 格式通吃。

总结

流式输出是 LLM 应用的体验分水岭。在技术选型上，记住一个原则：单向流式选 SSE，双向实时选 WebSocket。对于绝大多数“用户提问、模型流式回答”的场景，SSE 是更轻量、更兼容、更省心的选择——这也是 OpenAI 和 EnlyAI 官方 API 采用的方案。

实现层面，FastAPI + OpenAI SDK（指向 EnlyAI）+ 前端 fetch/ReadableStream，是一套经过验证的成熟组合。把 Nginx 缓冲关掉、超时调大、断开处理好，你的流式聊天接口就能稳定上线了。