LangGraph 2026 : construire des agents IA robustes en Python Mémoire persistante · Cycles · Outils · Human-in-the-loop

1. Pourquoi LangGraph — la limite des chaînes linéaires

LangChain a démocratisé l'accès aux LLM via des chaînes (chains) : on enchaîne un prompt, un modèle, un parser. Simple et efficace pour 80 % des cas d'usage. Mais dès qu'un agent doit reboucler, corriger sa propre sortie, ou maintenir une mémoire entre sessions, les chaînes montrent leurs limites structurelles.

LangGraph n'est pas un remplacement de LangChain — c'est une couche d'orchestration au-dessus. Il utilise les modèles, outils et retrievers LangChain, mais gère lui-même le flux d'exécution. La métaphore exacte : LangChain est la bibliothèque standard, LangGraph est le système d'exploitation.

1.1 Quand LangGraph est le bon outil — et quand il ne l'est pas

LangGraph apporte de la valeur quand votre workflow nécessite au moins l'un de ces besoins : mémoire persistante entre sessions (le même utilisateur revient), cycles (l'agent peut revenir en arrière et corriger), human-in-the-loop (pause pour validation), ou multi-agents coordonnés. Pour un agent simple qui appelle des outils sans mémoire ni rebouclage, create_react_agent de LangChain est plus simple à maintenir et tout aussi efficace. La doc officielle LangGraph v1 le reconnaît explicitement : "LangGraph is a low-level framework — use higher-level abstractions if they fit your needs."

LangGraph vs LangChain vs Agents classiques

Avant de plonger dans le code, un tableau comparatif pour clarifier où LangGraph se situe par rapport aux autres approches. La confusion entre LangGraph et LangChain est fréquente — ils répondent à des besoins différents, et sont souvent complémentaires.

Vue d'ensemble : comment LangGraph orchestre un agent

Voici le flux d'un agent LangGraph typique — chaque flèche représente une transition d'état dans le graphe :

Le checkpointer sauvegarde l'état à chaque étape — la boucle ReAct peut tourner plusieurs fois avant d'atteindre END.

2. Architecture fondamentale : StateGraph, nodes, edges

Tout programme LangGraph tourne autour de trois concepts : le StateGraph (le graphe), les nodes (les nœuds = fonctions Python), et les edges (les arêtes = connexions entre nœuds). Le graphe partage un état unique qui circule entre tous les nœuds.

2.1 StateGraph — le conteneur de workflow

StateGraph est la classe centrale de LangGraph. On lui passe le type du state (un TypedDict), on y ajoute des nœuds et des arêtes, et on l'instancie avec .compile(). L'objet compilé expose invoke(), stream(), ainvoke() et astream() — une interface unifiée quelle que soit la complexité du graphe. START et END sont des nœuds virtuels prédéfinis par LangGraph qui marquent l'entrée et la sortie du graphe.

2.2 Nodes — des fonctions Python ordinaires

Chaque nœud est une fonction Python qui reçoit l'état complet (state: MyState) et retourne un dict partiel des clés à mettre à jour. Rien de magique : pas d'héritage de classe, pas de décorateur obligatoire. La fonction peut appeler un LLM, une base de données, un outil externe, ou appliquer une logique Python pure. On l'enregistre avec graph.add_node("nom", ma_fonction). Le nom du nœud est une chaîne arbitraire — c'est ce nom qu'on utilise dans les arêtes et les fonctions de routage.

Le squelette minimal d'un StateGraph en Python :

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

# 1. Définir l'état partagé
class MyState(TypedDict):
    message: str
    result: str

# 2. Définir les nœuds (fonctions Python)
def process(state: MyState) -> dict:
    # Lit le state, retourne les clés à mettre à jour
    return {"result": state["message"].upper()}

# 3. Construire le graphe
graph = StateGraph(MyState)
graph.add_node("process", process)
graph.add_edge(START, "process")
graph.add_edge("process", END)

# 4. Compiler (obligatoire avant invoke)
app = graph.compile()

# 5. Invoquer
result = app.invoke({"message": "hello world"})
print(result)  # {'message': 'hello world', 'result': 'HELLO WORLD'}

Trois règles fondamentales à retenir :

• Un nœud reçoit le state complet et retourne un dict partiel (seulement les clés à modifier)
• Le state est immutable dans chaque nœud — vous ne le modifiez pas en place, vous retournez des mises à jour
• graph.compile() est obligatoire avant d'appeler invoke(), stream() ou ainvoke()

2.3 Edges — connexions normales et conditionnelles

Les arêtes définissent le flux d'exécution. graph.add_edge("a", "b") crée une arête déterministe : après a, toujours aller à b. graph.add_conditional_edges("a", routing_fn) crée une arête conditionnelle : la fonction routing_fn reçoit le state et retourne le nom du prochain nœud (ou END). C'est ce mécanisme qui permet les cycles — une fonction de routage peut retourner le nom d'un nœud déjà exécuté, créant une boucle contrôlée par la logique Python.

3. Le State — TypedDict, Annotated et reducers

Le state est le concept le plus important de LangGraph. C'est le seul canal de communication entre les nœuds. Chaque nœud lit le state en entrée et retourne les clés à mettre à jour.

3.1 TypedDict — la définition du state

Le state est un TypedDict Python standard. Il sert à la fois de schéma (quelles clés existent, quels types) et de contrat entre les nœuds. L'utilisation de TypedDict (et non d'un dataclass ou d'un objet) est intentionnelle : il permet la sérialisation/désérialisation automatique par les checkpointers (JSON, SQLite, PostgreSQL), sans configuration supplémentaire. Chaque nœud retourne un dict avec seulement les clés à modifier — les autres clés restent inchangées.

Par défaut, quand un nœud retourne {"key": new_value}, LangGraph écrase la valeur existante. Mais parfois on veut accumuler — par exemple, ajouter un message à une liste plutôt que de remplacer la liste. C'est là qu'interviennent les reducers.

3.2 Reducers — contrôler comment le state se met à jour

from typing import TypedDict, Annotated
from operator import add
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage

class AgentState(TypedDict):
    # Reducer add_messages : accumule les messages, déduplique par ID
    messages: Annotated[list[AnyMessage], add_messages]

    # Reducer operator.add : concatène les listes
    documents: Annotated[list[str], add]

    # Pas de reducer : le dernier nœud écrase
    current_step: str
    final_answer: str

3.3 MessagesState — le raccourci pour les agents conversationnels

# Utiliser MessagesState (raccourci pratique)
from langgraph.graph import MessagesState, StateGraph, START, END
from langchain_core.messages import HumanMessage, AIMessage

class MyState(MessagesState):
    # Étendre MessagesState avec des champs supplémentaires
    context: str
    iteration_count: int

# Un nœud qui ajoute un message (pas d'écrasement)
def respond(state: MyState) -> dict:
    return {
        "messages": [AIMessage(content="Bonjour !")],  # ajouté, pas écrasé
        "iteration_count": state.get("iteration_count", 0) + 1,  # écrasé
    }

4. Premier agent avec Ollama — de zéro à graph fonctionnel

On construit un agent conversationnel complet avec Ollama. Prérequis : Ollama installé, modèle qwen3:8b ou llama3.1:8b disponible. Consultez notre guide Ollama complet pour l'installation et le choix des modèles.

4.1 Installation et prérequis

# Installation
pip install -U langgraph langchain-ollama langchain-core

# Vérifier que Ollama tourne avec un modèle tool-capable
ollama pull qwen3:8b    # recommandé : bon rapport qualité/vitesse, supporte les tools
ollama pull llama3.1:8b # alternative Meta, aussi compatible tools

4.2 Construire l'agent minimal

# agent_simple.py
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_ollama import ChatOllama
from langchain_core.messages import HumanMessage, SystemMessage

# Modèle Ollama local
llm = ChatOllama(model="qwen3:8b", temperature=0.7)

# Nœud principal : appel au LLM
def call_llm(state: MessagesState) -> dict:
    messages = [
        SystemMessage(content="Tu es un assistant expert en IA. Réponds en français."),
        *state["messages"]
    ]
    response = llm.invoke(messages)
    return {"messages": [response]}

# Construire et compiler le graphe
graph = StateGraph(MessagesState)
graph.add_node("llm", call_llm)
graph.add_edge(START, "llm")
graph.add_edge("llm", END)
app = graph.compile()

# Invoquer
result = app.invoke({
    "messages": [HumanMessage(content="Explique LangGraph en 3 phrases.")]
})
print(result["messages"][-1].content)

4.3 Tester et inspecter le graphe

LangGraph expose app.get_graph().draw_ascii() pour visualiser la structure du graphe en console. Pour une visualisation graphique, app.get_graph().draw_mermaid_png() génère un PNG via l'API Mermaid (nécessite pip install grandalf). En production, LangSmith permet de tracer chaque exécution nœud par nœud — indispensable pour les graphes complexes. Pour déboguer en local, ajoutez print(state) au début de chaque nœud : le state complet est toujours disponible.

C'est intentionnellement verbose pour un LLM simple — la valeur de LangGraph apparaît quand on ajoute la mémoire, les cycles et les outils dans les sections suivantes.

5. Mémoire et checkpointing — MemorySaver, SqliteSaver, threads

Sans checkpointer, chaque appel à app.invoke() est apatride : le graphe ne se souvient pas des conversations précédentes. Le checkpointer sauvegarde l'état complet à chaque superstep (chaque exécution d'un nœud), permettant :

• La mémoire entre sessions (même utilisateur, conversations séparées)
• La reprise après erreur sans repartir de zéro
• Le human-in-the-loop (pause + reprise plus tard, même sur une autre machine)
• L'historique d'exécution complet (auditabilité)

Chaque conversation est identifiée par un thread_id dans la config. Deux thread_id différents = deux mémoires totalement isolées.

5.1 MemorySaver — développement rapide en RAM

# Mémoire en RAM — développement uniquement (perdu au redémarrage)
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# Le thread_id isole chaque conversation
config_user1 = {"configurable": {"thread_id": "user_alice_conv1"}}
config_user2 = {"configurable": {"thread_id": "user_bob_conv1"}}

# Tour 1 — Alice
app.invoke({"messages": [HumanMessage("Je m'appelle Alice.")]}, config=config_user1)

# Tour 2 — Alice (le graphe se souvient)
result = app.invoke({"messages": [HumanMessage("Comment je m'appelle ?")]}, config=config_user1)
print(result["messages"][-1].content)  # → "Tu t'appelles Alice."

# Bob n'a pas accès à la mémoire d'Alice
result2 = app.invoke({"messages": [HumanMessage("Comment je m'appelle ?")]}, config=config_user2)
print(result2["messages"][-1].content)  # → "Je ne sais pas votre prénom."

5.2 SqliteSaver et PostgresSaver — persistance en production

Pour la production locale, utilisez SqliteSaver — les checkpoints survivent aux redémarrages :

# Production locale — SQLite (persistant sur disque)
from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("./checkpoints.db") as checkpointer:
    app = graph.compile(checkpointer=checkpointer)

    # Même utilisation qu'avec MemorySaver
    config = {"configurable": {"thread_id": "session_42"}}
    app.invoke({"messages": [HumanMessage("Bonjour !")]}, config=config)

# Production distribuée — PostgreSQL
# pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string("postgresql://user:pass@host/db")

5.3 Choisir le bon checkpointer selon votre contexte

6. Tools et ToolNode — pattern ReAct avec Ollama

Un agent sans outils ne fait que générer du texte. Les tools permettent à l'agent d'agir : chercher sur le web, lire des fichiers, appeler des APIs, exécuter du code. LangGraph propose des composants prébuilt comme ToolNode (dans langgraph.prebuilt) pour gérer l'exécution des outils automatiquement.

Le pattern ReAct (Reasoning + Acting) est le plus courant : l'agent raisonne, décide d'appeler un outil, reçoit le résultat, raisonne à nouveau. Ce cycle continue jusqu'à ce que l'agent ait la réponse.

6.1 Définir des outils avec @tool

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_ollama import ChatOllama
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, SystemMessage

# Définir des outils avec le décorateur @tool
@tool
def calculer(expression: str) -> str:
    """Évalue une expression mathématique Python. Ex: '2 ** 10' → '1024'"""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"Erreur : {e}"

@tool
def meteo_actuelle(ville: str) -> str:
    """Retourne la météo actuelle pour une ville (simulation)."""
    meteo = {"paris": "18°C, ensoleillé", "lyon": "15°C, nuageux"}
    return meteo.get(ville.lower(), f"Météo inconnue pour {ville}")

tools = [calculer, meteo_actuelle]

# Modèle bindé avec les outils (génère des tool_calls)
llm = ChatOllama(model="qwen3:8b", temperature=0)
llm_with_tools = llm.bind_tools(tools)

# Nœud agent : appelle le LLM avec les outils disponibles
def agent(state: MessagesState) -> dict:
    messages = [
        SystemMessage(content="Tu es un assistant. Utilise les outils disponibles."),
        *state["messages"]
    ]
    return {"messages": [llm_with_tools.invoke(messages)]}

# ToolNode exécute automatiquement les tool_calls retournés par le LLM
tool_node = ToolNode(tools)

# Construction du graphe ReAct
graph = StateGraph(MessagesState)
graph.add_node("agent", agent)
graph.add_node("tools", tool_node)

graph.add_edge(START, "agent")

# tools_condition : route vers "tools" si l'agent a fait un tool_call, sinon END
graph.add_conditional_edges("agent", tools_condition)

# Après l'outil, toujours revenir à l'agent (cycle ReAct)
graph.add_edge("tools", "agent")

app = graph.compile()

result = app.invoke({
    "messages": [HumanMessage("Quel est 2^32 ? Et la météo à Paris ?")]
})
print(result["messages"][-1].content)
# → "2^32 = 4294967296. La météo à Paris est : 18°C, ensoleillé."

6.2 ToolNode et tools_condition — le cycle ReAct

Le flux d'exécution de ce graphe :

ToolNode est un nœud préconstruit disponible dans langgraph.prebuilt. Il inspecte automatiquement les tool_calls du dernier message, exécute les outils correspondants en parallèle si nécessaire, et retourne les ToolMessage résultants. tools_condition est sa fonction de routage complémentaire : elle retourne "tools" si des tool_calls sont présents, END sinon — implémentant ainsi le cycle ReAct en une seule ligne.

6.3 Modèles Ollama avec support tool calling

7. Cycles et logique conditionnelle — rebouclage et routage

La vraie puissance de LangGraph sur LangChain, c'est la capacité à reboucler. Un nœud peut renvoyer vers un nœud précédent — ce qui est impossible dans une chaîne. Cela permet des patterns comme la validation itérative (l'agent génère, vérifie, corrige jusqu'à satisfaire un critère) ou le reflexion (l'agent critique sa propre réponse).

7.1 Pattern validation itérative — génération avec correction d'erreurs

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated, Literal
from operator import add

class CodeState(TypedDict):
    task: str
    code: str
    errors: Annotated[list[str], add]
    attempts: int
    passed: bool

def generate_code(state: CodeState) -> dict:
    # LLM génère du code
    code = llm.invoke(f"Écris du Python pour : {state['task']}").content
    return {"code": code, "attempts": state.get("attempts", 0) + 1}

def test_code(state: CodeState) -> dict:
    # Exécute et vérifie le code
    try:
        exec(state["code"])
        return {"passed": True}
    except Exception as e:
        return {"passed": False, "errors": [str(e)]}

def route_after_test(state: CodeState) -> Literal["generate_code", "__end__"]:
    # Conditions : OK ou trop de tentatives → END, sinon reboucle
    if state["passed"] or state.get("attempts", 0) >= 3:
        return END
    return "generate_code"  # ← CYCLE : retour vers generate_code

graph = StateGraph(CodeState)
graph.add_node("generate_code", generate_code)
graph.add_node("test_code", test_code)

graph.add_edge(START, "generate_code")
graph.add_edge("generate_code", "test_code")
graph.add_conditional_edges("test_code", route_after_test)

app = graph.compile()

La fonction de routage route_after_test est une fonction Python ordinaire qui retourne le nom du prochain nœud (ou END). Elle a accès à l'état complet — ce qui permet des logiques de routage arbitrairement complexes.

7.2 Pattern Reflection — l'agent critique sa propre réponse

La reflection est une variante du cycle : l'agent génère une réponse, un second nœud "critique" évalue sa qualité (sur des critères comme la précision, la complétude, le format), et renvoie un feedback structuré au nœud de génération. Ce pattern est particulièrement efficace pour la génération de code, de texte long, ou de réponses devant respecter des contraintes formelles. La clé : le nœud critique est souvent un second appel LLM avec un prompt spécialisé, distinct du prompt de génération.

7.3 Protection contre les boucles infinies et recursion_limit

8. Human-in-the-Loop — interrupt() et validation humaine

Le human-in-the-loop (HITL) permet à un humain d'inspecter, modifier ou approuver l'état du graphe avant qu'il continue. C'est critique pour des actions irréversibles : envoyer un email, modifier une base de données, exécuter du code généré.

LangGraph offre deux approches. La méthode préconisée depuis la v0.4 est interrupt() — elle est plus simple et fonctionne nativement en production (contrairement aux breakpoints statiques qui bloquent le thread).

8.1 interrupt() — pause et reprise de l'exécution

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from langchain_ollama import ChatOllama
from langchain_core.messages import HumanMessage

llm = ChatOllama(model="qwen3:8b")

def draft_email(state: MessagesState) -> dict:
    # L'agent rédige un email
    response = llm.invoke([*state["messages"],
        HumanMessage("Rédige un email professionnel basé sur cette demande.")])
    return {"messages": [response]}

def human_review(state: MessagesState) -> Command:
    # Pause l'exécution, attend une validation humaine
    last_email = state["messages"][-1].content

    # interrupt() sauvegarde l'état et lève une exception interne
    # L'exécution reprend quand on appelle Command(resume=...)
    decision = interrupt({
        "action": "review_email",
        "email_draft": last_email,
        "question": "Approuver et envoyer ? (oui/non/modifier)"
    })

    if decision == "oui":
        return Command(goto="send_email")
    elif decision == "non":
        return Command(goto=END)
    else:
        # "modifier" → retour à la génération avec feedback
        return Command(
            goto="draft_email",
            update={"messages": [HumanMessage(f"Modifie l'email : {decision}")]}
        )

def send_email(state: MessagesState) -> dict:
    print("Email envoyé !")
    return {}

graph = StateGraph(MessagesState)
graph.add_node("draft_email", draft_email)
graph.add_node("human_review", human_review)
graph.add_node("send_email", send_email)
graph.add_edge(START, "draft_email")
graph.add_edge("draft_email", "human_review")
graph.add_edge("send_email", END)

app = graph.compile(checkpointer=MemorySaver())  # Un checkpointer est indispensable pour que interrupt() fonctionne

8.2 Command et reprise avec thread_id

# Utilisation
config = {"configurable": {"thread_id": "email_task_1"}}

# Première exécution : s'arrête à l'interrupt
result = app.invoke(
    {"messages": [HumanMessage("Écris un email pour demander un report de réunion.")]},
    config=config
)
# result contient l'interrupt avec l'email rédigé
# result['__interrupt__'][0].value['email_draft'] → email à relire

print("Email à valider :", result["__interrupt__"][0].value["email_draft"])

# L'humain valide — reprise avec Command(resume=...)
from langgraph.types import Command
final = app.invoke(Command(resume="oui"), config=config)
# → Email envoyé !

8.3 Cas d'usage production — au-delà du prototype

Le thread_id dans config identifie le checkpoint. L'exécution peut reprendre des heures ou des jours plus tard, depuis n'importe quelle machine (avec PostgresSaver), en passant le même config. C'est ce qui rend le HITL utilisable en production réelle — et pas seulement en démo. En pratique, le thread_id peut être l'identifiant de session utilisateur, l'ID d'une tâche asynchrone, ou l'ID d'un ticket de workflow : LangGraph ne l'interprète pas, il en fait juste une clé de checkpoint.

9. Multi-agents — architectures supervisor, swarm et subgraphs

Coordonner plusieurs agents spécialisés est un besoin courant dès que les tâches deviennent complexes. LangGraph propose plusieurs architectures, dont le pattern supervisor où un LLM orchestre des agents workers. Ce pattern est popularisé dans l'écosystème LangGraph depuis la v0.3 (mars 2025) avec le package langgraph-supervisor — c'est un pattern parmi d'autres, bien documenté et éprouvé, mais pas la seule approche valide.

9.1 Pattern supervisor — un LLM qui délègue

# pip install langgraph-supervisor
from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent
from langchain_ollama import ChatOllama
from langchain_core.tools import tool

llm = ChatOllama(model="qwen3:14b")

# Agent 1 : spécialisé recherche web
@tool
def recherche_web(query: str) -> str:
    """Effectue une recherche et retourne les résultats."""
    return f"Résultats pour '{query}': [résultats simulés]"

agent_recherche = create_react_agent(
    llm, tools=[recherche_web], name="agent_recherche",
    prompt="Tu es spécialisé dans la recherche d'informations."
)

# Agent 2 : spécialisé analyse et synthèse
@tool
def analyser_donnees(data: str) -> str:
    """Analyse des données et produit un résumé structuré."""
    return f"Analyse de : {data[:100]}..."

agent_analyse = create_react_agent(
    llm, tools=[analyser_donnees], name="agent_analyse",
    prompt="Tu es spécialisé dans l'analyse et la synthèse de données."
)

# Supervisor : coordonne les deux agents
supervisor = create_supervisor(
    llm,
    agents=[agent_recherche, agent_analyse],
    prompt="""Tu coordonnes une équipe d'agents spécialisés.
    - Pour les recherches d'informations → délègue à agent_recherche
    - Pour l'analyse de données → délègue à agent_analyse
    - Synthétise les résultats finaux toi-même."""
).compile()

# Utilisation identique à un agent simple
from langchain_core.messages import HumanMessage
result = supervisor.invoke({
    "messages": [HumanMessage("Analyse les tendances LLM open source en 2026.")]
})

9.2 Swarm, Subgraphs et Map-Reduce — les autres architectures

9.3 Note sur la compatibilité des packages multi-agents

10. Agent RAG — Self-RAG avec rebouclage

Un agent RAG classique (voir notre guide RAG avec Ollama) fait toujours la même chose : embed la question → cherche les chunks → génère. LangGraph permet d'aller plus loin avec le Self-RAG : l'agent évalue la pertinence des documents récupérés et peut reboucler pour améliorer la requête si les résultats sont insuffisants.

10.1 Limites du RAG classique et apport du Self-RAG

Le RAG classique souffre d'un problème structurel : si les documents récupérés sont hors sujet, le LLM génère quand même une réponse — souvent approximative ou inventée. Le Self-RAG ajoute deux nœuds critiques : grade_documents (évalue la pertinence) et rewrite_query (reformule la question si nécessaire). Le cycle se termine quand les documents sont jugés pertinents ou après N tentatives.

10.2 Architecture Self-RAG avec LangGraph

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated, Literal
from operator import add
from langchain_ollama import ChatOllama
from langchain_core.messages import HumanMessage

class RAGState(TypedDict):
    question: str
    rewritten_question: str
    documents: Annotated[list[str], add]  # accumule les docs
    answer: str
    relevant: bool
    attempts: int

llm = ChatOllama(model="qwen3:8b", temperature=0)

def retrieve(state: RAGState) -> dict:
    # Utilise la question réécrite si disponible, sinon l'originale
    query = state.get("rewritten_question") or state["question"]
    # Ici : votre retriever ChromaDB/FAISS
    docs = vectorstore.similarity_search(query, k=3)
    return {"documents": [d.page_content for d in docs]}

def grade_documents(state: RAGState) -> dict:
    # LLM évalue si les documents récupérés sont pertinents
    docs_text = "\n".join(state["documents"][-3:])  # derniers 3 docs
    prompt = f"""Question : {state['question']}
Documents : {docs_text}
Ces documents permettent-ils de répondre à la question ? Réponds UNIQUEMENT par 'oui' ou 'non'."""
    response = llm.invoke([HumanMessage(content=prompt)]).content.strip().lower()
    return {"relevant": "oui" in response}

def rewrite_query(state: RAGState) -> dict:
    # Si docs non pertinents, reformule la question
    prompt = f"Reformule cette question pour améliorer la recherche : {state['question']}"
    rewritten = llm.invoke([HumanMessage(content=prompt)]).content
    return {
        "rewritten_question": rewritten,
        "attempts": state.get("attempts", 0) + 1
    }

def generate(state: RAGState) -> dict:
    context = "\n".join(state["documents"])
    prompt = f"Contexte :\n{context}\n\nQuestion : {state['question']}\nRéponse :"
    answer = llm.invoke([HumanMessage(content=prompt)]).content
    return {"answer": answer}

def route_after_grade(state: RAGState) -> Literal["generate", "rewrite_query"]:
    if state["relevant"] or state.get("attempts", 0) >= 2:
        return "generate"
    return "rewrite_query"

graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve)
graph.add_node("grade_documents", grade_documents)
graph.add_node("rewrite_query", rewrite_query)
graph.add_node("generate", generate)

graph.add_edge(START, "retrieve")
graph.add_edge("retrieve", "grade_documents")
graph.add_conditional_edges("grade_documents", route_after_grade)
graph.add_edge("rewrite_query", "retrieve")  # ← cycle
graph.add_edge("generate", END)

rag_agent = graph.compile()

10.3 Intégrer votre retriever vectoriel existant

Le nœud retrieve de l'exemple utilise vectorstore.similarity_search() — remplacez-le par votre retriever ChromaDB, FAISS, ou Qdrant. La variable vectorstore peut être initialisée une fois en dehors du graphe et capturée par la closure du nœud. Pour un retriever plus sophistiqué, utilisez EnsembleRetriever (LangChain) qui combine dense (embeddings) et sparse (BM25) — cette combinaison améliore significativement le rappel dans les Self-RAG.

11. Streaming — astream() et astream_events()

Pour une interface utilisateur réactive, le streaming permet d'afficher la réponse token par token, ou de montrer l'avancement du graphe en temps réel.

11.1 astream() — streaming des mises à jour du graphe

import asyncio
from langchain_core.messages import HumanMessage

# stream_mode="values" → état complet à chaque superstep
async def stream_values():
    async for state in app.astream(
        {"messages": [HumanMessage("Quel est l'état de l'IA en 2026 ?")]},
        stream_mode="values"
    ):
        last_msg = state["messages"][-1]
        if hasattr(last_msg, "content"):
            print(f"[{last_msg.__class__.__name__}] {last_msg.content[:80]}...")

# stream_mode="updates" → seulement les deltas (nœud + mise à jour)
async def stream_updates():
    async for node_name, update in app.astream(
        {"messages": [HumanMessage("Explique les reducers LangGraph.")]},
        stream_mode="updates"
    ):
        print(f"Nœud '{node_name}' a mis à jour : {list(update.keys())}")

# astream_events() → streaming token par token du LLM
async def stream_tokens():
    async for event in app.astream_events(
        {"messages": [HumanMessage("Qu'est-ce que LangGraph ?")]},
        version="v2"
    ):
        # Filtrer les tokens de génération
        if event["event"] == "on_chat_model_stream":
            chunk = event["data"]["chunk"]
            if chunk.content:
                print(chunk.content, end="", flush=True)

asyncio.run(stream_tokens())

11.2 astream_events() et intégration FastAPI SSE

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

fastapi_app = FastAPI()

@fastapi_app.post("/chat/stream")
async def chat_stream(body: dict):
    async def generate():
        async for event in app.astream_events(
            {"messages": [HumanMessage(body["message"])]}, version="v2"
        ):
            if event["event"] == "on_chat_model_stream":
                chunk = event["data"]["chunk"].content
                if chunk:
                    yield f"data: {json.dumps({'token': chunk})}\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

12. Limites, pièges et quand ne pas utiliser LangGraph

12.1 Quand LangGraph est la mauvaise abstraction

12.2 Points de vigilance spécifiques à la production

Trois points qui ne ressortent pas toujours dans les tutoriels mais causent des problèmes en production :

• Serialisabilité du state : tout ce que vous mettez dans le state doit être sérialisable en JSON (pour SqliteSaver/PostgresSaver). Les objets Python custom, les lambdas, les connexions DB ne passent pas. Gardez le state simple : strings, lists, dicts, ints, bools.
• Thread safety : MemorySaver n'est pas thread-safe. En production multi-workers (gunicorn, uvicorn multi-process), utilisez SqliteSaver ou PostgresSaver — eux sont conçus pour l'accès concurrent.
• Taille des checkpoints : chaque superstep crée un checkpoint. Un agent avec 50 tours de conversation = 50 checkpoints. En SQLite, veillez à purger régulièrement les threads anciens pour éviter une croissance illimitée du fichier .db.

FAQ

Conclusion

LangGraph marque un changement de paradigme dans la construction d'agents IA. Là où les chaînes linéaires convenaient aux prototypes, les applications production de 2026 exigent des garanties que LangGraph est conçu pour apporter : état persistant (vos agents se souviennent), cycles contrôlés (ils peuvent corriger leurs erreurs), human-in-the-loop (les humains restent dans la boucle), et multi-agents (les tâches complexes sont décomposées).

La courbe d'apprentissage est réelle — StateGraph, reducers, checkpointers, Command — mais une fois ces concepts maîtrisés, le framework devient transparent. Les nœuds sont des fonctions Python ordinaires, le state est un TypedDict ordinaire, et le graphe est une description explicite du flux. Pas de magie cachée.