为什么需要LLM API网关
当企业级应用调用LLM API时，直接暴露上游接口会面临三个核心问题：成本失控、延迟波动和服务不可用。一个设计良好的API网关能在这三者之间取得平衡，同时提供统一的鉴权、计费和可观测能力。
本文将从工程实践角度，拆解LLM API网关的三大核心模块。
限流策略设计
多维度限流模型
与传统API不同，LLM调用具有token维度的特性。一个简单的QPS限流无法覆盖真实场景，需要多维度组合：
| 维度 | 限流键 | 典型阈值 | 适用场景 |
|---|---|---|---|
| QPS | API Key + 路径 | 50 req/s | 防恶意刷量 |
| TPM | API Key + 模型 | 100K tokens/min | 控制成本 |
| 并发数 | 用户ID | 5 concurrent | 防资源独占 |
| 日总量 | 租户ID | 10M tokens/day | 预算控制 |
令牌桶+滑动窗口的混合实现
以下是基于Redis的混合限流实现：
import asyncio
import time
import redis.asyncio as redis
class TokenBucketRateLimiter:
def __init__(self, redis_client: redis.Redis, capacity: int, refill_rate: float):
self.redis = redis_client
self.capacity = capacity
self.refill_rate = refill_rate # tokens per second
async def acquire(self, key: str, tokens: int = 1) -> bool:
lua_script = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local requested = tonumber(ARGV[3])
local now = tonumber(ARGV[4])
local bucket = redis.call('HMGET', key, 'tokens', 'timestamp')
local current_tokens = tonumber(bucket[1]) or capacity
local last_timestamp = tonumber(bucket[2]) or now
-- 补充令牌
local elapsed = now - last_timestamp
current_tokens = math.min(capacity, current_tokens + elapsed * refill_rate)
if current_tokens < requested then
return 0
end
current_tokens = current_tokens - requested
redis.call('HMSET', key, 'tokens', current_tokens, 'timestamp', now)
redis.call('EXPIRE', key, 3600)
return 1
"""
now = time.time()
result = await self.redis.eval(
lua_script, 1, f"ratelimit:{key}",
self.capacity, self.refill_rate, tokens, now
)
return bool(result)
关键设计点：使用Lua脚本保证原子性，避免竞态条件。对于TPM限流，需要在请求完成后根据实际token消耗做补充扣减。
多级缓存架构
LLM响应的生成成本高、延迟大，缓存命中率每提升1%都能带来显著的成本节省。
缓存层级设计
请求 → [本地LRU缓存] → [Redis语义缓存] → [上游LLM]
↑ 命中即返回 ↑ 近似匹配命中 ↑ 穿透回源
语义缓存实现
传统精确匹配缓存对LLM几乎无效——用户表述的微小差异就会导致cache miss。语义缓存通过embedding相似度判断是否可复用历史响应：
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
class SemanticCache:
def __init__(self, redis_client, embedding_model, threshold: float = 0.95):
self.redis = redis_client
self.embedding_model = embedding_model
self.threshold = threshold
async def get(self, query: str, top_k: int = 5):
query_emb = await self.embedding_model.embed(query)
# 从Redis获取近邻候选
candidates = await self.redis.execute_command(
'FT.SEARCH', 'semantic_cache_idx',
f'*=>[KNN {top_k} @embedding $query_vec]',
'PARAMS', 2, 'query_vec', query_emb.tobytes(),
'RETURN', 3, 'response', 'embedding', '__score',
'LIMIT', 0, top_k
)
for candidate in candidates[1:]:
score = float(candidate[2].get('__score', 1.0))
similarity = 1 - score # 距离转相似度
if similarity >= self.threshold:
return candidate[2].get('response')
return None
缓存策略权衡
| 策略 | 优点 | 风险 | 建议 |
|---|---|---|---|
| 精确匹配 | 零误缓存 | 命中率低 | 用于函数调用 |
| 语义缓存 | 命中率高 | 可能返回不精确 | 用于闲聊/FAQ |
| TTL过期 | 数据新鲜 | 频繁回源 | 设15-30分钟 |
| 主动失效 | 精确控制 | 实现复杂 | 关键业务用 |
故障转移机制
多供应商健康检查
class ProviderHealthMonitor:
def __init__(self, providers: list[str]):
self.providers = {p: {"healthy": True, "failures": 0, "latency": 0} for p in providers}
self.circuit_threshold = 3
async def record_result(self, provider: str, success: bool, latency: float):
state = self.providers[provider]
state["latency"] = 0.7 * state["latency"] + 0.3 * latency # EMA平滑
if success:
state["failures"] = 0
state["healthy"] = True
else:
state["failures"] += 1
if state["failures"] >= self.circuit_threshold:
state["healthy"] = False
def get_best_provider(self, exclude: set[str] = None) -> str:
exclude = exclude or set()
candidates = [(p, s) for p, s in self.providers.items()
if s["healthy"] and p not in exclude]
if not candidates:
return None
return min(candidates, key=lambda x: x[1]["latency"])[0]
故障转移流程
- 熔断触发：连续失败N次，标记provider为unhealthy
- 降级切换：自动路由到次优provider，记录降级日志
- 半开探测：每隔30秒发一个探测请求验证恢复
- 优雅降级：所有provider不可用时，返回缓存的最近响应或预设fallback
总结
LLM API网关不是简单的反向代理，而是成本控制、稳定性保障和性能优化三者的交汇点。核心设计原则：
- 限流要多维度（QPS + TPM + 并发 + 日总量）
- 缓存要分层级（精确 + 语义 + TTL）
- 故障转移要自动化（熔断 + 探测 + 降级）
在生产环境中，建议结合Prometheus + Grafana建立完整的可观测体系，实时监控缓存命中率、限流拒绝率、各provider延迟分布，形成闭环优化。