LLM API 网关实现：多模型路由/限流/缓存/审计

为什么需要 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     │
  └─────────┘└────────┘└─────────┘└──────────┘

多模型路由

路由引擎是网关的核心。它根据任务复杂度、成本预算、延迟要求将请求分发到最合适的模型。

路由策略对比

策略	实现复杂度	成本节省	适用场景
静态路由	低	中	固定场景，如翻译始终用同一模型
基于规则路由	中	高	按 token 数/任务类型分流
嵌入分类路由	高	最高	智能识别复杂度自动分流
级联路由	中	高	先小模型，失败再大模型

基于规则的路由实现

from dataclasses import dataclass
from enum import Enum

class ModelTier(Enum):
    SMALL = "gpt-4o-mini"      # 简单任务
    MEDIUM = "claude-sonnet"   # 中等任务
    LARGE = "gpt-4o"           # 复杂任务

@dataclass
class RouteRule:
    max_tokens: int           # 输入 token 上限
    task_type: str            # chat/summary/code/analyze
    required_tools: bool      # 是否需要 function calling
    tier: ModelTier

class ModelRouter:
    def __init__(self):
        self.rules = [
            RouteRule(max_tokens=500, task_type="chat",
                      required_tools=False, tier=ModelTier.SMALL),
            RouteRule(max_tokens=2000, task_type="summary",
                      required_tools=False, tier=ModelTier.SMALL),
            RouteRule(max_tokens=4000, task_type="code",
                      required_tools=True, tier=ModelTier.MEDIUM),
            RouteRule(max_tokens=8000, task_type="analyze",
                      required_tools=True, tier=ModelTier.LARGE),
        ]

    def route(self, request) -> str:
        # 1. 按 token 数过滤候选
        candidates = [r for r in self.rules
                      if r.max_tokens >= request.input_tokens]
        # 2. 按任务类型匹配
        for rule in candidates:
            if rule.task_type == request.task_type:
                if rule.required_tools and not request.use_tools:
                    continue
                return rule.tier.value
        # 3. 兜底用大模型
        return ModelTier.LARGE.value

级联路由（Cascading）

级联路由先用小模型尝试，失败或置信度低时自动升级到大模型：

class CascadingRouter:
    def __init__(self):
        self.chain = ["gpt-4o-mini", "claude-sonnet", "gpt-4o"]

    async def complete(self, messages, threshold=0.7):
        for model in self.chain:
            result = await self.call_model(model, messages)
            # 用另一个小模型快速评估回答质量
            score = await self.evaluate(result, messages)
            if score >= threshold:
                return result, model
            # 日志记录降级事件
            logger.info(f"Cascade fallback: {model} score={score}")
        return result, self.chain[-1]  # 兜底返回最后一个模型结果

速率限制

LLM API 的限流与传统 Web API 不同——不仅要限制请求数，还要限制 token 吞吐量。

Token Bucket 实现

import time
import asyncio
from collections import defaultdict

class TokenBucketRateLimiter:
    def __init__(self, rpm=60, tpm=90000):
        self.rpm = rpm             # 每分钟请求数
        self.tpm = tpm             # 每分钟 token 数
        self.request_buckets = defaultdict(list)
        self.token_buckets = defaultdict(list)

    async def acquire(self, user_id: str, estimated_tokens: int):
        now = time.time()
        window = 60  # 60 秒窗口

        # 清理过期记录
        self.request_buckets[user_id] = [
            t for t in self.request_buckets[user_id] if now - t < window
        ]
        self.token_buckets[user_id] = [
            (t, n) for t, n in self.token_buckets[user_id] if now - t < window
        ]

        # 检查 RPM
        if len(self.request_buckets[user_id]) >= self.rpm:
            wait = window - (now - self.request_buckets[user_id][0])
            raise RateLimitError(f"RPM exceeded, retry in {wait:.0f}s")

        # 检查 TPM
        current_tpm = sum(n for _, n in self.token_buckets[user_id])
        if current_tpm + estimated_tokens > self.tpm:
            raise RateLimitError(f"TPM exceeded: {current_tpm}/{self.tpm}")

        # 记录
        self.request_buckets[user_id].append(now)
        self.token_buckets[user_id].append((now, estimated_tokens))

分层限流策略

层级	维度	典型限制	说明
全局	总 TPM	500K/min	保护整体预算
租户	组织级	100K/min	多团队共享
用户	个人级	10K/min	防单用户滥用
模型	每模型	按配额	小模型宽松，大模型严格

语义缓存

传统缓存用 URL+Body 做缓存键，但 LLM 请求的 prompt 几乎不会完全相同。语义缓存通过嵌入相似度匹配实现"意思相近就命中"。

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class SemanticCache:
    def __init__(self, embed_model, threshold=0.95, ttl=3600):
        self.embed_model = embed_model
        self.threshold = threshold
        self.ttl = ttl
        self.entries = []  # 生产环境用 Vector DB

    async def get(self, messages):
        query = self._flatten(messages)
        query_emb = await self.embed_model.embed(query)

        for entry in self.entries:
            sim = cosine_similarity([query_emb], [entry["embedding"]])[0][0]
            if sim >= self.threshold:
                # 检查 TTL
                if time.time() - entry["timestamp"] < self.ttl:
                    entry["hit_count"] += 1
                    return entry["response"], sim

        return None, 0.0

    async def set(self, messages, response):
        query = self._flatten(messages)
        emb = await self.embed_model.embed(query)
        self.entries.append({
            "query": query,
            "embedding": emb,
            "response": response,
            "timestamp": time.time(),
            "hit_count": 0,
        })

注意：语义缓存要谨慎处理包含 PII 或时效性内容（如"今天天气"）的请求。建议对此类请求跳过缓存。

审计日志

审计日志不仅是合规要求，也是成本分析和质量改进的数据来源。

import json
from datetime import datetime

class AuditLogger:
    def __init__(self, sink):  # sink: elasticsearch / s3 / file
        self.sink = sink

    async def log(self, request_id, user_id, model, 
                  input_tokens, output_tokens, 
                  latency_ms, status, cost):
        entry = {
            "request_id": request_id,
            "timestamp": datetime.utcnow().isoformat(),
            "user_id": user_id,
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "latency_ms": latency_ms,
            "status": status,       # success/failure/degraded
            "cost_usd": cost,
        }
        await self.sink.write(json.dumps(entry))

    async def log_rejection(self, request_id, reason, detail):
        """记录被限流/拒绝的请求"""
        entry = {
            "request_id": request_id,
            "timestamp": datetime.utcnow().isoformat(),
            "event": "rejected",
            "reason": reason,       # rate_limit / content_policy / quota_exceeded
            "detail": detail,
        }
        await self.sink.write(json.dumps(entry))

网关选型：Kong vs APISIX vs 自研

维度	Kong	APISIX	自研
插件生态	丰富	丰富	完全自定义
LLM 特性	AI Gateway 插件	有限	需自己写
性能	高（C+Lua）	极高（Nginx+Lua）	取决于实现
多模型支持	原生支持	需定制	完全自定义
语义缓存	内置	需插件	需自己实现
学习曲线	中等	中等	高
生产成熟度	高	高	取决于团队

Kong AI Gateway 配置示例

# Kong declarative config
_format_version: "3.0"
services:
  - name: openai-service
    url: https://api.openai.com
    routes:
      - name: openai-chat
        paths: ["/openai/v1/chat"]
        strip_path: true
    plugins:
      - name: ai-proxy
        config:
          model: gpt-4o
          auth:
            header_name: Authorization
            header_value: "Bearer ${OPENAI_KEY}"
      - name: rate-limiting
        config:
          minute: 60
          policy: redis
          redis_host: redis.internal
      - name: ai-proxy-semantic-cache
        config:
          embed_model: text-embedding-3-small
          similarity_threshold: 0.95
          ttl: 3600

  - name: anthropic-service
    url: https://api.anthropic.com
    routes:
      - name: claude-chat
        paths: ["/claude/v1/messages"]
        strip_path: true
    plugins:
      - name: ai-proxy
        config:
          model: claude-sonnet-4-20250514

部署架构建议

生产环境推荐以下部署方式：

Internet → CDN/WAF → Load Balancer → Gateway Pods (3+)
                                              ↓
                                    Redis (限流计数器/缓存)
                                    PostgreSQL (审计日志)
                                    Vector DB (语义缓存)
                                    Vault (密钥管理)

关键要点：

密钥不落盘：API Key 存在 Vault/KMS，网关启动时注入内存
限流用 Redis：多实例部署时共享限流计数器
审计日志异步写：不阻塞主请求路径
健康检查：定期探测各模型 API 可用性，自动摘除故障节点
Graceful Shutdown：收到 SIGTERM 时等待在途请求完成

总结

LLM API 网关是企业 AI 基础设施的关键组件。核心功能包括：多模型路由（成本与能力平衡）、Token 级限流、语义缓存（命中率 20-40% 可显著降本）、全链路审计日志。选型上，快速起步用 Kong AI Gateway，需要深度定制则自研。无论哪种方式，都应将网关作为所有 LLM 调用的唯一出口，实现统一管控。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

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

为什么需要 LLM API 网关#

多模型路由#

路由策略对比#

基于规则的路由实现#

级联路由（Cascading）#

速率限制#

Token Bucket 实现#

分层限流策略#

语义缓存#

审计日志#

网关选型：Kong vs APISIX vs 自研#

Kong AI Gateway 配置示例#

部署架构建议#

总结#

加入讨论#

📌 相关推荐