🧠 图解记忆:网关统一鉴权、配额、路由和计费,并用降级与观测守住模型依赖的不确定性;点击图片可查看原图。
💡 答案要点
题目理解:
LLM API 网关:
- 限流:防止用户打爆 API 配额
- 路由:多模型选择、成本优化
- 计费:按使用量收费,支持多租户
- 核心挑战:高并发、低延迟、可观测整体架构:
外部请求
↓
┌──────────────────────────────────────────────────────────┐
│ API Gateway │
│ 统一入口 │ TLS 终结 │ 请求日志 │
└──────────────────────────────────────────────────────────┘
↓
┌──────────────────────────────────────────────────────────┐
│ Auth & Rate Limit Layer │
│ API Key 验证 │ 令牌桶限流 │ 额度扣减 │
└──────────────────────────────────────────────────────────┘
↓
┌──────────────────────────────────────────────────────────┐
│ Model Router │
│ 意图分类 │ 成本优先/质量优先 │ 模型选择 │
└──────────────────────────────────────────────────────────┘
↓
┌──────────────────────────────────────────────────────────┐
│ LLM Provider Adapters │
│ OpenAI │ Anthropic │ Azure │ 国内模型 │ 自部署 │
└──────────────────────────────────────────────────────────┘
↓
┌──────────────────────────────────────────────────────────┐
│ Usage & Billing │
│ 用量记录 │ 计费规则 │ 成本分析 │
└──────────────────────────────────────────────────────────┘限流实现(令牌桶 + 多维度):
展开 Python 代码示例(58 行)
python
import time
import redis
import hashlib
class RateLimiter:
"""多维度限流:用户 + API Key + IP"""
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
async def check_limit(
self,
api_key: str,
requests_per_minute: int = 60,
tokens_per_minute: int = 60000,
expected_tokens: int = 1000,
) -> tuple[bool, str]:
"""检查限流,返回 (是否允许, 原因)"""
# 1. 请求频率限流(令牌桶)
req_key = f"rate:req:{api_key}"
req_allow = await self.redis.evalsha(
RATE_LIMIT_SCRIPT,
1, req_key,
requests_per_minute, 60, # 每分钟 N 个请求
time.time()
)
if not req_allow:
return False, "rate_limit:requests"
# 2. Token 限流
tok_key = f"rate:tok:{api_key}"
current_tokens = await self.redis.get(tok_key) or 0
if current_tokens + expected_tokens > tokens_per_minute:
return False, "rate_limit:tokens"
await self.redis.incrby(tok_key, expected_tokens)
await self.redis.expire(tok_key, 60) # 1分钟窗口
return True, "ok"
RATE_LIMIT_SCRIPT = """
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local current = redis.call('GET', key) or 0
if current >= limit then
return 0
end
redis.call('INCR', key)
redis.call('EXPIRE', key, window)
return 1
"""智能路由(成本 + 质量平衡):
展开 Python 代码示例(61 行)
python
class ModelRouter:
"""模型路由:根据请求特征选择最优模型"""
MODELS = {
# (max_tokens, quality_score, cost_per_1k)
"gpt-4o": (128000, 0.95, 0.015),
"gpt-4o-mini": (128000, 0.85, 0.0015),
"claude-3.5-sonnet": (200000, 0.93, 0.010),
"qwen-plus": (131072, 0.80, 0.004),
}
def route(
self,
request: ChatRequest,
strategy: str = "cost_quality_balance"
) -> str:
"""
路由策略:
- cost_first: 成本优先
- quality_first: 质量优先
- cost_quality_balance: 成本质量平衡
"""
if strategy == "cost_first":
return self._cheapest_model(request)
if strategy == "quality_first":
return self._best_quality_model(request)
# 默认:成本质量平衡
return self._balanced_model(request)
def _balanced_model(self, request: ChatRequest) -> str:
"""综合评分:质量分数 / 成本"""
# 简单问题 → 便宜模型
if self._is_simple(request):
return "gpt-4o-mini"
# 代码相关 → Claude(代码能力强)
if self._is_code_related(request):
return "claude-3.5-sonnet"
# 长上下文 → 支持大的模型
if request.history_length > 20:
return "claude-3.5-sonnet"
# 中文 + 简单 → 国内模型(成本低)
if request.language == "zh" and self._is_simple(request):
return "qwen-plus"
return "gpt-4o"
def _is_simple(self, request: ChatRequest) -> bool:
"""判断是否为简单请求"""
simple_patterns = ["是什么", "介绍一下", "定义", "时间", "地点"]
return any(p in request.prompt for p in simple_patterns)
def _is_code_related(self, request: ChatRequest) -> bool:
"""判断是否代码相关"""
code_patterns = ["代码", "函数", "Python", "Java", "bug", "debug"]
return any(p in request.prompt.lower() for p in code_patterns)计费系统(用量记录 + 扣费):
展开 Python 代码示例(52 行)
python
class BillingService:
"""按量计费服务"""
def __init__(self, db: Database, llm_router: ModelRouter):
self.db = db
self.router = llm_router
async def record_usage(self, api_key: str, request: ChatRequest, response: ChatResponse):
"""记录用量并扣费"""
# 计算费用
model = request.model or self.router.route(request)
_, quality, cost_per_1k = self.router.MODELS[model]
input_tokens = response.usage.input_tokens
output_tokens = response.usage.output_tokens
total_tokens = input_tokens + output_tokens
cost = (total_tokens / 1000) * cost_per_1k
# 写入用量记录
await self.db.execute("""
INSERT INTO usage_logs (api_key, model, input_tokens, output_tokens, cost, created_at)
VALUES ($1, $2, $3, $4, $5, NOW())
""", api_key, model, input_tokens, output_tokens, cost)
# 扣减账户余额
await self.db.execute("""
UPDATE accounts
SET balance = balance - $1
WHERE api_key = $2 AND balance >= $1
""", cost, api_key)
async def get_cost_breakdown(self, api_key: str, start_date: datetime, end_date: datetime) -> dict:
"""获取成本分析"""
rows = await self.db.fetch("""
SELECT
model,
SUM(input_tokens) as total_input,
SUM(output_tokens) as total_output,
SUM(cost) as total_cost,
COUNT(*) as request_count
FROM usage_logs
WHERE api_key = $1 AND created_at BETWEEN $2 AND $3
GROUP BY model
""", api_key, start_date, end_date)
return {
"total_cost": sum(r["total_cost"] for r in rows),
"by_model": [dict(r) for r in rows],
"avg_cost_per_request": sum(r["total_cost"] for r in rows) / max(sum(r["request_count"] for r in rows), 1)
}多租户计费方案:
| 方案 | 说明 | 适用场景 |
|---|---|---|
| 预付费套餐 | 买 token 包,用完为止 | 个人用户 |
| 后付费月结 | 按月结算,月底扣费 | 企业客户 |
| 信用额度 | 设置信用额度,超额熔断 | 大客户 |
| 用量分层 | 阶梯定价,用量越大越便宜 | 大客户 |
面试话术:
"LLM API 网关的三大核心能力:限流、路由、计费。限流用令牌桶实现,Redis+Lua 脚本原子操作,支持请求数和 Token 数双维度限流。路由的核心是'让对的模型处理对的任务'——简单问答用 mini 模型(省 10 倍成本),代码生成用 Claude(代码能力强),长上下文用 Claude(200K)。计费用预写日志(WAL)+ 异步扣费,保证数据一致性。面试能说清楚这三件套的实现细节,说明你有 API 服务端开发经验。"
