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. La leçon précédente vous a appris à reconnaître les erreurs réessayables ; celle-ci vous apprend à réessayer intelligemment, c’est-à-dire sans aggraver la situation que vous tentez de corriger.

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. Le jitter mérite qu’on s’y arrête, car son absence produit une panne classique : cinquante instances de votre service reçoivent un 429 à la même seconde, attendent toutes exactement une seconde, puis repartent toutes ensemble et se refont refuser en bloc. En ajoutant un délai aléatoire, vous étalez ce redémarrage — c’est le remède au thundering herd.

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.6-terra",
                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)

Notez la dernière tentative : on propage l’erreur d’origine au lieu de l’étouffer. L’appelant reçoit ainsi le vrai diagnostic — 429 ou 500 — et peut décider quoi en faire.

Utiliser la bibliothèque tenacity

Écrire cette boucle à la main devient vite répétitif dès que vous avez cinq fonctions à protéger. Pour une gestion plus avancée des retries, tenacity est l’outil de référence : le décorateur déclare la politique et le corps de la fonction ne contient plus que l’appel métier. Vous y retrouvez les mêmes ingrédients, mais nommés et paramétrables — la liste des exceptions à réessayer, l’attente exponentielle bornée à soixante secondes, l’arrêt après cinq tentatives, et un log automatique avant chaque pause.

# 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.6-terra") -> 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 retry résout les incidents de quelques secondes. Il devient nuisible quand la panne dure : chaque requête utilisateur consomme alors cinq tentatives et trente secondes d’attente avant d’échouer, vos threads s’accumulent et c’est votre propre service qui tombe. Le circuit breaker empêche votre application de continuer à appeler une API défaillante.

Il fonctionne comme un disjoncteur électrique, avec trois états. Fermé, tout fonctionne normalement et les appels passent. Ouvert, l’API est considérée en panne et les appels sont bloqués immédiatement, sans même tenter le réseau — l’utilisateur reçoit une erreur en quelques millisecondes plutôt qu’après un timeout. Semi-ouvert, une fois le délai de repos écoulé, un appel test est envoyé pour vérifier si l’API est revenue : s’il réussit le circuit se referme, s’il échoue il se rouvre pour un nouveau cycle.

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")

Le compteur d’erreurs se remet à zéro dès qu’un appel réussit : ce sont bien des échecs consécutifs qui ouvrent le circuit, pas un cumul historique. L’intégration côté client tient alors en trois gestes — demander l’autorisation, signaler le succès, signaler l’échec.

# 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.6-terra",
            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. Observez le constructeur : le client est créé avec max_retries=0, ce qui désactive les retries internes du SDK. Sans cela, vos trois tentatives se multiplieraient par les deux du SDK et vous attendriez bien plus longtemps que prévu. Le timeout de trente secondes borne chaque appel individuel, la boucle gère les reprises, et le circuit breaker surveille la santé globale du service.

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.6-terra",
                    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."))

Avant de passer à la suite, testez ce client en simulant une panne : coupez le réseau, lancez une dizaine d’appels et vérifiez que les derniers échouent instantanément au lieu de patienter. Si ce n’est pas le cas, votre seuil est trop haut pour le volume traité.

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 tenacity pour 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)