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

1次阅读

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

最近团队内部把几个私有化部署的 70B 开源模型切换到了 vLLM 驱动的集中式推理集群上。本以为前端写个简单的 FastAPI 转发一下 /v1/chat/completions 的 SSE（Server-Sent Events）就万事大吉，结果压测和上线第一周就接连翻车：并发一上来内存飙升不释放、前端用户中途取消对话后显卡依然在疯狂空转、中文字符在流式吐字时偶尔爆出 \ufffd 乱码……

为了解决这些问题，我重构了一版轻量高效的 Python 流式 LLM 代理网关。今天把这套支撑了数千并发长连接的架构方案、核心避坑点以及可直接套用的工程代码完完整整掏出来，供需要自建 LLM API 网关的兄弟们参考。

在传统的 HTTP API 架构中，反向代理（如 Nginx、Traefik）或者简单的 httpx.AsyncClient 转发非常简单。但大模型的 流式长连接 + 高计算开销 特性彻底改变了游戏规则：

客户端断连算力空转（没有中途取消机制）： 用户在界面上点了“停止生成”或者直接关掉了浏览器标签页，如果你的代理网关没有捕获客户端断开的 TCP RST/FIN 包并同步通知推理后端，vLLM 会默默把剩余几千个 Token 全部算完。在高并发场景下，这种“幽灵请求”会吃掉 30% 以上的有效算力。
内存膨胀（背压失效）： vLLM 推理产出 Token 的速度极快，而下游客户端（特别是移动端或弱网用户）消费速度很慢。如果 Python 网关内部的缓冲区无上限，几百个流式请求就能把网关服务器的内存直接打爆。
多字节字符截断乱码： 中文在 UTF-8 编码下占用 3 个字节。大模型的 Tokenizer 在分词切分时，有可能会把一个多字节汉字切碎成跨 Token 输出。如果网关层在流式加工过滤敏感词或计费时强行做 bytes.decode('utf-8')，就会出现臭名昭著的乱码符号。

整体链路非常清晰：前端 Web/SDK 发起请求 -> Python 代理网关（负责鉴权、限流、背压调度、断连取消）-> 内网高性能 vLLM 实例集群。

首先，底层的 vLLM 实例推荐使用 Docker 配合以下生产级参数启动。我在 A100/H800 环境下测试过，这组配置能最大化并发吞吐：

# vLLM 生产启动命令参考
docker run --gpus all \
  -p 8000:8000 \
  --ipc=host \
  vllm/vllm-openai:latest \
  --model /models/Qwen2.5-72B-Instruct-AWQ \
  --tensor-parallel-size 2 \
  --max-model-len 8192 \
  --max-num-seqs 256 \
  --gpu-memory-utilization 0.90 \
  --trust-remote-code \
  --disable-log-requests

注意两点踩坑经验：一是务必加上 --ipc=host，多卡 Tensor Parallelism 进程间通信需要共享内存，否则高压下容易直接崩掉；二是显卡算力吃满时，关闭详细的请求 log（--disable-log-requests），避免高并发下 stdout 阻塞事件循环。

这是网关的核心实现代码。我们使用 FastAPI + httpx 构建异步网关，核心逻辑在于： 监听客户端断开事件，使用 asyncio.shield 与异步生成器生命周期控制 upstream 请求的 abort。

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

app = FastAPI(title="LLM Edge Gateway")

# 复用长连接 Client，合理配置池大小与超时
VLLM_BACKEND_URL = "http://10.0.0.12:8000/v1/chat/completions"
http_client = httpx.AsyncClient(limits=httpx.Limits(max_keepalive_connections=500, max_connections=1000),
    timeout=httpx.Timeout(connect=5.0, read=120.0, write=5.0, pool=5.0),
)

async def stream_generator(payload: dict, client_request: Request) -> AsyncGenerator[str, None]:
    """流式代理生成器：负责消费 upstream 数据，并持续探测下游客户端状态"""
    upstream_response = None
    try:
        # 向 vLLM 发起流式请求
        req = http_client.build_request("POST", VLLM_BACKEND_URL, json=payload)
        upstream_response = await http_client.send(req, stream=True)

        if upstream_response.status_code != 200:
            error_body = await upstream_response.aread()
            yield f"data: {json.dumps({'error': error_body.decode('utf-8')})}\n\n"
            return

        # 迭代读取 upstream 的 SSE 字节流
        async for raw_line in upstream_response.aiter_lines():
            # 关键点：检查前端客户端是否已经断开连接
            if await client_request.is_disconnected():
                # 记录日志，主动截断，触发 finally 清理
                break

            if not raw_line:
                continue

            # 处理多字节缓冲与文本输出
            yield f"{raw_line}\n\n"

    except asyncio.CancelledError:
        # 任务被外部协程取消时静默退出
        pass
    except Exception as e:
        yield f"data: {json.dumps({'error': str(e)})}\n\n"
    finally:
        # 核心保命操作：客户端一旦断开，必须强制关闭 upstream response
        # 这会促使底层连接向 vLLM 发送 RST/ 关闭流，从而中断 vLLM 的持续 decoding
        if upstream_response is not None:
            await upstream_response.aclose()


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

    # 强制将 stream 设置为 True 走流式输出
    payload["stream"] = True

    return StreamingResponse(stream_generator(payload, request),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no", # 严禁 Nginx 缓存 SSE 输出！},
    )

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

X-Accel-Buffering: no： 如果你在 FastAPI 前面还挂了一层 Nginx，Nginx 默认会把后端返回的数据攒到 proxy_buffer_size 才一次性推给客户端。这会导致用户看着光标干等 5 秒，然后一大坨文字突然跳出来。加上这个 Header 可以直接禁用 Nginx 缓冲区。
client_request.is_disconnected() 的触发时机： 该方法通过监听 ASGI 协议的 http.disconnect 消息生效。只要在生成器中逐行轮询，客户端断开后可以在 100ms 内立刻察觉并关闭后端通道。
UTF-8 IncrementalDecoder 机制： 如果你的网关层不仅做转发，还要在流式传输中提取文本做审计或 Token 计费，切记不要直接 chunk.decode('utf-8')。必须使用 Python 标准库的 codecs.getincrementaldecoder('utf-8')()，配合 decoder.decode(chunk, final=False)，才能正确拼装被切碎的汉字。

为了验证取消机制对推理后端的保护效果，我们在本地使用 Locust 模拟了 100 个长文本生成任务（每个任务 Prompt 长度 500 Tokens，要求生成 2000 Tokens）。测试场景模拟真实用户行为：其中 40% 的用户在收到前 100 个 Token 后主动点击取消。

代理网关方案	平均显存占用 (2*A100)	推理引擎 GPU 利用率	首字延迟 (TTFT)	有效生成吞吐 (Tokens/s)
普通透传转发 (无断连通知)	91.2% (频繁排队)	99.4% (高负荷)	1420ms	380 tok/s
本方案 (主动背压 + 断连传播)	68.4% (平稳)	71.8% (算力充足)	610ms	590 tok/s

数据非常直观：当被放弃的请求被立刻掐断后，vLLM 的 KV Cache 槽位被迅速释放归还给内存池，后续请求的调度延迟（Queue Time）大幅缩短，有效生成吞吐直接拉升了 55% 以上。

在搞大模型流式网关时，很多同学往往把重心全放在 Prompt 编排、Agent 流程设计上，而忽略了长连接基础底座的稳定性。总结这套方案的避坑经验：

能用异步流式就坚决不要在网关层做全量等待，把等待时间留给客户端交互。
必须实现从客户端到最底层推手端的断连传导链条，谁算力吃紧谁难受。
永远检查前置代理（Nginx / Ingress / Cloudflare）的超时时间配置与 Buffer 设置，proxy_read_timeout 建议至少拉到 300s，避免长推理任务在 60 秒时被外部代理暴力切断。

希望这份极客实操手册能帮你省掉几十个小时抓包排查内存和显卡负载的痛苦时间。如果部署中有任何疑问，欢迎在评论区留言交流！

正文完