LangGraph : construire des agents à états et à cycles

Jusqu’ici, create_agent nous rendait un grand service : il gérait tout seul la boucle « réfléchir, appeler un outil, recommencer ». Mais le jour où la coopérative vous demande « je veux que les réclamations passent par un humain, pas par l’IA », vous butez sur une limite. Cette logique-là — classer un message, puis aiguiller vers une branche ou une autre — n’est pas dans la boucle standard. Il vous faut reprendre la main sur le parcours.

C’est exactement le rôle de LangGraph : décrire votre agent comme un graphe de nœuds (des fonctions qui font le travail) reliés par des arêtes (qui décident de la suite). Vous dessinez le parcours, vous gardez un état qui circule de nœud en nœud, et vous pouvez créer des branches et des boucles que la boucle automatique ne permet pas. Dans ce tutoriel, on construit un aiguillage de support : les questions simples reçoivent une réponse automatique, les litiges sont escaladés vers un humain.

🎯 Ce que vous allez apprendre

• Modéliser un agent comme un graphe d’états avec StateGraph.
• Définir un état partagé et comprendre le rôle d’un réducteur comme add_messages.
• Créer des arêtes conditionnelles pour aiguiller selon le contexte.
• Donner une mémoire de conversation à votre graphe avec un checkpointer.

🛠️ Ce que vous allez construire

Un aiguillage de support pour l’Assistant Teranga : chaque message client est d’abord classé (« simple » ou « litige »), puis routé. Une question d’information reçoit une réponse immédiate ; une réclamation déclenche une escalade vers un conseiller humain. Et grâce à un checkpointer, l’assistant se souvient des échanges précédents avec le même client.

Prérequis

• Avoir suivi le premier tutoriel et compris create_agent.
• Python 3.10+. On installe LangGraph : pip install -U langgraph (au moment d’écrire, la 1.x).
• Test express : si vous savez ce qu’est une fonction qui prend un dictionnaire et en renvoie un autre, vous êtes prêt.
• ⏱️ Temps estimé : ~50 minutes.

Étape 1 — L’état, colonne vertébrale du graphe

Dans LangGraph, tout tourne autour d’un état : un dictionnaire typé qui circule entre les nœuds. Chaque nœud lit l’état, fait son travail, et renvoie les champs qu’il veut mettre à jour. La question délicate, c’est : quand deux nœuds touchent au même champ, comment les fusionner ? C’est le rôle d’un réducteur. Pour la liste des messages, le réducteur add_messages sait ajouter les nouveaux messages plutôt que d’écraser les anciens — sans lui, chaque nœud effacerait l’historique.

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class Etat(TypedDict):
    messages: Annotated[list, add_messages]   # add_messages = on ajoute, on n'écrase pas
    categorie: str                            # rempli par le nœud de classement

L’Annotated[list, add_messages] se lit ainsi : « ce champ est une liste, et pour la mettre à jour, utilise add_messages ». Le champ categorie, lui, n’a pas de réducteur : il sera simplement remplacé. Cette distinction est le cœur de LangGraph — maîtrisez-la et le reste coule de source.

✅ Point d’étape — vous savez expliquer pourquoi messages a besoin de add_messages mais pas categorie. Si ce n’est pas clair, imaginez deux nœuds qui répondent : sans réducteur, le second efface le premier.

Étape 2 — Écrire les nœuds

Un nœud est une simple fonction etat -> dict. On en écrit trois : un qui classe le message, un qui répond aux questions simples, un qui escalade les litiges. Chacun reste petit et testable isolément — c’est l’un des grands avantages de cette approche par rapport à une boucle opaque.

from langchain.chat_models import init_chat_model
modele = init_chat_model("openai:gpt-4o-mini")

def classer(etat: Etat) -> dict:
    dernier = etat["messages"][-1].content
    verdict = modele.invoke(
        "Réponds par un seul mot : 'litige' si ce message est une réclamation ou un "
        f"problème, sinon 'simple'. Message : {dernier}").content.lower()
    return {"categorie": "litige" if "litige" in verdict else "simple"}

def repondre(etat: Etat) -> dict:
    reponse = modele.invoke([
        {"role": "system", "content": "Tu es l'assistant Teranga. Réponds en français, brièvement."},
        *etat["messages"],
    ])
    return {"messages": [reponse]}

def escalader(etat: Etat) -> dict:
    return {"messages": [{"role": "assistant", "content":
        "Votre dossier mérite l'attention d'un conseiller. Je le transmets : "
        "vous serez recontacté sous 24 h. (Un ticket vient d'être ouvert.)"}]}

Notez que classer ne touche pas aux messages : il renvoie seulement categorie. repondre et escalader, eux, ajoutent un message — et grâce à add_messages, ils complètent l’historique sans l’effacer. Chaque nœud a une responsabilité unique, ce qui les rend faciles à tester et à faire évoluer.

✅ Point d’étape — appelez classer à la main avec un faux état pour vérifier qu’il renvoie bien « simple » ou « litige » avant de l’intégrer au graphe.

Étape 3 — Câbler le graphe avec une arête conditionnelle

Reste à relier les nœuds. On part de START vers classer. Puis vient le moment clé : une arête conditionnelle qui regarde la catégorie et choisit la suite. C’est ce branchement, impossible avec la boucle automatique, qui justifie LangGraph.

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

def router(etat: Etat) -> str:
    return "escalader" if etat["categorie"] == "litige" else "repondre"

graphe = StateGraph(Etat)
graphe.add_node("classer", classer)
graphe.add_node("repondre", repondre)
graphe.add_node("escalader", escalader)

graphe.add_edge(START, "classer")
graphe.add_conditional_edges("classer", router, {"repondre": "repondre", "escalader": "escalader"})
graphe.add_edge("repondre", END)
graphe.add_edge("escalader", END)

app = graphe.compile(checkpointer=InMemorySaver())

La fonction router ne fait pas le travail : elle renvoie le nom du prochain nœud, et le dictionnaire passé à add_conditional_edges fait la correspondance. La compilation est obligatoire : un graphe non compilé n’est pas exécutable. On lui passe au passage un InMemorySaver, dont on verra l’effet à l’étape suivante.

✅ Point d’étape — graphe.compile() ne lève pas d’erreur. Une erreur ici signale presque toujours un nom de nœud mal orthographié dans une arête.

Étape 4 — Exécuter et donner une mémoire

Le checkpointer fait plus que de la décoration : il sauvegarde l’état après chaque étape, identifié par un thread_id. Résultat, deux appels avec le même thread_id partagent l’historique — l’assistant se souvient. C’est la base d’une vraie conversation multi-tours.

config = {"configurable": {"thread_id": "client-kaolack-42"}}

r1 = app.invoke({"messages": [{"role": "user", "content": "Bonjour, vous livrez à Kaolack ?"}]}, config)
print(r1["messages"][-1].content)

# Deuxième message, MÊME thread : l'assistant garde le contexte
r2 = app.invoke({"messages": [{"role": "user", "content": "Et c'est à quel prix ?"}]}, config)
print(r2["messages"][-1].content)

Au premier message, « simple » → branche repondre. Au second, « Et c’est à quel prix ? » n’a de sens que parce que le checkpointer a conservé le fil : l’assistant sait qu’on parle de livraison à Kaolack. Changez le thread_id et le contexte disparaît : chaque client a son fil. Essayez maintenant un message de réclamation (« j’ai été livré un produit cassé ! ») : la catégorie bascule en « litige » et le graphe part vers escalader.

LangGraph ou create_agent : que choisir ?

Les deux coexistent, et le bon réflexe est de commencer simple. create_agent suffit tant que votre agent se résume à « réfléchir et appeler des outils en boucle » — c’est le cas de la majorité des assistants. On passe à LangGraph dès qu’on a besoin de contrôler le parcours : aiguiller selon une catégorie, imposer une étape de validation humaine, faire repasser par un nœud de vérification, ou orchestrer plusieurs agents. En pratique, beaucoup de projets démarrent avec create_agent et migrent un point précis vers LangGraph quand une exigence métier l’impose. Sachez aussi que create_agent est lui-même bâti sur LangGraph : apprendre LangGraph, c’est comprendre ce qui tournait sous le capot depuis le début.

Visualiser le graphe

Un graphe se débogue mieux quand on le voit. LangGraph sait exporter son schéma, par exemple en Mermaid, ce qui donne un diagramme des nœuds et des branches — précieux pour vérifier que l’aiguillage est correct avant de chercher un bug dans le code.

print(app.get_graph().draw_mermaid())   # colle le résultat dans un visualiseur Mermaid

Vous obtenez un texte décrivant START → classer → (repondre | escalader) → END. Sur un graphe de trois nœuds c’est anecdotique ; sur un agent réel de quinze nœuds, ce diagramme devient votre meilleur allié pour repérer une branche oubliée ou un nœud jamais atteint.

Suivre l’agent en direct avec le streaming

Quand un graphe enchaîne plusieurs nœuds, attendre la réponse finale sans rien voir est frustrant — et peu pratique pour déboguer. LangGraph sait diffuser chaque étape au fur et à mesure : au lieu d’invoke, on utilise stream, qui rend la main après chaque nœud.

for etape in app.stream({"messages": [{"role": "user", "content": "Vous livrez à Kaolack ?"}]}, config):
    print(etape)   # un dictionnaire {nom_du_noeud: mise_a_jour} à chaque étape

Vous voyez d’abord la sortie de classer, puis celle de repondre. C’est exactement ce mécanisme qui permet, dans une interface web, d’afficher « l’assistant réfléchit… » puis la réponse mot à mot, plutôt qu’un long silence suivi d’un bloc de texte. Pour un assistant destiné à des clients sur une connexion lente, ce retour visuel immédiat change tout dans la perception de rapidité.

Mettre un humain dans la boucle

Pour une action sensible — escalader, rembourser, modifier une commande — on veut parfois qu’un humain valide avant que ça parte. LangGraph permet d’interrompre le graphe juste avant un nœud, le temps d’une approbation, puis de reprendre exactement où on en était. On l’active à la compilation.

app = graphe.compile(checkpointer=InMemorySaver(), interrupt_before=["escalader"])

# Le graphe s'arrête avant 'escalader' et rend la main :
app.invoke({"messages": [{"role": "user", "content": "On m'a livré un panier cassé !"}]}, config)
# ... un humain relit, puis on reprend là où on s'était arrêté :
app.invoke(None, config)   # None = « continue », l'état est restauré par le checkpointer

Le checkpointer, encore lui, rend cette pause possible : il a sauvegardé l’état, donc reprendre revient à passer None avec le même thread_id. C’est la brique des parcours « humain dans la boucle » : l’IA prépare, l’humain valide les décisions à conséquence. Indispensable dès qu’une erreur de l’agent coûterait cher à la coopérative.

🐞 Pièges fréquents

🌍 Adaptation au contexte ouest-africain

L’aiguillage que vous venez de construire est précieux dans un contexte où le support humain est rare et précieux : l’IA absorbe les questions répétitives (horaires, livraison, prix) et réserve le temps des conseillers aux vrais litiges. Pour la mémoire, InMemorySaver suffit en développement, mais elle disparaît au redémarrage ; en production, LangGraph propose des checkpointers qui écrivent dans une base — par exemple PostgreSQL, que vous hébergez déjà peut-être pour l’application. Un seul petit VPS suffit à faire tourner ce graphe et sa base. Et comme chaque nœud est une fonction Python ordinaire, vous pouvez y brancher vos outils mobile money ou votre base de commandes sans rien réapprendre.

✅ Récapitulatif

Vous savez désormais reprendre la main sur le parcours d’un agent : modéliser un état avec son réducteur, écrire des nœuds testables, les câbler avec des arêtes conditionnelles, et donner une mémoire de conversation grâce à un checkpointer. L’Assistant Teranga ne suit plus une boucle imposée : il suit votre logique métier, avec une branche humaine pour les réclamations. C’est le passage de l’agent « boîte noire » à l’agent que vous orchestrez. Et ce n’est qu’un début : le même modèle — état, nœuds, arêtes — décrit aussi bien un correcteur qui boucle jusqu’à obtenir un résultat valide qu’un orchestrateur qui répartit le travail entre plusieurs agents spécialisés. Une fois la mécanique du graphe en main, vous tenez l’outil qui structure n’importe quel parcours d’agent, du plus simple au plus ambitieux.

🧾 Aide-mémoire

💪 À vous de jouer

Ajoutez une troisième catégorie « hors-sujet » (message sans rapport avec la coopérative) qui répond poliment que l’assistant ne traite que les sujets de la boutique. Il vous faut adapter le nœud classer, le router et la carte des arêtes.

def classer(etat: Etat) -> dict:
    dernier = etat["messages"][-1].content
    v = modele.invoke(
        "Réponds par un mot : 'litige', 'horssujet' (sans rapport avec une boutique "
        f"d'artisanat) ou 'simple'. Message : {dernier}").content.lower()
    for c in ("litige", "horssujet"):
        if c in v:
            return {"categorie": c}
    return {"categorie": "simple"}

def hors_sujet(etat: Etat) -> dict:
    return {"messages": [{"role": "assistant", "content":
        "Je suis l'assistant de la coopérative Teranga et ne peux aider que sur nos produits "
        "et commandes. Pour le reste, je vous invite à contacter le service concerné."}]}

graphe.add_node("hors_sujet", hors_sujet)
graphe.add_conditional_edges("classer", router,
    {"repondre": "repondre", "escalader": "escalader", "hors_sujet": "hors_sujet"})
graphe.add_edge("hors_sujet", END)
# et dans router : if etat["categorie"] == "horssujet": return "hors_sujet"

Tutoriels frères

• Donner des outils à un agent (tool calling) — brancher de vraies actions.
• CrewAI : une équipe d’agents collaboratifs — répartir le travail entre plusieurs agents.

Pour aller plus loin

• 🔝 Retour au guide principal : le guide des frameworks d’agents IA
• Documentation officielle : l’API graphe de LangGraph
• À lire ensuite : passer à une équipe d’agents avec CrewAI.

FAQ

Q : LangGraph remplace-t-il LangChain ?
R : Non, il le complète. LangChain fournit les modèles, outils et l’agent prêt à l’emploi ; LangGraph orchestre des parcours sur mesure. Les deux s’utilisent ensemble — vos nœuds appellent les modèles et outils de LangChain.

Q : Qu’est-ce qu’un réducteur, simplement ?
R : Une règle de fusion. Quand un nœud renvoie une valeur pour un champ déjà rempli, le réducteur dit comment combiner l’ancienne et la nouvelle. add_messages ajoute ; sans réducteur, on remplace.

Q : Puis-je faire des boucles dans le graphe ?
R : Oui, c’est même une de ses forces : une arête peut renvoyer vers un nœud précédent (par exemple « vérifier puis recommencer si invalide »). Pensez seulement à prévoir une condition de sortie pour éviter une boucle infinie.

Q : Le checkpointer en mémoire convient-il en production ?
R : Pour des tests, oui. En production, on utilise un checkpointer persistant (base de données) pour que la mémoire survive aux redémarrages et soit partagée entre plusieurs instances de l’application.