一个 FastAPI 服务挡在应用与 N 个 LLM 供应商之间：优先级路由、故障转移、熔断、响应缓存、按客户端限流、Prometheus 指标。应用侧只改一个 base_url，供应商挂了、超卖了、换模型了都由网关消化。
| 能力 | 行为 | 测试 |
|---|---|---|
| 优先级 + 权重路由 | 同优先级按 weight 加权分流；低优先级 tier 全挂时自动降级到下一 tier | test_routes_to_primary… / test_router_prefers_lower_priority_tier |
| 故障转移 | 上游 5xx/429/超时 → 记一次失败 → 立即重试 failover 候选；响应里带 gateway.attempted 链路 |
test_failover_on_primary_500（断言 ["primary","secondary"] 两跳） |
| 熔断器 | 连续 2 次失败 → 冷却 30s 内不再选中（half-open 半开试探） | test_circuit_opens_after_repeated_failures（断言跳过 primary 直达 secondary） |
| 响应缓存 | LRU + TTL，键 = (model, messages, temperature)；命中则不触上游 | test_cache_hit_on_identical_request（断言上游只被调用 1 次） |
| 令牌桶限流 | 按 X-Client-Id（缺省按 IP）每客户端独立限速，超限 429 |
test_rate_limit_returns_429 / test_rate_limit_isolated_per_client |
| 模型改写 | 网关把请求里的 model 改写为各供应商自己的模型 ID |
test_routes_to_primary_and_passes_model |
| 可观测 | /healthz（存活+各供应商可用性）、/metrics（Prometheus 文本）、每响应带 X-Request-ID |
test_healthz_metrics_and_providers_endpoints |
| 密钥安全 | 供应商 key 只从环境变量读取，/v1/providers 不回显任何密钥 |
test_api_key_sent_from_env + 密钥泄露断言 |
uv sync
uv run pytest # 16 passed（全部跑在 mock 上游上，零网络）
# 本地起服务（默认拓扑：DeepSeek 主 + OpenAI 备，key 从环境变量读）
export PRIMARY_API_KEY=... FALLBACK_API_KEY=...
uv run uvicorn llm_gateway.app:create_app --factory --port 8000应用侧只需要把 OpenAI SDK 的 base_url 指到网关：
from openai import OpenAI
client = OpenAI(base_url="http://localhost:8000/v1", api_key="anything")
resp = client.chat.completions.create(
model="gateway", # 网关改写为各供应商的真实模型 ID
messages=[{"role": "user", "content": "你好"}],
)
print(resp.choices[0].message.content, resp.model_extra["gateway"])
# gateway = {"provider": "primary", "cache": "miss", "attempted": ["primary"]}实测启动冒烟（本机）：
$ curl localhost:8000/healthz
{"ok":true,"providers":{"primary":true,"fallback":true}}
$ curl localhost:8000/v1/providers
[{"name":"primary","model":"deepseek-chat","priority":0,...},
{"name":"fallback","model":"gpt-4o-mini","priority":1,...}]
from llm_gateway.config import GatewayConfig, Provider
from llm_gateway.app import create_app
config = GatewayConfig(
providers=[
Provider(name="deepseek", base_url="https://api.deepseek.com/v1",
model="deepseek-chat", api_key_env="DEEPSEEK_KEY",
weight=0.8, priority=0), # 承接 80% 流量
Provider(name="vllm-selfhost", base_url="http://10.0.0.5:8000/v1",
model="qwen-72b", weight=0.2, priority=0), # 20% 分流，无 key
Provider(name="openai", base_url="https://api.openai.com/v1",
model="gpt-4o-mini", api_key_env="OPENAI_KEY",
priority=1), # 兜底 tier
],
rate_limit_per_minute=120,
cache_ttl_s=600,
)
app = create_app(config)llm_gateway/
├── config.py # Provider/GatewayConfig：拓扑声明式配置，key 只认环境变量
├── router.py # 优先级 tier + 权重选择 + 熔断（ProviderHealth）
├── cache.py # SHA-256 键的 LRU+TTL 响应缓存
├── ratelimit.py # 每客户端令牌桶
├── metrics.py # 无依赖 Prometheus 文本输出
└── app.py # FastAPI：/v1/chat/completions + 运维端点，failover 编排
tests/ # 16 个集成测试，httpx.MockTransport 假冒全部上游
Dockerfile # 容器部署
定位不同：LiteLLM 侧重上百个供应商的统一适配与 SDK 代理，one-api 侧重多渠道分发、令牌与计费管理后台。llm-gateway 是约 600 行源码的轻量自托管网关，专注「OpenAI 兼容入口 + 优先级路由 / 故障转移 / 熔断 / 响应缓存 / 限流」这一条链路，每项行为都有集成测试锁定；路由拓扑用 Python dataclass 声明（配置即代码），无数据库依赖、状态全内存。适合用来理解网关原理和做二次开发，不适合直接当多租户计费平台用。
触发阈值固定为「连续 2 次失败即熔断」（ProviderHealth 的默认 threshold=2，暂未暴露到 GatewayConfig）；冷却时长可配：GatewayConfig(circuit_cooldown_s=...)，默认 30 秒。冷却到期后进入 half-open，该供应商重新参与选择，一次成功即清零失败计数。
GET /metrics 直接返回 Prometheus 文本格式（metrics.py 自带实现，不依赖 prometheus_client），包含 gateway_requests_total、gateway_upstream_errors、gateway_upstream_rate_limited、gateway_rate_limited、gateway_cache_hits、gateway_latency_ms_sum，以及缓存条目数与命中率。Prometheus 侧把 scrape target 指到网关地址的 /metrics 即可；指标存于进程内存，重启后清零。
键 = SHA-256(model, messages, temperature, tools)，temperature 缺省按 1.0 参与（见 cache.py）。TTL 内完全相同的请求直接命中缓存、不触上游。当前没有按请求绕过缓存的开关——需要每次拿到不同结果的请求（如随机创意生成）要意识到这一点。
优先取请求头 X-Client-Id，缺省用对端 IP。每个客户端一个独立令牌桶，速率统一由 GatewayConfig(rate_limit_per_minute=...) 控制（默认 60/分钟），超限返回 OpenAI 风格的 429 rate_limit_error。
任何提供 OpenAI 兼容 /chat/completions 的端点都可以：在 GatewayConfig.providers 里加一个 Provider(name=..., base_url=..., model=..., api_key_env=..., priority=..., weight=...)，key 只从 api_key_env 指定的环境变量读取。vLLM、Ollama 等自托管端点同样适用——不填 api_key_env 即为无鉴权上游。
- 不支持流式（SSE）：
/v1/chat/completions把上游响应整体读成 JSON 后返回，stream=true不会被透传为 SSE（见 Roadmap 第一项）。 - 仅覆盖 chat/completions 一个端点：没有
/v1/models、embeddings、images 等；请求侧只校验messages非空，其余参数原样透传给上游。 - 状态全内存、面向单实例：缓存、熔断计数、限流桶、指标都在进程内存里，重启即清零；多副本部署时各副本独立计数（缓存命中率与限流精度都会打折），暂无 Redis 等外置存储。
- 上游调用是同步 HTTP 客户端：内部用同步
httpx.Client在 async 处理器里发请求，等待上游期间会占用事件循环，高并发吞吐受限——当前定位是低并发自托管场景。 - 配置即代码，无热更新：拓扑与阈值只能改 Python 代码后重启生效，没有管理后台，也没有 YAML/环境变量级配置。
- 未经生产长稳验证：16 个集成测试全部跑在
httpx.MockTransport假上游上，真实多供应商环境的长期稳定性与极端场景（上游 hang 住、TLS 抖动等）未做系统性验证。
MIT