负载均衡与多密钥策略
在多个 API 密钥和账户之间实现轮询与加权负载均衡，以扩大速率限制余量并减少 p99 延迟尖峰。
负载均衡与多密钥策略 是 CoddyKit 上的免费 AI Engineering Academy 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Engineering Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。
为什么一个 API 密钥不够
单个 OpenAI API 密钥具有固定的速率限制，以每分钟请求数（RPM）和每分钟令牌数（TPM）衡量。在第 1 级别，GPT-4o 允许 500 RPM 和 30,000 TPM。对于拥有数百名并发用户的生产应用，单个密钥会不断触及这些限制。多个 API 密钥可以按比例增加可用容量。
创建多个 API 密钥
您可以在同一个 OpenAI 组织中创建多个 API 密钥，也可以创建多个 OpenAI 账户（每个账户分别计费）。请将每个密钥存储在环境配置中，并将它们视为一个池。请使用 AWS Secrets Manager 或 HashiCorp Vault 等机密管理器保存密钥，不要将其放在源代码或提交到版本控制的 .env 文件中。
import os
API_KEYS = [
os.environ['OPENAI_KEY_1'],
os.environ['OPENAI_KEY_2'],
os.environ['OPENAI_KEY_3'],
os.environ['OPENAI_KEY_4'],
]
# Total effective RPM = 500 * 4 = 2000 RPM
# Total effective TPM = 30000 * 4 = 120000 TPM轮询负载均衡
轮询通过按顺序循环使用所有密钥，在它们之间均匀分配请求。它实现简单，并能确保随着时间推移，每个密钥处理的负载大致相同。请使用线程安全的计数器或原子整数，避免两个并发请求同时选择同一个密钥。当所有密钥具有相同的速率限制时，轮询效果良好。
import itertools
import threading
from openai import OpenAI
class RoundRobinPool:
def __init__(self, keys: list):
self._clients = [OpenAI(api_key=k) for k in keys]
self._cycle = itertools.cycle(range(len(self._clients)))
self._lock = threading.Lock()
def get_client(self) -> OpenAI:
with self._lock:
idx = next(self._cycle)
return self._clients[idx]
pool = RoundRobinPool(API_KEYS)
client = pool.get_client()加权负载均衡
加权负载均衡会根据密钥的容量，为速率限制更高的高等级密钥分配更大的流量份额。如果密钥 A 为第 3 级（10,000 RPM），密钥 B 为第 1 级（500 RPM），则密钥 A 应接收约 95% 的请求。当低等级密钥与高等级密钥混用时，加权均衡可以防止低等级密钥成为瓶颈。
import random
class WeightedPool:
def __init__(self, key_configs: list):
# key_configs = [{'key': '...', 'weight': 10}, ...]
self._clients = [OpenAI(api_key=c['key']) for c in key_configs]
self._weights = [c['weight'] for c in key_configs]
def get_client(self) -> OpenAI:
return random.choices(self._clients, weights=self._weights, k=1)[0]
pool = WeightedPool([
{'key': os.environ['OPENAI_KEY_TIER3'], 'weight': 20},
{'key': os.environ['OPENAI_KEY_TIER1'], 'weight': 1},
])跟踪每个密钥的速率限制状态
OpenAI API 会在每个响应中返回速率限制标头：x-ratelimit-remaining-requests 和 x-ratelimit-remaining-tokens。请按密钥跟踪这些标头，以了解哪些密钥即将耗尽。当某个密钥报告当前分钟剩余请求数少于 10 时，请暂时将流量转移到其他密钥，以便在 429 错误发生前避免它们。
class SmartPool:
def __init__(self, keys: list):
self._clients = [OpenAI(api_key=k) for k in keys]
self._remaining = {i: 500 for i in range(len(keys))} # initial RPM
def get_best_client(self):
# Pick key with most remaining capacity
best_idx = max(self._remaining, key=lambda i: self._remaining[i])
return self._clients[best_idx], best_idx
def update_remaining(self, idx: int, response_headers: dict):
remaining = int(response_headers.get('x-ratelimit-remaining-requests', 0))
self._remaining[idx] = remaining处理 429 速率限制错误
当某个密钥返回 429 错误时，请立即将该密钥从池中移除，持续时间以 Retry-After 标头指定的时长为准（通常为 60 秒）。将其标记为冷却中，并将所有流量转发到剩余密钥。冷却窗口结束后，再将该密钥恢复到池中。这样可以防止级联故障：对同一密钥的重试不会让情况进一步恶化。
import time
from openai import RateLimitError
class CooldownPool:
def __init__(self, keys: list):
self._clients = [(OpenAI(api_key=k), None) for k in keys] # (client, cooldown_until)
def get_available_clients(self):
now = time.time()
return [
(i, c) for i, (c, until) in enumerate(self._clients)
if until is None or until <= now
]
def mark_cooling(self, idx: int, retry_after: int = 60):
client, _ = self._clients[idx]
self._clients[idx] = (client, time.time() + retry_after)
print(f'Key {idx} cooling down for {retry_after}s')使用 OpenRouter 作为多路复用器
OpenRouter 是一种代理服务，通过单个兼容 OpenAI 的 API 端点提供数百个模型。通过 OpenRouter 路由请求，您可以自动获得多个底层提供商账户之间的负载均衡、备用提供商回退，以及将开源模型作为备份的能力。鉴于它带来的运维简便性，其成本加价幅度很小。
from openai import OpenAI
# OpenRouter uses the same OpenAI SDK interface
client = OpenAI(
api_key=os.environ['OPENROUTER_API_KEY'],
base_url='https://openrouter.ai/api/v1'
)
response = client.chat.completions.create(
model='openai/gpt-4o', # OpenRouter model name format
messages=[{'role': 'user', 'content': prompt}]
)
# Automatic failover if OpenAI is down通过指标监控密钥健康状况
请跟踪每个密钥的指标，包括已发送请求数、收到的 429 错误数，以及过去一小时内的冷却时间。429 比例较高的密钥需要减少流量或升级等级。请以 Prometheus 格式在 /metrics 端点公开这些指标，以便监控系统在任何密钥持续触及限制时发出警报。
from dataclasses import dataclass, field
from collections import defaultdict
@dataclass
class KeyMetrics:
requests_sent: int = 0
rate_limit_errors: int = 0
total_tokens_used: int = 0
cooldown_count: int = 0
class MetricPool:
def __init__(self, keys: list):
self._clients = [OpenAI(api_key=k) for k in keys]
self._metrics = [KeyMetrics() for _ in keys]
def report(self):
for i, m in enumerate(self._metrics):
error_rate = m.rate_limit_errors / max(m.requests_sent, 1)
print(f'Key {i}: {m.requests_sent} req, {error_rate:.1%} 429 rate')按地理区域分配密钥
如果您的用户分布在全球各地，请考虑为每个地理区域维护独立的 API 密钥，并将请求路由到距离用户最近的密钥。减少网络往返时间可以改善 TTFT。请在每个区域部署轻量级负载均衡器（AWS Lambda@Edge 或 Cloudflare Worker），由它选择适当的密钥并代理请求，从而避免将密钥暴露给客户端。
REGIONAL_KEYS = {
'us-east': os.environ['OPENAI_KEY_US_EAST'],
'eu-west': os.environ['OPENAI_KEY_EU_WEST'],
'ap-southeast': os.environ['OPENAI_KEY_AP'],
}
def get_key_for_region(user_region: str) -> str:
# Default to us-east if region unknown
return REGIONAL_KEYS.get(user_region, REGIONAL_KEYS['us-east'])测试负载均衡器
请编写一个负载测试，通过您的均衡池发起 100 个并发请求，并测量分配情况、错误率和延迟百分位数。验证没有任何单个密钥处理超出其按比例分配的份额，并确保 429 错误低于 0.1%。请使用 asyncio.gather 或 Locust 等工具，模拟生产系统实际会遇到的并发负载。
import asyncio
import time
async def load_test(pool, concurrency=100, total=1000):
sem = asyncio.Semaphore(concurrency)
results = []
async def one_request():
async with sem:
client = pool.get_client()
start = time.perf_counter()
try:
await client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': 'Ping'}],
max_tokens=5
)
results.append(('ok', time.perf_counter() - start))
except Exception as e:
results.append(('error', str(e)))
await asyncio.gather(*[one_request() for _ in range(total)])
ok = [r for r in results if r[0] == 'ok']
print(f'Success rate: {len(ok)/total:.1%}')
return results选择合适的均衡策略
请根据速率限制结构选择均衡策略。当所有密钥的等级限制相同且流量分布均匀时，使用轮询。当密钥的等级限制不同时，使用加权均衡。当您需要在突发流量下尽量减少 429 错误时，使用感知健康状况的路由（跳过接近耗尽的密钥）。对于大多数生产系统，结合指数退避的感知健康状况路由可以在简单性和韧性之间取得最佳平衡。
# Strategy selection guide:
# Scenario A: 4 keys all Tier 2 (same limits)
# -> Round-robin: simple, even distribution
#
# Scenario B: 1 Tier 3 key + 3 Tier 1 keys
# -> Weighted: Tier 3 gets 10x weight
#
# Scenario C: Variable traffic with burst periods
# -> Health-aware: track remaining headers, skip near-limit keys
#
# Scenario D: Multi-region, latency-sensitive
# -> Geographic: regional keys, route by user location快速检查
请检验您对 LLM API 负载均衡策略的理解。
课程回顾
在本课中，您学到了：轮询和加权均衡可以将流量分配到多个 API 密钥，从而增加速率限制容量；冷却跟踪通过暂时移除受到限制的密钥，防止 429 错误级联；OpenRouter提供了带自动回退功能的托管多路复用方案。接下来，我们将实现备用提供商和熔断器。
常见问题解答
「负载均衡与多密钥策略」课时是免费的吗？
是的 — 「负载均衡与多密钥策略」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Engineering Academy 课程的其余内容，请升级到 CoddyKit PRO。 AI Engineering Academy 课程共包含 4 节课。
「负载均衡与多密钥策略」这节课中我会学到什么？
在多个 API 密钥和账户之间实现轮询与加权负载均衡，以扩大速率限制余量并减少 p99 延迟尖峰。 你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Engineering Academy 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Engineering Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。
「负载均衡与多密钥策略」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Engineering Academy 课中编写并运行代码吗？
能。每节 AI Engineering Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。
此课程中的所有课时
- 衡量 LLM 延迟：TTFT 与 TPOT
- 负载均衡与多密钥策略
- 备用提供商与熔断器
- 超时预算与平稳降级