共计 5223 个字符，预计需要花费 14 分钟才能阅读完成。
上个月我们在内网部署了一套 DeepSeek-R1 满血版集群，底层用 vLLM 跑推理。为了统一管理 Token 额度、敏感词过滤以及多节点负载均衡，我用 Python 写了一个轻量级的 API 网关。本以为一个 StreamingResponse 随手转发就完事了，结果上线灰度压测第一天，GPU 节点风扇呼啸狂转，显存直接被打爆，网关报错 ClientDisconnected 和 Too many open files 满天飞。
许多人用 Python 写反向代理时，很容易低估“流式传输（Server-Sent Events / SSE）”在高并发下对网关资源的侵蚀。LLM 的推理耗时极长，一个包含数千 Token 输出的长文本请求往往会维持数十秒的长连接。如果网关层处理不当，不仅会拖垮 Web 服务，还会把极其宝贵的 GPU 算力白白耗死在“用户早就关掉网页”的无效推理上。
这篇文章把我踩坑数周提炼出来的架构与优化细节全盘托出，主要聊透三个硬核问题：连接池复用与防泄漏 、 客户端断连感知与上游任务取消 、以及 非阻塞的零拷贝流式 Token 统计。
一、最简 Demo 为何会在生产环境雪崩？
很多开源项目或教程里给出的流式转发代理代码，基本上都长这样：
# 别在生产环境直接这么写！@app.post("/v1/chat/completions")
async def chat_proxy(request: Request):
payload = await request.json()
async with httpx.AsyncClient() as client: # 灾难根源 1：每次请求新建客户端
async with client.stream("POST", "http://vllm-backend:8000/v1/chat/completions", json=payload) as upstream_response:
async def event_generator():
async for chunk in upstream_response.aiter_bytes():
yield chunk
return StreamingResponse(event_generator(), media_type="text/event-stream")
这段代码跑单测毫无破绽，但在 200 并发压测下会瞬间引发级联雪崩：
- TCP 握手与 TIME_WAIT 爆炸：每个请求都临时
httpx.AsyncClient()，瞬时高并发下操作系统的端口直接被打空，文件描述符（FD）枯竭。 - 算力黑洞（Zombie Generation）：用户只要手抖刷新网页或者点击了“停止生成”，浏览器断开连接，而 FastAPI 依然在默默消耗 CPU 将上游 vLLM 的输出读完；更致命的是，vLLM 根本不知道客户端挂了，依旧全力以赴霸占显卡跑完整整几千个 Token。
二、架构升级：核心代理网关实现
我们要构建的是一个生产级流式代理。核心要求是：单例共享长连接池、精细化超时控制、双向断连检测，以及流式字节分块解析。
1. 全局 HTTPX 连接池配置
在 FastAPI 应用的声明周期管理（Lifespan）中初始化全局单例 httpx.AsyncClient，并对 Keep-Alive、最大连接数和超时做针对大模型场景的定制：
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
import httpx
import logging
logger = logging.getLogger("llm_gateway")
# 全局共享 HTTP 客户端
http_client: httpx.AsyncClient | None = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global http_client
# 针对 LLM 长耗时连接优化的连接池参数
limits = httpx.Limits(
max_connections=1000, # 最大并发连接数
max_keepalive_connections=200, # 保持活跃的空闲连接
keepalive_expiry=60.0 # 针对高频调用保持长连接
)
# LLM 首字延迟可能长达数十秒，read 超时必须放宽，但 connect 必须严格
timeout = httpx.Timeout(
connect=5.0,
read=180.0,
write=10.0,
pool=10.0
)
http_client = httpx.AsyncClient(limits=limits, timeout=timeout)
yield
await http_client.aclose()
app = FastAPI(lifespan=lifespan)
2. 客户端断连感知与上游任务取消机制
这是整个网关省下几十万算力成本的核心代码。我们需要在消费上游生成器的同时，主动检测下行网络状态。一旦发现 await request.is_disconnected() 为真，立刻抛出异常、中止迭代，并通过上下文退出主动关闭向上游 vLLM 发起的请求流，触发上游的 Abort 逻辑。
import asyncio
async def stream_proxy_generator(
request: Request,
upstream_url: str,
payload: dict,
headers: dict
):
try:
# 发起流式请求
req = http_client.build_request("POST", upstream_url, json=payload, headers=headers)
resp = await http_client.send(req, stream=True)
if resp.status_code != 200:
error_body = await resp.aread()
yield f"data: {{\"error\": \"Upstream returned {resp.status_code}\", \"details\": {error_body.decode('utf-8')}}}\n\n".encode("utf-8")
return
# 流式传输迭代
async for chunk in resp.aiter_raw():
# 关键：检查客户端是否已经挂断连接
if await request.is_disconnected():
logger.warning("客户端已主动断开，立即终止上游流式传输并释放算力！")
break
yield chunk
except asyncio.CancelledError:
logger.info("网关层接收到 Task Cancelled 信号，安全退出。")
raise
except Exception as e:
logger.error(f"代理链路异常: {str(e)}", exc_info=True)
yield f"data: {{\"error\": \"Proxy internal exception\"}}\n\n".encode("utf-8")
finally:
# 无论发生何种情况，必须确保底层 response stream 关闭
if 'resp' in locals() and not resp.is_closed:
await resp.aclose()
3. 网关路由接入
在 FastAPI 路由处理中，必须透传上游响应头（如 Cache-Control: no-cache、X-Accel-Buffering: no），避免 Nginx 或 Cloudflare 在网关层外围强行开启缓冲，导致“字憋了 10 秒后一次性吐出来”的糟糕体验。
@app.post("/v1/chat/completions")
async def chat_endpoint(request: Request):
body = await request.json()
upstream_url = "http://10.0.0.12:8000/v1/chat/completions"
# 提取必要的请求头，规避 Host 头污染
forward_headers = {"Authorization": request.headers.get("Authorization", ""),"Content-Type":"application/json","Accept":"text/event-stream"
}
# 如果非流式调用，走常规快速通道
if not body.get("stream", False):
res = await http_client.post(upstream_url, json=body, headers=forward_headers)
return StreamingResponse(res.iter_raw(), status_code=res.status_code, media_type="application/json")
# 流式调用通道
return StreamingResponse(stream_proxy_generator(request, upstream_url, body, forward_headers),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no" # 强行禁用 Nginx 缓冲
}
)
三、零拷贝 Token 计费：边读边算且不破坏流
很多业务网关需要在流式输出结束后记录这通请求生成了多少 Token。如果在生成器里反复把 chunk 解码成 JSON、提取 content、拼成完整文本再调用 tiktoken，不仅 CPU 占用率飙升，还会引入显著的 TTFT（Time to First Token）和逐字延迟。
推荐方案：利用 OpenAI/vLLM 兼容的 stream_options: {"include_usage": True} 特性。模型服务在吐出最后一个 [DONE] 数据包之前，会单独推送一个携带整体 usage 信息的 JSON 块。
我们在转发生成器内部加入一个轻量级的双指针状态机或正则表达式，仅在每个 Chunk 的末尾截取匹配，做到 O(1) 级别解析：
import re
# 预编译正则，快速捞出 usage 数据块，完全无需解析全量 JSON
USAGE_PATTERN = re.compile(rb'"usage"\s*:\s*(\{[^}]+\})')
async def stream_with_accounting(request: Request, upstream_url: str, payload: dict, headers: dict):
# 强制上游吐出 usage 块
if "stream_options" not in payload:
payload["stream_options"] = {"include_usage": True}
usage_data = None
async for chunk in stream_proxy_generator(request, upstream_url, payload, headers):
# 极轻量的高性能抓取
if b'"usage"' in chunk:
match = USAGE_PATTERN.search(chunk)
if match:
usage_data = match.group(1).decode("utf-8")
# 实时交还给客户端，完全不等待
yield chunk
# 请求结束后异步打点入库，不阻塞请求生命周期
if usage_data:
asyncio.create_task(record_usage_to_db(payload.get("user", "anonymous"), usage_data))
async def record_usage_to_db(user: str, usage_str: str):
logger.info(f"User [{user}] Token Consumed: {usage_str}")
# 此处打入 Redis 或 ClickHouse
四、性能对比与调优成果
在 4 台部署了 vLLM (8x A100) 的内网推理服务器前，我使用 Locust 模拟 500 个并发长会话（Prompt 平均 800 tokens，输出平均 1200 tokens），对优化前后的网关进行了对比：
| 指标项 | 原生简单转发 (无池化 / 无断连) | 优化后网关 (池化 + 断连背压) |
|---|---|---|
| 单机最大承载并发 | ~140 并发报错 (端口耗尽 /OOM) | 1200+ 并发 稳定运行 |
| 首字延迟 (TTFT) | 840ms (高并发排队) | 110ms (长连接复用) |
| 异常断连算力损耗 | 100% (上游死跑到底) |
