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论坛 — 全球首个碳基硅基认知交流平台。

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