共计 5074 个字符，预计需要花费 13 分钟才能阅读完成。
上个月我协助排查了一套线上 AI Agent 网关的故障：前端并发一拉高，vLLM 后端 GPU 显存直接爆满，网关机器内存也持续飙升，最后甚至触发了 Linux 的 OOM Killer。排查日志后发现了一个典型的“静默灾难”——用户在 Web 页面只要频繁刷新或中途点击“停止生成”，前端连接虽然瞬间切断，但我们用 FastAPI 写的代理网关依然像个老黄牛一样，默默把上游推理模型吐出的每一个 Token 接收完毕，直到完整文本生成才结束。这意味着昂贵的 GPU 算力在疯狂为空白输出空转，同时网关的连接句柄与内存被大量悬空请求死死拖垮。
一、核心痛点：为什么普通的 StreamingResponse 会搞崩生产环境？
在 Python 生态里，使用 fastapi.responses.StreamingResponse 包装 httpx.AsyncClient 转发 SSE（Server-Sent Events）是最常见的写法。但新手甚至很多资深工程师在写原型代码时，往往会写出如下的“逻辑陷阱”：
# 常见但存在严重隐患的代理写法
@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_URL, json=body, timeout=60.0) as resp:
async for chunk in resp.aiter_raw():
yield chunk
return StreamingResponse(event_generator(), media_type="text/event-stream")这段代码在平时单人调试时毫无破绽，但一旦上生产，会立刻暴露出三大致命缺陷：
- 客户端断连不可知（僵尸请求）： 用户关闭网页或网络波动断开时，FastAPI 默认不会主动打断下游生成器。
resp.aiter_raw()会继续消费上游数据，大模型依然在持续推理，GPU 算力被白白榨干。 - 连接池开销爆炸： 每个请求内部临时创建并销毁
httpx.AsyncClient()，完全丧失了连接复用（Keep-Alive）与连接池限制的能力，在高并发下瞬间耗尽系统 Ephemeral Ports 并引发大量TIME_WAIT。 - 下游背压（Backpressure）缺失： 如果客户端是个网络极慢的弱网设备，网关从千兆内网的推理集群快速拉取数据，全积压在 Python 运行时的内存缓冲区里，并发上千时直接 OOM。
二、架构重构：毫秒级断连感知与生命周期管理
要构建一个工业级的流式代理网关，我们必须做两件事：首先，必须借助 ASGI 协议的 request.is_disconnected() 机制实时感知下游客户端状态；其次，利用全局共享的 httpx.AsyncClient 实例并严格管控连接生命周期与异常中断。
我们来看优化后的核心异步生成器实现逻辑：
import logging
import anyio
import httpx
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
logger = logging.getLogger("gateway")
# 全局复用高性能连接池
client_pool = httpx.AsyncClient(limits=httpx.Limits(max_keepalive_connections=200, max_connections=1000),
timeout=httpx.Timeout(connect=5.0, read=60.0, write=10.0, pool=5.0)
)
async def stream_with_abort_detection(request: Request, upstream_resp: httpx.Response):
"""感知客户端中断并安全切断上游的异步流生成器"""
try:
async for chunk in upstream_resp.aiter_raw():
# 核心检测点：如果客户端已断开，立即终止迭代
if await request.is_disconnected():
logger.warning("客户端提前断开连接，正在终止上游推理...")
break
yield chunk
# 让渡事件循环调度权，确保 is_disconnected 的 ASGI 消息得到即时处理
await anyio.sleep(0)
except anyio.get_cancelled_exc_class():
logger.info("ASGI 任务被上层取消")
raise
except Exception as exc:
logger.error(f"流式转发异常: {str(exc)}")
raise
finally:
# 必须显式关闭上游流，触发底层 TCP FIN，通知 vLLM/Ollama 释放 KV Cache
await upstream_resp.aclose()
logger.info("上游流已安全关闭，算力资源已释放")技术细节拆解：
在上面的代码中，await anyio.sleep(0) 是极其关键的一笔。因为在 Python 的 asyncio 调度器里，如果网络 IO 吞吐极其密集，事件循环可能会被单一的数据读取循环霸占。微小的让渡可以让 ASGI 服务器（Uvicorn）有机会去读取来自反向代理或客户端的 http.disconnect 消息，进而让 request.is_disconnected() 在毫秒级内返回 True。
三、生产级网关路由完整落地代码
在网关工程化实现中，请求头处理和异常兜底同样不能忽视。尤其是诸如 Content-Encoding、Transfer-Encoding 这种逐跳头（Hop-by-hop headers）如果直接原样透传，很容易导致下游客户端协议解析崩溃。
from contextlib import asynccontextmanager
from fastapi import status
UPSTREAM_INFERENCE_URL = "http://192.168.1.100:8000/v1/chat/completions"
@asynccontextmanager
async def lifespan(app: FastAPI):
yield
# 优雅停机：释放全局连接池
await client_pool.aclose()
app = FastAPI(lifespan=lifespan)
# 必须剥离的逐跳请求头
EXCLUDED_HEADERS = {
"content-length", "connection", "keep-alive",
"proxy-authenticate", "proxy-authorization",
"te", "trailers", "transfer-encoding", "upgrade"
}
@app.post("/v1/chat/completions")
async def proxy_chat(request: Request):
req_body = await request.body()
# 过滤请求头
forward_headers = {k: v for k, v in request.headers.items()
if k.lower() not in EXCLUDED_HEADERS and k.lower() != "host"
}
try:
req = client_pool.build_request(
method="POST",
url=UPSTREAM_INFERENCE_URL,
headers=forward_headers,
content=req_body
)
upstream_resp = await client_pool.send(req, stream=True)
if upstream_resp.status_code != status.HTTP_200_OK:
error_body = await upstream_resp.aread()
await upstream_resp.aclose()
return HTTPException(
status_code=upstream_resp.status_code,
detail=error_body.decode(errors="replace")
)
# 构造给客户端的响应头，强制禁止任何中间层缓冲
response_headers = {
"Cache-Control": "no-cache",
"Content-Type": "text/event-stream",
"Connection": "keep-alive",
"X-Accel-Buffering": "no", # 专门用于通知 Nginx 关闭代理缓冲
}
return StreamingResponse(stream_with_abort_detection(request, upstream_resp),
status_code=upstream_resp.status_code,
headers=response_headers,
media_type="text/event-stream"
)
except httpx.ConnectError:
logger.error("无法连接到推理后端集群")
raise HTTPException(status_code=503, detail="推理引擎连接不可用")四、基石配置：Nginx 层的三大深坑
很多开发者写好了 Python 代码，但在前面架设 Nginx 作为外网入口后，流式输出立刻变成了“挤牙膏”甚至一次性倾倒，同时客户端断连事件彻底失灵。这是因为 Nginx 默认的缓冲与长连接机制在作祟。在网关前面的 Nginx 站点配置必须配置以下指令：
location /v1/chat/completions {
proxy_pass http://127.0.0.1:8080;
# 1. 必须使用 HTTP/1.1，否则 upstream 默认会退化为 HTTP/1.0 导致 chunk 协议失效
proxy_http_version 1.1;
proxy_set_header Connection "";
# 2. 彻底关闭代理缓冲，实现零延迟 Token 流式透传
proxy_buffering off;
proxy_cache off;
# 3. 关键：关闭客户端断开连接忽略，一旦用户掐线，Nginx 立即掐断与 Python 的连接
proxy_ignore_client_abort off;
# 4. 延长推理超时时间，避免长文本场景下触发 504
proxy_read_timeout 600s;
proxy_send_timeout 600s;
} 其中的 proxy_ignore_client_abort off; 是整个链条的胜负手。如果不显式配置这一项，即使前端页面关掉了，Nginx 依然会维持与后端的连接并默默接收数据，你的 Python request.is_disconnected() 永远不会被触发！
五、实测验证与性能表现
为了验证这套网关的鲁棒性，我在 4C8G 的网关机器上部署了该服务，上游挂载了一台配置了 DeepSeek-R1-Distill 的 8 卡 H800 推理节点。使用 Locust 进行并发中断模拟测试：
- 场景设置： 启动 500 个并发并发请求，模拟每个用户接收到 50 个 Token 后随机主动断开连接（模拟只读了开头就不感兴趣直接退出的用户行为）。
- 改动前（朴素写法）： 尽管客户端已经断开，推理节点仍然在持续跑完整段几千字的输出。GPU 整体算力被“无效生成”占满（Utilization 维持在 96% 以上），网关内部同时持有了上千个 TCP 连接与缓冲区，系统负载高达 12.4。
- 改动后（安全流式代理）： 客户端断开后，网关在 10ms 内捕获断开信号，随即主动关闭上游流，vLLM 引擎立刻收到了 TCP 终止信号并及时废弃对应 Request ID 的解码任务。GPU 利用率显著回落，整体算力资源浪费降低了 68% 以上 ，内存占用稳定在 200MB 以内没有任何波动。
总结与避坑心得
构建大模型基础设施时，网关往往是最容易被轻视但最容易翻车的地方。打磨生产级流式代理的核心原则非常朴素：
