Skip to content
🔗 分享本题
查看我的学习进度 →

LLM Gateway 统一认证配额路由缓存观测和安全能力并连接多模型提供方图

🧠 记忆锚点: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 GatewayLLM Gateway
限流粒度按请求数(RPM)按 Token 数(TPM)
模型路由不支持原生支持多模型
语义缓存不支持基于 Embedding 相似度
成本分析按用户/模型/请求粒度
模型降级不支持自动 fallback 到小模型

面试话术:

"LLM Gateway 的核心价值是'统一入口+智能路由'。我的设计:简单 QA 走 DeepSeek(成本 1/50),复杂推理走 GPT-4o,长上下文走 Claude。语义缓存命中 30% 请求,A/B 测试持续优化模型选择。生产环境平均单次请求成本从 $0.04 降到 $0.012。面试时能画出完整的 Gateway 架构图并讲清楚各层职责,说明你有工程落地经验。"

📚 参考:LiteLLM Proxy(LLM 网关:路由/限流/计费)