Retry, backoff exponentiel et circuit breaker
Retry, backoff exponentiel et circuit breaker
Les erreurs transitoires sont normales en production. Le SDK OpenAI gère automatiquement 2 retries, mais pour une résilience maximale, vous devez implémenter vos propres stratégies.
Le backoff exponentiel
Le principe est simple : chaque retry attend plus longtemps que le précédent. Cela évite de surcharger l’API quand elle est sous pression.
import time
import random
from openai import OpenAI, RateLimitError, InternalServerError, APIConnectionError
client = OpenAI()
def appel_avec_backoff(prompt: str, max_retries: int = 5) -> str:
"""Appel API avec backoff exponentiel et jitter."""
for tentative in range(max_retries):
try:
response = client.responses.create(
model="gpt-5.3",
input=prompt
)
return response.output_text
except (RateLimitError, InternalServerError, APIConnectionError) as e:
if tentative == max_retries - 1:
raise # Dernière tentative, on propage l'erreur
# Backoff exponentiel : 1s, 2s, 4s, 8s, 16s
base_delay = 2 ** tentative
# Jitter aléatoire pour éviter le thundering herd
jitter = random.uniform(0, base_delay * 0.5)
delay = base_delay + jitter
print(f"Tentative {tentative + 1}/{max_retries} échouée. "
f"Retry dans {delay:.1f}s...")
time.sleep(delay)
raise RuntimeError("Ne devrait jamais arriver")
# Utilisation
resultat = appel_avec_backoff("Expliquez le backoff exponentiel.")
print(resultat)Utiliser la bibliothèque tenacity
Pour une gestion plus avancée des retries, tenacity est l’outil de référence :
# pip install tenacity
from tenacity import (
retry,
stop_after_attempt,
wait_exponential,
retry_if_exception_type,
before_sleep_log,
)
from openai import RateLimitError, InternalServerError, APIConnectionError
import logging
logger = logging.getLogger(__name__)
@retry(
retry=retry_if_exception_type(
(RateLimitError, InternalServerError, APIConnectionError)
),
wait=wait_exponential(multiplier=1, min=1, max=60),
stop=stop_after_attempt(5),
before_sleep=before_sleep_log(logger, logging.WARNING),
)
def appel_resilient(prompt: str, model: str = "gpt-5.3") -> str:
"""Appel API avec retry automatique via tenacity."""
response = client.responses.create(
model=model,
input=prompt
)
return response.output_text
# Utilisation — les retries sont transparents
resultat = appel_resilient("Bonjour !")
print(resultat)Le pattern circuit breaker
Le circuit breaker empêche votre application de continuer à appeler une API défaillante. Il a trois états :
- Fermé : tout fonctionne normalement
- Ouvert : l’API est en panne, les appels sont bloqués immédiatement
- Semi-ouvert : un appel test est envoyé pour vérifier si l’API est revenue
import time
from enum import Enum
class EtatCircuit(Enum):
FERME = "ferme"
OUVERT = "ouvert"
SEMI_OUVERT = "semi_ouvert"
class CircuitBreaker:
"""Circuit breaker pour les appels API."""
def __init__(self, seuil_erreurs: int = 5, timeout_reset: float = 30.0):
self.seuil = seuil_erreurs
self.timeout_reset = timeout_reset
self.erreurs_consecutives = 0
self.etat = EtatCircuit.FERME
self.derniere_erreur = 0.0
def peut_appeler(self) -> bool:
"""Vérifie si un appel est autorisé."""
if self.etat == EtatCircuit.FERME:
return True
if self.etat == EtatCircuit.OUVERT:
if time.monotonic() - self.derniere_erreur > self.timeout_reset:
self.etat = EtatCircuit.SEMI_OUVERT
return True # Un appel test
return False
# Semi-ouvert : un seul appel autorisé
return True
def enregistrer_succes(self):
"""Enregistre un appel réussi."""
self.erreurs_consecutives = 0
self.etat = EtatCircuit.FERME
def enregistrer_erreur(self):
"""Enregistre un échec."""
self.erreurs_consecutives += 1
self.derniere_erreur = time.monotonic()
if self.erreurs_consecutives >= self.seuil:
self.etat = EtatCircuit.OUVERT
print(f"Circuit OUVERT apres {self.seuil} erreurs consecutives")
# Intégration avec le client OpenAI
circuit = CircuitBreaker(seuil_erreurs=3, timeout_reset=30.0)
def appel_avec_circuit_breaker(prompt: str) -> str:
"""Appel API protégé par un circuit breaker."""
if not circuit.peut_appeler():
raise RuntimeError(
f"Circuit ouvert — API indisponible. "
f"Retry dans {circuit.timeout_reset}s"
)
try:
response = client.responses.create(
model="gpt-5.3",
input=prompt
)
circuit.enregistrer_succes()
return response.output_text
except (RateLimitError, InternalServerError, APIConnectionError) as e:
circuit.enregistrer_erreur()
raise
# Utilisation
try:
resultat = appel_avec_circuit_breaker("Bonjour !")
print(resultat)
except RuntimeError as e:
print(f"Erreur : {e}")Combiner les trois patterns
En production, combinez backoff, circuit breaker et timeout :
class APIClient:
"""Client API résilient combinant tous les patterns."""
def __init__(self):
self.client = OpenAI(timeout=30.0, max_retries=0)
self.circuit = CircuitBreaker(seuil_erreurs=5, timeout_reset=60.0)
def appeler(self, prompt: str, max_retries: int = 3) -> str:
if not self.circuit.peut_appeler():
raise RuntimeError("Service temporairement indisponible")
for tentative in range(max_retries):
try:
response = self.client.responses.create(
model="gpt-5.3",
input=prompt
)
self.circuit.enregistrer_succes()
return response.output_text
except (RateLimitError, InternalServerError) as e:
self.circuit.enregistrer_erreur()
if tentative < max_retries - 1:
delay = (2 ** tentative) + random.uniform(0, 1)
time.sleep(delay)
else:
raise
raise RuntimeError("Échec apres toutes les tentatives")
# Utilisation
api = APIClient()
print(api.appeler("Expliquez la résilience en ingénierie logicielle."))Points clés à retenir
- Le SDK gère 2 retries par défaut — ajustez avec
max_retries - Le backoff exponentiel avec jitter évite le thundering herd
- Le circuit breaker protège votre application contre les pannes prolongées
- Utilisez
tenacitypour une gestion déclarative des retries en production - Combinez les trois patterns pour une résilience maximale
- Ne réessayez jamais les erreurs 400/401/403 (erreurs client)