LangGraph est devenu en 2026 la bibliothèque de référence pour orchestrer des agents IA multi-étapes en Python : là où un appel direct à Claude répond à une question, LangGraph permet de construire des workflows entiers — un agent qui recherche, un autre qui rédige, un troisième qui relit, avec des branchements conditionnels, de la persistance d’état entre exécutions et la capacité de reprendre une conversation interrompue. Maintenue par LangChain AI, la bibliothèque a atteint la version 1.2 en mai 2026 et s’utilise aujourd’hui en production chez des startups SaaS, des plateformes mobile money et des outils éditoriaux qui automatisent leurs pipelines avec Claude Opus 4.7 ou Claude Sonnet 4.6.

Ce tutoriel construit un workflow complet pas à pas, depuis l’installation jusqu’au déploiement en production, en expliquant pour chaque bloc de code pourquoi on l’écrit, ce qu’il fait exactement, et comment vérifier que ça marche. Tout le code est testé contre LangGraph 1.2 (package PyPI langgraph 0.3.x) et langchain-anthropic 0.3.x au moment d’écrire ces lignes.

Ce que vous saurez faire à la fin

Comprendre la différence entre LangChain (chaînage simple) et LangGraph (graphes d’état)
Construire un StateGraph avec plusieurs nœuds spécialisés
Ajouter des branchements conditionnels (router) et des boucles d’amélioration
Persister l’état dans PostgreSQL pour reprendre les conversations
Faire du human-in-the-loop avec interrupt_before
Brancher des outils externes (recherche, paiement) via bind_tools
Observer la production avec LangSmith

Étape 1 — Comprendre LangGraph vs LangChain

LangChain a popularisé le chaînage linéaire d’appels LLM : prompt → LLM → parseur → prompt suivant → LLM, etc. Le pattern est élégant pour des tâches simples mais devient ingérable dès qu’on veut des branchements, des boucles ou de la persistance. LangGraph résout cela en modélisant le workflow comme un graphe d’état : chaque nœud est une fonction Python qui lit et écrit dans un état partagé typé, et les arêtes du graphe (déterministes ou conditionnelles) décident du nœud suivant.

Trois différences majeures à retenir. État typé : un TypedDict Python décrit toutes les variables qui circulent entre nœuds, ce qui évite les bugs de clés manquantes. Branchements et boucles : un nœud peut décider de revenir en arrière ou de bifurquer selon une condition. Checkpoints : chaque transition peut être sauvegardée automatiquement, permettant de reprendre une conversation où elle s’était arrêtée — fondamental pour les chatbots persistants ou les workflows interrompus par un humain.

Étape 2 — Installer LangGraph et configurer la clé API

L’installation se fait en deux commandes Python : la bibliothèque LangGraph elle-même, et l’intégration langchain-anthropic qui expose la classe ChatAnthropic permettant d’appeler Claude depuis le code Python.

python -m venv .venv
source .venv/bin/activate  # macOS / Linux ; sous Windows: .venv\Scripts\activate

pip install langgraph langchain-anthropic langgraph-checkpoint-postgres

export ANTHROPIC_API_KEY="sk-ant-api03-..."

Un environnement virtuel isole les dépendances du projet de votre installation système — c’est une pratique standard Python qui évite les conflits de versions. La clé API Anthropic est exportée en variable d’environnement ; LangChain la détecte automatiquement quand on instancie ChatAnthropic(). Un test rapide pour valider l’installation : lancer python -c "from langgraph.graph import StateGraph; print('OK')" dans le terminal — la sortie doit être simplement OK sans erreur d’import.

Étape 3 — Définir l’État (StateGraph)

L’état est le contrat entre tous les nœuds du workflow. On le déclare une fois pour toutes avec TypedDict de la bibliothèque standard Python. Pour les champs qu’on veut accumuler au fil des nœuds (par exemple une liste de messages), on utilise Annotated avec un réducteur — typiquement add du module operator pour les listes.

from typing import TypedDict, Annotated, List
from operator import add

class WorkflowState(TypedDict):
    user_query: str
    research_results: Annotated[List[dict], add]
    draft_article: str
    review_feedback: str
    final_article: str
    iteration_count: int

Cette structure dit à LangGraph que research_results est une liste qui sera append-only entre nœuds (chaque nœud qui retourne {"research_results": [...]} additionne sa contribution au lieu d’écraser). Les autres champs sont des strings ou ints remplacés à chaque écriture. Si vous oubliez l’annotation pour une liste que plusieurs nœuds remplissent, vous perdrez les données antérieures — c’est une des erreurs les plus fréquentes en LangGraph.

Étape 4 — Premier workflow simple à trois nœuds

On construit maintenant un workflow trivial mais réaliste : un nœud recherche, un nœud rédaction, un nœud relecture. Chaque nœud est une fonction Python qui prend l’état complet en entrée et retourne un dictionnaire partiel à fusionner.

from langchain_anthropic import ChatAnthropic
from langgraph.graph import StateGraph, END

llm = ChatAnthropic(model="claude-sonnet-4-6", max_tokens=2048)

def research_node(state: WorkflowState):
    """Agent qui fait de la recherche sur le sujet."""
    query = state["user_query"]
    response = llm.invoke(f"Liste 5 points clés à connaître sur : {query}")
    return {"research_results": [{"content": response.content}]}

def writer_node(state: WorkflowState):
    """Agent qui rédige un brouillon."""
    research = state["research_results"][-1]["content"]
    response = llm.invoke(
        f"À partir de cette recherche :\n{research}\n\n"
        f"Rédige un article de 400 mots sur : {state['user_query']}"
    )
    return {"draft_article": response.content}

def review_node(state: WorkflowState):
    """Agent qui relit et donne un feedback."""
    response = llm.invoke(
        f"Relis ce brouillon et donne un feedback en 3 lignes max.\n"
        f"Si le texte est prêt à publier, écris exactement 'PRET'.\n\n"
        f"{state['draft_article']}"
    )
    return {
        "review_feedback": response.content,
        "final_article": state["draft_article"] if "PRET" in response.content.upper() else "",
        "iteration_count": state.get("iteration_count", 0) + 1
    }

# Construction du graphe
workflow = StateGraph(WorkflowState)
workflow.add_node("research", research_node)
workflow.add_node("write", writer_node)
workflow.add_node("review", review_node)

workflow.set_entry_point("research")
workflow.add_edge("research", "write")
workflow.add_edge("write", "review")
workflow.add_edge("review", END)

app = workflow.compile()

Trois choses à observer. Le modèle claude-sonnet-4-6 est le modèle généraliste de la génération 4.6 d’Anthropic, équilibre raison/coût optimal pour ce type d’orchestration. Chaque nœud reçoit l’état complet mais ne retourne que les champs modifiés — LangGraph fusionne automatiquement. Les arêtes workflow.add_edge("a", "b") sont déterministes : après le nœud « a », on passe systématiquement à « b ». Au prochain démarrage, app.invoke() exécutera research → write → review → fin.

Étape 5 — Exécuter et visualiser le workflow

L’exécution se fait via la méthode invoke sur l’application compilée. On passe un état initial qui doit contenir au moins les champs sans valeur par défaut.

result = app.invoke({
    "user_query": "Stratégie pricing pour PME e-commerce à Dakar",
    "research_results": [],
    "draft_article": "",
    "review_feedback": "",
    "final_article": "",
    "iteration_count": 0
})

print("Feedback :", result["review_feedback"])
print("Article final :", result["final_article"] or result["draft_article"])

# Visualiser le graph sous forme de diagramme Mermaid
print(app.get_graph().draw_mermaid())

Le résultat est l’état final après que tous les nœuds ont tourné séquentiellement. La sortie attendue contient un feedback texte de 1-3 phrases et l’article final, soit le brouillon validé soit le brouillon brut si la relecture n’a pas accepté. La méthode draw_mermaid() produit un diagramme texte que vous pouvez copier dans n’importe quel éditeur Mermaid Live pour visualiser le flux — pratique pour documenter le workflow ou debugger un branchement.

Étape 6 — Ajouter des branchements conditionnels

La puissance de LangGraph apparaît dès qu’on introduit des branchements. Plutôt que de toujours aller du nœud « review » à END, on veut une boucle : si le brouillon n’est pas prêt, on retourne au nœud « write » pour itérer ; sinon on publie. C’est le rôle de add_conditional_edges.

def should_iterate(state: WorkflowState) -> str:
    """Décide si on itère ou si on termine."""
    if state["iteration_count"] >= 3:
        return "publish"  # cap à 3 itérations pour éviter la boucle infinie
    if "PRET" in state.get("review_feedback", "").upper():
        return "publish"
    return "rewrite"

# Remplacer l'edge déterministe review→END par un branchement
workflow = StateGraph(WorkflowState)
workflow.add_node("research", research_node)
workflow.add_node("write", writer_node)
workflow.add_node("review", review_node)
workflow.set_entry_point("research")
workflow.add_edge("research", "write")
workflow.add_edge("write", "review")
workflow.add_conditional_edges(
    "review",
    should_iterate,
    {
        "rewrite": "write",   # boucle de retour pour améliorer
        "publish": END
    }
)
app = workflow.compile()

La fonction should_iterate est un simple Python qui retourne une chaîne — clé du dictionnaire passé à add_conditional_edges. LangGraph appelle cette fonction après chaque exécution de « review » et oriente vers le nœud correspondant. Le plafond à 3 itérations est essentiel : sans lui, un modèle qui ne décide jamais « PRET » produit une boucle infinie. Un signal de réussite typique : l’article s’améliore visiblement entre l’itération 1 et 2, et la 3ème itération sort sur « PRET ».

Étape 7 — Persistance avec PostgresSaver (checkpoints)

Pour un chatbot ou un workflow long, il faut pouvoir interrompre puis reprendre exactement là où on s’était arrêté — par exemple si l’utilisateur ferme l’onglet et revient une heure plus tard. LangGraph gère cela avec un checkpointer qui sérialise l’état à chaque transition. Pour la production, on utilise PostgresSaver qui stocke dans une base PostgreSQL.

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@localhost:5432/langgraph_db?sslmode=disable"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # À exécuter UNE SEULE FOIS pour créer les tables nécessaires
    checkpointer.setup()
    app = workflow.compile(checkpointer=checkpointer)

    # Reprendre une conversation interrompue avec un thread_id stable
    config = {"configurable": {"thread_id": "user-221771234567"}}
    result = app.invoke({"user_query": "..."}, config=config)

    # Time travel : inspecter l'historique des états
    for snapshot in app.get_state_history(config):
        print(snapshot.values.get("iteration_count"), snapshot.next)

Trois points importants. checkpointer.setup() doit être appelé une seule fois au premier déploiement pour créer les tables internes — l’oublier provoque l’erreur « relation does not exist » au premier appel. Le thread_id est l’identifiant logique de la conversation : avec le même thread_id, l’état précédent est restauré automatiquement. Le time travel via get_state_history liste tous les états passés et permet de relancer à partir de n’importe lequel — extrêmement utile pour debugger ou pour proposer un « undo » à l’utilisateur.

Étape 8 — Système multi-agents spécialisés

Jusqu’ici nos nœuds appellent le même LLM avec des prompts différents. La logique multi-agents pousse plus loin : chaque agent a son propre rôle, son propre modèle (Haiku pour les tâches simples, Opus pour les complexes), et le graphe orchestre leur collaboration.

llm_fast = ChatAnthropic(model="claude-haiku-4-5", max_tokens=1024)
llm_smart = ChatAnthropic(model="claude-opus-4-7", max_tokens=4096)

class MultiAgentState(TypedDict):
    task: str
    research_data: dict
    code_implementation: str
    tests_results: str
    final_output: str

def researcher_agent(state):
    """Cherche les bonnes pratiques — Haiku suffit ici."""
    response = llm_fast.invoke(
        f"Liste les bonnes pratiques pour : {state['task']}"
    )
    return {"research_data": {"content": response.content}}

def architect_agent(state):
    """Conçoit la solution — exige Opus pour le raisonnement."""
    response = llm_smart.invoke(
        f"En t'appuyant sur cette recherche :\n{state['research_data']['content']}\n"
        f"Propose une architecture pour : {state['task']}"
    )
    return {"code_implementation": response.content}

def reviewer_agent(state):
    """Relit et challenge — Opus pour le sens critique."""
    response = llm_smart.invoke(
        f"Critique l'architecture suivante en listant les risques :\n"
        f"{state['code_implementation']}"
    )
    return {"tests_results": response.content, "final_output": state["code_implementation"]}

Utiliser Haiku pour la recherche et Opus pour la conception réduit typiquement le coût d’exécution de 60 à 80 % par rapport à tout faire avec Opus. La règle empirique : prendre Haiku pour les tâches déterministes (classification, extraction de liste, reformulation simple) et Opus pour les tâches qui demandent du raisonnement multi-étapes ou de la créativité contrainte. Pour un workflow facturé à l’usage, ce mix peut faire la différence entre une marge de 5 % et une marge de 40 %.

Étape 9 — Parallélisation pour la performance

Quand plusieurs nœuds n’ont pas de dépendance entre eux, on peut les exécuter en parallèle. LangGraph 1.2 le supporte nativement via la primitive Send ou via l’API asynchrone ainvoke. L’exemple ci-dessous lance trois recherches simultanées sur trois sous-sujets.

import asyncio
from typing import List

async def parallel_research(state) -> dict:
    """Lance N recherches en parallèle via asyncio.gather."""
    topics = state["topics"]  # ex: ["fiscalité", "logistique", "marketing"]

    async def one_search(topic: str) -> str:
        resp = await llm_fast.ainvoke(f"Recherche concise sur : {topic}")
        return resp.content

    results = await asyncio.gather(*[one_search(t) for t in topics])
    return {"research_results": [{"topic": t, "content": r} for t, r in zip(topics, results)]}

# Le nœud est désormais une coroutine ; LangGraph la reconnaît automatiquement
workflow.add_node("research", parallel_research)
app = workflow.compile()

# Lancement asynchrone
result = await app.ainvoke({"topics": ["fiscalité", "logistique", "marketing"]})

Sans parallélisation, trois recherches Haiku séquentielles prennent typiquement 9 à 12 secondes. En parallèle, le temps total est celui de la recherche la plus longue, soit 3 à 4 secondes. Pour un agent qui doit consulter plusieurs sources avant de répondre à un utilisateur, c’est la différence entre une perception « lente » et « instantanée ». Attention : ainvoke et await exigent que vous tourniez dans une boucle asyncio — pas de magie côté top-level scripts synchrones.

Étape 10 — Outils externes (Tools / Function Calling)

Un agent purement LLM est limité par ce que Claude sait à sa date de cutoff. Les tools changent la donne : on déclare des fonctions Python avec le décorateur @tool, et Claude peut décider de les appeler lui-même quand il en a besoin. C’est le mécanisme de function calling standardisé.

from langchain_core.tools import tool

@tool
def recherche_stock(produit: str) -> str:
    """Recherche un produit dans le stock du magasin à Dakar.
    Retourne les références disponibles avec leur prix en FCFA."""
    # Logique réelle : appel à votre base PostgreSQL / API ERP
    catalogue = {
        "laptop hp": "HP Pavilion 15 — 489 000 FCFA — 3 en stock",
        "iphone": "iPhone 15 — 720 000 FCFA — rupture",
    }
    return catalogue.get(produit.lower(), "Aucune référence trouvée pour : " + produit)

@tool
def envoyer_facture(client_phone: str, montant_fcfa: int) -> str:
    """Envoie une demande de paiement Wave au client (mobile money Sénégal)."""
    # Logique réelle : appel à l'API Wave Business
    return f"Demande Wave de {montant_fcfa} FCFA envoyée au {client_phone}"

llm_with_tools = llm_smart.bind_tools([recherche_stock, envoyer_facture])

def agent_with_tools(state):
    """Le LLM décide seul d'appeler les tools quand pertinent."""
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": state["messages"] + [response]}

Le décorateur @tool de langchain_core.tools est l’import moderne (l’ancien from langchain.tools import tool est déprécié depuis LangChain 0.1). La docstring de chaque fonction est cruciale : c’est ce que Claude lit pour décider quand l’appeler. Une docstring vague produit des appels manqués ; une docstring précise donne des appels pertinents. Quand Claude décide d’appeler recherche_stock("laptop hp"), LangGraph intercepte, exécute la fonction Python locale, et ré-injecte le résultat dans la conversation pour que Claude continue son raisonnement.

Étape 11 — Human-in-the-loop avec interrupt_before

Pour les actions à risque — un paiement, un envoi d’email, une suppression — on veut souvent qu’un humain valide avant exécution. LangGraph le permet en posant un point d’arrêt avant un nœud sensible.

app = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["envoi_paiement"]  # le graph s'arrête AVANT ce nœud
)

# 1. Premier appel — le graph tourne jusqu'au point d'arrêt
config = {"configurable": {"thread_id": "session-42"}}
result = app.invoke(initial_state, config=config)
# Inspection : où en sommes-nous ?
state = app.get_state(config)
print("Prochain nœud :", state.next)  # ("envoi_paiement",)
print("État actuel :", state.values)

# 2. Validation humaine (UI, Slack, email...)
# Si OK, on reprend ; sinon on modifie l'état avant de continuer

# 3. Reprendre l'exécution
final = app.invoke(None, config=config)

L’astuce est app.invoke(None, config=config) : passer None comme input signale à LangGraph de reprendre à partir du checkpoint précédent. Entre le premier appel et le second, votre application peut afficher l’état à un opérateur dans une UI, attendre une confirmation, voire modifier l’état via app.update_state(config, {"montant_fcfa": 50000}) si l’humain corrige une valeur. Ce pattern est fondamental pour tout workflow qui touche à de l’argent ou à des données critiques.

Étape 12 — Monitoring avec LangSmith

En production, on veut tracer chaque exécution, mesurer les coûts par étape, et identifier les nœuds qui échouent. LangSmith est le service de tracing géré par LangChain AI ; son activation tient en trois variables d’environnement.

export LANGSMITH_API_KEY="lsv2_pt_..."
export LANGSMITH_TRACING=true
export LANGSMITH_PROJECT="kafka-ecommerce-prod"

Une fois ces variables exportées, chaque invocation de votre app est automatiquement loggée vers le dashboard LangSmith — durée par nœud, tokens consommés, prompts complets, réponses LLM. L’interface web permet de filtrer par projet, de comparer deux exécutions, d’identifier le nœud qui consomme 80 % du budget. Pour une équipe qui débute en production, brancher LangSmith dès le J1 économise des heures de debugging.

Étape 13 — Déploiement Docker + API REST

Pour exposer votre workflow comme service, on l’encapsule dans une API FastAPI et on conteneurise. Le squelette minimal ci-dessous fournit un endpoint POST /workflow qui reçoit la requête utilisateur et retourne le résultat final.

# app.py
from fastapi import FastAPI
from pydantic import BaseModel
# ... imports du workflow ci-dessus ...

api = FastAPI(title="LangGraph workflow API")

class Request(BaseModel):
    query: str
    thread_id: str | None = None

@api.post("/workflow")
def run_workflow(req: Request):
    config = {"configurable": {"thread_id": req.thread_id or "anon"}}
    result = app.invoke({"user_query": req.query, "research_results": []}, config=config)
    return {"final": result.get("final_article") or result.get("draft_article")}

# Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app:api", "--host", "0.0.0.0", "--port", "8000"]

Le requirements.txt contient les mêmes paquets que l’étape 2 plus fastapi et uvicorn. L’image Python 3.12-slim pèse environ 130 Mo une fois construite, ce qui tient confortablement sur un Hetzner CX22 à 4,49 euros par mois. En production, on ajoute un reverse proxy Nginx ou Caddy en façade pour TLS et rate-limiting, et on monitore via docker logs couplé à LangSmith.

Étape 14 — Cas d’usage concret : assistant e-commerce Dakar

Pour rassembler tout ce qui précède, voici un workflow type pour un assistant e-commerce sénégalais : il classifie l’intention du client (qualifier, produit, commander), route vers le bon agent, consulte le stock, propose un paiement Wave, et passe par une validation humaine avant tout débit. La trame du graphe :

workflow = StateGraph(AgentState)
workflow.add_node("intention", noeud_intention)
workflow.add_node("recherche_produit", noeud_produit)
workflow.add_node("creer_commande", noeud_commander)
workflow.add_node("qualifier", noeud_qualifier)
workflow.add_node("envoi_paiement", noeud_paiement)

workflow.set_entry_point("intention")
workflow.add_conditional_edges("intention", lambda s: {
    "produit": "recherche_produit",
    "commander": "creer_commande",
    "qualifier": "qualifier"
}.get(s["intention"], END))

workflow.add_edge("recherche_produit", END)
workflow.add_edge("qualifier", END)
workflow.add_edge("creer_commande", "envoi_paiement")
workflow.add_edge("envoi_paiement", END)

with PostgresSaver.from_conn_string(DB_URI) as cp:
    cp.setup()
    app = workflow.compile(checkpointer=cp, interrupt_before=["envoi_paiement"])

Ce graphe combine cinq techniques vues précédemment : classification d’intention (étape 4), branchement conditionnel (étape 6), persistance Postgres (étape 7), tools pour la recherche de stock et l’API Wave (étape 10), et point d’arrêt humain avant débit (étape 11). Pour une PME de Dakar qui veut industrialiser son support WhatsApp, le coût d’exploitation est de l’ordre de 30 à 50 euros par mois en VPS plus 0,5 à 2 USD par 1000 conversations chez Anthropic — largement rentable dès la première dizaine de ventes générées.

Erreurs classiques en LangGraph

Erreur	Cause	Solution
`KeyError: 'research_results'` dans un nœud	État initial incomplet à `invoke`	Fournir tous les champs `TypedDict` au démarrage
Liste écrasée entre deux nœuds	Champ liste sans `Annotated[..., add]`	Ajouter le réducteur dans la définition d’état
`relation "checkpoints" does not exist`	`checkpointer.setup()` jamais appelé	L’invoquer une fois au premier déploiement
Boucle infinie review → write → review	Pas de plafond d’itérations dans le router	Cap à 3-5 dans `should_iterate`
Tool jamais appelé par Claude	Docstring vague ou ambiguë	Reformuler : « Recherche X et retourne Y »
Latence x3 après ajout d’un nœud	Appel séquentiel d’opérations indépendantes	Paralléliser avec `ainvoke` + `asyncio.gather`

LangGraph : orchestrer des agents Claude multi-étapes