Activités

Une activité est l'endroit où le travail s'effectue réellement : envoyer un e-mail, appeler une API externe, exécuter une requête en base de données, traiter un fichier, générer une réponse à partir d'un modèle d'IA. Tout ce qui interagit avec le monde extérieur, effectue des entrées/sorties ou produit des effets de bord doit se trouver dans une activité, et non dans le workflow lui-même.

Les activités sont des fonctions asynchrones ordinaires. Elles ne sont soumises à aucune contrainte de déterminisme : elles peuvent tout appeler, utiliser n'importe quelle bibliothèque, lire sur le disque ou écrire dans une base de données. La plateforme traite chaque appel d'activité comme une unité atomique. Si une activité échoue, elle est automatiquement relancée selon une politique configurable. Une fois qu'elle réussit, son résultat est consigné dans l'historique d'exécution et le workflow se poursuit.

Comme les résultats sont persistés, une activité qui a réussi n'est jamais réexécutée lors d'un replay ; seul son résultat consigné est utilisé. En revanche, si une activité échoue ou expire en cours d'exécution, la plateforme la relance, potentiellement alors que des effets de bord ont déjà été partiellement appliqués.

Cela signifie que les activités doivent pouvoir être réexécutées sans risque : une nouvelle tentative après un échec partiel laisse le système dans un état cohérent. On parle parfois d'idempotence, mais la propriété précise est que les effets de bord convergent vers le même état observable, et non que les sorties soient identiques octet par octet.

Une activité LLM, par exemple, peut renvoyer une complétion différente lors d'une nouvelle tentative, mais elle reste réexécutable sans risque car son effet de bord (la consignation d'une complétion) n'intervient qu'une seule fois par tentative réussie.

import mistralai.workflows as workflows

@workflows.activity()
async def fetch_user_data(user_id: str) -> dict:
    response = await http_client.get(f"/users/{user_id}")
    return response.json()

import mistralai.workflows as workflows

@workflows.activity()
async def fetch_user_data(user_id: str) -> dict:
    response = await http_client.get(f"/users/{user_id}")
    return response.json()

Les activités sont appelées avec await directement dans le code du workflow :

@workflows.workflow.define(name="user_report")
class UserReportWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, user_id: str) -> dict:
        user = await fetch_user_data(user_id)
        report = await generate_report(user)
        return report

@workflows.workflow.define(name="user_report")
class UserReportWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, user_id: str) -> dict:
        user = await fetch_user_data(user_id)
        report = await generate_report(user)
        return report

Configurez les nouvelles tentatives et les timeouts au niveau du décorateur. Définissez toujours start_to_close_timeout ; sans cela, une activité qui ne répond plus bloque indéfiniment :

from datetime import timedelta

@workflows.activity(
    start_to_close_timeout=timedelta(minutes=5),
    retry_policy_max_attempts=3,
    retry_policy_backoff_coefficient=2.0,
)
async def call_external_api(params: ApiParams) -> ApiResponse:
    async with httpx.AsyncClient() as client:
        response = await client.post(params.url, json=params.data)
        return ApiResponse(data=response.json())

from datetime import timedelta

@workflows.activity(
    start_to_close_timeout=timedelta(minutes=5),
    retry_policy_max_attempts=3,
    retry_policy_backoff_coefficient=2.0,
)
async def call_external_api(params: ApiParams) -> ApiResponse:
    async with httpx.AsyncClient() as client:
        response = await client.post(params.url, json=params.data)
        return ApiResponse(data=response.json())

Pour les activités qui s'exécutent pendant plus de quelques minutes, utilisez des heartbeats pour signaler que l'activité est toujours active. Un heartbeat est un ping périodique que l'activité envoie à la plateforme ; il peut également transporter des données de progression afin qu'une nouvelle tentative reprenne à partir du dernier point de contrôle au lieu de repartir de zéro.

Si la plateforme ne reçoit plus de heartbeats dans le délai du heartbeat_timeout configuré, elle considère que le worker est mort et replanifie l'activité sur un autre worker.

from datetime import timedelta
import mistralai.workflows as workflows
from mistralai.workflows import activity_heartbeat

@workflows.activity(
    start_to_close_timeout=timedelta(minutes=30),
    heartbeat_timeout=timedelta(seconds=30),
)
async def process_large_file(path: str) -> int:
    processed = 0
    async for chunk in read_chunks(path):
        await do_work(chunk)
        processed += 1
        activity_heartbeat(processed)  # ping + checkpoint
    return processed

from datetime import timedelta
import mistralai.workflows as workflows
from mistralai.workflows import activity_heartbeat

@workflows.activity(
    start_to_close_timeout=timedelta(minutes=30),
    heartbeat_timeout=timedelta(seconds=30),
)
async def process_large_file(path: str) -> int:
    processed = 0
    async for chunk in read_chunks(path):
        await do_work(chunk)
        processed += 1
        activity_heartbeat(processed)  # ping + checkpoint
    return processed

Consultez Construire des Workflows > Activités > Principes de base pour connaître toutes les options de configuration.

Utilisez des bibliothèques asynchrones pour les entrées/sorties. Les appels bloquants immobilisent le worker pour toutes les activités concurrentes :

# ✅ Async HTTP
async with httpx.AsyncClient() as client:
    response = await client.post(url, json=data)

# ❌ Blocking HTTP, stalls the worker
response = requests.post(url, json=data)

# ✅ Async HTTP
async with httpx.AsyncClient() as client:
    response = await client.post(url, json=data)

# ❌ Blocking HTTP, stalls the worker
response = requests.post(url, json=data)

Limite de 2 Mo en entrée/sortie. Chaque appel d'activité est limité à 2 Mo en entrée et à 2 Mo en sortie. Consultez Déchargement des payloads pour les payloads plus volumineux.