共计 5495 个字符，预计需要花费 14 分钟才能阅读完成。
上个月我把公司内部的知识库 Agent 对接到底层私有化部署的 vLLM 和 DeepSeek API 时，直接遇到了生产事故：前端用户在生成长文本时频繁刷新页面或关闭标签页，但后台的 Python 异步服务毫无察觉，依然按部就班地从上游大模型节点吞吐 Token，硬生生把 GPU 推理卡队列占满，引发后续请求大面积排队超时，一晚上白白损耗了数千万无意义的 Token。
很多工程师用 FastAPI 写 LLM 接口时，习惯性用官方文档里的 StreamingResponse 随便套一个 httpx.stream() 就交差上线。但在真实生产流量下，网络抖动丢包、客户端中途关闭断连、跨模型厂商 SSE 格式不兼容、连接池耗尽导致的协程饥饿，分分钟教你做人。今天我们直接上干货，手拆一个工业级、低延迟且具备自动断流感知的 Python 流式网关。
一、为什么简单的 httpx.stream 会在生产翻车？
一个朴素的流式转发代码通常长这样：
# 典型错误示范：看似能跑，实则漏水
@app.post("/v1/chat/completions")
async def chat(request: Request):
client = httpx.AsyncClient()
response = await client.send(req, stream=True)
return StreamingResponse(response.aiter_raw(), media_type="text/event-stream")这套逻辑潜藏着三大致命隐患：
- 连接泄漏（Socket Leak）：每次请求实例化
AsyncClient，TCP 三次握手和 TLS 握手开销巨大，瞬间压测直接报Too many open files。 - 半开连接孤儿请求（Zombie Stream）：客户端网络断开或用户点了取消生成，ASGI 层的断开信号未被消费，生成器继续向上游拉取直到模型输出
<|end of sentence|>，昂贵算力被白白浪费。 - 流劫持解析困难：如果你想在流式传输过程中同步计算计费 Token 数，或者给敏感词打补丁，纯二进制透传根本无法插入业务中间件。
二、架构设计：带生命周期感知的高性能流中继
网关的核心职责是建立两段解耦的管道：客户端 <-> 网关 (SSE) 与 网关 <-> 上游推理集群 (HTTP/2 keep-alive)，并在网关层建立强生命周期检测。
│ (HTTP/1.1 SSE – 可随时断开)
▼
[FastAPI Gateway] ──> [asyncio.Task 监听 request.is_disconnected()]
│ (复用 HTTP/2 连接池 + 缓冲区背压控制)
▼
[Upstream vLLM / DeepSeek API] ──> (检测断开立即触发上游连接 close())
三、核心源码落地：生产级流式中继器
废话不多说，直接看我提炼后的核心模块代码，依赖仅需 fastapi, uvicorn[standard] 和 httpx：
# gateway_streamer.py
import asyncio
import json
import logging
from typing import AsyncGenerator
import httpx
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("stream_gateway")
app = FastAPI()
# 全局共享的长连接池，严格限制连接数并配置合理的超时时间
upstream_client: httpx.AsyncClient = None
@app.on_event("startup")
async def startup_event():
global upstream_client
limits = httpx.Limits(max_keepalive_connections=200, max_connections=1000, keepalive_expiry=60.0)
timeout = httpx.Timeout(connect=5.0, read=60.0, write=10.0, pool=5.0)
upstream_client = httpx.AsyncClient(limits=limits, timeout=timeout, http2=True)
@app.on_event("shutdown")
async def shutdown_event():
await upstream_client.aclose()
async def proxy_sse_stream(
raw_request: Request,
target_url: str,
payload: dict,
headers: dict
) -> AsyncGenerator[str, None]:
"""
流式代理生成器：1. 逐行读取上游 SSE 数据并进行缓冲校验
2. 轮询检测下游客户端连接状态
3. 支持中途 Token 计数与审计插桩
"""
total_tokens = 0
upstream_response = None
try:
req = upstream_client.build_request("POST", target_url, json=payload, headers=headers)
upstream_response = await upstream_client.send(req, stream=True)
if upstream_response.status_code != 200:
error_body = await upstream_response.aread()
logger.error(f"上游错误 [{upstream_response.status_code}]: {error_body.decode('utf-8', errors='ignore')}")
yield f"data: {json.dumps({'error':'Upstream LLM error','code': upstream_response.status_code})}\n\n"
return
async for chunk in upstream_response.aiter_lines():
# 关键检测点：下游客户端是否已经主动挂断（如关掉网页窗口）if await raw_request.is_disconnected():
logger.warning("客户端已断开，立即终止上游流管道，释放推理显存！")
break
if not chunk:
continue
# 处理 SSE 数据规范
if chunk.startswith("data:"):
data_content = chunk[6:].strip()
if data_content == "[DONE]":
yield "data: [DONE]\n\n"
break
try:
# 可以在此处无缝插入流式审核逻辑或 Token 累加
chunk_json = json.loads(data_content)
delta = chunk_json.get("choices", [{}])[0].get("delta", {})
if "content" in delta:
total_tokens += 1
except json.JSONDecodeError:
pass
yield f"{chunk}\n\n"
except asyncio.CancelledError:
logger.info("ASGI Task Cancelled，捕获到协程取消信号")
raise
except Exception as e:
logger.exception(f"流传输遇到未捕获异常: {str(e)}")
yield f"data: {json.dumps({'error':'Stream aborted unexpectedly'})}\n\n"
finally:
if upstream_response:
await upstream_response.aclose()
logger.info(f"流结束，本次请求代理上游实际传输约 {total_tokens} 个 Chunks")
@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
try:
body = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="Invalid JSON body")
# 强制将 stream 设为 True
body["stream"] = True
# 填入实际内网上游推理地址，如 vLLM 原生接口
UPSTREAM_URL = "http://10.0.0.12:8000/v1/chat/completions"
headers = {
"Authorization": "Bearer sk-your-internal-auth-token",
"Content-Type": "application/json"
}
return StreamingResponse(proxy_sse_stream(request, UPSTREAM_URL, body, headers),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no", # 核心：禁止 Nginx/ 反代缓冲区拦截流式传输
}
)四、关键细节与生产避坑实录
1. 警惕 Nginx 缓冲截流：X-Accel-Buffering: no
我曾遇到过本地通过 localhost:8000 压测流畅无比，一上测试环境挂了 Nginx 反向代理，前端就变成“卡顿 10 秒后一次性吐出所有文字”的情况。这是因为 Nginx 默认启用了 proxy_buffering。在网关响应头中强制下发 X-Accel-Buffering: no，或者在 Nginx 对应 location 配置里加入：
proxy_buffering off;
proxy_cache off;
proxy_set_header Connection '';
proxy_http_version 1.1;
chunked_transfer_encoding on;2. 协程并发饥饿与 Uvicorn 工作模型
Python 原生事件循环在处理密集型 JSON 序列化时依然存在单核性能瓶颈。如果你单节点开单 Worker 处理 500 个并发 SSE 管道，哪怕全是异步 I/O，每秒数万次 JSON parse 也会让 Event Loop 的延迟拉升到上百毫秒。
生产环境务必采用 Gunicorn 配合 UvicornWorker 的多进程模式，并将单核分配严格绑定在 CPU 计算节点上：
# 启动命令：4 个 worker，启用 uvloop 高性能驱动
gunicorn gateway_streamer:app \
--workers 4 \
--worker-class uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8080 \
--backlog 2048 \
--access-logfile -3. 断流测试：验证显卡显存是否及时释放
部署完成后，别光用 curl 测试成功路径。用 Python 写一段脚本故意建立请求后中途直接 sock.close()：
# test_drop.py
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(("127.0.0.1", 8080))
request = (
"POST /v1/chat/completions HTTP/1.1\r\n"
"Host: 127.0.0.1:8080\r\n"
"Content-Type: application/json\r\n"
"Content-Length: 108\r\n\r\n"
'{"model":"deepseek-coder","messages": [{"role":"user","content":" 写一篇一万字的长文 "}],"stream": true}'
)
s.sendall(request.encode())
# 接收几百字节后粗暴关闭连接
data = s.recv(512)
s.close()
print("客户端提前强行断开连接")此时查看网关后台日志，若能在 1 秒内看到 客户端已断开，立即终止上游流管道 且 vLLM 监控指标中该请求的状态直接变成 FINISHED_ABORTED，说明你已经成功堵住了这一致命资源漏洞。
总结与避坑心得
构建流式推理网关看似门槛不高，但真正将服务推向生产时，核心早已不是“调通 API”，而是 生命周期的精准控制与资源的按需截断。通过配置全局 HTTP/2 异步连接池、注入反向代理穿透头，并时刻监听底层 ASGI 的 is_disconnected 状态，我们不仅能大幅降低长连接消耗的 Socket 文件句柄，更能直接保住昂贵的上游 GPU 算力。后续如果你需要加入多模型动态灰度路由或敏感词流式打码替换，只需在这个生成器管道内插桩即可实现完全掌控。
