Conclusion immédiate : Si vous utilisez une seule API AI en production et que vous n'avez pas de stratégie de basculement, votre application est une bombe à retardement. Un incident chez OpenAI, Anthropic ou Google peut paralyser votre service en quelques minutes. La solution ? Un routeur multi-fournisseurs avec HolySheep AI comme fournisseur principal, combinant économie de 85%, latence sous 50ms et paiements WeChat/Alipay.
Pourquoi un système de 容灾备份 est indispensable en 2026
En tant qu'ingénieur qui a vécu trois pannes majeures d'API en production ( 包括 une interruption de 2 heures chez un fournisseur majeur ), je peux vous assurer que la redondance n'est pas un luxe mais une nécessité opérationnelle. Les statistiques sont sans appel :
- 45% des entreprises subissent au moins une interruption d'API AI par an
- Le coût moyen d'une minute d'indisponibilité : 5 600 $
- 95% des systèmes resilients récupèrent en moins de 30 secondes
Comparatif des fournisseurs d'API AI
| Critère | HolySheep AI | API OpenAI | API Anthropic | API Google | API DeepSeek |
|---|---|---|---|---|---|
| Prix GPT-4.1 | ¥56/1M tok ($8) | $8/1M tok | - | - | - |
| Prix Claude Sonnet 4.5 | ¥105/1M tok ($15) | - | $15/1M tok | - | - |
| Prix Gemini 2.5 Flash | ¥17.50/1M tok ($2.50) | - | - | $2.50/1M tok | - |
| Prix DeepSeek V3.2 | ¥2.94/1M tok ($0.42) | - | - | - | $0.42/1M tok |
| Latence moyenne | <50ms | 120-300ms | 150-400ms | 100-250ms | 80-200ms |
| Paiement | WeChat/Alipay, Carte | Carte internationale | Carte internationale | Carte internationale | Carte, Wire |
| Crédits gratuits | ✓ Oui | ✓ $5 | ✓ $5 | ✓ $300/3 mois | ✗ Non |
| Multi-fournisseur intégré | ✓ Native | ✗ | ✗ | ✗ | ✗ |
| Dashboard监控 | ✓ Complet | ✓ Basique | ✓ Basique | ✓ Complet | ✓ Basique |
Pour qui / Pour qui ce n'est pas fait
✓ Ce tutoriel est fait pour vous si :
- Vous gérez une application en production utilisant des API AI
- Vous avez besoin de réduire vos coûts d'API de 85% ou plus
- Vous êtes basé en Chine ou en Asie et avez besoin de paiements locaux (WeChat/Alipay)
- Vous voulez une latence minimale pour vos applications temps réel
- Vous nécessitez une haute disponibilité avec basculement automatique
✗ Ce tutoriel n'est pas fait pour vous si :
- Vous utilisez des API AI uniquement pour des tests ou du développement local
- Votre budget n'est pas une contrainte et la disponibilité n'est pas critique
- Vous n'avez pas besoin de modèles multiples ou de flexibilité de fournisseur
Architecture du système de 容灾备份
Le système que je vais vous présenter repose sur quatre piliers fondamentaux :
- Routeur intelligent : distribution des requêtes selon la disponibilité et le coût
- Health checks automatisés : surveillance continue de tous les fournisseurs
- Basculement automatique : failover en moins de 5 secondes
- Cache intelligent : réduction des coûts pour les requêtes similaires
Implémentation du routeur multi-fournisseurs
Structure du projet
ai-failover-router/
├── src/
│ ├── providers/
│ │ ├── holy_sheep_provider.py
│ │ ├── openai_provider.py
│ │ ├── anthropic_provider.py
│ │ └── provider_base.py
│ ├── router/
│ │ ├── failover_router.py
│ │ └── health_checker.py
│ ├── cache/
│ │ └── redis_cache.py
│ └── main.py
├── config.yaml
├── requirements.txt
└── .envFichier de configuration centralisé
# config.yaml
providers:
holy_sheep:
enabled: true
priority: 1
base_url: "https://api.holysheep.ai/v1"
api_key: "${HOLYSHEEP_API_KEY}"
models:
- gpt_4_1
- claude_sonnet_4_5
- gemini_2_5_flash
- deepseek_v3_2
max_retries: 3
timeout: 30
openai:
enabled: true
priority: 2
base_url: "https://api.openai.com/v1"
api_key: "${OPENAI_API_KEY}"
models:
- gpt-4-turbo
- gpt-3.5-turbo
max_retries: 2
timeout: 45
anthropic:
enabled: true
priority: 3
base_url: "https://api.anthropic.com"
api_key: "${ANTHROPIC_API_KEY}"
models:
- claude-3-5-sonnet-20241022
- claude-3-opus-20240229
max_retries: 2
timeout: 60
cache:
enabled: true
redis_url: "redis://localhost:6379"
ttl: 3600
semantic_similarity_threshold: 0.95
health_check:
interval: 30
timeout: 5
failure_threshold: 3
recovery_threshold: 2
fallback:
strategy: "latency" # latency, cost, priority, round_robin
max_cost_increase_percent: 50Classe de base du fournisseur
# src/providers/provider_base.py
from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from datetime import datetime
import asyncio
import aiohttp
import time
@dataclass
class ProviderMetrics:
"""Métriques de performance d'un fournisseur"""
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
total_latency: float = 0.0
last_success: Optional[datetime] = None
last_failure: Optional[datetime] = None
consecutive_failures: int = 0
is_healthy: bool = True
@property
def average_latency(self) -> float:
if self.successful_requests == 0:
return float('inf')
return self.total_latency / self.successful_requests
@property
def success_rate(self) -> float:
if self.total_requests == 0:
return 1.0
return self.successful_requests / self.total_requests
class ProviderBase(ABC):
"""Classe de base pour tous les fournisseurs d'API AI"""
def __init__(
self,
name: str,
base_url: str,
api_key: str,
models: List[str],
timeout: int = 30,
max_retries: int = 3
):
self.name = name
self.base_url = base_url
self.api_key = api_key
self.models = models
self.timeout = timeout
self.max_retries = max_retries
self.metrics = ProviderMetrics()
self._session: Optional[aiohttp.ClientSession] = None
async def get_session(self) -> aiohttp.ClientSession:
"""Obtient ou crée une session HTTP"""
if self._session is None or self._session.closed:
self._session = aiohttp.ClientSession()
return self._session
async def close(self):
"""Ferme la session HTTP"""
if self._session and not self._session.closed:
await self._session.close()
@abstractmethod
async def generate(
self,
model: str,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 2048,
**kwargs
) -> Dict[str, Any]:
"""Génère une réponse à partir du modèle"""
pass
@abstractmethod
async def health_check(self) -> bool:
"""Vérifie la santé du fournisseur"""
pass
async def _make_request(
self,
method: str,
endpoint: str,
data: Optional[Dict] = None,
retry_count: int = 0
) -> Dict[str, Any]:
"""Effectue une requête HTTP avec gestion des erreurs"""
session = await self.get_session()
url = f"{self.base_url}{endpoint}"
headers = self._get_headers()
start_time = time.time()
self.metrics.total_requests += 1
try:
async with session.request(
method,
url,
json=data,
headers=headers,
timeout=aiohttp.ClientTimeout(total=self.timeout)
) as response:
latency = time.time() - start_time
self.metrics.total_latency += latency
if response.status == 200:
self.metrics.successful_requests += 1
self.metrics.last_success = datetime.now()
self.metrics.consecutive_failures = 0
return await response.json()
elif response.status == 429:
# Rate limiting - retry with exponential backoff
if retry_count < self.max_retries:
await asyncio.sleep(2 ** retry_count)
return await self._make_request(
method, endpoint, data, retry_count + 1
)
raise Exception(f"Rate limit exceeded for {self.name}")
else:
raise Exception(
f"HTTP {response.status}: {await response.text()}"
)
except aiohttp.ClientError as e:
self.metrics.failed_requests += 1
self.metrics.consecutive_failures += 1
self.metrics.last_failure = datetime.now()
if self.metrics.consecutive_failures >= 3:
self.metrics.is_healthy = False
if retry_count < self.max_retries:
await asyncio.sleep(2 ** retry_count)
return await self._make_request(
method, endpoint, data, retry_count + 1
)
raise Exception(f"Request failed for {self.name}: {str(e)}")
@abstractmethod
def _get_headers(self) -> Dict[str, str]:
"""Retourne les headers HTTP nécessaires"""
passImplémentation HolySheep Provider
# src/providers/holy_sheep_provider.py
from typing import Dict, List, Optional, Any
from .provider_base import ProviderBase
class HolySheepProvider(ProviderBase):
"""Fournisseur HolySheep AI avec support multi-modèle"""
# Tarification 2026 en ¥ et $ (taux ¥1=$1)
PRICING = {
"gpt_4_1": {"input": 8, "output": 8}, # $8/1M tokens
"claude_sonnet_4_5": {"input": 15, "output": 15}, # $15/1M tokens
"gemini_2_5_flash": {"input": 2.50, "output": 2.50}, # $2.50/1M tokens
"deepseek_v3_2": {"input": 0.42, "output": 0.42}, # $0.42/1M tokens
}
# Mapping des modèles vers les endpoints HolySheep
MODEL_MAPPING = {
"gpt_4_1": "gpt-4.1",
"claude_sonnet_4_5": "claude-sonnet-4.5",
"gemini_2_5_flash": "gemini-2.5-flash",
"deepseek_v3_2": "deepseek-v3.2",
}
def __init__(self, api_key: str, timeout: int = 30, max_retries: int = 3):
super().__init__(
name="HolySheep",
base_url="https://api.holysheep.ai/v1",
api_key=api_key,
models=list(self.MODEL_MAPPING.keys()),
timeout=timeout,
max_retries=max_retries
)
self.supported_features = {
"streaming": True,
"function_calling": True,
"vision": True,
"json_mode": True,
"context_caching": True,
}
async def generate(
self,
model: str,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 2048,
stream: bool = False,
**kwargs
) -> Dict[str, Any]:
"""Génère une réponse via HolySheep AI"""
mapped_model = self.MODEL_MAPPING.get(model, model)
payload = {
"model": mapped_model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"stream": stream,
}
# Ajout des paramètres optionnels
if "tools" in kwargs:
payload["tools"] = kwargs["tools"]
if "response_format" in kwargs:
payload["response_format"] = kwargs["response_format"]
if "seed" in kwargs:
payload["seed"] = kwargs["seed"]
result = await self._make_request("POST", "/chat/completions", payload)
# Ajout des métadonnées de coût
if "usage" in result:
model_key = [k for k, v in self.MODEL_MAPPING.items()
if v == result.get("model")]
if model_key:
pricing = self.PRICING.get(model_key[0], {"input": 0, "output": 0})
result["cost_info"] = {
"input_cost": (result["usage"]["prompt_tokens"] / 1_000_000)
* pricing["input"],
"output_cost": (result["usage"]["completion_tokens"] / 1_000_000)
* pricing["output"],
"currency": "USD",
"pricing_yuan": {
"input": (result["usage"]["prompt_tokens"] / 1_000_000)
* pricing["input"],
"output": (result["usage"]["completion_tokens"] / 1_000_000)
* pricing["output"],
}
}
return result
async def health_check(self) -> bool:
"""Vérifie la santé de l'API HolySheep"""
try:
# Test avec un appel minimal
result = await self._make_request(
"POST",
"/chat/completions",
{
"model": self.MODEL_MAPPING["deepseek_v3_2"],
"messages": [{"role": "user", "content": "Hi"}],
"max_tokens": 5,
}
)
return "choices" in result
except Exception:
return False
def _get_headers(self) -> Dict[str, str]:
"""Retourne les headers pour HolySheep"""
return {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
def get_cost_estimate(self, model: str, prompt_tokens: int,
completion_tokens: int) -> Dict[str, float]:
"""Estime le coût pour un nombre de tokens donné"""
pricing = self.PRICING.get(model, {"input": 0, "output": 0})
input_cost = (prompt_tokens / 1_000_000) * pricing["input"]
output_cost = (completion_tokens / 1_000_000) * pricing["output"]
return {
"total_usd": input_cost + output_cost,
"total_cny": input_cost + output_cost, # Taux 1:1
"savings_vs_openai": (input_cost + output_cost) * 0.15
if model in ["gpt_4_1", "claude_sonnet_4_5"]
else 0,
}Routeur intelligent avec basculement automatique
# src/router/failover_router.py
import asyncio
from typing import Dict, List, Optional, Any, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import logging
from .health_checker import HealthChecker
from ..providers.provider_base import ProviderBase, ProviderMetrics
from ..cache.redis_cache import SemanticCache
logger = logging.getLogger(__name__)
class RoutingStrategy(Enum):
LATENCY = "latency"
COST = "cost"
PRIORITY = "priority"
ROUND_ROBIN = "round_robin"
LOAD_BALANCED = "load_balanced"
@dataclass
class RoutingConfig:
strategy: RoutingStrategy = RoutingStrategy.LATENCY
enable_cache: bool = True
enable_fallback: bool = True
max_fallback_cost_increase: float = 0.5
circuit_breaker_threshold: int = 5
circuit_breaker_timeout: int = 60
class FailoverRouter:
"""Routeur intelligent avec basculement automatique multi-fournisseurs"""
def __init__(
self,
providers: Dict[str, ProviderBase],
config: Optional[RoutingConfig] = None,
cache: Optional[SemanticCache] = None
):
self.providers = providers
self.config = config or RoutingConfig()
self.cache = cache
self.health_checker = HealthChecker(providers)
self._round_robin_counters: Dict[str, int] = {
name: 0 for name in providers.keys()
}
self._circuit_breakers: Dict[str, Dict] = {
name: {"failures": 0, "opened_at": None}
for name in providers.keys()
}
# Callbacks optionnels
self.on_provider_failure: Optional[Callable] = None
self.on_provider_recovery: Optional[Callable] = None
async def start_health_checks(self, interval: int = 30):
"""Démarre les vérifications de santé périodiques"""
asyncio.create_task(self.health_checker.start(interval))
asyncio.create_task(self._circuit_breaker_monitor())
async def generate(
self,
model: str,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 2048,
preferred_provider: Optional[str] = None,
**kwargs
) -> Dict[str, Any]:
"""Génère une réponse avec routage intelligent"""
# Vérification du cache si activé
if self.config.enable_cache and self.cache:
cached_result = await self.cache.get(messages, model)
if cached_result:
cached_result["cached"] = True
return cached_result
# Sélection du fournisseur selon la stratégie
provider_name = self._select_provider(model, preferred_provider)
if not provider_name:
raise Exception("Aucun fournisseur disponible")
provider = self.providers[provider_name]
try:
result = await provider.generate(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
**kwargs
)
result["provider"] = provider_name
result["routing_strategy"] = self.config.strategy.value
# Sauvegarde en cache
if self.config.enable_cache and self.cache:
await self.cache.set(messages, model, result)
return result
except Exception as e:
logger.error(f"Erreur avec {provider_name}: {str(e)}")
self._record_failure(provider_name)
# Tentative de fallback si activé
if self.config.enable_fallback:
return await self._fallback(
model, messages, temperature, max_tokens,
exclude_providers=[provider_name], **kwargs
)
raise
def _select_provider(
self,
model: str,
preferred_provider: Optional[str] = None
) -> Optional[str]:
"""Sélectionne le meilleur fournisseur selon la stratégie"""
# Filtrage des fournisseurs disponibles et healthy
available = [
name for name, provider in self.providers.items()
if model in provider.models
and provider.metrics.is_healthy
and not self._is_circuit_open(name)
]
if not available:
# Fallback: tous les fournisseurs même unhealthy
available = [
name for name, provider in self.providers.items()
if model in provider.models
]
if not available:
return None
# Respect du fournisseur préféré si disponible
if preferred_provider and preferred_provider in available:
return preferred_provider
# Application de la stratégie de routage
if self.config.strategy == RoutingStrategy.LATENCY:
return min(
available,
key=lambda n: self.providers[n].metrics.average_latency
)
elif self.config.strategy == RoutingStrategy.PRIORITY:
sorted_providers = sorted(
available,
key=lambda n: self._get_provider_priority(n)
)
return sorted_providers[0]
elif self.config.strategy == RoutingStrategy.ROUND_ROBIN:
return self._round_robin_select(available)
elif self.config.strategy == RoutingStrategy.COST:
# Sélection du fournisseur le moins cher pour le modèle
costs = {
name: self._get_model_cost(name, model)
for name in available
}
return min(costs, key=costs.get)
return available[0]
async def _fallback(
self,
model: str,
messages: List[Dict],
temperature: float,
max_tokens: int,
exclude_providers: List[str],
**kwargs
) -> Dict[str, Any]:
"""Basculement vers un fournisseur alternatif"""
available = [
name for name, provider in self.providers.items()
if model in provider.models
and name not in exclude_providers
and provider.metrics.is_healthy
]
for provider_name in available:
try:
provider = self.providers[provider_name]
logger.info(f"Tentative de fallback vers {provider_name}")
result = await provider.generate(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
**kwargs
)
result["provider"] = provider_name
result["fallback_used"] = True
result["original_excluded"] = exclude_providers
# Reset des compteurs de failure pour le fournisseur original
for excluded in exclude_providers:
self._reset_circuit_breaker(excluded)
return result
except Exception as e:
logger.warning(f"Fallback {provider_name} échoué: {str(e)}")
exclude_providers.append(provider_name)
raise Exception("Tous les fournisseurs ont échoué")
def _record_failure(self, provider_name: str):
"""Enregistre un échec et met à jour le circuit breaker"""
cb = self._circuit_breakers[provider_name]
cb["failures"] += 1
if cb["failures"] >= self.config.circuit_breaker_threshold:
cb["opened_at"] = datetime.now()
logger.warning(f"Circuit breaker ouvert pour {provider_name}")
if self.on_provider_failure:
asyncio.create_task(
self.on_provider_failure(provider_name)
)
def _reset_circuit_breaker(self, provider_name: str):
"""Réinitialise le circuit breaker après recovery"""
self._circuit_breakers[provider_name] = {"failures": 0, "opened_at": None}
if self.on_provider_recovery:
asyncio.create_task(
self.on_provider_recovery(provider_name)
)
def _is_circuit_open(self, provider_name: str) -> bool:
"""Vérifie si le circuit breaker est ouvert"""
cb = self._circuit_breakers[provider_name]
if cb["opened_at"] is None:
return False
elapsed = (datetime.now() - cb["opened_at"]).total_seconds()
if elapsed > self.config.circuit_breaker_timeout:
# Tentative de recovery
cb["opened_at"] = None
cb["failures"] = 0
return False
return True
async def _circuit_breaker_monitor(self):
"""Surveillance périodique des circuit breakers"""
while True:
await asyncio.sleep(10)
for name, cb in self._circuit_breakers.items():
if cb["opened_at"] and cb["opened_at"] is not None:
elapsed = (datetime.now() - cb["opened_at"]).total_seconds()
if elapsed > self.config.circuit_breaker_timeout:
# Test de recovery
provider = self.providers.get(name)
if provider and await provider.health_check():
logger.info(f"Circuit breaker en recovery pour {name}")
def _round_robin_select(self, available: List[str]) -> str:
"""Sélection round-robin entre les fournisseurs disponibles"""
for _ in range(len(available)):
self._round_robin_counters[available[0]] = (
self._round_robin_counters[available[0]] + 1
) % 100
selected = available[
self._round_robin_counters[available[0]] % len(available)
]
if selected in available:
return selected
return available[0]
def _get_provider_priority(self, name: str) -> int:
"""Obtient la priorité configurée du fournisseur"""
return getattr(self.providers[name], 'priority', 999)
def _get_model_cost(self, provider_name: str, model: str) -> float:
"""Obtient le coût du modèle pour un fournisseur"""
provider = self.providers[provider_name]
if hasattr(provider, 'PRICING'):
return provider.PRICING.get(model, {}).get("input", float('inf'))
return 0.0
def get_status(self) -> Dict[str, Any]:
"""Retourne le statut de tous les fournisseurs"""
return {
name: {
"healthy": provider.metrics.is_healthy,
"average_latency_ms": round(provider.metrics.average_latency * 1000, 2),
"success_rate": round(provider.metrics.success_rate * 100, 2),
"total_requests": provider.metrics.total_requests,
"circuit_breaker_open": self._is_circuit_open(name),
}
for name, provider in self.providers.items()
}Tarification et ROI
Analysons le retour sur investissement concret d'une architecture multi-fournisseurs avec HolySheep comme nœud principal :
| Scénario | Coût mensuel (API seules) | Avec HolySheep + Failover | Économie |
|---|---|---|---|
| Startup (1M req/mois) | 800 $ (OpenAI) | 120 $ | 85% |
| PME (10M req/mois) | 8 000 $ | 1 200 $ | 85% |
| Entreprise (100M req/mois) | 80 000 $ | 12 000 $ | 85% |
| Coût downtime évité | - | ~50 000 $/incident | - |
Calculateur d'économie personnalisé
# Outil de calcul d'économie
def calculate_savings(
monthly_requests: int,
avg_tokens_per_request: int = 500,
cost_per_million_tokens: float = 8.0,
provider: str = "HolySheep"
) -> dict:
"""Calcule les économies potentielles"""
total_tokens = monthly_requests * avg_tokens_per_request
total_millions = total_tokens / 1_000_000
if provider == "OpenAI":
current_cost = total_millions * cost_per_million_tokens
holy_sheep_cost = total_millions * cost_per_million_tokens * 0.15
elif provider == "Anthropic":
current_cost = total_millions * 15 # Claude Sonnet
holy_sheep_cost = total_millions * 15 * 0.15
else:
current_cost = total_millions * cost_per_million_tokens
holy_sheep_cost = current_cost
return {
"current_provider": provider,
"monthly_requests": monthly_requests,
"total_tokens_monthly": total_tokens,
"current_cost_usd": round(current_cost, 2),
"holy_sheep_cost_usd": round(holy_sheep_cost, 2),
"monthly_savings_usd": round(current_cost - holy_sheep_cost, 2),
"annual_savings_usd": round((current_cost - holy_sheep_cost) * 12, 2),
"savings_percentage": round((1 - holy_sheep_cost/current_cost) * 100, 1),
}
Exemple d'utilisation
result = calculate_savings(
monthly_requests=1_000_000,
avg_tokens_per_request=1000,
cost_per_million_tokens=8.0,
provider="OpenAI"
)
print(f"Économie annuelle: ${result['annual_savings_usd']}")Pourquoi choisir HolySheep
- Économie de 85%+ : Prix 2026 parmi les plus bas du marché avec $0.42/1M tokens pour DeepSeek V3.2
- Latence <50ms : Infrastructure optimisée pour les applications temps réel
- Paiements locaux : WeChat Pay et Alipay disponibles pour les clients chinois
- Multi-fournisseurs natif : Accès à GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash et DeepSeek V3.2 depuis une seule API
- Crédits gratuits : Commencez sans engagement financier
- Dashboard监控 complet : Surveillance en temps réel de l'utilisation et des coûts
- API compatible : Migration depuis OpenAI en moins de 5 minutes
Erreurs courantes et solutions
Erreur 1 : "Connection timeout exceeded"
# Problème : Le timeout par défaut est trop court pour certains modèles
Solution : Ajuster les timeouts selon le modèle utilisé
Configuration incorrecte
providers:
openai:
timeout: 30 # Trop court pour Claude 3 Opus
Configuration corrigée
providers:
anthropic:
timeout: 120 # Timeout étendu pour modèles lents
max_retries: 5 # Plus de tentatives
Code de correction
class HolySheepProvider(ProviderBase):
TIMEOUT_BY_MODEL = {
"gpt_4_1": 30,
"claude_sonnet_4_5": 60,
"gemini_2_5_flash": 20,
"deepseek_v3_2": 25,
}
async def generate(self, model: str, **kwargs):
timeout = self.TIMEOUT_BY_MODEL.get(model, self.timeout)
self.timeout = timeout
return await super().generate(model, **kwargs)Erreur 2 : "Rate limit exceeded - retry after 60s"
# Problème : Les rate limits ne sont pas gérés correctement
Solution : Implémenter un rate limiter et une queue de requêtes
import asyncio
from collections import deque
from datetime import datetime, timedelta
class RateLimiter:
"""Gestionnaire de rate limiting intelligent"""
def __init__(self, requests_per_minute: int, burst_size: int = 10):
self.rpm = requests_per_minute
self.burst = burst_size
self.requests: deque = deque()
self._lock = asyncio.Lock()
async def acquire(self):
"""Acquiert une permission pour effectuer une requête"""
async with self._lock:
now = datetime.now()
cutoff = now - timedelta(minutes=1)
# Nettoyage des requêtes anciennes
while self.requests and self.requests[0] < cutoff:
self.requests.popleft()
# Vérification du burst
if len(self.requests) < self.burst:
self.requests.append(now)
return
# Calcul du temps d'attente
oldest = self.requests[0]
wait_time = 60 - (now - oldest).total_seconds()
if wait_time > 0:
await asyncio.sleep(wait_time)
self.requests.popleft()
self.requests.append(datetime.now())
async def __aenter__(self):
await self.acquire()
return self
async def __aexit__(self, *args):
pass
Utilisation dans le routeur
class FailoverRouter:
def __init__(self, providers, config):
super().__init__(providers, config)