Équilibrage de charge et stratégies multi-clés

Why One API Key Is Not Enough

Creating Multiple API Keys

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

Round-Robin Load Balancing

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()

Weighted Load Balancing

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},
])

Tracking Per-Key Rate Limit State

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

Handling 429 Rate Limit Errors

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')

Using OpenRouter as a Multiplexer

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

Monitoring Key Health with 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')

Geographic Key Distribution

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'])

Testing Your Load Balancer

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

Choosing the Right Balancing Strategy

# 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

Quick Check

Lesson Recap