Il y a trois semaines, à 2 h 47 du matin, mon téléphone a vibré en mode silencieux. Notre pipeline de génération de rapports juridiques, qui brassait 1 200 requêtes/min vers Claude Opus, venait de s'effondrer. Le monitoring affichait 429 Too Many Requests en boucle, et le backlog client atteignait 18 000 messages. En ouvrant les logs, j'ai compris : nous avions dépassé le quota RPM du fournisseur et personne n'avait prévu de repli. C'est ce week-end-là, après 36 h de debug, que j'ai écrit le scheduler que je vous partage aujourd'hui.
Dans cet article, je vous montre comment j'ai conçu un répartiteur qui route les requêtes vers GPT-5.5 ou Claude Opus 4.7 selon le contexte, le coût et la latence — le tout en passant par l'API unifiée de S'inscrire ici pour bénéficier d'une facturation locale et d'une latence intra-région inférieure à 50 ms.
Pourquoi l'équilibrage de charge devient indispensable
Dès que vous dépassez 50 requêtes concurrentes sur un seul modèle, trois problèmes apparaissent :
- Quota RPM/TPM atteint : le fournisseur renvoie
429et votre SLA explose. - Coût non maîtrisé : un même prompt envoyé à un modèle premium au lieu d'un modèle léger peut multiplier la facture par 7.
- Latence en queue de file : sans orchestration, les requêtes lentes bloquent les rapides.
La solution : répartir intelligemment vers deux modèles phares — GPT-5.5 (rapide, polyvalent, 600 RPM) et Claude Opus 4.7 (profond, raisonnement long, 200 K de contexte) — en s'appuyant sur une couche d'abstraction unique.
Architecture du scheduler
Le répartiteur suit quatre règles de routage :
- Type de tâche : raisonnement multi-étapes → Opus 4.7 ; extraction, reformulation, JSON strict → GPT-5.5.
- Budget restant : si le quota Opus est à 80 %, bascule automatique vers GPT-5.5.
- Taille du contexte : au-delà de 60 000 tokens, Opus 4.7 (200 K) prend le relais.
- Latence observée : si p95 dépasse 1 200 ms sur Opus, repli temporaire vers Sonnet 4.5.
Implémentation en Python
Voici le répartiteur que j'ai mis en production. Il utilise httpx asynchrone, un token-bucket par modèle et un sémaphore pour borner la concurrence.
import asyncio
import httpx
import time
import random
from dataclasses import dataclass, field
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
BASE_URL = "https://api.holysheep.ai/v1"
@dataclass
class ModelProfile:
name: str
rpm_limit: int
input_cost: float # dollars par million de tokens
output_cost: float
p95_latency_ms: int
PROFILES = {
"gpt-5.5": ModelProfile("gpt-5.5", 600, 3.00, 12.00, 820),
"claude-opus-4.7": ModelProfile("claude-opus-4.7", 400, 5.00, 25.00, 1450),
"claude-sonnet-4.5": ModelProfile("claude-sonnet-4.5", 800, 3.00, 15.00, 690),
}
@dataclass
class TokenBucket:
capacity: int
tokens: float = field(init=False)
last_refill: float = field(init=False)
refill_rate: float # tokens par seconde
def __post_init__(self):
self.tokens = self.capacity
self.last_refill = time.monotonic()
def consume(self, amount: float = 1.0) -> bool:
now = time.monotonic()
self.tokens = min(self.capacity, self.tokens + (now - self.last_refill) * self.refill_rate)
self.last_refill = now
if self.tokens >= amount:
self.tokens -= amount
return True
return False
buckets = {m: TokenBucket(capacity=p.rpm_limit, refill_rate=p.rpm_limit / 60.0) for m, p in PROFILES.items()}
def choose_model(task_type: str, context_tokens: int, latency_budget_ms: int = 2000) -> str:
if context_tokens > 60_000 or task_type == "deep_reasoning":
candidate = "claude-opus-4.7