API

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

引言 2023 年 OpenAI 的 Chat Completions API 成为事实标准后，整个 AI 生态出现了一个有趣的现象：几乎所有的 LLM 推理引擎、本地模型运行时和云服务提供商都实现了「OpenAI 兼容 API」。这种兼容性让开发者可以仅修改 base_url 就切换底层模型，极大降低了供应商锁定风险。本文将深入解析这一生态的现状、标准规范、主流实现方案和最佳实践。 OpenAI API 标准解析 核心端点 OpenAI 兼容 API 的核心端点包括： 端点 路径 功能 Chat Completions /v1/chat/completions 对话补全（最核心） Completions /v1/completions 文本补全（旧版） Embeddings /v1/embeddings 文本向量化 Models /v1/models 模型列表 Files /v1/files 文件管理 Fine-tuning /v1/fine_tuning/jobs 微调任务 Chat Completions 规范 { "model": "gpt-4o", "messages": [ {"role": "system", "content": "你是一个助手。"}, {"role": "user", "content": "你好"} ], "temperature": 0.7, "top_p": 0.9, "max_tokens": 1024, "stream": false, "stop": ["\n\n"], "response_format": {"type": "json_object"}, "tools": [...], "tool_choice": "auto", "seed": 42, "n": 1, "logprobs": false } 兼容性分级 不同实现对「兼容」的定义各不相同，可以按兼容程度分为四级： ...

为什么需要 LLM API 网关 当你的团队同时使用 OpenAI、Anthropic、本地模型等多个 LLM 提供商时，直接在业务代码中调用各家 API 会带来一系列问题：密钥散落各处、无法统一限流、缺乏调用审计、模型切换成本高。LLM API 网关就是解决这些问题的中间层。 ┌──────────────────────────────────────────────────────┐ │ 业务服务层 │ │ (Chat UI / RAG / Agent / Code Gen) │ └──────────────────┬───────────────────────────────────┘ │ 统一 API: POST /v1/chat/completions ┌──────────────────▼───────────────────────────────────┐ │ LLM API Gateway │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌────────────┐ │ │ │ 路由引擎 │ │ 限流器 │ │ 缓存层 │ │ 审计日志 │ │ │ └─────────┘ └─────────┘ └─────────┘ └────────────┘ │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ 密钥管理 │ │ 重试器 │ │ 负载均衡│ │ │ └─────────┘ └─────────┘ └─────────┘ │ └──────┬──────────┬──────────┬──────────┬─────────────┘ │ │ │ │ ┌────▼────┐┌───▼────┐┌───▼─────┐┌───▼──────┐ │ OpenAI ││Claude ││ Llama ││ Qwen │ │ API ││ API ││ Local ││ API │ └─────────┘└────────┘└─────────┘└──────────┘ 多模型路由 路由引擎是网关的核心。它根据任务复杂度、成本预算、延迟要求将请求分发到最合适的模型。 ...

为什么需要 OpenAI 兼容 OpenAI 的 Chat Completions API 已成为 LLM 推理的事实标准。兼容这个接口意味着： ┌─────────────────────────────────────────────┐ │ 任意 OpenAI SDK 客户端 │ │ (Python / TypeScript / Go / Rust ...) │ ├─────────────────────────────────────────────┤ │ OpenAI 兼容 API 层 │ │ /v1/chat/completions │ │ /v1/completions │ │ /v1/embeddings │ │ /v1/models │ ├─────────────┬───────────┬───────────────────┤ │ vLLM │ TGI │ Ollama │ │ │ │ LM Studio │ └─────────────┴───────────┴───────────────────┘ 核心价值： 零代码迁移：只需改 base_url，现有应用立即切换到本地模型 生态复用：LangChain、LlamaIndex、AutoGen 等框架原生支持 可替换性：推理引擎可热切换，应用层无需修改 from openai import OpenAI # 同一份代码，切换后端只需改一行 # OpenAI: client = OpenAI(base_url="https://api.openai.com/v1", api_key="sk-xxx") # vLLM: client = OpenAI(base_url="http://vllm:8000/v1", api_key="none") # Ollama: client = OpenAI(base_url="http://ollama:11434/v1", api_key="ollama") # TGI: client = OpenAI(base_url="http://tgi:8080/v1", api_key="none") response = client.chat.completions.create( model="meta-llama/Llama-3.1-8B-Instruct", messages=[{"role": "user", "content": "Hello"}], stream=True ) API 规范差异 基础端点覆盖 端点 vLLM TGI Ollama LM Studio /v1/chat/completions ✅ ✅ ✅ ✅ /v1/completions ✅ ✅ ✅ ✅ /v1/embeddings ✅ ✅ ✅ ✅ /v1/models ✅ ✅ ✅ ✅ /v1/files ✅ ❌ ❌ ❌ /v1/audio/transcriptions ✅ ❌ ❌ ✅ /v1/images/generations ✅ ❌ ❌ ❌ 请求参数支持 参数 vLLM TGI Ollama LM Studio stream ✅ ✅ ✅ ✅ temperature ✅ ✅ ✅ ✅ top_p ✅ ✅ ✅ ✅ top_k ✅ ✅ ✅ ✅ max_tokens ✅ ✅ ✅ ✅ stop ✅ ✅ ✅ ✅ n (多选) ✅ ⚠️ 限1 ❌ ✅ logprobs ✅ ✅ ⚠️ ✅ presence_penalty ✅ ✅ ✅ ✅ frequency_penalty ✅ ✅ ✅ ✅ seed ✅ ✅ ✅ ✅ response_format (JSON) ✅ ✅ ⚠️ ✅ Function Calling 支持 response = client.chat.completions.create( model="llama3.1:8b", messages=[{"role": "user", "content": "What's the weather in Tokyo?"}], tools=[{ "type": "function", "function": { "name": "get_weather", "description": "Get current weather", "parameters": { "type": "object", "properties": { "city": {"type": "string"} }, "required": ["city"] } } }], tool_choice="auto" ) 特性 vLLM TGI Ollama LM Studio 基础 Function Calling ✅ ✅ ✅ ✅ Parallel Function Calling ✅ ⚠️ ✅ ✅ tool_choice: "required" ✅ ⚠️ ✅ ⚠️ 指定函数调用 ✅ ⚠️ ✅ ⚠️ JSON Schema 结构化输出 ✅ ✅ ⚠️ ✅ 流式输出差异 所有引擎均支持 SSE 流式输出，但存在细微差异： ...