微服务与 AI 网关设计
原创
灵阙教研团队
S 精选 进阶 |
约 7 分钟阅读
更新于 2026-02-27 AI 导读
微服务与 AI 网关设计 统一模型入口：路由、缓存、限流、容灾的架构实践 Maurice | 灵阙学院 一、为什么需要 AI 网关 当应用对接多个 LLM Provider 时，每个调用点直接集成 SDK 会导致以下问题： Key 散落：API Key 分散在各服务中，轮转和审计困难 无统一限流：单个服务超额调用导致全局 Key 被封 无容灾：Provider 宕机时需要手动切换，业务中断...
微服务与 AI 网关设计
统一模型入口：路由、缓存、限流、容灾的架构实践
Maurice | 灵阙学院
一、为什么需要 AI 网关
当应用对接多个 LLM Provider 时，每个调用点直接集成 SDK 会导致以下问题：
- Key 散落：API Key 分散在各服务中，轮转和审计困难
- 无统一限流：单个服务超额调用导致全局 Key 被封
- 无容灾：Provider 宕机时需要手动切换，业务中断
- 无成本可见：Token 消耗分散在各处，无法统一计量
AI 网关将这些横切关注点集中到一个架构层解决。
┌─────────────────────────────────────────────────────────────┐
│ AI 网关架构全景 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Service A ──┐ │
│ Service B ──┼── AI Gateway ──┬── OpenAI │
│ Service C ──┤ │ ├── Anthropic │
│ Agent D ──┘ │ ├── Google │
│ │ ├── 自部署 (vLLM) │
│ ┌───┴────┐ └── SiliconFlow │
│ │ 路由 │ │
│ │ 缓存 │ │
│ │ 限流 │ │
│ │ 重试 │ │
│ │ 计量 │ │
│ │ 日志 │ │
│ └────────┘ │
└─────────────────────────────────────────────────────────────┘
二、核心能力拆解
2.1 模型路由
请求进入 → 解析 model 字段 → 路由决策
│
├─ 直接映射: "gpt-4o" → OpenAI gpt-4o
├─ 别名映射: "fast" → gpt-4o-mini, "premium" → claude-opus
├─ 条件路由: if (tokens > 10000) → use claude (200K context)
├─ A/B 路由: 50% → gpt-4o, 50% → claude-sonnet (效果对比)
└─ 权重路由: 70% OpenAI + 30% 自部署 (成本优化)
路由配置示例：
routes:
- model: "chat-balanced"
strategy: "weighted"
targets:
- provider: openai
model: gpt-4o-mini
weight: 60
- provider: siliconflow
model: Qwen/Qwen2.5-72B-Instruct
weight: 40
- model: "chat-premium"
strategy: "fallback"
targets:
- provider: anthropic
model: claude-sonnet-4-20250514
priority: 1
- provider: openai
model: gpt-4o
priority: 2
2.2 多 Provider Failover
┌─── Provider A (主) ───── 成功 → 返回
│ 失败/超时
请求 → Failover ────┼─── Provider B (备) ───── 成功 → 返回
Controller │ 失败/超时
└─── Provider C (兜底) ─── 成功 → 返回
失败 → 报错
class FailoverChain:
def __init__(self, providers: list[ProviderConfig]):
self.providers = providers
self.circuit_breakers: dict[str, CircuitBreaker] = {}
async def execute(self, request: ChatRequest) -> ChatResponse:
errors = []
for provider in self.providers:
cb = self.circuit_breakers[provider.name]
if cb.is_open:
continue # 跳过已熔断的 Provider
try:
response = await provider.chat(
request, timeout=provider.timeout_seconds
)
cb.record_success()
return response
except (TimeoutError, ProviderError) as e:
cb.record_failure()
errors.append((provider.name, str(e)))
raise AllProvidersFailedError(errors)
2.3 限流策略
| 限流维度 | 说明 | 实现方式 |
|---|---|---|
| 全局 RPM | 每分钟总请求数 | Token Bucket |
| Per-Key RPM | 单个 API Key 的请求限制 | 滑动窗口 |
| Per-User TPM | 单用户每分钟 Token 消耗 | 令牌桶 + Token 计数 |
| Per-Model RPM | 单个模型的请求限制 | 与 Provider 限额对齐 |
| 并发数 | 同时在途请求数 | 信号量 |
class TokenBucketLimiter:
def __init__(self, rpm: int, tpm: int):
self.request_bucket = TokenBucket(
capacity=rpm, refill_rate=rpm / 60
)
self.token_bucket = TokenBucket(
capacity=tpm, refill_rate=tpm / 60
)
async def acquire(self, estimated_tokens: int) -> bool:
if not self.request_bucket.consume(1):
return False
if not self.token_bucket.consume(estimated_tokens):
self.request_bucket.refund(1)
return False
return True
三、语义缓存设计
传统缓存基于精确字符串匹配，但 LLM 场景中语义相同的不同措辞应该命中同一缓存。
3.1 架构
查询进入 → Embedding → 向量检索 (相似度 > 阈值?)
│
是 ────┼──── 否
│ │
返回缓存 调用 LLM
(省钱) 结果写入缓存
3.2 实现要点
class SemanticCache:
def __init__(
self,
similarity_threshold: float = 0.95,
ttl_seconds: int = 3600,
):
self.threshold = similarity_threshold
self.ttl = ttl_seconds
self.vectorstore = Redis(
index_name="semantic_cache",
embedding=OpenAIEmbeddings(model="text-embedding-3-small"),
)
async def get(self, prompt: str) -> str | None:
results = self.vectorstore.similarity_search_with_score(
prompt, k=1
)
if results and results[0][1] >= self.threshold:
entry = results[0][0]
if not self._is_expired(entry):
return entry.metadata["response"]
return None
async def set(self, prompt: str, response: str):
self.vectorstore.add_texts(
texts=[prompt],
metadatas=[{
"response": response,
"created_at": time.time(),
}],
)
3.3 缓存策略
| 场景 | 是否适合缓存 | 原因 |
|---|---|---|
| FAQ 查询 | 适合 (高命中) | 问题有限且重复率高 |
| RAG 问答 | 适合 (中命中) | 相似问题指向相同文档 |
| 代码生成 | 谨慎 | 细微差异可能导致错误代码 |
| 创意写作 | 不适合 | 用户期望多样化输出 |
| 实时数据 | 不适合 | 结果时效性强 |
四、API Key 管理
4.1 安全架构
┌──────────────────────────────────────────────┐
│ API Key 安全层级 │
├──────────────────────────────────────────────┤
│ │
│ 应用层: 使用内部 Gateway Token │
│ │ (不含任何 Provider Key) │
│ ▼ │
│ 网关层: Gateway Token → Provider Key 映射 │
│ │ Key 存储在 Vault/KMS │
│ ▼ │
│ Provider: 实际的 API Key │
│ │
│ 原则: │
│ - 应用代码中永远不出现 Provider Key │
│ - Key 轮转不影响应用服务 │
│ - 每个应用发放独立的 Gateway Token │
│ - 审计日志记录每次 Key 使用 │
└──────────────────────────────────────────────┘
4.2 Key 池化与轮转
class KeyPool:
"""多 Key 轮转，分散限流压力"""
def __init__(self, keys: list[ProviderKey]):
self.keys = keys
self.usage: dict[str, KeyUsage] = {}
def get_key(self, provider: str) -> str:
available = [
k for k in self.keys
if k.provider == provider
and not self._is_rate_limited(k)
and k.enabled
]
if not available:
raise NoAvailableKeyError(provider)
# 选择用量最低的 Key
return min(available, key=lambda k: self.usage[k.id].rpm_used)
五、使用计量与计费
5.1 计量数据模型
每次请求记录:
{
"request_id": "req_abc123",
"timestamp": "2026-02-27T10:30:00Z",
"user_id": "user_001",
"app_id": "chatbot_v2",
"model": "gpt-4o",
"provider": "openai",
"input_tokens": 1250,
"output_tokens": 380,
"cached": false,
"latency_ms": 2340,
"cost_usd": 0.0068,
"status": "success"
}
5.2 成本聚合
-- 按应用/日期聚合成本
SELECT
app_id,
DATE(timestamp) AS date,
SUM(input_tokens) AS total_input_tokens,
SUM(output_tokens) AS total_output_tokens,
SUM(cost_usd) AS total_cost,
COUNT(*) AS request_count,
AVG(latency_ms) AS avg_latency
FROM llm_usage_log
WHERE timestamp >= NOW() - INTERVAL '7 days'
GROUP BY app_id, DATE(timestamp)
ORDER BY total_cost DESC;
六、现有方案对比
| 维度 | LiteLLM | Portkey | Cloudflare AI Gateway |
|---|---|---|---|
| 部署方式 | 自托管 (开源) | SaaS + 自托管 | SaaS (Cloudflare) |
| Provider 数量 | 100+ | 200+ | 主流 ~15 |
| OpenAI 兼容 | 完整 | 完整 | 完整 |
| 语义缓存 | 内置 | 内置 | 内置 |
| 限流 | 内置 | 内置 | 内置 |
| Failover | 内置 | 内置 | 内置 |
| 计量/审计 | 内置 | 内置 (高级) | 内置 |
| 评估/Guardrails | 基础 | 高级 | 无 |
| 开源协议 | MIT | 部分开源 | 不适用 |
| 适合场景 | 全场景自托管 | 企业级 SaaS | CDN 用户 |
6.1 LiteLLM 快速部署
# litellm_config.yaml
model_list:
- model_name: "gpt-4o"
litellm_params:
model: "openai/gpt-4o"
api_key: "os.environ/OPENAI_API_KEY"
- model_name: "gpt-4o" # 同名 = 负载均衡
litellm_params:
model: "azure/gpt-4o-deployment"
api_key: "os.environ/AZURE_API_KEY"
api_base: "https://my-endpoint.openai.azure.com"
- model_name: "claude-sonnet"
litellm_params:
model: "anthropic/claude-sonnet-4-20250514"
api_key: "os.environ/ANTHROPIC_API_KEY"
router_settings:
routing_strategy: "latency-based-routing"
num_retries: 3
timeout: 30
fallbacks:
- {"gpt-4o": ["claude-sonnet"]}
litellm --config litellm_config.yaml --port 4000
七、生产架构参考
┌──────────────────────────────────────────────────────────────┐
│ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │Service A│ │Service B│ │Agent C │ Application Layer │
│ └────┬────┘ └────┬────┘ └────┬────┘ │
│ └────────────┼────────────┘ │
│ ▼ │
│ ┌────────────────────────────────┐ │
│ │ Load Balancer (Nginx/Envoy) │ │
│ └────────────┬───────────────────┘ │
│ ▼ │
│ ┌────────────────────────────────┐ ┌──────────────┐ │
│ │ AI Gateway (LiteLLM) │───→│ Redis │ │
│ │ - Auth + Rate Limit │ │ - Cache │ │
│ │ - Route + Failover │ │ - Rate State │ │
│ │ - Log + Meter │ └──────────────┘ │
│ └──────┬──────────┬──────────┬───┘ │
│ ▼ ▼ ▼ │
│ ┌──────────┐ ┌─────────┐ ┌──────────┐ │
│ │ OpenAI │ │Anthropic│ │Self-host │ Provider Layer │
│ └──────────┘ └─────────┘ │(vLLM) │ │
│ └──────────┘ │
│ │
│ ┌────────────────────────────────┐ │
│ │ PostgreSQL (usage_log) │ Persistence Layer │
│ │ + Grafana Dashboard │ │
│ └────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘
八、实施路线
| 阶段 | 目标 | 产出 | 周期 |
|---|---|---|---|
| Phase 1 | 统一入口 + Failover | LiteLLM 部署 + 基础路由 | 3 天 |
| Phase 2 | 计量 + 限流 | 用量仪表盘 + 预算告警 | 1 周 |
| Phase 3 | 语义缓存 | Redis 向量缓存 + 命中率监控 | 1 周 |
| Phase 4 | 高级路由 | A/B 测试 + 智能路由 + 成本优化 | 2 周 |
Maurice | maurice_wen@proton.me