📚 AI 工程与智能体教程站

第 15 章 · 多模型网关与成本优化 ​

本章目标：

  • 掌握多 Provider 路由策略
  • 实现基于任务复杂度的智能路由
  • 设计成本监控和预算告警
  • 实现故障降级机制

15.1 多网关架构 ​

python
from ai import createGateway, createOpenAICompatible

# Vercel AI Gateway
gateway = createGateway({
    apiKey: process.env.AI_GATEWAY_API_KEY
})

# 自定义 OpenAI 兼容 Provider
my_provider = createOpenAICompatible({
    name: "my-provider",
    baseURL: process.env.BASE_URL,
    apiKey: process.env.API_KEY
})

15.2 智能路由 ​

python
from enum import Enum
from typing import Literal

class TaskComplexity(Enum):
    SIMPLE = "simple"      # 简单问答
    MEDIUM = "medium"     # 需要推理
    COMPLEX = "complex"   # 复杂规划

class Router:
    def __init__(self):
        self.gateway = createGateway({apiKey: "xxx"})
        self.providers = {
            "cheap": self.gateway("openai/gpt-4o-mini"),
            "medium": self.gateway("anthropic/claude-haiku-4-5"),
            "strong": self.gateway("anthropic/claude-sonnet-4-6"),
            "expert": self.gateway("openai/o3-mini")
        }
    
    def classify(self, prompt: str) -> TaskComplexity:
        """简单规则分类（生产环境可用 LLM 判断）"""
        if len(prompt) < 50 and "?" not in prompt:
            return TaskComplexity.SIMPLE
        elif "分析" in prompt or "解释" in prompt:
            return TaskComplexity.COMPLEX
        return TaskComplexity.MEDIUM
    
    def route(self, prompt: str) -> str:
        complexity = self.classify(prompt)
        
        models = {
            TaskComplexity.SIMPLE: "cheap",
            TaskComplexity.MEDIUM: "medium",
            TaskComplexity.COMPLEX: "expert"
        }
        
        return self.providers[models[complexity]]

15.3 成本监控 ​

python
import asyncio
from dataclasses import dataclass
from typing import Dict

@dataclass
class CostTracker:
    budget_limit: float
    spent: float = 0.0
    
    # 各模型价格（每 1M tokens）
    prices = {
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "claude-haiku": {"input": 0.25, "output": 1.25},
        "o3-mini": {"input": 3.00, "output": 12.00}
    }
    
    def track(self, model: str, input_tokens: int, output_tokens: int):
        prices = self.prices.get(model, self.prices["gpt-4o-mini"])
        cost = (input_tokens * prices["input"] + output_tokens * prices["output"]) / 1_000_000
        self.spent += cost
        
        if self.spent >= self.budget_limit * 0.9:
            print(f"⚠️  预算警告：已使用 {self.spent:.2f}/${self.budget_limit:.2f}")
        
        return cost

# 使用
tracker = CostTracker(budget_limit=100.0)  # $100 月度预算

async def tracked_call(prompt: str, model: str):
    result = await call_llm(prompt, model)
    tracker.track(model, len(prompt), len(result))
    return result

15.4 故障降级 ​

python
from tenacity import retry, stop_after_attempt

class FallbackRouter:
    def __init__(self):
        self.primary = self.gateway("anthropic/claude-sonnet-4-6")
        self.fallback = self.gateway("openai/gpt-4o")
    
    @retry(stop=stop_after_attempt(2))
    async def call(self, prompt: str) -> str:
        try:
            return await self.primary(prompt)
        except Exception as e:
            print(f"主模型失败: {e}，切换到备用")
            return await self.fallback(prompt)

本章小结 ​

  • 多网关架构支持模型级联和故障转移
  • 智能路由根据任务复杂度选模型
  • 成本追踪防止预算超支
  • 降级机制提升服务韧性

🛠️ 动手实践 ​

  1. 实现三档路由（简单/中等/复杂）映射到不同模型
  2. 添加预算告警，超支时自动切换到低成本模型
  3. 实现主备降级，主模型超时 5s 自动切备用