实战踩坑：用Python构建高并发LLM流式代理网关（SSE+背压控制+断连回收）

共计 4336 个字符，预计需要花费 11 分钟才能阅读完成。

最近在为团队的 AI Agent 平台重构接入层网关时，我们遇到了一个极其典型的“伪高并发”翻车现场：前端用户在生成回答时频繁点击“停止生成”或直接关掉网页，而后端接入的 vLLM 集群 GPU 算力却依然吃满，响应队列被堵死，导致真正等待结果的用户延迟飙升到 15 秒以上。

很多同学在写大模型（LLM）API 代理或网关时，往往以为只要在 FastAPI 里套一个 StreamingResponse 就算完事。实际上，流式代理（SSE / Server-Sent Events）如果没做 客户端断开探测 、 生成器协程取消 和内存背压（Backpressure）控制，只要并发一上来，网关和推理端就会双双雪崩。今天我们不聊虚的概念，直接拿生产踩坑排查出来的代码，手把手拆解一套抗打的 Python LLM 流式代理架构。

一、为什么常规的 StreamingResponse 会拖垮推理集群？

看一个最常见的错误示范：

# 常见但有重大隐患的代码
@app.post("/v1/chat/completions")
async def chat_proxy(request: Request):
    body = await request.json()
    async def event_generator():
        async with httpx.AsyncClient() as client:
            async with client.stream("POST", UPSTREAM_LLM_URL, json=body) as response:
                async for chunk in response.aiter_bytes():
                    yield chunk
    return StreamingResponse(event_generator(), media_type="text/event-stream")

这段代码在平时自测时毫无破绽，但一旦上生产压测，致命问题接踵而至：

1. 上游算力白嫖幽灵：当客户端网络抖动或用户主动点击“取消”，下游连接已断开，但 Python 的 event_generator 依然在不知疲倦地从上游 vLLM 读取剩余的几百个 Token。vLLM 根本不知道前端走人了，依然在吭哧吭哧跑 KV Cache，GPU 算力被白白浪费。
2. HTTPX 连接池耗尽：每次请求都 async with httpx.AsyncClient() 动态建连，没有持久连接复用，瞬时并发突破 300 时，系统的 TIME_WAIT 端口直接被打爆。
3. 消费速率不对等引起的内存膨胀：上游推理端产生 Token 的速度是每秒 80 tokens，但如果某个手机客户端网络奇差，接收速度只有每秒 5 tokens，网关层如果没有缓冲流控，所有的 chunks 就会堆积在网关的内存中。

二、架构重构：支持主动取消与背压控制的网关核心

解决上述问题的核心思路是：将客户端连接状态与上游异步生成器强绑定。当监测到 request.is_disconnected() 为真时，第一时间向上游发送 abort 请求或主动关闭 httpx 流，让推理端立刻截断采样。

1. 连接复用与全局生命周期管理

首先抽离 HTTP 客户端，使用全局单例的 httpx.AsyncClient，并调大连接池上限和持久保持机制：

from contextlib import asynccontextmanager
from fastapi import FastAPI
import httpx

# 全局客户端，避免每次请求频繁创建和销毁
http_client: httpx.AsyncClient = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global http_client
    # 限制连接池并配置长连接保活
    limits = httpx.Limits(max_keepalive_connections=200, max_connections=1000)
    timeout = httpx.Timeout(connect=5.0, read=60.0, write=5.0, pool=10.0)
    http_client = httpx.AsyncClient(limits=limits, timeout=timeout)
    yield
    await http_client.aclose()

app = FastAPI(lifespan=lifespan)

2. 生产级流式转发与断连感知发生器

我们编写一个核心流式转发发生器，通过 asyncio.wait 实现对请求取消的毫秒级感知：

import asyncio
import json
from fastapi import Request, HTTPException
from fastapi.responses import StreamingResponse

UPSTREAM_API = "http://10.0.0.88:8000/v1/chat/completions"

async def stream_with_cancellation(request: Request, payload: dict):
    # 确保强制开启上游流式输出
    payload["stream"] = True
    
    # 构造上游请求（使用全局 client）req = http_client.build_request("POST", UPSTREAM_API, json=payload)
    response = await http_client.send(req, stream=True)

    if response.status_code != 200:
        error_detail = await response.aread()
        await response.aclose()
        raise HTTPException(status_code=response.status_code, detail=error_detail.decode())

    async def body_iterator():
        try:
            async for line in response.aiter_lines():
                # 每吐出一个 chunk 前，优先检查客户端是否断开
                if await request.is_disconnected():
                    # 触发日志记录，立即退出循环
                    print("[Stream] 监测到下游客户端主动断开，开始阻断上游...")
                    break
                
                if line:
                    yield f"{line}\r\n\r\n".encode("utf-8")
        except asyncio.CancelledError:
            # 当请求被 ASGI 容器强行 abort 时触发
            print("[Stream] 协程被强制取消")
            raise
        finally:
            # 无论是因为正常结束、客户端断连还是异常退出，务必显式关闭 response
            # 对于支持的推理引擎（如 vLLM），关闭流会自动中止该 prompt 的继续 decoding
            await response.aclose()
            print("[Stream] 上游流式连接已安全释放")

    return body_iterator()

3. 路由层接入与 Header 规范

前端的 EventSource 或 fetch 流读取对响应头非常敏感，特别是 Nginx 反代场景下，如果缺少特定 Header，Nginx 可能会把流数据缓存起来，导致前端“卡顿几秒后突然吐出一大段”。

@app.post("/v1/chat/completions")
async def chat_endpoint(request: Request):
    payload = await request.json()
    
    stream_generator = await stream_with_cancellation(request, payload)
    
    headers = {
        "Content-Type": "text/event-stream; charset=utf-8",
        "Cache-Control": "no-cache, no-transform",
        "Connection": "keep-alive",
        "X-Accel-Buffering": "no", # 关键：严禁 Nginx 缓冲 SSE 数据流
    }
    
    return StreamingResponse(stream_generator, headers=headers)

三、Nginx 与网关层调优实战

光改 Python 代码还不够。如果你的架构前置了 Nginx 反向代理，默认的代理缓冲配置会彻底破坏你的流式体验。我在测试机上踩过一个深坑：后端代码跑得很完美，但前端拿到的流总是 4KB 一顿一顿地出。

在你的 Nginx 虚拟主机配置中，必须加入如下优化指令：

location /v1/chat/completions {
    proxy_pass http://127.0.0.1:8080;
    
    # 强制关闭缓冲，保障 chunk 实时下发
    proxy_buffering off;
    proxy_cache off;
    
    # 针对 HTTP/1.1 协议优化连接复用
    proxy_http_version 1.1;
    proxy_set_header Connection "";
    
    # 延长超时时间，防止长文本推理生成阶段被 Nginx 切断
    proxy_read_timeout 300s;
    proxy_send_timeout 300s;
}

四、实测数据：优化前后的 GPU 吞吐对比

我们在单台搭载 4 卡 NVIDIA A100 (80GB) 的内网节点上，使用 Locust 模拟 500 个并发用户会话。其中 40% 的用户在生成前 10 个 Token 后随机触发中断（模拟真实场景下的打断和刷新）。

总结与避坑心得

在工程化落地大模型应用时，模型参数与算法固然关键，但中间这层 管道工程 往往直接决定了产品的生死线。总结下来三个核心原则：

1. 永远不要让推理引擎盲目计算：客户端只要断连，网关就必须毫不留情地阻断上游连接，保护你极其昂贵的 GPU 显存。
2. 管理好你的 ASGI 协程与连接池：单例 httpx.AsyncClient 结合合理的保活时间是防范连接数打满的基本功。
3. 注意链路全流程的无缓冲规范：从 Python 层的 StreamingResponse 到 Nginx 的 proxy_buffering off，每一环都不能丢掉防缓冲标志。

按照这套方案改造后，不仅解决了算力浪费的问题，整个 API 网关的稳定性也直接迈上了一个台阶。建议大家把断连监控日志加入到 Prometheus 指标中，时刻关注团队系统里的“幽灵流量”。