共计 5097 个字符，预计需要花费 13 分钟才能阅读完成。
上周在帮内部 Agent 平台重构网关层时，遇到了一个极其典型却容易被忽略的事故：前端用户提问后，觉得生成速度慢或者问题输错了，频繁点击“停止生成”或直接关掉网页，然而后端的 vLLM 集群依然在疯狂吐字，显卡显存和算力全部被毫无意义的推理占满；同时，FastAPI 服务在高峰期连接数暴增，最终抛出 Too many open files 崩溃。
很多开发者用 Python 写 LLM 网关，以为简单的 StreamingResponse(client.chat.completions.create(stream=True)) 就完事了。实际上，流式代理（SSE, Server-Sent Events）是有状态且长生命周期的连接，一旦缺乏背压控制、连接池优化以及断连感知机制，只要稍微来点并发，服务分分钟被击穿。这篇文章直接贴出我经过压测验证的生产级代码，聊聊怎么把这个流式代理做扎实。
一、核心痛点拆解：为什么你写的流式代理会崩？
在普通 HTTP 请求中，Request 和 Response 周期极短（几十毫秒到几百毫秒），但在 LLM 流式场景下，一个请求可能持续输出 30 秒甚至数分钟。这引发了三个致命问题：
- 算力幽灵消耗（Zombie Inferences）： 客户端断开连接（如关掉浏览器），上游应用服务器如果不检测 TCP 连接状态，下游继续向 OpenAI / vLLM 读取并等待后续 Token，白白烧钱烧 GPU。
- HTTPX 连接池枯竭与泄漏： 很多教程教大家在函数内部临时创建
async with httpx.AsyncClient() as client:，在高并发下频繁进行 TCP/TLS 握手，连接池不仅起不到复用效果，还会产生海量TIME_WAIT状态。 - 反向代理缓冲导致的“假死”： Nginx 默认会缓冲上游响应，流式输出被攒成一个大包才发给客户端，导致 TTFT（首字响应时间）严重劣化。
二、架构设计与生产级代码实现
我们需要利用 asyncio 的协同调度机制，把 FastAPI 的 Request.is_disconnected()、异步生成器以及 httpx.AsyncClient 的上下文做深度绑定。
先看完整的网关核心代码实现（保存为 llm_gateway.py）：
import asyncio
import logging
from contextlib import asynccontextmanager
from typing import AsyncGenerator
import httpx
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import StreamingResponse
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("llm-proxy")
# 全局 httpx 客户端，复用连接池
http_client: httpx.AsyncClient | None = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global http_client
# 调优连接池：流式响应长连接多，max_keepalive_connections 和 max_connections 需调高
limits = httpx.Limits(max_keepalive_connections=500, max_connections=2000, keepalive_expiry=60.0)
# 超时配置：流式输出 connect 设短，read 设长
timeout = httpx.Timeout(connect=5.0, read=120.0, write=10.0, pool=10.0)
http_client = httpx.AsyncClient(limits=limits, timeout=timeout)
logger.info("HTTPX 连接池已初始化")
yield
await http_client.aclose()
logger.info("HTTPX 连接池已优雅关闭")
app = FastAPI(lifespan=lifespan)
UPSTREAM_LLM_URL = "http://10.0.0.12:8000/v1/chat/completions" # 你的 vLLM 或 Ollama 地址
UPSTREAM_API_KEY = "EMPTY"
async def stream_generator(request: Request, payload: dict) -> AsyncGenerator[bytes, None]:
headers = {"Authorization": f"Bearer {UPSTREAM_API_KEY}",
"Content-Type": "application/json",
}
# 确保上游以流式返回
payload["stream"] = True
try:
# 使用 build_request + send，方便精细控制流
req = http_client.build_request("POST", UPSTREAM_LLM_URL, json=payload, headers=headers)
response = await http_client.send(req, stream=True)
if response.status_code != 200:
error_body = await response.aread()
logger.error(f"上游错误: {response.status_code} - {error_body.decode()}")
yield f"data: {{\"error\": \"Upstream returned {response.status_code}\"}}\n\n".encode()
return
async with response:
async for chunk in response.aiter_bytes():
# 核心机制：检查客户端是否主动断开（网络掉线或用户点停止）if await request.is_disconnected():
logger.warning("客户端已主动断开，立即终止上游流拉取，释放算力！")
break
# 透传 chunk 给前端
yield chunk
except asyncio.CancelledError:
# 当客户端强制 abort 请求，FastAPI 会触发 CancelledError
logger.info("捕获到 asyncio.CancelledError，安全退出协程")
raise
except httpx.ReadTimeout:
logger.error("读取上游模型超时，模型可能卡死或显存溢出")
yield b"data: {\"error\": \"LLM Inference Timeout\"}\n\n"
except Exception as e:
logger.exception(f"流式转发未知异常: {str(e)}")
yield f"data: {{\"error\": \"Proxy internal error: {str(e)}\"}}\n\n".encode()
finally:
logger.debug("流式管道资源清理完成")
@app.post("/v1/chat/completions")
async def chat_proxy(request: Request):
try:
body = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="Invalid JSON")
# 关键点：禁用响应缓冲，设置标准的 text/event-stream 响应头
return StreamingResponse(stream_generator(request, body),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no", # 告知 Nginx 不要缓冲响应！}
)
三、核心避坑要点与排障经验
1. 为什么不能漏掉 X-Accel-Buffering: no？
很多极客在本地用 uvicorn 测流式输出很丝滑，一部署到带 Nginx 反代的生产机器上就发现：前端总是卡上 3~5 秒，然后一大坨文字突然同时蹦出来。这就是典型的被 Nginx 缓冲区吃掉导致的。在响应头显式加上 X-Accel-Buffering: no，配合 Nginx 配置：
location /v1/chat/completions {
proxy_pass http://127.0.0.1:8000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
# 流式特调关键配置
proxy_http_version 1.1;
proxy_set_header Connection "";
proxy_buffering off;
proxy_cache off;
proxy_read_timeout 300s; # 避免生成长文本时超时被 nginx 切断
chunked_transfer_encoding on;
}2. 客户端断开感知（Disconnection Detection）
在上面的 Python 代码中，我们使用了 await request.is_disconnected()。注意：FastAPI 底层基于 Starlette，is_disconnected() 的原理是探测客户端 TCP socket 的 receive() 状态。如果在一次循环读取中客户端断开了，HTTP/2 会抛出 Reset Frame，HTTP/1.1 则关闭连接，ASGI server 会将其包装为取消事件。如果不显式 break，生成器会继续阻塞在 response.aiter_bytes()，导致显卡依然在推理下文，白白浪费成百上千个输出 token。
3. HTTP 客户端连接池生命周期管理
千万不要在每个请求里写 async with httpx.AsyncClient()。在我的单机压测中（使用 Locust 模拟 500 并发流式打入）：
- 临时 Client 方案： 随着连接快速建立和销毁，系统端口号迅速占满，大量
Cannot assign requested address错误，且 CPU 被 TLS 协商占满，首字响应时间（TTFT）劣化到 1800ms。 - 全局 Client 复用连接池： 端口稳定，保持持久的 TCP Keep-Alive，TTFT 稳定在 220ms 左右，CPU 占用率下降约 45%。
四、压测与验证命令
你可以直接用 curl 验证流式代理是否生效以及中断机制：
# 测试流式透传是否流畅
curl -N -X POST http://localhost:8000/v1/chat/completions \
-H "Content-Type: application/json" \
-d '{"model":"qwen-7b","messages": [{"role":"user","content":" 写一篇 3000 字关于火星移民的科幻小说 "}]}'
# 测试中断感知：直接按 Ctrl+C 终止上面的 curl 命令
# 然后观察网关后台日志，应该秒级输出：# WARNING:llm-proxy: 客户端已主动断开，立即终止上游流拉取，释放算力！总结与避坑心得
把 Python 仅仅当作“胶水语言”写 LLM 业务逻辑很容易，但在并发场景下，长连接运维才是核心内功。总结三条硬原则：
- 长连接必须池化： 跨服务转发一定要维护全局
httpx.AsyncClient，合理设置 Keepalive 数量与读写超时时间。 - 流式响应全链路去缓冲： 从代码层
media_type="text/event-stream"，到网关头X-Accel-Buffering: no，再到 Nginxproxy_buffering off，链路任何一环开了缓冲，流式体验就荡然无存。 - 警惕算力空转： 做 LLM 网关，中断检测不是锦上添花，而是省钱保活的刚需。前端只要 abort，后端必须无条件截断上游拉取。
按照这套方案搭建的流式网关，我目前在单台 4C8G 的机器上稳定调度着两台 4x A100 的推理集群，即使在面对突发突刺流量时，网关自身的 CPU 和内存消耗也极低，完全告别了内存泄漏与无用推理。
