Notes pratiques : Orchestration multi-agents : Création et observation de systèmes multi-agents

Ce guide reconstitue le parcours allant des matières premières à un système fonctionnel pour : l’orchestration multi-agents : création et observation de systèmes multi-agents avec LangGraph et LangSmith. L’accent est mis sur des étapes opérationnelles, des vérifications explicites, ainsi que du code que vous pouvez intégrer directement dans un dépôt sans devoir deviner son intention. Pour l’étape d’aperçu, définissez les entrées, le responsable de l’étape et 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é. 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.

Introduction : Qu’est-ce que l’orchestration d’agents LangSmith ?

Lorsque vous travaillez sur l’étape « Qu’est-ce que LangSmith ? » de l’Introduction, 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. Documentez 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 non livrés font partie intégrante du produit, et non d’améliorations apportées ultérieurement. Créez un point 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 tente à nouveau un nœud ultérieur.

L’architecture : comment l’orchestration fonctionne dans LangGraph

Lorsque vous travaillez sur l’étape « L’architecture de l’orchestration », 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. Préférez des unités petites et testables plutôt que des scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité et non un processus embrouillé. Faites 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 d’LLM lorsque l’opérateur réessaie un nœud ultérieur.

Comment l’orchestration est configurée :

Lors de l’étape « Comment fonctionne l’orchestration ? », 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. Considérez cette étape 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. Faites 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 du LLM lorsque l’opérateur réessaie un nœud ultérieur. Lors de l’étape « Comment fonctionne l’orchestration ? », 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 flags fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à lire l’ensemble du graphique.

Exemple reproductible étape par étape

La phase « Exemple reproductible étape par étape » 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 rollback avant d’élargir le périmètre. Documentez ensemble le parcours réussi et le parcours de récupération. Les tentatives répétées, les contrôles humains et le traitement des messages non livrés font partie intégrante du produit, et non d’une mise en forme ultérieure. Maintenez l’état des graphes simple et typé. Les blocs imbriqués masquent le fait que tel nœud a écrit telle champ et perturbent la reprise après interruption.

Prérequis

La phase des prérequis 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. 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é. Maintenez l’état du graphe plat et typé. Les blocs imbriqués cachent le fait que tel nœud a écrit telle champ et perturbent la reprise après interruption.

Étape 1 : Mise en place de l’environnement

La phase de configuration de l’environnement, étape 1, fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemplaire idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Considérez cette phase comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définites 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. La phase de configuration de l’environnement, étape 1, fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemplaire idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Gardez la configuration en dehors du code de l’application. Les fichiers d’environnement, les stocks de secrets et les indicateurs fonctionnels doivent se trouver en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du graphe.

uv add langgraph langchain-anthropic langsmith python-dotenv

Étape 2 : Le code Python

Pour l’étape 2, la phase Python, définissez les entrées, le responsable de l’étape et 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é. Documentez 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 non traités font partie du produit, et non d’une amélioration ultérieure. Séparez la construction du client du cycle de messages afin que les fournisseurs puissent être remplacés sans avoir à réécrire la machine d’états de la conversation.

import os
from typing import TypedDict, Literal
from dotenv import load_dotenv
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, END

# ==========================================
# 0. Load Environment Variables
# ==========================================
# This loads the API keys and LangSmith configs from the .env file
load_dotenv()

# ==========================================
# 1. Define the Shared State
# ==========================================
class AgentState(TypedDict):
    messages: list
    next_agent: str

# ==========================================
# 2. Define the Nodes (The Agents)
# ==========================================
# Initialize Claude 3.5 Sonnet
llm = ChatAnthropic(model="claude-sonnet-4-5-20250929", temperature=0)

def router_node(state: AgentState):
    """Acts as the router. Classifies the user query and directs it to the correct department."""
    system_prompt = SystemMessage(content=(
        "You are a router agent. Look at the user's message and classify it as either "
        "'billing' or 'technical'. Reply with ONLY the word 'billing' or 'technical'."
    ))
    response = llm.invoke([system_prompt] + state["messages"])
    classification = response.content.strip().lower()

    # Update state with the routing decision
    return {"next_agent": classification, "messages": [response]}

def billing_node(state: AgentState):
    """Handles billing-related queries."""
    system_prompt = SystemMessage(content=(
        "You are a billing support agent. Help the user with invoices, refunds, and payments. "
        "Be polite and professional."
    ))
    response = llm.invoke([system_prompt] + state["messages"])
    return {"messages": [response]}

def tech_support_node(state: AgentState):
    """Handles technical issues."""
    system_prompt = SystemMessage(content=(
        "You are a technical support agent. Help the user troubleshoot bugs, login issues, "
        "and software errors. Be analytical and helpful."
    ))
    response = llm.invoke([system_prompt] + state["messages"])
    return {"messages": [response]}

# ==========================================
# 3. Define the Routing Logic
# ==========================================
def route_decision(state: AgentState) -> Literal["billing", "technical"]:
    """Reads the state to decide which node to visit next."""
    next_agent = state.get("next_agent", "technical")
    # Claude is highly instruction-following, but we use 'in' to safely handle
    # any edge cases where it might add conversational filler.
    if "billing" in next_agent:
        return "billing"
    return "technical"

# ==========================================
# 4. Build and Compile the Graph
# ==========================================
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("router", router_node)
workflow.add_node("billing", billing_node)
workflow.add_node("technical", tech_support_node)

# Define edges
workflow.set_entry_point("router")
# The magic of orchestration: Conditional routing based on state
workflow.add_conditional_edges(
    "router",
    route_decision,
    {
        "billing": "billing",
        "technical": "technical",
    }
)

# Both specialized agents end the workflow
workflow.add_edge("billing", END)
workflow.add_edge("technical", END)

# Compile the graph
app = workflow.compile()

# ==========================================
# 5. Run the Orchestration
# ==========================================
if __name__ == "__main__":
    # Test Case 1: Billing Query
    print("--- Running Billing Test ---")
    inputs = {"messages": [HumanMessage(content="I was charged twice for my subscription!")]}
    result = app.invoke(inputs)
    print(result["messages"][-1].content)
    print("\n")

    # Test Case 2: Tech Support Query
    print("--- Running Tech Support Test ---")
    inputs = {"messages": [HumanMessage(content="My app keeps crashing when I click the save button.")]}
    result = app.invoke(inputs)
    print(result["messages"][-1].content)

Étape 3 : Exécuter le script pour tester

Pour l’Étape 3 « Exécuter la phase », définissez les entrées, le responsable de l’étape ainsi que les critères d’arrêt 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é. 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 processus embrouillé. Faites approuver par un humain les actions qui entraînent des dépenses ou modifient des données de production. La connexion en temps de compilation ne garantit pas la complétude du processus métier.

uv run multiagent-orchestration.py

Étape 4 : Afficher l’orchestration dans LangSmith

Pour l’étape 4 « View the stage », définissez les entrées, le responsable de l’étape et les critères de sortie 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é. 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 les terminations partielles silencieuses. Faites approuver par un humain les actions qui entraînent des dépenses ou modifient des données de production. La connexion en temps de compilation ne correspond pas à une complétude opérationnelle. Pour l’étape 4 « View the stage », définissez les entrées, le responsable de l’étape et les critères de sortie 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é. Gardez 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.

Ce que vous verrez dans LangSmith :

Lors de l’étape « Ce que vous verrez », notez d’abord les éléments requis : les données à fournir, 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. Documentez à la fois le parcours normal et les scénarios de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie intégrante du produit, et non d’améliorations apportées ultérieurement. 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 d’LLM lorsque l’opérateur réessaie un nœud ultérieur.

Conclusion

Liste de contrôle opérationnelle