Politique de retry — Workflows Mistral — Orchestration Durable

Pourquoi les retries sont essentiels

Dans un système distribué, les erreurs transitoires sont inévitables : timeouts réseau, limites de débit dépassées, services temporairement indisponibles. Aucune de ces défaillances ne signale un bug dans votre code, et pourtant chacune suffit à faire échouer un traitement si rien ne les absorbe. La politique de retry permet à vos activités de survivre à ces erreurs sans intervention manuelle, la plateforme se chargeant de relancer selon la configuration que vous définissez activité par activité.

Configurer une politique de retry

Le décorateur @workflows.activity() accepte les paramètres qui gouvernent ce comportement. Voici une activité d’appel d’API externe correctement armée.

from datetime import timedelta
import mistralai.workflows as workflows

@workflows.activity(
    start_to_close_timeout=timedelta(minutes=5),
    retry_policy_max_attempts=3,
    retry_policy_backoff_coefficient=2.0,
)
async def appeler_api_externe(url: str, donnees: dict) -> dict:
    """Appelle une API externe avec retry automatique."""
    async with httpx.AsyncClient(timeout=60) as client:
        response = await client.post(url, json=donnees)
        response.raise_for_status()
        return response.json()

Trois paramètres, trois rôles complémentaires.

start_to_close_timeout — Durée maximale d’une seule tentative d’exécution. Si l’activité ne termine pas dans ce délai, elle est considérée comme échouée et une nouvelle tentative est lancée.
retry_policy_max_attempts — Nombre maximum de tentatives. La valeur par défaut dépend de la plateforme. Après ce nombre de tentatives, l’activité est définitivement en échec et le workflow reçoit une erreur.
retry_policy_backoff_coefficient — Multiplicateur du délai entre les tentatives. Avec un coefficient de 2.0, les délais doublent à chaque tentative (1s, 2s, 4s, 8s…). C’est le backoff exponentiel.

Ce backoff mérite qu’on s’y arrête, car son intérêt n’est pas seulement de patienter : il s’agit de laisser au service en difficulté le temps de se rétablir au lieu de le marteler. Le déroulé d’un appel qui finit par aboutir illustre la progression.

Tentative 1 : exécution immédiate
  → Échec (timeout réseau)
  → Attente 1 seconde

Tentative 2 : exécution
  → Échec (service indisponible)
  → Attente 2 secondes (1 × 2.0)

Tentative 3 : exécution
  → Échec (limite de débit)
  → Attente 4 secondes (2 × 2.0)

Tentative 4 : exécution
  → Succès !

Face à un « 429 Too Many Requests », ce mécanisme est particulièrement efficace : une politique qui réessaierait immédiatement ne ferait qu’entretenir la saturation, alors que l’espacement croissant finit par retomber sous le seuil du fournisseur.

Ne jamais omettre start_to_close_timeout

Ce timeout est critique et devrait figurer sur chaque activité, sans exception. Sans lui, une activité qui se bloque — deadlock, connexion pendante, service qui accepte la requête mais ne répond jamais — reste en attente indéfiniment et gèle tout le workflow. Pire : elle ne déclenche aucun retry, puisque de son point de vue elle n’a pas encore échoué.

# MAUVAISE PRATIQUE : pas de timeout
@workflows.activity()
async def activite_dangereuse(params: dict) -> dict:
    # Si le service ne répond jamais, l'activité attend indéfiniment
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.lente.example.com/data")
    return response.json()

# BONNE PRATIQUE : timeout explicite
@workflows.activity(
    start_to_close_timeout=timedelta(minutes=2)
)
async def activite_sure(params: dict) -> dict:
    async with httpx.AsyncClient(timeout=90) as client:
        response = await client.get("https://api.lente.example.com/data")
    return response.json()

Pour fixer la valeur, mesurez d’abord le temps d’exécution normal de l’activité, puis ajoutez une marge de deux à trois fois pour couvrir les cas lents. Veillez enfin à ce que le timeout du client HTTP interne reste inférieur au start_to_close_timeout — dans l’exemple ci-dessus, 90 secondes contre 2 minutes — afin que l’erreur remonte proprement depuis votre code plutôt que d’être tranchée par la plateforme.

L’idempotence, condition non négociable

Puisqu’une activité peut être rejouée plusieurs fois avec les mêmes paramètres, elle doit produire le même effet quel que soit le nombre d’exécutions. Cette propriété s’appelle l’idempotence, et son absence transforme une politique de retry en générateur d’incidents.

@workflows.activity(retry_policy_max_attempts=3)
async def ajouter_credit(user_id: str, montant: float) -> dict:
    """DANGEREUX : si retried, le crédit est ajouté plusieurs fois !"""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"https://api.billing.com/credits/add",
            json={"user_id": user_id, "montant": montant}
        )
    return response.json()

Le scénario redouté est banal : l’API de facturation traite bien la requête, mais la réponse se perd sur le réseau. L’activité conclut à un échec, la plateforme réessaie, et le client est crédité deux fois. Une clé d’idempotence supprime le problème en donnant au service distant le moyen de reconnaître une requête déjà traitée.

@workflows.activity(retry_policy_max_attempts=3)
async def ajouter_credit(user_id: str, montant: float, idempotency_key: str) -> dict:
    """CORRECT : la clé d'idempotence empêche la duplication."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"https://api.billing.com/credits/add",
            json={
                "user_id": user_id,
                "montant": montant,
                "idempotency_key": idempotency_key
            }
        )
    return response.json()

La clé se génère dans le workflow avec workflow.uuid4(), ce qui garantit qu’elle reste identique d’un replay à l’autre — un uuid.uuid4() classique produirait une nouvelle valeur à chaque tentative et annulerait toute la protection. Quand le service distant ne propose pas ce mécanisme, deux autres voies existent : l’upsert en base, avec un INSERT ... ON CONFLICT UPDATE au lieu d’un INSERT seul, ou la vérification préalable, qui consiste à interroger l’état avant d’agir.

Trois profils de configuration

Les valeurs se calibrent selon la nature de l’opération. Un paiement ou une notification doit aboutir vite ou échouer vite : timeout court, trois tentatives, backoff modéré.

@workflows.activity(
    start_to_close_timeout=timedelta(seconds=30),
    retry_policy_max_attempts=3,
    retry_policy_backoff_coefficient=2.0,
)
async def traiter_paiement(params: dict) -> dict:
    # ...
    pass

Un traitement long, génération de rapport ou inférence lourde, inverse la logique : chaque tentative coûte cher en temps de calcul, on en autorise donc moins, avec un espacement plus large.

@workflows.activity(
    start_to_close_timeout=timedelta(minutes=15),
    retry_policy_max_attempts=2,
    retry_policy_backoff_coefficient=3.0,
)
async def generer_rapport_ml(params: dict) -> dict:
    # ...
    pass

Face à une API tierce instable et soumise à du rate limiting, on assume enfin la persévérance : cinq tentatives, quitte à s’étaler sur plusieurs minutes.

@workflows.activity(
    start_to_close_timeout=timedelta(minutes=2),
    retry_policy_max_attempts=5,
    retry_policy_backoff_coefficient=2.0,
)
async def appeler_api_tierce(params: dict) -> dict:
    # ...
    pass

Points clés à retenir

Configurez toujours start_to_close_timeout pour éviter les activités bloquantes
Le backoff exponentiel (retry_policy_backoff_coefficient) augmente le délai entre les tentatives
retry_policy_max_attempts définit le nombre maximum de tentatives avant échec définitif
Les activités avec retry doivent être idempotentes — même résultat si exécutées plusieurs fois
Utilisez des clés d’idempotence ou des upserts pour garantir l’idempotence
Adaptez les timeouts et le nombre de retries au type d’opération (rapide vs. longue, stable vs. instable)