LangGraph : des agents IA pilotés par un graphe

Les trois guides précédents traitaient un agent comme une boucle : on appelle le modèle, on exécute un outil, on recommence. Ce modèle atteint vite ses limites — il gère mal la persistance (reprendre une exécution interrompue) et le human-in-the-loop (mettre l'agent en pause pour qu'un humain tranche). LangGraph change de modèle mental : un agent y est un graphe d'états, des nœuds qui transforment un état partagé reliés par des arêtes qui décident du parcours. Vous construirez un agent de triage de tickets qui classe un ticket de support, puis suspend son exécution quand l'action recommandée est destructive — le temps qu'un opérateur valide. Au passage : l'état, le routage conditionnel, le checkpointer et la fonction interrupt(). Public visé : développeur ayant lu les guides sur les agents et le function calling.

Ce que vous allez apprendre

Pourquoi le modèle graphe d'états dépasse la boucle écrite à la main.
Construire un graphe : état partagé, nœuds, arêtes, compile().
Aiguiller le parcours avec un routage conditionnel.
Persister l'état avec un checkpointer — et pourquoi il est obligatoire.
Mettre l'agent en pause pour une validation humaine avec interrupt().

Prérequis

Ce guide prolonge Comprendre les agents IA et Agents typés avec PydanticAI. Il vous faut :

Une instance Ollama avec le modèle qwen2.5.
Les bases de Python et de Pydantic.
Python 3.10+.

De la boucle au graphe d'états

Une boucle d'agent est linéaire : un seul fil d'exécution qui tourne jusqu'à une réponse. Tant que le travail reste simple, c'est suffisant. Mais dès qu'on veut brancher le parcours selon le résultat d'une étape, interrompre puis reprendre, ou revenir en arrière, la boucle devient un enchevêtrement de if et de variables d'état éparpillées.

LangGraph propose un autre modèle : l'agent est un graphe. On y déclare trois choses. L'état — un dictionnaire typé qui circule de bout en bout, la mémoire de travail du graphe. Les nœuds — des fonctions qui reçoivent l'état et renvoient les clés qu'elles modifient. Les arêtes — les liens qui décident quel nœud s'exécute ensuite. Le parcours n'est plus codé en dur dans une boucle : il émerge de la structure du graphe.

Ce changement débloque deux capacités difficiles à obtenir avec une boucle, et c'est tout l'objet de ce guide : la persistance de l'état entre deux exécutions, et la pause du graphe pour laisser un humain décider.

L'état partagé

L'état se déclare comme un TypedDict. C'est le contrat du graphe : chaque nœud sait quelles clés il peut lire et écrire.

from typing_extensions import TypedDict

class EtatTriage(TypedDict):
    ticket: str       # le texte du ticket à trier
    categorie: str    # rempli par le nœud d'analyse
    priorite: str     # rempli par le nœud d'analyse
    action: str       # action recommandée par le modèle
    approuve: bool    # décision de l'opérateur
    resultat: str     # bilan final, rempli par le nœud d'exécution

Un nœud ne renvoie jamais l'état entier : il renvoie un dictionnaire des seules clés qu'il met à jour. LangGraph fusionne cette mise à jour dans l'état partagé. Le ticket entre vide de tout sauf son texte ; chaque nœud le complète au passage.

Les nœuds : des fonctions état vers mise à jour

Un nœud est une fonction ordinaire. Le premier nœud, analyser, appelle le modèle pour classer le ticket. On réutilise ici la sortie structurée du guide précédent — un modèle Pydantic — branchée sur Ollama via ChatOllama, l'intégration officielle de LangChain.

from langchain_ollama import ChatOllama
from pydantic import BaseModel, Field
from typing import Literal

LLM = ChatOllama(
    model="qwen2.5", base_url="http://localhost:11434", temperature=0
)

class Analyse(BaseModel):
    categorie: str = Field(description="Thème du ticket, ex : facturation")
    priorite: Literal["haute", "moyenne", "basse"]
    action: Literal["repondre_au_client", "escalader_n2", "fermer_le_ticket"]

ANALYSEUR = LLM.with_structured_output(Analyse)

def analyser(etat: EtatTriage) -> dict:
    """Classe le ticket : catégorie, priorité et action recommandée."""
    analyse = ANALYSEUR.invoke(
        [
            ("system", "Tu tries des tickets de support client."),
            ("human", etat["ticket"]),
        ]
    )
    return {
        "categorie": analyse.categorie,
        "priorite": analyse.priorite,
        "action": analyse.action,
    }

Le temperature=0 rend le triage reproductible : le même ticket donne le même classement. Les types Literal contraignent le modèle — il ne peut renvoyer qu'une des trois actions prévues, jamais une valeur inattendue.

Le dernier nœud, appliquer, exécute l'action. C'est lui qui porte la règle de sécurité : une action destructive non approuvée ne s'exécute pas.

ACTIONS_DESTRUCTIVES = {"fermer_le_ticket"}

def appliquer(etat: EtatTriage) -> dict:
    """Exécute l'action — sauf si une validation humaine l'a refusée."""
    action = etat["action"]
    if action in ACTIONS_DESTRUCTIVES and not etat.get("approuve", False):
        return {"resultat": f"Action « {action} » refusée — ticket conservé."}
    return {"resultat": f"Action « {action} » appliquée."}

Le routage conditionnel

Tous les tickets ne suivent pas le même chemin. Un ticket d'aide va directement à l'exécution ; un spam, dont l'action est destructive, doit d'abord passer par une validation. Cet aiguillage est une fonction de routage : elle lit l'état et renvoie le nom du prochain nœud.

def router(etat: EtatTriage) -> Literal["controle_humain", "appliquer"]:
    """Aiguille vers la validation humaine si l'action est destructive."""
    if etat["action"] in ACTIONS_DESTRUCTIVES:
        return "controle_humain"
    return "appliquer"

Cette fonction ne fait aucun travail : elle ne fait que décider. C'est la différence clé avec une boucle, où le branchement se mêle au traitement. Ici, la décision de parcours est isolée, lisible, testable seule.

Construire et compiler le graphe

Reste à assembler les pièces. On crée un StateGraph, on y ajoute les nœuds, puis les arêtes. Les constantes START et END sont les points d'entrée et de sortie du graphe.

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

constructeur = StateGraph(EtatTriage)
constructeur.add_node("analyser", analyser)
constructeur.add_node("controle_humain", controle_humain)
constructeur.add_node("appliquer", appliquer)

constructeur.add_edge(START, "analyser")
constructeur.add_conditional_edges(
    "analyser", router, ["controle_humain", "appliquer"]
)
constructeur.add_edge("controle_humain", "appliquer")
constructeur.add_edge("appliquer", END)

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

Deux types d'arêtes se côtoient. add_edge crée un lien fixe : après appliquer, on va toujours à END. add_conditional_edges délègue la décision à la fonction router : après analyser, le graphe va là où router l'indique. Le troisième argument — la liste des destinations possibles — sert à dessiner le graphe et à valider les noms.

L'appel à compile() fige le graphe et le rend exécutable. C'est ici qu'on branche le checkpointer, dont dépend toute la suite.

Le checkpointer : persister l'état

Un checkpointer sauvegarde l'état du graphe à chaque étape. Sans lui, une exécution est éphémère : elle vit en mémoire le temps d'un appel, puis disparaît. Avec lui, l'état est persisté — on peut inspecter une exécution passée, la reprendre, ou la poursuivre plus tard.

LangGraph en fournit plusieurs. InMemorySaver garde l'état en mémoire : parfait pour un lab ou des tests, perdu à l'arrêt du processus. Pour la production, on branche un checkpointer durable — SqliteSaver pour une persistance locale sur fichier, PostgresSaver pour une base partagée entre plusieurs instances.

Chaque exécution est identifiée par un thread_id, passé dans la configuration. C'est la clé qui relie les invocations entre elles : réutiliser le même thread_id reprend la même exécution ; un nouveau démarre de zéro.

config = {"configurable": {"thread_id": "ticket-aide"}}
sortie = graphe.invoke({"ticket": "Je n'arrive plus à exporter mes factures."}, config)

La validation humaine avec interrupt()

C'est la capacité phare de LangGraph pour des agents sûrs : suspendre l'exécution et rendre la main à un humain. La fonction interrupt(), appelée dans un nœud, met le graphe en pause, sauvegarde son état via le checkpointer, et renvoie le contrôle à l'appelant.

from langgraph.types import interrupt

def controle_humain(etat: EtatTriage) -> dict:
    """Suspend le graphe : une action destructive exige une validation."""
    decision = interrupt(
        {
            "action": etat["action"],
            "ticket": etat["ticket"],
            "question": "Confirmer cette action destructive ?",
        }
    )
    return {"approuve": bool(decision)}

Côté appelant, l'invocation se déroule en deux temps. Le premier invoke() exécute le graphe jusqu'au interrupt(), puis s'arrête : son résultat contient la clé __interrupt__ avec la charge utile transmise — de quoi présenter la décision à l'opérateur.

from langgraph.types import Command

config = {"configurable": {"thread_id": "ticket-spam"}}
sortie = graphe.invoke({"ticket": "PROMO !!! Achetez 10000 abonnés..."}, config)

if "__interrupt__" in sortie:
    demande = sortie["__interrupt__"][0].value
    print("Validation requise :", demande["question"])
    # L'opérateur décide, puis on reprend l'exécution.
    sortie = graphe.invoke(Command(resume=True), config)

Le second invoke() ne repart pas du début : il reprend là où le graphe s'était arrêté, grâce au thread_id partagé. On lui passe un objet Command(resume=...) ; la valeur qu'il porte devient la valeur de retour de l'appel interrupt() dans le nœud. Le nœud controle_humain se ré-exécute alors en entier, interrupt() renvoie cette fois True, et le graphe poursuit jusqu'à appliquer.

Le résultat : une action destructive ne s'exécute jamais sans un feu vert humain explicite. Un refus — Command(resume=False) — conserve le ticket. C'est exactement le garde-fou qu'on veut sur un agent qui agit en production.

Faire tourner l'agent

Sur deux tickets contrastés, le graphe emprunte deux parcours différents :

=== Ticket d'aide — aucune pause ===
Catégorie : Problème technique
Action    : repondre_au_client
Résultat  : Action « repondre_au_client » appliquée.

=== Ticket de spam — graphe en pause ===
Validation requise : Confirmer cette action destructive ?
Action proposée    : fermer_le_ticket

=== Ticket de spam — après validation ===
Action    : fermer_le_ticket
Résultat  : Action « fermer_le_ticket » appliquée.

Le ticket d'aide file de START à END sans interruption. Le spam, lui, est aiguillé vers controle_humain, met le graphe en pause, et n'aboutit qu'après le feu vert de l'opérateur. Un seul graphe, deux parcours — décidés par l'état, pas par du code de branchement éparpillé.

La suite de tests du lab le vérifie sur quatre contrôles : le routage (router aiguille bien selon l'action) et la règle de sécurité (appliquer respecte un refus) sont testés sans modèle, de façon déterministe ; deux tests d'intégration confirment qu'un ticket anodin ne déclenche aucune pause et qu'un spam, lui, suspend bien le graphe.

Dépannage

Symptôme	Cause probable	Solution
`interrupt()` ne met pas le graphe en pause	Graphe compilé sans checkpointer	Passer `checkpointer=` à `compile()`
`Command(resume=...)` repart du début	`thread_id` différent entre les deux `invoke()`	Réutiliser le même `thread_id` dans la config
`KeyError` sur une clé d'état	Nœud qui lit une clé non encore remplie	Vérifier l'ordre des nœuds ; utiliser `etat.get(...)`
Le modèle renvoie une action hors liste	Sortie non contrainte	Typer le champ en `Literal[...]` et `with_structured_output`
`add_conditional_edges` n'atteint jamais un nœud	La fonction de routage ne renvoie pas son nom exact	Faire renvoyer une chaîne identique au nom passé à `add_node`

À retenir

LangGraph modélise un agent comme un graphe d'états : nœuds, arêtes, état partagé — pas une boucle linéaire.
Un nœud renvoie seulement les clés qu'il modifie ; LangGraph les fusionne dans l'état.
Le routage conditionnel isole la décision de parcours du traitement — lisible et testable seule.
Le checkpointer persiste l'état ; InMemorySaver pour un lab, SqliteSaver/PostgresSaver en production.
interrupt() suspend le graphe ; Command(resume=...) le reprend, sa valeur devient le retour de interrupt().
Le checkpointer est obligatoire pour interrupt(), et le thread_id est la clé de reprise.

Prochaines étapes

Agents typés avec PydanticAI La sortie structurée et les retries, réutilisés ici dans un nœud.

Function calling et outils Le contrat de schéma qui contraint la sortie du modèle.

Comprendre le MCP Standardiser les outils d'un agent avec un protocole.

Formation IA générative Le parcours complet : comprendre, exécuter, construire, industrialiser.

Pour aller plus loin

Documentation LangGraph — la référence du framework, dont l'API graphe et la persistance.
Human-in-the-loop avec interrupt() — le mécanisme de pause et de reprise en détail.