1. 为什么需要 LLM 网关

当企业接入多个 LLM 供应商（OpenAI、Anthropic、Google、本地模型等），直接调用会面临：

  • 供应商锁定：切换模型需要修改所有调用方代码
  • 成本失控：无法统一监控和控制 token 消耗
  • 可靠性差：单供应商宕机即服务不可用
  • 安全风险：API Key 分散在各处，难以统一管理

LLM 网关作为统一接入层，解决以上所有问题。

2. 整体架构

┌─────────────────────────────────────────────────┐
│                  LLM Gateway                     │
│                                                  │
│  ┌─────────┐  ┌──────────┐  ┌───────────────┐ │
│  │  Router  │  │   Load   │  │    Rate       │ │
│  │  Engine  │→ │ Balancer │→ │   Limiter     │ │
│  └─────────┘  └──────────┘  └───────────────┘ │
│       │                                       │
│  ┌────┴────────────────────────────────────┐  │
│  │          Model Adapter Pool             │  │
│  │  ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐  │  │
│  │  │OpenAI│ │Claude│ │Gemini│ │Local │  │  │
│  │  └──────┘ └──────┘ └──────┘ └──────┘  │  │
│  └─────────────────────────────────────────┘  │
│                                                  │
│  ┌──────────┐  ┌──────────┐  ┌────────────┐  │
│  │  Cache   │  │  Metrics │  │   Audit    │  │
│  │  Layer   │  │  Collector│  │   Logger   │  │
│  └──────────┘  └──────────┘  └────────────┘  │
└─────────────────────────────────────────────────┘

3. 多模型路由引擎

3.1 路由策略

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Optional
import hashlib
import time

@dataclass
class LLMRequest:
    model: str
    messages: list[dict]
    temperature: float = 0.7
    max_tokens: int = 2048
    stream: bool = False
    metadata: dict = field(default_factory=dict)

@dataclass
class ModelEndpoint:
    provider: str        # openai, anthropic, local
    model_id: str        # gpt-4o, claude-3.5-sonnet
    endpoint_url: str
    api_key: str
    max_concurrent: int = 10
    cost_per_1k_input: float = 0.0
    cost_per_1k_output: float = 0.0
    avg_latency_ms: float = 500
    availability: float = 99.9
    current_load: int = 0

class RoutingStrategy(ABC):
    @abstractmethod
    def select(self, request: LLMRequest, candidates: list[ModelEndpoint]) -> ModelEndpoint:
        pass

class CostOptimizedRouting(RoutingStrategy):
    """最低成本优先路由"""
    def select(self, request: LLMRequest, candidates: list[ModelEndpoint]) -> ModelEndpoint:
        available = [c for c in candidates if c.current_load < c.max_concurrent]
        if not available:
            raise RuntimeError("No available endpoints")
        return min(available, key=lambda c: c.cost_per_1k_input)

class LatencyOptimizedRouting(RoutingStrategy):
    """最低延迟优先路由"""
    def select(self, request: LLMRequest, candidates: list[ModelEndpoint]) -> ModelEndpoint:
        available = [c for c in candidates if c.current_load < c.max_concurrent]
        if not available:
            raise RuntimeError("No available endpoints")
        return min(available, key=lambda c: c.avg_latency_ms)

class WeightedRoundRobinRouting(RoutingStrategy):
    """加权轮询路由"""
    def __init__(self):
        self._index = 0
        self._weights: dict[str, int] = {}

    def set_weight(self, model_id: str, weight: int):
        self._weights[model_id] = weight

    def select(self, request: LLMRequest, candidates: list[ModelEndpoint]) -> ModelEndpoint:
        available = [c for c in candidates if c.current_load < c.max_concurrent]
        if not available:
            raise RuntimeError("No available endpoints")
        # 构建加权列表
        weighted = []
        for c in available:
            w = self._weights.get(c.model_id, 1)
            weighted.extend([c] * w)
        selected = weighted[self._index % len(weighted)]
        self._index += 1
        return selected

class CapabilityBasedRouting(RoutingStrategy):
    """基于任务能力的智能路由"""
    CAPABILITY_MAP = {
        "code_generation": ["gpt-4o", "claude-3.5-sonnet", "deepseek-coder"],
        "long_context": ["gemini-1.5-pro", "claude-3.5-sonnet"],
        "vision": ["gpt-4o", "gemini-1.5-pro"],
        "low_cost": ["gpt-4o-mini", "claude-3-haiku", "gemini-1.5-flash"],
    }

    def select(self, request: LLMRequest, candidates: list[ModelEndpoint]) -> ModelEndpoint:
        required_cap = request.metadata.get("capability")
        if not required_cap:
            return candidates[0]
        preferred_models = self.CAPABILITY_MAP.get(required_cap, [])
        for model_id in preferred_models:
            for c in candidates:
                if c.model_id == model_id and c.current_load < c.max_concurrent:
                    return c
        # 降级：选任意可用的
        return next((c for c in candidates if c.current_load < c.max_concurrent), candidates[0])

3.2 路由策略对比

策略优势劣势适用场景
成本优先最低开销可能高延迟批量处理
延迟优先响应快成本高实时对话
加权轮询负载均衡不考虑成本通用场景
能力路由任务匹配最优需维护映射混合任务

4. 限流与降级

4.1 多维度限流

from collections import defaultdict, deque
import time

class MultiDimensionRateLimiter:
    def __init__(self):
        # 按用户限流
        self.user_limits: dict[str, tuple[int, float]] = {}  # user -> (max_rpm, window_s)
        self.user_counters: dict[str, deque] = defaultdict(deque)
        # 按模型限流
        self.model_limits: dict[str, tuple[int, float]] = {}
        self.model_counters: dict[str, deque] = defaultdict(deque)
        # 全局限流
        self.global_limit = (1000, 60)  # 1000 rpm
        self.global_counter: deque = deque()

    def check(self, user_id: str, model: str) -> bool:
        now = time.time()
        if not self._check_counter(self.global_counter, self.global_limit, now):
            return False
        if user_id in self.user_limits:
            if not self._check_counter(self.user_counters[user_id], self.user_limits[user_id], now):
                return False
        if model in self.model_limits:
            if not self._check_counter(self.model_counters[model], self.model_limits[model], now):
                return False
        return True

    def _check_counter(self, counter: deque, limit: tuple, now: float) -> bool:
        max_count, window = limit
        while counter and counter[0] < now - window:
            counter.popleft()
        if len(counter) >= max_count:
            return False
        counter.append(now)
        return True

4.2 熔断降级链

class FallbackChain:
    """模型降级链：主模型失败时自动切换到备选模型"""
    def __init__(self):
        self.chains: dict[str, list[str]] = {
            "gpt-4o": ["claude-3.5-sonnet", "gemini-1.5-pro", "gpt-4o-mini"],
            "claude-3.5-sonnet": ["gpt-4o", "gemini-1.5-pro"],
        }

    async def execute_with_fallback(self, gateway, request: LLMRequest) -> dict:
        primary = request.model
        chain = [primary] + self.chains.get(primary, [])

        last_error = None
        for model_id in chain:
            try:
                request.model = model_id
                result = await gateway.forward(request)
                return result
            except Exception as e:
                last_error = e
                continue

        raise last_error

5. 统一适配器层

class ModelAdapter(ABC):
    @abstractmethod
    async def chat(self, request: LLMRequest) -> dict: pass
    @abstractmethod
    async def stream_chat(self, request: LLMRequest): pass
    @abstractmethod
    def count_tokens(self, messages: list[dict]) -> int: pass

class OpenAIAdapter(ModelAdapter):
    def __init__(self, endpoint: ModelEndpoint):
        self.endpoint = endpoint
        self.client = httpx.AsyncClient(
            base_url=endpoint.endpoint_url,
            headers={"Authorization": f"Bearer {endpoint.api_key}"},
            timeout=120
        )

    async def chat(self, request: LLMRequest) -> dict:
        resp = await self.client.post("/v1/chat/completions", json={
            "model": request.model,
            "messages": request.messages,
            "temperature": request.temperature,
            "max_tokens": request.max_tokens,
        })
        resp.raise_for_status()
        data = resp.json()
        return {
            "content": data["choices"][0]["message"]["content"],
            "model": data["model"],
            "usage": data["usage"],
        }

    async def stream_chat(self, request: LLMRequest):
        async with self.client.stream("POST", "/v1/chat/completions", json={
            "model": request.model, "messages": request.messages,
            "stream": True,
        }) as resp:
            async for line in resp.aiter_lines():
                if line.startswith("data: "):
                    yield line[6:]

    def count_tokens(self, messages: list[dict]) -> int:
        # tiktoken 计算
        import tiktoken
        enc = tiktoken.encoding_for_model("gpt-4o")
        total = sum(len(enc.encode(m["content"])) for m in messages)
        return total

class AnthropicAdapter(ModelAdapter):
    async def chat(self, request: LLMRequest) -> dict:
        # 适配 Anthropic API 格式
        system = next((m["content"] for m in request.messages if m["role"] == "system"), "")
        user_messages = [m for m in request.messages if m["role"] != "system"]
        resp = await self.client.post("/v1/messages", json={
            "model": request.model,
            "system": system,
            "messages": user_messages,
            "max_tokens": request.max_tokens,
        })
        # 转换为统一格式
        ...

class LocalModelAdapter(ModelAdapter):
    """本地 vLLM/Ollama 适配器"""
    async def chat(self, request: LLMRequest) -> dict:
        resp = await self.client.post("/v1/chat/completions", json={
            "model": request.model,
            "messages": request.messages,
            "temperature": request.temperature,
        })
        ...

6. 成本控制与计量

@dataclass
class UsageRecord:
    user_id: str
    model: str
    input_tokens: int
    output_tokens: int
    cost: float
    timestamp: float
    request_id: str

class CostTracker:
    def __init__(self, redis_client):
        self.redis = redis_client

    async def record(self, record: UsageRecord):
        pipe = self.redis.pipeline()
        # 按用户累计
        pipe.hincrby(f"cost:user:{record.user_id}:daily:{date.today()}", "total", int(record.cost * 10000))
        pipe.hincrby(f"cost:user:{record.user_id}:monthly:{date.today().strftime('%Y-%m')}", "total", int(record.cost * 10000))
        # 按模型累计
        pipe.hincrby(f"cost:model:{record.model}:daily:{date.today()}", "total", int(record.cost * 10000))
        # 预算检查
        pipe.hget(f"budget:user:{record.user_id}:monthly", "limit")
        results = await pipe.execute()
        budget_limit = results[-1]
        if budget_limit and results[1] > int(budget_limit):
            raise BudgetExceededError(f"User {record.user_id} exceeded monthly budget")

    async def get_usage_report(self, user_id: str, period: str = "daily") -> dict:
        key = f"cost:user:{user_id}:{period}:{date.today()}"
        data = await self.redis.hgetall(key)
        return {"user": user_id, "period": period, "cost_cents": int(data.get("total", 0)) / 10000}

7. 可观测性

class GatewayMetrics:
    def __init__(self):
        self.request_count = Counter("gateway_requests_total", ["model", "status"])
        self.latency = Histogram("gateway_latency_seconds", ["model"])
        self.token_usage = Counter("gateway_tokens_total", ["model", "direction"])
        self.cost = Counter("gateway_cost_total", ["model"])
        self.active_connections = Gauge("gateway_active_connections")

    def record_request(self, model: str, status: str, duration: float,
                       input_tokens: int, output_tokens: int, cost: float):
        self.request_count.labels(model=model, status=status).inc()
        self.latency.labels(model=model).observe(duration)
        self.token_usage.labels(model=model, direction="input").inc(input_tokens)
        self.token_usage.labels(model=model, direction="output").inc(output_tokens)
        self.cost.labels(model=model).inc(cost)

8. 部署架构

                    ┌─────────┐
                    │   CDN   │
                    └────┬────┘
                         │
              ┌──────────┼──────────┐
              │          │          │
         ┌────┴───┐ ┌───┴────┐ ┌──┴─────┐
         │GW #1   │ │GW #2   │ │GW #3   │
         │(active)│ │(active)│ │(active)│
         └────┬───┘ └───┬────┘ └──┬─────┘
              │         │         │
         ┌────┴─────────┴─────────┴────┐
         │       Redis Cluster          │
         │  (rate limits, cache, cost)  │
         └──────────────────────────────┘
              │              │
         ┌────┴────┐   ┌────┴────┐
         │ Model A │   │ Model B │
         │ Provider│   │ Provider│
         └─────────┘   └─────────┘

9. 总结

LLM 网关是企业级 AI 应用的必备基础设施。核心设计要点：

  1. 统一接口：适配器模式屏蔽供应商差异
  2. 智能路由：根据成本、延迟、能力多维度决策
  3. 弹性容错：限流 + 熔断 + 降级链保障可用性
  4. 成本治理：实时计量 + 预算控制防失控
  5. 可观测：全链路指标追踪每一次调用

建议技术选型：基于 Python FastAPI / Go 实现网关核心，Redis 做限流和缓存，Prometheus + Grafana 做监控，从开源项目 LiteLLM、OneAPI 中汲取经验，但根据企业需求自建以获得完整控制力。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。