Suivre un pipeline LangGraph pour des blogs avec Langfuse : segments, scores, avertissements.
Configurer un pipeline LangGraph multi-agents avec Langfuse : une trace par article, des segments imbriqués, des avertissements, des scores et une routine de débogage pour les drafts de qualité faible.
Sources: Building Multi-Agent System From Scratch — Part 6: Observability and Debugging
Les mécanismes de protection (vérifications des sources, vérifications des citations, limite de tentatives, escalade vers un humain) ne sont utiles que si une exécution échouée peut être reconstituée par la suite. Un pipeline de blog à plusieurs agents soulève alors de nombreuses questions : les sources de recherche étaient-elles de mauvaise qualité ? L’auteur a-t-il ignoré les avertissements de citation ? Le réviseur avait-il raison de rejeter le travail ? La recherche a-t-elle expiré ? Quel a été le coût en appels de modèles et en tokens pour cette exécution ?
En l’absence d’observabilité, le système n’est qu’une boîte noire entourée d’autres boîtes noires. Le suivi des actions menées par plusieurs agents constitue une tâche particulièrement difficile. Cette configuration ajoute Langfuse à un pipeline de blog LangGraph, générant une trace par article traité qui contient les transferts de responsabilités, les appels aux modèles et aux outils, les temps d’exécution, les coûts, les erreurs et les résultats d’évaluation.
Déterminer ce qu’est une trace
Au préalable, choisissez l’unité à examiner. Dans ce pipeline, une trace correspond à une tentative de créer un article. Elle commence lorsque un sujet est soumis et se termine lorsque le pipeline s’achève, fait une pause en attendant une intervention humaine ou échoue. Elle contient des observations pour chaque tâche effectuée :
blog-pipeline
├── research-agent
│ ├── search-web
│ └── research-model
├── writer-agent
│ └── writer-model
├── citation-check
│ └── citation-review-model
└── reviewer-agent
└── reviewer-model
Langfuse fonctionne avec trois types de noms. Une trace correspond à une exécution complète du début à la fin. Une observation est une étape individuelle au sein de cette trace, qui peut être un segment (logique d’application), une génération (appel à un LLM) ou un événement (un marqueur minuscule). Une session relie plusieurs traces entre elles, comme toutes les tentatives concernant le même sujet de blog.
La hiérarchie reflète la méthode de débogage : on part de l’exécution complète pour ensuite examiner en détail l’agent, l’outil ou l’appel de modèle qui a causé le problème.
Mise en place de Langfuse
Créez un projet Langfuse et fournissez ses identifiants via des variables d’environnement. La clé secrète ne doit jamais être inscrite dans les fichiers sources.
pip install -U langfuse
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_BASE_URL="https://cloud.langfuse.com"
export LANGFUSE_TRACING_ENVIRONMENT="development"
Le paquet langfuse est le SDK Python. Les deux clés authentifient l’application par rapport au projet, et LANGFUSE_BASE_URL sélectionne le site de déploiement. L’exemple vise la région cloud de l’UE, il convient donc de le remplacer par une autre région ou une instance auto-hébergée selon les besoins.
Une étiquette d’environnement permet de séparer les traces de développement de celles de production. Des noms courts et stables tels que development, staging et production conviennent bien. Sans elles, les expérimentations et les exécutions réelles des articles se mélangent dans une même vue, ce qui rend le débogage confus.
Intégration du gestionnaire d’appels de retour
Langfuse fournit un gestionnaire d’appels de retour LangChain. LangGraph s’appuie sur le mécanisme d’exécution et de gestion des appels de retour de LangChain, de sorte que le même gestionnaire capture les événements liés aux graphes, aux chaînes, aux outils et aux modèles :
from langfuse import get_client, propagate_attributes
from langfuse.langchain import CallbackHandler
langfuse = get_client()
langfuse_handler = CallbackHandler()
get_client() crée un client à partir des variables d’environnement. Le gestionnaire doit être créé une seule fois au démarrage et réutilisé à chaque exécution. Il écoute les événements de LangChain et les transmet sous forme d’observations, ce qui élimine toute nécessité de tracer manuellement chaque appel au modèle.
propagate_attributes intervient ensuite. Il applique les mêmes étiquettes à chaque observation au cours d’une même exécution, ce qui permet de filtrer une requête de recherche, une génération par l’écriteur et une génération par le réviseur en tant que parties d’un même pipeline d’article.
L’appel au graphe est enveloppé dans un span racine, le gestionnaire étant transmis via la configuration de LangGraph :
def run_blog_pipeline(topic: str, blog_id: str, user_id: str):
initial_state = {
"topic": topic,
"audience": "developers new to agent systems",
"research_brief": "",
"sources": [],
"open_questions": [],
"article_draft": "",
"review_feedback": "",
"citation_issues": [],
"approved": False,
"revision_count": 0,
"status": "researching",
}
with langfuse.start_as_current_observation(
as_type="span",
name="blog-pipeline",
input={"topic": topic, "audience": initial_state["audience"]},
) as pipeline_span:
with propagate_attributes(
trace_name="blog-pipeline",
session_id=blog_id,
user_id=user_id,
tags=["blog-pipeline", "langgraph"],
version="1.0.0",
metadata={"workflow": "research-write-review"},
):
trace_id = langfuse.get_current_trace_id()
result = graph.invoke(
initial_state,
config={"callbacks": [langfuse_handler]},
)
pipeline_span.update(
output={
"status": result["status"],
"approved": result["approved"],
"revision_count": result["revision_count"],
}
)
return result, trace_id
Divers détails de cet enveloppe sont importants. L’appel externe start_as_current_observation() crée l’observation racine, et ses paramètres d’entrée restent délibérément succincts, ne comprenant que le sujet et le public cible. C’est ce qui apparaît immédiatement dans la liste des traces, de sorte qu’un affichage de l’état interne n’a pas sa place là.
propagate_attributes() affecte chaque observation imbriquée. La valeur blog_id sert d’ID de session, regroupant les tentatives répétées sur un même article. Une valeur version identifie le flux de travail et la version du modèle derrière un résultat, permettant ainsi de filtrer les traces en fonction de la version plutôt que par supposition lorsque le comportement change.
Le gestionnaire parcourt le config standard de LangGraph, et à partir de là il collecte automatiquement les détails au niveau du framework : chaque appel de nœud, la génération par LLM, les appels d’outils, les entrées, les sorties, la latence, le nombre de tokens et le coût associé.
À la fin, l’élément racine est mis à jour avec trois résultats au niveau du pipeline : si l’exécution s’est terminée, si le brouillon a été approuvé et combien de révisions ont été effectuées. Le trace_id retourné doit être stocké dans les journaux de l’application ou dans l’interface utilisateur, permettant un accès direct depuis l’exécution d’un article jusqu’à son trace Langfuse.
Éléments de suivi au niveau métier à l’intérieur des agents
Le suivi automatique prouve qu’un modèle a été appelé. Il ne montre pas ce que chaque spécialiste a transmis au suivant. Pour cela, ajoutez un élément de suivi léger à l’intérieur d’un nœud d’agent, et transmettez l’argument config dans des appels imbriqués de LangChain ou LangGraph afin que le gestionnaire conserve le même contexte de suivi :
from langchain_core.runnables import RunnableConfig
def research_node(state: BlogState, config: RunnableConfig) -> dict:
with langfuse.start_as_current_observation(
as_type="span",
name="research-agent",
input={"topic": state["topic"], "audience": state["audience"]},
) as span:
result = research_agent.invoke(
{"topic": state["topic"], "audience": state["audience"]},
config=config,
)
span.update(
output={
"source_count": len(result["sources"]),
"open_question_count": len(result["open_questions"]),
"brief": result["research_brief"],
}
)
return {
"research_brief": result["research_brief"],
"sources": result["sources"],
"open_questions": result["open_questions"],
"status": "writing",
}
Ce segment complète le gestionnaire d’appel de retour sans le remplacer. Il ajoute une signification que le framework ne peut pas deviner, à savoir ce que "la recherche a réussi" signifie pour cette chaîne de traitement. Ici, il enregistre le nombre de sources, le nombre de questions non résolues ainsi que les instructions transmises à l’auteur.
En passant config: RunnableConfig à research_agent.invoke(), on transmet également les callbacks et le contexte de suivi. Oublier cela entraîne une erreur classique liée à l’observabilité : le graphe de niveau supérieur est suivi, mais l’appel imbriqué apparaît comme un suivi non pertinent ou disparaît.
Conservez les noms d’observation constants : research-agent, writer-agent, citation-check. Un sujet, un nom de modèle ou un ID d’utilisateur ne doit pas être intégré dans un nom, car ces données relèvent des entrées, des attributs du modèle ou des métadonnées. Des noms constants permettent aux tableaux de bord et aux filtres de rester fonctionnels malgré les changements d’implémentation.
Avertissements et erreurs
Tout résultat anormal n’est pas nécessairement une panne. Une source douteuse, un budget de révision épuisé ou une citation non prise en charge comptent également et doivent apparaître sous forme d’avertissement.
def record_source_assessment(assessment: SourceAssessment) -> None:
if assessment.suspicious_content:
langfuse.update_current_span(
level="WARNING",
status_message="Untrusted source contained agent-directed instructions.",
)
def record_pipeline_failure(error: Exception) -> None:
langfuse.update_current_span(
level="ERROR",
status_message=f"Pipeline failed: {type(error).__name__}",
)
Langfuse prend en charge des niveaux de journalisation pour les observations. WARNING s’applique aux exécutions terminées mais nécessitant une vérification, tandis que ERROR est réservé aux véritables pannes. Les messages d’état doivent décrire le problème opérationnel sans inclure de texte source sensible ou de secrets dans les données de télémétrie.
Le enregistreur d’évaluation de la source doit faire partie du processus de vérification de la source, et l’enregistreur de pannes doit être intégré à un gestionnaire d’erreurs capable de récupérer la situation ou de générer un rapport. Il est inapproprié d’intercepter chaque exception uniquement pour la journaliser, car les erreurs inattendues doivent toujours être signalées au système de surveillance des erreurs afin d’être détectées clairement.
Évaluation du résultat
Une trace indique ce qui s’est passé, tandis qu’une note indique si le résultat est satisfaisant. Une fois une exécution terminée, il convient d’attribuer une première note déterministe basée sur l’état du pipeline :
result, trace_id = run_blog_pipeline(
topic="How AI agents use tools",
blog_id="blog-ai-tools-001",
user_id="philip",
)
if trace_id:
langfuse.create_score(
trace_id=trace_id,
name="review_approved",
value=1 if result["approved"] else 0,
data_type="BOOLEAN",
comment=result["status"],
)
langfuse.create_score(
trace_id=trace_id,
name="revision_count",
value=float(result["revision_count"]),
data_type="NUMERIC",
)
Le premier score reflète le verdict structuré du réviseur, soit approbation soit rejet, tandis que le second indique le volume de travail supplémentaire nécessaire pour améliorer un brouillon. Accumulés dans le temps, ils révèlent bien plus que de simples anecdotes. Quel modèle, quelle version de prompt, quelle catégorie de sujet ou quelle méthode de sourcing permet le moins de révisions et le plus d’approbations ?
Aucun de ces scores n’est une vérité absolue, car un réviseur peut se tromper. Un score humain ajouté après lecture du brouillon final peut également être postérieur. L’habitude importante est de rendre la qualité mesurable et de comparer les résultats entre différentes traces, plutôt que de se fier à quelques exemples mémorables.
Une routine de débogage
Lorsqu’un article s’avère faible, ouvrez sa trace et posez ces questions dans l’ordre suivant :
- Le graphe a-t-il suivi le parcours attendu ? Cherchez les étapes Recherche, Rédaction, Vérification des citations et Réviseur, ainsi que tout cycle de révision.
C’est pourquoi la forme du suivi est importante. Un arbre bien nommé transforme « le blog était faux » en une constatation précise, telle que « l’auteur a ajouté une affirmation absente du bref après la deuxième révision. »
Préserver les données sensibles
Le suivi peut enregistrer les instructions, les résultats, les sorties des outils ainsi que les métadonnées, ce qui signifie également qu’il peut capturer des informations qui ne devraient pas être stockées. Les clés API, les en-têtes d’authentification et toute configuration secrète ne doivent jamais être envoyées à Langfuse. Si des données d’utilisateur entrent plus tard dans le pipeline, examinez ce qui est enregistré et mettez en place un masquage avant d’exporter les traces. Langfuse propose des fonctions de masquage qui suppriment les valeurs sensibles avant qu’elles ne quittent l’application.
Règle générale : enregistrez suffisamment d’informations pour expliquer une décision, mais pas plus que ce qui peut être conservé en toute sécurité.
Avantages du pipeline
Chaque exécution laisse désormais un enregistrement opérationnel :
- Chaque tentative de traitement d’un article correspond à une trace unique.
- Les agents, les modèles et les outils apparaissent sous forme d’observations imbriquées.
- Les noms constants, ainsi que les métadonnées, la version, la session et l’environnement, facilitent le filtrage.
Le débogage cesse d’être une question de navigation à travers les journaux de terminal et de devinage pour devenir un examen précis du moment où les choses ont déraillé.