大模型API网关实战：多模型智能路由、限流降级与成本管控完整方案
引言：为什么你需要一个 LLM API 网关
当团队从「试用一个大模型」走向「在生产环境中长期使用大模型」时，几乎都会撞上同一堵墙。
一开始，你可能在业务代码里直接 import openai，把 API Key 硬编码在配置里，所有请求都打到 GPT-4o。看起来很美好，直到出现下面这些情况：
- 某天 OpenAI 出现区域性故障，整个业务跟着宕机，没有任何降级手段；
- 月底财务找上门，说这个月大模型账单翻了三倍，但没人能说清钱花在哪；
- 产品想换一批请求到更便宜的 Claude Haiku，结果发现业务代码里上百处调用都要改；
- 高峰期 Token 消耗把上游限流打满，关键业务请求被拒，而非关键请求却在抢额度。
这些问题的本质是：你的业务系统和大模型供应商之间，缺少一个解耦层。 这个解耦层就是 LLM API 网关。它和传统的 API 网关一脉相承，但针对大模型的特性（按 Token 计费、延迟差异大、模型能力分层、上下文长）做了专门设计。
一个设计良好的 LLM API 网关，能让你在不改动业务代码的前提下，完成模型切换、成本治理、故障容灾和合规审计。本文将从架构到代码，带你实现一个生产级方案。
API 网关核心能力总览
在动手之前，先明确我们要建设的网关应该具备哪些核心能力。我把它归纳为五项：
-
统一接口：对业务方暴露一套 OpenAI 兼容的 REST 接口。无论后端实际调用的是 OpenAI、Anthropic 还是本地模型，业务方都只对接这一套接口，做到「换模型不换代码」。
-
多模型路由：根据请求的任务类型、成本预算、延迟要求，把请求智能分发到最合适的模型。例如简单分类任务走小模型，复杂推理走大模型，预算紧张时自动降级。
-
限流降级：通过令牌桶做请求级和 Token 级限流，防止突发流量击穿上游配额；当某个供应商持续报错时触发熔断，自动故障转移到备用通道。
-
成本管控：为每个业务线、每个用户设置 Token 预算和费用上限，实时追踪消耗，超预算自动熔断请求，避免「账单失控」。
-
审计日志：记录每一次请求的元信息（调用方、模型、Token 用量、延迟、成本、是否成功），用于事后追溯、成本归因和合规审计。
这五项能力相互配合：路由决定「找谁干」，限流降级保证「干得稳」，成本管控决定「能不能花」，审计日志回答「花了多少、花得值不值」。
架构设计
整体架构采用分层设计，自上而下分为接入层、网关核心层、供应商适配层。
┌─────────────────────────────┐
业务系统 ───────────▶ │ 接入层 (Nginx + FastAPI) │
│ TLS 终止 / 负载均衡 / 鉴权 │
└──────────────┬──────────────┘
│
┌──────────────▼──────────────┐
│ 网关核心层 │
│ ┌───────────────────────┐ │
│ │ 路由决策引擎 │ │
│ │ (任务/成本/延迟策略) │ │
│ └───────────┬───────────┘ │
│ ┌───────────▼───────────┐ │
│ │ 限流 + 熔断 + 重试 │ │
│ └───────────┬───────────┘ │
│ ┌───────────▼───────────┐ │
│ │ 成本管控 + 审计日志 │ │
│ └───────────┬───────────┘ │
└──────────────┬──────────────┘
│
┌────────────┬───────────┼───────────┬────────────┐
▼ ▼ ▼ ▼ ▼
┌────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐
│ OpenAI │ │Anthropic│ │ 通义千问 │ │ DeepSeek│ │ 本地模型│
└────────┘ └─────────┘ └─────────┘ └─────────┘ └─────────┘
数据流说明：请求进入接入层后，先完成 API Key 鉴权，随后进入核心层。路由引擎根据请求标签（如 task_type）和实时状态（各模型延迟、剩余预算）选出目标模型；限流模块判断是否放行；请求经供应商适配层转换为对应厂商的协议格式后发出；响应回流时，成本管控模块按厂商定价计算费用并扣减预算，审计模块写入日志。任何环节失败，都会触发重试或故障转移。
这层的解耦设计让新增一个供应商只需要写一个适配器，而不需要碰路由和限流逻辑。
实战：用 Python 实现一个生产级 LLM API 网关
下面我们用 Python + FastAPI 实现上述架构。代码完整可运行，依赖安装命令如下：
pip install fastapi uvicorn httpx pydantic
为突出核心逻辑，代码用内存存储替代数据库；生产环境可把预算、日志、计数器替换为 Redis 和 PostgreSQL。
1. 统一 OpenAI 兼容接口 + 供应商适配
我们对外暴露 OpenAI 的 /v1/chat/completions 接口，内部通过适配器对接不同厂商。
# llm_gateway.py
import asyncio
import time
import random
import logging
from collections import defaultdict, deque
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import httpx
from fastapi import FastAPI, Header, HTTPException, Request
from pydantic import BaseModel
logging.basicConfig(level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("llm-gateway")
# ---------- 数据模型 ----------
class Message(BaseModel):
role: str
content: str
class ChatRequest(BaseModel):
model: Optional[str] = None # 允许为空，由路由决定
messages: list[Message]
temperature: float = 0.7
max_tokens: int = 1024
# 业务扩展字段，用于路由决策
task_type: str = "chat" # chat / classify / reason / embed
budget_tier: str = "standard" # low / standard / premium
# ---------- 模型与定价配置 ----------
@dataclass
class ModelConfig:
name: str # 网关内统一别名
provider: str # openai / anthropic / deepseek
endpoint: str
api_key: str
price_in_per_1k: float # 输入价格 (元/1k tokens)
price_out_per_1k: float # 输出价格
max_latency_ms: int = 8000
capacity_weight: int = 1 # 并发权重
tags: list[str] = field(default_factory=list)
MODEL_REGISTRY: dict[str, ModelConfig] = {
"gpt-4o": ModelConfig("gpt-4o", "openai",
"https://api.openai.com/v1/chat/completions",
"sk-OPENAI-KEY", 0.0175, 0.07, tags=["reason", "premium"]),
"gpt-4o-mini": ModelConfig("gpt-4o-mini", "openai",
"https://api.openai.com/v1/chat/completions",
"sk-OPENAI-KEY", 0.00105, 0.0042, tags=["chat", "low"]),
"claude-haiku": ModelConfig("claude-haiku", "anthropic",
"https://api.anthropic.com/v1/messages",
"sk-ANTHROPIC-KEY", 0.0014, 0.007, tags=["chat", "low"]),
"deepseek-chat": ModelConfig("deepseek-chat", "deepseek",
"https://api.deepseek.com/chat/completions",
"sk-DEEPSEEK-KEY", 0.001, 0.0028, tags=["chat", "reason", "low"]),
}
2. 多模型智能路由
路由策略综合考虑三件事：任务类型匹配、成本预算、实时延迟。先按任务标签筛出候选模型，再按预算档位和滚动平均延迟打分排序。
class Router:
"""基于任务类型 / 成本 / 延迟的路由引擎"""
def __init__(self, registry: dict[str, ModelConfig]):
self.registry = registry
self.latency_samples: dict[str, deque] = defaultdict(
lambda: deque(maxlen=20))
def record_latency(self, model: str, ms: int):
self.latency_samples[model].append(ms)
def avg_latency(self, model: str) -> float:
s = self.latency_samples[model]
return sum(s) / len(s) if s else 3000
def select(self, req: ChatRequest) -> str:
candidates = []
for alias, cfg in self.registry.items():
# 任务类型必须匹配标签
if req.task_type not in cfg.tags:
continue
# 预算档位过滤
if req.budget_tier == "low" and "low" not in cfg.tags:
continue
if req.budget_tier == "premium" and "premium" not in cfg.tags:
continue
# 指定模型则直接返回
if req.model and req.model != alias:
continue
candidates.append((alias, cfg))
if not candidates:
# 兜底：退回最便宜的可用模型
fallback = min(self.registry.values(),
key=lambda c: c.price_in_per_1k)
return fallback.name
# 打分：成本权重 0.5 + 延迟权重 0.5（越低越好）
def score(cfg: ModelConfig) -> float:
cost_score = cfg.price_in_per_1k + cfg.price_out_per_1k
latency_score = self.avg_latency(cfg.name) / 1000.0
return 0.5 * cost_score + 0.5 * latency_score
best = min(candidates, key=lambda x: score(x[1]))
return best[0]
3. 限流、熔断与指数退避重试
令牌桶控制整体 QPS，熔断器在连续失败时隔离故障模型，重试采用指数退避并配合故障转移。
class TokenBucket:
"""令牌桶限流器"""
def __init__(self, rate: float, capacity: int):
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.last = time.monotonic()
self._lock = asyncio.Lock()
async def acquire(self) -> bool:
async with self._lock:
now = time.monotonic()
self.tokens = min(self.capacity,
self.tokens + (now - self.last) * self.rate)
self.last = now
if self.tokens >= 1:
self.tokens -= 1
return True
return False
class CircuitBreaker:
"""熔断器：CLOSED -> OPEN -> HALF_OPEN -> CLOSED"""
class State(Enum):
CLOSED = "closed"; OPEN = "open"; HALF_OPEN = "half_open"
def __init__(self, fail_threshold=5, recovery_secs=30):
self.fail_threshold = fail_threshold
self.recovery_secs = recovery_secs
self.state = self.State.CLOSED
self.fail_count = 0
self.opened_at = 0.0
def allow(self) -> bool:
if self.state == self.State.OPEN:
if time.monotonic() - self.opened_at > self.recovery_secs:
self.state = self.State.HALF_OPEN
return True
return False
return True
def record(self, success: bool):
if success:
self.fail_count = 0
self.state = self.State.CLOSED
else:
self.fail_count += 1
if self.fail_count >= self.fail_threshold:
self.state = self.State.OPEN
self.opened_at = time.monotonic()
4. 成本管控与审计日志
为每个调用方维护 Token 预算和实时费用；每次请求写入审计日志。
@dataclass
class BudgetAccount:
monthly_token_budget: int
used_tokens: int = 0
used_cost: float = 0.0
@property
def remaining(self) -> int:
return self.monthly_token_budget - self.used_tokens
class CostController:
def __init__(self):
self.accounts: dict[str, BudgetAccount] = defaultdict(
lambda: BudgetAccount(monthly_token_budget=2_000_000))
def check(self, api_key: str) -> bool:
return self.accounts[api_key].remaining > 0
def charge(self, api_key: str, in_tokens: int, out_tokens: int,
cfg: ModelConfig):
acc = self.accounts[api_key]
acc.used_tokens += in_tokens + out_tokens
acc.used_cost += (in_tokens * cfg.price_in_per_1k
+ out_tokens * cfg.price_out_per_1k) / 1000
def usage(self, api_key: str) -> dict:
a = self.accounts[api_key]
return {"used_tokens": a.used_tokens, "used_cost": round(a.used_cost, 4),
"remaining": a.remaining}
@dataclass
class AuditRecord:
ts: float
api_key: str
model: str
task_type: str
in_tokens: int
out_tokens: int
latency_ms: int
cost: float
status: str
error: str = ""
class AuditLog:
def __init__(self):
self.records: list[AuditRecord] = []
def write(self, rec: AuditRecord):
self.records.append(rec)
logger.info(
f"AUDIT key={rec.api_key[:8]}.. model={rec.model} "
f"in={rec.in_tokens} out={rec.out_tokens} "
f"latency={rec.latency_ms}ms cost={rec.cost:.5f} status={rec.status}")
5. LLMGateway 主类：串起所有组件
LLMGateway 把路由、限流、熔断、重试、成本、审计整合到一次调用里。供应商适配在这里统一转换为各厂商协议。
class LLMGateway:
def __init__(self, registry: dict[str, ModelConfig]):
self.registry = registry
self.router = Router(registry)
self.bucket = TokenBucket(rate=10, capacity=20) # 10 QPS, 突发 20
self.breakers: dict[str, CircuitBreaker] = defaultdict(CircuitBreaker)
self.cost = CostController()
self.audit = AuditLog()
self.client = httpx.AsyncClient(timeout=30)
async def _call_provider(self, cfg: ModelConfig, req: ChatRequest):
"""统一适配不同厂商协议，返回 (output_text, in_tokens, out_tokens)"""
if cfg.provider == "openai" or cfg.provider == "deepseek":
payload = {
"model": cfg.name,
"messages": [m.model_dump() for m in req.messages],
"temperature": req.temperature,
"max_tokens": req.max_tokens,
}
headers = {"Authorization": f"Bearer {cfg.api_key}"}
resp = await self.client.post(cfg.endpoint, json=payload,
headers=headers)
resp.raise_for_status()
data = resp.json()
text = data["choices"][0]["message"]["content"]
usage = data.get("usage", {})
return text, usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0)
if cfg.provider == "anthropic":
payload = {
"model": cfg.name,
"max_tokens": req.max_tokens,
"messages": [m.model_dump() for m in req.messages],
}
headers = {"x-api-key": cfg.api_key,
"anthropic-version": "2023-06-01",
"content-type": "application/json"}
resp = await self.client.post(cfg.endpoint, json=payload,
headers=headers)
resp.raise_for_status()
data = resp.json()
text = data["content"][0]["text"]
usage = data.get("usage", {})
return text, usage.get("input_tokens", 0), usage.get("output_tokens", 0)
raise ValueError(f"unsupported provider: {cfg.provider}")
async def chat(self, req: ChatRequest, api_key: str) -> dict:
# 1. 限流
if not await self.bucket.acquire():
raise HTTPException(429, "rate limit exceeded")
# 2. 预算检查
if not self.cost.check(api_key):
raise HTTPException(402, "token budget exhausted")
# 3. 路由出候选模型列表（主 + 备）
primary = self.router.select(req)
candidates = [primary] + [m for m in self.registry
if m != primary and req.task_type
in self.registry[m].tags]
# 4. 逐个尝试，带指数退避 + 故障转移
last_err = ""
for alias in candidates:
cfg = self.registry[alias]
cb = self.breakers[alias]
if not cb.allow():
last_err = f"circuit open for {alias}"
continue
for attempt in range(3):
start = time.monotonic()
try:
text, in_tok, out_tok = await self._call_provider(cfg, req)
latency = int((time.monotonic() - start) * 1000)
self.router.record_latency(alias, latency)
cb.record(True)
self.cost.charge(api_key, in_tok, out_tok, cfg)
self.audit.write(AuditRecord(
ts=time.time(), api_key=api_key, model=alias,
task_type=req.task_type, in_tokens=in_tok,
out_tokens=out_tok, latency_ms=latency,
cost=(in_tok * cfg.price_in_per_1k
+ out_tok * cfg.price_out_per_1k) / 1000,
status="ok"))
return {"model": alias, "content": text,
"usage": {"prompt_tokens": in_tok,
"completion_tokens": out_tok},
"latency_ms": latency}
except Exception as e:
cb.record(False)
last_err = str(e)
logger.warning(f"call {alias} failed (attempt {attempt+1}): {e}")
# 指数退避
await asyncio.sleep(0.5 * (2 ** attempt))
self.audit.write(AuditRecord(
ts=time.time(), api_key=api_key, model=primary,
task_type=req.task_type, in_tokens=0, out_tokens=0,
latency_ms=0, cost=0, status="error", error=last_err))
raise HTTPException(503, f"all models failed: {last_err}")
6. FastAPI 服务入口
app = FastAPI(title="LLM API Gateway")
gateway = LLMGateway(MODEL_REGISTRY)
@app.post("/v1/chat/completions")
async def chat_completions(req: ChatRequest,
authorization: str = Header(...)):
api_key = authorization.replace("Bearer ", "").strip()
return await gateway.chat(req, api_key)
@app.get("/v1/usage")
async def usage(authorization: str = Header(...)):
api_key = authorization.replace("Bearer ", "").strip()
return gateway.cost.usage(api_key)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
启动服务后即可用标准 OpenAI 客户端调用：
curl -X POST http://localhost:8000/v1/chat/completions \
-H "Authorization: Bearer biz-key-001" \
-H "Content-Type: application/json" \
-d '{"messages":[{"role":"user","content":"用一句话解释什么是向量数据库"}],
"task_type":"chat","budget_tier":"low","max_tokens":100}'
网关会自动把这条低成本闲聊请求路由到 gpt-4o-mini 或 deepseek-chat，并记录成本与延迟。
部署方案：Docker + Nginx 反向代理
生产部署推荐用 Nginx 做前置反向代理（TLS 终止、限流、静态资源），后端跑多个 FastAPI 容器实例做水平扩展。
Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY llm_gateway.py .
RUN pip install --no-cache-dir fastapi uvicorn httpx pydantic
EXPOSE 8000
CMD ["uvicorn", "llm_gateway:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
docker-compose.yml
version: "3.9"
services:
gateway:
build: .
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
deploy:
replicas: 3
restart: always
nginx:
image: nginx:alpine
ports:
- "443:443"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf:ro
- ./certs:/etc/nginx/certs:ro
depends_on:
- gateway
nginx.conf（关键片段）
upstream llm_gateway {
least_conn;
server gateway:8000;
}
server {
listen 443 ssl;
server_name llm-gateway.internal;
ssl_certificate /etc/nginx/certs/server.crt;
ssl_certificate_key /etc/nginx/certs/server.key;
location /v1/ {
proxy_pass http://llm_gateway;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_read_timeout 120s; # 大模型响应慢，超时调大
limit_req zone=api burst=50 nodelay; # 二级限流
}
}
# 在 http 块中定义限流区
# limit_req_zone $binary_remote_addr zone=api:10m rate=20r/s;
实操部署步骤：
- 把
llm_gateway.py、Dockerfile、docker-compose.yml、nginx.conf放到同一目录，准备好 TLS 证书目录./certs。 - 在
.env里填入真实的各厂商 API Key。 - 执行
docker compose up -d --scale gateway=3，启动 3 个网关实例 + 1 个 Nginx。 - 用
curl验证/v1/chat/completions与/v1/usage接口正常返回。 - 把内部业务系统的
OPENAI_BASE_URL指向https://llm-gateway.internal/v1，即可无缝切换到网关。
生产环境建议把内存中的预算、审计日志、令牌桶计数迁到 Redis（多实例共享状态），把审计记录落库到 PostgreSQL 或 ClickHouse 做长期分析。
常见问题 FAQ
Q1：业务方已经在用 OpenAI SDK，迁移到网关需要改代码吗？
不需要。OpenAI SDK 支持 base_url 参数，只要把它指向网关地址（如 https://llm-gateway.internal/v1），并把 API Key 换成网关签发的业务 Key 即可。网关对外暴露的是 OpenAI 兼容协议，对 SDK 透明。
Q2：多模型路由会不会导致回答质量不稳定？
会有这种风险，需要通过任务标签来约束。建议把「质量敏感型」任务（如复杂推理、长文写作）固定路由到高端模型，而把「成本敏感型」任务（分类、摘要、闲聊）开放给小模型路由。在 Router.select 里为质量敏感任务加 if req.task_type == "reason": return "gpt-4o" 这样的硬规则即可锁定。
Q3：令牌桶限流的 rate 和 capacity 怎么定？ rate 决定稳态 QPS，capacity 决定允许的瞬时突发。可以先按上游供应商的 RPM 限额倒推：例如 OpenAI 某档位是 500 RPM，换算约 8.3 QPS，可以把 rate 设为 8、capacity 设为 16 留一点缓冲。同时要在网关层做 Token 维度的限流（不只是请求数），因为一个长上下文请求消耗的配额远大于短请求。
Q4：熔断器打开后，多久恢复合适？
取决于上游故障类型。区域性故障通常几十秒到几分钟恢复，recovery_secs=30 是个比较稳妥的起点；如果是配额耗尽类故障，恢复时间要长得多，可以设到 5-10 分钟，并配合告警人工介入。建议把 fail_threshold 设为 5 左右，避免偶发抖动误触发熔断。
Q5：成本管控如何防止「某个用户用光所有人的预算」？
关键是按调用方（业务线/租户）独立核算预算。本文代码里 CostController 以 api_key 为 key 维护账户，每个账户有独立的 monthly_token_budget。生产环境还应加上「单请求 Token 上限」和「日预算预警」（用到 80% 时告警），做到事前拦截而非事后追责。
Q6：流式输出（streaming）怎么处理？
本文示例是非流式返回。要支持流式，需在 /v1/chat/completions 里判断 stream=True，然后用 FastAPI 的 StreamingResponse 把上游的 SSE 分片透传给客户端。成本计算需要在流结束时累加每个 chunk 的 usage（OpenAI 在最后一个 chunk 返回总 usage），熔断和限流逻辑保持不变。
总结
LLM API 网关的本质，是把「调用大模型」这件原本散落在业务各处、强耦合于单一供应商的事，收敛成一个可治理的中间层。本文从架构到代码给出了一套完整方案：
- 统一接口让业务和供应商解耦，换模型不改业务代码；
- 多模型路由按任务、成本、延迟动态选择最优模型；
- 令牌桶限流 + 熔断 + 指数退避 + 故障转移保障高可用；
- Token 预算与实时成本追踪让大模型花费可观测、可控；
- 审计日志支撑事后追溯与成本归因。
这套方案的价值不在于代码本身有多复杂，而在于它建立了一种治理范式：当大模型成为基础设施，它就必须像数据库、消息队列一样被网关化管理。你可以把本文的代码当作起点，逐步往里补 Redis 共享状态、Prometheus 监控、多租户权限、Prompt 缓存、敏感词过滤等能力，演进出贴合自身业务的网关体系。
相关文章推荐：
本文发布于 1630.top，转载请注明出处。