Créer un workflow

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. Le workflow coordonne les activités, attend des événements externes, gère les timers et les branchements conditionnels.

En résumé : les activités font le travail, le workflow dirige.

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

Analysons chaque élément :

• @workflows.workflow.define(name="...") — Enregistre la classe comme workflow. Le name est l’identifiant unique utilisé pour déclencher l’exécution
• @workflows.workflow.entrypoint — Marque la méthode qui sera appelée au démarrage du workflow
• self — Le premier paramètre est toujours l’instance de la classe
• await activite() — Le workflow appelle les activités de manière asynchrone

Exemple complet : pipeline de traitement

Voici un workflow réaliste qui orchestre plusieurs activités pour traiter un document :

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. Si le processus tombe entre l’étape 2 et l’étape 3, le workflow reprendra à l’étape 3 sans réexécuter les deux premières.

Durabilité du workflow

Les workflows survivent aux redémarrages, pannes d’infrastructure et erreurs transitoires. La plateforme enregistre chaque événement et reconstruit l’état du workflow lors du replay.

Ce mécanisme repose sur la contrainte de déterminisme : le code du workflow doit toujours produire la même séquence d’opérations pour les mêmes entrées. Les détails seront approfondis dans la leçon 8.

Workflow avec branchement conditionnel

Un workflow peut contenir de la logique conditionnelle :

@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 est que le résultat du branchement doit dépendre du retour d’une activité (déterministe), pas d’un appel direct à un service externe dans le workflow.

Inputs avec Pydantic

Pour des paramètres d’entrée structurés, utilisez des modèles Pydantic :

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

Lorsque vous utilisez un BaseModel, les champs deviennent des clés de premier niveau dans l’input JSON envoyé au workflow.

Bonnes pratiques

• Un workflow = une responsabilité — Ne mélangez pas des logiques métier différentes dans un même workflow
• Nommage clair — Le name du workflow doit décrire son objectif : "traitement_commande", pas "workflow_1"
• Pas d’I/O dans le workflow — Tout appel réseau, fichier ou base de données doit être dans une activité
• Gardez le code du workflow simple — La logique complexe (parsing, transformation) va dans les activités

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