Créer un workflow — Workflows Mistral — Orchestration Durable

Le rôle du workflow

Le workflow est la couche d’orchestration de votre application. Il définit ce qui doit se passer, dans quel ordre, et sous quelles conditions — sans exécuter le travail lui-même. Concrètement, il coordonne les activités, attend des événements externes, gère les timers et les branchements conditionnels, mais il ne touche jamais au réseau ni au disque.

La formule tient en une phrase que vous pouvez garder en tête à chaque relecture de code : les activités font le travail, le workflow dirige. Si vous voyez un httpx ou un open() dans une classe de workflow, quelque chose est à déplacer.

Structure d’un workflow

Un workflow est une classe Python décorée avec @workflows.workflow.define(), contenant une méthode d’entrée décorée avec @workflows.workflow.entrypoint :

import mistralai.workflows as workflows

@workflows.workflow.define(name="mon_premier_workflow")
class MonPremierWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, nom: str) -> dict:
        """Point d'entrée du workflow."""
        resultat = await saluer(nom)
        return resultat

Le décorateur @workflows.workflow.define(name="...") enregistre la classe comme workflow, et le name qu’il reçoit est l’identifiant unique que vous utiliserez pour déclencher une exécution — c’est cette chaîne, pas le nom de la classe Python, que verra l’appelant. Le décorateur @workflows.workflow.entrypoint marque la méthode appelée au démarrage. Le premier paramètre self reste l’instance de la classe, comme dans n’importe quel objet Python, et chaque appel d’activité se fait avec await.

Exemple complet : pipeline de traitement

Voici un workflow réaliste qui orchestre plusieurs activités pour traiter un document. Observez la répartition : les trois fonctions du haut contiennent tout le code qui touche l’extérieur, la classe du bas ne fait qu’enchaîner des await et assembler un résultat.

import mistralai.workflows as workflows
from mistralai import Mistral

# --- Activités ---

@workflows.activity()
async def extraire_texte(url_document: str) -> dict:
    """Télécharge et extrait le texte d'un document."""
    async with httpx.AsyncClient() as client:
        response = await client.get(url_document)
        texte = response.text
    return {"texte": texte, "longueur": len(texte)}

@workflows.activity()
async def analyser_avec_mistral(texte: str) -> dict:
    """Analyse le texte avec Mistral pour en extraire les points clés."""
    client = Mistral()
    response = await client.chat.complete_async(
        model="mistral-large-latest",
        messages=[
            {"role": "system", "content": "Extrayez les 5 points clés de ce texte."},
            {"role": "user", "content": texte}
        ]
    )
    return {"analyse": response.choices[0].message.content}

@workflows.activity()
async def sauvegarder_resultat(analyse: dict) -> dict:
    """Sauvegarde le résultat de l'analyse."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.example.com/rapports",
            json=analyse
        )
        return {"id_rapport": response.json()["id"]}

# --- Workflow ---

@workflows.workflow.define(name="pipeline_analyse_document")
class PipelineAnalyseDocument:
    @workflows.workflow.entrypoint
    async def run(self, url_document: str) -> dict:
        """Orchestre l'extraction, l'analyse et la sauvegarde."""
        # Étape 1 : Extraire le texte
        extraction = await extraire_texte(url_document)

        # Étape 2 : Analyser avec Mistral
        analyse = await analyser_avec_mistral(extraction["texte"])

        # Étape 3 : Sauvegarder
        sauvegarde = await sauvegarder_resultat(analyse)

        return {
            "url_source": url_document,
            "longueur_texte": extraction["longueur"],
            "id_rapport": sauvegarde["id_rapport"]
        }

Chaque await dans le workflow constitue un point de persistance. Cela vaut la peine d’être vérifié mentalement sur cet exemple : si le processus tombe entre l’étape 2 et l’étape 3, le document ne sera pas retéléchargé et le modèle ne sera pas rappelé — donc pas de nouvelle facture de tokens — le workflow reprendra directement à la sauvegarde.

Durabilité du workflow

Les workflows survivent aux redémarrages, pannes d’infrastructure et erreurs transitoires, parce que la plateforme enregistre chaque événement et reconstruit l’état du workflow lors du replay. Cette reconstruction n’est fiable qu’à une condition, déjà évoquée : le code du workflow doit toujours produire la même séquence d’opérations pour les mêmes entrées, c’est la contrainte de déterminisme. Les détails seront approfondis dans la leçon 8.

Workflow avec branchement conditionnel

Un workflow n’est pas condamné à une séquence linéaire. Il peut contenir de la logique conditionnelle, ici sur un traitement de commande où le chemin dépend de l’état du stock :

@workflows.workflow.define(name="traitement_commande")
class TraitementCommande:
    @workflows.workflow.entrypoint
    async def run(self, commande: dict) -> dict:
        # Vérifier le stock
        stock = await verifier_stock(commande["produit_id"])

        if stock["disponible"]:
            # Chemin nominal
            paiement = await traiter_paiement(commande["montant"])
            expedition = await preparer_expedition(commande)
            await envoyer_confirmation(commande["email"], expedition)
            return {"status": "confirmée", "tracking": expedition["tracking"]}
        else:
            # Chemin alternatif
            await notifier_rupture_stock(commande["email"], commande["produit_id"])
            return {"status": "rupture_stock"}

Les branchements if/else sont parfaitement supportés. La seule contrainte tient au matériau de la décision : le résultat du branchement doit dépendre du retour d’une activité — donc d’une valeur enregistrée dans le journal, rejouée à l’identique — et non d’un appel direct à un service externe depuis le workflow. Ici, verifier_stock interroge l’inventaire dans une activité, et le if ne lit qu’un dictionnaire déjà persisté.

Inputs avec Pydantic

Un workflow qui prend six paramètres positionnels devient vite illisible. Pour des entrées structurées, utilisez des modèles Pydantic, qui apportent en prime la validation des types au démarrage de l’exécution :

from pydantic import BaseModel

class ParametresRapport(BaseModel):
    type_rapport: str
    inclure_details: bool = False
    date_debut: str
    date_fin: str

@workflows.workflow.define(name="generation_rapport")
class GenerationRapport:
    @workflows.workflow.entrypoint
    async def run(self, params: ParametresRapport) -> dict:
        donnees = await collecter_donnees(params.type_rapport, params.date_debut, params.date_fin)
        
        if params.inclure_details:
            donnees = await enrichir_donnees(donnees)
        
        rapport = await generer_rapport_pdf(donnees)
        return {"url_rapport": rapport["url"]}

Un point de vigilance au moment de déclencher ce workflow : lorsque vous utilisez un BaseModel, les champs deviennent des clés de premier niveau dans l’input JSON envoyé au workflow. Vous enverrez donc {"type_rapport": ..., "date_debut": ...} et non un objet imbriqué sous une clé params.

Bonnes pratiques

Quatre habitudes font la différence sur la durée. Tenez-vous d’abord à un workflow pour une responsabilité : mélanger la facturation et la relance commerciale dans la même classe rend impossible de rejouer l’une sans l’autre. Soignez ensuite le nommage, puisque le name du workflow est un identifiant public appelé depuis d’autres systèmes — "traitement_commande" se maintient, "workflow_1" ne se maintient pas. N’introduisez aucune I/O dans le workflow : tout appel réseau, fichier ou base de données appartient à une activité. Gardez enfin le code du workflow simple, en déportant la logique complexe — parsing, transformations, calculs lourds — dans les activités, où elle est testable isolément et sans contrainte de déterminisme.

Points clés à retenir

Un workflow est une classe décorée avec @workflows.workflow.define() qui orchestre des activités
La méthode @workflows.workflow.entrypoint est le point d’entrée appelé au démarrage
Chaque await d’une activité est un point de persistance (reprise automatique en cas de panne)
Les branchements conditionnels et les boucles sont supportés
Utilisez des modèles Pydantic pour des paramètres d’entrée structurés
Le workflow ne doit contenir aucune opération I/O directe