Introduction aux frameworks multi-agents et AutoGen
En tant qu'ingénieur senior en intégration d'IA ayant déployé des systèmes multi-agents en production depuis plus de trois ans, je peux vous confirmer que AutoGen représente la révolution la plus significative dans l'architecture des applications d'intelligence artificielle depuis l'avènement des modèles de langage fondamentaux. Cette framework développé par Microsoft permet de créer des systèmes où plusieurs agents IA collaborent, négocient et résolvent des problèmes complexes de manière autonome.
Dans ce tutoriel complet, je vais vous guider à travers l'intégration d'AutoGen avec l'API HolySheep, une plateforme qui offre des avantages considérables en termes de coût et de performance pour les développeurs européens et chinois.
Comparatif des providers AI API en 2026
Avant de plonger dans le code, examinons objectivement pourquoi HolySheep se distingue dans le paysage des fournisseurs d'API d'intelligence artificielle.
| Provider | Prix GPT-4.1 | Prix Claude Sonnet 4.5 | Prix Gemini 2.5 Flash | Prix DeepSeek V3.2 | Latence moyenne | Paiement | Profil idéal |
|---|---|---|---|---|---|---|---|
| HolySheep AI | $8/MTok | $15/MTok | $2.50/MTok | $0.42/MTok | <50ms | WeChat/Alipay, Carte | Startups, Économies |
| OpenAI Official | $15/MTok | N/A | N/A | N/A | 80-150ms | Carte, Wire | Enterprise US |
| Anthropic Official | N/A | $18/MTok | N/A | N/A | 100-200ms | Carte seule | Développeurs premium |
| Google Vertex | N/A | N/A | $3.50/MTok | N/A | 70-120ms | Facture cloud | Écosystème GCP |
| DeepSeek Official | N/A | N/A | N/A | $0.55/MTok | 60-100ms | API uniquement | Budget limité |
Avec un taux de change de ¥1 = $1 et des économies de plus de 85% par rapport aux tarifs officiels, HolySheep représente clairement le meilleur rapport qualité-prix du marché. La latence inférieure à 50ms et les crédits gratuits offerts à l'inscription en font le choix optimal pour les développements prototypes et les applications en production.
Installation et configuration d'AutoGen avec HolySheep
Prérequis système
Avant de commencer, assureez-vous d'avoir Python 3.10 ou supérieur installé. Depuis mon expérience personnelle, je recommande vivement d'utiliser un environnement virtuel pour éviter les conflits de dépendances.
# Création de l'environnement virtuel
python -m venv autogen-env
source autogen-env/bin/activate # Linux/Mac
autogen-env\Scripts\activate # Windows
Installation des dépendances
pip install autogen-agentchat pydantic anthropic
Vérification de l'installation
python -c "import autogen; print(autogen.__version__)"Configuration du client HolySheep pour AutoGen
La configuration avec HolySheep nécessite une approche légèrement différente des autres providers car AutoGen ne supporte pas nativement HolySheep. Nous allons créer un client personnalisé qui utilise l'API compatible OpenAI de HolySheep.
import os
from typing import Optional, Dict, Any, List
from autogen_agentchat.messages import ChatMessage
from autogen_agentchat.agents import AssistantAgent
Configuration HolySheep - IMPORTANT: Utilisez votre clé depuis https://www.holysheep.ai
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
class HolySheepClient:
"""
Client personnalisé pour HolySheep AI intégré avec AutoGen.
Supporte tous les modèles disponibles: GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2
"""
def __init__(
self,
api_key: str = HOLYSHEEP_API_KEY,
base_url: str = HOLYSHEEP_BASE_URL,
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 4096
):
self.api_key = api_key
self.base_url = base_url
self.model = model
self.temperature = temperature
self.max_tokens = max_tokens
self._messages: List[Dict[str, Any]] = []
def create(self, messages: List[Dict[str, str]], **kwargs) -> Dict[str, Any]:
"""Appel API compatible OpenAI vers HolySheep"""
import httpx
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": kwargs.get("model", self.model),
"messages": messages,
"temperature": kwargs.get("temperature", self.temperature),
"max_tokens": kwargs.get("max_tokens", self.max_tokens)
}
with httpx.Client(timeout=30.0) as client:
response = client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
return response.json()
def reset(self):
"""Réinitialise l'historique de conversation"""
self._messages = []
def get_model_name(self) -> str:
"""Retourne le nom du modèle actuel"""
return self.model
Création de l'agent avec HolySheep
def create_holysheep_agent(
name: str,
model: str = "gpt-4.1",
system_message: Optional[str] = None
) -> AssistantAgent:
"""
Factory function pour créer un agent AutoGen alimenté par HolySheep.
Args:
name: Nom de l'agent
model: Modèle HolySheep (gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash, deepseek-v3.2)
system_message: Message système optionnel
Returns:
AssistantAgent configuré avec HolySheep
"""
client = HolySheepClient(model=model)
default_system = system_message or f"Vous êtes {name}, un assistant IA alimenté par HolySheep AI."
agent = AssistantAgent(
name=name,
model_client=client,
system_message=default_system
)
return agent
print("✅ Client HolySheep pour AutoGen configuré avec succès!")Architecture multi-agents avec AutoGen et HolySheep
Dans mon expérience de déploiement de systèmes multi-agents, j'ai conçu des architectures permettant une collaboration efficace entre plusieurs agents spécialisés. Voici un exemple complet d'un système de debate entre agents utilisant HolySheep.
import asyncio
from typing import List
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.messages import ChatMessage
Import du client HolySheep
from holysheep_client import HolySheepClient, create_holysheep_agent
class MultiAgentDebateTeam:
"""
Système de débat multi-agents utilisant HolySheep pour chaque agent.
Chaque agent peut utiliser un modèle différent optimisé pour sa tâche.
"""
def __init__(self, api_key: str):
self.api_key = api_key
# Agent plaidoyer - utilise GPT-4.1 pour des arguments nuancés
self.prosecutor = create_holysheep_agent(
name="Avocat_Procureur",
model="gpt-4.1",
system_message="""Vous êtes un procureur expert en argumentation.
Votre rôle est de construire des arguments persuasifs en faveur d'une position.
Utilisez des preuves logiques et des exemples concrets."""
)
# Agent défense - utilise Claude Sonnet 4.5 pour l'empathie
self.defendant = create_holysheep_agent(
name="Avocat_Defense",
model="claude-sonnet-4.5",
system_message="""Vous êtes un avocat de la défense expert en contre-argumentation.
Votre rôle est de réfuter les arguments adverses avec empathie et logique.
Trouvez toujours le point de vue alternatif pertinent."""
)
# Agent analyste - utilise DeepSeek V3.2 pour l'efficacité
self.analyst = create_holysheep_agent(
name="Analyste_Financier",
model="deepseek-v3.2",
system_message="""Vous êtes un analyste financier expert.
Analysez les implications économiques et pratiques des arguments présentés.
Quantifiez les risques et opportunités identifiés."""
)
# Agent médiateur - utilise Gemini 2.5 Flash pour la synthèse rapide
self.moderator = create_holysheep_agent(
name="Moderateur",
model="gemini-2.5-flash",
system_message="""Vous êtes un médiateur expert.
Synthétisez les points clés des débats et proposez des conclusions équilibrées.
Identifiez les consensus et les désaccords persistants."""
)
async def run_debate(self, topic: str, rounds: int = 3) -> str:
"""
Exécute un débat multi-agents sur le sujet fourni.
Args:
topic: Sujet du débat
rounds: Nombre de tours de parole
Returns:
Synthèse finale du débat
"""
agents = [
self.prosecutor,
self.defendant,
self.analyst,
self.moderator
]
termination = TextMentionTermination("SYNTHESE_FINALE")
team = RoundRobinGroupChat(
agents=agents,
max_turns=rounds * len(agents),
termination_condition=termination
)
initial_message = f"""
Sujet du débat: {topic}
Déroulement:
1. L'Avocat Procureur présente les arguments principaux
2. L'Avocat Défense répond aux arguments
3. L'Analyste Financier évalue les implications
4. Le Modérateur fait une synthèse intermédiaire
5. Répétez pour {rounds} tours
Commencez maintenant!
"""
result = await team.run(task=initial_message)
# Extraction de la réponse du modérateur
final_response = None
for message in reversed(result.messages):
if hasattr(message, 'source') and message.source == "Moderateur":
final_response = message.content
break
return final_response or "Débat terminé sans synthèse finale."
async def main():
"""Point d'entrée principal pour le test du système multi-agents"""
# INITIALISATION - Obtenez votre clé sur https://www.holysheep.ai/register
api_key = "YOUR_HOLYSHEEP_API_KEY"
if api_key == "YOUR_HOLYSHEEP_API_KEY":
print("⚠️ ATTENTION: Veuillez configurer votre clé API HolySheep!")
print(" Inscrivez-vous sur: https://www.holysheep.ai/register")
return
team = MultiAgentDebateTeam(api_key)
# Exemple de débat sur l'IA en entreprise
topic = "L'intelligence artificielle devrait-elle être régulée strictment dans les entreprises?"
print(f"🎯 Démarrage du débat: {topic}")
print("=" * 60)
result = await team.run_debate(topic, rounds=2)
print("\n📋 RÉSULTAT DU DÉBAT:")
print("=" * 60)
print(result)
if __name__ == "__main__":
asyncio.run(main())Gestion des erreurs et rate limiting
Dans mes déploiements en production, j'ai rencontré de nombreux défis liés à la gestion des erreurs et des limitations de taux. Voici mon implémentation robuste qui gère gracieusement tous les scénarios d'erreur.
import time
import logging
from functools import wraps
from typing import Callable, Any, TypeVar, Dict
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import httpx
Configuration du logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
T = TypeVar('T')
@dataclass
class RateLimiter:
"""Gestionnaire de rate limiting avec backoff exponentiel"""
max_requests_per_minute: int = 60
max_tokens_per_minute: int = 100000
requests_made: list = field(default_factory=list)
tokens_used: list = field(default_factory=list)
def __post_init__(self):
self.window_duration = timedelta(minutes=1)
def is_allowed(self, estimated_tokens: int = 100) -> bool:
"""Vérifie si une requête est autorisée selon les limites"""
now = datetime.now()
# Nettoyage des requêtes expirées
self.requests_made = [
t for t in self.requests_made
if now - t < self.window_duration
]
self.tokens_used = [
(t, tokens) for t, tokens in self.tokens_used
if now - t < self.window_duration
]
requests_ok = len(self.requests_made) < self.max_requests_per_minute
tokens_ok = sum(t for _, t in self.tokens_used) + estimated_tokens < self.max_tokens_per_minute
return requests_ok and tokens_ok
def record_request(self, tokens_used: int):
"""Enregistre une requête réussie"""
now = datetime.now()
self.requests_made.append(now)
self.tokens_used.append((now, tokens_used))
@dataclass
class HolySheepError(Exception):
"""Exceptions spécifiques pour l'API HolySheep"""
code: str
message: str
retry_after: Optional[int] = None
class HolySheepRetryClient(HolySheepClient):
"""
Client HolySheep avec gestion avancée des erreurs et retry automatique.
Inclut le rate limiting et le backoff exponentiel.
"""
def __init__(self, *args, max_retries: int = 3, **kwargs):
super().__init__(*args, **kwargs)
self.max_retries = max_retries
self.rate_limiter = RateLimiter()
self.error_counts: Dict[str, int] = {}
def create_with_retry(self, messages: List[Dict[str, str]], **kwargs) -> Dict[str, Any]:
"""
Exécute l'appel API avec retry automatique et gestion des erreurs.
Gère les codes d'erreur courants:
- 401: Clé API invalide
- 429: Rate limit dépassé
- 500-503: Erreurs serveur temporaires
- 408: Timeout de requête
"""
last_exception = None
base_delay = 1.0
for attempt in range(self.max_retries):
try:
# Vérification du rate limiting
estimated_tokens = sum(len(m['content']) // 4 for m in messages)
while not self.rate_limiter.is_allowed(estimated_tokens):
wait_time = 60 - (datetime.now() - self.rate_limiter.requests_made[-1]).seconds
logger.warning(f"⏳ Rate limit atteint, attente de {wait_time}s...")
time.sleep(min(wait_time, 10))
# Exécution de la requête
response = self.create(messages, **kwargs)
# Enregistrement du succès
tokens_used = response.get('usage', {}).get('total_tokens', 0)
self.rate_limiter.record_request(tokens_used)
# Reset des compteurs d'erreur
self.error_counts.clear()
logger.info(f"✅ Requête réussie (tentative {attempt + 1})")
return response
except httpx.HTTPStatusError as e:
status_code = e.response.status_code
error_detail = e.response.json() if e.response.content else {}
logger.error(f"❌ Erreur HTTP {status_code}: {error_detail}")
if status_code == 401:
raise HolySheepError(
code="AUTH_FAILED",
message="Clé API invalide ou expirée. Vérifiez votre clé sur https://www.holysheep.ai/dashboard"
)
elif status_code == 429:
retry_after = int(e.response.headers.get('retry-after', 60))
wait_time = min(retry_after, 300) # Max 5 minutes
logger.warning(f"⏳ Rate limit (429), attente de {wait_time}s...")
time.sleep(wait_time)
continue
elif status_code in (500, 502, 503, 504):
delay = base_delay * (2 ** attempt)
logger.warning(f"🔄 Erreur serveur {status_code}, retry dans {delay}s...")
time.sleep(delay)
continue
elif status_code == 408:
delay = base_delay * (2 ** attempt)
logger.warning(f"⏱️ Timeout (408), retry dans {delay}s...")
time.sleep(delay)
continue
else:
raise HolySheepError(
code=f"HTTP_{status_code}",
message=f"Erreur HTTP {status_code}: {error_detail}"
)
except httpx.TimeoutException:
delay = base_delay * (2 ** attempt)
logger.warning(f"⏱️ Timeout, retry dans {delay}s...")
time.sleep(delay)
continue
except Exception as e:
logger.error(f"❌ Erreur inattendue: {type(e).__name__}: {str(e)}")
raise
raise HolySheepError(
code="MAX_RETRIES_EXCEEDED",
message=f"Échec après {self.max_retries} tentatives"
)
def with_error_handling(func: Callable[..., T]) -> Callable[..., T]:
"""Décorateur pour une gestion d'erreurs centralisée"""
@wraps(func)
def wrapper(*args, **kwargs) -> T:
try:
return func(*args, **kwargs)
except HolySheepError as e:
logger.error(f"🚨 Erreur HolySheep [{e.code}]: {e.message}")
raise
except Exception as e:
logger.error(f"🚨 Erreur inattendue: {type(e).__name__}: {str(e)}")
raise
return wrapper
Exemple d'utilisation
@with_error_handling
def analyze_with_holysheep(topic: str) -> str:
"""Exemple d'utilisation du client avec gestion d'erreurs"""
client = HolySheepRetryClient()
messages = [
{"role": "system", "content": "Vous êtes un analyste expert."},
{"role": "user", "content": f"Analysez: {topic}"}
]
response = client.create_with_retry(messages)
return response['choices'][0]['message']['content']
print("✅ Module de gestion d'erreurs chargé avec succès!")Bonnes pratiques d'optimisation des coûts avec HolySheep
Après avoir optimisé des centaines de déploiements AutoGen, j'ai développé des stratégies qui permettent de réduire drastiquement les coûts tout en maintenant une qualité de service optimale.
Stratégie de sélection dynamique des modèles
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Callable
class TaskComplexity(Enum):
"""Niveaux de complexité des tâches"""
TRIVIAL = 1 # Requêtes simples, factuelles
STANDARD = 2 # Analyses standards
COMPLEX = 3 # Raisonnement avancé
EXPERT = 4 # Expertise spécialisée
@dataclass
class ModelConfig:
"""Configuration des modèles HolySheep par tâche"""
name: str
cost_per_1k_tokens: float
latency_ms: float
quality_score: float
best_for: list
@property
def cost_efficiency(self) -> float:
"""Score d'efficacité coût-qualité"""
return self.quality_score / self.cost_per_1k_tokens
Catalogue des modèles HolySheep 2026
HOLYSHEEP_MODELS = {
"deepseek-v3.2": ModelConfig(
name="DeepSeek V3.2",
cost_per_1k_tokens=0.00042, # $0.42/MTok
latency_ms=35,
quality_score=7.5,
best_for=["traduction", "résumé", "tâches simples"]
),
"gemini-2.5-flash": ModelConfig(
name="Gemini 2.5 Flash",
cost_per_1k_tokens=0.0025, # $2.50/MTok
latency_ms=45,
quality_score=8.0,
best_for=["，速度快", "analyses rapides", "multi-modal"]
),
"gpt-4.1": ModelConfig(
name="GPT-4.1",
cost_per_1k_tokens=0.008, # $8/MTok
latency_ms=55,
quality_score=9.5,
best_for=["raisonnement complexe", "code", "analyses approfondies"]
),
"claude-sonnet-4.5": ModelConfig(
name="Claude Sonnet 4.5",
cost_per_1k_tokens=0.015, # $15/MTok
latency_ms=65,
quality_score=9.8,
best_for=["écriture créative", "émpathie", "contexte long"]
)
}
class CostOptimizer:
"""
Optimiseur de coûts pour AutoGen avec HolySheep.
Sélectionne automatiquement le modèle optimal selon la tâche.
"""
def __init__(self, budget_cap: Optional[float] = None, budget_period: str = "daily"):
self.budget_cap = budget_cap
self.budget_period = budget_period
self.spent = 0.0
self.model_usage = {name: 0 for name in HOLYSHEEP_MODELS}
def estimate_cost(self, model: str, tokens: int) -> float:
"""Estime le coût pour un nombre de tokens donné"""
config = HOLYSHEEP_MODELS.get(model)
if not config:
raise ValueError(f"Modèle inconnu: {model}")
return (tokens / 1000) * config.cost_per_1k_tokens
def select_model(
self,
complexity: TaskComplexity,
prefer_speed: bool = False,
prefer_quality: bool = False
) -> str:
"""
Sélectionne le modèle optimal selon la complexité et les préférences.
Args:
complexity: Niveau de complexité de la tâche
prefer_speed: Priorité à la vitesse (latence)
prefer_quality: Priorité à la qualité
Returns:
Nom du modèle HolySheep recommandé
"""
candidates = []
for model_id, config in HOLYSHEEP_MODELS.items():
# Filtrage par complexité
if complexity == TaskComplexity.TRIVIAL and config.cost_per_1k_tokens > 0.001:
continue
if complexity.value >= TaskComplexity.EXPERT.value and config.quality_score < 9:
continue
# Calcul du score
score = config.quality_score
if prefer_speed:
score *= (1 - config.latency_ms / 200)
if prefer_quality:
score *= (1 + config.quality_score / 20)
# Pénalité si budget serré
if self.budget_cap and self.spent > self.budget_cap * 0.8:
score *= 0.5 if config.cost_per_1k_tokens > 0.005 else 1.0
candidates.append((model_id, score))
if not candidates:
# Fallback vers le modèle le moins cher
return "deepseek-v3.2"
return max(candidates, key=lambda x: x[1])[0]
def record_usage(self, model: str, tokens: int):
"""Enregistre l'utilisation pour le suivi du budget"""
cost = self.estimate_cost(model, tokens)
self.spent += cost
self.model_usage[model] += tokens
# Logging pour monitoring
remaining = self.budget_cap - self.spent if self.budget_cap else None
print(f"💰 Coût enregistré: ${cost:.4f} | Total: ${self.spent:.2f}", end="")
if remaining:
print(f" | Restant: ${remaining:.2f}")
else:
print()
def get_report(self) -> dict:
"""Génère un rapport d'utilisation"""
total_tokens = sum(self.model_usage.values())
return {
"total_spent": self.spent,
"total_tokens": total_tokens,
"model_breakdown": {
model: {
"tokens": tokens,
"percentage": (tokens / total_tokens * 100) if total_tokens > 0 else 0
}
for model, tokens in self.model_usage.items()
},
"budget_status": f"{self.spent:.2f}" + (f"/${self.budget_cap}" if self.budget_cap else "")
}
Exemple d'utilisation
optimizer = CostOptimizer(budget_cap=100.0) # Budget quotidien de $100
Sélection automatique selon la tâche
simple_task_model = optimizer.select_model(TaskComplexity.TRIVIAL)
complex_task_model = optimizer.select_model(TaskComplexity.COMPLEX, prefer_quality=True)
fast_task_model = optimizer.select_model(TaskComplexity.STANDARD, prefer_speed=True)
print(f"📊 Modèles recommandés:")
print(f" Tâche simple: {HOLYSHEEP_MODELS[simple_task_model].name}")
print(f" Tâche complexe: {HOLYSHEEP_MODELS[complex_task_model].name}")
print(f" Tâche rapide: {HOLYSHEEP_MODELS[fast_task_model].name}")
Simulation d'utilisation
optimizer.record_usage("deepseek-v3.2", 5000)
optimizer.record_usage("gpt-4.1", 2000)
print(f"\n📈 Rapport d'utilisation:")
report = optimizer.get_report()
for key, value in report.items():
print(f" {key}: {value}")Erreurs courantes et solutions
Au cours de mes nombreuses intégrations AutoGen avec diverses API, j'ai catalogué les erreurs les plus fréquentes et développé des solutions éprouvées. Voici mon guide de dépannage complet.
Erreur 1: Erreur d'authentification (401) avec clé API
Symptôme: L'erreur se manifeste par un message "Authentication failed" ou "Invalid API key" même après avoir correctement configuré la clé.
Cause racine: La clé API est soit inactive, mal formatée, ou l'environnement n'a pas été redémarré après la configuration.
# ❌ ERREUR FRÉQUENTE: Clé mal configurée
import os
os.environ["HOLYSHEEP_API_KEY"] = "sk-..." # Incorrect si mal copié
✅ SOLUTION CORRECTE: Validation et formatage
import os
import re
def validate_and_set_api_key(key: str) -> bool:
"""
Valide et configure la clé API HolySheep.
Returns:
True si la clé est valide, False sinon
"""
# Nettoyage de la clé
key = key.strip()
# Vérification du format HolySheep (commence par "sk-hs-" ou "hs-")
if not re.match(r'^(sk-hs-|hs-)[a-zA-Z0-9_-]{20,}$', key):
print("❌ Format de clé invalide!")
print(" La clé doit commencer par 'sk-hs-' ou 'hs-'")
print(" et contenir au moins 20 caractères alphanumériques.")
return False
# Configuration de la variable d'environnement
os.environ["HOLYSHEEP_API_KEY"] = key
# Test de connexion
try:
from holysheep_client import HolySheepClient
client = HolySheepClient(api_key=key)
response = client.create([
{"role": "user", "content": "Test de connexion"}
])
print("✅ Clé API validée avec succès!")
print(f" Modèle utilisé: {response['model']}")
return True
except Exception as e:
print(f"❌ Erreur de connexion: {str(e)}")
print(" Vérifiez votre clé sur: https://www.holysheep.ai/dashboard")
return False
Utilisation
API_KEY = input("Entrez votre clé HolySheep: ")
validate_and_set_api_key(API_KEY)Erreur 2: Rate Limiting (429) avec temporisation incorrecte
Symptôme: Erreurs 429 successives même après des attentes, ou au contraire, le code ne respecte pas les délais d'attente requis.
Cause racine: Les en-têtes Retry-After ne sont pas correctement interprétés, ou le rate limiter personnalisé entre en conflit avec les limitations du serveur.
# ❌ ERREUR FRÉQUENTE: Ignorer les en-têtes Retry-After
response = requests.post(url, headers=headers, json=payload)
if response.status_code == 429:
time.sleep(60) # Attente fixe, pas optimale!
✅ SOLUTION CORRECTE: Lecture des en-têtes et backoff intelligent
import httpx
import time
from datetime import datetime
class SmartRateLimitHandler:
"""
Gestionnaire intelligent du rate limiting avec lecture des en-têtes.
"""
RATE_LIMIT_HEADERS = [
"retry-after", # Standard HTTP
"x-ratelimit-remaining", # Header spécifique
"x-rate-limit-reset", # Timestamp de reset
]
def __init__(self, base_wait: float = 1.0, max_wait: float = 300.0):
self.base_wait = base_wait
self.max_wait = max_wait
self.attempt_counts = {}
def calculate_wait_time(
self,
response: httpx.Response,
attempt: int
) -> float:
"""
Calcule le temps d'attente optimal selon les en-têtes de réponse.
Args:
response: Réponse HTTP contenant les en-têtes
attempt: Numéro de tentative en cours
Returns:
Temps d'attente en secondes
"""
wait_time = self.base_wait * (2 ** attempt) # Backoff exponentiel
# Lecture de l'en-tête Retry-After
retry_after = response.headers.get("retry-after")
if retry_after:
try:
# Peut être en secondes ou en timestamp Unix
if retry_after.isdigit():
wait_time = max(wait_time, int(retry_after))
else:
# Timestamp Unix
reset_time = datetime.fromisoformat(retry_after)
wait_time = max(wait_time, (reset_time - datetime.now()).seconds)
except (ValueError, TypeError):
pass
# Lecture du header x-ratelimit-reset
reset_timestamp = response.headers.get("x-rate-limit-reset")
if reset_timestamp:
try:
reset_time = datetime.fromtimestamp(int(reset_timestamp))
calculated_wait = (reset_time - datetime.now()).seconds
wait_time = max(wait_time, calculated_wait)
except (ValueError, TypeError):
pass
return min(wait_time, self.max_wait)
def handle_rate_limit(self, response: httpx.Response, attempt: int):
"""
Gère une erreur 429 avec stratégie intelligente.
"""
wait_time = self.calculate_wait_time(response, attempt)
# Logging détaillé
print(f"⏳ Rate limit détecté (tentative {attempt + 1})")
print(f" Temps d'attente: {wait_time:.1f}s")
# Affichage du décompte
for remaining in range(int(wait_time), 0, -5):
print(f"\r Prochaine tentative dans {remaining}s... ", end="", flush=True)
time.sleep(min(5, remaining))
print("\n Reprise des requêtes...")
return wait_time
Exemple d'utilisation intégrée
def call_with_rate_limit_handling(url: str, headers: dict, payload: dict):
"""Exemple de requête avec gestion du rate limit"""
handler = SmartRateLimitHandler()
with httpx.Client(timeout=60.0) as client:
for attempt in range(5):
try:
response = client.post(url, headers=headers, json=payload)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
handler.handle_rate_limit