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, et elle découle directement du mécanisme de replay que vous avez découvert plus tôt. Lorsqu’un worker reprend un workflow interrompu, il rejoue le code du workflow depuis le début et compare, opération après opération, ce que le code produit avec ce que l’historique a enregistré. Tant que les deux séquences coïncident, la reprise est invisible. Dès qu’elles divergent, la plateforme n’a plus aucun moyen de savoir où elle en était : elle détecte l’incohérence et échoue le workflow avec une erreur de non-déterminisme.

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. Attention à la nuance : les résultats des activités, eux, peuvent parfaitement varier d’une exécution à l’autre, puisqu’ils proviennent du journal. C’est l’ordre des appels qui doit rester identique.

Ce que cela donne dans le code

Prenons un workflow qui envoie un rapport à un utilisateur selon son statut d’abonnement.

@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 alors même qu’il contient un branchement. La raison tient à la source de la condition : profil["premium"] provient du résultat d’une activité, donc du journal. Lors du replay, recuperer_profil renverra exactement la même valeur qu’à la première exécution, et le même chemin sera emprunté.

Comparez maintenant avec cette version, qui semble anodine mais casse tout.

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é"}

Imaginez que le worker tombe à 11 h 58 et redémarre à 12 h 03. Au replay, datetime.now() renvoie une heure d’après-midi alors que l’historique contient un appel à envoyer_email_matin. Le workflow échoue. random.random() produit le même effet, de façon encore plus aléatoire à diagnostiquer.

Les remplacements fournis par le SDK

Le SDK ne se contente pas d’interdire : il fournit des équivalents déterministes pour les trois opérations les plus courantes.

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

Le principe est le même que pour les activités : la valeur est calculée puis enregistrée dans le journal lors de la première exécution, et simplement rejouée ensuite. L’exemple précédent devient alors parfaitement sûr.

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

La frontière à mémoriser

Sont proscrits dans le code d’un workflow : datetime.now(), datetime.utcnow() et time.time() ; uuid.uuid4() et uuid.uuid1() ; random.random(), random.randint() et random.choice() ; tout appel réseau direct (httpx, aiohttp, requests) ; toute lecture ou écriture de fichier (open(), aiofiles) ; tout accès base de données ; os.environ, car les variables d’environnement peuvent changer entre deux replays ; enfin threading et multiprocessing, le parallélisme interne étant géré par la plateforme.

Reste autorisé tout ce qui est reproductible : appeler des activités via await, utiliser workflow.now(), workflow.uuid4() et workflow.random(), écrire de la logique conditionnelle ou des boucles dont la condition dépend de résultats d’activités, manipuler des données en mémoire (calculs, chaînes de caractères), et journaliser — le logging étant considéré comme un effet de bord acceptable.

Diagnostiquer une erreur de non-déterminisme

Quand l’erreur tombe, quatre vérifications suffisent le plus souvent à la localiser.

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

Ce dernier point est le plus sournois : une librairie de sérialisation qui trie par identifiant généré aléatoirement, ou un client HTTP instancié au chargement du module, suffisent à faire échouer un replay sans qu’aucune ligne suspecte n’apparaisse dans votre workflow.

La parade la plus efficace consiste à garder le workflow aussi maigre que possible et à repousser toute la logique complexe dans les activités. Un workflow qui se contente d’enchaîner trois appels n’a mécaniquement aucune occasion d’être non déterministe.

# 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é