实战踩坑：基于 FastAPI 与 Asyncio 构建高并发 LLM 流式代理网关

4次阅读

共计 4435 个字符，预计需要花费 12 分钟才能阅读完成。

最近在为团队的一套多模型聚合服务做网关改造，后端挂着 vLLM 集群和若干商用 API（OpenAI、DeepSeek 等），前端统一走 Server-Sent Events (SSE) 流式打字机交互。原以为用 FastAPI 随手写一个 StreamingResponse 几行代码就能搞定，结果上压测工具跑了 500 并发，服务器内存瞬间飙升，还出现了大量客户端断开连接后后端大模型仍在持续跑推理、白白浪费算力和 token 费用的离谱现象。

很多开发者在写异步流式代理时，往往会低估网络 I/O 挂起、缓冲池膨胀和客户端非正常断连带来的连锁反应。今天这篇文章，我把自己在产线调优和踩坑总结出来的全套生产级实现方案彻底扒开，聊聊怎么用纯原生 Python + FastAPI 搞定一个高吞吐、零漏水、带主动断流的 LLM 流式代理网关。

大部分教程或脚手架里教的流式代理，大概率长成这样：

# 典型反例：玩具级代码，请勿直接上生产
@app.post("/v1/chat/completions")
async def chat_proxy(request: Request):
    body = await request.json()
    async with httpx.AsyncClient() as client:
        req = client.build_request("POST", UPSTREAM_URL, json=body)
        res = await client.send(req, stream=True)
        return StreamingResponse(res.aiter_bytes(), media_type="text/event-stream")

这段代码跑单测非常丝滑，但一上并发就原形毕露：

频繁新建与销毁 Client：每次请求都走 AsyncClient() 上下文，TLS 握手开销巨大，短时间内端口被 TIME_WAIT 状态打满。
没有连接超时与保活隔离：LLM 生成首字耗时（TTFT）通常较长，默认超时经常抛出 ReadTimeout，或者连接死锁假死。
幽灵调用（Ghost Requests）：当用户在网页端点击“停止生成”或直接关掉标签页时，ASGI 服务器虽然感知到了 HTTP 断开，但背后的 res.aiter_bytes() 依然在后台默默接收上游推流，后端算力集群根本不会停。

要解决上述问题，我们需要做三件事：全局长连接池复用 、 基于 request.is_disconnected() 的主动背压 / 断流 ，以及 无缓冲的块传输流式生成器。

必须将 httpx.AsyncClient 挂载到应用的声明周期（Lifespan）中，合理设置连接限制与长超时：

from contextlib import asynccontextmanager
from fastapi import FastAPI
import httpx

# 针对大模型场景深度定制的 Limits 与 Timeout
http_limits = httpx.Limits(
    max_connections=2000,          # 允许的最大并发连接数
    max_keepalive_connections=500, # 保持活跃的空闲连接数
    keepalive_expiry=30.0          # 保活过期时间（秒）)

http_timeout = httpx.Timeout(
    connect=10.0,  # 建立连接不能超时
    read=60.0,     # 单块数据读取超时（防止模型卡死）write=10.0,
    pool=15.0      # 从连接池拿不到连接的排队上限
)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 容器初始化时启动全局客户端池
    app.state.client = httpx.AsyncClient(
        limits=http_limits,
        timeout=http_timeout,
        http2=True  # 强烈推荐开启 HTTP/2 多路复用
    )
    yield
    await app.state.client.aclose()

app = FastAPI(lifespan=lifespan)

我们需要写一个专属的异步生成器，一边从上游泵数据，一边轮询客户端的 TCP 连接状态。只要前端断开，立即终止上游流：

from fastapi import Request, HTTPException, status
from fastapi.responses import StreamingResponse
import asyncio
import logging

logger = logging.getLogger("gateway")

async def forward_llm_stream(upstream_response: httpx.Response, request: Request):
    try:
        # aiter_raw() 避开内部文本编码转换，性能最佳
        async for chunk in upstream_response.aiter_raw():
            # 关键：检查客户端是否已经断开（取消生成或关闭页面）if await request.is_disconnected():
                logger.warning("客户端已主动断开，立即切断上游 LLM 推理！")
                break
            
            if chunk:
                yield chunk
                # 极小让步让出事件循环，避免大块数据处理时饿死其它协程
                await asyncio.sleep(0)

    except (httpx.ReadTimeout, httpx.RemoteProtocolError) as err:
        logger.error(f"上游模型服务异常中途断流: {err}")
    finally:
        # 确保哪怕抛出 CancelledError 也把上游响应关闭
        await upstream_response.aclose()

反向代理流式传输，千万不要漏掉 X-Accel-Buffering: no，否则一旦网关前面套了 Nginx，Nginx 默认的 buffer 会直接毁掉打字机效果。

@app.post("/v1/chat/completions")
async def chat_proxy_endpoint(request: Request):
    client: httpx.AsyncClient = request.app.state.client
    
    # 获取原始请求头并做清洗
    headers = dict(request.headers)
    headers.pop("host", None)
    headers.pop("content-length", None)
    
    body = await request.body()
    
    try:
        # 建立到大模型推理引擎的上游请求
        upstream_req = client.build_request(
            method="POST",
            url="http://10.0.0.120:8000/v1/chat/completions",
            headers=headers,
            content=body
        )
        upstream_res = await client.send(upstream_req, stream=True)
        
        # 上游非 200 直接透传错误
        if upstream_res.status_code != 200:
            error_content = await upstream_res.aread()
            return StreamingResponse(content=iter([error_content]),
                status_code=upstream_res.status_code,
                media_type="application/json"
            )

        # 核心响应：禁用一切中间代理缓冲
        return StreamingResponse(forward_llm_stream(upstream_res, request),
            media_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
                "X-Accel-Buffering": "no",  # 关键：阻止 Nginx 缓冲
            }
        )
        
    except httpx.ConnectTimeout:
        raise HTTPException(status_code=504, detail="LLM 上游建立连接超时")
    except Exception as e:
        logger.exception("代理转发未知异常")
        raise HTTPException(status_code=500, detail=str(e))

我在局域网万兆环境下，用模拟并发客户端对上述代码进行了针对性压测，总结了三个极易掉坑的细节：

不要用默认的 asyncio 原生事件循环。在处理超高并发的网络 I/O 代理时，uvloop 和 httptools 能够带来 30%~40% 的吞吐提升。生产环境启动命令一定要锁死组件：

# 安装核心加速库
pip install uvloop httptools

# 生产环境启动方式（4 核机型示例）gunicorn main:app \
  --workers 4 \
  --worker-class uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8080 \
  --timeout 120 \
  --keep-alive 65

很多工程习惯给 FastAPI 加全局 GZipMiddleware(minimum_size=1000)。在流式传输场景下，中间件会尝试收集足够的 chunk 凑齐 1000 字节后再压缩，导致原本顺畅的单个 token 流变成“一坨一坨”地往外吐，直接毁掉打字机动画。

解决方案：在中间件层面显式排查路径，或者对 media_type == "text/event-stream" 绕过压缩逻辑。

以 500 个并发长连接请求 vLLM（单次返回 800 tokens）为例：

方案	首字延迟增加 (TTFT)	客户端中断后端算力释放率	网关平均内存占用
基础 AsyncClient 上下文方案	+182ms (频繁握手)	0%（算力持续空转）	~1.4 GB（连接缓冲泄漏）
本方案（池化 + 主动断流）	+3ms (极小开销)	100%（20ms 内终止）	~180 MB（恒定稳定）

在 AI 原生时代，网关不再是简单的反向代理，它承担着模型调用成本控制的第一道防线。客户端取消请求后立刻掐死上游流，单这一项调优就能在真实业务场景中节省 15% 以上的无谓算力开销。核心原则只有八个字：连接池化、无缓存流、主动感知。把这些底层 I/O 细节做好，哪怕用纯 Python 也能扛起相当客观的企业级并发流量。

===EOF===

正文完