AI API 网关实战：多模型统一调度、自动降级与成本追踪

┌─────────────────────────────────────────────────┐
│                  AI API 网关                      │
├─────────────────────────────────────────────────┤
│  应用层                                          │
│  POST /v1/chat/completions  ← 统一 OpenAI 兼容接口│
├─────────────────────────────────────────────────┤
│  路由层                                          │
│  ModelRouter: 根据任务类型/成本/可用性选择模型      │
├─────────────────────────────────────────────────┤
│  适配层                                          │
│  OpenAIAdapter / ClaudeAdapter / DeepSeekAdapter  │
│  统一格式 → 各模型原生格式                        │
├─────────────────────────────────────────────────┤
│  管理层                                          │
│  FallbackManager: 故障自动降级                    │
│  CostTracker: 实时费用追踪                        │
│  RateLimiter: 令牌桶限流                          │
└─────────────────────────────────────────────────┘

import os
import time
import hashlib
from typing import Optional
from dataclasses import dataclass, field
from openai import OpenAI

@dataclass
class ModelConfig:
    """模型配置"""
    name: str                    # 显示名称
    provider: str                # openai / anthropic / deepseek / glm
    model_id: str                # API model ID
    api_key: str
    base_url: str
    tier: str = "mid"            # cheap / mid / premium
    cost_input_per_1m: float = 0  # ¥/百万输入token
    cost_output_per_1m: float = 0
    max_retries: int = 2
    timeout: int = 60
    is_available: bool = True


class ModelAdapter:
    """统一模型适配器——对外暴露统一接口，内部处理各模型差异"""

    def __init__(self, config: ModelConfig):
        self.config = config
        self.client = OpenAI(
            api_key=config.api_key,
            base_url=config.base_url,
            timeout=config.timeout,
            max_retries=config.max_retries,
        )
        self._total_tokens = 0
        self._total_cost = 0.0
        self._call_count = 0

    def chat(self, messages: list, **kwargs) -> dict:
        """统一的 chat 接口。自动处理各模型的格式差异。"""
        # 构建模型特定参数
        params = self._build_params(messages, **kwargs)

        start = time.time()
        response = self.client.chat.completions.create(**params)
        latency_ms = int((time.time() - start) * 1000)

        # 提取统一的返回格式
        result = self._normalize_response(response)

        # 记录用量
        usage = response.usage
        tokens = usage.total_tokens
        cost = (
            usage.prompt_tokens * self.config.cost_input_per_1m / 1_000_000 +
            usage.completion_tokens * self.config.cost_output_per_1m / 1_000_000
        )
        self._total_tokens += tokens
        self._total_cost += cost
        self._call_count += 1

        return {
            "content": result["content"],
            "model": self.config.name,
            "tokens": tokens,
            "cost": cost,
            "latency_ms": latency_ms,
            "finish_reason": result["finish_reason"],
        }

    def _build_params(self, messages: list, **kwargs) -> dict:
        """构建各模型的原生请求参数"""
        base = {
            "model": self.config.model_id,
            "temperature": kwargs.get("temperature", 0.7),
            "max_tokens": kwargs.get("max_tokens", 2048),
        }

        if self.config.provider == "anthropic":
            # Claude 的 system prompt 在顶层
            system_msg = None
            chat_messages = []
            for m in messages:
                if m["role"] == "system":
                    system_msg = m["content"]
                else:
                    chat_messages.append(m)
            base["messages"] = chat_messages
            if system_msg:
                base["system"] = system_msg
            if "max_tokens" in base:
                base["max_tokens"] = base.pop("max_tokens")
        else:
            base["messages"] = messages

        return base

    def _normalize_response(self, response) -> dict:
        """统一各模型的返回格式"""
        if self.config.provider == "anthropic":
            return {
                "content": response.choices[0].message.content,
                "finish_reason": response.choices[0].finish_reason or "stop",
            }
        else:
            return {
                "content": response.choices[0].message.content or "",
                "finish_reason": response.choices[0].finish_reason or "stop",
            }

    @property
    def stats(self) -> dict:
        return {
            "model": self.config.name,
            "calls": self._call_count,
            "total_tokens": self._total_tokens,
            "total_cost": round(self._total_cost, 4),
        }

用户请求 → 复杂度分析 → 选择 tier
├─ 简单（翻译/分类/摘要） → cheap: DeepSeek-V4 (¥2/1M)
├─ 中等（问答/分析）       → mid: GLM-5.2 (¥15/1M)
└─ 复杂（写代码/推理）     → premium: Claude Opus (¥180/1M)

class ModelRouter:
    """模型路由器——任务分类 + 模型选择"""

    # 任务复杂度判断规则
    COMPLEX_PATTERNS = [
        "写代码", "实现一个", "重构", "架构设计",
        "debug", "优化这段代码", "解释这个算法",
        "数学证明", "推导", "设计模式",
    ]
    SIMPLE_PATTERNS = [
        "翻译", "总结", "摘要", "分类",
        "提取", "转换格式", "纠错", "润色",
    ]

    def __init__(self, adapters: dict[str, list[ModelAdapter]]):
        """
        adapters: {"cheap": [deepseek, ...], "mid": [glm, ...], "premium": [claude, gpt, ...]}
        每个 tier 可以有多个备用模型
        """
        self.adapters = adapters

    def route(self, messages: list) -> tuple[str, ModelAdapter]:
        """根据任务复杂度选择 tier 和模型"""
        user_msg = ""
        for m in reversed(messages):
            if m["role"] == "user":
                user_msg = m["content"]
                break

        # 判断复杂度
        msg_lower = user_msg.lower()
        for pattern in self.COMPLEX_PATTERNS:
            if pattern.lower() in msg_lower:
                return self._pick_adapter("premium")

        for pattern in self.SIMPLE_PATTERNS:
            if pattern.lower() in msg_lower:
                return self._pick_adapter("cheap")

        return self._pick_adapter("mid")

    def _pick_adapter(self, tier: str) -> tuple[str, ModelAdapter]:
        """从指定 tier 选择可用的适配器（负载均衡：选调用次数最少的）"""
        adapters = self.adapters.get(tier, [])
        available = [a for a in adapters if a.config.is_available]
        if not available:
            # 降级到下一个 tier
            if tier == "premium":
                return self._pick_adapter("mid")
            elif tier == "mid":
                return self._pick_adapter("cheap")
            else:
                raise RuntimeError("All models unavailable")

        # 最少使用负载均衡
        adapter = min(available, key=lambda a: a._call_count)
        return tier, adapter

class FallbackManager:
    """故障降级管理器"""

    def __init__(self, router: ModelRouter):
        self.router = router
        self.failure_counts: dict[str, int] = {}  # 每个模型的连续失败次数
        self.cooldown_until: dict[str, float] = {}  # 冷却结束时间
        self.max_failures = 3       # 连续失败N次后标记为不可用
        self.cooldown_seconds = 60  # 冷却60秒后重试

    def execute_with_fallback(self, messages: list, **kwargs) -> dict:
        """执行请求，带自动降级"""
        last_error = None

        # 尝试最多 3 次（不同 tier 的降级）
        for attempt in range(3):
            tier, adapter = self.router.route(messages)

            # 检查是否在冷却中
            if adapter.config.name in self.cooldown_until:
                if time.time() = self.max_failures:
                    adapter.config.is_available = False
                    self.cooldown_until[name] = time.time() + self.cooldown_seconds
                    print(f"[FALLBACK] {name} 连续失败 {self.max_failures} 次，标记不可用，{self.cooldown_seconds}秒后重试")

        raise RuntimeError(f"All fallback attempts exhausted. Last error: {last_error}")

class CostTracker:
    """实时成本追踪器"""

    def __init__(self):
        self.daily_costs: dict[str, list] = {}  # date -> [(timestamp, model, tokens, cost)]

    def record(self, model: str, tokens: int, cost: float):
        today = time.strftime("%Y-%m-%d")
        if today not in self.daily_costs:
            self.daily_costs[today] = []
        self.daily_costs[today].append((int(time.time()), model, tokens, cost))

    def get_daily_summary(self, date: str = None) -> dict:
        date = date or time.strftime("%Y-%m-%d")
        records = self.daily_costs.get(date, [])
        if not records:
            return {"date": date, "total_cost": 0, "total_tokens": 0, "calls": 0}

        by_model = {}
        for _, model, tokens, cost in records:
            if model not in by_model:
                by_model[model] = {"tokens": 0, "cost": 0.0, "calls": 0}
            by_model[model]["tokens"] += tokens
            by_model[model]["cost"] += cost
            by_model[model]["calls"] += 1

        return {
            "date": date,
            "total_cost": round(sum(r[3] for r in records), 4),
            "total_tokens": sum(r[2] for r in records),
            "total_calls": len(records),
            "by_model": by_model,
        }

    def get_cost_alert(self, daily_budget: float = 10.0) -> Optional[str]:
        """预算告警——超过日预算80%时触发"""
        summary = self.get_daily_summary()
        if summary["total_cost"] > daily_budget * 0.8:
            return f"⚠️ 今日费用 ¥{summary['total_cost']:.2f}，已达日预算 {daily_budget} 元的 {summary['total_cost']/daily_budget*100:.0f}%"
        return None

class AIGateway:
    """AI API 网关——统一入口"""

    def __init__(self):
        # 初始化模型适配器
        self.adapters = {
            "cheap": [
                ModelAdapter(ModelConfig(
                    name="DeepSeek-V4", provider="deepseek",
                    model_id="deepseek-chat",
                    api_key=os.getenv("DEEPSEEK_API_KEY"),
                    base_url="https://api.deepseek.com/v1",
                    tier="cheap",
                    cost_input_per_1m=2, cost_output_per_1m=8,
                )),
            ],
            "mid": [
                ModelAdapter(ModelConfig(
                    name="GLM-5.2", provider="glm",
                    model_id="glm-5.2-flash",
                    api_key=os.getenv("GLM_API_KEY"),
                    base_url="https://open.bigmodel.cn/api/paas/v4",
                    tier="mid",
                    cost_input_per_1m=15, cost_output_per_1m=30,
                )),
            ],
            "premium": [
                ModelAdapter(ModelConfig(
                    name="Claude-Opus-4.8", provider="anthropic",
                    model_id="claude-opus-4-8-20250219",
                    api_key=os.getenv("ANTHROPIC_API_KEY"),
                    base_url="https://api.anthropic.com/v1",
                    tier="premium",
                    cost_input_per_1m=180, cost_output_per_1m=720,
                )),
                ModelAdapter(ModelConfig(
                    name="GPT-5.5", provider="openai",
                    model_id="gpt-5.5",
                    api_key=os.getenv("OPENAI_API_KEY"),
                    base_url="https://api.openai.com/v1",
                    tier="premium",
                    cost_input_per_1m=180, cost_output_per_1m=720,
                )),
            ],
        }

        self.router = ModelRouter(self.adapters)
        self.fallback = FallbackManager(self.router)
        self.cost_tracker = CostTracker()

    def chat(self, messages: list, **kwargs) -> str:
        """统一的聊天接口——自动路由 + 降级 + 成本追踪"""
        result = self.fallback.execute_with_fallback(messages, **kwargs)
        self.cost_tracker.record(result["model"], result["tokens"], result["cost"])
        return result["content"]

    def get_stats(self):
        return {
            "adapters": {tier: [a.stats for a in adapters]
                         for tier, adapters in self.adapters.items()},
            "daily": self.cost_tracker.get_daily_summary(),
            "alert": self.cost_tracker.get_cost_alert(),
        }