[!info] related notes

• 所属 MOC: AI Agent Application MOC
• 相关: LLM Provider Abstraction, Model Router, Fallback Strategy

Model Gateway

一句话定义

Model Gateway 是 AI Service 内部统一对接多个 LLM Provider 的网关层。它在 LLM Provider Abstraction 之上，增加了路由选择、负载均衡、Fallback、重试、token 追踪和成本控制等能力。

它解决什么问题

有了 LLM Provider Abstraction 还不够：

• 同一个 Provider 可能有多个 API Key（负载均衡）
• 一个 Provider 挂了需要切换到另一个（Fallback）
• 不同任务用不同模型更经济（路由）
• 需要追踪每次调用的 token 消耗（成本控制）

核心原理

Gateway 架构

Agent Engine
    │
    ▼
Model Gateway
    ├─ Router: 根据任务类型选择模型
    ├─ Load Balancer: 多个 API Key 轮询
    ├─ Fallback: Provider A 失败 → Provider B
    ├─ Retry: 失败重试（指数退避）
    ├─ Token Tracker: 记录每次调用的 token
    └─ Cost Controller: 检查预算是否超限
    │
    ▼
LLM Provider (OpenAI / Anthropic / 本地)

Python 实现

class ModelGateway:
    def __init__(self, providers: dict, router: ModelRouter):
        self.providers = providers
        self.router = router
        self.token_tracker = TokenTracker()
        self.cost_controller = CostController()

    async def chat(self, request: ChatRequest) -> ChatResponse:
        # 1. 路由选择模型
        model_config = self.router.select(request.task_type)

        # 2. 成本检查
        if not self.cost_controller.can_afford(model_config.model):
            model_config = self.router.get_cheaper_alternative(model_config)

        # 3. 带重试和 Fallback 的调用
        for provider_name in model_config.providers:
            provider = self.providers[provider_name]
            try:
                response = await self._call_with_retry(
                    provider, request, model_config.model
                )
                # 4. 记录 token
                self.token_tracker.record(
                    model=model_config.model,
                    input_tokens=response.usage["input_tokens"],
                    output_tokens=response.usage["output_tokens"],
                )
                return response
            except Exception as e:
                logger.warning(f"Provider {provider_name} failed: {e}")
                continue

        raise AllProvidersFailedError()

    async def _call_with_retry(self, provider, request, model, max_retries=3):
        for attempt in range(max_retries):
            try:
                return await provider.chat(
                    messages=request.messages,
                    model=model,
                    tools=request.tools,
                )
            except RateLimitError:
                await asyncio.sleep(2 ** attempt)
            except Exception:
                if attempt == max_retries - 1:
                    raise

常见坑

1. 不做 Fallback: 单 Provider 依赖
2. 不做成本控制: token 费用超预算
3. 重试太激进: 触发 Provider 限流
4. 不追踪 token: 无法分析成本

参考资料

• Model Gateway
• LLM Provider Abstraction