Python高并发LLM流式代理实战：基于FastAPI与httpx解决SSE背压与内存泄露

2次阅读

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

上个月在公司内网上线了一套聚合网关，用于统一管理前端发往下游私有化 vLLM 集群和 DeepSeek 服务的 SSE（Server-Sent Events）流式请求。上线第二天早高峰，监控告警直接拉响：代理服务的内存从开机时的 150MB 一路飙升至 6GB 触发 OOM Killer，下游 GPU 显存也因为大量“已取消”的任务继续空跑推理而彻底堵死。

许多开发者在用 Python 写流式转发时，直觉上都觉得不过是调一下下游 API 再用 StreamingResponse 吐出去而已。然而，在真实的高并发生产环境下，大模型的长连接流式传输隐藏着几个非常恶劣的隐患：客户端提前断开导致的协程游离、数据消费过慢引发的隐式内存积压（缺乏背压），以及 HTTPX 客户端连接池的复用陷阱。今天聊聊我踩平这些坑后的最终落地方案。

很多工程文档里提供的流式转发代码，通常长成下面这个极简模样：

# 典型的“反面教材”式写法
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import httpx

app = FastAPI()
client = httpx.AsyncClient()

@app.post("/v1/chat/completions")
async def chat_proxy(payload: dict):
    req = client.build_request("POST", "http://vllm-backend:8000/v1/chat/completions", json=payload)
    resp = await client.send(req, stream=True)
    
    async def event_generator():
        async for chunk in resp.aiter_bytes():
            yield chunk

    return StreamingResponse(event_generator(), media_type="text/event-stream")

这段代码跑单测时顺滑无比，但在真实生产业务中会直接暴露出三个致命缺陷：

下游调用失控（Zombie Request）： 当终端用户点击“停止生成”或直接关掉网页标签页时，FastAPI 默认不会主动打断内部的 event_generator。下游的 vLLM 或外部商用 API 依然在拼命生成 Token，不仅吞噬显存算力，更让 token 费用白白打水漂。
内存无上限膨胀（Missing Backpressure）： 遇到弱网客户端（例如手机在电梯里）接收速率骤降到 1KB/s 时，Python 异步框架如果拉取下游速度过快，未被消费的 chunks 就会全部被丢进异步队列的缓冲区，长长的高速数据流瞬间撑爆物理内存。
全局 AsyncClient 导致的连接池穿透： 默认配置下的 httpx.AsyncClient 在应对数千并发长连接时，Keep-Alive 与连接池上限配置若不合理，很容易引发 PoolTimeout 或本地 socket 耗尽。

解决上述痛点的核心思路是：将客户端网络状态感知、下游 HTTP 上下文管理、以及基于块大小的背压调度强绑定在生命周期内。

以下是我重构并在生产环境稳定承载日均千万级 SSE 请求的核心网关模块：

import asyncio
import logging
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
import httpx

logger = logging.getLogger("gateway")

# 1. 深度定制的下游连接池
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 针对长连接调优的连接池参数
    limits = httpx.Limits(
        max_connections=2000,          # 最大连接总数
        max_keepalive_connections=500,  # 保持空闲的连接数
        keepalive_expiry=30.0           # 保持存活时间
    )
    # 针对大模型特化超时：连接必须极快，但读超时必须放宽（防止 TTFT 过长）timeouts = httpx.Timeout(
        connect=5.0,
        read=60.0,
        write=10.0,
        pool=5.0
    )
    app.state.client = httpx.AsyncClient(limits=limits, timeout=timeouts)
    yield
    await app.state.client.aclose()

app = FastAPI(lifespan=lifespan)

async def stream_generator(
    client: httpx.AsyncClient,
    payload: dict,
    request: Request
) -> AsyncGenerator[bytes, None]:
    """带有双向生命周期感知的流式转发器"""
    upstream_url = "http://upstream-engine:8000/v1/chat/completions"
    
    try:
        # 使用流式上下文管理器，保证异常或退出时连接必须释放
        async with client.stream("POST", upstream_url, json=payload) as upstream_resp:
            if upstream_resp.status_code != 200:
                error_body = await upstream_resp.aread()
                logger.error(f"下游异常: {upstream_resp.status_code} - {error_body.decode()}")
                yield f"data: {{\"error\": \" 下游服务响应失败 \"}}\n\n".encode()
                return

            # 控制每一批次流出，不拉取无限大，规避内存积压
            async for chunk in upstream_resp.aiter_raw():
                # 关键守护逻辑：一旦检测到下游客户端断链，立刻 raise 中断协程
                if await request.is_disconnected():
                    logger.warning("客户端已中断连接，立即熔断下游请求")
                    break
                
                # 显式让出事件循环控制权，避免密集 CPU 计算挂起
                yield chunk
                await asyncio.sleep(0)

    except asyncio.CancelledError:
        logger.info("捕获 CancelledError：当前流式任务已被客户端终止")
        raise
    except httpx.ReadTimeout:
        logger.error("下游大模型吐出 Token 超时")
        yield b"data: {\"error\": \"Upstream inference timeout\"}\n\n"
    except Exception as e:
        logger.error(f"网关层未知故障: {str(e)}")
        yield b"data: {\"error\": \"Gateway internal error\"}\n\n"

@app.post("/v1/chat/completions")
async def secure_chat_proxy(payload: dict, request: Request):
    if not payload.get("stream", False):
        raise HTTPException(status_code=400, detail="本网关专用通道仅支持 stream=True 模式")

    return StreamingResponse(stream_generator(app.state.client, payload, request),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no", # 严禁上游 Nginx 对 SSE 进行块缓存
        }
    )

在生产排障时，我发现代码逻辑全通，但前端就是收不到流式打字效果，而是卡顿 5 秒后瞬间喷出一大坨数据。这往往不是 Python 的问题，而是网关前层的 Nginx 在作祟。必须在响应头显式返回 X-Accel-Buffering: no，同时确保 Nginx 反向代理配置里禁用了代理缓冲：

location /v1/chat/completions {
    proxy_pass http://127.0.0.1:8080;
    proxy_http_version 1.1;
    proxy_set_header Connection "";
    
    # 彻底关闭 proxy buffer，保障打字机效果毫秒级送达
    proxy_buffering off;
    proxy_cache off;
    chunked_transfer_encoding on;
}

在 FastAPI 中，request.is_disconnected() 是一个异步方法，它通过检查底层 ASGI Receive 队列中的 http.disconnect 消息来确认对端状态。必须注意两点：

不要每一个 chunk 都频繁 await 嗅探几百次。如果 Token 吐出极快，每 5~10 个 chunk 检测一次可以减少事件循环调度开销。
配合 async with client.stream() 上下文退出机制，只要跳出循环，底层 HTTPX 请求就会发送 FIN 报文强制下游断流，有效阻断 GPU 显存浪费。

长连接架构下，传统的 Gunicorn 同步 / 短连接配置全部失效。启动流式网关时，工作进程推荐绑定单核以杜绝 GIL 竞争导致的事件循环延迟：

uvicorn main:app \
  --host 0.0.0.0 \
  --port 8080 \
  --workers 4 \
  --loop uvloop \
  --http httptools \
  --limit-concurrency 4000 \
  --timeout-keep-alive 30

其中，--loop uvloop 和 --http httptools 无论如何都要配上，实测在吞吐极限下，原生 asyncio loop 的延迟抖动会比 uvloop 高出近 40%。

为了验证这套网关的鲁棒性，我在 4C8G 的云服务器上，使用 Locust 模拟了 1000 个并发长连接，其中 30% 的请求在接收到第 15 个 Token 时主动中断（模拟用户点“停止”或刷新页面）。

方案实现	1000 并发平均内存占用	断开时协程回收时延	下游无效 GPU 算力浪费率
初始方案 (直觉透传)	3.8 GB ~ 5.2 GB (波动剧烈)	下游直至生成完毕才释放	~ 28.5%
优化方案 (感知断开 + 背压)	410 MB ~ 620 MB (平稳)	< 15 ms (即刻熔断)	< 0.5%

可以看到，优化的重点不仅仅在于网关自身的内存压减（从几 G 骤降至几百兆），更关键的是成功把无效的推理算力在 15 毫秒内彻底截断，大幅减轻了后端推理引擎的压力。

做 LLM 流式系统和做传统的 Web RESTful 架构有着本质区别。RESTful 关注的是 QPS（每秒请求数），而流式网关关注的是 并发连接数（Concurrence）与长连接生命周期的精确受控。在写 Python 异步代码时，永远不要假定客户端会体面地等待响应完成，处理好 CancelledError、掐死下游的僵尸请求、为慢速客户端做好背压保护，才是一套工业级 AI 网关能稳定扛住流量冲击的底气所在。

===EOF===

正文完