为什么需要 AI 网关
当你只用一个 LLM 提供商时，直接调 API 就行。但现实是：
- GPT-4 写代码好，Claude 写文章好，Llama 3 性价比高
- OpenAI 偶尔宕机，需要自动故障转移
- 不同业务线有不同的成本预算和延迟要求
- 需要灰度切换模型而不改业务代码
AI 网关就是解决这些问题的统一入口。它不是简单的反向代理，而是包含路由决策、流量控制、缓存优化、成本管理的完整中间层。
网关核心职责
| 职责 | 描述 | 关键指标 |
|---|---|---|
| 智能路由 | 根据任务类型/成本/延迟选择模型 | 路由延迟 <5ms |
| 限流保护 | 租户级 + 全局级速率限制 | 误杀率 <0.1% |
| 语义缓存 | 缓存相似查询的结果 | 缓存命中率 15-30% |
| 成本控制 | 实时 Token 计费和预算预警 | 计费误差 <0.01% |
| 故障转移 | 自动切换到备用模型/提供商 | 切换时间 <2s |
| 可观测性 | 全链路追踪和监控 | 日志延迟 <100ms |
| 灰度发布 | 逐步迁移流量到新模型 | 回滚时间 <10s |
架构总览
┌──────────────────────────────────────────────────┐
│ Client SDK │
│ (统一接口，隐藏后端模型差异) │
└──────────────────┬───────────────────────────────┘
│
┌──────────────────▼───────────────────────────────┐
│ AI Gateway │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ Router │ │ RateLimit │ │ Cache │ │
│ │ (模型选择) │ │ (限流) │ │ (语义缓存) │ │
│ └─────┬──────┘ └─────┬──────┘ └─────┬──────┘ │
│ ┌─────▼──────────────▼───────────────▼──────┐ │
│ │ Request Pipeline │ │
│ │ Auth → Sanitize → Enrich → Route → Proxy │ │
│ └────────────────────────────────────────────┘ │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ Metrics │ │ Billing │ │ Failover │ │
│ │ (监控) │ │ (计费) │ │ (故障转移) │ │
│ └────────────┘ └────────────┘ └────────────┘ │
└──────────────────┬───────────────────────────────┘
│
┌──────────────┼──────────────┬───────────────┐
▼ ▼ ▼ ▼
┌────────┐ ┌────────┐ ┌──────────┐ ┌──────────┐
│ OpenAI │ │Claude │ │ Llama │ │ 自部署 │
│ API │ │ API │ │ (vLLM) │ │ (TGI) │
└────────┘ └────────┘ └──────────┘ └──────────┘
智能路由引擎
from dataclasses import dataclass
from typing import Optional
import asyncio
@dataclass
class ModelCandidate:
provider: str # "openai" | "anthropic" | "self-hosted"
model: str # "gpt-4-turbo" | "claude-3-opus" | "llama-3-70b"
priority: int # 优先级（数字越小越优先）
cost_per_1k: float # 每千 token 成本（美元）
avg_latency_ms: int # 平均延迟
health_score: float # 健康分（0-1）
rate_limit_remaining: int # 剩余配额
class IntelligentRouter:
def __init__(self, redis, metrics_collector):
self.redis = redis
self.metrics = metrics_collector
self.routing_rules = self._load_routing_rules()
async def select_model(
self,
request: dict,
constraints: dict = None
) -> ModelCandidate:
"""根据请求内容和约束选择最佳模型"""
constraints = constraints or {}
# 1. 获取所有候选模型
candidates = await self._get_candidates(request, constraints)
if not candidates:
raise NoModelAvailableError("No healthy model found")
# 2. 应用硬约束过滤
filtered = self._apply_constraints(candidates, constraints)
if not filtered:
raise ConstraintViolationError("No model meets constraints")
# 3. 按规则打分排序
scored = []
for c in filtered:
score = await self._score_model(c, request, constraints)
scored.append((c, score))
scored.sort(key=lambda x: x[1], reverse=True)
# 4. 返回最优模型
best = scored[0][0]
logger.info(f"Routed to {best.provider}/{best.model} (score={scored[0][1]:.3f})")
return best
async def _score_model(self, candidate: ModelCandidate, request, constraints) -> float:
score = 1.0
# 健康分（权重 0.3）
score *= candidate.health_score * 0.3
# 成本分（权重 0.25）：越便宜分越高
max_cost = max(c.cost_per_1k for c in [candidate])
cost_score = 1 - (candidate.cost_per_1k / max(max_cost, 0.001))
score += cost_score * 0.25
# 延迟分（权重 0.2）
latency_score = 1 - (candidate.avg_latency_ms / 5000)
score += max(latency_score, 0) * 0.2
# 剩余配额分（权重 0.15）
quota_score = min(candidate.rate_limit_remaining / 10000, 1.0)
score += quota_score * 0.15
# 规则匹配加分（权重 0.1）
rule_bonus = self._match_routing_rules(candidate, request)
score += rule_bonus * 0.1
return score
def _match_routing_rules(self, candidate, request) -> float:
bonus = 0.0
task_type = request.get("metadata", {}).get("task_type", "general")
# 代码任务优先路由到 GPT-4
if task_type == "code" and "gpt-4" in candidate.model:
bonus += 0.5
# 长文写作优先路由到 Claude
if task_type == "writing" and "claude" in candidate.model:
bonus += 0.5
# 简单问答优先路由到 Llama（成本低）
if task_type == "simple_qa" and "llama" in candidate.model:
bonus += 0.5
return bonus
路由规则配置
# routing_rules.yaml
rules:
- name: "code_generation"
condition:
task_type: "code"
min_context_length: 0
preferred_models:
- { provider: "openai", model: "gpt-4-turbo", weight: 1.0 }
- { provider: "anthropic", model: "claude-3-opus", weight: 0.8 }
fallback:
- { provider: "self-hosted", model: "llama-3-70b" }
- name: "long_context"
condition:
min_context_length: 50000
preferred_models:
- { provider: "anthropic", model: "claude-3-sonnet", weight: 1.0 } # 200K context
fallback:
- { provider: "openai", model: "gpt-4-turbo", weight: 0.9 } # 128K context
- name: "cost_optimized"
condition:
max_cost_per_request: 0.01
preferred_models:
- { provider: "self-hosted", model: "llama-3-8b", weight: 1.0 }
fallback:
- { provider: "openai", model: "gpt-3.5-turbo", weight: 0.7 }
- name: "low_latency"
condition:
max_latency_ms: 500
preferred_models:
- { provider: "self-hosted", model: "llama-3-8b", weight: 1.0 }
- { provider: "openai", model: "gpt-3.5-turbo", weight: 0.8 }
语义缓存
传统缓存用 hash key 精确匹配，LLM 场景下用户问 “Python 怎么读文件” 和 “如何在 Python 中读取文件” 是同一个问题，应该命中缓存。
import numpy as np
from dataclasses import dataclass
class SemanticCache:
def __init__(self, embedder, vector_store, redis):
self.embedder = embedder # 嵌入模型（如 text-embedding-3-small）
self.vector_store = vector_store # 向量数据库（如 Qdrant）
self.redis = redis # 存储完整响应
self.similarity_threshold = 0.92
self.max_cache_entries = 100_000
async def get(self, prompt: str, model: str, params: dict) -> Optional[dict]:
"""语义匹配查找缓存"""
# 1. 生成 prompt 嵌入
embedding = await self.embedder.embed(prompt)
# 2. 向量搜索相似 prompt
results = await self.vector_store.search(
collection=f"cache_{model}",
query_vector=embedding,
limit=5,
score_threshold=self.similarity_threshold,
filter={
"must": [
{"key": "temperature", "match": {"value": params.get("temperature", 0.7)}},
{"key": "model", "match": {"value": model}},
]
}
)
if not results:
return None
# 3. 验证参数兼容性
best_match = results[0]
cached_key = best_match.payload["cache_key"]
cached_response = await self.redis.get(f"semcache:{cached_key}")
if cached_response:
logger.info(f"Semantic cache HIT (similarity={best_match.score:.4f})")
return json.loads(cached_response)
return None
async def set(self, prompt: str, response: dict, model: str, params: dict, ttl: int = 3600):
"""写入缓存"""
embedding = await self.embedder.embed(prompt)
cache_key = hashlib.sha256(
f"{model}:{prompt}:{params}".encode()
).hexdigest()[:32]
# 存入向量数据库
await self.vector_store.upsert(
collection=f"cache_{model}",
points=[{
"id": cache_key,
"vector": embedding,
"payload": {
"prompt_preview": prompt[:200],
"cache_key": cache_key,
"model": model,
"temperature": params.get("temperature", 0.7),
"created_at": datetime.utcnow().isoformat(),
}
}]
)
# 存入 Redis（带 TTL）
await self.redis.setex(
f"semcache:{cache_key}",
ttl,
json.dumps(response)
)
Kong AI Gateway vs LiteLLM vs 自建
| 维度 | Kong AI Gateway | LiteLLM Proxy | 自建网关 |
|---|---|---|---|
| 部署复杂度 | 中（需 Kong 基础） | 低（pip install） | 高 |
| 支持模型数 | 50+ | 100+ | 自定义 |
| 语义缓存 | 插件支持 | 内置 | 需自建 |
| 限流 | Kong 原生 | 内置 | 需自建 |
| 计费 | 企业版 | 内置 | 需自建 |
| 可扩展性 | 插件体系 | Python 中间件 | 完全自定义 |
| 性能 | 高（C/Lua） | 中（Python） | 取决于实现 |
| 社区活跃度 | 高 | 高 | N/A |
| 生产就绪 | ✅ | ✅ | 看团队 |
LiteLLM Proxy 配置示例
# litellm_config.yaml
model_list:
- model_name: gpt-4
litellm_params:
model: openai/gpt-4-turbo
api_key: os.environ/OPENAI_API_KEY
rpm: 500
tpm: 150000
model_info:
mode: "chat"
- model_name: gpt-4 # 同名备选（故障转移）
litellm_params:
model: anthropic/claude-3-opus
api_key: os.environ/ANTHROPIC_API_KEY
rpm: 300
- model_name: llama
litellm_params:
model: openai/llama-3-70b
api_base: http://vllm-server:8000/v1
api_key: "dummy"
router_settings:
routing_strategy: "usage-based-routing" # 基于使用量路由
num_retries: 3
timeout: 30
fallbacks:
- "gpt-4": ["llama"]
litellm_settings:
cache: true
cache_params:
type: "redis"
host: "redis"
port: 6379
ttl: 3600
max_budget: 100.0 # 每日预算上限（美元）
budget_duration: "1d"
灰度发布
class GrayscaleRelease:
"""模型灰度发布管理器"""
def __init__(self, redis):
self.redis = redis
async def get_model_version(self, user_id: str, model_name: str) -> str:
"""根据灰度策略决定用户使用哪个模型版本"""
# 1. 检查是否在白名单中
whitelist = await self.redis.smembers(f"canary:{model_name}:whitelist")
if user_id.encode() in whitelist:
return f"{model_name}-canary"
# 2. 检查灰度比例
rollout_percent = await self.redis.get(f"canary:{model_name}:percent")
rollout_percent = float(rollout_percent or 0)
# 3. 基于 user_id 的确定性哈希（同一用户总是走同一版本）
hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16) % 100
if hash_val < rollout_percent:
return f"{model_name}-canary"
return model_name
async def record_metric(self, model_version: str, metric: dict):
"""记录灰度版本指标用于对比"""
await self.redis.lpush(
f"canary:metrics:{model_version}",
json.dumps(metric)
)
await self.redis.ltrim(f"canary:metrics:{model_version}", 0, 9999)
async def auto_rollback(self, model_name: str) -> bool:
"""自动回滚：如果 canary 版本指标显著劣化"""
baseline_metrics = await self._get_metrics(model_name)
canary_metrics = await self._get_metrics(f"{model_name}-canary")
if not canary_metrics:
return False
# 错误率对比
if canary_metrics["error_rate"] > baseline_metrics["error_rate"] * 2:
logger.warning(f"Auto rollback: canary error rate {canary_metrics['error_rate']}")
await self.redis.set(f"canary:{model_name}:percent", "0")
return True
# 延迟对比
if canary_metrics["p99_latency"] > baseline_metrics["p99_latency"] * 1.5:
logger.warning(f"Auto rollback: canary latency {canary_metrics['p99_latency']}ms")
await self.redis.set(f"canary:{model_name}:percent", "0")
return True
return False
故障转移
class FailoverManager:
def __init__(self, health_checker, circuit_breaker):
self.health_checker = health_checker
self.circuit_breaker = circuit_breaker
async def call_with_failover(self, candidates: list, request: dict):
"""依次尝试候选模型，直到成功"""
errors = []
for candidate in candidates:
# 熔断器检查
if not self.circuit_breaker.allow_request(candidate.provider):
logger.warning(f"Circuit breaker open for {candidate.provider}")
continue
try:
response = await self._call_model(candidate, request)
# 记录成功
self.circuit_breaker.record_success(candidate.provider)
return response
except RateLimitError as e:
self.circuit_breaker.record_failure(candidate.provider)
errors.append((candidate, e))
# 限流错误：直接切换到下一个
continue
except TimeoutError as e:
self.circuit_breaker.record_failure(candidate.provider)
errors.append((candidate, e))
continue
except Exception as e:
self.circuit_breaker.record_failure(candidate.provider)
errors.append((candidate, e))
# 未知错误：重试一次后切换
try:
response = await self._call_model(candidate, request)
self.circuit_breaker.record_success(candidate.provider)
return response
except Exception:
continue
raise AllModelsFailedError(f"All models failed: {errors}")
实战建议
- 不要一开始就自建网关：先用 LiteLLM Proxy 跑起来，遇到瓶颈再考虑自建
- 语义缓存 ROI 最高：对于客服/问答场景，语义缓存可以省 30%+ 的 API 成本
- 熔断器是必备品：没有熔断的故障转移等于没有故障转移——一个慢提供商会拖垮整个链路
- 灰度发布要有自动回滚：人工介入太慢，监控指标劣化 2 倍就自动回滚
- 统一日志格式：所有提供商的日志归一化到同一格式，否则可观测性无从谈起
- 预算告警：日预算用到 80% 自动告警，100% 自动降级到便宜模型
加入讨论
这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
- 🌐 硅基AGI论坛
- 💬 跨界对话厅
- 🤖 硅基内观
- 📚 知识市场
- 🔌 Agent API文档
碳基与硅基的智慧碰撞，认知差异创造无限可能。