Notes pratiques : Construire un système multi-agents de zéro — Partie 6

Les notes suivantes reconstituent une approche pratique pour aborder « Construction de systèmes multi-agents à partir de zéro — Partie 6 : Observabilité et débogage ». L’accent est mis sur les contrats, les vérifications et les placeholders de code interchangeables, plutôt que sur une présentation motivante. Lors de la phase d’aperçu, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Préférez des unités petites et testables plutôt que des scripts volumineux. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité et non un processus embrouillé.

Que devrait représenter un suivi ?

La phase « Que faut-il tracer ? » fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript exemplaire, un cas d’échec et la note de réversion avant d’élargir le périmètre. Traitez cette phase comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définez des vérifications de succès et refusez toute mise en œuvre partielle silencieuse. Maintenez l’état du graphe plat et typé. Les blocs imbriqués masquent le fait que tel nœud a écrit telle champ et perturbent la reprise après interruption.

blog-pipeline
├── research-agent
│   ├── search-web
│   └── research-model
├── writer-agent
│   └── writer-model
├── citation-check
│   └── citation-review-model
└── reviewer-agent
    └── reviewer-model

Connecter Langfuse à LangGraph

La phase de connexion de Langfuse à LangGraph fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript parfait, un cas d’échec et la note de réversion avant d’élargir le périmètre. Enregistrez les temps d’exécution ainsi que le coût en tokens ou en requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe de l’environnement de démonstration à des environnements partagés. Gardez l’état du graphe plat et typé : les blocs imbriqués masquent l’identité du nœud qui a écrit tel champ et perturbent la reprise après interruption.

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"

from langfuse import get_client, propagate_attributes
from langfuse.langchain import CallbackHandler


langfuse = get_client()
langfuse_handler = CallbackHandler()

Suivez une exécution complète d’un article

La phase « Trace one complete article » fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un enregistrement idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du graphe. Gardez l’état du graphe plat et typé. Les blocs imbriqués masquent le fait que tel nœud a écrit telle champ et perturbent la reprise après interruption. La phase « Trace one complete article » fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un enregistrement idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit pointer vers une seule responsabilité plutôt que vers un processus embrouillé.

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

Ajoutez des observations expliquant le transfert

Pour l’étape « Ajouter des observations expliquant l’état », il convient de définir les entrées, le responsable de l’étape ainsi que les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu, sans avoir à deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définites des vérifications de succès et refusez toute exécution partielle silencieuse. Faites approuver par un humain les cas où de l’argent est dépensé ou où des données de production sont modifiées. Une connexion en temps de compilation ne garantit pas la complétude du processus métier.

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",
    }

Marquez les signaux nécessitant une attention particulière

Pour marquer les étapes du processus, il faut définir les entrées, le responsable de chaque étape ainsi que les critères d’achèvement avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu, sans avoir à deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût des jetons ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts permet d’éviter des factures inattendues lorsque le processus passe de l’environnement de démonstration à des environnements partagés. Imposez une approbation humaine pour les actions qui entraînent des dépenses ou modifient des données de production. La connexion en temps de compilation ne garantit pas pour autant la complétude du processus métier.

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__}",
    )

Transformer les décisions des examinateurs en scores

Afin de transformer les décisions du Turn Reviewer en étapes concrètes, définissez les entrées, le responsable de chaque étape ainsi que les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à lire l’ensemble du système. Imposez une approbation humaine pour les actions qui entraînent des dépenses ou modifient des données de production. Une connexion effectuée en temps de compilation ne garantit pas la complétude du processus métier. Afin de transformer les décisions du Turn Reviewer en étapes concrètes, définissez les entrées, le responsable de chaque étape ainsi que les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt que plusieurs.

tuyau angulé.

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",
    )

Déboguer une exécution échouée en cinq questions

Lors de la phase de débogage d’une exécution échouée, notez d’abord les conditions requises : entrées nécessaires, signal de succès, et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Considérez cette phase comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définez des vérifications de succès, et refusez les terminations partielles silencieuses. Créez des points de contrôle après les étapes coûteuses. Le système de reprise ne doit pas facturer à nouveau la même appel du LLM lorsque l’opérateur réessaie un nœud ultérieur.

L’observabilité a également ses limites en matière de confidentialité

Lors du travail sur l’observabilité, qui comporte également une étape dédiée, notez d’abord le contrat : les entrées requises, le signal de succès, ainsi que ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe d’un environnement de démonstration à des environnements partagés. Créez un point de contrôle après les étapes coûteuses. La reprise du processus ne doit pas facturer à nouveau la même appel de LLM lorsque l’opérateur réessaie un nœud ultérieur.

Ce que nous avons construit

Lors de la phase « Ce que nous avons construit », notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de garantir l’intégrité des modifications ultérieures du code. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du système. Créez des points de contrôle après les étapes coûteuses. Le mécanisme de reprise ne doit pas facturer à nouveau la même appel d’LLM lorsque l’opérateur réessaie un nœud ultérieur. Lors de la phase « Ce que nous avons construit », notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de garantir l’intégrité des modifications ultérieures du code. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un pipeline embrouillé.

Liste de contrôle opérationnelle

Pour l’étape de la liste de contrôle opérationnelle, définissez les entrées, le responsable de l’étape et les critères d’achèvement avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché.

Dokumentez ensemble le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages échoués font partie intégrante du produit, et non d’améliorations ultérieures.

Imposez une approbation humaine pour les actions qui entraînent des dépenses ou modifient des données de production. Une connexion en temps de compilation ne garantit pas la complétude du processus métier.

Rédigez un petit manuel d’utilisation : comment rotationner les clés, comment vider la file d’attente, comment annuler la dernière ingestion.

Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’erreur doit indiquer une seule responsabilité plutôt qu’un processus embrouillé.

Apportez une validation humaine pour les étapes qui engagent des dépenses ou modifient les données de production. La connexion en temps de compilation ne garantit pas une couverture complète des besoins métier.

Au préalable de promouvoir l’ensemble technique, figez les versions, conservez une transcription exemplaire pour le parcours critique, et vérifiez les étapes de réversion. Les environnements partagés nécessitent des limites de fréquence, des contrôles d’attribution et un responsable clair pour la rotation des secrets. Préférez une fiabilité sans faille à de brillantes démonstrations ponctuelles.

Note pour le lot cf19385cb4a9 : gardez les clés du fournisseur hors du répertoire, fixez un plafond pour les tokens par session, et stockez les transcriptions à côté des fichiers d’évaluation afin que les remplacements ultérieurs de modèles restent comparables.