Créer un worker

Le rôle du worker

Le worker est le processus Python qui exécute concrètement vos workflows et activités. Il se connecte à la plateforme Mistral, écoute les tâches en attente, et les exécute localement sur votre machine ou serveur.

Sans worker en cours d’exécution, vos workflows ne peuvent pas s’exécuter. Le worker est le lien entre la plateforme d’orchestration (côté Mistral) et votre code (côté local).

Créer un worker minimal

Voici le code minimal pour un worker fonctionnel :

import asyncio
import mistralai.workflows as workflows

# Importez vos workflows et activités
@workflows.activity()
async def dire_bonjour(nom: str) -> dict:
    return {"message": f"Bonjour, {nom} !"}

@workflows.workflow.define(name="salutation_workflow")
class SalutationWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, nom: str) -> dict:
        resultat = await dire_bonjour(nom)
        return resultat

# Démarrer le worker
async def main() -> None:
    await workflows.run_worker([SalutationWorkflow])

if __name__ == "__main__":
    asyncio.run(main())

La fonction clé est workflows.run_worker(). Elle prend en paramètre la liste des classes de workflows que ce worker peut exécuter.

Ce que fait run_worker()

Quand vous appelez workflows.run_worker([MonWorkflow]), voici ce qui se passe :

1. Le worker s’authentifie auprès de la plateforme Mistral avec votre clé API
2. Il enregistre les workflows et activités déclarés
3. Il ouvre une connexion persistante et attend les tâches
4. Quand une exécution est déclenchée, le worker exécute le code localement
5. Les résultats sont renvoyés à la plateforme pour persistance

Le worker reste actif tant que le processus Python tourne. Utilisez Ctrl+C pour l’arrêter proprement.

Worker avec plusieurs workflows

Un worker peut exécuter plusieurs workflows :

import asyncio
import mistralai.workflows as workflows

# --- Activités ---

@workflows.activity()
async def extraire_texte(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
    return {"texte": response.text}

@workflows.activity()
async def analyser_sentiment(texte: str) -> dict:
    client = Mistral()
    response = await client.chat.complete_async(
        model="mistral-large-latest",
        messages=[
            {"role": "system", "content": "Analysez le sentiment : positif, négatif ou neutre."},
            {"role": "user", "content": texte}
        ]
    )
    return {"sentiment": response.choices[0].message.content}

@workflows.activity()
async def envoyer_alerte(message: str) -> dict:
    # Envoi vers Slack, email, etc.
    return {"status": "envoyé"}

# --- Workflows ---

@workflows.workflow.define(name="analyse_document")
class AnalyseDocument:
    @workflows.workflow.entrypoint
    async def run(self, url: str) -> dict:
        extraction = await extraire_texte(url)
        sentiment = await analyser_sentiment(extraction["texte"])
        return sentiment

@workflows.workflow.define(name="surveillance_mentions")
class SurveillanceMentions:
    @workflows.workflow.entrypoint
    async def run(self, url: str, seuil: str) -> dict:
        extraction = await extraire_texte(url)
        sentiment = await analyser_sentiment(extraction["texte"])
        if sentiment["sentiment"] == "négatif":
            await envoyer_alerte(f"Mention négative détectée : {url}")
        return sentiment

# --- Worker ---

async def main() -> None:
    await workflows.run_worker([
        AnalyseDocument,
        SurveillanceMentions,
    ])

if __name__ == "__main__":
    asyncio.run(main())

Structure de fichiers recommandée

Pour un projet en production, séparez le code en modules :

mon-projet/
├── activities/
│   ├── __init__.py
│   ├── extraction.py      # extraire_texte()
│   ├── analyse.py          # analyser_sentiment()
│   └── notifications.py    # envoyer_alerte()
├── workflows/
│   ├── __init__.py
│   ├── analyse_document.py
│   └── surveillance.py
└── worker.py               # Point d'entrée unique

Le fichier worker.py importe tous les workflows :

import asyncio
import mistralai.workflows as workflows

from workflows.analyse_document import AnalyseDocument
from workflows.surveillance import SurveillanceMentions

async def main() -> None:
    await workflows.run_worker([
        AnalyseDocument,
        SurveillanceMentions,
    ])

if __name__ == "__main__":
    asyncio.run(main())

Lancer le worker

uv run python worker.py

Le worker affiche un message de connexion et reste actif en attente de tâches. Vous devez voir quelque chose comme :

Worker connected. Listening for tasks...

Worker en production

Pour un déploiement en production, quelques recommandations :

Utiliser un gestionnaire de processus

Utilisez systemd, supervisor ou pm2 pour garder le worker actif :

# Avec pm2 (si Node.js est disponible)
pm2 start "uv run python worker.py" --name workflow-worker

# Avec systemd (créer un fichier .service)

Variables d’environnement

Ne codez jamais la clé API en dur. Utilisez les variables d’environnement :

import os

# Le SDK lit automatiquement MISTRAL_API_KEY
# Pas besoin de le passer explicitement

Logs

Le worker émet des logs via le module logging standard de Python. Configurez le niveau selon vos besoins :

import logging
logging.basicConfig(level=logging.INFO)

Points clés à retenir

• Le worker est le processus qui exécute vos workflows et activités localement
• workflows.run_worker([...]) prend la liste des classes de workflows à exécuter
• Un seul worker peut exécuter plusieurs workflows différents
• Le worker reste actif et écoute les tâches tant que le processus tourne
• En production, utilisez un gestionnaire de processus pour maintenir le worker actif
• La clé API est lue automatiquement depuis MISTRAL_API_KEY