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 应用的必备基础设施。核心设计要点：
...
