适用场景与目标

为什么今天要聊 AI Gateway

昨天我们讲了推理引擎选型（SGLang vs vLLM），但即便选了最快的引擎，如果对 LLM API 调用没有统一层，成本会失控——相同语义的问题重复发送给 OpenAI/Anthropic，Token 浪费 30%~70%，缓存完全靠业务代码拼凑。

2026 年，AI Gateway（智能路由层）已经从”可选项”变成”必需品”。核心能力：

• 语义缓存：即使 Prompt 措辞略有不同，也能命中缓存（例如用户问”How do I reset password” vs “怎么重置密码”语义相近）
• 智能路由：按模型可用性/价格/延迟自动切换供应商
• 预算控制：按团队/用户/功能维度配额，超限自动降级
• 可观测性：Token 消费、延迟、命中率一目了然

目标读者

有实际 LLM 调用量（>100万 Token/月）、使用多个模型（OpenAI + Anthropic + 自部署）的工程团队。

---

最小可行方案（MVP）步骤

方案选型：LiteLLM vs Bifrost vs 自研

方案	部署难度	语义缓存	成本控制	适用规模
LiteLLM（开源）	⭐⭐ 中	❌ 仅精确匹配	基础监控	中小型
Bifrost（Maxim）	⭐⭐⭐ 简单	✅ 原生支持	原生 + 预算分配	中大型
自研 FastAPI 层	⭐⭐⭐⭐ 复杂	✅ 需自建 Embedding	灵活自控	超大型

推荐 MVP 起步：LiteLLM（免费，生产可用），进阶后平滑迁移 Bifrost。

Step 1：安装 & 基础配置（LiteLLM v1.x）

pip install litellm 'litellm[proxy]'
# 创建配置目录
mkdir -p ~/ai-gateway && cd ~/ai-gateway

Step 2：编写 liteconfig.yaml

model_list:
  - model_name: gpt-4o
    litellm_params:
      model: gpt-4o
      api_key: os.environ/AZURE_OPENAI_API_KEY
      api_base: os.environ/AZURE_OPENAI_API_BASE
  
  - model_name: claude-3-5-sonnet
    litellm_params:
      model: anthropic/claude-3-5-sonnet
      api_key: os.environ/ANTHROPIC_API_KEY

  - model_name: deepseek-chat
    litellm_params:
      model: deepseek/deepseek-chat-v3-0324
      api_base: https://api.deepseek.com/v1
      api_key: os.environ/DEEPSEEK_API_KEY

litellm_settings:
  drop_params: true
  set_verbose: true

Step 3：启动 Gateway

litellm --config ~/ai-gateway/liteconfig.yaml --port 4000

Step 4：验证部署

# 测试 OpenAI 兼容接口
curl http://localhost:4000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer sk-test" \
  -d '{
    "model": "gpt-4o",
    "messages": [{"role": "user", "content": "解释一下 Go 的协程"}]
  }'

---

关键实现细节

1. 语义缓存实现（基于 Embedding + 向量数据库）

LiteLLM 开源版只支持精确匹配缓存，要实现语义缓存需要扩展层：

# semantic_cache.py
from sentence_transformers import SentenceTransformer
import numpy as np
import redis
import hashlib

model = SentenceTransformer('all-MiniLM-L6-v2')
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

def get_cache_key(prompt: str, model: str) -> str:
    """基于语义生成缓存 key"""
    embedding = model.encode(prompt)
    # 取前256维压缩（节省存储 + 加速）
    compressed = embedding[:256]
    vector_str = ','.join(f'{v:.6f}' for v in compressed)
    hash_val = hashlib.sha256(vector_str.encode()).hexdigest()[:16]
    return f"sem_cache:{model}:{hash_val}"

def get_cached_response(prompt: str, model: str) -> str | None:
    key = get_cache_key(prompt, model)
    return redis_client.get(key)

def set_cached_response(prompt: str, model: str, response: str, ttl: int = 3600):
    key = get_cache_key(prompt, model)
    redis_client.setex(key, ttl, response)

关键参数说明：

• 向量维度压缩到 256：compressed = embedding[:256]（原始 384 维压缩，语义损失 <5%）
• TTL 默认 1 小时（避免过期信息占用缓存）
• Embedding 模型：all-MiniLM-L6-v2（最快，CPU 可跑）

2. 智能路由：按价格 + 可用性自动切换

# smart_router.py
import asyncio
from litellm import completion

# 定义模型路由策略（价格从低到高）
ROUTING_RULES = {
    "deepseek-chat":  {"price_per_1k_tokens": 0.001, "max_rpm": 2000},
    "gpt-4o-mini":    {"price_per_1k_tokens": 0.003, "max_rpm": 500},
    "claude-3-5-sonnet": {"price_per_1k_tokens": 0.015, "max_rpm": 100},
}

async def smart_route(messages: list, budget_limit: float = 0.05):
    """按预算限制自动选择最便宜的可用模型"""
    for model_name, info in ROUTING_RULES.items():
        if info["price_per_1k_tokens"] <= budget_limit:
            try:
                response = await completion(
                    model=model_name,
                    messages=messages,
                    timeout=30
                )
                return response
            except Exception as e:
                # 模型不可用，跳到下一个
                continue
    raise RuntimeError("所有模型均不可用或超出预算")

3. 预算控制（按用户/团队维度）

# budget_enforcer.py
from collections import defaultdict
from datetime import datetime, timedelta

class BudgetTracker:
    def __init__(self):
        self.spending = defaultdict(float)
        self.limits = defaultdict(lambda: defaultdict(float))
    
    def set_limit(self, user_id: str, monthly_limit_usd: float):
        self.limits[user_id]["monthly"] = monthly_limit_usd
    
    def check_and_charge(self, user_id: str, model: str, input_tokens: int, output_tokens: int) -> bool:
        price = ROUTING_RULES[model]["price_per_1k_tokens"]
        cost = (input_tokens + output_tokens) / 1000 * price
        
        self.spending[user_id] += cost
        limit = self.limits[user_id]["monthly"]
        
        if limit > 0 and self.spending[user_id] > limit:
            return False  # 超预算，拒绝请求
        return True

budget_tracker = BudgetTracker()
budget_tracker.set_limit("user_abc", monthly_limit_usd=50.0)

---

常见坑与规避清单

坑 1：缓存 Key 膨胀（Redis 内存爆炸）

问题：语义缓存 key 数量随请求量线性增长，Redis 内存失控。

规避：

• 使用 SCAN + TTL 组合定期清理：redis-cli --scan --pattern 'sem_cache:*' | xargs redis-cli EXPIRE 3600
• 压缩向量到 128 维（embedding[:128]），key 长度减少 50%
• 设置缓存容量上限（Redis maxmemory-policy allkeys-lru）

坑 2：路由时模型”自杀”（某模型突发高延迟）

问题：某供应商突发延迟，但路由层没有超时即熔断，导致请求堆积。

规避：

# 熔断器实现
from circuitbreaker import circuit

@circuit(error_threshold=5, timeout_duration=30)
async def safe_complete(model: str, messages: list):
    # 连续5次失败则熔断30秒
    return await completion(model=model, messages=messages, timeout=10)

坑 3：Token 计数不准确（成本估算偏差）

问题：直接用 Embedding 模型生成缓存 key，无法反映实际 Token 消耗。

规避：

• Token 计数统一走 tiktoken（OpenAI）或 anthropic-tokenizer（Claude）
• 日志中记录 prompt_tokens + completion_tokens 双方数据

坑 4：多语言 Prompt 缓存失效（RAG 场景常见）

问题：中文问题”Did you mean X”被翻译后无法命中缓存。

规避：

• Prompt 在入缓存前统一做”语义标准化”：去除语气词、统一语言、去除变量占位符内容
• 建议在 Embedding 前先做 prompt cleanup，而非直接全文 Embedding

坑 5：LiteLLM 日志过大（磁盘爆满）

问题：set_verbose: true 导致日志量暴增。

规避：

# liteconfig.yaml 中关闭 verbose
litellm_settings:
  set_verbose: false  # 生产环境必须关闭
  # 使用结构化日志输出到文件
  json_logs: true

---

成本/性能/维护权衡

成本对比（以 1000 万 Token/月为例）

方案	基础设施成本	缓存节省	实际月成本	适用规模
直接调 OpenAI	$420（gpt-4o）	0	$420	<100万 Token
LiteLLM + 精确缓存	$50 EC2	~30%	~$300	100-1000万
Bifrost + 语义缓存	$200（托管）	~60%	~$170	>1000万
自研（Redis + Embedding）	$150 EC2 + Redis	~60%	~$200	中大型

性能影响

• 语义缓存命中时延：<5ms（Redis 读取 + 反序列化）
• 语义缓存未命中时延：额外 200~500ms（Embedding 计算）
• 智能路由延迟增加：<10ms（可忽略）

维护成本

• LiteLLM：社区活跃，升级频率高（建议每2周更新一次）
• Bifrost：托管方案免运维，适合没有专职 DevOps 的团队
• 自研方案：维护成本最高，需要专门的人力投入

---

一周内可执行行动清单

Day 1-2：环境搭建

• 在测试服务器安装 LiteLLM：pip install litellm
• 创建 liteconfig.yaml，配置现有的 API Key
• 启动 Gateway 并用 curl 验证

Day 3-4：接入现有业务

• 选定一个低流量 API 端点作为试点
• 在业务代码中替换 LLM 调用为 Gateway 调用
• 开启 LiteLLM 内置日志，对比请求数和 Token 消耗

Day 5：缓存层搭建

• 部署 Redis 实例（或使用 Redis Cloud）
• 集成语义缓存代码（semantic_cache.py）
• 验证”相同语义不同措辞”的请求确实能命中缓存

Day 6：可观测性

• 接入 Prometheus + Grafana（LiteLLM 支持 /metrics 端点）
• 配置关键看板：Token 消耗、请求延迟、缓存命中率

Day 7：成本评估与优化

• 统计一周 Token 消费数据
• 计算缓存命中率（目标 >30%）
• 决定是否需要迁移 Bifrost 或自建高级路由

---

参考资源

• LiteLLM 官方文档：https://docs.litellm.ai/
• Bifrost by Maxim：https://www.getmaxim.ai/
• Semantic Cache 技术细节：https://docs.litellm.ai/docs/tutorials/semantic_caching