Observabilité et OpenTelemetry — Workflows Mistral — Orchestration Durable

Pourquoi observer vos workflows

Un workflow en production peut enchaîner des dizaines d’activités, appeler plusieurs services externes, et durer de quelques secondes à plusieurs jours. Sans observabilité, diagnostiquer un problème revient à chercher une aiguille dans une botte de foin.

La plateforme Workflows de Mistral intègre nativement OpenTelemetry (OTel) pour capturer les détails d’exécution : chaque activité, chaque erreur, chaque durée est enregistrée comme un span dans une trace distribuée.

Concepts clés d’OpenTelemetry

Traces

Une trace représente l’ensemble du parcours d’une exécution de workflow, du déclenchement initial au résultat final. Elle est composée de multiples spans.

Spans

Un span est une opération individuelle dans la trace. Chaque activité génère automatiquement un span contenant :

Le nom de l’activité
L’heure de début et de fin
La durée d’exécution
Le statut (succès ou erreur)
Les éventuels attributs personnalisés

Relation parent-enfant

Les spans forment une hiérarchie : le span du workflow est le parent, et les spans des activités sont les enfants. Cela permet de visualiser la structure complète de l’exécution.

Nommer les activités pour la lisibilité

Le paramètre name du décorateur d’activité définit le nom qui apparaîtra dans les traces :

import mistralai.workflows as workflows

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

@workflows.activity(name="Analyse de sentiment Mistral")
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."},
            {"role": "user", "content": texte}
        ]
    )
    return {"sentiment": response.choices[0].message.content}

@workflows.activity(name="Envoi de la notification email")
async def envoyer_notification(destinataire: str, contenu: str) -> dict:
    # ...
    return {"status": "envoyé"}

Des noms clairs facilitent considérablement le débogage. Comparez dans une vue de trace :

fetch_user_data vs. Récupération du profil utilisateur — le second est immédiatement compréhensible.

Récupérer les traces par programmation

Le SDK Mistral fournit trois méthodes pour accéder aux données de trace :

Trace OpenTelemetry brute

from mistralai import Mistral

client = Mistral()

# Récupérer les données OTel complètes
trace = await client.get_workflow_execution_trace_otel(
    execution_id="exec_abc123def456"
)

# La trace contient tous les spans au format OTel standard
for span in trace.spans:
    print(f"Activité : {span.name}")
    print(f"Durée : {span.duration_ms} ms")
    print(f"Statut : {span.status}")
    print()

Résumé de trace

Pour une vue synthétique sans les détails bruts :

summary = await client.get_workflow_execution_trace_summary(
    execution_id="exec_abc123def456"
)

print(f"Workflow : {summary.workflow_name}")
print(f"Durée totale : {summary.total_duration_ms} ms")
print(f"Nombre d'activités : {summary.activity_count}")
print(f"Erreurs : {summary.error_count}")

Événements détaillés

Pour un filtrage fin des événements :

events = await client.get_workflow_execution_trace_events(
    execution_id="exec_abc123def456",
    include_internal_events=False  # Exclure les événements internes de la plateforme
)

for event in events:
    print(f"Type : {event.type}")
    print(f"Timestamp : {event.timestamp}")
    print(f"Données : {event.data}")
    print()

Le paramètre include_internal_events=False filtre les événements techniques de la plateforme (scheduling, replay) pour ne garder que les événements métier (démarrage/fin d’activité, erreurs).

Échantillonnage de traces

En production, collecter 100 % des traces peut générer un volume de données important. La plateforme supporte l’échantillonnage par parent (parent-based sampling).

Comment ça fonctionne

Le sampling est contrôlé par le header traceparent envoyé lors du déclenchement du workflow. Si le flag de sampling est activé dans le traceparent, la trace sera collectée. Sinon, elle sera ignorée.

Forcer la collecte d’une trace

Pour diagnostiquer un problème spécifique, vous pouvez forcer la collecte en passant un traceparent avec le flag de sampling activé :

execution = await client.workflows.execute_workflow_async(
    workflow_identifier="mon_workflow",
    input={"donnees": "test"},
    # Le flag "01" dans le traceparent active le sampling
    headers={"traceparent": "00-trace_id-span_id-01"}
)

Désactiver la collecte

Inversement, pour les exécutions de routine où vous ne souhaitez pas collecter de traces :

# Le flag "00" désactive le sampling
headers={"traceparent": "00-trace_id-span_id-00"}

Monitoring en production

Métriques clés à surveiller

Pour un workflow en production, surveillez ces indicateurs :

Taux de succès — Pourcentage de workflows qui terminent sans erreur
Durée moyenne — Temps moyen d’exécution d’un workflow complet
Durée P95/P99 — Durée des workflows les plus lents (95e et 99e percentile)
Taux de retry — Fréquence des retries par activité (un taux élevé indique un service instable)
Activité la plus lente — Identifiez le goulot d’étranglement

Streamer vers un backend externe

Les traces OTel standard peuvent être exportées vers des systèmes de monitoring classiques :

Datadog — Via le collecteur OTel Datadog
Grafana / Tempo — Via le collecteur OTel
Jaeger — Pour la visualisation de traces distribuées
New Relic — Via l’exporteur OTel

Débogage d’un workflow échoué

Quand un workflow échoue, suivez cette procédure :

Récupérez l’ID d’exécution depuis les logs ou la réponse API
Consultez le résumé de trace pour identifier l’activité en erreur
Examinez les événements détaillés pour comprendre la séquence qui a mené à l’erreur
Vérifiez les retries : l’activité a-t-elle été retried ? Combien de fois ?
Analysez le message d’erreur : timeout, erreur réseau, erreur métier ?

async def diagnostiquer_echec(execution_id: str):
    client = Mistral()

    # Étape 1 : Résumé
    summary = await client.get_workflow_execution_trace_summary(execution_id)
    print(f"Statut : {summary.status}")
    print(f"Erreurs : {summary.error_count}")

    # Étape 2 : Événements détaillés
    events = await client.get_workflow_execution_trace_events(
        execution_id,
        include_internal_events=False
    )

    for event in events:
        if event.type == "activity_failed":
            print(f"Activité échouée : {event.data['activity_name']}")
            print(f"Erreur : {event.data['error_message']}")
            print(f"Tentative : {event.data['attempt']}")

Récapitulatif du cours

Félicitations, vous avez terminé ce cours sur les Workflows Mistral ! Voici un résumé des concepts couverts :

Leçons 1-3 — Introduction à l’orchestration durable, le mécanisme d’event sourcing et l’installation du SDK
Leçons 4-7 — Création d’activités, de workflows et de workers, déclenchement et test
Leçons 8-11 — Déterminisme, inputs Pydantic, timeouts, limitations et patterns avancés (parallélisme, continuation)
Leçons 12-14 — Politiques de retry, async I/O obligatoire, et observabilité avec OpenTelemetry

La plateforme est en Public Preview — les APIs et fonctionnalités peuvent évoluer. Consultez régulièrement la documentation officielle sur docs.mistral.ai/workflows pour suivre les mises à jour.

Points clés à retenir

Chaque activité génère automatiquement des spans OpenTelemetry avec nom, durée et statut
Nommez vos activités avec le paramètre name pour faciliter le débogage
Trois méthodes d’accès aux traces : OTel brut, résumé et événements détaillés
L’échantillonnage est contrôlé par le header traceparent lors du déclenchement
En production, surveillez le taux de succès, la durée et le taux de retry
Les traces OTel standard sont exportables vers Datadog, Grafana, Jaeger ou New Relic