你的 LLM 流在本地 localhost 正常，到生产环境就挂——SSE 端到端修复指南

流式输出 LLM 响应看起来是你产品中最简单的功能。SDK 给你一个迭代器，你打印 token，在笔记本上不到一分钟就能跑通。

然后你部署上去，就会发生以下情况之一：响应在 40 秒后一次性到达，而不是逐 token 输出；或者连接在半途断开，日志里没有任何错误；或者用户关闭了标签页，但你仍在为 60000 个永远不会被读取的 token 付费。

这些都不是模型问题。它们是发生在 api.anthropic.com 和浏览器标签页之间的六个问题——代理、负载均衡器、HTTP 客户端和 JavaScript API 都会对长连接有自己的处理逻辑。

我在 Cursuri-AI.ro（一个东欧的 AI 教育平台）教 AI 工程，这是我在生产审查中最常遇到的故障面——因为每一层单独看都没问题。以下是整条链路的完整分析，一次一个故障点。

首先：流式输出不再只是用户体验的锦上添花

值得明确你为什么要这样做，因为这改变了你对待失败的方式。

关于感知延迟的论证很有名，而且确实有效。但在当前模型上，流式输出对于越来越多的请求来说也是一个正确性要求。Claude Opus 5、Sonnet 5 和 4.6/4.7/4.8 系列支持高达 128K 的输出 token，而 Anthropic 的 SDK 会拒绝它认为会超过连接容忍度的非流式请求——Python SDK 会抛出 ValueError，而不是让你构建一个会挂起和丢弃的东西。默认客户端超时是 10 分钟（注意不同 SDK 的单位不同：Python 和 Ruby 是秒，TypeScript 是毫秒），而在 Opus 5 上 thinking 默认开启，一个困难任务可能在第一个可见字符生成前花费数分钟。

所以：任何具有大 max_tokens、长输入或重推理提示的请求都是流式请求。不是为了那个动画效果——而是为了连接。

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-opus-5",
    max_tokens=64000,
    thinking={"type": "adaptive", "display": "summarized"},
    messages=[{"role": "user", "content": "Analyze this incident report..."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

final = stream.get_final_message()

这段代码中有一处不是装饰性的。在 Opus 5、Opus 4.8/4.7、Fable 5 和 Sonnet 5 上，thinking display 默认是 "omitted"——thinking 块打开，发送一个 signature_delta，然后关闭，中间没有 thinking_delta 事件。如果你在流式传输推理过程给用户看但保留默认设置，你的 UI 会显示一个漫长的死机然后突然出现一堵文字墙。display: "summarized" 是让你在思考期间有东西可渲染。它不需要额外成本：thinking 发生并以相同方式计费，无论 display 设置如何。

故障点 #1：你的代理扣押了 token

这是"流式输出在生产环境不工作"的头号 bug，特征非常明显：本地你看到 token 一个一个出现；部署后，整个响应在最后一次性到达。

你的应用没有任何问题。是 nginx 在做缓冲。从 nginx 文档来看，proxy_buffering 默认为 on，当启用缓冲时，nginx 会将上游的响应读入自己的缓冲区，然后再转发。当它 off 时，"响应会同步传递给客户端，立即收到立即转发。"

有两种修复方法，你要用第二种。

location /api/chat/stream {
    proxy_pass http://app;
    proxy_buffering off;
    proxy_read_timeout 300s;   # default is 60s
}

这可以工作，但这把基础设施规则写到了一个应用程序团队不拥有的文件中。nginx 还支持响应头："Buffering can also be enabled or disabled by passing yes or no in the X-Accel-Buffering response header field." 所以流式端点可以自己关闭缓冲：

from fastapi.responses import StreamingResponse

def sse_response(generator):
    return StreamingResponse(
        generator,
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",   # the important one
        },
    )

这个头跟着端点走。添加一个路由，得到正确的行为。没有配置漂移，没有"六个月后有人重建 ingress 时 staging 环境正常但生产环境故障"的情况。

当你在这里时，注意 proxy_read_timeout——它在 nginx 中默认是 60 秒。一个模型在第一个 token 前硬思考 90 秒，会被你自己的代理切断，而你看到的错误是一个通用的上游超时，根本不会提到流式输出。

故障点 #2：负载均衡器在你暂停期间杀死你

同类问题，往外一层，它在最你在意的请求上咬你最深。

AWS Application Load Balancer 的 idle_timeout.timeout_seconds 属性有效范围是 1–4000 秒，默认是 60（参见 ELB API 参考中的 LoadBalancerAttribute）。"Idle"意味着双向都没有字节。一个 LLM 在第一个 token 前思考 75 秒，产生的正是这个：一条安静的、活跃的、完全健康的 TCP 连接，但你的负载均衡器认为它死了。

你可以提高超时时间，而且你可能应该。但单独提高超时时间是一个脆弱的修复，因为它只是把悬崖推得更远。稳健的修复是确保连接实际上永远不会空闲——而 SSE 有一个为此专门构建的机制。

SSE wire 格式将以冒号开头的行视为注释。MDN 说得很清楚："一行开头以冒号作为第一个字符本质上是注释，会被忽略"，以及"注释行可用于防止连接超时；服务器可以定期发送注释以保持连接活跃。"

import asyncio

async def sse_stream(request, params):
    queue: asyncio.Queue = asyncio.Queue()

async def heartbeat():
        while True:
            await asyncio.sleep(15)
            await queue.put(": keepalive\n\n")   # ignored by every SSE client

# ... producer task pushes real events onto the same queue ...

十五秒是一个好默认值——轻松低于 60 秒的空闲超时，足够便宜以至于没人会注意到。这些字节被客户端丢弃，但保持路径上的每一跳都相信连接是活着的。

顺便说一句，Anthropic 在他们那边也做同样的事情："事件流也可以包含任意数量的 ping 事件。"你的解析器需要期望它们并忽略它们，这就直接引向下一个故障点。

故障点 #3：错误以 HTTP 200 到达

这里是其他方面都做对了的团队会栽跟头的地方。

一旦流开始，HTTP 状态码已经发送了。是 200。无论接下来发生什么，它都会保持 200。这之后的失败作为 body 中的事件到达，如果你的解析器只处理快乐路径，它们会静默消失。

Messages API 直接记录了这一点："API 可能会在事件流中偶尔发送错误。例如，在高使用率期间，你可能会收到 overloaded_error，这在非流式上下文中通常对应 HTTP 529"：

event: error
data: {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}}

如果你使用 SDK 辅助函数，这是为你处理好的。如果你正在解析原始 SSE——你在浏览器端是这样的，而且服务器端也经常是这样的——你需要明确地对其分支处理。

还有两件事属于同一个解析器：

未知事件类型不能抛出异常。文档明确说："根据版本管理策略，可能会添加新的事件类型，你的代码应该优雅地处理未知事件类型。"一个带 else: raise 的解析器是一个定时故障，只是日期未知。记录并跳过。

stop_reason 并不总是 end_turn。在 Opus 4.7 及更高版本上，安全分类器可以拒绝请求并返回 HTTP 200 和 stop_reason: "refusal"，stop_details 中包含类别。在你读取 content 之前先检查 stop_reason。（stop_details 只对拒绝填充，对其他所有 stop 原因都是 null，所以在读取前要保护。）在 Opus 5 和 Fable 5 上，你也可以选择服务端回退——betas: ["server-side-fallback-2026-07-01"] 配合 fallbacks: "default"——按拒绝类别路由。如果你这样做，期望在每个模型边界处的流中有一个回退 content 块：一个没有 delta 之间的 content_block_start / content_block_stop 对。一个假设每个块都有 delta 的解析器会在此卡住。

以下是一个能扛住所有这些情况的解析器形状：

for event in raw_events:
    t = event.get("type")

```python
if t == "content_block_delta":
    d = event["delta"]
    if d["type"] == "text_delta":
        yield sse("token", {"text": d["text"]})
    elif d["type"] == "thinking_delta":
        yield sse("thinking", {"text": d["thinking"]})
    # input_json_delta / signature_delta: accumulate, don't render

elif t == "message_delta":
    # NOTE: usage counts in message_delta are CUMULATIVE, not incremental
    usage = event.get("usage") or {}
    output_tokens = usage.get("output_tokens", output_tokens)
    stop_reason = event["delta"].get("stop_reason", stop_reason)

elif t == "error":
    yield sse("error", {"code": event["error"]["type"]})
    return

elif t in ("ping", "content_block_start", "content_block_stop", "message_start", "message_stop"):
    pass

else:
    log.info("unknown stream event, ignoring", extra={"event_type": t})

关于累计 usage 的那条注释值得深入理解——文档里用警告框特别标注了这一点。把 message_delta 的 usage 跨事件累加，会得到一个二次增长的 token 计数和一个严重失真的成本看板。

中断 #4：用户已经离开，但你仍在付费

用户提出一个问题，听完三句话后发现答案不对，关闭了标签页。

你启动的 60,000 token 生成会怎样？许多生产环境中的应用会让它跑完、全额计费，然后把结果写入一个没人会读的数据库行。乘以你的跳出率试试看。

取消必须逐跳显式传播。服务端侧，Starlette（进而 FastAPI）暴露了断连信号：

async def generate(request, params):
    with client.messages.stream(**params) as stream:
        for text in stream.text_stream:
            if await request.is_disconnected():
                break          # exiting the with-block closes the upstream connection
            yield sse("token", {"text": text})

退出上下文管理器才是关键所在——它关闭了到 Anthropic 的 HTTP 连接，生成才会停止。没有上下文管理器的 break，或者一个拥有流且生命周期长于请求的后台任务，会继续燃烧 token。

浏览器侧，fetch 取消通过 AbortController 实现：

const controller = new AbortController();
stopButton.onclick = () => controller.abort();

const res = await fetch("/api/chat/stream", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ prompt }),
  signal: controller.signal,
});

中止会关闭 TCP 连接，这就是你的 is_disconnected() 检查所观察到的现象。这条链路只有在每一环都存在时才能正常工作。

中断 #5：EventSource 做不了你的聊天应用需要的事

浏览器原生的 SSE 客户端是 EventSource，用它是显而易见的选择。但对大多数 LLM 聊天界面来说，它也是错误的选择。

构造函数接受一个 URL 和一个选项对象，其唯一有意义的成员是 withCredentials。没有任何地方可以放 HTTP 方法、请求体或请求头。这就排除了用 POST body 发送 prompt 以及使用 Authorization 头的可能。而它内置的自动重连——"默认情况下，如果客户端与服务器之间的连接关闭，连接会重新启动"——对于 LLM 生成来说是主动出击：一根掉线的连接会静默触发一次全新的生成，全额计费，且不记得已经送出的 token。一次不稳定的连接变成一张重复账单。

改用 fetch 搭配 ReadableStream reader。你得到 POST、请求头、AbortController，以及——关键的是——没有你没要求过的重连：

const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
let buf = "";

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

  const frames = buf.split("\n\n");
  buf = frames.pop();                       // keep the incomplete tail

  for (const frame of frames) {
    const dataLines = frame
      .split("\n")
      .filter((l) => l.startsWith("data:"))  // ':' comment lines fall out here
      .map((l) => l.slice(5).trimStart());
    if (!dataLines.length) continue;         // heartbeat frame
    handle(JSON.parse(dataLines.join("\n")));
  }
}

两个会造成真实 bug 的细节。第一，TCP 不会完整交付你的帧——reader.read() 会给你半个事件，而 buf.split("\n\n") / buf.pop() 模式正是防止你解析到截断 JSON 对象的原因。第二，按照规范，一个帧中多个连续的 data: 行之间用换行符连接；用 "" 连接它们会损坏任何包含换行符的 payload。

中断 #6：部分输出是需要你主动设计的一种状态

每个流有三种结束方式：完成、取消、飞行中失败。大多数代码库只持久化第一种。

这会产生两个丑陋的症状。用户遇到网络抖动后刷新，他们写了一半的答案直接消失了——连接是无状态的，所以 token 也是。而你的成本跟踪会少报，因为你只在干净完成时记录 usage，而每一条被放弃的生成都已全额计费。

修复方案不是可恢复流——那是困难的、大多不必要的功能。它是在生成过程中持久化 assistant 消息，并带上状态：

在第一个 token 之前插入 status = "streaming" 的消息行。

定期刷新累积的文本（每约 50 个 token 或每秒一次——不要每个 token 都刷，除非你喜欢写放大）。

在 message_stop 时，设置 status = "complete" 并写入 get_final_message() 的最终 usage。

在断连或错误时，设置 status = "partial" 或 "failed" 并记录你观察到的 usage。

现在刷新会渲染带有诚实"生成已中断"标记的部分答案，而你的成本看板会统计被放弃的生成——这个数字才能告诉你中断 #4 是否在让你花真金白银。将持久化、usage 计算和流状态连接起来，正是我们在课程中从头到尾构建生产级 AI SaaS 时所做的 plumbing 工作，因为"演示能用"和"产品能用"真正分叉的地方就在这里。

以上所有内容，压缩成你今天下午就可以去验证的事项：

如果你想知道这些是否真的在让你花钱，诚实的测试是用 30% 中途放弃率做一次负载测试，对 staging 环境运行，并在两端测量 token 使用量。你的看板报告的数字和你的 Anthropic 控制台报告的数字之间的差距，就是问题的规模。

流式处理是很多 AI 产品悄悄损失金钱和信任的地方——这是一个 plumbing 问题，而不是 prompting 问题。如果你想知道从第一次 API 调用到构建出一个在真实流量下稳定运行的应用程序的完整路径，那是我们的 Advanced LLM Integration in Production 课程的脉络；其背后的 failure-mode discipline——知道一个改变是否真的有帮助——来自 LLM Evaluation and Testing。如果你的下一个功能是语音，这六个中断每一个都会更难，那就是 Voice AI and Realtime Multimodal Agents 中的另一个话题了。

Sources: Streaming Messages — Claude API docs · Using server-sent events — MDN · ngx_http_proxy_module — nginx · LoadBalancerAttribute — AWS ELB API Reference