实战踩坑：用 Python + vLLM 搭建生产级大模型流式网关（高并发+SSE+背压控制）

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

很多同学在本地或者私有云环境把 vLLM、Ollama 跑起来之后，就直接在前端拿原生 fetch 或 EventSource 直连模型端口。在个人调试时这没问题，但一旦挂上业务网关或者多租户并发访问，问题接踵而至：客户端关闭标签页导致后台 GPU 依然在疯狂空转、长连接堆积压垮应用服务器、上游生成过快导致 Node/Python 代理层内存被打爆（OOM）。

上周我在一台双卡 RTX 4090 的宿主机上处理内部 Agent 网关的压测任务，用 Locust 压到 300 并发流式输出时，网关内存从 200MB 一路飙到 4GB，显存利用率更是被无意义的死连接占满。这篇文章拆解我最终重构的 Python 高并发流式网关核心实现，彻底解决 SSE（Server-Sent Events）背压控制 、 客户端断开级联取消机制 以及 异步连接池复用 问题。

一、为什么简单的 FastAPI StreamingResponse 会崩？

绝大多数教程给出的标准写法通常长这样：

# 典型的“教科书式但不可用于生产”的代码
@app.post("/v1/chat/completions")
async def chat_stream(request: ChatRequest):
    async def event_generator():
        async for chunk in call_llm_stream(request):
            yield f"data: {chunk}\n\n"
    return StreamingResponse(event_generator(), media_type="text/event-stream")

这段代码在生产环境存在三个致命隐患：

1. 无显式客户端断开检测（Orphan Tasks）：用户问了一半觉得不对劲直接点了“停止生成”或关掉浏览器，FastAPI 默认不会主动打断内部生成器，call_llm_stream 会继续向下游推理引擎（如 vLLM）拉取 token，白白浪费稀缺的算力。
2. 背压（Backpressure）缺失：vLLM 的 PagedAttention 吞吐极高（每秒可能产出数十个 chunk），如果消费端（如弱网手机端）接收变慢，网关异步队列就会无限缓存这些数据，导致内存暴涨。
3. 连接池耗尽（File Descriptor Leak）：如果不精确复用 httpx.AsyncClient，每个请求新建 HTTP 连接，在高 QPS 下很快就会打爆操作系统的 nofile 限制，抛出 Too many open files。

二、架构重构：工业级异步流式网关实现

我们采用 FastAPI + httpx + 异步生成器监听 的架构，在代理层注入三个核心防御逻辑：检测 request.is_disconnected()、精确配置 HTTP 缓冲与 Keep-Alive，以及流式协议封装。

1. 完整的流式反代网关核心代码

import json
import logging
import asyncio
from typing import AsyncGenerator
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
import httpx

app = FastAPI(title="LLM-Gateway-Proxy")
logger = logging.getLogger("gateway")

# 必须全局复用 Client，并精细调整连接池参数
limits = httpx.Limits(max_keepalive_connections=100, max_connections=500, keepalive_expiry=30.0)
# 针对流式场景，禁用读超时，但严格控制连接建立超时
timeout = httpx.Timeout(connect=5.0, read=None, write=10.0, pool=5.0)
client = httpx.AsyncClient(base_url="http://127.0.0.1:8000", limits=limits, timeout=timeout)

@app.on_event("shutdown")
async def shutdown_event():
    await client.aclose()

async def sse_stream_wrapper(
    upstream_response: httpx.Response, 
    client_request: Request
) -> AsyncGenerator[bytes, None]:
    """负责将 upstream 的原始流转发给客户端，同时注入背压与断联监控"""
    try:
        # aiter_raw / aiter_lines：避免全量解码开销
        async for line in upstream_response.aiter_lines():
            # 关键：检查下游客户端是否已挂断
            if await client_request.is_disconnected():
                logger.warning("Downstream client disconnected prematurely. Aborting upstream stream.")
                break

            if not line:
                continue

            # 保持 SSE 协议格式：行末补充换行
            yield f"{line}\r\n\r\n".encode("utf-8")
            
            # 微小的协作式让渡，避免协程饥饿
            await asyncio.sleep(0)

    except asyncio.CancelledError:
        logger.info("Task cancelled by framework.")
        raise
    except Exception as e:
        logger.error(f"Error streaming response: {str(e)}")
        yield f"event: error\ndata: {json.dumps({'error':'stream_failed'})}\r\n\r\n".encode("utf-8")
    finally:
        # 确保主动释放上游流连接，触发 vLLM 侧断开并停止推理
        await upstream_response.aclose()

@app.post("/api/v1/chat")
async def chat_proxy(request: Request):
    try:
        payload = await request.json()
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid JSON payload")

    # 强制确保请求带有 stream 标志
    payload["stream"] = True

    # 构建发送给 vLLM 的请求对象
    upstream_req = client.build_request("POST", "/v1/chat/completions", json=payload)
    
    try:
        # 使用 send(stream=True) 获取底层流响应，不加载至内存
        upstream_resp = await client.send(upstream_req, stream=True)
    except httpx.ConnectError:
        raise HTTPException(status_code=503, detail="vLLM engine unavailable")

    if upstream_resp.status_code != 200:
        error_content = await upstream_resp.aread()
        await upstream_resp.aclose()
        return StreamingResponse(iter([error_content]), status_code=upstream_resp.status_code)

    return StreamingResponse(sse_stream_wrapper(upstream_resp, request),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"  # 禁用 Nginx 缓冲区，关键！}
    )

三、高并发调优的核心实战参数

代码跑通只是第一步，要让网关真正扛住数百路并发流式传输，必须解决传输链路上的缓冲截断与系统限制。

1. 反向代理层（Nginx）的缓冲陷阱

如果你的 FastAPI 前面挂了 Nginx，必须在 location 块中显式关闭缓冲。否则 Nginx 会默认等积攒了 4KB 数据才一次性 flush 给前端，你的打字机效果会直接变成“卡顿式吐字”。

location /api/ {
    proxy_pass http://127.0.0.1:8888;
    proxy_http_version 1.1;
    proxy_set_header Connection "";
    
    # 彻底关闭缓存与缓冲，保证逐字推流
    proxy_buffering off;
    proxy_cache off;
    chunked_transfer_encoding on;

    # 设置较长的流式超时时间（防止长推理被中断）proxy_read_timeout 600s;
    proxy_send_timeout 600s;
}

2. Uvicorn 生产启动参数

流式网关属于典型的 I/O 密集型场景，单进程往往无法充分利用多核解析 JSON 和维护长连接。推荐使用 uvloop 和 httptools，进程数设为 CPU 核心数 * 2：

# 生产部署启动命令
uvicorn gateway:app \
  --host 0.0.0.0 \
  --port 8888 \
  --workers 4 \
  --loop uvloop \
  --http httptools \
  --limit-concurrency 1000 \
  --backlog 2048 \
  --timeout-keep-alive 65

四、实测压测数据对比

我们在双路 Xeon 6330、128G 内存、双卡 RTX 4090 环境下，使用 Locust 对比了“直接访问 vLLM 暴露的端口”与“通过本网关代理”的性能表现（并发模型：Qwen-14B-Chat-AWQ，输入 512 tokens，输出 256 tokens，设定 30% 用户在生成至第 50 token 时主动断开）：

实测数据表明，在引入主动断开机制后，由于大量无效任务在客户端取消时被立即从 vLLM 调度队列中剥离，TTFT（Time To First Token）下降了 22.5%，显卡避免了大量“鬼影计算”，整体系统吞吐量提升了将近三分之一。

总结与避坑心得

在大模型工程落地时，流式响应绝对不是简单的一个 yield 就能高枕无忧。总结三条核心避坑军规：

• 一定要处理 client.is_disconnected()：把算力留给真正的有效请求，保护 vLLM 的 KV Cache 队列。
• 善用 X-Accel-Buffering: no 与 Nginx 配置：绝大多数前端“打字机失效、变成整段吐出”的故障，根源都在网关缓冲配置上。
• 连接池的超时颗粒度必须分立：对 LLM 场景，read_timeout 可以为 None，但 connect_timeout 必须严格限制在 3~5 秒内，防止上游挂死连带拖垮网关。