Nous avons migré notre pipeline d’agents vers LangGraph (Supervisor + Human-in-the-loop)

Salut tout le monde.

Honnêtement, si vous avez déjà essayé de construire un système multi-agents avec des frameworks de chaînage classiques, vous connaissez la douleur. Dès que les agents doivent collaborer, on se retrouve vite avec des boucles infinies, des pertes de contexte ou des agents qui font un peu n’importe quoi dans leur coin.

Après pas mal de maux de tête, je me suis penché sur LangGraph (pas confondre avec longchain). Pour le coup, l’approche sous forme de graphe d’états (State Graph) change vraiment la donne pour gérer la mémoire et la coordination.

Je vous partage aujourd’hui mon implémentation d’un pattern Supervisor-Worker (un agent superviseur qui distribue le travail à des agents spécialisés) avec une validation humaine (Human-in-the-loop). C’est le setup idéal pour générer des rapports de recherche sans laisser l’IA publier n’importe quoi sans surveillance.

Le Scénario

On veut créer une équipe de recherche autonome :

1. Le Chercheur (Researcher) : cherche les infos sur le web.
2. Le Rédacteur (Writer) : synthétise les notes et rédige le rapport.
3. Le Superviseur (Supervisor) : orchestre le tout, décide qui travaille et quand c’est fini.
4. L’Humain (Vous et moi) : valide le brouillon final ou renvoie l’équipe au charbon avec des feedbacks précis.

1. Définir l’état global (State)

La force de LangGraph, c’est ce dictionnaire partagé (le State) qui circule entre les nœuds. On y stocke l’historique des messages, la décision de routage du superviseur, et un flag d’approbation humaine.

from typing import Literal
from typing_extensions import TypedDict
from langchain_core.messages import HumanMessage
from langgraph.graph import MessagesState

# Nos agents spécialisés
members = ["researcher", "writer"]

class RouterState(MessagesState):
    next: str       # Prochaine étape décidée par le superviseur
    approved: bool  # Statut de l'approbation humaine

2. Le Superviseur (avec Structured Output)

Pour éviter que le superviseur ne divague, on utilise le tool-calling pour forcer le LLM à renvoyer un format structuré précis (via Pydantic). Il doit obligatoirement choisir entre appeler le chercheur, le rédacteur, ou déclarer que c’est terminé (FINISH).

from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI

# Le schéma qui force la décision du superviseur
class RouteResponse(BaseModel):
    next: Literal["researcher", "writer", "FINISH"] = Field(
        description="Le prochain agent à exécuter, ou FINISH si le rapport est prêt."
    )

llm = ChatOpenAI(model="gpt-4o", temperature=0)
# On lie le schéma au modèle
supervisor_agent = llm.with_structured_output(RouteResponse)

def supervisor_node(state: RouterState):
    system_prompt = (
        f"Tu es le superviseur d'une équipe de rédacteurs composée de : {members}. "
        "Analyse l'historique de la discussion et sélectionne le prochain agent. "
        "Si le travail est suffisant pour donner la réponse finale, réponds par FINISH."
    )

    messages = [{"role": "system", "content": system_prompt}] + state["messages"]
    response = supervisor_agent.invoke(messages)

    return {"next": response.next}

3. Les Workers (Chercheur et Rédacteur)

Les workers exécutent leur tâche, puis écrivent leur résultat directement dans l’état sous forme de message. Une fois leur action terminée, le flux repasse automatiquement par le superviseur.

from langchain_core.messages import AIMessage

def researcher_node(state: RouterState):
    last_message = state["messages"][-1].content
    # (Ici, on simule une recherche web, par exemple via Tavily ou une API tierce)
    result = f"Notes de recherche sur '{last_message}' : LangGraph intègre nativement des outils de supervision en 2026."

    return {"messages": [AIMessage(content=result, name="researcher")]}

def writer_node(state: RouterState):
    history = "\n".join([m.content for m in state["messages"]])
    result = f"Rapport Final basé sur les recherches :\n{history}"

    return {"messages": [AIMessage(content=result, name="writer")]}

4. Le point de contrôle humain (Human Gate Node)

C’est là que la magie de LangGraph opère. On va créer un nœud “passerelle” qui ne fait rien d’autre que servir de point d’arrêt (breakpoint). Le graphe va s’y mettre en pause en attendant notre action.

def human_gate_node(state: RouterState):
    pass # Nœud vide servant de checkpoint

5. Construction du Graphe avec Gestion des Interruptions

On assemble le tout. Au lieu d’aller directement vers la fin (END) quand le superviseur dit FINISH, on redirige le flux vers notre human_gate_node. On configure ensuite un MemorySaver pour sauvegarder l’état à chaque étape.

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

workflow = StateGraph(RouterState)

# Ajout des nœuds
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)
workflow.add_node("human_gate_node", human_gate_node)

# Les workers renvoient toujours vers le superviseur
for member in members:
    workflow.add_edge(member, "supervisor")

# Routage dynamique depuis le superviseur
def route_dest(state: RouterState):
    if state["next"] == "FINISH":
        return "human_gate_node" # On intercepte avant la fin !
    return state["next"]

workflow.add_conditional_edges(
    "supervisor",
    route_dest,
    {"researcher": "researcher", "writer": "writer", "human_gate_node": "human_gate_node"}
)

# Décision suite à l'avis humain
def evaluate_human_choice(state: RouterState):
    if state.get("approved") is True:
        return END
    return "supervisor" # Si rejeté, on repart dans la boucle

workflow.add_conditional_edges(
    "human_gate_node",
    evaluate_human_choice,
    {END: END, "supervisor": "supervisor"}
)

workflow.set_entry_point("supervisor")

# Activation de la persistance de l'état ET du point d'arrêt
memory = MemorySaver()
app = workflow.compile(checkpointer=memory, interrupt_before=["human_gate_node"])

6. L’exécution (et l’interaction en direct)

C’est la partie la plus cool. On lance le run (Phase A), le graphe travaille en coulisse puis se fige au breakpoint. On récupère le draft, on donne notre avis (Phase B) et on relance la machine.

Phase A : Lancement et pause automatique

# Un thread_id unique permet de suivre cette session précise dans l'historique
config = {"configurable": {"thread_id": "projet_rapport_cyber"}}
inputs = {"messages": [HumanMessage(content="Cherche les nouveautés de LangGraph en 2026 et fais-en un résumé.")]}

# Le graphe s'exécute : superviseur -> chercheur -> superviseur -> rédacteur -> PAUSE
for event in app.stream(inputs, config, stream_mode="values"):
    if "messages" in event:
        print(f"Dernier message : {event['messages'][-1].content}\n")

# Vérification du statut
state = app.get_state(config)
print(f"Prochain nœud en attente d'exécution : {state.next}") 
# Devrait afficher : ('human_gate_node',)

Phase B : Approbation ou correction par l’utilisateur

# Lecture du brouillon rédigé par l'agent
current_state = app.get_state(config)
draft_propose = current_state.values["messages"][-1].content
print(f"--- BROUILLON REÇU ---\n{draft_propose}\n----------------------")

# Input utilisateur dans la console
decision = input("Tapez 'oui' pour valider, ou écrivez vos corrections pour rejeter : ")

if decision.lower() == 'oui':
    # On valide l'état et on relance le graphe qui va s'arrêter sur END
    app.update_state(config, {"approved": True}, as_node="human_gate_node")
    app.stream(None, config) 
    print("Rapport validé et finalisé avec succès !")
else:
    # On injecte le feedback humain dans la mémoire et on réactive le flux
    feedback = HumanMessage(content=f"Modifications demandées par l'humain : {decision}")
    app.update_state(
        config, 
        {"messages": [feedback], "approved": False}, 
        as_node="human_gate_node"
    )
    # Le graphe reprend et retourne directement voir le superviseur avec le feedback
    app.stream(None, config)
    print("Corrections envoyées à l'équipe d'agents...")

Pourquoi cette architecture change tout ?

Avant d’utiliser ça, mes agents tournaient souvent en boucle fermée et consommaient mes tokens d’API pour rien (ce qui fait toujours plaisir en fin de mois…).

Avec ce pattern :

• Le superviseur sait exactement où on en est grâce au format d’état partagé.
• Le checkpointer (MemorySaver) permet de figer l’application à un instant T sans perdre une seule variable.
• Le fait de pouvoir mettre à jour l’état à la volée (app.update_state) en tant que nœud virtuel permet d’injecter du feedback de manière super propre sans casser la logique de développement de nos agents.

Bref, si vous prévoyez de mettre des agents en production, ne faites pas l’impasse sur le Human-in-the-loop. C’est le seul moyen d’éviter les hallucinations sauvages sur vos rapports finaux.

Des questions sur l’implémentation ou sur la gestion du State ? N’hésitez pas !