一、问题：你的 LLM 调用散落在各处

# service_a.py
from openai import OpenAI
client = OpenAI(api_key="sk-xxx")
result = client.chat.completions.create(model="gpt-4o", messages=[...])

# service_b.py
from anthropic import Anthropic
claude = Anthropic(api_key="sk-yyy")
result = claude.messages.create(model="claude-sonnet-4-20250514", messages=[...])

问题：

• API Key 散落各处，轮换需改 N 个地方
• 无统一重试/超时策略
• 无流量控制，配额可能被打满
• 换模型需改代码重部署
• 看不到全局成本

方案：在应用和 LLM 之间加网关层。

二、网关架构

                    ┌──────────────────────┐
                    │    LLMGateway        │
                    │                      │
  chat(messages,    │  ┌────────────────┐  │
   strategy="cost") │  │   Router       │  │
         ──────────►│  │ · ModelRegistry│  │
                    │  └───────┬────────┘  │
                    │          │           │
                    │  ┌───────▼────────┐  │
                    │  │ Rate Limiter   │  │
                    │  │ · TokenBucket  │  │
                    │  └───────┬────────┘  │
                    │          │           │
                    │  ┌───────▼────────┐  │
                    │  │ Circuit Breaker│  │
                    │  └───────┬────────┘  │
                    │          │           │
                    │  ┌───────▼────────┐  │
                    │  │ Provider Client│  │
                    │  └───────┬────────┘  │
                    │          │           │
                    │  ┌───────▼────────┐  │
                    │  │ Cost Tracker   │  │
                    │  └────────────────┘  │
                    └──────────────────────┘

三、完整实现

# llm_gateway.py - LLM API 网关
# 核心能力：多模型路由、令牌桶限流、熔断器、退避重试、成本追踪

import time
import threading
import hashlib
from enum import Enum
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any
from collections import defaultdict
from openai import OpenAI


# ============================================================
# 1. 模型注册表
# ============================================================

class ModelCapability(Enum):
    CHAT = "chat"
    EMBEDDING = "embedding"
    VISION = "vision"
    FUNCTION_CALL = "function_call"


@dataclass
class ModelEntry:
    # 单个模型条目
    provider: str                     # openai / anthropic / deepseek
    model_id: str                     # gpt-4o-mini / claude-sonnet-4-...
    capabilities: List[ModelCapability]
    input_price: float                # USD per 1M input tokens
    output_price: float               # USD per 1M output tokens
    avg_latency_ms: float = 500
    max_tokens: int = 128000
    rpm_limit: int = 500
    enabled: bool = True
    priority: int = 1                 # 越小越优先
    _api_key: str = ""
    _base_url: Optional[str] = None

    @property
    def api_key(self) -> str:
        return self._api_key

    def set_credentials(self, api_key: str, base_url: Optional[str] = None):
        self._api_key = api_key
        self._base_url = base_url


class ModelRegistry:
    # 模型注册表, key = "provider:model_id"

    def __init__(self):
        self._models: Dict[str, ModelEntry] = {}

    def register(self, entry: ModelEntry, api_key: str,
                 base_url: Optional[str] = None) -> str:
        key = f"{entry.provider}:{entry.model_id}"
        entry.set_credentials(api_key, base_url)
        self._models[key] = entry
        return key

    def get(self, key: str) -> Optional[ModelEntry]:
        return self._models.get(key)

    def list_by_capability(self, cap: ModelCapability) -> List[ModelEntry]:
        return [m for m in self._models.values()
                if cap in m.capabilities and m.enabled]

    def list_enabled(self) -> List[ModelEntry]:
        return [m for m in self._models.values() if m.enabled]

    def disable(self, key: str):
        if key in self._models:
            self._models[key].enabled = False

    def enable(self, key: str):
        if key in self._models:
            self._models[key].enabled = True


# ============================================================
# 2. 路由策略
# ============================================================

class RoutingStrategy(Enum):
    COST_OPTIMIZED = "cost"          # 选最便宜的
    LATENCY_OPTIMIZED = "latency"    # 选最快的
    QUALITY_OPTIMIZED = "quality"    # 选最强的
    ROUND_ROBIN = "round_robin"      # 轮询
    FALLBACK_CHAIN = "fallback"      # 优先级链式降级


class Router:
    # 路由器：根据策略选择模型

    def __init__(self, registry: ModelRegistry):
        self.registry = registry
        self._rr_idx: Dict[str, int] = defaultdict(int)

    def select(self, capability: ModelCapability = ModelCapability.CHAT,
               strategy: RoutingStrategy = RoutingStrategy.COST_OPTIMIZED,
               exclude: Optional[List[str]] = None) -> Optional[ModelEntry]:
        candidates = self.registry.list_by_capability(capability)
        if exclude:
            ek = set(exclude)
            candidates = [m for m in candidates
                          if f"{m.provider}:{m.model_id}" not in ek]
        if not candidates:
            return None

        if strategy == RoutingStrategy.COST_OPTIMIZED:
            return min(candidates, key=lambda m: m.input_price + m.output_price)
        elif strategy == RoutingStrategy.LATENCY_OPTIMIZED:
            return min(candidates, key=lambda m: m.avg_latency_ms)
        elif strategy == RoutingStrategy.QUALITY_OPTIMIZED:
            return min(candidates, key=lambda m: m.priority)
        elif strategy == RoutingStrategy.ROUND_ROBIN:
            k = capability.value
            idx = self._rr_idx[k] % len(candidates)
            self._rr_idx[k] += 1
            return candidates[idx]
        else:  # FALLBACK_CHAIN
            return sorted(candidates, key=lambda m: m.priority)[0]

    def fallback_chain(self, cap: ModelCapability = ModelCapability.CHAT
                       ) -> List[ModelEntry]:
        return sorted(self.registry.list_by_capability(cap),
                      key=lambda m: m.priority)


# ============================================================
# 3. 令牌桶限流器
# ============================================================

class TokenBucket:
    # 令牌桶: 固定速率填充, 请求消耗令牌, 令牌不足即拒绝

    def __init__(self, rate: float, capacity: int):
        self.rate = rate          # 填充速率 (个/秒)
        self.capacity = capacity  # 桶容量 (允许突发)
        self._tokens = float(capacity)
        self._last_refill = time.monotonic()
        self._lock = threading.Lock()

    def consume(self, tokens: int = 1) -> bool:
        with self._lock:
            self._refill()
            if self._tokens >= tokens:
                self._tokens -= tokens
                return True
            return False

    def _refill(self):
        now = time.monotonic()
        elapsed = now - self._last_refill
        self._tokens = min(self.capacity, self._tokens + elapsed * self.rate)
        self._last_refill = now


class RateLimiter:
    # 多维度限流: 全局 + 单模型 + 单用户

    def __init__(self):
        self._buckets: Dict[str, TokenBucket] = {}
        self._lock = threading.Lock()

    def _get_bucket(self, key: str, rate: float, cap: int) -> TokenBucket:
        with self._lock:
            if key not in self._buckets:
                self._buckets[key] = TokenBucket(rate, cap)
            return self._buckets[key]

    def allow(self, model_key: str, user_id: Optional[str] = None,
              global_rpm: int = 5000, model_rpm: int = 500,
              user_rpm: int = 60) -> bool:
        # 全局限流
        gb = self._get_bucket("global", global_rpm / 60.0, global_rpm // 10)
        if not gb.consume(1):
            return False
        # 模型限流
        mb = self._get_bucket(f"model:{model_key}",
                              model_rpm / 60.0, model_rpm // 10)
        if not mb.consume(1):
            return False
        # 用户限流
        if user_id:
            ub = self._get_bucket(f"user:{user_id}",
                                  user_rpm / 60.0, max(user_rpm // 5, 10))
            if not ub.consume(1):
                return False
        return True


# ============================================================
# 4. 熔断器 (Circuit Breaker)
# ============================================================

class CircuitState(Enum):
    CLOSED = "closed"       # 正常
    OPEN = "open"           # 熔断
    HALF_OPEN = "half_open" # 探测


class CircuitBreakerOpenError(Exception):
    pass


class CircuitBreaker:
    # CLOSED -> (失败阈值) -> OPEN -> (冷却) -> HALF_OPEN -> CLOSED

    def __init__(self, name: str, failure_threshold: int = 5,
                 cooldown_seconds: float = 30.0,
                 half_open_max: int = 3):
        self.name = name
        self.failure_threshold = failure_threshold
        self.cooldown_seconds = cooldown_seconds
        self.half_open_max = half_open_max
        self.state = CircuitState.CLOSED
        self._failure_count = 0
        self._last_failure = 0.0
        self._half_open_count = 0
        self._lock = threading.Lock()

    def __enter__(self):
        if not self.allow_request():
            raise CircuitBreakerOpenError(
                f"Circuit '{self.name}' is OPEN"
            )
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            self.on_success()
        else:
            self.on_failure()
        return False  # 不吞异常

    def allow_request(self) -> bool:
        with self._lock:
            now = time.monotonic()
            if self.state == CircuitState.CLOSED:
                return True
            if self.state == CircuitState.OPEN:
                if now - self._last_failure >= self.cooldown_seconds:
                    self.state = CircuitState.HALF_OPEN
                    self._half_open_count = 0
                    return True
                return False
            if self.state == CircuitState.HALF_OPEN:
                if self._half_open_count < self.half_open_max:
                    self._half_open_count += 1
                    return True
                return False
            return True

    def on_success(self):
        with self._lock:
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.CLOSED
                self._failure_count = 0

    def on_failure(self):
        with self._lock:
            self._failure_count += 1
            self._last_failure = time.monotonic()
            if (self.state == CircuitState.CLOSED
                    and self._failure_count >= self.failure_threshold):
                self.state = CircuitState.OPEN
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.OPEN


# ============================================================
# 5. LLM Gateway 主类
# ============================================================

@dataclass
class GatewayResponse:
    model: str
    content: str
    usage: Dict[str, int]
    latency_ms: float
    cost_usd: float
    retries: int
    provider: str


class LLMGateway:
    # LLM API 网关 - 统一入口

    def __init__(self):
        self.registry = ModelRegistry()
        self.rate_limiter = RateLimiter()
        self._cbs: Dict[str, CircuitBreaker] = {}
        self._lock = threading.Lock()
        self.total_cost = 0.0
        self.total_requests = 0
        self.total_tokens = 0

    def register_model(self, provider: str, model_id: str,
                       api_key: str, input_price: float = 0.0,
                       output_price: float = 0.0,
                       capabilities: List[ModelCapability] = None,
                       base_url: Optional[str] = None, **kwargs):
        entry = ModelEntry(
            provider=provider, model_id=model_id,
            capabilities=capabilities or [ModelCapability.CHAT],
            input_price=input_price, output_price=output_price, **kwargs,
        )
        return self.registry.register(entry, api_key, base_url)

    def _get_cb(self, key: str) -> CircuitBreaker:
        with self._lock:
            if key not in self._cbs:
                self._cbs[key] = CircuitBreaker(
                    name=key, failure_threshold=5, cooldown_seconds=30.0,
                )
            return self._cbs[key]

    def chat(self, messages: List[Dict[str, str]],
             model: Optional[str] = None,
             strategy: RoutingStrategy = RoutingStrategy.COST_OPTIMIZED,
             user_id: Optional[str] = None,
             temperature: float = 0.7, max_tokens: int = 4096,
             max_retries: int = 3, timeout: float = 60.0,
             **kwargs) -> GatewayResponse:
        # 核心方法: 路由 -> 限流 -> 熔断 -> 调用 -> 记录

        router = Router(self.registry)
        if model:
            entry = self.registry.get(model)
            if not entry:
                raise ValueError(f"Model '{model}' not registered")
            candidates = [entry]
        else:
            candidates = router.fallback_chain()

        failed = set()
        last_error = None

        for entry in candidates:
            mk = f"{entry.provider}:{entry.model_id}"
            if mk in failed:
                continue

            if not self.rate_limiter.allow(mk, user_id, model_rpm=entry.rpm_limit):
                last_error = Exception(f"Rate limit: {mk}")
                failed.add(mk)
                continue

            cb = self._get_cb(mk)
            retries = 0

            for attempt in range(max_retries):
                try:
                    with cb:
                        result = self._call(entry, messages,
                                           temperature, max_tokens, timeout, **kwargs)
                    self.total_requests += 1
                    self.total_cost += result.cost_usd
                    self.total_tokens += result.usage.get("total_tokens", 0)
                    result.retries = retries
                    return result
                except CircuitBreakerOpenError:
                    failed.add(mk)
                    break
                except Exception as e:
                    last_error = e
                    retries = attempt
                    if attempt < max_retries - 1:
                        time.sleep(min(0.5 * (2 ** attempt), 10.0))
                    else:
                        failed.add(mk)

        raise RuntimeError(
            f"All models exhausted. Failed: {failed}. "
            f"Last error: {last_error}"
        )

    def _call(self, entry: ModelEntry, messages: List[Dict],
              temperature: float, max_tokens: int,
              timeout: float, **kwargs) -> GatewayResponse:
        start = time.time()
        ck = {"api_key": entry.api_key}
        if entry._base_url:
            ck["base_url"] = entry._base_url

        client = OpenAI(**ck)
        resp = client.chat.completions.create(
            model=entry.model_id, messages=messages,
            temperature=temperature, max_tokens=max_tokens,
            timeout=timeout, **kwargs,
        )
        latency = (time.time() - start) * 1000

        usage = {
            "prompt_tokens": resp.usage.prompt_tokens if resp.usage else 0,
            "completion_tokens": resp.usage.completion_tokens if resp.usage else 0,
            "total_tokens": resp.usage.total_tokens if resp.usage else 0,
        }
        cost = (
            usage["prompt_tokens"] / 1_000_000 * entry.input_price
            + usage["completion_tokens"] / 1_000_000 * entry.output_price
        )

        return GatewayResponse(
            model=entry.model_id,
            content=resp.choices[0].message.content,
            usage=usage, latency_ms=latency,
            cost_usd=round(cost, 6), retries=0,
            provider=entry.provider,
        )

    def stats(self) -> dict:
        return {
            "total_requests": self.total_requests,
            "total_cost_usd": round(self.total_cost, 4),
            "total_tokens": self.total_tokens,
            "avg_cost": round(
                self.total_cost / max(self.total_requests, 1), 6
            ),
            "models": {
                k: {
                    "state": self._get_cb(k).state.value,
                    "enabled": self.registry.get(k).enabled,
                }
                for k in self.registry._models
            },
        }


# ============================================================
# 6. 使用示例
# ============================================================

if __name__ == "__main__":
    gw = LLMGateway()

    gw.register_model(
        "openai", "gpt-4o-mini",
        api_key="sk-your-key",
        input_price=0.15, output_price=0.60,
        capabilities=[ModelCapability.CHAT, ModelCapability.FUNCTION_CALL],
        priority=1, rpm_limit=500,
    )
    gw.register_model(
        "deepseek", "deepseek-chat",
        api_key="sk-your-key",
        base_url="https://api.deepseek.com/v1",
        input_price=0.14, output_price=0.28,
        capabilities=[ModelCapability.CHAT],
        priority=2, rpm_limit=100,
    )

    print("=" * 60)
    print("LLM Gateway 模型列表:")
    for e in gw.registry.list_enabled():
        print(f"  {e.provider}:{e.model_id} "
              f"cost=(${e.input_price}, ${e.output_price})/1M")

    # 限流演示
    print("\n限流演示 (model_rpm=3):")
    rl = RateLimiter()
    for i in range(10):
        ok = rl.allow("openai:gpt-4o-mini", "user_1",
                      global_rpm=5000, model_rpm=3, user_rpm=60)
        status = "ALLOWED" if ok else "REJECTED"
        print(f"  Request {i+1}: {status}")

    print("\nLLM Gateway 框架就绪!")

四、路由策略对比

五、生产化建议

1. 配置外置 — 模型信息放 gateway.yaml，不要硬编码
2. 可观测性 — 接入 Prometheus + Grafana，监控 QPS/延迟/成本
3. 灰度切换 — 新模型先用 5% 流量，验证后全量
4. 成本告警 — 日成本超预算自动告警（飞书/钉钉/企微）
5. 密钥管理 — 对接 Vault / 云 KMS
6. 流式透传 — 网关需透传 SSE stream（下篇讲）

六、总结

网关四层能力：

1. 路由层 — 多模型注册 + 策略选择，业务无感知
2. 限流层 — 全局限流 + 模型限流 + 用户限流，保护配额
3. 容错层 — 熔断 + 退避重试 + fallback 链，高可用
4. 成本层 — 全局成本追踪，自动选便宜模型