Tableau comparatif : HolySheep vs API officielle vs services relais
| Critère | 🔴 HolySheep AI | API OpenAI Officielle | Services relais tiers |
|---|---|---|---|
| Prix GPT-4.1 | $8/Mtok | $15/Mtok | $10-12/Mtok |
| Prix Claude Sonnet 4.5 | $15/Mtok | $18/Mtok | $16-17/Mtok |
| Prix Gemini 2.5 Flash | $2.50/Mtok | $3.50/Mtok | $2.80-3/Mtok |
| Latence moyenne | <50ms | 80-200ms | 60-150ms |
| Erreurs 429/503 | Taux personnalisé, haute disponibilité | Fréquentes en heure de pointe | Variables selon charge |
| Méthodes de paiement | WeChat Pay, Alipay, USDT, Carte | Carte internationale uniquement | Limité |
| Crédits gratuits | ✅ Oui, dès l'inscription | ❌ Non | ❌ Rarement |
| Support circuit breaker | Intégré via SDK | Manuel | Variable |
Introduction
En tant qu'ingénieur senior qui a intégré des API d'IA dans des environnements de production depuis plus de trois ans, j'ai rencontré d'innombrables problèmes de stabilité liés aux erreurs 503 (Service Unavailable) et 429 (Too Many Requests). Ces erreurs sont particulièrement critiques dans les workflows d'agents IA où une seule requête peut déclencher une chaîne de 10 à 50 appels API. Après avoir testé десятки de solutions, j'ai trouvé que HolySheep AI offrait le meilleur équilibre entre fiabilité, coût et facilité d'implémentation pour la容错设计 (conception de tolérance aux pannes).
Dans cet article, je vais vous montrer comment implémenter un système robuste de retry automatique avec backoff exponentiel et un circuit breaker en production, en utilisant l'API HolySheep comme endpoint principal.
Pourquoi la容错设计 est critique pour les agents IA
Les agents IA modernes ne sont pas de simples chatbots. Un agent comme celui que nous avons déployé pour l'analyse de documents peut effectuer :
- Extraction du contenu (1-3 requêtes)
- Classification du texte (2-5 requêtes)
- Résumé automatique (3-8 requêtes)
- Génération de métadonnées (5-10 requêtes)
Avec une probabilité d'erreur de 2% par requête et 30 requêtes par tâche, la probabilité de succès sans retry tombe à 54%. Avec un bon système de容错, nous atteignons 99.7% de succès en production.
Architecture du système de容错
1. Retry automatique avec backoff exponentiel
#!/usr/bin/env python3
"""
HolySheep AI Agent - Système de Retry avec Backoff Exponentiel
Auteur: HolySheep AI Technical Team
Version: 2.0
"""
import time
import random
import asyncio
from typing import Optional, Dict, Any, Callable
from dataclasses import dataclass, field
from enum import Enum
import aiohttp
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class RetryStrategy(Enum):
"""Stratégies de retry disponibles"""
EXPONENTIAL = "exponential"
LINEAR = "linear"
FIBONACCI = "fibonacci"
@dataclass
class RetryConfig:
"""Configuration du système de retry"""
max_retries: int = 5
base_delay: float = 1.0 # Délai initial en secondes
max_delay: float = 60.0 # Délai maximum
exponential_base: float = 2.0 # Base pour le backoff exponentiel
jitter: bool = True # Ajout de aléatoire pour éviter le thundering herd
retry_on_status: tuple = field(
default_factory=lambda: (429, 500, 502, 503, 504)
)
timeout: float = 120.0 # Timeout total en secondes
@dataclass
class RequestMetrics:
"""Métriques de requête pour le monitoring"""
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
total_retries: int = 0
total_latency_ms: float = 0.0
class HolySheepAIClient:
"""
Client HolySheep AI avec retry automatique et métriques
Endpoint: https://api.holysheep.ai/v1
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
retry_config: Optional[RetryConfig] = None
):
self.api_key = api_key
self.base_url = base_url.rstrip('/')
self.retry_config = retry_config or RetryConfig()
self.metrics = RequestMetrics()
def _calculate_delay(self, attempt: int, strategy: RetryStrategy = RetryStrategy.EXPONENTIAL) -> float:
"""Calcule le délai avant le prochain retry"""
if strategy == RetryStrategy.EXPONENTIAL:
delay = self.retry_config.base_delay * (self.retry_config.exponential_base ** attempt)
elif strategy == RetryStrategy.LINEAR:
delay = self.retry_config.base_delay * (attempt + 1)
else: # FIBONACCI
delay = self.retry_config.base_delay * self._fibonacci(attempt + 2)
# Application du jitter pour éviter le thundering herd
if self.retry_config.jitter:
delay = delay * (0.5 + random.random())
return min(delay, self.retry_config.max_delay)
@staticmethod
def _fibonacci(n: int) -> int:
"""Calcule le nième nombre de Fibonacci"""
if n <= 1:
return n
a, b = 0, 1
for _ in range(n - 1):
a, b = b, a + b
return b
async def _make_request(
self,
session: aiohttp.ClientSession,
endpoint: str,
payload: Dict[str, Any],
headers: Dict[str, str]
) -> Dict[str, Any]:
"""Effectue une requête HTTP vers l'API HolySheep"""
url = f"{self.base_url}/{endpoint.lstrip('/')}"
async with session.post(
url,
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=self.retry_config.timeout)
) as response:
return {
"status": response.status,
"body": await response.json() if response.content_type == 'application/json' else await response.text(),
"headers": dict(response.headers)
}
async def chat_completion(
self,
messages: list,
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 2000,
**kwargs
) -> Dict[str, Any]:
"""
Envoie une requête de chat completion avec retry automatique
Args:
messages: Liste des messages [{"role": "user", "content": "..."}]
model: Modèle à utiliser (gpt-4.1, claude-sonnet-4.5, etc.)
temperature: Température de génération (0.0 - 2.0)
max_tokens: Nombre maximum de tokens à générer
Returns:
Réponse de l'API HolySheep
"""
self.metrics.total_requests += 1
start_time = time.time()
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
**kwargs
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": f"req_{int(time.time() * 1000)}"
}
last_error = None
for attempt in range(self.retry_config.max_retries + 1):
try:
async with aiohttp.ClientSession() as session:
result = await self._make_request(session, "chat/completions", payload, headers)
if result["status"] == 200:
self.metrics.successful_requests += 1
self.metrics.total_latency_ms += (time.time() - start_time) * 1000
logger.info(f"✓ Requête réussie en {attempt} tentative(s)")
return result["body"]
elif result["status"] in self.retry_config.retry_on_status:
last_error = result
# Calcul du délai de retry
if result["status"] == 429:
# Vérification du header Retry-After
retry_after = result["headers"].get("Retry-After")
if retry_after:
delay = float(retry_after)
else:
delay = self._calculate_delay(attempt)
logger.warning(f"⚠ Rate limit (429) - Retry dans {delay:.2f}s (tentative {attempt + 1})")
else:
delay = self._calculate_delay(attempt)
logger.warning(f"⚠ Erreur {result['status']} - Retry dans {delay:.2f}s (tentative {attempt + 1})")
self.metrics.total_retries += 1
await asyncio.sleep(delay)
elif result["status"] == 401:
raise PermissionError("Clé API invalide ou expirée")
elif result["status"] == 400:
raise ValueError(f"Requête invalide: {result['body']}")
else:
raise Exception(f"Erreur inattendue: {result['status']}")
except aiohttp.ClientError as e:
last_error = e
delay = self._calculate_delay(attempt)
logger.warning(f"⚠ Erreur réseau: {e} - Retry dans {delay:.2f}s")
await asyncio.sleep(delay)
# Tous les retries ont échoué
self.metrics.failed_requests += 1
logger.error(f"✗ Échec après {self.retry_config.max_retries + 1} tentatives")
raise RetryExhaustedError(
f"Échec après {self.retry_config.max_retries} retries. "
f"Dernière erreur: {last_error}"
)
def get_metrics(self) -> Dict[str, Any]:
"""Retourne les métriques actuelles"""
success_rate = (
(self.metrics.successful_requests / self.metrics.total_requests * 100)
if self.metrics.total_requests > 0 else 0
)
avg_latency = (
self.metrics.total_latency_ms / self.metrics.successful_requests
if self.metrics.successful_requests > 0 else 0
)
return {
"total_requests": self.metrics.total_requests,
"successful_requests": self.metrics.successful_requests,
"failed_requests": self.metrics.failed_requests,
"total_retries": self.metrics.total_retries,
"success_rate": f"{success_rate:.2f}%",
"avg_latency_ms": f"{avg_latency:.2f}ms"
}
class RetryExhaustedError(Exception):
"""Exception levée quand tous les retries ont échoué"""
pass
Exemple d'utilisation
async def main():
"""Exemple d'utilisation du client HolySheep avec retry"""
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
retry_config=RetryConfig(
max_retries=5,
base_delay=1.0,
max_delay=30.0,
exponential_base=2.0,
jitter=True
)
)
messages = [
{"role": "system", "content": "Tu es un assistant technique expert en容错设计."},
{"role": "user", "content": "Explique-moi comment implémenter un circuit breaker en Python."}
]
try:
response = await client.chat_completion(
messages=messages,
model="gpt-4.1",
temperature=0.7
)
print(f"Réponse: {response['choices'][0]['message']['content']}")
# Affichage des métriques
print(f"\nMétriques: {client.get_metrics()}")
except RetryExhaustedError as e:
print(f"Erreur fatale: {e}")
if __name__ == "__main__":
asyncio.run(main())2. Implémentation du Circuit Breaker
#!/usr/bin/env python3
"""
HolySheep AI Agent - Circuit Breaker Pattern
Implémentation complète pour protéger les appels API en cascade
"""
import time
import threading
from enum import Enum
from typing import Callable, Any, Optional
from dataclasses import dataclass, field
from collections import deque
import logging
logger = logging.getLogger(__name__)
class CircuitState(Enum):
"""États du circuit breaker"""
CLOSED = "closed" # Normal - les requêtes passent
OPEN = "open" # Protection activée - requêtes bloquées
HALF_OPEN = "half_open" # Test - une requête pour tester la récupération
@dataclass
class CircuitBreakerConfig:
"""Configuration du circuit breaker"""
failure_threshold: int = 5 # Nombre d'échecs pour ouvrir le circuit
success_threshold: int = 3 # Nombre de succès pour fermer le circuit (depuis HALF_OPEN)
timeout: float = 30.0 # Temps avant passage en HALF_OPEN (secondes)
half_open_max_calls: int = 1 # Nombre d'appels autorisés en HALF_OPEN
window_size: int = 60 # Fenêtre de temps pour compter les erreurs (secondes)
@dataclass
class CircuitBreakerMetrics:
"""Métriques du circuit breaker"""
total_calls: int = 0
successful_calls: int = 0
failed_calls: int = 0
rejected_calls: int = 0
state_changes: int = 0
last_failure_time: Optional[float] = None
last_success_time: Optional[float] = None
# Pour le calcul du taux d'échec
failure_timestamps: deque = field(default_factory=deque)
class CircuitBreaker:
"""
Circuit Breaker pour protéger les appels API HolySheep
États:
- CLOSED: Fonctionnement normal, toutes les requêtes passent
- OPEN: Après failure_threshold échecs, les requêtes sont rejetées immédiatement
- HALF_OPEN: Après timeout, une requête de test est autorisée
"""
def __init__(self, name: str, config: Optional[CircuitBreakerConfig] = None):
self.name = name
self.config = config or CircuitBreakerConfig()
self.metrics = CircuitBreakerMetrics()
# État du circuit
self._state = CircuitState.CLOSED
self._lock = threading.RLock()
# Compteurs
self._failure_count = 0
self._success_count = 0
self._half_open_calls = 0
# Timestamps
self._opened_at: Optional[float] = None
self._last_state_change = time.time()
# Historique des échecs dans la fenêtre
self._failure_history: deque = deque(maxlen=self.config.failure_threshold * 2)
@property
def state(self) -> CircuitState:
"""Retourne l'état actuel du circuit (avec vérification du timeout)"""
with self._lock:
if self._state == CircuitState.OPEN:
# Vérifier si le timeout est écoulé
if self._opened_at and (time.time() - self._opened_at) >= self.config.timeout:
self._transition_to(CircuitState.HALF_OPEN)
return self._state
def _transition_to(self, new_state: CircuitState):
"""Change l'état du circuit"""
old_state = self._state
self._state = new_state
self._last_state_change = time.time()
self.metrics.state_changes += 1
logger.info(
f"Circuit '{self.name}': {old_state.value} → {new_state.value} "
f"(failures: {self._failure_count}, time since last change: {time.time() - self._last_state_change:.1f}s)"
)
if new_state == CircuitState.OPEN:
self._opened_at = time.time()
self._failure_count = 0
elif new_state == CircuitState.HALF_OPEN:
self._half_open_calls = 0
self._failure_count = 0
self._success_count = 0
elif new_state == CircuitState.CLOSED:
self._failure_count = 0
self._success_count = 0
def _record_success(self):
"""Enregistre un succès"""
with self._lock:
self.metrics.successful_calls += 1
self.metrics.last_success_time = time.time()
if self._state == CircuitState.HALF_OPEN:
self._success_count += 1
if self._success_count >= self.config.success_threshold:
self._transition_to(CircuitState.CLOSED)
elif self._state == CircuitState.CLOSED:
# Réduction progressive du compteur d'échecs
self._failure_count = max(0, self._failure_count - 1)
def _record_failure(self):
"""Enregistre un échec"""
with self._lock:
self.metrics.failed_calls += 1
self.metrics.last_failure_time = time.time()
self._failure_history.append(time.time())
if self._state == CircuitState.HALF_OPEN:
self._transition_to(CircuitState.OPEN)
elif self._state == CircuitState.CLOSED:
self._failure_count += 1
if self._failure_count >= self.config.failure_threshold:
self._transition_to(CircuitState.OPEN)
def can_execute(self) -> bool:
"""Vérifie si une requête peut être exécutée"""
with self._lock:
if self._state == CircuitState.CLOSED:
return True
elif self._state == CircuitState.OPEN:
return False
elif self._state == CircuitState.HALF_OPEN:
return self._half_open_calls < self.config.half_open_max_calls
return False
def _before_call(self):
"""Hook avant l'appel"""
with self._lock:
self.metrics.total_calls += 1
if self._state == CircuitState.HALF_OPEN:
self._half_open_calls += 1
def execute(self, func: Callable[[], Any], *args, **kwargs) -> Any:
"""
Exécute une fonction avec protection du circuit breaker
Args:
func: Fonction à exécuter
*args, **kwargs: Arguments de la fonction
Returns:
Résultat de la fonction
Raises:
CircuitBreakerOpen: Si le circuit est ouvert
Exception: Si la fonction échoue
"""
if not self.can_execute():
self.metrics.rejected_calls += 1
raise CircuitBreakerOpen(
f"Circuit '{self.name}' is OPEN. "
f"Last failure: {self.metrics.last_failure_time}. "
f"Will retry after {self._get_time_until_retry():.1f}s"
)
self._before_call()
try:
result = func(*args, **kwargs)
self._record_success()
return result
except Exception as e:
self._record_failure()
raise
def _get_time_until_retry(self) -> float:
"""Calcule le temps restant avant le prochain retry"""
if self._opened_at:
elapsed = time.time() - self._opened_at
return max(0, self.config.timeout - elapsed)
return 0
def get_status(self) -> dict:
"""Retourne le statut complet du circuit breaker"""
return {
"name": self.name,
"state": self.state.value,
"metrics": {
"total_calls": self.metrics.total_calls,
"successful_calls": self.metrics.successful_calls,
"failed_calls": self.metrics.failed_calls,
"rejected_calls": self.metrics.rejected_calls,
"state_changes": self.metrics.state_changes,
"last_failure": self.metrics.last_failure_time,
"last_success": self.metrics.last_success_time
},
"current_failure_count": self._failure_count,
"time_until_retry": self._get_time_until_retry(),
"last_state_change": self._last_state_change
}
def reset(self):
"""Réinitialise le circuit breaker"""
with self._lock:
self._state = CircuitState.CLOSED
self._failure_count = 0
self._success_count = 0
self._half_open_calls = 0
self._opened_at = None
self._failure_history.clear()
self.metrics = CircuitBreakerMetrics()
logger.info(f"Circuit '{self.name}' has been reset")
class CircuitBreakerOpen(Exception):
"""Exception levée quand le circuit breaker est ouvert"""
pass
class HolySheepAgentWithCircuitBreaker:
"""
Agent IA HolySheep avec Circuit Breaker intégré
Combine le retry automatique avec la protection circuit breaker
"""
def __init__(self, api_key: str, circuit_config: Optional[CircuitBreakerConfig] = None):
self.api_client = HolySheepAIClient(api_key)
self.circuit_breaker = CircuitBreaker(
name="holySheep_api",
config=circuit_config or CircuitBreakerConfig(
failure_threshold=5,
success_threshold=2,
timeout=30.0,
half_open_max_calls=1
)
)
async def chat(self, messages: list, model: str = "gpt-4.1") -> dict:
"""Envoie un message avec protection circuit breaker"""
async def _make_api_call():
return await self.api_client.chat_completion(messages, model=model)
try:
# Exécution protégée par le circuit breaker
result = await asyncio.get_event_loop().run_in_executor(
None,
lambda: self.circuit_breaker.execute(
lambda: asyncio.run(_make_api_call())
)
)
return result
except CircuitBreakerOpen as e:
logger.error(f"Circuit breaker ouvert: {e}")
raise
except RetryExhaustedError as e:
logger.error(f"Tous les retries épuisés: {e}")
raise
def get_circuit_status(self) -> dict:
"""Retourne le statut du circuit breaker"""
return self.circuit_breaker.get_status()
Exemple d'utilisation avancée
async def demo_agent():
"""Démonstration de l'agent avec circuit breaker"""
# Configuration du circuit breaker
circuit_config = CircuitBreakerConfig(
failure_threshold=3, # Ouverture après 3 échecs
success_threshold=2, # Fermeture après 2 succès en HALF_OPEN
timeout=20.0, # Test après 20 secondes
half_open_max_calls=1 # 1 appel de test à la fois
)
# Initialisation de l'agent
agent = HolySheepAgentWithCircuitBreaker(
api_key="YOUR_HOLYSHEEP_API_KEY",
circuit_config=circuit_config
)
messages = [
{"role": "user", "content": "Bonjour, comment vas-tu?"}
]
try:
# Test normal
response = await agent.chat(messages, model="gpt-4.1")
print(f"Réponse: {response['choices'][0]['message']['content']}")
except CircuitBreakerOpen as e:
print(f"Circuit ouvert - Utiliser le fallback: {e}")
# Implémenter la logique de fallback ici
# Affichage du statut
print(f"\nStatut circuit breaker: {agent.get_circuit_status()}")
if __name__ == "__main__":
asyncio.run(demo_agent())3. Integration complète avec patterns de fallback
#!/usr/bin/env python3
"""
HolySheep AI - Agent Workflow complet avec容错设计 multi-niveaux
Inclut: Retry + Circuit Breaker + Fallback + Bulkhead + Rate Limiter
"""
import asyncio
import time
from typing import Optional, List, Dict, Any, Union
from dataclasses import dataclass
from enum import Enum
import logging
from collections import defaultdict
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ModelType(Enum):
"""Types de modèles disponibles"""
PREMIUM = "premium" # GPT-4.1, Claude Sonnet 4.5
STANDARD = "standard" # Gemini 2.5 Flash
ECONOMY = "economy" # DeepSeek V3.2
@dataclass
class ModelConfig:
"""Configuration par modèle"""
name: str
provider: str
type: ModelType
base_cost_per_1k: float
fallback_models: List[str]
max_tokens: int
avg_latency_ms: float
Catalogue des modèles HolySheep avec leurs configurations
HOLYSHEEP_MODELS = {
"gpt-4.1": ModelConfig(
name="gpt-4.1",
provider="OpenAI",
type=ModelType.PREMIUM,
base_cost_per_1k=8.0, # $8/Mtok
fallback_models=["claude-sonnet-4.5", "gemini-2.5-flash"],
max_tokens=128000,
avg_latency_ms=45
),
"claude-sonnet-4.5": ModelConfig(
name="claude-sonnet-4.5",
provider="Anthropic",
type=ModelType.PREMIUM,
base_cost_per_1k=15.0, # $15/Mtok
fallback_models=["gemini-2.5-flash", "gpt-4.1"],
max_tokens=200000,
avg_latency_ms=52
),
"gemini-2.5-flash": ModelConfig(
name="gemini-2.5-flash",
provider="Google",
type=ModelType.STANDARD,
base_cost_per_1k=2.50, # $2.50/Mtok
fallback_models=["deepseek-v3.2", "gpt-4.1"],
max_tokens=1000000,
avg_latency_ms=35
),
"deepseek-v3.2": ModelConfig(
name="deepseek-v3.2",
provider="DeepSeek",
type=ModelType.ECONOMY,
base_cost_per_1k=0.42, # $0.42/Mtok - Économie 85%+
fallback_models=["gemini-2.5-flash"],
max_tokens=64000,
avg_latency_ms=28
)
}
class ResilientAgent:
"""
Agent IA résilient avec容错设计 multi-niveaux
Stratégies de protection:
1. Circuit Breaker - Protection contre les pannes en cascade
2. Retry avec Backoff - Gestion des erreurs temporaires
3. Fallback automatique - Basculement vers modèles alternatifs
4. Rate Limiter - Contrôle du taux de requêtes
5. Bulkhead - Isolation des ressources
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.client = HolySheepAIClient(
api_key=api_key,
retry_config=RetryConfig(
max_retries=3,
base_delay=0.5,
max_delay=10.0
)
)
# Circuit breakers par modèle
self.circuit_breakers: Dict[str, CircuitBreaker] = {}
for model_name in HOLYSHEEP_MODELS:
self.circuit_breakers[model_name] = CircuitBreaker(
name=f"model_{model_name}",
config=CircuitBreakerConfig(
failure_threshold=5,
success_threshold=2,
timeout=30.0
)
)
# Rate limiter
self.rate_limiter = RateLimiter(
requests_per_second=50,
burst_size=100
)
# Métriques globales
self.global_metrics = defaultdict(int)
async def chat_with_fallback(
self,
messages: List[Dict],
primary_model: str = "gpt-4.1",
max_cost_factor: float = 2.0,
context: Optional[str] = None
) -> Dict[str, Any]:
"""
Chat avec fallback automatique multi-niveau
Args:
messages: Messages de conversation
primary_model: Modèle préféré
max_cost_factor: Facteur de coût maximum acceptable vs modèle principal
context: Contexte additionnel pour le logging
Returns:
Réponse de l'API avec métadonnées
Raises:
AllModelsFailedError: Si tous les modèles échouent
"""
start_time = time.time()
attempt_history = []
# Obtenir la configuration du modèle principal
primary_config = HOLYSHEEP_MODELS.get(primary_model)
if not primary_config:
primary_model = "gpt-4.1"
primary_config = HOLYSHEEP_MODELS[primary_model]
# Construire la liste des modèles à essayer (avec fallbacks)
models_to_try = [primary_model]
for fallback in primary_config.fallback_models:
if fallback not in models_to_try:
models_to_try.append(fallback)
last_error = None
for model_name in models_to_try:
model_config = HOLYSHEEP_MODELS.get(model_name)
if not model_config:
continue
# Vérification du coût vs modèle principal
if model_config.base_cost_per_1k > primary_config.base_cost_per_1k * max_cost_factor:
logger.warning(
f"Modèle {model_name} trop coûteux ({model_config.base_cost_per_1k}$/1k) "
f"vs budget ({primary_config.base_cost_per_1k * max_cost_factor}$/1k)"
)
continue
# Vérification du circuit breaker
cb = self.circuit_breakers.get(model_name)
if cb and cb.state == CircuitState.OPEN:
logger.info(f"Circuit breaker ouvert pour {model_name}, skipping")
continue
# Vérification du rate limiter
if not self.rate_limiter.allow_request():
logger.warning(f"Rate limit atteint, waiting...")
await asyncio.sleep(0.1)
continue
logger.info(
f"Tentative avec {model_name} "
f"(latence: {model_config.avg_latency_ms}ms, "
f"coût: ${model_config.base_cost_per_1k}/1k)"
)
try:
response = await self._call_model(model_name, messages)
# Succès - mise à jour des métriques
elapsed_ms = (time.time() - start_time) * 1000
self.global_metrics[f"model_{model_name}_success"] += 1
self.global_metrics["total_success"] += 1
attempt_history.append({
"model": model_name,
"success": True,
"latency_ms": elapsed_ms
})
return {
"content": response["choices"][0]["message"]["content"],
"model_used": model_name,
"latency_ms": elapsed_ms,
"cost_per_1k": model_config.base_cost_per_1k,
"attempt_history": attempt_history,
"success": True
}
except (RetryExhaustedError, CircuitBreakerOpen) as e:
last_error = e
self.global_metrics[f"model_{model_name}_failure"] += 1
attempt_history.append({
"model": model_name,
"success": False,
"error": str(e)
})
logger.warning(f"Échec avec {model_name}: {e}")
continue
except Exception as e:
last_error = e
self.global_metrics[f"model_{model_name}_failure"] += 1
attempt_history.append({
"model": model_name,
"success": False,
"error": str(e)
})
logger.error(f"Erreur inattendue avec {model_name}: {e}")
continue
# Tous les modèles ont échoué
self.global_metrics["total_failure"] += 1
raise AllModelsFailedError(
f"Aucun modèle disponible après {len(models_to_try)} tentatives. "
f"Dernier erreur: {last_error}",
attempt_history=attempt_history
)
async def _call_model(self, model_name: str, messages: List[Dict]) -> Dict:
"""Appelle un modèle spécifique avec protection circuit breaker"""
cb = self.circuit_breakers.get(model_name)
if cb:
def sync_call():
return asyncio.run(self.client.chat_completion(
messages=messages,
model=model_name
))
return cb.execute(sync_call)
else:
return await self.client.chat_completion(
messages=messages,
model=model_name
)
def get_health_report(self) -> Dict[str, Any]:
"""Génère un rapport de santé du système"""
return {
"circuit_breakers": {
name: cb.get_status()
for name, cb in self.circuit_breakers.items()
},
"rate_limiter": self.rate_limiter.get_status(),
"global_metrics": dict(self.global_metrics),
"models_available": [
name for name, cb in self.circuit_breakers.items()
if