更新于

LLM 流式响应背压处理：防止服务端被打爆的实战指南

LLM流式响应应用工程后端架构

在把 LLM 流式输出接入生产系统时，大多数工程师先遇到的是「怎么用 SSE 把 token 推给前端」，但很快就会碰到一个更棘手的问题：当前端消费速度跟不上后端生产速度时，数据该怎么办？

这就是「背压」（Backpressure）问题。忽视它的后果轻则内存溢出、连接超时，重则整个服务雪崩。本文从原理到代码，给出一套可落地的处理方案。

一、什么是背压，LLM 场景为何特别容易出现

背压（Backpressure）是流式系统中的经典问题：当数据生产速率 > 消费速率时，中间的缓冲区被撑满，系统要么丢数据，要么生产端被迫等待，要么缓冲区无限增长最终 OOM。

LLM 场景有三个特点放大了这个风险：

1. 生产端速率不可控 模型推理速率取决于 GPU 负载、批处理状态，高峰期可能骤降，低峰期又可能突增。你的 streaming consumer 必须应对这种波动。

2. 消费端差异极大 同一个服务后端，连接的可能是 100ms 延迟的数据中心客户端，也可能是 5G 切换中的移动端用户。两者消费速率差一到两个数量级。

3. 中间层往往不透明 Nginx → Node.js/Python → Claude API 这条链路上，每一层都有自己的缓冲区（TCP socket buffer、HTTP/1.1 chunked buffer、SSE event buffer），背压问题可能在任何一层静默积压。

二、背压问题的具体表现

在 LLM 应用中，背压失控通常以以下形态出现：

症状根因
Node.js 进程内存持续增长，重启后恢复流式数据在内存队列中堆积
客户端断连后服务端仍在消费 Claude API没有监听 close 事件，未及时取消上游请求
大并发时响应延迟陡增Event loop 被同步的流处理逻辑阻塞
SSE 连接偶发”假死”（客户端无新 token，但连接未断）TCP 缓冲区满，写入操作阻塞但未超时

三、Node.js 中的流式背压处理

3.1 基础：监听 drain 事件

Node.js 的 stream.Writable 内置背压信号：当 write() 返回 false 时，说明内部缓冲区已满，应该暂停生产；当 drain 事件触发时，可以恢复生产。

import Anthropic from "@anthropic-ai/sdk";
import http from "http";

const client = new Anthropic();

http
  .createServer(async (req, res) => {
    res.setHeader("Content-Type", "text/event-stream");
    res.setHeader("Cache-Control", "no-cache");
    res.setHeader("Connection", "keep-alive");

    const stream = await client.messages.stream({
      model: "claude-opus-4-8",
      max_tokens: 1024,
      messages: [{ role: "user", content: "请写一篇关于量子计算的科普文章" }],
    });

    // 客户端断连时，立即取消上游流
    req.on("close", () => {
      stream.controller.abort();
    });

    for await (const event of stream) {
      if (
        event.type === "content_block_delta" &&
        event.delta.type === "text_delta"
      ) {
        const data = `data: ${JSON.stringify({ text: event.delta.text })}\n\n`;

        // write() 返回 false = 缓冲区满，需要等待 drain
        const canContinue = res.write(data);
        if (!canContinue) {
          // 暂停 stream 消费，等待 drain
          await new Promise((resolve) => res.once("drain", resolve));
        }
      }
    }

    res.write("data: [DONE]\n\n");
    res.end();
  })
  .listen(3000);

关键点：await new Promise(resolve => res.once('drain', resolve)) 这一行会暂停 for await 循环，直到写入缓冲区被清空，避免无限堆积。

3.2 生产级封装：带超时和错误处理

async function streamToResponse(stream, res, timeoutMs = 30000) {
  let lastActivity = Date.now();

  const timeoutCheck = setInterval(() => {
    if (Date.now() - lastActivity > timeoutMs) {
      stream.controller.abort();
      res.end();
      clearInterval(timeoutCheck);
    }
  }, 5000);

  try {
    for await (const event of stream) {
      if (
        event.type === "content_block_delta" &&
        event.delta.type === "text_delta"
      ) {
        lastActivity = Date.now();
        const data = `data: ${JSON.stringify({ text: event.delta.text })}\n\n`;

        if (!res.writable) break; // 连接已关闭

        const canContinue = res.write(data);
        if (!canContinue) {
          await Promise.race([
            new Promise((resolve) => res.once("drain", resolve)),
            new Promise((_, reject) =>
              setTimeout(() => reject(new Error("drain timeout")), 10000)
            ),
          ]);
        }
      }
    }
  } catch (err) {
    if (err.name !== "AbortError") {
      console.error("Stream error:", err);
    }
  } finally {
    clearInterval(timeoutCheck);
    if (res.writable) {
      res.write("data: [DONE]\n\n");
      res.end();
    }
  }
}

四、Python（asyncio）中的背压处理

Python 的 asyncio 有类似机制：asyncio.Queue 配合 maxsize 实现有界缓冲，生产者在队列满时自动挂起。

import asyncio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import anthropic

app = FastAPI()
client = anthropic.AsyncAnthropic()

async def stream_generator(queue: asyncio.Queue):
    """从队列消费并生成 SSE 数据"""
    while True:
        item = await queue.get()
        if item is None:  # 结束信号
            yield "data: [DONE]\n\n"
            break
        yield f"data: {item}\n\n"

async def produce_to_queue(queue: asyncio.Queue, user_message: str):
    """从 Claude API 生产数据到有界队列"""
    try:
        async with client.messages.stream(
            model="claude-opus-4-8",
            max_tokens=1024,
            messages=[{"role": "user", "content": user_message}],
        ) as stream:
            async for text in stream.text_stream:
                # 队列满时，put() 会自动等待（背压）
                await queue.put(text)
    finally:
        await queue.put(None)  # 发送结束信号

@app.get("/stream")
async def stream_endpoint(q: str = "你好"):
    # maxsize=50：队列最多缓冲 50 个 token，超出则生产者暂停
    queue = asyncio.Queue(maxsize=50)

    # 并发启动生产者
    asyncio.create_task(produce_to_queue(queue, q))

    return StreamingResponse(
        stream_generator(queue),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

asyncio.Queue(maxsize=50) 的 maxsize 就是背压阀门：当队列中已有 50 个未消费的 token 时，queue.put() 会挂起，让出 event loop，等待消费者腾出空间。

五、Nginx 层的配置陷阱

很多工程师在应用层处理好了背压，却忘了 Nginx 本身也会缓冲。如果 Nginx 将 SSE 响应整体缓存后再转发，背压就完全失效了。

必须添加的 Nginx 配置：

location /api/stream {
    proxy_pass http://backend;
    proxy_http_version 1.1;
    
    # 关闭 Nginx 代理缓冲，让数据实时穿透
    proxy_buffering off;
    proxy_cache off;
    
    # 告知 Nginx 不要缓冲响应体
    proxy_set_header X-Accel-Buffering no;
    
    # 超时配置：LLM 响应可能很长
    proxy_read_timeout 300s;
    proxy_send_timeout 300s;
    
    # 保持 HTTP/1.1 keepalive
    proxy_set_header Connection '';
}

同时在应用层响应头中也加上 X-Accel-Buffering: no，双重保险。

六、客户端断连的处理：别让上游白白烧钱

背压的另一面是：当客户端主动断开连接时，服务端必须立即停止向 Claude API 拉取数据。否则 Claude API 还在持续生成和计费，但数据已经无处发送。

// Express 示例
app.get("/stream", async (req, res) => {
  const abortController = new AbortController();

  // 客户端断连 → 立即 abort 上游请求
  req.on("close", () => {
    abortController.abort();
  });

  try {
    const stream = await client.messages.stream(
      {
        model: "claude-opus-4-8",
        max_tokens: 1024,
        messages: [{ role: "user", content: req.query.q }],
      },
      { signal: abortController.signal }
    );

    // ... 流处理逻辑
  } catch (err) {
    if (err.name !== "AbortError") {
      console.error(err);
    }
  }
});

在生产环境中，这个模式每天可以节省大量无效的 token 消耗。

七、监控与告警：发现背压问题的信号

背压问题往往是渐进式的，需要主动监控：

import time
from prometheus_client import Gauge, Counter

# 关键指标
stream_queue_depth = Gauge("llm_stream_queue_depth", "当前流式队列深度")
stream_drain_waits = Counter("llm_stream_drain_waits_total", "背压等待次数")
stream_client_disconnects = Counter("llm_stream_client_disconnects_total", "客户端主动断连次数")

# 在生产代码中埋点
async def monitored_produce(queue, message):
    async with client.messages.stream(...) as stream:
        async for text in stream.text_stream:
            current_depth = queue.qsize()
            stream_queue_depth.set(current_depth)
            
            if current_depth >= queue.maxsize * 0.8:
                # 队列已达 80% 容量，记录背压事件
                stream_drain_waits.inc()
            
            await queue.put(text)

告警规则建议：

  • 队列深度持续 > 80% 超过 30 秒：触发告警
  • drain 等待次数每分钟 > 100：说明消费端明显慢于生产端
  • 客户端断连率 > 5%：需要检查前端实现或网络质量

八、相关阅读

流式 API 的稳定性离不开可靠的基础设施，YoTradeApi 提供低延迟、高稳定的 Claude API 中转，国内网络环境下流式响应的首 token 延迟可低至 1 秒内。

相关阅读