Créer une activité

Qu’est-ce qu’une activité ?

Une activité est l’unité opérationnelle de base dans un workflow. C’est là que le travail réel se passe : appels API externes, requêtes base de données, envoi d’emails, traitement de fichiers, appels aux modèles IA. La règle de partage est simple à énoncer et vous servira de boussole tout au long du cours — toute opération qui implique des entrées/sorties (I/O) ou des effets de bord doit être encapsulée dans une activité.

Contrairement au workflow qui orchestre, l’activité exécute. C’est la distinction fondamentale, et elle a une conséquence directe : c’est parce que le travail risqué est isolé dans les activités que le moteur peut retenter, mesurer et rejouer sans jamais dupliquer d’effet.

Anatomie d’une activité

Une activité est une fonction Python async décorée avec @workflows.activity() :

import mistralai.workflows as workflows

@workflows.activity()
async def recuperer_profil_utilisateur(user_id: str) -> dict:
    """Récupère le profil d'un utilisateur depuis l'API."""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/users/{user_id}")
        response.raise_for_status()
        return response.json()

Quatre éléments méritent votre attention dans ces quelques lignes. Le décorateur @workflows.activity() enregistre la fonction comme activité auprès du moteur, ce qui la rend appelable depuis un workflow. Le async def est obligatoire : toutes les activités doivent être asynchrones, sans exception. Les paramètres doivent être JSON-sérialisables, donc des types simples — str, int, float, bool, list, dict — ou des modèles Pydantic ; un objet de connexion ou un handle de fichier passé en argument échouera à la sérialisation. La même règle vaut pour le retour, qui traverse le réseau avant d’être inscrit dans le journal d’événements.

Exemples concrets d’activités

Le premier cas, sans doute le plus courant dans ce cours, encapsule un appel à un modèle Mistral. Notez que l’activité renvoie aussi la consommation de tokens : ce genre de métadonnée, une fois dans le journal, vous permet de reconstituer le coût d’une exécution après coup.

import mistralai.workflows as workflows
from mistralai import Mistral

@workflows.activity()
async def generer_resume(texte: str) -> dict:
    """Génère un résumé du texte fourni via Mistral."""
    client = Mistral()
    response = await client.chat.complete_async(
        model="mistral-large-latest",
        messages=[
            {"role": "system", "content": "Vous êtes un assistant spécialisé en résumé de texte."},
            {"role": "user", "content": f"Résumez ce texte en 3 phrases :\n\n{texte}"}
        ]
    )
    return {
        "resume": response.choices[0].message.content,
        "tokens_utilises": response.usage.total_tokens
    }

Le deuxième cas illustre l’accès base de données. Le try/finally n’est pas décoratif : si l’activité échoue et sera retentée, une connexion laissée ouverte à chaque tentative épuise rapidement le pool PostgreSQL.

import mistralai.workflows as workflows
import asyncpg

@workflows.activity()
async def sauvegarder_rapport(rapport: dict) -> dict:
    """Sauvegarde un rapport dans PostgreSQL."""
    conn = await asyncpg.connect("postgresql://user:pass@localhost/db")
    try:
        row = await conn.fetchrow(
            "INSERT INTO rapports (titre, contenu, date_creation) VALUES ($1, $2, NOW()) RETURNING id",
            rapport["titre"],
            rapport["contenu"]
        )
        return {"id": row["id"], "status": "sauvegardé"}
    finally:
        await conn.close()

Le troisième cas, l’envoi d’email, est celui où les effets de bord se voient le mieux : une fois le message parti, aucun retry ne le rattrape.

import mistralai.workflows as workflows
import aiosmtplib

@workflows.activity()
async def envoyer_notification(destinataire: str, sujet: str, corps: str) -> dict:
    """Envoie un email de notification."""
    message = f"Subject: {sujet}\nTo: {destinataire}\n\n{corps}"
    await aiosmtplib.send(
        message,
        hostname="smtp.example.com",
        port=587,
        username="[email protected]",
        password="secret",
        use_tls=True
    )
    return {"status": "envoyé", "destinataire": destinataire}

Caractéristiques fondamentales

Première liberté, et elle est appréciable : contrairement aux workflows, les activités n’ont aucune contrainte de déterminisme. Vous pouvez utiliser datetime.now(), uuid.uuid4(), random.random(), ou tout autre appel non déterministe librement dans une activité. C’est précisément à cela que servent les activités — absorber tout ce que le workflow n’a pas le droit de faire.

Chaque appel à une activité est ensuite traité comme une unité atomique par la plateforme : elle réussit entièrement ou échoue entièrement, sans état intermédiaire visible par le workflow. Si votre activité fait trois choses et que la deuxième plante, le workflow ne voit qu’un échec, pas un tiers de succès — d’où l’intérêt de ne pas empiler cinq opérations distinctes dans une seule fonction.

Une fois qu’une activité a réussi, son résultat est définitivement enregistré dans le journal d’événements du workflow. Lors d’un replay, l’activité n’est jamais réexécutée : seul le résultat enregistré est utilisé. En contrepartie, comme une activité peut être retentée en cas d’échec — timeout réseau, erreur transitoire — elle doit être idempotente : exécutée plusieurs fois avec les mêmes paramètres, elle doit produire le même résultat. Le passage d’un POST à un PUT sur un identifiant stable suffit souvent à obtenir cette propriété, comme ici où l’email sert de clé :

@workflows.activity()
async def creer_ou_mettre_a_jour_client(email: str, nom: str) -> dict:
    """Idempotent : utilise un upsert au lieu d'un insert."""
    async with httpx.AsyncClient() as client:
        response = await client.put(
            f"https://api.crm.com/clients/{email}",
            json={"nom": nom, "email": email}
        )
        return response.json()

Nommer une activité

Le décorateur accepte un nom explicite, ce qui améliore nettement la lisibilité des traces lorsqu’une fonction technique porte un nom anglais que personne d’autre dans l’équipe ne relie au métier :

@workflows.activity(name="Traitement des emails clients")
async def process_customer_emails(batch_id: str) -> dict:
    # ...
    pass

Ce nom apparaîtra dans les traces OpenTelemetry et facilitera le débogage. Prenez l’habitude de le renseigner dès l’écriture : le jour où vous cherchez pourquoi une exécution a duré quarante minutes, vous lirez des libellés parlants plutôt qu’une liste de noms de fonctions.

Contraintes de taille

Deux limites encadrent les échanges : 2 MB maximum pour les paramètres d’entrée et 2 MB maximum pour la valeur de retour. Elles découlent directement de l’event sourcing, puisque ces valeurs sont sérialisées et stockées dans le journal.

La conséquence pratique est nette. Si vous devez traiter des fichiers volumineux — un PDF scanné de 40 MB, une archive d’export — ne les passez pas d’activité en activité. Stockez-les dans un service externe (S3, GCS) et faites circuler uniquement l’URL ou l’identifiant : l’activité télécharge le fichier, le traite, réécrit le résultat dans le stockage et renvoie une référence de quelques centaines d’octets.

Points clés à retenir

• Une activité est une fonction async décorée avec @workflows.activity() qui effectue le travail réel (I/O, API, BDD)
• Les paramètres et le retour doivent être JSON-sérialisables
• Les activités n’ont pas de contrainte de déterminisme
• Une activité réussie n’est jamais réexécutée lors d’un replay
• Les activités doivent être idempotentes (même résultat si exécutées plusieurs fois)
• Limite de 2 MB en entrée et en sortie par activité