La règle de déterminisme — Workflows Mistral — Orchestration Durable

Pourquoi le déterminisme est obligatoire

Le déterminisme est la contrainte la plus importante des workflows Mistral. Pour comprendre pourquoi, rappelons le mécanisme de replay : lorsqu’un worker reprend un workflow interrompu, il rejoue le code du workflow depuis le début et compare les opérations avec l’historique enregistré.

Si votre workflow produit une séquence d’opérations différente lors du replay, la plateforme détecte une incohérence et échoue le workflow avec une erreur de non-déterminisme.

Ce que déterminisme signifie concrètement

Un workflow déterministe produit exactement la même séquence d’opérations chaque fois qu’il est exécuté avec les mêmes entrées. Les résultats des activités peuvent varier (ils proviennent du journal), mais l’ordre des appels doit être identique.

Exemple : workflow déterministe

@workflows.workflow.define(name="workflow_deterministe")
class WorkflowDeterministe:
    @workflows.workflow.entrypoint
    async def run(self, user_id: str) -> dict:
        # Toujours la même séquence d'appels pour le même input
        profil = await recuperer_profil(user_id)
        
        if profil["premium"]:
            rapport = await generer_rapport_complet(user_id)
        else:
            rapport = await generer_rapport_basique(user_id)
        
        await envoyer_email(profil["email"], rapport)
        return {"status": "terminé"}

Ce workflow est déterministe car le branchement if/else dépend du résultat d’une activité (recuperer_profil), dont le résultat est enregistré dans le journal. Lors du replay, le même résultat sera réutilisé, donc le même chemin sera suivi.

Exemple : workflow NON déterministe

import datetime
import random

@workflows.workflow.define(name="workflow_non_deterministe")
class WorkflowNonDeterministe:
    @workflows.workflow.entrypoint
    async def run(self, user_id: str) -> dict:
        # INTERDIT : datetime.now() change à chaque exécution
        if datetime.datetime.now().hour < 12:
            await envoyer_email_matin(user_id)
        else:
            await envoyer_email_apres_midi(user_id)
        
        # INTERDIT : random.random() donne un résultat différent à chaque fois
        if random.random() > 0.5:
            await envoyer_notification(user_id)
        
        return {"status": "terminé"}

Ce workflow va échouer lors du replay car datetime.now() et random.random() donnent des résultats différents à chaque exécution.

Les remplacements du SDK

Le SDK fournit des alternatives déterministes pour les opérations courantes non déterministes :

from mistralai.workflows import workflow

# Obtenir l'heure actuelle (déterministe lors du replay)
current_time = workflow.now()          # Au lieu de datetime.now()

# Générer un identifiant unique (déterministe lors du replay)
request_id = workflow.uuid4()          # Au lieu de uuid.uuid4()

# Obtenir un nombre aléatoire (déterministe lors du replay)
rand_value = workflow.random()         # Au lieu de random.random()

Ces fonctions retournent un résultat qui est enregistré dans le journal lors de la première exécution, puis rejoué lors des exécutions suivantes.

Exemple corrigé

from mistralai.workflows import workflow

@workflows.workflow.define(name="workflow_corrige")
class WorkflowCorrige:
    @workflows.workflow.entrypoint
    async def run(self, user_id: str) -> dict:
        # CORRECT : workflow.now() est déterministe lors du replay
        heure_actuelle = workflow.now()
        if heure_actuelle.hour < 12:
            await envoyer_email_matin(user_id)
        else:
            await envoyer_email_apres_midi(user_id)
        
        # CORRECT : workflow.random() est déterministe lors du replay
        if workflow.random() > 0.5:
            await envoyer_notification(user_id)
        
        # CORRECT : workflow.uuid4() est déterministe lors du replay
        trace_id = workflow.uuid4()
        await enregistrer_trace(str(trace_id), user_id)
        
        return {"status": "terminé", "trace_id": str(trace_id)}

Règles d’or du déterminisme

Ce qui est INTERDIT dans un workflow

datetime.now(), datetime.utcnow(), time.time()
uuid.uuid4(), uuid.uuid1()
random.random(), random.randint(), random.choice()
Tout appel réseau direct (httpx, aiohttp, requests)
Toute lecture/écriture de fichier (open(), aiofiles)
Tout accès base de données
os.environ (les variables d’environnement peuvent changer entre les replays)
threading, multiprocessing (le parallélisme interne est géré par la plateforme)

Ce qui est AUTORISÉ dans un workflow

Appeler des activités via await
Utiliser workflow.now(), workflow.uuid4(), workflow.random()
Logique conditionnelle basée sur les résultats d’activités
Boucles dont la condition dépend de résultats d’activités
Opérations sur des données en mémoire (calculs, manipulation de chaînes, etc.)
Logging (considéré comme un effet de bord acceptable)

Comment diagnostiquer une erreur de déterminisme

Lorsqu’un workflow échoue avec une erreur de non-déterminisme, vérifiez systématiquement :

Présence d’appels non déterministes — Recherchez datetime.now, uuid.uuid4, random.random dans le code du workflow
I/O directes — Tout appel réseau ou fichier doit être dans une activité
Variables d’environnement — Si vous utilisez os.environ pour un branchement, déplacez la lecture dans une activité
Bibliothèques tierces — Certaines librairies font des appels réseau ou utilisent le hasard en interne

Astuce : séparer logique déterministe et non déterministe

Une bonne pratique consiste à garder le workflow le plus simple possible et à déplacer toute logique complexe dans les activités :

# BONNE PRATIQUE : workflow simple, logique dans les activités
@workflows.workflow.define(name="pipeline_simple")
class PipelineSimple:
    @workflows.workflow.entrypoint
    async def run(self, params: dict) -> dict:
        etape1 = await preparer_donnees(params)
        etape2 = await traiter_donnees(etape1)
        etape3 = await finaliser(etape2)
        return etape3

Points clés à retenir

Le code du workflow doit être déterministe : mêmes entrées = même séquence d’opérations
Utilisez workflow.now(), workflow.uuid4(), workflow.random() au lieu des équivalents Python standard
Toute opération I/O (réseau, fichier, BDD) doit être dans une activité, jamais dans le workflow
Les branchements conditionnels sont autorisés s’ils dépendent de résultats d’activités
Une violation du déterminisme provoque un échec du workflow lors du replay
En cas de doute, déplacez la logique dans une activité