大模型网关：多模型路由、限流、缓存与成本归因怎么搭

本文是 AI 应用系统学习系列的 L3 实战篇。前置：成本与延迟优化：让 AI 应用跑得起。 学完可以配合面试题食用：大模型 API 工程化：兼容层设计、降级与重试策略、成本与延迟优化：让 AI 应用跑得起。

为什么需要网关：key 散在各服务里的代价

公司只有一部总机时，谁打了外线、打了多久，一张话单全清楚。每人私拉一条外线之后，月底没人说得清话费去哪了。多数团队的 LLM 调用正处在第二种状态：审核服务拿着 DeepSeek 的 key，工单系统拿着 GLM 的 key，客服机器人手里还有一把 Claude 的 key。

由此带来四个具体麻烦：换一次模型要改 N 个服务、发 N 次版；月底账单是一笔糊涂账，没人答得上「AI 功能每月烧多少钱」；某家供应商故障，值班同学要挨个改配置重启；各服务自己写重试，故障时重试流量互相放大。

网关做的事就是把外线收回总机：业务方只认一个地址、一把业务 key，模型选择、fallback、限流、记账、密钥管理收拢到一层。

路由与 fallback：网关的核心调度

第一件事是统一接口。对业务方暴露 OpenAI 兼容的 /v1/chat/completions，供应商之间的差异（Anthropic 的 messages 结构、各家错误码不同）在网关内部转换，换模型只改路由表。

第二件事是按任务分级路由。路由表就是「任务类型 → 模型链」的映射：分类、摘要走便宜的小模型，代码生成走代码模型，多步 Agent 推理才上旗舰。37 篇算过账，简单任务用旗舰模型，价格差 10 倍是常事——路由表就是把这笔账自动化。

第三件事是 fallback。主模型超时或返回 5xx，自动切链上的下一个；但要分清什么情况该切：超时、5xx、限流可以切，400 参数错误切了也是错，直接把错误还给调用方。重试必须有次数上限和指数退避（0.5s、1s、2s）。把 fallback 写成「失败了就再试」的无限循环，供应商一故障，重试流量每分钟能滚到几万次，对方限流名单第一个就是你。

sequenceDiagram
    participant C as 业务服务
    participant G as 网关
    participant M1 as 主模型
    participant M2 as 备用模型
    C->>G: POST /v1/chat/completions
    G->>G: 查语义缓存 / 校验配额
    G->>M1: 转发（30s 超时）
    M1--xG: 超时
    G->>M2: 转发同一请求
    M2-->>G: 200 + usage
    G->>G: 落账：团队/模型/token/耗时
    G-->>C: 返回响应

限流、成本归因、密钥审计：网关的治理面

限流为什么必须在网关做？指望各服务自觉，等于指望新同事复制旧代码时记得补上配额逻辑——漏配是必然。配额按三个维度记：按团队（谁的预算）、按用户（防单个用户刷爆）、按模型（旗舰模型配额单独收紧）。计数放 Redis INCRBY，不放进程内存：网关一部署两个实例，内存计数就分叉了。

成本归因是每次调用落一条账：时间、团队、任务类型、模型、prompt_tokens、completion_tokens、耗时、是否缓存命中、是否发生 fallback。起步用 JSONL 追加写就够，月底一条 groupby 拆出每个团队、每个功能的成本。没有这张表，37 篇讲的成本优化无从下手——不知道钱花在哪，就不知道该优化哪。

密钥与审计：供应商 key 只存网关（环境变量或密钥管理服务），业务方只拿网关签发的业务 key，泄露了可以单独吊销。全量请求记 trace 日志，落库前做 PII 脱敏，手机号、身份证号正则替换后再存。用户的 prompt 里什么都有，这不是合规洁癖，是基本卫生。

缓存三层：省钱主力与它的边界

按实现难度递增讲三层：

• 精确缓存：prompt 完全一致才命中，键就是 prompt 的 hash。客服场景「怎么开发票」这类高频问题，命中率能到三四成，这部分请求的上游调用直接归零。
• 前缀缓存：供应商侧的能力（Anthropic 的 prompt caching、DeepSeek 的上下文硬盘缓存），公共前缀一致时输入按折扣计费。网关能帮上忙的是把公共 system prompt 规整成一致版本——每个业务方自己写一版 system prompt，前缀缓存等于白送不要。
• 语义缓存：把 query 向量化，和历史问题的相似度超过阈值（0.95 左右）就直接返回缓存答案，完全不调模型。

语义缓存有一条必须划清的边界：相似不等于可复用。「王女士的订单号是多少」和「李女士的订单号是多少」，余弦相似度轻松过 0.95，答案却互斥。规则很简单：含个人信息、含时效数据的请求跳过语义缓存。

动手实操：FastAPI 最小网关

约 100 行，路由表、配额、语义缓存、fallback 各一个骨架：

# gateway.py — FastAPI 最小 LLM 网关：路由表 + 配额 + 语义缓存 + fallback
# 演示用：配额与缓存放进程内存，生产换 Redis；上游 key 按供应商分别配
import os, time, json, math
import httpx
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

app = FastAPI()

# ---------- 路由表：任务类型 -> 模型链，列表顺序即 fallback 顺序 ----------
ROUTES = {
    "summary": ["qwen3-8b", "deepseek-chat"],           # 便宜模型优先
    "chat":    ["deepseek-chat", "glm-4-air"],
    "code":    ["claude-sonnet-4-5", "deepseek-chat"],
}
PROVIDERS = {                                           # 模型 -> base_url
    "qwen3-8b":          "http://localhost:8000/v1",    # vLLM 自托管
    "deepseek-chat":     "https://api.deepseek.com/v1",
    "glm-4-air":         "https://open.bigmodel.cn/api/paas/v4",
    "claude-sonnet-4-5": "https://api.anthropic.com/v1",
}
BIZ_KEYS = {"biz-team-a-001": "team-a"}                 # 业务 key -> 团队名

# ---------- 配额：每团队每模型 token 上限（生产用 Redis INCRBY）----------
QUOTA = {"team-a": {"*": 5_000_000, "claude-sonnet-4-5": 1_000_000}}
used  = {"team-a": {}}

def check_quota(team: str, model: str, est_tokens: int):
    cap = QUOTA.get(team, {}).get(model) or QUOTA.get(team, {}).get("*", 0)
    if used[team].get(model, 0) + est_tokens > cap:
        raise HTTPException(429, f"quota exceeded for {team}/{model}")

def record_usage(team, model, pt, ct):
    used[team][model] = used[team].get(model, 0) + pt + ct
    with open("usage.jsonl", "a") as f:                 # 成本归因：一调用一账
        f.write(json.dumps({"ts": time.time(), "team": team, "model": model,
                            "prompt_tokens": pt, "completion_tokens": ct}) + "\n")

# ---------- 语义缓存：embedding 相似度过阈值且不含敏感词才命中 ----------
EMBED_URL = "http://localhost:8000/v1/embeddings"       # vLLM 跑 bge-m3 之类
sem_cache = []                                          # [{q_emb, answer}]

def embed(text: str) -> list[float]:
    r = httpx.post(EMBED_URL, json={"model": "bge-m3", "input": text}, timeout=5)
    return r.json()["data"][0]["embedding"]

def cosine(a, b) -> float:
    dot = sum(x * y for x, y in zip(a, b))
    return dot / (math.sqrt(sum(x*x for x in a)) * math.sqrt(sum(x*x for x in b)))

def cache_lookup(messages) -> str | None:
    q = messages[-1]["content"]
    if any(w in q for w in ("订单", "余额", "密码")):   # 个人/时效数据不进缓存
        return None
    try:
        e = embed(q)
        return next((it["answer"] for it in sem_cache if cosine(e, it["q_emb"]) >= 0.95), None)
    except Exception:                                   # embedding 服务挂了就跳过缓存
        return None

def cache_store(messages, answer):
    try:
        sem_cache.append({"q_emb": embed(messages[-1]["content"]), "answer": answer})
    except Exception:
        pass

# ---------- 主入口 ----------
class ChatReq(BaseModel):
    task: str = "chat"                                  # 业务方声明任务类型，路由表按它分流
    messages: list[dict]
    temperature: float = 0.7

@app.post("/v1/chat/completions")
def chat(req: ChatReq, x_biz_key: str = Header()):
    team = BIZ_KEYS.get(x_biz_key)
    if not team:
        raise HTTPException(401, "invalid biz key")

    hit = cache_lookup(req.messages)                    # 1) 语义缓存
    if hit:
        record_usage(team, "semantic-cache", 0, 0)
        return {"choices": [{"message": {"role": "assistant", "content": hit}}],
                "cached": True}

    chain = ROUTES.get(req.task)                        # 2) 路由 + fallback
    if not chain:
        raise HTTPException(400, f"unknown task: {req.task}")
    pt_est = sum(len(str(m.get("content", ""))) for m in req.messages) // 2
    last_err = None
    for model in chain:
        try:
            check_quota(team, model, pt_est)
            r = httpx.post(
                f"{PROVIDERS[model]}/chat/completions",
                headers={"Authorization": f"Bearer {os.environ['UPSTREAM_KEY']}"},
                json={"model": model, "messages": req.messages,
                      "temperature": req.temperature},
                timeout=30,
            )
            r.raise_for_status()                        # 4xx 抛 HTTPStatusError
            data = r.json()
            u = data["usage"]
            record_usage(team, model, u["prompt_tokens"], u["completion_tokens"])
            cache_store(req.messages, data["choices"][0]["message"]["content"])
            return data
        except (httpx.TimeoutException, httpx.HTTPStatusError) as e:
            # 4xx（参数错误）切供应商也没用，直接还给调用方；只对超时/5xx 降级
            if isinstance(e, httpx.HTTPStatusError) and e.response.status_code < 500:
                raise HTTPException(e.response.status_code, e.response.text)
            last_err = e
    raise HTTPException(502, f"all providers failed: {last_err}")

跑起来后建议试三件事：

# 1) 正常调用
curl -s localhost:9000/v1/chat/completions -H 'X-Biz-Key: biz-team-a-001' \
  -H 'Content-Type: application/json' \
  -d '{"task":"summary","messages":[{"role":"user","content":"总结这段会议纪要，列出三个待办"}]}'

# 2) 原样再发一次：响应里多出 "cached": true，上游调用为零，usage.jsonl 里记一条 semantic-cache

# 3) 把 ROUTES["chat"] 主模型的 base_url 改成错误端口重启，观察：30s 超时后切到备用模型，账照落

现成方案：什么规模该自研

• LiteLLM Proxy：100 多家供应商统一成 OpenAI 格式，自带路由、fallback、预算和虚拟 key。供应商两三家、团队十来个人，先用它，一天能跑通。
• One API：Go 写的，Web 界面管理渠道和令牌，个人和小团队友好。
• Portkey：商业网关，可观测性和 guardrails 是卖点，不想自己维护时考虑。

自研的分界线：当你开始为了满足需求改这些方案的源码（比如语义缓存要接公司内部权限体系），才值得自研。上面那 100 行就是自研的种子——先跑通路由和记账，缓存、配额逐步往里加。

常见误区与小结

• 网关里写业务逻辑。网关只管转发、路由、记账，一旦掺业务分支，它就变成所有服务的前置依赖加故障点。
• 语义缓存对所有请求生效。个人数据和时效数据命中缓存就是事故，必须按规则跳过。
• 重试无上限、无退避。供应商故障时重试流量自我放大，等于自己攻击自己。
• 配额计数放进程内存。多实例部署计数分叉、重启清零，上 Redis。
• 供应商 key 下发给业务服务「方便调试」。key 一散开就收不回来，泄露范围不可控。

小结：37 篇讲的是单次调用怎么省钱，本篇把所有调用收拢到一个执行层——路由、缓存、配额、记账只有在这一层才做得起来。下一篇讲结构化输出：为什么 prompt 里写「严格输出 JSON」根本不够。

参考

• LiteLLM Proxy 官方文档（路由、fallback、虚拟 key）
• songquanpeng/one-api 的 GitHub README（渠道与令牌模型）
• Anthropic、DeepSeek 官方文档中 prompt caching 一节