热点

AI SCORE

技术实践2026-08-29 23:18

SSE 流式调用常见的性能误区

dev.to · AI#SSE#性能优化#流式处理

Editor brief · 编辑速览

澄清 WebSocket vs SSE 的适用场景，以及 fetch+res.json() 实际是缓冲而非流式，需要用 body.getReader() 分块读取。

文章思维导图

Knowledge map

拖拽缩放

Full translation

完整中文译文

你怪免费模型服务器。我能理解。一半的时候，服务器没问题。是你的客户端代码在吞噬 token。我们来聊聊流式传输。具体来说，是关于 Server-Sent Events (SSE) 的误区。

误区一：SSE 就是 WebSocket 上的 JSON

到处都是这种说法。人们说："流式传输需要 WebSocket。是双向的，对吧？"

SSE 是 HTTP。MIME 类型是 text/event-stream。WebSocket 需要协议升级。SSE 可以直接在普通 HTTP 上工作。

单向。服务器到客户端。就这样。

对于 AI 对话，这正是你需要的。你把 prompt 发上去，模型把响应流下来。

在这里使用 WebSocket 增加了开销。你要管理：ws、onopen、onmessage。为什么？HTTP 本身就很好用。

误区二：await res.json() 适合流式传输

这是最大的一个误区。

当你用 fetch 时，数据是一个流。调用 res.json() 会在解析前缓冲整个响应。

你不是在流式传输。你是在等整个文件下载完。

解决办法？一块一块地读取 body 流。

const resp = await fetch("/v1/chat/completions", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify(payload),
});

if (!resp.ok) throw new Error(`HTTP ${resp.status}`);

const reader = resp.body.getReader();
const decoder = new TextDecoder();
let buffer = "";

while (true) {
  const { value, done } = await reader.read();
  if (done) break;

  buffer += decoder.decode(value, { stream: true });
  const events = buffer.split("\n\n");

  // Last chunk might be incomplete. Keep it in the buffer.
  buffer = events.pop();

  for (const evt of events) {
    if (!evt.startsWith("data:")) continue;
    const raw = evt.slice(5).trim();
    if (raw === "[DONE]") continue;

    const json = JSON.parse(raw);
    const token = json.choices?.[0]?.delta?.content || "";
    process.stdout.write(token);
  }
}

你看到区别了吗？

第一个 token 出现在屏幕上。它不需要等待最后一个 token。

误区三：Nginx 反向代理和边缘缓存缓冲区是个谜

你的代理可能在往磁盘缓冲所有内容后再发送。

Nginx 有一个叫 X-Accel-Buffering 的特性。FastAPI、Node 或你的反向代理都信任它。如果你的代理缓冲了数据块，你的同事们就遭殃了。

如果你的第一个字节刚好落在 500ms 的边界上，你的低 TTFB 毫无意义。

在流式传输之前尝试添加这个 header：

res.setHeader("X-Accel-Buffering", "no");
res.setHeader("Cache-Control", "no-cache");
res.setHeader("Connection", "keep-alive");

如果你用 FastAPI，你还需要 StreamingResponse：

from fastapi.responses import StreamingResponse

def stream_result(prompt: str):
    for token in call_free_model(prompt):
        yield token

@app.post("/v1/chat/completions")
def chat_route(payload: dict):
    return StreamingResponse(
        stream_result(payload["messages"]),
        media_type="text/event-stream",
        headers={"X-Accel-Buffering": "no"},
    )

如果你省略了这个 header，你的模型会显得很慢。它从来不慢。只是你的基础设施在说："等一下，让我先缓冲一下。"

误区四：curl 显示真实行为

我有个老习惯。我运行这个：

curl -N https://api.example.com/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"model":"free","stream":true}'

看到 -N 标志了吗？那会禁用缓冲。

很多人会省略它。然后 curl 就会缓冲。然后他们告诉我 API 很慢。

再说一遍：分清服务器行为和客户端行为至关重要。

这里有一个小型可执行脚本。它测量首 token 时间和完整响应时间的对比。

import time
import httpx
import statistics

def time_to_first_token(url, payload, runs=5):
    first = []
    total = []

    for _ in range(runs):
        start = time.monotonic()
        with httpx.stream("POST", url, json=payload, timeout=60) as resp:
            seen_first = False
            for line in resp.iter_lines():
                if line.startswith("data:") and not seen_first:
                    first.append(time.monotonic() - start)
                    seen_first = True
        total.append(time.monotonic() - start)

    return statistics.median(first), statistics.median(total)

# Example
url = "http://localhost:8000/v1/chat/completions"
payload = {"model": "my-free-model", "stream": True, "messages": [{"role": "user", "content": "hi"}]}

first_ms, total_ms = time_to_first_token(url, payload)
print(f"First token (median): {first_ms * 1000:.1f} ms")
print(f"Full output (median): {total_ms * 1000:.1f} ms")

如果 first_ms 在 80-120ms 范围内但 total_ms 是 8 秒，你的服务器实际上在流式传输。用户觉得慢是因为模型花了 7 秒生成 token。

MonkeyCode 代理检查

声明：本文是 MonkeyCode 产品推广的一部分。

当我针对免费模型端点运行这个时，让代理参与其中很有帮助。MonkeyCode 的免费服务器选项让你可以启动一个小代理。那个代理可以按设计隐藏模型端点。

但如果你设置了那个代理，要尊重流式路径。不要让你的代理控制器收集完整 payload 然后再转发。这是一个非常常见的失败：你的云函数是不可见的，但它会杀死流式传输。

我保持代理文件简单：

service: proxy
build:
  command: python -m app

再说一遍，不要添加会缓冲的中间层。你的免费模型 API 会变成一个巨大的谎言。端点响应 200ms，但你的网关显示 9 秒。

谁应该忽略这个建议

不是每个人都需要追求首 token 延迟。

如果你在写一个处理批处理作业的 CLI 工具，老实调用 res.json()。它更容易阅读，更容易维护。

如果你在构建一个有进度条或聊天气泡的 UI，你需要流式传输。你需要首 token 可观测性。

如果你在构建调试测试套件，写我上面分享的脚本。把它放在你的 CI 里。保护你的代理免受缓冲回归。

正确的思维模型

每个 SSE 增量都是一个 TCP 数据包。

每个缓冲区都是一个即将发生的谎言。

你的免费模型不是你的车。它是一个中间人。如果你用错误的方式发送响应，中间人就是交通堵塞。

在 curl 中使用 -N。在浏览器中使用 getReader()。在 Python 中使用 iter_lines()。在你的网关上强制使用 X-Accel-Buffering: no。

然后你就可以诚实地抱怨模型了——因为有时候它确实很慢。

但至少你会有数据来证明它。