FastAPI 打造高并发 LLM 流式代理：从客户端断连到显存止血实战

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

上个月帮朋友排查一套内网部署的 vLLM 问答平台，后台显卡整天处于 98% 以上满载状态，但实际并发用户数其实也就几十个人。排查日志后发现一个典型现象：大量用户在模型生成长文本时，只要发现前几个词不对，就会频繁点击“停止生成”或直接关掉网页，然而后端的 Python 代理层却还在老老实实地拉取上游数据，直到整个 prompt 的 token 全部吐完。

这就导致算力和显存白白浪费在已经被丢弃的请求上。今天聊聊怎么用 FastAPI 和 httpx 搓一个能扛并发、且能感知客户端断连并立刻“止血”的大模型流式代理中间件。

一、为什么简单的 SSE 流式透传是“灾难级”的？

很多开发者初学 FastAPI 流式响应时，习惯直接抄官方示例写出类似下面的代码：

# 典型反面教材
@app.post("/v1/chat/completions")
async def chat_proxy(request: Request):
    client = httpx.AsyncClient()
    upstream_resp = await client.send(...)
    return StreamingResponse(upstream_resp.aiter_bytes())

这种写法在本地 Demo 里跑得顺风顺水，一上线就会遇到三个致命硬伤：

1. 上游算力白白空转 ：前端一旦关闭连接（用户点 Stop 或关标签页），StreamingResponse 默认并不会主动通知上游。如果用户问了个“写一篇一万字的小说”，哪怕用户 1 秒后就走人，上游推理引擎（vLLM、Ollama 或按 Token 收费的商业 API）仍会吭哧吭哧把这一万字推理完。
2. 临时创建客户端导致端口耗尽 ：每个请求临时 new 一个 httpx.AsyncClient()，短连接迅速耗尽宿主机的 Ephemeral Ports，压测一过 300 并发直接满屏 Cannot assign requested address。
3. 内存无底线缓冲（Backpressure 失控）：如果上游吐 token 速度极快（比如本地双卡 H800 的 vLLM 可以飙到上百 tokens/s），而客户端网络延迟高、消费慢，代理服务内部的缓冲区就会迅速膨胀，几十个并发下来 Python 进程的内存能直接飙到几个 G。

二、优雅实现：带断连感知与生命周期的代理

要解决算力浪费，核心在于让我们的异步生成器（Async Generator）不仅要往下游吐数据，还要同时监听客户端的 Socket 状态。一旦发现下游断开，立刻中断循环并通知上游取消请求。

这里给出我们优化后的核心中间层代码：

import httpx
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from contextlib import asynccontextmanager

# 全局共享 HTTP 连接池，避免频繁握手
upstream_client: httpx.AsyncClient = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global upstream_client
    # 限制连接池，防止压垮上游或被本地文件描述符限制搞死
    limits = httpx.Limits(
        max_keepalive_connections=100, 
        max_connections=500, 
        keepalive_expiry=30.0
    )
    # LLM 流式输出耗时长，连接超时必须按需设大，但 connect 必须严格受限
    timeout = httpx.Timeout(timeout=300.0, connect=5.0)
    upstream_client = httpx.AsyncClient(limits=limits, timeout=timeout)
    yield
    await upstream_client.aclose()

app = FastAPI(lifespan=lifespan)

async def sse_stream_generator(request: Request, body: dict):
    req = upstream_client.build_request(
        "POST", 
        "http://192.168.1.100:8000/v1/chat/completions",
        json=body,
        headers={"Authorization": "Bearer internal-token"}
    )
    
    response = await upstream_client.send(req, stream=True)
    
    try:
        # 严格按字节流读取，不做无意义的字符编解码
        async for chunk in response.aiter_raw():
            # 核心检查点：主动探测客户端断连
            if await request.is_disconnected():
                # 这一行打在日志里，你会发现止血频率惊人地高
                print("Client disconnected, aborting upstream task.")
                break
            
            if chunk:
                yield chunk
    finally:
        # 务必在 finally 块里关闭上游流，给 vLLM/ 商业网关发 RST 信号
        await response.aclose()

@app.post("/api/v1/chat")
async def chat_proxy(request: Request):
    payload = await request.json()
    return StreamingResponse(sse_stream_generator(request, payload),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"  # 禁用 Nginx 缓冲区，保证首字实时性
        }
    )

避坑要点剖析：

• request.is_disconnected()：Starlette/FastAPI 底层基于 ASGI 协议，它会不断轮询接收 channel 的 http.disconnect 消息。在 yield 之前检查一下，只要前端按了 Stop，下一次读取直接跳出循环。
• response.aclose()：如果不显式调用 aclose()，即便跳出循环，底层 HTTP 连接也可能直到 GC 回收时才关闭。显式关闭会立刻向后端断开连接，后端的 vLLM 捕捉到底层连接断开后，会触发其内部调度器销毁对应的 Sequence，直接省下后续的显存计算。
• aiter_raw() 替代 aiter_lines() 或 aiter_bytes()：流式透传时千万别自作聪明在代理层用 decode("utf-8") 再 encode，遇到中文字符跨 chunk 被截断时极易抛出解码异常，直接原样透传二进制原始流最稳。

三、必须注意的网络与中间件层暗坑

光把 Python 逻辑改对还不够，很多时候架构链路上的中间件会让你前功尽弃：

1. 反向代理层（Nginx）的缓冲灾难

如果你的代理前端挂着 Nginx，很多老哥会发现前端根本收不到流式，而是卡了十几秒后直接“哗”地一下把全部文本吐出来。这就是 Nginx 的 proxy buffering 搞的鬼。

除了在 FastAPI 代码中加入 X-Accel-Buffering: no 响应头，最稳妥的还是直接在 Nginx 对应 location 中压制缓冲：

location /api/v1/chat {
    proxy_pass http://127.0.0.1:8080;
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    proxy_buffering off;
    proxy_cache off;
    chunked_transfer_encoding on;
    proxy_read_timeout 600s;
}

2. Uvicorn Worker 与异步死锁

LLM 代理基本上属于纯 I/O 密集型应用，不需要开一堆多进程 Worker。我在 8 核机器上压测的经验配置是：开 4 个 Worker，启动时显式使用 httptools 和 uvloop。千万不要只开 1 个单线程 worker 还忘了给 Uvicorn 设置连接上限：

uvicorn main:app --host 0.0.0.0 --port 8080 \
  --workers 4 \
  --loop uvloop \
  --http httptools \
  --limit-concurrency 1000

四、实测压测效果对比

为了验证断连取消的收益，我在测试集群搭建了包含 1 个代理服务与 1 个单卡 RTX 4090 运行 Qwen-14B 的 vLLM 实例环境。使用 Locust 模拟 50 个高并发用户，其中 40% 的用户在收到前 50 个 token 时主动模拟断开（类似于正常用户不满意输出果断重试的行为）。

数据非常直观：当大量垃圾计算被掐死在摇篮里后，GPU 的 KV Cache 调度器不需要频繁抢占或将上下文换入换出，真正的有效请求处理速度提升了近 56%，同时代理层的内存因为消除了无脑 buffer 与野蛮创建 client，直接稳在了 100MB 级别。

总结与避坑心得

写 Python 代理网关，很多时候瓶颈不是 Python 这门语言不够快，而是我们在异步流程中遗漏了太多协议细节：

• 生命周期优先 ：永远把 httpx.AsyncClient 放到全局或 lifespan 集中管理，单例连接池是 I/O 吞吐的基石。
• 及时刹车 ：LLM 推理是极度昂贵的计算，客户端断联探测不仅仅是网络健壮性问题，更是实打实的算力成本控制。
• 链路透明 ：从 Nginx、网关代理到上游，确保每一层都关闭缓冲，让每一滴 token 都能低延迟、零积压地推送到前端。