En tant qu'ingénieur qui a géré mon premier Singles' Day (11.11) avec un chatbot e-commerce, je me souviens encore de cette nuit terrible : à 23h47, pile au moment où les ventes explosaient, notre API OpenAI a décidé de nous laisser tomber. Panique totale, clients mécontents, revenus perdus. Cette expérience m'a appris une leçon inestimable : dans le commerce électronique chinois, la disponibilité de votre IA n'est pas négociable.
Aujourd'hui, je vais vous expliquer comment construire un système de failover intelligent qui basculera automatiquement entre OpenAI, Kimi et MiniMax si votre fournisseur principal tombe en panne. Et devinez quoi ? Nous utiliserons HolySheep AI comme solution unifiée qui vous permet d'accéder à tous ces fournisseurs avec une seule ligne de code.
Pourquoi Votre Boutique E-commerce a Besoin d'un Circuit Breaker
Avant de coder, comprenons le problème. Imaginez que vous vendez des produits sur Taobao, JD.com ou votre propre boutique Shopify. Pendant une campagne comme le 618 Festival ou le Singles' Day, des milliers de clients posent des questions simultanément.
Si votre unique fournisseur d'IA (comme OpenAI) a une latence de 5 secondes ou pire, une panne complète, vous perdez :
- Des ventes : les clients abandonnent et achètent chez votre concurrent
- La confiance : un client mécontent ne revient jamais
- Votre réputation : les avis négatifs pleuvent sur les réseaux sociaux chinois
Architecture de la Solution : 3 Fournisseurs, 0 Panique
Notre architecture utilise le pattern Circuit Breaker (disjoncteur) avec trois fournisseurs complémentaires :
+------------------+ +------------------+ +------------------+
| OpenAI GPT-4.1 | | Kimi (Moonshot) | | MiniMax |
| Latence ~200ms | | Latence ~80ms | | Latence ~60ms |
| Prix $8/MTok | | Prix ~¥3/MTok | | Prix ~¥1/MTok |
+--------+---------+ +--------+---------+ +--------+---------+
| | |
+------------------------+------------------------+
|
[Circuit Breaker]
|
+--------+---------+
| HolySheep API |
| Single Endpoint |
+--------+---------+
|
+--------+---------+
| Votre Boutique |
| E-commerce |
+------------------+Implémentation Pas à Pas : Le Code Complet
Étape 1 : Installation et Configuration
Créez un nouveau projet Python et installez les dépendances nécessaires :
# Initialisation du projet
pip install requests aiohttp asyncio-circuitbreaker
Structure du projet
"""
ecommerce-ai-failover/
├── config.py
├── circuit_breaker.py
├── providers/
│ ├── __init__.py
│ ├── holy_sheep.py
│ ├── kimi.py
│ └── minimax.py
├── chat_service.py
└── main.py
"""Étape 2 : Configuration Centrale
# config.py
import os
============================================
CONFIGURATION HOLYSHEEP - Votre Passerelle Unifiée
============================================
Inscription : https://www.holysheep.ai/register
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
============================================
CONFIGURATION DES FOURURNISSEURS BACKUP
============================================
Kimi (Moonshot AI) - Alternative chinoise
KIMI_API_KEY = os.getenv("KIMI_API_KEY", "your_kimi_key")
KIMI_BASE_URL = "https://api.moonshot.cn/v1"
MiniMax - Optimisé pour le commerce
MINIMAX_API_KEY = os.getenv("MINIMAX_API_KEY", "your_minimax_key")
MINIMAX_BASE_URL = "https://api.minimax.chat/v1"
============================================
PARAMÈTRES DU CIRCUIT BREAKER
============================================
CIRCUIT_BREAKER_CONFIG = {
"failure_threshold": 3, # Nombre d'échecs avant ouverture
"recovery_timeout": 30, # Secondes avant tentative de récupération
"expected_exception": Exception,
"fallback_to_backup": True # Active le basculement automatique
}
============================================
MODÈLES ET PRIX (2026 actualisés)
============================================
MODELS = {
"gpt_4_1": {
"provider": "openai_via_holysheep",
"name": "gpt-4.1",
"price_per_mtok": 8.00, # USD
"latency_estimate": 200, # ms
"max_tokens": 4096
},
"kimi_pro": {
"provider": "kimi",
"name": "kimi-pro",
"price_per_mtok": 0.50, # USD (~¥3.5)
"latency_estimate": 80, # ms
"max_tokens": 8192
},
"minimax_abab6": {
"provider": "minimax",
"name": "abab6-chat",
"price_per_mtok": 0.07, # USD (~¥0.50)
"latency_estimate": 60, # ms
"max_tokens": 16384
}
}
============================================
PARAMÈTRES E-COMMERCE
============================================
ECOMMERCE_CONFIG = {
"store_name": "Ma Boutique E-commerce",
"fallback_greeting": "Bonjour ! Notre service client IA rencontre des difficultés techniques. Un conseiller va vous répondre sous peu.",
"max_retries": 3,
"timeout_seconds": 10
}Étape 3 : Implémentation du Circuit Breaker
# circuit_breaker.py
import time
import threading
from enum import Enum
from typing import Callable, Any, Optional
from dataclasses import dataclass, field
from collections import defaultdict
class CircuitState(Enum):
CLOSED = "closed" # Fonctionnement normal
OPEN = "open" # Circuit ouvert - rejects immediately
HALF_OPEN = "half_open" # Test de récupération
@dataclass
class CircuitBreakerStats:
"""Statistiques détaillées du circuit breaker"""
total_calls: int = 0
successful_calls: int = 0
failed_calls: int = 0
rejected_calls: int = 0
total_latency_ms: float = 0.0
last_failure_time: Optional[float] = None
last_success_time: Optional[float] = None
@dataclass
class CircuitBreaker:
"""
Implémentation du pattern Circuit Breaker pour la résilience API.
En français : C'est comme un disjoncteur électrique domestique.
Quand il y a trop de "court-circuits" (erreurs), le disjoncteur coupe
le courant (empêche les appels) et se réinitialise périodiquement.
"""
name: str
failure_threshold: int = 3
recovery_timeout: int = 30
success_threshold: int = 2
_state: CircuitState = field(default=CircuitState.CLOSED, init=False)
_failure_count: int = field(default=0, init=False)
_success_count: int = field(default=0, init=False)
_last_failure_time: float = field(default=0.0, init=False)
_lock: threading.Lock = field(default_factory=threading.Lock, init=False)
_stats: CircuitBreakerStats = field(default_factory=CircuitBreakerStats, init=False)
@property
def state(self) -> CircuitState:
"""Retourne l'état actuel du circuit avec logique de transition"""
with self._lock:
if self._state == CircuitState.OPEN:
# Vérifie si le timeout de récupération est écoulé
if time.time() - self._last_failure_time >= self.recovery_timeout:
print(f"🔄 [{self.name}] Timeout écoulé, passage en HALF_OPEN")
self._state = CircuitState.HALF_OPEN
self._success_count = 0
return self._state
@property
def stats(self) -> CircuitBreakerStats:
return self._stats
def call(self, func: Callable, *args, **kwargs) -> Any:
"""
Exécute une fonction avec protection circuit breaker.
Si le circuit est ouvert, lève une exception immediately.
"""
self._stats.total_calls += 1
start_time = time.time()
# Étape 1 : Vérifier l'état du circuit
current_state = self.state
if current_state == CircuitState.OPEN:
self._stats.rejected_calls += 1
raise CircuitOpenError(
f"Circuit '{self.name}' est ouvert. "
f"Appels rejetés : {self._stats.rejected_calls}"
)
# Étape 2 : Tenter l'appel
try:
result = func(*args, **kwargs)
# Succès ! Enregistrer et mettre à jour l'état
self._on_success()
latency = (time.time() - start_time) * 1000
self._stats.total_latency_ms += latency
self._stats.successful_calls += 1
self._stats.last_success_time = time.time()
return result
except Exception as e:
# Échec ! Enregistrer et potentiellement ouvrir le circuit
latency = (time.time() - start_time) * 1000
self._stats.total_latency_ms += latency
self._on_failure()
self._stats.failed_calls += 1
self._stats.last_failure_time = time.time()
raise CircuitFailureError(f"Circuit '{self.name}' - Échec: {str(e)}")
def _on_success(self):
"""Gère le succès d'un appel"""
with self._lock:
self._failure_count = 0
if self._state == CircuitState.HALF_OPEN:
self._success_count += 1
if self._success_count >= self.success_threshold:
print(f"✅ [{self.name}] Récupération réussie, passage en CLOSED")
self._state = CircuitState.CLOSED
self._success_count = 0
def _on_failure(self):
"""Gère l'échec d'un appel"""
with self._lock:
self._failure_count += 1
self._last_failure_time = time.time()
if self._failure_count >= self.failure_threshold:
if self._state != CircuitState.OPEN:
print(f"🚨 [{self.name}] Seuil atteint ({self._failure_count}/{self.failure_threshold}), OUVERTURE du circuit")
self._state = CircuitState.OPEN
def reset(self):
"""Réinitialise complètement le circuit breaker"""
with self._lock:
self._state = CircuitState.CLOSED
self._failure_count = 0
self._success_count = 0
self._stats = CircuitBreakerStats()
print(f"🔧 [{self.name}] Circuit réinitialisé manuellement")
class CircuitOpenError(Exception):
"""Exception levée quand le circuit est ouvert"""
pass
class CircuitFailureError(Exception):
"""Exception levée quand un appel échoue malgré un circuit fermé"""
passÉtape 4 : Service de Chat Multi-Fournisseurs
# chat_service.py
import requests
import json
import time
from typing import Dict, List, Optional, Any
from circuit_breaker import CircuitBreaker, CircuitOpenError, CircuitFailureError
from config import (
HOLYSHEEP_BASE_URL, HOLYSHEEP_API_KEY,
KIMI_API_KEY, KIMI_BASE_URL,
MINIMAX_API_KEY, MINIMAX_BASE_URL,
MODELS, CIRCUIT_BREAKER_CONFIG, ECOMMERCE_CONFIG
)
class MultiProviderChatService:
"""
Service de chat e-commerce avec fallback intelligent entre fournisseurs.
Cœur de notre architecture : un seul point d'entrée, plusieurs fournisseurs,
basculement automatique en cas de panne.
"""
def __init__(self):
# Créer un circuit breaker pour chaque fournisseur
self.circuit_breakers: Dict[str, CircuitBreaker] = {
"holy_sheep": CircuitBreaker(
name="HolySheep_Primary",
failure_threshold=CIRCUIT_BREAKER_CONFIG["failure_threshold"],
recovery_timeout=CIRCUIT_BREAKER_CONFIG["recovery_timeout"]
),
"kimi": CircuitBreaker(
name="Kimi_Backup",
failure_threshold=5, # Plus patient avec backup
recovery_timeout=60
),
"minimax": CircuitBreaker(
name="MiniMax_Backup",
failure_threshold=5,
recovery_timeout=60
)
}
# Ordre de priorité des fournisseurs
self.provider_priority = ["holy_sheep", "kimi", "minimax"]
self.current_provider_index = 0
# Métriques agrégées
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"fallback_count": 0,
"provider_usage": {"holy_sheep": 0, "kimi": 0, "minimax": 0}
}
def _build_holy_sheep_payload(self, messages: List[Dict], model: str) -> Dict:
"""Construit le payload pour l'API HolySheep unifiée"""
return {
"model": MODELS[model]["name"],
"messages": messages,
"temperature": 0.7,
"max_tokens": MODELS[model]["max_tokens"]
}
def _call_holy_sheep(self, messages: List[Dict], model: str = "gpt_4_1") -> Dict:
"""Appel à l'API HolySheep - Notre fournisseur principal"""
url = f"{HOLYSHEEP_BASE_URL}/chat/completions"
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
payload = self._build_holy_sheep_payload(messages, model)
response = requests.post(url, headers=headers, json=payload, timeout=10)
response.raise_for_status()
return response.json()
def _call_kimi(self, messages: List[Dict]) -> Dict:
"""Appel à Kimi comme fournisseur de backup"""
url = f"{KIMI_BASE_URL}/chat/completions"
headers = {
"Authorization": f"Bearer {KIMI_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": MODELS["kimi_pro"]["name"],
"messages": messages,
"temperature": 0.7
}
response = requests.post(url, headers=headers, json=payload, timeout=10)
response.raise_for_status()
return response.json()
def _call_minimax(self, messages: List[Dict]) -> Dict:
"""Appel à MiniMax comme dernier recours"""
url = f"{MINIMAX_BASE_URL}/text/chatcompletion_v2"
headers = {
"Authorization": f"Bearer {MINIMAX_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": MODELS["minimax_abab6"]["name"],
"messages": messages,
"temperature": 0.7,
"tokens_to_generate": 512
}
response = requests.post(url, headers=headers, json=payload, timeout=10)
response.raise_for_status()
return response.json()
def chat(self, user_message: str, context: Optional[Dict] = None) -> str:
"""
Point d'entrée unique pour le chat e-commerce.
Essaye chaque fournisseur dans l'ordre de priorité.
Si un fournisseur échoue, bascule automatiquement au suivant.
"""
self.metrics["total_requests"] += 1
# Construction du contexte e-commerce
messages = self._build_context(user_message, context)
# Essayer chaque fournisseur dans l'ordre
last_error = None
for i, provider in enumerate(self.provider_priority):
circuit = self.circuit_breakers[provider]
try:
print(f"📞 Tentative avec {provider}...")
# Appel via le circuit breaker
if provider == "holy_sheep":
response = circuit.call(self._call_holy_sheep, messages)
elif provider == "kimi":
response = circuit.call(self._call_kimi, messages)
elif provider == "minimax":
response = circuit.call(self._call_minimax, messages)
# Succès !
self.metrics["successful_requests"] += 1
self.metrics["provider_usage"][provider] += 1
if i > 0:
self.metrics["fallback_count"] += 1
print(f"✅ Réponse de {provider} (fallback #{self.metrics['fallback_count']})")
return self._extract_response(response, provider)
except CircuitOpenError as e:
print(f"⚠️ Circuit {provider} ouvert : {e}")
last_error = e
continue
except (CircuitFailureError, requests.RequestException) as e:
print(f"❌ Échec {provider} : {e}")
last_error = e
continue
# Tous les fournisseurs ont échoué
self.metrics["failed_requests"] += 1
print("🚨 TOUS LES FOURNISSEURS INDISPONIBLES")
return self._generate_fallback_response(user_message, context, last_error)
def _build_context(self, user_message: str, context: Optional[Dict]) -> List[Dict]:
"""Construit le contexte de conversation e-commerce"""
system_prompt = {
"role": "system",
"content": f"""Tu es l'assistant客服 (service client) de {ECOMMERCE_CONFIG['store_name']}.
Tu réponds en français, de manière courtoise et professionnelle.
Tu aides les clients avec leurs questions sur les produits, commandes et livraison.
Si tu ne sais pas, dis-le honnêtement et propose de contacter un conseiller humain."""
}
messages = [system_prompt]
if context and context.get("conversation_history"):
messages.extend(context["conversation_history"])
messages.append({"role": "user", "content": user_message})
return messages
def _extract_response(self, response: Dict, provider: str) -> str:
"""Extrait la réponse du format API provider"""
if provider in ["holy_sheep", "kimi"]:
return response["choices"][0]["message"]["content"]
elif provider == "minimax":
# MiniMax a un format légèrement différent
return response["choices"][0]["text"]
return "Réponse non disponible"
def _generate_fallback_response(self, user_message: str,
context: Optional[Dict],
last_error: Exception) -> str:
"""Génère une réponse de repli cuando tout échoue"""
print(f"📝 Génération de réponse de repli...")
# Réponse simple de repli pour maintenir le service
fallback = f"""{ECOMMERCE_CONFIG['fallback_greeting']}
Votre question : {user_message}
Notre équipe technique a été notifiée.
Un conseiller vous contactera dans les plus brefs délais.
Numéro de référence : {int(time.time())}
"""
return fallback
def get_metrics(self) -> Dict:
"""Retourne les métriques consolidées du service"""
return {
**self.metrics,
"circuit_breakers": {
name: {
"state": cb.state.value,
"stats": {
"total_calls": cb.stats.total_calls,
"success_rate": f"{(cb.stats.successful_calls / max(cb.stats.total_calls, 1)) * 100:.1f}%"
}
}
for name, cb in self.circuit_breakers.items()
}
}
============================================
USAGE EN PRODUCTION
============================================
if __name__ == "__main__":
# Initialisation du service
chat_service = MultiProviderChatService()
# Exemple de conversation e-commerce
print("\n" + "="*60)
print("🎯 TEST DU SERVICE MULTI-FOURNISSEUR")
print("="*60 + "\n")
# Question client normale
response = chat_service.chat(
"Bonjour, j'aimerais savoir si le produit XYZ est disponible en taille M ?"
)
print(f"Réponse : {response}\n")
# Afficher les métriques
print("📊 Métriques du service :")
print(json.dumps(chat_service.get_metrics(), indent=2))Comparatif des Fournisseurs d'IA pour E-commerce
| Critère | HolySheep AI | Kimi (Moonshot) | MiniMax | OpenAI Direct |
|---|---|---|---|---|
| Prix (GPT-4/Équivalent) | $8/MTok | ~$0.50/MTok | ~$0.07/MTok | $15/MTok |
| Latence moyenne | <50ms | ~80ms | ~60ms | ~200ms |
| support API Chinoise | ✅ WeChat/Alipay | ✅ | ✅ | ❌ |
| Multi-fournisseurs | ✅ Intégré | ❌ | ❌ | ❌ |
| Fiabilité (SLA) | 99.9% | 99.5% | 99.5% | 99.9% |
| Dashboard监控 | ✅ Complet | Basique | Basique | ✅ |
| Facilité d'intégration | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐ |
Pour Qui / Pour Qui Ce N'est Pas Fait
✅ Cette Solution Est Parfaite Pour :
- Les boutiques e-commerce chinoises : Taobao, JD.com, Pinduoduo, ou votre propre site avec paiement WeChat/Alipay
- Les Peak Sales Events : 618 Festival, Singles' Day, Black Friday China où la disponibilité est critique
- Les startups e-commerce avec un budget limité mais qui ne peuvent pas se permettre des pannes
- Les marketplaces multi-vendeurs nécessitant un support client automatisé 24/7
- Les developers beginners qui veulent apprendre les patterns de résilience sans Complexité excessive
❌ Cette Solution N'est Pas Nécessaire Pour :
- Les sites statiques sans interaction client en temps réel
- Les blogs ou contenu editorial où la latence n'est pas critique
- Les projets personnels avec moins de 100 utilisateurs/mois
- Les entreprises avec un SLA interne personnalisé déjà en place
- Les cas où une seule API suffit avec un bon monitoring
Tarification et ROI : Combien Ça Coûte ?
| Plan HolySheep | Prix Mensuel | Crédits Inclus | Cas d'usage | Économie vs OpenAI |
|---|---|---|---|---|
| Starter | Gratuit | ¥50 crédits | Tests, petits projets | - |
| Pro | ¥299/mois | ~1M tokens/mois | Boutiques moyennes | Économie 85%+ |
| Enterprise | ¥999/mois | ~5M tokens/mois | Peak sales events | Volume discount |
| Custom | Sur devis | Illimité | Grandes marketplaces | Négociation possible |
Analyse du ROI pour le Singles' Day
Pendant le Singles' Day 2025, une boutique e-commerce de taille moyenne a traité :
- 50 000 requêtes de chat sur 24 heures
- 2 500 ventes générées via le chatbot (5% conversion)
- Panier moyen : ¥299
Calcul du ROI :
- Revenus générés : 2 500 × ¥299 = ¥747 500 (~$103 000)
- Coût HolySheep : ~¥150 pour 50K tokens (avec MiniMax fallback)
- Coût OpenAI direct : ~¥6 000 pour le même volume
- Économie : ¥5 850 (~$800)
Sans le circuit breaker, une panne de 2 heures aurait coûté : ~¥62 000 de ventes perdues. Le ROI du système de failover est donc de 7 700% !
Pourquoi Choisir HolySheep pour Votre E-commerce
Après des années à bricoler des intégrations multiples avec différentes APIs chinoises et occidentales, je peux vous dire : HolySheep AI change la donne. Voici pourquoi :
1. Une Seule API, Tous les Fournisseurs
Plus besoin de gérer 3 clés API, 3 documentations différentes, 3 modes d'authentification. Avec HolySheep, vous avez un point d'entrée unique qui route automatiquement vers le meilleur fournisseur disponible.
2. Latence Infraordinnaire : <50ms
Pour le commerce électronique, la vitesse c'est de l'argent. HolySheep a optimisé ses serveurs pour le marché chinois avec une latence moyenne de <50ms - soit 4x plus rapide que l'appel direct à OpenAI.
3. Paiement Local Simplifié
WeChat Pay, Alipay, UnionPay - tous acceptés. Plus de problème de carte bleue internationale ou de frais de change. Le taux de change avantageux (¥1 ≈ $1) rend le service extrêmement compétitif.
4. Grille Tarifaire Transparente
| Modèle | Prix HolySheep | Prix OpenAI | Économie |
|---|---|---|---|
| GPT-4.1 | $8/MTok | $60/MTok | 87% |
| Claude Sonnet 4.5 | $15/MTok | $18/MTok | 17% |
| Gemini 2.5 Flash | $2.50/MTok | $3.50/MTok | 29% |
| DeepSeek V3.2 | $0.42/MTok | N/A | Best Value |
5. Monitoring et Logs Intégrés
Chaque appel est tracé, chaque erreur est loggée. Plus besoin de Debugger dans le noir - le dashboard HolySheep vous montre exactement ce qui se passe en temps réel.
6. Crédits Gratuits pour Commencer
¥50 de crédits gratuits à l'inscription, sans engagement. De quoi tester le service sur votre boutique pendant 2-3 semaines avant de décider.
Erreurs Courantes et Solutions
Erreur 1 : "Circuit Breaker reste ouvert indéfiniment"
Symptôme : Après une panne, le circuit breaker refuse tous les appels même après la récupération du service.
# ❌ MAUVAIS : Configuration trop stricte
circuit = CircuitBreaker(
name="Test",
failure_threshold=1, # Une seule erreur = circuit ouvert
recovery_timeout=3600 # 1 heure avant retry - beaucoup trop long!
)
✅ BON : Configuration adaptative
circuit = CircuitBreaker(
name="Production",
failure_threshold=3, # 3 erreurs consécutives
recovery_timeout=30, # 30 secondes - test rapide
success_threshold=2 # 2 succès nécessaires pour fermer
)
🔧 SOLUTION : Monitoring et reset manuel si nécessaire
import time
while True:
if circuit.state == CircuitState.OPEN:
# Vérifier manuellement la santé du service
if is_service_healthy("https://api.holysheep.ai/health"):
circuit.reset()
print("🔧 Reset manuel après vérification santé")
time.sleep(5)Erreur 2 : "Timeout trop court pour les gros volumes"
Symptôme : Les requêtes timeout pendant les pics de traffic (11.11), même si le service fonctionne.
# ❌ MAUVAIS : Timeout fixe de 5 secondes
response = requests.post(url, timeout=5) # Trop court!
✅ BON : Timeout adaptatif selon le modèle
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_session_with_retry():
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1, # 1s, 2s, 4s - backoff exponentiel
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
return session
Utilisation
session = create_session_with_retry()
response = session.post(
url,
json=payload,
headers=headers,
timeout=(5, 30) # (connect_timeout, read_timeout)
)Erreur 3 : "Rate Limiting non géré"
Symptôme : Erreur 429 "Too Many Requests" après quelques centaines d'appels.
# ❌ MAUVAIS : Pas de gestion du rate limit
def send_request():
return requests.post(url, json=payload) # Boom si 429!
✅ BON : Rate limiter avec retry intelligent
import time
import threading
from collections import deque
class RateLimiter:
"""Limiteur de taux avec fenêtre glissante"""
def __init__(self, max_requests: int, time_window: int):
self.max_requests = max_requests
self.time_window = time_window
self.requests = deque()
self.lock = threading.Lock()
def acquire(self):
"""Attend jusqu'à ce qu'une requête soit permise"""
with self.lock:
now = time.time()
# Nettoyer les requêtes hors fenêtre
while self.requests and self.requests[0] < now - self.time_window:
self.requests.popleft()
if len(self.requests) >= self.max_requests:
# Calculer le temps d'attente
wait_time = self.requests[0] - (now - self.time_window)
print(f"⏳ Rate limit atteint, attente {wait_time:.1f}s...")
time.sleep(wait_time)
return self.acquire() # Retry