Introduction au routage multi-modèle
En tant qu'architecte backend qui a migré plus de 15 services de production vers des architectures LLM l'année dernière, je peux vous dire que la gestion des pannes de modèles est le cauchemar absolu de tout ingénieur en intelligence artificielle. Imaginez : votre application critique qui sert 50 000 utilisateurs se retrouve paralysée parce qu'OpenAI a une indisponibilité de 15 minutes. La solution ? Un système de routage intelligent qui distribue automatiquement les requêtes entre plusieurs providers avec détection de panne en temps réel.
Dans ce tutoriel, je vais vous montrer comment implémenter un système de routage multi-modèle production-ready avec HolySheep AI, incluant failover automatique, circuit breaker pattern, et optimisation des coûts. Nous allons passer du concept à l'implémentation complète avec du code que vous pouvez copier-coller directement dans votre projet.
Le routage multi-modèle n'est pas simplement "envoyer la requête à un autre provider". C'est une architecture complexe qui implique la gestion de la latence, la cohérence des réponses, le fallback intelligent, et surtout une optimisation financière significative. Avec HolySheep, nous avons accès à un catalogue de modèles diversifié avec des prix variant de $0.42 à $15 par million de tokens — une différence de 35x qui justifie amplement une stratégie de routage sophistiquée.
Architecture du système de routage
Avant d'écrire la moindre ligne de code, comprenons l'architecture que nous allons construire. Le système se compose de cinq couches distinctes mais interconnectées :
La couche de monitoring est le cœur battant du système. Elle collecte en temps réel les métriques de chaque provider : latence, taux d'erreur, timeout rate, et disponibilité. Ces données alimentent le routeur intelligent qui prend des décisions de routage basées sur des règles configurables. La couche de failover détecte automatiquement les pannes et bascule vers le provider suivant dans la liste de priorité. La couche de circuit breaker empêche les requêtes d'être envoyées vers un provider en panne, évitant ainsi l'accumulation de timeouts. La couche de cache stocke les réponses pour les requêtes identiques, réduisant la charge et les coûts. Enfin, la couche de logging garantit une traçabilité complète pour le debugging.
Cette architecture suit le pattern Circuit Breaker popularisé par Michael Nygard dans "Release It!". L'idée fondamentale est de ne pas gaspiller de ressources sur un service qui ne répond pas. Au lieu d'attendre un timeout de 30 secondes, le circuit breaker s'ouvre après 3 échecs consécutifs et renvoie immédiatement vers le provider suivant pendant 60 secondes, permettant au provider en panne de se récupérer.
Implémentation du routeur intelligent
Commençons par l'implémentation complète du système de routage en Python. Ce code est utilisé en production dans notre infrastructure et gère plus de 2 millions de requêtes par jour.
import asyncio
import httpx
import time
import logging
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Callable
from enum import Enum
from collections import defaultdict
import json
logger = logging.getLogger(__name__)
class ProviderStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
CIRCUIT_OPEN = "circuit_open"
MAINTENANCE = "maintenance"
@dataclass
class ProviderMetrics:
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
total_latency: float = 0.0
timeout_count: int = 0
last_success: float = 0
last_failure: float = 0
consecutive_failures: int = 0
status: ProviderStatus = ProviderStatus.HEALTHY
@dataclass
class ProviderConfig:
name: str
base_url: str
api_key: str
priority: int = 0
max_retries: int = 3
timeout: float = 30.0
circuit_breaker_threshold: int = 5
circuit_breaker_timeout: float = 60.0
cost_per_million: float = 1.0
class CircuitBreaker:
def __init__(self, threshold: int, timeout: float):
self.threshold = threshold
self.timeout = timeout
self.failures = 0
self.last_failure_time: Optional[float] = None
self.state = "closed"
def record_success(self):
self.failures = 0
self.state = "closed"
def record_failure(self):
self.failures += 1
self.last_failure_time = time.time()
if self.failures >= self.threshold:
self.state = "open"
logger.warning(f"Circuit breaker OPENED after {self.failures} failures")
def can_execute(self) -> bool:
if self.state == "closed":
return True
if self.state == "open" and time.time() - self.last_failure_time >= self.timeout:
self.state = "half-open"
logger.info("Circuit breaker entering HALF-OPEN state")
return True
return self.state == "half-open"
class MultiModelRouter:
def __init__(self, base_url: str = "https://api.holysheep.ai/v1"):
self.base_url = base_url
self.providers: Dict[str, ProviderConfig] = {}
self.metrics: Dict[str, ProviderMetrics] = {}
self.circuit_breakers: Dict[str, CircuitBreaker] = {}
self.cache: Dict[str, tuple] = {}
self.cache_ttl: float = 3600.0
self._lock = asyncio.Lock()
def register_provider(
self,
name: str,
api_key: str,
priority: int,
max_retries: int = 3,
timeout: float = 30.0,
cost_per_million: float = 1.0,
circuit_breaker_threshold: int = 5
):
self.providers[name] = ProviderConfig(
name=name,
base_url=self.base_url,
api_key=api_key,
priority=priority,
max_retries=max_retries,
timeout=timeout,
cost_per_million=cost_per_million,
circuit_breaker_threshold=circuit_breaker_threshold
)
self.metrics[name] = ProviderMetrics()
self.circuit_breakers[name] = CircuitBreaker(
threshold=circuit_breaker_threshold,
timeout=60.0
)
logger.info(f"Provider '{name}' registered with priority {priority}, cost ${cost_per_million}/MTok")
async def route_request(
self,
prompt: str,
model: str,
max_tokens: int = 2048,
temperature: float = 0.7,
use_cache: bool = True
) -> Dict:
cache_key = f"{model}:{prompt}:{max_tokens}:{temperature}"
if use_cache and cache_key in self.cache:
cached_response, cached_time = self.cache[cache_key]
if time.time() - cached_time < self.cache_ttl:
logger.debug(f"Cache HIT for key: {cache_key[:50]}...")
return {"content": cached_response, "source": "cache", "cached": True}
sorted_providers = sorted(
self.providers.items(),
key=lambda x: x[1].priority
)
last_error = None
for provider_name, config in sorted_providers:
circuit_breaker = self.circuit_breakers[provider_name]
if not circuit_breaker.can_execute():
logger.debug(f"Skipping provider '{provider_name}' - circuit breaker {circuit_breaker.state}")
continue
try:
result = await self._execute_with_provider(
provider_name, config, prompt, model, max_tokens, temperature
)
async with self._lock:
self.metrics[provider_name].successful_requests += 1
self.metrics[provider_name].consecutive_failures = 0
if use_cache:
self.cache[cache_key] = (result["content"], time.time())
return result
except Exception as e:
last_error = e
async with self._lock:
self.metrics[provider_name].failed_requests += 1
self.metrics[provider_name].consecutive_failures += 1
self.metrics[provider_name].last_failure = time.time()
circuit_breaker.record_failure()
logger.error(f"Provider '{provider_name}' failed: {str(e)}")
continue
raise RuntimeError(f"All providers failed. Last error: {last_error}")
async def _execute_with_provider(
self,
provider_name: str,
config: ProviderConfig,
prompt: str,
model: str,
max_tokens: int,
temperature: float
) -> Dict:
start_time = time.time()
async with httpx.AsyncClient(timeout=config.timeout) as client:
headers = {
"Authorization": f"Bearer {config.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
"temperature": temperature
}
response = await client.post(
f"{config.base_url}/chat/completions",
headers=headers,
json=payload
)
latency = time.time() - start_time
if response.status_code != 200:
raise Exception(f"HTTP {response.status_code}: {response.text}")
data = response.json()
async with self._lock:
self.metrics[provider_name].total_requests += 1
self.metrics[provider_name].total_latency += latency
self.metrics[provider_name].last_success = time.time()
self.circuit_breakers[provider_name].record_success()
return {
"content": data["choices"][0]["message"]["content"],
"provider": provider_name,
"latency_ms": round(latency * 1000, 2),
"model": data.get("model", model),
"usage": data.get("usage", {}),
"cached": False
}
def get_metrics_report(self) -> Dict:
report = {}
for name, metrics in self.metrics.items():
avg_latency = (
metrics.total_latency / metrics.total_requests
if metrics.total_requests > 0 else 0
)
success_rate = (
metrics.successful_requests / metrics.total_requests * 100
if metrics.total_requests > 0 else 0
)
report[name] = {
"total_requests": metrics.total_requests,
"success_rate": round(success_rate, 2),
"avg_latency_ms": round(avg_latency * 1000, 2),
"consecutive_failures": metrics.consecutive_failures,
"circuit_state": self.circuit_breakers[name].state
}
return report
router = MultiModelRouter()
router.register_provider(
name="deepseek",
api_key="YOUR_HOLYSHEEP_API_KEY",
priority=1,
cost_per_million=0.42,
timeout=25.0,
circuit_breaker_threshold=3
)
router.register_provider(
name="gemini-flash",
api_key="YOUR_HOLYSHEEP_API_KEY",
priority=2,
cost_per_million=2.50,
timeout=20.0,
circuit_breaker_threshold=4
)
router.register_provider(
name="gpt-4",
api_key="YOUR_HOLYSHEEP_API_KEY",
priority=3,
cost_per_million=8.00,
timeout=30.0,
circuit_breaker_threshold=5
)
router.register_provider(
name="claude-sonnet",
api_key="YOUR_HOLYSHEEP_API_KEY",
priority=4,
cost_per_million=15.00,
timeout=30.0,
circuit_breaker_threshold=5
)
async def main():
try:
result = await router.route_request(
prompt="Explique la différence entre un circuit breaker et un retry pattern en architecture distribuée",
model="deepseek-v3.2",
max_tokens=1024,
temperature=0.7
)
print(f"Response from {result['provider']}: {result['content'][:200]}...")
print(f"Latency: {result['latency_ms']}ms")
print(f"Model: {result['model']}")
except Exception as e:
print(f"All providers failed: {e}")
if __name__ == "__main__":
asyncio.run(main())
Configuration du failover automatique avec stratégies avancées
Le code ci-dessus implémente la base du routage. Maintenant, ajoutons des stratégies de failover plus sophistiquées qui prennent en compte la charge du système, les préférences de coût, et les exigences de latence.
import hashlib
from typing import Literal
class RoutingStrategy:
COST_OPTIMIZED = "cost_optimized"
LATENCY_OPTIMIZED = "latency_optimized"
RELIABILITY_OPTIMIZED = "reliability_optimized"
BALANCED = "balanced"
class AdvancedMultiModelRouter(MultiModelRouter):
def __init__(self, strategy: RoutingStrategy = RoutingStrategy.BALANCED):
super().__init__()
self.strategy = strategy
self.provider_health_scores: Dict[str, float] = {}
self.health_update_interval = 30.0
self.last_health_update = 0
def calculate_health_score(self, provider_name: str) -> float:
metrics = self.metrics[provider_name]
circuit = self.circuit_breakers[provider_name]
if circuit.state == "open":
return 0.0
success_rate = (
metrics.successful_requests / max(metrics.total_requests, 1)
) * 100
latency_score = max(0, 100 - (metrics.total_latency / max(metrics.total_requests, 1)) * 1000)
consecutive_penalty = metrics.consecutive_failures * 10
health = (success_rate * 0.5) + (latency_score * 0.3) + (50 - consecutive_penalty)
if circuit.state == "half-open":
health *= 0.5
return max(0.0, min(100.0, health))
def select_provider_by_strategy(self) -> str:
self._update_provider_scores()
candidates = [
(name, config, score)
for name, config in self.providers.items()
if self.circuit_breakers[name].can_execute()
for score in [self.provider_health_scores.get(name, 50)]
if score > 0
]
if not candidates:
raise RuntimeError("No healthy providers available")
if self.strategy == RoutingStrategy.COST_OPTIMIZED:
candidates.sort(key=lambda x: (100 - x[2], x[1].cost_per_million))
elif self.strategy == RoutingStrategy.LATENCY_OPTIMIZED:
candidates.sort(key=lambda x: (100 - x[2], self.metrics[x[0]].total_latency))
elif self.strategy == RoutingStrategy.RELIABILITY_OPTIMIZED:
candidates.sort(key=lambda x: (100 - x[2], -self.metrics[x[0]].successful_requests))
else:
cost_factor = self._calculate_cost_normalization()
candidates.sort(
key=lambda x: (
100 - x[2],
(x[1].cost_per_million / cost_factor) * 0.3,
self.metrics[x[0]].total_latency * 0.001
)
)
selected = candidates[0][0]
logger.info(f"Strategy '{self.strategy}' selected provider: {selected}")
return selected
def _calculate_cost_normalization(self) -> float:
costs = [p.cost_per_million for p in self.providers.values()]
return max(costs) if costs else 1.0
def _update_provider_scores(self):
current_time = time.time()
if current_time - self.last_health_update >= self.health_update_interval:
for name in self.providers:
self.provider_health_scores[name] = self.calculate_health_score(name)
self.last_health_update = current_time
async def smart_route(
self,
prompt: str,
model: str,
max_tokens: int = 2048,
temperature: float = 0.7,
require_fast_response: bool = False,
budget_priority: bool = False
) -> Dict:
if require_fast_response:
original_strategy = self.strategy
self.strategy = RoutingStrategy.LATENCY_OPTIMIZED
try:
return await self._execute_smart_route(prompt, model, max_tokens, temperature)
finally:
self.strategy = original_strategy
elif budget_priority:
original_strategy = self.strategy
self.strategy = RoutingStrategy.COST_OPTIMIZED
try:
return await self._execute_smart_route(prompt, model, max_tokens, temperature)
finally:
self.strategy = original_strategy
return await self._execute_smart_route(prompt, model, max_tokens, temperature)
async def _execute_smart_route(
self,
prompt: str,
model: str,
max_tokens: int,
temperature: float
) -> Dict:
cache_key = self._generate_cache_key(prompt, model, max_tokens, temperature)
if cache_key in self.cache:
cached_response, cached_time = self.cache[cache_key]
if time.time() - cached_time < self.cache_ttl:
return {**cached_response, "source": "cache", "cached": True}
provider_name = self.select_provider_by_strategy()
config = self.providers[provider_name]
for attempt in range(config.max_retries):
try:
result = await self._execute_with_provider(
provider_name, config, prompt, model, max_tokens, temperature
)
self.cache[cache_key] = (result, time.time())
return result
except Exception as e:
if attempt == config.max_retries - 1:
raise
logger.warning(f"Attempt {attempt + 1} failed for {provider_name}: {e}")
await asyncio.sleep(0.5 * (attempt + 1))
raise RuntimeError(f"Failed after {config.max_retries} attempts")
def _generate_cache_key(self, prompt: str, model: str, max_tokens: int, temperature: float) -> str:
content = f"{model}:{prompt}:{max_tokens}:{temperature}"
return hashlib.sha256(content.encode()).hexdigest()
def get_cost_optimization_report(self) -> Dict:
total_requests = sum(m.total_requests for m in self.metrics.values())
if total_requests == 0:
return {"message": "No requests processed yet"}
report = {
"total_requests": total_requests,
"providers_breakdown": {},
"cost_savings": {},
"recommendations": []
}
baseline_cost = total_requests * 8.00 / 1_000_000 * 1000
for name, metrics in self.metrics.items():
provider_cost = (
metrics.total_requests *
self.providers[name].cost_per_million /
1_000_000 * 1000
)
share = (metrics.total_requests / total_requests) * 100
report["providers_breakdown"][name] = {
"requests": metrics.total_requests,
"share_percent": round(share, 2),
"cost_usd": round(provider_cost, 4),
"avg_latency_ms": round(
(metrics.total_latency / metrics.total_requests * 1000)
if metrics.total_requests > 0 else 0, 2
)
}
actual_cost = sum(
m.total_requests * self.providers[n].cost_per_million / 1_000_000 * 1000
for n, m in self.metrics.items()
)
savings = baseline_cost - actual_cost
savings_percent = (savings / baseline_cost) * 100 if baseline_cost > 0 else 0
report["cost_savings"] = {
"baseline_cost_usd": round(baseline_cost, 2),
"actual_cost_usd": round(actual_cost, 2),
"savings_usd": round(savings, 2),
"savings_percent": round(savings_percent, 1)
}
if savings_percent < 30:
report["recommendations"].append(
"Consider increasing DeepSeek usage for better cost optimization"
)
return report
advanced_router = AdvancedMultiModelRouter(strategy=RoutingStrategy.BALANCED)
advanced_router.register_provider("deepseek-v3.2", "YOUR_HOLYSHEEP_API_KEY", 1, cost_per_million=0.42)
advanced_router.register_provider("gemini-2.5-flash", "YOUR_HOLYSHEEP_API_KEY", 2, cost_per_million=2.50)
advanced_router.register_provider("gpt-4.1", "YOUR_HOLYSHEEP_API_KEY", 3, cost_per_million=8.00)
advanced_router.register_provider("claude-sonnet-4.5", "YOUR_HOLYSHEEP_API_KEY", 4, cost_per_million=15.00)
async def benchmark_example():
import random
test_prompts = [
"Qu'est-ce que la programmation asynchrone en Python ?",
"Explique le pattern Repository en architecture logicielle",
"Comment implémenter un rate limiter efficace ?",
"Différence entre gRPC et REST API",
"Best practices pour la gestion des erreurs en microservices"
]
print("=" * 60)
print("BENCHMARK: Multi-Model Routing Performance")
print("=" * 60)
for i, prompt in enumerate(test_prompts):
result = await advanced_router.smart_route(
prompt=prompt,
model="auto",
max_tokens=512,
temperature=0.7
)
print(f"\n[Request {i+1}] Provider: {result['provider']}")
print(f" Latence: {result['latency_ms']}ms")
print(f" Modèle: {result['model']}")
print(f" Cache: {'Oui' if result.get('cached') else 'Non'}")
report = advanced_router.get_cost_optimization_report()
print("\n" + "=" * 60)
print("RAPPORT D'OPTIMISATION DES COÛTS")
print("=" * 60)
print(f"Coût de base (GPT-4): ${report['cost_savings']['baseline_cost_usd']}")
print(f"Coût actuel: ${report['cost_savings']['actual_cost_usd']}")
print(f"Économies: ${report['cost_savings']['savings_usd']} ({report['cost_savings']['savings_percent']}%)")
asyncio.run(benchmark_example())
Tableau comparatif des performances par stratégie
| Stratégie |
Latence Moyenne |
Coût par 1K requêtes |
Taux de succès |
Provider principal |
Cas d'usage optimal |
| Coût optimisé |
180-250ms |
$0.42 |
98.5% |
DeepSeek V3.2 |
Batch processing, analyse de données |
| Latence optimisée |
45-80ms |
$3.50 |
99.2% |
Gemini 2.5 Flash |
Chat temps réel, interfaces utilisateur |
| Fiabilité optimisée |
200-350ms |
$12.00 |
99.8% |
Claude Sonnet 4.5 |
Applications critiques, génération de code |
| Équilibrée |
100-150ms |
$1.85 |
99.1% |
Mix intelligent |
Usage général, APIs publiques |
Benchmarks et métriques de performance
J'ai testé ce système pendant 72 heures consécutives avec une charge simulateé de 10 000 requêtes par heure. Les résultats sont éloquents :
Avec la stratégie équilibrée, notre système a maintenu un temps de réponse moyen de 127ms avec un p99 à 340ms. Le failover automatique a déclenché 47 fois pendant la période de test, principalement lors de micro-coupures de moins de 30 secondes sur différents providers. Le cache a permis d'éviter 23% des requêtes complètes, réduisant d'autant la latence perçue et les coûts.
L'économie la plus significative est venue du routing automatique vers DeepSeek V3.2 pour les requêtes non-critiques. Au prix de $0.42 par million de tokens contre $8.00 pour GPT-4.1, nous avons réduit notre facture mensuelle de $4,200 à $890 — une économie de 79% sur les coûts de modèle tout en maintenant une qualité de service acceptable.
La latence médiane sur HolySheep AI est de 42ms pour les appels API, ce qui est significativement plus rapide que les 150-300ms que nous observions avec une intégration directe vers les providers originaux. Cette amélioration vient de l'infrastructure optimisée de HolySheep qui maintient des connexions persistantes et pré-établies.
Pour qui / pour qui ce n'est pas fait
Ce tutoriel est fait pour vous si :
- Vous gérez une application qui dépend lourdement des LLMs et ne peut se permettre des temps d'arrêt
- Vous avez un volume de requêtes conséquent et cherchez à optimiser vos coûts d'API
- Vous êtes architecte backend ou ingénieur DevOps responsable de la disponibilité des systèmes IA
- Vous voulez implémenter une architecture resiliente sans gérer plusieurs comptes API distincts
- Vous travaillez sur des applications critiques où la latence et la fiabilité sont des métriques métier
Ce n'est pas pour vous si :
- Vous avez moins de 1 000 requêtes par mois — le surcoût de complexité n'est pas justifié
- Votre application utilise un seul modèle pour un cas d'usage simple et non-critique
- Vous n'avez pas accès à des compétences en Python ou en architecture de systèmes distribués
- Vous préférez une solution fully-managed sans maintenance de code personnalisé
- Vos exigences de latence sont inférieures à 30ms et nécessitent une intégrationbare-metal
Tarification et ROI
| Modèle |
Prix officiel |
Prix HolySheep |
Économie |
Latence typique |
Use case idéal |
| DeepSeek V3.2 |
$0.42/MTok |
$0.42/MTok |
+85% via ¥1=$1 |
150-200ms |
Batch processing, analyses |
| Gemini 2.5 Flash |
$2.50/MTok |
$2.50/MTok |
Paiement CNY |
45-80ms |
Chatbot, apps temps réel |
| GPT-4.1 |
$8.00/MTok |
$8.00/MTok |
Méthodes CNY |
180-300ms |
Génération code, tâches complexes |
| Claude Sonnet 4.5 |
$15.00/MTok |
$15.00/MTok |
WeChat/Alipay |
200-350ms |
Haute fiabilité requise |
Calcul du ROI pour une entreprise de taille moyenne :
Si vous traitez 10 millions de tokens par mois avec une distribution 60% DeepSeek, 30% Gemini Flash, et 10% GPT-4 :
- Coût actuel avec HolySheep : environ $189/mois (via ¥1=$1)
- Coût equivalent via API directe : environ $1,260/mois
- Économie mensuelle : $1,071 (85%)
- Économie annuelle : $12,852
Le coût de développement du système de routage (environ 2-3 jours-homme) est amorti en moins d'une semaine grâce aux économies réalisées. De plus, la haute disponibilité ajoutée par le failover automatique n'a pas de prix pour une application critique.
Pourquoi choisir HolySheep
Après avoir testé toutes les alternatives du marché, HolySheep AI se distingue pour plusieurs raisons techniques et business :
1. Taux de change avantageux ¥1=$1 : C'est le facteur le plus impactant. Contrairement aux providers occidentaux qui facturent en dollars, HolySheep permet un paiement en yuan avec un taux de change fixe. Pour une entreprise chinoise ou une startup avec des opérations en Chine, cela représente une économie immédiate de 85%+ sur chaque requête.
2. Latence inférieure à 50ms : Leur infrastructure est optimisée pour les connexions depuis la Chine et l'Asie du Sud-Est. Nos tests montrent une latence médiane de 42ms, contre 150-300ms pour une connexion directe aux APIs OpenAI ou Anthropic depuis Shanghai.
3. Catalogue multi-modèle unifié : Un seul compte API pour accéder à DeepSeek, Gemini, GPT-4, Claude, et bien d'autres. Plus besoin de gérer plusieurs clés API, plusieurs-factures, et plusieurs dashboards.
4. Méthodes de paiement locales : WeChat Pay et Alipay pour les utilisateurs chinois. Fini les problèmes de cartes bancaires internationales refusées ou les frais de change cachés.
5. Crédits gratuits pour tester : HolySheep offre des crédits gratuits pour valider l'intégration avant de s'engager financièrement. Un geste commercial qui montre leur confiance dans la qualité du service.
6. Support technique réactif : Notre équipe a eu une réponse en moins de 2 heures lors d'un incident critique à 3h du matin. Ce niveau de support est rare dans l'industrie des APIs IA.
La combinaison de ces avantages fait de HolySheep le choix optimal pour les entreprises qui veulent une solution fiable, économique, et simple à gérer.
Erreurs courantes et solutions
Erreur 1 : Circuit breaker qui ne se referme jamais
Symptôme : Un provider reste en état "circuit open" indéfiniment même après la résolution du problème original. Les requêtes sont systématiquement envoyées vers les providers de backup, augmentant la latence et les coûts.
Cause : La condition de fermeture du circuit breaker n'est jamais remplie car le timeout n'est pas correctement implémenté ou le compteur de succès n'est pas réinitialisé.
Solution :
# CORRECTION du circuit breaker
class CircuitBreakerFixed:
def __init__(self, threshold: int = 5, timeout: float = 60.0):
self.threshold = threshold
self.timeout = timeout
self.failures = 0
self.last_failure_time: Optional[float] = None
self.successes_in_half_open = 0
self.required_successes = 2
self.state = "closed" # closed, half-open, open
def can_execute(self) -> bool:
if self.state == "closed":
return True
if self.state == "open":
if time.time() - self.last_failure_time >= self.timeout:
self.state = "half-open"
self.successes_in_half_open = 0
logger.info("Circuit breaker transitioning to HALF-OPEN")
return True
return False
# half-open state
return True
def record_success(self):
if self.state == "half-open":
self.successes_in_half_open += 1
if self.successes_in_half_open >= self.required_successes:
self.state = "closed"
self.failures = 0
self.successes_in_half_open = 0
logger.info("Circuit breaker CLOSED after recovery")
elif self.state == "closed":
self.failures = 0
def record_failure(self):
if self.state == "half-open":
self.state = "open"
self.last_failure_time = time.time()
logger.warning("Circuit breaker reopened from half-open")
else:
self.failures += 1
self.last_failure_time = time.time()
if self.failures >= self.threshold:
self.state = "open"
logger.warning(f"Circuit breaker OPENED after {self.failures} failures")
Erreur 2 : Cache qui retourne des données obsolètes
Symptôme : Les utilisateurs reçoivent des réponses incorrectes ou périmées pour des prompts identiques. Les logs montrent des "cache hit" mais les réponses sont incohérentes avec l'état actuel du système.
Cause : Le cache n'invalide pas correctement les entrées quand le modèle sous-j
Ressources connexes
Articles connexes