共计 5415 个字符，预计需要花费 14 分钟才能阅读完成。
上个月我把公司内部的知识库 Agent 接入国产 DeepSeek 和线上 OpenAI 时，网关服务连续两天在下午业务高峰期莫名其妙地出现 Worker 假死。查看监控，CPU 利用率并不高，但内存曲线呈阶梯状一路狂飙直到 OOM，同时大量的客户端开始报 504 Gateway Timeout。
很多开发者用 Python 写大模型中继代理时，逻辑通常非常简单：用 httpx 发请求，把得到的响应塞进 FastAPI 的 StreamingResponse。如果你在本地写 Demo 或者单并发测试，这种写法丝滑无比；但在 50+ 并发、长上下文（32k+ tokens）、耗时普遍在 10 秒以上的流式连接冲击下，这种“玩具级”写法会迅速把服务器拖垮。
一、玩具代码背后的三大致命死穴
先看一段最容易在网上见到的“标准”流式转发代码：
# 典型反面教材：千万不要直接扔到生产环境
@app.post("/v1/chat/completions")
async def chat_proxy(request: Request):
payload = await request.json()
async def event_generator():
client = httpx.AsyncClient()
async with client.stream("POST", UPSTREAM_URL, json=payload) as resp:
async for chunk in resp.aiter_bytes():
yield chunk
return StreamingResponse(event_generator(), media_type="text/event-stream")这段代码在真实生产流量下有三个足以致命的缺陷：
- 客户端生命周期失控（Client Disconnect）：前端用户关掉标签页或点击“停止生成”时，默认情况下 FastAPI 的生成器并不会立刻中断上游连接。上游大模型还在继续吐字计费，下游代理还在傻傻缓冲，直接浪费算力和带宽。
- 连接池与 Client 频繁创建销毁：在生成器内部实例化
httpx.AsyncClient()，高并发下会导致大量处于TIME_WAIT的 TCP 连接，端口迅速耗尽。 - 无法在流式中无感计费与审计：流式传输直接转发了原始字节，若想提取返回文本统计 Token 消耗量，传统做法要么让前端上报，要么在内存中完整拼接整个字符串，这在长回复时极易造成内存驻留峰值。
二、工业级异步流式网关实现
针对上述痛点，我们需要设计一个包含 全局连接池复用 、 下游断开实时检测 、 流式 Chunk 无额外内存分配解析 的代理中继。以下是提炼自生产环境的完整核心模块：
import json
import logging
from typing import AsyncGenerator
import httpx
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
import tiktoken
app = FastAPI()
logger = logging.getLogger("llm_proxy")
# 1. 生产环境严禁临时创建 Client，维护全局单例并合理配置超时与连接池
limits = httpx.Limits(max_keepalive_connections=100, max_connections=500, keepalive_expiry=30.0)
timeout = httpx.Timeout(connect=5.0, read=60.0, write=10.0, pool=10.0)
http_client = httpx.AsyncClient(limits=limits, timeout=timeout)
@app.on_event("shutdown")
async def shutdown_event():
await http_client.aclose()
async def stream_processor(
request: Request,
upstream_response: httpx.Response,
encoding_name: str = "cl100k_base"
) -> AsyncGenerator[bytes, None]:
"""
流式代理核心：- 监听下游断开信号
- 逐块产出 SSE
- 统计 Output Tokens
"""
total_tokens = 0
enc = tiktoken.get_encoding(encoding_name)
try:
async for raw_line in upstream_response.aiter_lines():
# 关键：检查客户端是否已经断开连接
if await request.is_disconnected():
logger.warning("Downstream client disconnected. Aborting upstream task.")
break
if not raw_line:
continue
# 透传原始数据流
yield f"{raw_line}\n\n".encode("utf-8")
# 仅处理 SSE 规范的数据行（避免全量 JSON decode 开销）if raw_line.startswith("data:"):
data_str = raw_line[6:].strip()
if data_str == "[DONE]":
break
try:
chunk_json = json.loads(data_str)
choices = chunk_json.get("choices", [])
if choices:
delta = choices[0].get("delta", {})
content = delta.get("content", "")
if content:
# 增量计算 Token，不拼接全局长文本，避免大堆内存驻留
total_tokens += len(enc.encode(content, disallowed_special=()))
except json.JSONDecodeError:
continue
except Exception as e:
logger.error(f"Error during stream forwarding: {str(e)}")
raise
finally:
# 在这里执行异步审计或数据库扣费写入
logger.info(f"Stream finished. Total output tokens metered: {total_tokens}")
# 显式关闭上游流式连接
await upstream_response.aclose()
@app.post("/v1/chat/completions")
async def proxy_chat(request: Request):
body = await request.body()
# 构建转发请求（剥离不必要的 Host 头）headers = dict(request.headers)
headers.pop("host", None)
headers.pop("content-length", None)
# 建立上游长连接
req = http_client.build_request(
method="POST",
url="https://api.openai.com/v1/chat/completions",
headers=headers,
content=body
)
try:
upstream_resp = await http_client.send(req, stream=True)
except httpx.ConnectError:
raise HTTPException(status_code=502, detail="Upstream LLM Provider Unreachable")
except httpx.PoolTimeout:
raise HTTPException(status_code=503, detail="Proxy Worker Exhausted")
# 如果上游直接报错（例如 401、429），直接回抛结构化错误，不要继续走流式
if upstream_resp.is_error:
error_content = await upstream_resp.aread()
return StreamingResponse(iter([error_content]),
status_code=upstream_resp.status_code,
media_type="application/json"
)
return StreamingResponse(stream_processor(request, upstream_resp),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no" # 关掉 Nginx 缓冲区，保证打字机实时体验
}
)三、踩坑记录：高并发下必须修复的三个底层细节
1. Nginx 反向代理层吞掉首字延迟（TTFT）
在本地调试很正常，一旦部署到 Linux 服务器、前面顶了一层 Nginx，很多同学就会发现流式输出变成了“一段一段吐”，甚至等到模型生成了四五百个字才突然哗啦一下全部显示出来。
这是因为 Nginx 默认开启了 proxy_buffering。大模型的 SSE 传输本质是小分块 HTTP 报文，Nginx 会尝试把它攒满一个 proxy_buffer_size（通常是 4k/8k）才发给前端。除了在应用层加上 X-Accel-Buffering: no 头之外，在 Nginx 对应 Location 块中必须显式指定：
location /v1/chat/completions {
proxy_pass http://127.0.0.1:8000;
proxy_http_version 1.1;
proxy_set_header Connection "";
# 强制关闭缓冲，保障极低首字时延
proxy_buffering off;
proxy_cache off;
chunked_transfer_encoding on;
}2. Uvicorn Worker 数量与事件循环饥饿
很多小伙伴习惯用 gunicorn -w 4 -k uvicorn.workers.UvicornWorker。但在 LLM 这种纯粹的 I/O 密集、高长连接占比场景下，每个 Worker 的事件循环中挂载了上千个可能持续几十秒的连接。
如果在生成器内部进行了耗时的 CPU 密集操作（比如同步执行高耗时的正则脱敏，或者在同一线程中使用 tiktoken 编码长达几千字的 prompt），会直接把当前 Worker 的 Event Loop 堵死，导致其他所有正在流式输出的连接瞬间卡顿。
实操原则：
- 所有的分词计量必须分片化，绝对不要在流式循环中一次性加载大文本。
- 如果有敏感词过滤或内容安全审查，必须交给
asyncio.to_thread放到单独线程池，或者通过 Redis 队列转交给独立 Worker 处理。
3. `request.is_disconnected()` 的轮询代价与陷阱
在 FastAPI 中，request.is_disconnected() 是一个异步方法，它底层依赖 Starlette 的接收队列。如果你的客户端主动断开连接，只要你没有调用 await request.receive()，该状态有时无法立即更新。
在上面的代码实现中，我们在 async for raw_line in upstream_response.aiter_lines() 的每一次迭代中调用该方法。由于上游模型持续吐出 token（频率大约每秒 20~80 次），这恰好构成了天然的心跳检测探针。实测在断开连接后的 50ms 内即可探测到底层套接字断开，及时执行 break 进而触发 finally 中的 upstream_response.aclose()，直接为 GPU 推理后端释放了一个 Batch Slot。
四、实测压测数据对比
我们使用 Locust 模拟 200 个并发长连接持续请求大模型中继，每个请求要求输出约 1000 个 Tokens（耗时约 12 秒），连续压测 10 分钟：
| 方案 | 200 并发吞吐失败率 | 首字延迟 (TTFT P95) | 网关内存占用 (峰值) |
|---|---|---|---|
| 未优化的基础代码（每次实例化 Client） | 18.4% (大量的 ConnectTimeout) | 3800ms | 1.8 GB (且回落极慢) |
| 优化后方案（全局池化 + 实时防爆） | 0.0% | 420ms | 180 MB (平稳不抖动) |
总结与避坑心得
在 Python 异步生态中，大模型中继网关绝对不是一个简单的“requests 替换为 httpx”就能搞定的活。处理 SSE 流式传输的核心只有八个字：连接受控、资源闭环。
上线前务必在终端用 netstat -an | grep 8000 | awk '{print $6}' | sort | uniq -c 观察连接状态，如果看到数千个 CLOSE_WAIT 或 TIME_WAIT，就说明你的上游或下游资源没有正确在上下文退出时调用 aclose()。把本文的连接池配置和断开监测放到你的代码里，把流式数据当成流水而不是蓄水池，你的网关服务才能扛得住业务洪峰。
===EOF===
