Le jour où mon système RAG a crashé en pleine nuit
Il est 3h du matin quand mon téléphone vibre. Mon système RAG d'entreprise — celui qui permet aux commerciaux de rechercher des documents internes — vient de tomber. 500 collaborateurs paralysés. Le problème ? Une cascade de timeouts vers mon fournisseur d'IA qui avait décidé de ralentir.
Cette nuit-là, j'ai compris l'importance critique du pattern Circuit Breaker. Aujourd'hui, je vais vous partager comment implémenter cette architecture indispensable pour protéger vos intégrations IA.
Qu'est-ce que le Circuit Breaker ?
Le Circuit Breaker est un patron de conception qui monitore les appels à un service externe. Quand le taux d'échec dépasse un seuil critique, le circuit "s'ouvre" — les appels suivants sont bloqués immédiatement pendant un temps défini, évitant ainsi d'épuiser vos ressources et de dégrader votre système.
Cas d'Utilisation : Pic de Service Client E-commerce
Imaginons une boutique en ligne avec un chatbot IA pour le support client. Pendant les soldes, le trafic explose. Sans Circuit Breaker, si votre API IA ralentit :
- Les timeouts s'accumulent
- Votre serveur sature de connexions en attente
- L'expérience utilisateur se dégrade catastrophiquement
- Votre facture API grimpe en flèche avec des requêtes échouées
Avec HolySheep AI, j'ai réduit ma latence à moins de 50ms tout en bénéficiant d'un taux de change avantageux (¥1 = $1). Combiné au Circuit Breaker, c'est la solution idéale pour des pics de charge maîtrisés.
Implémentation Python avec le Framework Tenacity
import asyncio
from tenacity import (
retry, stop_after_attempt, wait_exponential,
circuit_breaker
)
import aiohttp
import time
from dataclasses import dataclass
from typing import Optional
@dataclass
class CircuitState:
failures: int = 0
last_failure_time: Optional[float] = None
state: str = "CLOSED" # CLOSED, OPEN, HALF_OPEN
# Seuils configurables
FAILURE_THRESHOLD: int = 5
RECOVERY_TIMEOUT: int = 30 # secondes
SUCCESS_THRESHOLD: int = 2
class HolySheepCircuitBreaker:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.state = CircuitState()
def _should_allow_request(self) -> bool:
current_time = time.time()
if self.state.state == "CLOSED":
return True
elif self.state.state == "OPEN":
# Vérifie si le timeout de récupération est écoulé
if (current_time - self.state.last_failure_time) >= self.state.RECOVERY_TIMEOUT:
self.state.state = "HALF_OPEN"
print("🔄 Circuit en état HALF_OPEN — test en cours")
return True
return False
elif self.state.state == "HALF_OPEN":
return True
return False
def _record_success(self):
if self.state.state == "HALF_OPEN":
self.state.state = "CLOSED"
self.state.failures = 0
print("✅ Circuit refermé — fonctionnement normal")
elif self.state.state == "CLOSED":
self.state.failures = max(0, self.state.failures - 1)
def _record_failure(self):
self.state.failures += 1
self.state.last_failure_time = time.time()
if self.state.state == "HALF_OPEN":
self.state.state = "OPEN"
print("❌ Échec en HALF_OPEN — Circuit rouvert")
elif self.state.failures >= self.state.FAILURE_THRESHOLD:
self.state.state = "OPEN"
print(f"⚠️ Seuil atteint ({self.state.FAILURE_THRESHOLD}) — Circuit ouvert")
async def chat_completion(self, messages: list, model: str = "deepseek-v3"):
if not self._should_allow_request():
raise Exception("CIRCUIT_OPEN: Service temporairement indisponible")
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": 1000
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=10)
) as response:
if response.status == 200:
self._record_success()
return await response.json()
else:
self._record_failure()
raise Exception(f"API Error: {response.status}")
except asyncio.TimeoutError:
self._record_failure()
raise Exception("TIMEOUT: Réponse de l'API expirée")
except Exception as e:
self._record_failure()
raise
Utilisation
breaker = HolySheepCircuitBreaker("YOUR_HOLYSHEEP_API_KEY")
async def main():
messages = [{"role": "user", "content": "Explique-moi le circuit breaker"}]
try:
result = await breaker.chat_completion(messages, model="deepseek-v3")
print(result)
except Exception as e:
print(f"Erreur gérée: {e}")
asyncio.run(main())
Implémentation TypeScript pour Node.js
interface CircuitBreakerConfig {
failureThreshold: number;
recoveryTimeout: number;
successThreshold: number;
}
enum CircuitState {
CLOSED = 'CLOSED',
OPEN = 'OPEN',
HALF_OPEN = 'HALF_OPEN'
}
class HolySheepAIClient {
private baseUrl = 'https://api.holysheep.ai/v1';
private apiKey: string;
private state: CircuitState = CircuitState.CLOSED;
private failures: number = 0;
private lastFailureTime: number = 0;
private config: CircuitBreakerConfig;
constructor(apiKey: string, config?: Partial) {
this.apiKey = apiKey;
this.config = {
failureThreshold: config?.failureThreshold ?? 5,
recoveryTimeout: config?.recoveryTimeout ?? 30000,
successThreshold: config?.successThreshold ?? 2
};
}
private canAttempt(): boolean {
const now = Date.now();
switch (this.state) {
case CircuitState.CLOSED:
return true;
case CircuitState.OPEN:
if (now - this.lastFailureTime >= this.config.recoveryTimeout) {
this.state = CircuitState.HALF_OPEN;
console.log('🔄 Circuit en mode test (HALF_OPEN)');
return true;
}
return false;
case CircuitState.HALF_OPEN:
return true;
default:
return false;
}
}
private recordSuccess(): void {
if (this.state === CircuitState.HALF_OPEN) {
this.state = CircuitState.CLOSED;
this.failures = 0;
console.log('✅ Circuit refermé avec succès');
} else if (this.state === CircuitState.CLOSED) {
Ressources connexes
Articles connexes