AI 网关实战：限流、路由、缓存是怎么保护你的 LLM 服务的

从一次生产事故说起

去年双十一前两周，某电商团队的 AI 客服突然全线崩溃。不是代码 bug，是大模型 API 被限流了——上游供应商的 QPS 限制是 500，他们的系统峰值到了 2000。200 万个客服请求被拒绝，用户收到的回复是"抱歉，我暂时无法回答"。

事后复盘，问题不在大模型，而在没有人在 LLM 前面放一个网关。

这篇文章不聊 AI 网关的概念，直接拆解一个生产级的 AI 网关长什么样——限流怎么配、路由怎么设计、缓存怎么命中，以及每一步踩过的坑。

AI 网关到底干什么

简单说，AI 网关就是 LLM 前面的那一层代理。所有请求先经过它，再由它转发给大模型。听起来和 API 网关一样，但 AI 场景有几个特殊需求：

Token 级别的限流——不是按请求数，是按输入+输出的 token 数
多模型路由——同一个请求，根据成本、延迟、可用性选不同的后端模型
语义缓存——相似的问题直接返回缓存结果，不调用模型
流式响应处理——SSE 流的分发、重试、降级
成本追踪——每个请求花了多少 token、多少钱

我们一个个拆。

限流：Token 级别的精准控制

大多数 API 网关的限流是按请求数来的，比如每分钟 1000 次请求。但 LLM 不一样——一个 1000 token 的请求和一个 50000 token 的请求，对后端的压力差了几十倍。所以 AI 网关的限流必须是按 token 计数的。

// 基于 token 的滑动窗口限流器
type TokenRateLimiter struct {
    mu           sync.Mutex
    window       time.Duration  // 滑动窗口大小
    maxTokens    int            // 窗口内最大 token 数
    buckets      map[string]*TokenBucket
}

type TokenBucket struct {
    tokens      int
    lastReset   time.Time
    requestLog  []TokenEntry
}

type TokenEntry struct {
    timestamp time.Time
    tokens    int  // 这次请求消耗的 token 数
}

func (rl *TokenRateLimiter) Allow(key string, estimatedTokens int) bool {
    rl.mu.Lock()
    defer rl.mu.Unlock()

    bucket := rl.getBucket(key)
    now := time.Now()

    // 清理窗口外的记录
    cutoff := now.Add(-rl.window)
    var valid []TokenEntry
    currentTokens := 0
    for _, entry := range bucket.requestLog {
        if entry.timestamp.After(cutoff) {
            valid = append(valid, entry)
            currentTokens += entry.tokens
        }
    }
    bucket.requestLog = valid

    // 判断是否允许
    if currentTokens+estimatedTokens > rl.maxTokens {
        return false
    }

    // 放行，记录这次请求
    bucket.requestLog = append(bucket.requestLog, TokenEntry{
        timestamp: now,
        tokens:    estimatedTokens,
    })
    return true
}

这段代码的核心逻辑：每个用户一个滑动窗口，窗口内累计消耗的 token 数不超过上限。和固定窗口限流相比，滑动窗口不会在窗口边界产生突发流量。

一个真实的限流配置

生产环境我们一般是三层限流：

层级	维度	限制	目的
全局	所有请求	100K tokens/min	保护后端模型不被打满
用户	单用户	10K tokens/min	防止单个用户耗尽配额
端点	特定 API	50K tokens/min	保护高成本接口

这里有个坑：估算 token 数和实际 token 数的偏差。输入文本的 token 数可以通过 tiktoken 库准确估算，但输出的 token 数没法提前知道。所以通常的做法是：

输入时估算 input tokens，先扣额度
响应回来后计算实际 output tokens，多退少补
如果用户实际消耗远超估算（比如设置了 max_tokens=4000 但模型真的输出 4000），下次请求就会被限流

// 实际 token 数回填
func (rl *TokenRateLimiter) Adjust(key string, estimated, actual int) {
    rl.mu.Lock()
    defer rl.mu.Unlock()

    bucket := rl.getBucket(key)
    if len(bucket.requestLog) == 0 {
        return
    }

    // 找到最近一条记录，调整 token 数
    last := &bucket.requestLog[len(bucket.requestLog)-1]
    diff := actual - estimated
    last.tokens += diff  // 可能为负（估算偏高），也可能为正（估算偏低）
}

路由：多模型智能分发

你的系统可能同时接了 GPT-4、Claude、本地部署的 Llama——不同场景用不同模型，这不是"选择困难"，是成本和性能的 trade-off。

AI 网关的路由层要解决三个问题：

1. 基于内容复杂度的路由

简单问题（"今天天气怎么样"）走便宜模型，复杂问题（"分析这份财报的风险因素"）走强模型。

python

def route_by_complexity(user_input: str) -> str:
    """根据输入内容复杂度选择模型"""
    # 简单启发式规则
    word_count = len(user_input.split())
    has_code = bool(re.search(r'```|def |class |import ', user_input))
    has_numbers = bool(re.search(r'\d{4,}', user_input))

    if word_count < 10 and not has_code and not has_numbers:
        return "gpt-3.5-turbo"       # 简单问题
    elif has_code or word_count > 50:
        return "gpt-4"               # 代码/长文本用强模型
    else:
        return "claude-sonnet"        # 中等复杂度

生产环境不会只靠规则，通常会加一个轻量级分类模型（几十 MB 的那种），在网关层先判断请求属于哪一类，再决定路由。

2. 基于成本和延迟的路由

type ModelEndpoint struct {
    Name        string
    CostPerMTok float64   // 每百万 token 价格
    AvgLatency  time.Duration  // 平均延迟
    SuccessRate float64   // 成功率
    QuotaLeft   int       // 剩余配额
}

func SelectBestEndpoint(endpoints []ModelEndpoint, budget float64, maxLatency time.Duration) *ModelEndpoint {
    var best *ModelEndpoint
    for i := range endpoints {
        ep := &endpoints[i]
        if ep.CostPerMTok > budget {
            continue
        }
        if ep.AvgLatency > maxLatency {
            continue
        }
        if ep.SuccessRate < 0.95 {
            continue
        }
        if best == nil || ep.CostPerMTok < best.CostPerMTok {
            best = ep
        }
    }
    return best
}

3. 降级策略

主模型挂了怎么办？网关的降级策略决定了用户体验：

请求 → GPT-4（主模型）
        ↓ 超时/失败
      → Claude-Sonnet（备选 1）
        ↓ 超时/失败
      → GPT-3.5-Turbo（备选 2）
        ↓ 超时/失败
      → 缓存命中（语义缓存兜底）
        ↓ 未命中
      → 返回"服务繁忙，请稍后重试"

降级不是简单换模型，要注意模型行为差异——GPT-4 的输出格式和 GPT-3.5 不同，如果你的下游依赖特定 JSON 结构，降级后可能会解析失败。所以生产环境通常会在网关层做一次响应格式标准化。

语义缓存：最容易被低估的优化

大模型最贵的不是计算，是重复计算。用户问"Java 和 Python 哪个适合做 AI"，五分钟后另一个人问"AI 开发用 Java 好还是 Python 好"——同一个问题，调了两次模型，花了两份钱。

语义缓存解决的就是这个问题。

实现原理

用户输入 → 向量化（embedding） → 向量库相似度搜索 → 相似度 > 阈值 → 返回缓存
                                                         ↓ 不满足
                                                   调用模型 → 缓存结果

python

import hashlib
from sentence_transformers import SentenceTransformer

class SemanticCache:
    def __init__(self, threshold=0.92):
        self.model = SentenceTransformer('all-MiniLM-L6-v2')
        self.threshold = threshold
        self.cache = {}  # embedding_hash -> {question, answer, tokens}

    def get(self, question: str) -> dict | None:
        q_embedding = self.model.encode(question)
        q_hash = hashlib.md5(q_embedding.tobytes()).hexdigest()

        # 精确匹配（哈希碰撞概率极低）
        if q_hash in self.cache:
            return self.cache[q_hash]

        # 语义相似度匹配
        best_match = None
        best_score = 0
        for cached_hash, cached in self.cache.items():
            score = cosine_similarity(
                q_embedding.reshape(1, -1),
                cached['embedding'].reshape(1, -1)
            )[0][0]
            if score > best_score:
                best_score = score
                best_match = cached

        if best_score >= self.threshold:
            return best_match
        return None

    def put(self, question: str, answer: str, embedding):
        q_hash = hashlib.md5(embedding.tobytes()).hexdigest()
        self.cache[q_hash] = {
            'question': question,
            'answer': answer,
            'embedding': embedding,
            'created_at': time.time(),
        }

阈值怎么设

阈值是语义缓存的核心参数：

0.95+：几乎只缓存完全相同的重写，命中率低但准确
0.90-0.92：适合 FAQ 类场景，"怎么安装"和"如何安装"能匹配上
0.85-：风险太高，可能把不相关的问题误判为相同

生产环境建议从 0.92 开始，然后根据实际命中日志调。

缓存失效策略

缓存不能无限增长，几个策略：

TTL 过期：24 小时后自动失效（适合时效性强的内容）
LRU 淘汰：内存满了淘汰最久没用的（适合固定内存的场景）
手动失效：模型更新或 prompt 变更后主动清空（避免返回旧版本模型的输出）

可观测性：你不知道的钱去哪了

没有监控的 AI 网关就是闭着眼睛花钱。以下是必须追踪的指标：

// 请求级别的追踪数据
type RequestTrace struct {
    RequestID     string
    UserID        string
    Model         string
    InputTokens   int
    OutputTokens  int
    TotalCost     float64
    Latency       time.Duration
    CacheHit      bool
    RouteDecision string     // 为什么选了这个模型
    Error         string     // 如果有错误
}

// 聚合指标
type Metrics struct {
    TotalRequests      int
    TotalTokens        int
    TotalCost          float64
    AvgLatency         time.Duration
    CacheHitRate       float64    // 缓存命中率
    ErrorRate          float64    // 错误率
    ModelDistribution  map[string]int  // 各模型调用次数
}

这些指标不只是"看看而已"——它们驱动决策：

缓存命中率低于 30%？调低相似度阈值，或者缓存过期时间太短
某个模型错误率飙升？自动从路由表中摘除
成本突然上涨？检查是否有异常用户或 prompt 注入攻击

一个完整的请求生命周期

最后，把一个请求从进来到出去的完整流程串起来：

1. 请求到达网关
   ↓
2. 用户级别 token 限流检查
   ↓ 通过
3. 语义缓存查询（embedding 相似度匹配）
   ↓ 未命中
4. 路由决策（根据复杂度/成本/可用性选模型）
   ↓
5. 发送到目标模型
   ↓
6. 接收流式响应，同时统计 token 消耗
   ↓
7. 实际 token 数回填限流器
   ↓
8. 响应写入语义缓存
   ↓
9. 记录追踪数据到指标系统
   ↓
10. 返回给客户端

整个过程在网关层完成，业务代码不需要关心这些细节。

写在最后

AI 网关不是"又一个 API 网关"。它的核心价值是在 LLM 前面加一层可控性——你能控制成本（路由+缓存）、控制稳定性（限流+降级）、控制可观测性（追踪+指标）。

没有网关的 AI 应用就像没有刹车的车——跑得挺快，但停不下来。

ai_gateway_arch.png

从一次生产事故说起 ​

AI 网关到底干什么 ​

限流：Token 级别的精准控制 ​

一个真实的限流配置 ​

路由：多模型智能分发 ​

1. 基于内容复杂度的路由 ​

2. 基于成本和延迟的路由 ​

3. 降级策略 ​

语义缓存：最容易被低估的优化 ​

实现原理 ​

阈值怎么设 ​

缓存失效策略 ​

可观测性：你不知道的钱去哪了 ​

一个完整的请求生命周期 ​

写在最后 ​