RH
RouterHub AI API 中转站导航与平台推荐
登录
博客

LLM 流式输出工程:SSE 重连、Last-Event-ID、背压与中途失败

流式不是"加个 stream=true"那么简单 绝大多数用 LLM 的产品都接了流式输出,但能把流式做到生产稳定的不到 30%。

muchacha 2026-06-16 22:11:31

正文

流式不是"加个 stream=true"那么简单

绝大多数用 LLM 的产品都接了流式输出,但能把流式做到生产稳定的不到 30%。原因是流式涉及的边界情况比想象多得多——重连、断流、背压、tool 调用中途、token 计数、客户端崩了怎么续——任何一处没处理好,用户就看到"AI 卡死"或"突然消失了一半文字"。

这篇文章把 LLM 流式输出的工程化做法讲完整。从 transport 选择,到 SSE 的标准重连,到背压策略,到 tool call 期间的特殊处理。

一、Transport 选什么

主流方案有 SSE、WebSocket、gRPC streaming、HTTP/2 server push。LLM 业务场景的实战选择:

Transport 优点 缺点 适用
SSE 浏览器原生支持、自动重连、HTTP-friendly 单向(server → client) 绝大多数 LLM 流式
WebSocket 双向、低延迟 需要自己实现重连、proxy 友好度差 双向交互(语音 agent)
gRPC streaming 高性能、强 schema 浏览器原生不支持、要 envoy/grpc-web 后端到后端
HTTP chunked 兼容性好 没有标准重连、消息边界要自己定 legacy / 简单 demo

99% 的场景用 SSE。它的理由:

  • 浏览器 EventSource API 自带重连,用 Last-Event-ID header 续传
  • HTTP/2 之后,单连接里能跑 100+ 个 SSE 流(老的 6 连接限制不存在了)
  • 标准消息格式,不用自己设计帧协议

二、SSE 协议的关键三件套

data: {"type": "token", "value": "Hello"}
id: 42
event: message

data: {"type": "token", "value": " world"}
id: 43
event: message

data: [DONE]
id: 44
event: done

三个关键字段:

  • data:实际数据,可以多行
  • id:用于断线重连续传(Last-Event-ID)
  • event:事件类型,客户端可以分类监听

服务端实现要点:

  1. Content-Type: text/event-stream
  2. Cache-Control: no-cache, no-transform(no-transform 极重要,否则中间代理可能 buffer)
  3. X-Accel-Buffering: no(nginx 反代必加)
  4. 每条消息 \n\n 结尾
  5. 必须立即 flush,否则代理 buffer 会让流变批量

三、最小可用的 Node 服务端

import express from 'express'
import Anthropic from '@anthropic-ai/sdk'

const app = express()
const client = new Anthropic()

app.get('/chat/stream', async (req, res) => {
  res.setHeader('Content-Type', 'text/event-stream')
  res.setHeader('Cache-Control', 'no-cache, no-transform')
  res.setHeader('Connection', 'keep-alive')
  res.setHeader('X-Accel-Buffering', 'no')
  
  let eventId = Number(req.headers['last-event-id'] ?? 0)
  
  // 心跳,防代理超时
  const heartbeat = setInterval(() => {
    res.write(': ping\n\n')
  }, 15000)
  
  try {
    const stream = await client.messages.stream({
      model: 'claude-sonnet-4.6',
      max_tokens: 4096,
      messages: [{ role: 'user', content: req.query.q as string }],
    })
    
    for await (const event of stream) {
      if (event.type === 'content_block_delta' && event.delta.type === 'text_delta') {
        eventId++
        res.write(`id: ${eventId}\n`)
        res.write(`event: token\n`)
        res.write(`data: ${JSON.stringify({ text: event.delta.text })}\n\n`)
      }
    }
    
    eventId++
    res.write(`id: ${eventId}\n`)
    res.write(`event: done\n`)
    res.write(`data: [DONE]\n\n`)
  } catch (err) {
    res.write(`event: error\n`)
    res.write(`data: ${JSON.stringify({ message: (err as Error).message })}\n\n`)
  } finally {
    clearInterval(heartbeat)
    res.end()
  }
})

这个版本能跑,但还没处理重连续传——下一节讲。

四、重连与续传

4.1 浏览器侧

const source = new EventSource('/chat/stream?q=hello')

source.addEventListener('token', (e) => {
  const { text } = JSON.parse(e.data)
  appendToUI(text)
})

source.addEventListener('done', () => source.close())

source.onerror = () => {
  console.warn('reconnecting...')
  // EventSource 会自动重连,带上 Last-Event-ID
}

EventSource 在断线时自动重连,自动带上最后收到的 id 作为 Last-Event-ID header。这是 SSE 比 WebSocket 省心的地方。

4.2 服务端续传

看到 Last-Event-ID 应该:

  • 找到这次会话的生成进度
  • 跳过已经发送的 token
  • 从下一个 token 继续

实现关键:生成内容必须落盘,不能只在 LLM stream buffer 里。一种常见做法:

// 用户首次请求时
const sessionId = randomUUID()
await redis.set(`session:${sessionId}:full_text`, '', 'EX', 600)

// 每个 token 都追加
for await (const event of stream) {
  if (event.type === 'content_block_delta') {
    const text = event.delta.text
    await redis.append(`session:${sessionId}:full_text`, text)
    // ... 然后再发给客户端
  }
}

// 重连时
const lastId = req.headers['last-event-id']
const fullText = await redis.get(`session:${sessionId}:full_text`)
const tokensAlreadySent = parseInt(lastId ?? '0')
// 从 tokensAlreadySent 开始重新发剩余部分

这套机制让用户在地铁里突然没信号、隧道出来再连上,看到的是完整内容,而不是"AI 输出截断了一半"。

五、Tool Call 期间的特殊处理

LLM 在生成中途调 tool 时,客户端如果只看到 token 流,会一脸懵。需要额外的事件类型告诉客户端"AI 在调工具,请稍等"。

event: tool_use_start
data: {"tool": "search_sites", "input": {"q": "..."}}

event: tool_use_result
data: {"tool": "search_sites", "result": [...]}

event: token
data: {"text": "根据搜索结果..."}

UI 侧据此显示一个"AI 正在搜索…"的状态条,用户体验远好于看着光标卡 5 秒。

注意 tool 执行期间客户端连接必须保持,继续发心跳。否则 nginx/cloudflare 会因为 60 秒无数据而断连。

六、背压(Backpressure)

LLM 生成速度通常 30-100 tokens/秒,客户端绝大多数情况下能消费过来。但有几种情况会出问题:

6.1 客户端解析慢

UI 侧逐 token 渲染,如果每个 token 触发昂贵的 markdown 渲染或 syntax highlight,会跟不上。

应对:在客户端做渲染节流——每 50ms 批量更新 UI 一次,而不是每个 token 都重渲。

6.2 网络拥堵

客户端在弱网下,TCP buffer 满了,服务端 write 会阻塞。Node 的 res.write() 返回 false 表示需要等待 drain。

应对:

function writeWithBackpressure(res: Response, chunk: string): Promise<void> {
  return new Promise((resolve) => {
    if (!res.write(chunk)) {
      res.once('drain', resolve)
    } else {
      resolve()
    }
  })
}

服务端代码改成 await writeWithBackpressure(...),自动等客户端跟上。如果不处理,服务端会内存膨胀,严重时 OOM。

6.3 客户端干脆没了

用户关浏览器了,但服务端还在花钱跑 LLM 生成。

应对:监听 res.close / req.on('close'),触发 stream cancel:

req.on('close', () => {
  stream.controller.abort()
  console.log('client gone, aborted upstream stream')
})

这一步必须做,不然你的账单会被"用户连点 10 次重新生成、每次都没等完"打爆。

七、错误处理与重试

7.1 中途失败

LLM 可能在生成中间报错(rate limit、内部错误)。绝不能让客户端只看到一半文本然后连接静默关闭。

try {
  // ... stream loop
} catch (err) {
  res.write(`event: error\n`)
  res.write(`data: ${JSON.stringify({
    code: classifyError(err),
    retryable: isRetryable(err),
    message: err.message,
  })}\n\n`)
}

客户端拿到 error 事件后,根据 retryable 决定是否自动重试。

7.2 智能重试

客户端重试时,如果服务端做了内容落盘,可以从断点续传(见第 4 节)。如果没做,把已经收到的 partial text 作为上下文发回:

请基于已经写到的内容继续:
"<partial_text>"

这种"接力"模式比从头重写省钱省时间。

7.3 错误分类

function classifyError(err: unknown): string {
  if (err instanceof Anthropic.RateLimitError) return 'rate_limit'
  if (err instanceof Anthropic.APITimeoutError) return 'timeout'
  if (err instanceof Anthropic.AuthenticationError) return 'auth'
  if (err instanceof Anthropic.BadRequestError) return 'bad_request'
  return 'unknown'
}

不同错误对应不同处理:

  • rate_limit:指数退避重试
  • timeout:立即重试
  • auth:不要重试,告警
  • bad_request:不要重试,记录 + 上报

八、可观测性

流式接口的指标比普通 HTTP 多几项:

  • TTFT(Time To First Token):用户从发请求到看到第一个 token 的时间。这是流式的灵魂指标,< 1s 是优秀,1-2s 可接受,> 2s 用户开始抱怨
  • token rate:tokens/秒,反映上游模型负载
  • stream complete rate:正常完成 vs 中途断流的比例
  • reconnect rate:多少 session 触发了重连
  • partial bytes lost:重连续传中丢失的字节数(理想为 0)

把这些打到 Prometheus / Datadog,看板上一眼能看出"LLM provider 是不是抽风了"。

九、Edge Function 部署的坑

很多团队把 LLM stream API 部署到 Cloudflare Workers / Vercel Edge Functions。注意几个坑:

9.1 流式必须用 Response stream

// Cloudflare Workers 的写法
export default {
  async fetch(req: Request): Promise<Response> {
    const { readable, writable } = new TransformStream()
    const writer = writable.getWriter()
    const encoder = new TextEncoder()
    
    streamToClient(writer, encoder)  // 异步运行
    
    return new Response(readable, {
      headers: {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache, no-transform',
      },
    })
  },
}

Workers 的 CPU time 限制(默认 30s,付费 50s)是硬上限。生成超过这个时间会被强杀。

9.2 部分 CDN 仍会缓冲

即使设置了 no-transform,某些 CDN 边缘节点仍可能缓冲短数据。规避方法:

  • 在响应头加 X-Accel-Buffering: no(Nginx)
  • Cloudflare 上禁用 “Auto Minify”
  • 第一个事件后立即发送 4-8KB padding,踢出缓冲区

9.3 Vercel 函数超时

Vercel Hobby plan 默认 10 秒超时。流式不能超过这个时间。Pro plan 才有 60 秒。生产请用 Pro 或自托管。

十、给 RouterHub 用户的话

通过中转 API 做流式,确认中转的几件事:

  1. 是否原样转发 SSE 帧:有些中转把 SSE 转成 chunked 但格式错误,客户端 EventSource 解析不了
  2. 是否保留 token 字段名:OpenAI 的 delta.content vs Anthropic 的 delta.text 是不同字段,中转转换时要正确映射
  3. 是否支持取消:客户端 abort 时,中转必须把 cancel 信号传到上游,否则"省费"功能失效
  4. 是否压缩 SSE:gzip/brotli 在 SSE 上有时会增加首字延迟,中转默认应该禁用

跑一段简单测试可以验证:

const source = new EventSource('https://your-relay.example.com/...')
source.addEventListener('message', e => console.log(performance.now(), e.data))

第一条消息 < 1.5s 到达且后续连续输出,证明流式链路通畅。> 3s 才到第一个、或者所有 token 一次性涌出,说明中转在 buffer。

相关阅读