搭建 AI Gateway：统一模型访问入口与多模型路由

前两篇文章搭建了 Agent 并加上了监控。但 Agent 现在直连 LLM——API Key 散落在代码里，模型挂了就直接报错，没有灾备。这就像所有服务器直连公网，没有网关和负载均衡。

这篇文章搭建一个 AI Gateway，放 Agent 和 LLM 之间。

---

为什么需要 AI Gateway

AI Gateway 和 API Gateway 的定位一样——统一入口、路由、认证、限流、监控。AI 特有的额外需求：

能力	传统 API Gateway	AI Gateway 额外需求
统一入口	✅	所有模型走同一个 /v1/chat/completions
认证管理	✅	多 Provider 多 API Key 集中管理
路由	✅	按 model 名路由到不同 Provider
限流	✅	按模型+用户双重限流，防止账单爆炸
故障转移	❌	主模型故障 → 自动切备用模型
成本控制	❌	按 token 成本路由到最便宜的可用模型
可观测性	部分	Token 用量、成本、模型级延迟

最终架构：

Agent (LangGraph)
  │  base_url=http://localhost:8000/v1
  ▼
AI Gateway (:8000)
  │ 路由表: gpt-4o-mini → OpenAI
  │         deepseek-chat → DeepSeek
  │  fallback: gpt-4o-mini → deepseek-chat → qwen-plus
  ├─→ OpenAI API
  ├─→ DeepSeek API
  └─→ Qwen API

---

核心设计：Provider 注册表 + 模型路由表

从环境变量加载 Provider 配置，启动时自动构建路由表：

class Provider:
    def __init__(self, name: str, base_url: str, api_key: str, models: list[str]):
        self.name = name
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.models = models

def load_providers() -> list[Provider]:
    """从环境变量加载，支持 PROVIDER_OPENAI, PROVIDER_DEEPSEEK 等"""
    providers = []
    for key, value in os.environ.items():
        if not key.startswith("PROVIDER_") or not key.endswith("_API_KEY"):
            continue
        prefix = key.replace("_API_KEY", "")
        name = prefix.replace("PROVIDER_", "").lower()
        api_key = value
        base_url = os.getenv(f"{prefix}_BASE_URL", "")
        models_str = os.getenv(f"{prefix}_MODELS", "")
        models = [m.strip() for m in models_str.split(",") if m.strip()]
        if api_key and base_url and models:
            providers.append(Provider(name, base_url, api_key, models))
    return providers

配置示例（.env）：

PROVIDER_OPENAI_API_KEY=sk-xxx
PROVIDER_OPENAI_BASE_URL=https://api.openai.com/v1
PROVIDER_OPENAI_MODELS=gpt-4o-mini,gpt-4o

PROVIDER_DEEPSEEK_API_KEY=sk-xxx
PROVIDER_DEEPSEEK_BASE_URL=https://api.deepseek.com/v1
PROVIDER_DEEPSEEK_MODELS=deepseek-chat

设计要点：命名约定 PROVIDER_<NAME>_API_KEY / _BASE_URL / _MODELS 让增加新 Provider 只需加三行环境变量，零代码改动。

---

核心逻辑：路由 + 故障转移

# 模型名 → (Provider, provider_side_model_name)
MODEL_ROUTING: dict[str, tuple[Provider, str]] = {}

# 故障转移链：主模型不可用时，按顺序尝试
FALLBACK_CHAIN = {
    "gpt-4o":         ["gpt-4o-mini", "deepseek-chat"],
    "gpt-4o-mini":    ["deepseek-chat", "qwen-plus"],
    "deepseek-chat":  ["qwen-plus"],
}

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    body = await request.json()
    model_name = body.get("model")

    # 1. 限流检查
    if not check_rate_limit():
        raise HTTPException(status_code=429)

    # 2. 查找路由
    entry = MODEL_ROUTING.get(model_name)
    if not entry:
        raise HTTPException(status_code=400, detail=f"Unknown model: {model_name}")

    # 3. 尝试调用（含 fallback）
    fallback_models = FALLBACK_CHAIN.get(model_name, [])
    for fb_model in [model_name] + fallback_models:
        provider, provider_model = MODEL_ROUTING.get(fb_model, (None, None))
        if not provider:
            continue
        try:
            result = await _call_provider(provider, provider_model, body)
            return result
        except Exception as e:
            logger.error(f"  ✗ {fb_model}: {e}")
            await asyncio.sleep(0.5)  # 短暂等待后重试
            continue

    # 4. 所有 provider 都失败
    raise HTTPException(status_code=502, detail="All models failed")

为什么这样设计：

• 用 for ... in [model_name] + fallback_models 循环，把主模型和备用模型统一处理
• 每次失败后 await asyncio.sleep(0.5) 避免雪崩——如果 provider 侧正在恢复，短暂等待可以提高成功率
• 返回 502 Bad Gateway 而非 500——语义上更准确，调用方可以据此判断是网关下游问题

---

限流：滑动窗口

_request_counts: dict[int, list[float]] = defaultdict(list)

def check_rate_limit() -> bool:
    now = time.time()
    window_start = now - 60
    minute_key = int(now / 60)
    _request_counts[minute_key] = [
        t for t in _request_counts.get(minute_key, []) if t > window_start
    ]
    if len(_request_counts[minute_key]) >= RATE_LIMIT:
        return False
    _request_counts[minute_key].append(now)
    return True

按分钟分桶，每个桶内保留最近 60 秒的记录。不依赖 Redis，单进程即可工作。生产环境可以换成 Redis 实现跨进程限流。

---

指标端点：/metrics

输出 Prometheus 兼容格式，可以直接接入 Grafana：

# HELP ai_gateway_requests_total Total requests
# TYPE ai_gateway_requests_total counter
ai_gateway_requests_total 42

# HELP ai_gateway_requests_success Successful requests
ai_gateway_requests_success 40

# HELP ai_gateway_requests_errors Failed requests
ai_gateway_requests_errors 2

# HELP ai_gateway_latency_ms_avg Average latency (last 100)
ai_gateway_latency_ms_avg 1234.5

# HELP ai_gateway_tokens_total Total tokens processed
ai_gateway_tokens_total 28450

---

Agent 端接入

Agent 端只需改一行——把 base_url 指向 Gateway：

# 之前：直连 OpenAI
llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url="https://api.openai.com/v1",
)

# 之后：走 Gateway
llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key="not-needed",       # Gateway 不验证 client key
    base_url="http://localhost:8000/v1",  # ← 指向 Gateway
)

Agent 代码其余部分完全不变。Gateway 对客户端透明——这就是 OpenAI 兼容协议的价值。

---

实际效果

启动 Gateway 后运行测试脚本：

🚀 AI Gateway 启动
  监听端口: 8000
  限流: 60 req/min
  Provider 数: 2
  路由表: 3 个模型
    gpt-4o-mini → openai → fallback: ['deepseek-chat', 'qwen-plus']
    gpt-4o → openai → fallback: ['gpt-4o-mini', 'deepseek-chat']
    deepseek-chat → deepseek → fallback: ['qwen-plus']

🧪 AI Gateway 功能测试

1. 健康检查
  GET /health → 200 {"status":"ok","providers":2,"models":3}

2. 基本对话
  ✓ gpt-4o-mini → openai (1234ms, 156 tokens)
  回复: Kubernetes 是一个开源的容器编排平台...

3. 流式响应 (SSE)
  ✓ gpt-4o-mini → openai (stream)
  共收到 15 个 chunk

4. 故障转移
  ✗ gpt-4o (openai): HTTP 401
  ↳ 故障转移: gpt-4o → gpt-4o-mini (openai)
  ✓ gpt-4o-mini → openai (987ms, 89 tokens)

5. 未知模型
  400 {"detail":"Unknown model 'nonexistent-model'..."}

6. 限流测试
  请求 1: 200
  请求 2: 200
  ...
  请求 7: 429 (限流)

注意第 4 步：请求 gpt-4o 但没配 API Key → 自动 fallback 到 gpt-4o-mini → 成功返回。对调用方完全透明。

---

生产环境还需要什么

能力	当前实现	生产增强
配置管理	环境变量	配置中心 / 热更新
限流	单进程内存	Redis 分布式限流
认证	无	Client API Key 验证
缓存	无	相同请求命中缓存，省 Token
日志	stdout	结构化日志 → ELK / Loki
高可用	单实例	多实例 + 负载均衡
流式 Token 统计	未实现	解析 SSE 事件累计

---

完整代码

• Gateway 主程序：demo/ai-gateway/gateway.py
• 测试脚本：demo/ai-gateway/test_gateway.py

---

下一步

Gateway 让 Agent 和模型之间有了统一入口。下一篇文章回到 Agent 工程主线——RAG 系统部署实战：从本地 Chroma 到生产环境，包括索引构建、检索优化和部署考量。