手搓低延迟本地AI网关：基于FastAPI与LiteLLM的高并发流式缓存实战

8次阅读

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

上个月我把工作站的本地大模型服务暴露给团队写代码和跑 Agent，结果不到半天网关就频频超时。后端两张 RTX 4090 跑 Qwen2.5-32B 与 DeepSeek-R1-Distill，只要有三个同事同时在 Cursor 里狂按 Tab 补全，伴随几个 Agent 并发发起长上下文检索，显卡显存直接打满，排队延迟（Time To First Token, TTFT）飙到 12 秒开外，编辑器直接报错中断。

市面上现成的商业 API 网关往往太重，且对 LLM 特有的 SSE（Server-Sent Events）流式响应、Token 计费与语义缓存支持得很别扭。为了压榨这套本地算力，我花了一个周末基于 FastAPI、LiteLLM 与 Redis 手搓了一个轻量级流式 AI 反向代理。实测上线后，高频重复 Prompt 首字响应压缩到 20ms 以内，突发并发下的请求成功率从 68% 拉升至 99.5%。今天把整套落地架构、踩坑代码和压测参数全盘盘出。

传统 HTTP 反向代理（如 Nginx 默认配置）处理无状态短连接游刃有余，但遇到本地 LLM 推理服务时会踩三颗大雷：

缓冲机制破坏流式体验：Nginx 默认开启proxy_buffering，模型生成的 Token 会被 Nginx 攒成大块才发给前端，用户端打字机效果直接变成“便秘式卡顿”。
完全相同的上下文反复推理：Agent 运行时的大量 Prompt（如 System Prompt、工具定义、固定 Few-shot）存在极高重合度。如果网关层不做精确与语义缓存，宝贵的 GPU 算力全在重复计算 Prefill。
缺乏自适应并发保护（Backpressure）：vLLM/Ollama 在 KV Cache 耗尽时，要么 OOM 挂掉，要么将新请求推入等待队列导致 HTTP Client 主动超时断开。网关层必须具备平滑排队与自适应熔断能力。

为了兼顾开发速度与性能，我放弃了臃肿的 Kong 与 APISIX，采用如下轻量化极客组合：

接入与调度层（FastAPI + Asyncio）：纯异步处理长连接，负责鉴权、流式代理转发、分块哈希匹配。
路由与统一接口层（LiteLLM）：抹平 vLLM、Ollama 与公有云备用 API 之间的输入输出差异，原生支持多模型故障转移（Failover）。
缓存与限流状态层（Redis）：存储精确匹配的流式响应 Chunks 与基于令牌桶的客户端频控。

在流式传输（Streaming）下做缓存是公认的难点。模型是一个 Token 一个 Token 吐出来的，如果等全部流式结束再存缓存，客户端首次调用体验良好，但第二次请求就失去了流式的真实打字节奏。更优的做法是在网关层拦截 chunk 流：边向客户端 Yield 输出，边在内存缓存块，最后原子化写入 Redis，并记录首字耗时。

下面是我提炼后的网关核心路由模块，拿掉业务鉴权后可直接跑起来：

import json
import hashlib
from typing import AsyncGenerator
import redis.asyncio as aioredis
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import httpx

app = FastAPI(title="Geek-LLM-Gateway")
redis_client = aioredis.from_url("redis://localhost:6379/0", decode_responses=False)
UPSTREAM_VLLM_URL = "http://192.168.1.120:8000/v1/chat/completions"

def generate_cache_key(model: str, messages: list, temperature: float) -> str:
    payload = json.dumps({"m": model, "msg": messages, "t": temperature}, sort_keys=True)
    return f"llm_cache:{hashlib.sha256(payload.encode()).hexdigest()}"

async def stream_from_cache(cached_chunks: list[bytes]) -> AsyncGenerator[bytes, None]:
    for chunk in cached_chunks:
        yield chunk

async def stream_and_cache(cache_key: str, req_body: dict) -> AsyncGenerator[bytes, None]:
    client = httpx.AsyncClient(timeout=120.0)
    cached_buffer = []
    
    try:
        async with client.stream("POST", UPSTREAM_VLLM_URL, json=req_body) as response:
            async for line in response.aiter_lines():
                if not line:
                    continue
                formatted_chunk = f"{line}\n\n".encode("utf-8")
                cached_buffer.append(formatted_chunk)
                yield formatted_chunk
                
        # 请求顺利结束，异步写入 Redis，TTL 设置为 2 小时
        pipe = redis_client.pipeline()
        pipe.delete(cache_key)
        pipe.rpush(cache_key, *cached_buffer)
        pipe.expire(cache_key, 7200)
        await pipe.execute()
    finally:
        await client.aclose()

@app.post("/v1/chat/completions")
async def chat_proxy(request: Request):
    body = await request.json()
    is_stream = body.get("stream", False)
    temperature = body.get("temperature", 0.7)
    
    # 仅针对确定性较高或固定 Prompt 触发精确缓存
    can_cache = temperature <= 0.3 and is_stream
    cache_key = generate_cache_key(body.get("model", ""), body.get("messages", []), temperature)
    
    if can_cache and await redis_client.exists(cache_key):
        chunks = await redis_client.lrange(cache_key, 0, -1)
        return StreamingResponse(stream_from_cache(chunks),
            media_type="text/event-stream",
            headers={"X-Cache-Lookup": "HIT"}
        )

    if is_stream:
        return StreamingResponse(stream_and_cache(cache_key, body),
            media_type="text/event-stream",
            headers={"X-Cache-Lookup": "MISS"}
        )
    
    # 非流式情况回退至默认转发...

我在网关前面挂了一层 Nginx 用来做局域网 SSL 卸载，起初发现 Cursor 经常莫名其妙报 Unexpected end of JSON。抓包发现，Nginx 将 SSE 的data: [DONE] 分片和上一个分片合并了，且经常延迟 8KB 才刷一次屏。在 Nginx 的虚拟主机配置中必须显式加上这两行：

proxy_http_version 1.1;
proxy_set_header Connection "";
proxy_buffering off;
proxy_cache off;
chunked_transfer_encoding on;

压测期间并发一上 100，网关直接抛出 Too many open files 和PoolTimeout。解决方式有两个：第一，把操作系统的 nofile 拉高；第二，严禁在每个请求中实例化httpx.AsyncClient()。必须在 FastAPI 生命周期（Lifespan）中维护全局唯一的长连接客户端池，并调高限制：

# 推荐的全局连接池设置
limits = httpx.Limits(max_keepalive_connections=200, max_connections=500)
timeout = httpx.Timeout(connect=5.0, read=120.0, write=5.0, pool=5.0)
global_client = httpx.AsyncClient(limits=limits, timeout=timeout)

测试硬件环境：双卡 Nvidia RTX 4090 24G，后端运行 vLLM（启用 PagedAttention，Tensor Parallel=2），加载 Qwen2.5-32B-Instruct-GPTQ-Int4。测试场景为 50 个虚拟用户并发发送 400 Tokens 上下文请求，包含 30% 的高频重复 Prompt。

指标项	直连 vLLM (无网关)	挂载自研流式网关 (冷启动)	自研网关 (30% 缓存命中)
平均首字延迟 (TTFT)	4,820 ms	4,910 ms (+1.8%)	1,720 ms (-64.3%)
P99 首字延迟	11,300 ms	11,540 ms	4,200 ms (-62.8%)
系统并发容量 (QPS)	11.2 req/s	10.9 req/s	26.4 req/s (+135%)
请求超时率 (Timeout > 15s)	18.4%	4.2% (受控排队)	0.0%

实测数据表明，自研网关在冷启动下的微小损耗（约 90ms，主要是异步中间件调度与 Header 解析开销）几乎可以忽略不计；但一旦引入精确流式缓存与自适应连接池，在对抗开发团队的高频补全和相同上下文检索时，整体并发上限直接翻倍，且杜绝了超时抛错的惨剧。

折腾私有 AI 算力，不要只盯着显卡算力参数和量化级别。在本地算力有限（通常在 1~4 张消费级显卡）的情况下，前端网关调度能榨出的性能往往超出预期：

不要指望通用网关搞定流式 LLM：它们对 SSE 连接特性的支持往往隐藏着各种 Buffer 黑洞。
Agent 场景必须上缓存：Agent 在反思、规划阶段输入的巨大 System Prompt 基本一致，善用 temperature=0 并缓存结果，是拯救显存带宽成本最高效的办法。
严控连接生命周期：异步 Python 虽然写起来快，但在高并发流式场景下，任何未被回收的 AsyncGenerator 和悬空连接都会在几分钟内耗尽宿主机的 Socket 资源。

下一步我打算给这个网关接入基于嵌入向量的语义近似缓存（Semantic Cache），以及模型过载时的本地降级（自动切换至云端 API）。如果你也在折腾自己的家庭实验室或团队本地算力，这套方案值得你立刻上手试一试。

===EOF===

正文完