
🧠 记忆锚点:Gateway 承担跨应用通用控制,业务 Prompt、RAG 和工具编排仍留在应用;实验需粘性分流、可追踪、可回退。
💡 答案要点
LLM Gateway 的核心职责:
LLM Gateway 是 AI 应用的统一入口,负责限流、路由、监控、缓存四件事。它位于应用层和各模型 Provider 之间,屏蔽底层复杂度。
多模型路由的实现:
python
class LLMRouter:
def __init__(self):
self.routes = {
"gpt-4o": {"provider": "openai", "cost_per_1k": 0.005, "latency_p50": 1.2},
"claude-3-5-sonnet": {"provider": "anthropic", "cost_per_1k": 0.003, "latency_p50": 1.5},
"deepseek-chat": {"provider": "deepseek", "cost_per_1k": 0.00014, "latency_p50": 2.0},
}
def route(self, request: LLMRequest, context: RoutingContext) -> str:
if request.task_type == "qa" and request.complexity == "low":
return "deepseek-chat"
if request.complexity == "high" or request.needs_reasoning:
return "gpt-4o"
if request.context_length > 128000:
return "claude-3-5-sonnet"
if context.user_tier == "free" and request.complexity == "medium":
return "deepseek-chat"
return "claude-3-5-sonnet"A/B 测试框架:
展开 Python 代码示例(30 行)
python
from collections import defaultdict
class LLMABExperiment:
def __init__(self, experiment_id: str):
self.experiment_id = experiment_id
self.variants = {
"control": {"model": "gpt-4o", "weight": 0.5},
"treatment": {"model": "claude-3-5-sonnet", "weight": 0.5},
}
self.metrics = defaultdict(lambda: {"requests": 0, "latencies": [], "errors": 0})
def get_variant(self, user_id: str) -> str:
bucket = hash(user_id) % 100
cumulative = 0
for variant, config in self.variants.items():
cumulative += config["weight"] * 100
if bucket < cumulative:
return variant
return "control"
def record(self, user_id: str, variant: str, latency: float, success: bool, quality_score: float = None):
m = self.metrics[variant]
m["requests"] += 1
m["latencies"].append(latency)
if not success:
m["errors"] += 1
if quality_score is not None:
if "quality_scores" not in m:
m["quality_scores"] = []
m["quality_scores"].append(quality_score)LLM Gateway 完整架构:
python
class LLMGateway:
def __init__(self):
self.rate_limiter = TokenRateLimiter(tpm_limit=400000)
self.router = LLMRouter()
self.cache = SemanticCache()
async def handle(self, request: LLMRequest) -> LLMResponse:
tokens = estimate_tokens(request.prompt)
if not await self.rate_limiter.acquire(tokens):
raise RateLimitException("Too many requests")
cached = self.cache.get(request.prompt)
if cached:
return cached
model = self.router.route(request, self.get_context(request))
response = await self.call_model(model, request)
self.cache.set(request.prompt, response, ttl=3600)
return response与 Nginx/Kong 等 API Gateway 的区别:
| 维度 | 传统 API Gateway | LLM Gateway |
|---|---|---|
| 限流粒度 | 按请求数(RPM) | 按 Token 数(TPM) |
| 模型路由 | 不支持 | 原生支持多模型 |
| 语义缓存 | 不支持 | 基于 Embedding 相似度 |
| 成本分析 | 无 | 按用户/模型/请求粒度 |
| 模型降级 | 不支持 | 自动 fallback 到小模型 |
面试话术:
"LLM Gateway 的核心价值是'统一入口+智能路由'。我的设计:简单 QA 走 DeepSeek(成本 1/50),复杂推理走 GPT-4o,长上下文走 Claude。语义缓存命中 30% 请求,A/B 测试持续优化模型选择。生产环境平均单次请求成本从 $0.04 降到 $0.012。面试时能画出完整的 Gateway 架构图并讲清楚各层职责,说明你有工程落地经验。"