LangGraph : Créer des agents IA avancés avec des graphes d'états

20 min de lecture 4159 mots
tl;dr: LangGraph est le framework avancé de LangChain pour créer des agents avec des graphes d'états. Il permet de définir des workflows complexes avec cycles, conditions, et états persistants. Idéal pour agents multi-étapes, systèmes de révision, et architectures multi-agents sophistiquées.

Les agents classiques LangChain sont puissants, mais montrent leurs limites face à des workflows complexes nécessitant des cycles, des révisions, ou des états partagés. LangGraph résout ces défis en introduisant une approche basée sur les graphes d’états, permettant de créer des agents véritablement sophistiqués et contrôlables. Dans cet article, nous explorons ce framework de bout en bout avec des exemples pratiques.

💡 🎯 LangGraph = LangChain + State Machines. Construisez des workflows IA arbitrairement complexes avec logique de contrôle avancée, impossibles avec des chains simples.

Diagramme d’architecture LangChain illustrant LangGraph pour des workflows complexes pour le développement d’applications IA

Qu’est-ce que LangGraph ?

LangGraph est un framework développé par LangChain pour construire des agents avec état (stateful agents) sous forme de graphes cycliques. Contrairement aux agents ReAct classiques qui suivent un pattern linéaire, LangGraph permet de définir explicitement :

  • Nœuds (Nodes) : Les étapes de traitement (LLM, outils, fonctions)
  • Arêtes (Edges) : Les transitions entre étapes (conditionnelles ou fixes)
  • État (State) : Les données partagées et persistantes entre nœuds
  • Cycles : Possibilité de boucler et réviser

Pourquoi LangGraph ?

Limites des agents classiques :

Agent ReAct standard : Input → Thought → Action → Observation → Final Answer
                       (linéaire, pas de révision possible)

Avec LangGraph :

Input → Recherche → Rédaction → Révision → [KO] Recherche (boucle)
                              → [OK] Validation → Output

Cas d’usage idéaux :

  • ✅ Workflows avec révision/validation (génération de contenu, code review)
  • ✅ Agents multi-étapes avec décisions complexes
  • ✅ Systèmes multi-agents avec orchestration
  • ✅ Persistance d’état entre sessions
  • ✅ Workflows cycliques (itération jusqu’à succès)

Concepts fondamentaux

Le State (État)

L’état est la mémoire partagée entre tous les nœuds du graphe.

from typing import TypedDict, Annotated
from langgraph.graph import add_messages

class AgentState(TypedDict):
    """État partagé de l'agent"""
    messages: Annotated[list, add_messages]  # Historique des messages
    current_task: str  # Tâche en cours
    iteration_count: int  # Compteur d'itérations
    final_output: str  # Résultat final

Reducer functions : add_messages fusionne intelligemment les nouveaux messages.

Les Nodes (Nœuds)

Chaque nœud est une fonction qui reçoit l’état, le modifie, et le retourne.

def research_node(state: AgentState) -> AgentState:
    """Nœud de recherche"""
    query = state["current_task"]
    results = search_tool(query)

    # Mise à jour de l'état
    state["messages"].append({
        "role": "system",
        "content": f"Résultats de recherche : {results}"
    })

    return state

Les Edges (Arêtes)

Les arêtes définissent les transitions :

Arêtes fixes : Toujours la même transition

graph.add_edge("research", "analysis")  # research → analysis

Arêtes conditionnelles : Décision basée sur l’état

def should_continue(state: AgentState) -> str:
    """Décide de la prochaine étape"""
    if state["iteration_count"] > 5:
        return "finish"
    if quality_check(state["final_output"]):
        return "validate"
    return "revise"

graph.add_conditional_edges(
    "analysis",
    should_continue,
    {
        "finish": END,
        "validate": "validation",
        "revise": "research"  # Boucle !
    }
)

Le Graph (Graphe)

Assemble nœuds et arêtes en un workflow exécutable.

from langgraph.graph import StateGraph, END

# Création du graphe
workflow = StateGraph(AgentState)

# Ajout des nœuds
workflow.add_node("research", research_node)
workflow.add_node("analysis", analysis_node)
workflow.add_node("validation", validation_node)

# Définition du point d'entrée
workflow.set_entry_point("research")

# Ajout des arêtes
workflow.add_edge("research", "analysis")
workflow.add_conditional_edges("analysis", should_continue)

# Compilation
app = workflow.compile()

Installation et configuration

Installation

# LangGraph et dépendances
pip install langgraph langchain langchain-openai

# Pour la persistance (optionnel)
pip install langgraph-checkpoint-sqlite

# Pour la visualisation (optionnel)
pip install grandalf

Configuration de base

import os
from dotenv import load_dotenv

load_dotenv()
os.environ["OPENAI_API_KEY"] = "sk-..."

# LangSmith pour le monitoring (optionnel)
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-..."
os.environ["LANGCHAIN_PROJECT"] = "langgraph-experiments"

Premier exemple : Chatbot simple

Commençons par un chatbot basique pour comprendre les mécaniques.

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI

# 1. Définir l'état
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 2. Créer le nœud de chat
llm = ChatOpenAI(model="gpt-4o-mini")

def chatbot(state: State):
    """Nœud qui appelle le LLM"""
    return {"messages": [llm.invoke(state["messages"])]}

# 3. Construire le graphe
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

# 4. Compiler
graph = graph_builder.compile()

# 5. Utiliser
def chat(user_input: str):
    response = graph.invoke({
        "messages": [{"role": "user", "content": user_input}]
    })
    return response["messages"][-1].content

# Test
print(chat("Bonjour, qui es-tu ?"))
print(chat("Quelle est la capitale de la France ?"))
💡 📝 Le state est passé automatiquement entre les nodes. Chaque node reçoit le state actuel, le modifie, et le retourne.

Exemple intermédiaire : Agent de recherche avec révision

Agent qui recherche, rédige, puis révise jusqu’à qualité suffisante.

from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun
import re

# État
class ResearchState(TypedDict):
    query: str
    research_results: str
    draft: str
    revision_count: int
    final_report: str
    quality_score: float

# Outils
llm = ChatOpenAI(model="gpt-4", temperature=0)
search = DuckDuckGoSearchRun()

# Nœuds
def research_node(state: ResearchState) -> ResearchState:
    """Recherche d'informations"""
    print(f"\n🔍 Recherche sur : {state['query']}")
    results = search.run(state["query"])

    return {
        **state,
        "research_results": results[:2000]  # Limiter la taille
    }

def writing_node(state: ResearchState) -> ResearchState:
    """Rédaction du rapport"""
    print(f"\n✍️ Rédaction (tentative {state['revision_count'] + 1})")

    prompt = f"""En te basant sur ces informations de recherche, rédige un rapport clair et structuré.

Informations :
{state['research_results']}

Sujet : {state['query']}

Rapport (3-5 paragraphes) :"""

    response = llm.invoke(prompt)
    draft = response.content

    return {
        **state,
        "draft": draft,
        "revision_count": state["revision_count"] + 1
    }

def review_node(state: ResearchState) -> ResearchState:
    """Révision et notation de qualité"""
    print("\n📋 Révision du rapport")

    prompt = f"""Évalue ce rapport sur 10 selon ces critères :
- Clarté et structure
- Exactitude des informations
- Complétude

Rapport :
{state['draft']}

Réponds uniquement avec un score sur 10 suivi de tes commentaires.
Format : "Score: X/10. Commentaires: ..."
"""

    response = llm.invoke(prompt)
    review = response.content

    # Extraire le score
    match = re.search(r'Score:\s*(\d+)/10', review)
    score = float(match.group(1)) / 10 if match else 0.5

    print(f"Score : {score * 10}/10")

    return {
        **state,
        "quality_score": score
    }

def finalize_node(state: ResearchState) -> ResearchState:
    """Finalisation"""
    print("\n✅ Rapport finalisé")
    return {
        **state,
        "final_report": state["draft"]
    }

# Routage conditionnel
def should_continue(state: ResearchState) -> Literal["revise", "finalize"]:
    """Décide si on révise ou finalise"""

    # Limite de sécurité
    if state["revision_count"] >= 3:
        print("⚠️ Limite de révisions atteinte")
        return "finalize"

    # Qualité suffisante ?
    if state["quality_score"] >= 0.7:
        return "finalize"

    print("🔄 Qualité insuffisante, nouvelle révision")
    return "revise"

# Construction du graphe
workflow = StateGraph(ResearchState)

# Nœuds
workflow.add_node("research", research_node)
workflow.add_node("writing", writing_node)
workflow.add_node("review", review_node)
workflow.add_node("finalize", finalize_node)

# Arêtes
workflow.set_entry_point("research")
workflow.add_edge("research", "writing")
workflow.add_edge("writing", "review")

workflow.add_conditional_edges(
    "review",
    should_continue,
    {
        "revise": "writing",  # Boucle de révision
        "finalize": "finalize"
    }
)

workflow.add_edge("finalize", END)

# Compilation
app = workflow.compile()

# Utilisation
def generate_report(query: str) -> str:
    initial_state = {
        "query": query,
        "research_results": "",
        "draft": "",
        "revision_count": 0,
        "final_report": "",
        "quality_score": 0.0
    }

    result = app.invoke(initial_state)
    return result["final_report"]

# Test
report = generate_report("Quelles sont les dernières avancées en IA générative en 2025 ?")
print("\n" + "="*50)
print("RAPPORT FINAL")
print("="*50)
print(report)

Ce que fait cet agent :

  1. Recherche sur le web
  2. Rédige un rapport
  3. Évalue la qualité
  4. Si score < 7/10 : refait la rédaction (max 3 fois)
  5. Sinon : finalise

Exemple avancé : Agent multi-outils avec état complexe

Agent capable de rechercher, calculer, et générer du code selon les besoins.

from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain.tools import tool

# Définition de l'état
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    next_action: str

# Outils
@tool
def search_web(query: str) -> str:
    """Recherche sur le web"""
    from langchain_community.tools import DuckDuckGoSearchRun
    search = DuckDuckGoSearchRun()
    return search.run(query)

@tool
def calculate(expression: str) -> str:
    """Calcule une expression mathématique"""
    try:
        import ast
        import operator

        ops = {
            ast.Add: operator.add,
            ast.Sub: operator.sub,
            ast.Mult: operator.mul,
            ast.Div: operator.truediv,
            ast.Pow: operator.pow
        }

        def eval_expr(node):
            if isinstance(node, ast.Num):
                return node.n
            elif isinstance(node, ast.BinOp):
                return ops[type(node.op)](
                    eval_expr(node.left),
                    eval_expr(node.right)
                )
            else:
                raise TypeError(node)

        result = eval_expr(ast.parse(expression, mode='eval').body)
        return str(result)
    except Exception as e:
        return f"Erreur : {str(e)}"

@tool
def generate_python_code(task: str) -> str:
    """Génère du code Python pour une tâche"""
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    prompt = f"Génère du code Python concis et fonctionnel pour : {task}"
    response = llm.invoke(prompt)
    return response.content

tools = [search_web, calculate, generate_python_code]

# LLM avec tools
llm = ChatOpenAI(model="gpt-4", temperature=0)
llm_with_tools = llm.bind_tools(tools)

# Nœud d'agent
def agent_node(state: AgentState) -> AgentState:
    """Nœud principal qui décide des actions"""
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# Routage
def should_continue(state: AgentState) -> Literal["tools", "end"]:
    """Décide si on utilise des outils ou on termine"""
    last_message = state["messages"][-1]

    # Si le LLM veut utiliser des outils
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"

    # Sinon, on termine
    return "end"

# Construction du graphe
workflow = StateGraph(AgentState)

# Nœuds
workflow.add_node("agent", agent_node)
workflow.add_node("tools", ToolNode(tools))

# Flux
workflow.set_entry_point("agent")

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        "end": END
    }
)

workflow.add_edge("tools", "agent")  # Retour vers l'agent après outil

# Compilation
app = workflow.compile()

# Utilisation
def run_agent(query: str):
    """Exécute l'agent"""
    response = app.invoke({
        "messages": [HumanMessage(content=query)]
    })

    return response["messages"][-1].content

# Tests
print(run_agent("Quelle est la racine carrée de 144 ?"))
# → Utilise calculate

print(run_agent("Cherche les dernières nouvelles sur l'IA"))
# → Utilise search_web

print(run_agent("Génère du code pour trier une liste de nombres"))
# → Utilise generate_python_code

print(run_agent("Combien font 15 * 23 et qui a inventé la multiplication ?"))
# → Utilise calculate puis search_web

Persistance avec checkpoints

LangGraph permet de sauvegarder l’état pour reprendre plus tard.

from langgraph.checkpoint.sqlite import SqliteSaver

# Création du checkpoint
memory = SqliteSaver.from_conn_string(":memory:")  # Ou chemin fichier

# Compilation avec mémoire
app = workflow.compile(checkpointer=memory)

# Configuration de thread
config = {"configurable": {"thread_id": "conversation-1"}}

# Première interaction
response1 = app.invoke(
    {"messages": [HumanMessage(content="Je m'appelle Alice")]},
    config=config
)

# Deuxième interaction (se souvient)
response2 = app.invoke(
    {"messages": [HumanMessage(content="Quel est mon nom ?")]},
    config=config
)

print(response2["messages"][-1].content)
# → "Votre nom est Alice"

Cas d’usage :

  • Chatbots conversationnels persistants
  • Workflows longs avec pauses
  • Reprise après erreur
⚠️ Warning
👥 Human-in-the-loop nécessite checkpointing pour sauvegarder le state entre interruptions. Sans checkpointer, impossible de mettre en pause et reprendre.

Streaming des résultats

Afficher les résultats au fur et à mesure.

# Mode streaming
for event in app.stream(
    {"messages": [HumanMessage(content="Explique la photosynthèse")]},
    stream_mode="values"
):
    # Affiche chaque mise à jour de l'état
    last_message = event["messages"][-1]
    print(f"{last_message.type}: {last_message.content[:100]}")

# Ou streaming token par token
for chunk in app.stream(
    {"messages": [HumanMessage(content="Raconte une histoire")]},
    stream_mode="messages"
):
    print(chunk.content, end="", flush=True)

Visualisation du graphe

from IPython.display import Image, display

# Génère un diagramme du workflow
try:
    display(Image(app.get_graph().draw_mermaid_png()))
except Exception:
    # Fallback : affiche le graphe en ASCII
    print(app.get_graph().draw_ascii())

Exemple de sortie Mermaid :

graph TD
    START --> research
    research --> writing
    writing --> review
    review -->|quality < 0.7| writing
    review -->|quality >= 0.7| finalize
    finalize --> END

Pattern : Agent humain-dans-la-boucle

Demander validation humaine avant certaines actions.

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.prebuilt import ToolNode

# État
class HumanInLoopState(TypedDict):
    messages: Annotated[list, add_messages]
    pending_approval: bool

# Nœud de demande d'approbation
def approval_node(state: HumanInLoopState) -> HumanInLoopState:
    """Demande validation humaine"""
    last_message = state["messages"][-1]

    print("\n⚠️ ACTION NÉCESSITANT APPROBATION")
    print(f"Action : {last_message.tool_calls[0]['name']}")
    print(f"Arguments : {last_message.tool_calls[0]['args']}")

    approval = input("\nApprouver ? (o/n) : ")

    if approval.lower() != 'o':
        return {
            **state,
            "messages": [AIMessage(content="Action annulée par l'utilisateur")],
            "pending_approval": False
        }

    return {**state, "pending_approval": False}

# Routage
def needs_approval(state: HumanInLoopState) -> Literal["approve", "execute"]:
    """Vérifie si l'action nécessite approbation"""
    last_message = state["messages"][-1]

    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        tool_name = last_message.tool_calls[0]["name"]

        # Actions sensibles
        if tool_name in ["delete_file", "send_email", "make_payment"]:
            return "approve"

    return "execute"

# Graphe
workflow = StateGraph(HumanInLoopState)
workflow.add_node("agent", agent_node)
workflow.add_node("approve", approval_node)
workflow.add_node("tools", ToolNode(tools))

workflow.set_entry_point("agent")

workflow.add_conditional_edges(
    "agent",
    needs_approval,
    {
        "approve": "approve",
        "execute": "tools"
    }
)

workflow.add_edge("approve", "tools")
workflow.add_edge("tools", "agent")

memory = SqliteSaver.from_conn_string(":memory:")
app = workflow.compile(checkpointer=memory)

Pattern : Multi-agents avec spécialisation

Plusieurs agents collaborent, chacun avec son expertise.

from typing import Literal

class MultiAgentState(TypedDict):
    messages: Annotated[list, add_messages]
    current_agent: str
    research_data: str
    analysis_results: str
    final_report: str

# Agents spécialisés
researcher_llm = ChatOpenAI(model="gpt-4", temperature=0.3)
analyst_llm = ChatOpenAI(model="gpt-4", temperature=0)
writer_llm = ChatOpenAI(model="gpt-4", temperature=0.7)

def researcher_agent(state: MultiAgentState) -> MultiAgentState:
    """Agent de recherche"""
    print("\n🔬 Agent Recherche au travail")

    prompt = f"""Tu es un chercheur expert. Trouve des informations factuelles sur :
    {state['messages'][-1].content}

    Fournis des sources et des données précises."""

    response = researcher_llm.invoke(prompt)

    return {
        **state,
        "research_data": response.content,
        "current_agent": "analyst"
    }

def analyst_agent(state: MultiAgentState) -> MultiAgentState:
    """Agent d'analyse"""
    print("\n📊 Agent Analyse au travail")

    prompt = f"""Tu es un analyste expert. Analyse ces données et tire des conclusions :

    Données de recherche :
    {state['research_data']}

    Fournis une analyse structurée avec insights clés."""

    response = analyst_llm.invoke(prompt)

    return {
        **state,
        "analysis_results": response.content,
        "current_agent": "writer"
    }

def writer_agent(state: MultiAgentState) -> MultiAgentState:
    """Agent de rédaction"""
    print("\n✍️ Agent Rédaction au travail")

    prompt = f"""Tu es un rédacteur expert. Crée un rapport clair et engageant basé sur :

    Recherche :
    {state['research_data']}

    Analyse :
    {state['analysis_results']}

    Rédige un rapport structuré et accessible."""

    response = writer_llm.invoke(prompt)

    return {
        **state,
        "final_report": response.content,
        "current_agent": "done"
    }

# Routeur
def route_to_agent(state: MultiAgentState) -> str:
    """Dirige vers le bon agent"""
    return state.get("current_agent", "researcher")

# Graphe
workflow = StateGraph(MultiAgentState)

workflow.add_node("researcher", researcher_agent)
workflow.add_node("analyst", analyst_agent)
workflow.add_node("writer", writer_agent)

workflow.set_entry_point("researcher")

workflow.add_conditional_edges(
    "researcher",
    route_to_agent,
    {"analyst": "analyst"}
)

workflow.add_conditional_edges(
    "analyst",
    route_to_agent,
    {"writer": "writer"}
)

workflow.add_conditional_edges(
    "writer",
    route_to_agent,
    {"done": END}
)

app = workflow.compile()

# Utilisation
def multi_agent_research(topic: str):
    result = app.invoke({
        "messages": [HumanMessage(content=topic)],
        "current_agent": "researcher",
        "research_data": "",
        "analysis_results": "",
        "final_report": ""
    })

    return result["final_report"]

# Test
report = multi_agent_research("Impact de l'IA sur le marché du travail en 2025")
print("\n" + "="*60)
print("RAPPORT MULTI-AGENTS")
print("="*60)
print(report)

Comparaison : ReAct Agent vs LangGraph

AspectAgent ReActLangGraph
StructureLinéaire (boucle simple)Graphe (cycles complexes)
ContrôleAutomatique par le LLMExplicite par le développeur
CyclesLimitésNatifs et contrôlables
ÉtatImplicite (historique)Explicite et typé
RévisionDifficileFacile (boucles)
Multi-agentsComplexeNaturel
DebuggingTraces texteGraphe visuel + checkpoints
PersistanceManuelleIntégrée (checkpoints)
Cas d’usageTâches simples, explorationWorkflows complexes, production

Quand utiliser quoi ? :

  • ReAct : Prototypes, tâches exploratoires simples
  • LangGraph : Production, workflows complexes, multi-agents
💡 🔄 Les cycles permettent retry logic, amélioration itérative, et agents autonomes qui s’auto-corrigent jusqu’à atteindre l’objectif.

Comparaison avec d’autres frameworks

LangGraph vs Alternatives

Bien que LangGraph soit puissant, d’autres frameworks existent pour construire des agents IA. Voici une comparaison :

FrameworkFocusStrengthsWeaknesses
LangGraphState graphs, agentsCycles, conditions, checkpointing, intégration LangChainCourbe d’apprentissage
AutoGen (Microsoft)Multi-agent conversationsAgent-to-agent chat naturel, rôles prédéfinisMoins de contrôle fin du workflow
CrewAIRole-based agentsSimple à utiliser, rôles prédéfinis (researcher, writer)Moins flexible, limité aux patterns prédéfinis
HaystackSearch + LLMsExcellent pour RAG et recherche, pipelinesMoins adapté aux agents autonomes complexes

Choisir LangGraph si :

  • ✅ Vous avez besoin de logique de contrôle complexe (cycles, conditions multiples)
  • ✅ Human-in-the-loop est essentiel à votre workflow
  • ✅ Vous voulez persister l’état et reprendre après interruption
  • ✅ Vous utilisez déjà l’écosystème LangChain
  • ✅ Vous avez besoin d’un contrôle total sur le flux d’exécution

Choisir AutoGen si :

  • Vous voulez des agents qui conversent naturellement entre eux
  • Vous préférez une approche “conversationnelle” plutôt que “graphe”
  • Vous avez des rôles d’agents bien définis (manager, developer, tester)

Choisir CrewAI si :

  • Vous débutez et voulez quelque chose de simple
  • Vos agents ont des rôles classiques (researcher, analyst, writer)
  • Vous ne voulez pas gérer les détails d’orchestration

Choisir Haystack si :

  • Votre focus principal est RAG et recherche documentaire
  • Vous voulez des pipelines de traitement de documents

Bonnes pratiques

Typage strict de l’état

from typing import TypedDict, List, Optional

class StrictState(TypedDict, total=False):
    # Champs requis
    query: str
    messages: List[dict]

    # Champs optionnels
    research_data: Optional[str]
    iteration: int

Limites de sécurité

def safe_node(state: AgentState) -> AgentState:
    """Nœud avec limites"""
    MAX_ITERATIONS = 10

    if state.get("iteration_count", 0) >= MAX_ITERATIONS:
        raise Exception("Limite d'itérations atteinte")

    # Logique du nœud
    result = process(state)

    return {
        **state,
        **result,
        "iteration_count": state.get("iteration_count", 0) + 1
    }

Gestion d’erreurs

def robust_node(state: AgentState) -> AgentState:
    """Nœud avec gestion d'erreurs"""
    try:
        result = risky_operation(state)
        return {**state, "result": result, "error": None}
    except Exception as e:
        print(f"❌ Erreur dans le nœud : {e}")
        return {**state, "error": str(e), "should_retry": True}

# Routage avec gestion d'erreur
def error_router(state: AgentState) -> Literal["retry", "fail", "continue"]:
    if state.get("error"):
        if state.get("retry_count", 0) < 3:
            return "retry"
        return "fail"
    return "continue"

Logging et monitoring

from datetime import datetime

class LoggingState(TypedDict):
    messages: Annotated[list, add_messages]
    logs: list

def logged_node(state: LoggingState) -> LoggingState:
    """Nœud avec logging"""
    start = datetime.now()

    # Logique
    result = process(state)

    duration = (datetime.now() - start).total_seconds()

    log_entry = {
        "node": "logged_node",
        "timestamp": start.isoformat(),
        "duration_s": duration,
        "success": True
    }

    return {
        **state,
        **result,
        "logs": state.get("logs", []) + [log_entry]
    }

Exercices pratiques

Mettre en pratique est essentiel pour maîtriser LangGraph. Voici trois exercices progressifs :

Exercice 1 : Workflow de Génération d’Article (Intermédiaire)

Objectif : Créer un pipeline complet Recherche → Draft → Review humaine → Publish

Spécifications :

  1. Node recherche : Utiliser DuckDuckGoSearchRun pour chercher des informations
  2. Node rédaction : GPT-4 génère un article basé sur la recherche
  3. Node review humaine : Interruption avec interrupt_before pour validation
  4. Conditional edge : Si approuvé → publish, si rejeté → amélioration → retour review

Extension avancée :

  • Ajouter un scorer de qualité automatique (évalue clarté, pertinence, structure)
  • Implémenter cycle d’amélioration itérative (max 3 tentatives)
  • Ajouter métriques (temps, coût, nombre d’itérations)

Solution squelette :

class ArticleState(TypedDict):
    topic: str
    research: str
    draft: str
    score: float
    status: str  # "pending", "approved", "rejected"
    iteration: int

def research_node(state): ...
def draft_node(state): ...
def score_node(state): ...
def human_review_node(state): ...
def publish_node(state): ...

def should_continue(state):
    if state["status"] == "approved": return "publish"
    if state["iteration"] >= 3: return "publish"
    if state["score"] < 0.7: return "improve"
    return "review"

Exercice 2 : Agent Customer Support (Avancé)

Objectif : Agent autonome avec tools pour support client

Architecture cible :

     START
       ↓
   [agent] ←──┐
      ↓       │
  tool call?  │
   Yes  No    │
    ↓    ↓   │
[tools] [eval]
    ↓    ↓
    └────┘
       ↓
   resolved?
   Yes   No
    ↓     ↓
  [close][escalate]
    ↓     ↓
   END   END

Tools à implémenter :

  1. search_faq(query: str) -> str : Recherche dans base FAQ
  2. create_ticket(issue: str, priority: str) -> str : Crée ticket support
  3. check_order_status(order_id: str) -> dict : Vérifie statut commande

Challenges :

  • L’agent doit décider quand utiliser quel tool
  • Évaluer si la question est résolue après chaque réponse
  • Escalader automatiquement si 3 tentatives échouent
  • Persister la conversation avec checkpointing

Exercice 3 : Multi-Agent Debate System (Expert)

Objectif : Système de débat avec 2 agents opposés + 1 juge

Agents :

  • Agent Pro : Argumente POUR une proposition
  • Agent Con : Argumente CONTRE la proposition
  • Judge : Évalue qualité des arguments, déclare un gagnant

Graph :

    START
      ↓
  [initialize]
      ↓
    Round 1-3:
   [pro] → [con]
      ↓
  [evaluate_round]
      ↓
    continue?
    Yes   No
     ↓     ↓
   loop  [judge]
          ↓
        [end]

Features avancées :

  • Chaque agent voit les arguments précédents de l’adversaire
  • Judge score chaque round (clarté, logique, preuves)
  • Maximum 3 rounds de débat
  • State accumule historique complet du débat
  • Output final : gagnant + justification

State structure :

class DebateState(TypedDict):
    topic: str
    rounds: List[dict]  # [{round: 1, pro: "...", con: "...", score: ...}]
    current_round: int
    max_rounds: int
    winner: str
    justification: str

Checklist Production

Avant de déployer votre application LangGraph en production, vérifiez ces points essentiels :

Architecture & Design

  • Graph testé end-to-end avec données réelles
  • Cycles terminés correctement : Pas de boucles infinies possibles
  • Conditional edges exhaustifs : Tous les cas de figure couverts
  • State schema bien défini : TypedDict avec types explicites
  • Nodes idempotents : Réexécuter un node produit même résultat
  • Entry point clair : Point de départ évident et documenté

Persistance & Checkpointing

  • Checkpointer configuré : PostgreSQL (production) ou SQLite (dev)
  • Thread IDs uniques : Par session utilisateur ou conversation
  • Cleanup automatique : Suppression des vieux checkpoints (>30 jours)
  • Backup strategy : Sauvegarde de la base checkpoints
  • Migration plan : Stratégie si schema state change

Résilience & Error Handling

  • Try/catch dans chaque node : Gestion explicite des erreurs
  • Timeouts configurés : Max 30s par appel LLM, 2min par node
  • Retry logic : 3 tentatives avec backoff exponentiel pour LLM calls
  • Fallbacks définis : Comportement par défaut si erreur critique
  • Circuit breaker : Désactiver temporairement si trop d’erreurs
  • Dead letter queue : Logger requêtes qui échouent définitivement

Monitoring & Observability

  • Logging structuré : JSON logs avec node_name, duration, tokens
  • Métriques trackées :
    • Latence par node et workflow complet
    • Coûts LLM (tokens in/out, prix par requête)
    • Taux d’erreur par node
    • Distribution des conditional paths
  • Alertes configurées :
    • Latence >10s
    • Taux d’erreur >5%
    • Coûts >seuil quotidien
  • Tracing distribué : Correlation IDs pour suivre requête E2E
  • Dashboard : Grafana/Datadog avec métriques clés

Sécurité & Compliance

  • Validation inputs : Sanitize user inputs avant LLM
  • Secrets management : API keys dans variables d’environnement
  • Rate limiting : Par utilisateur et global
  • PII handling : Masquage données sensibles dans logs
  • RGPD compliance : Droit à l’effacement des checkpoints
  • Audit logs : Traçabilité actions sensibles

Performance & Scalabilité

  • Caching intelligent : Résultats similaires mis en cache
  • Streaming activé : Pour UX temps réel
  • Parallélisation : Nodes indépendants en parallèle si possible
  • Load testing : Testé à 10x charge attendue
  • Auto-scaling : Kubernetes HPA ou équivalent

Documentation

  • Graph visualisé : Mermaid diagram à jour dans docs
  • State schema documenté : Chaque champ expliqué
  • Runbook : Procédures debugging et recovery
  • Exemples d’utilisation : Code samples pour cas courants

Testing

  • Tests unitaires : Chaque node testé isolément
  • Tests d’intégration : Workflows complets end-to-end
  • Tests de régression : Suite de tests sur cas connus
  • Mocks LLM : Tests sans appels réels pour CI/CD
  • Chaos engineering : Injection erreurs pour tester resilience
⚠️ Warning
⚠️ Ne déployez JAMAIS sans checkpointing configuré en production. Une interruption sans sauvegarde = perte totale de state.

Déploiement en production

API REST avec FastAPI

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langgraph.checkpoint.sqlite import SqliteSaver

app_api = FastAPI()

# Checkpoint persistant
memory = SqliteSaver.from_conn_string("./checkpoints.db")
graph_app = workflow.compile(checkpointer=memory)

class QueryRequest(BaseModel):
    query: str
    thread_id: str

class QueryResponse(BaseModel):
    result: str
    thread_id: str

@app_api.post("/query", response_model=QueryResponse)
async def handle_query(request: QueryRequest):
    """Endpoint pour exécuter l'agent"""
    try:
        config = {"configurable": {"thread_id": request.thread_id}}

        result = graph_app.invoke(
            {"messages": [HumanMessage(content=request.query)]},
            config=config
        )

        return QueryResponse(
            result=result["messages"][-1].content,
            thread_id=request.thread_id
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# Lancer : uvicorn main:app_api --reload

Avec LangServe

from fastapi import FastAPI
from langserve import add_routes

app = FastAPI()

# Ajouter le graphe comme route
add_routes(
    app,
    graph_app,
    path="/agent",
    enable_feedback_endpoint=True
)

# Auto-disponible :
# POST /agent/invoke
# POST /agent/stream
# POST /agent/batch

Ressources et aller plus loin

Documentation officielle

Exemples avancés

  1. Plan-and-Execute : Agent qui planifie puis exécute
  2. Multi-agent Collaboration : Équipe d’agents spécialisés
  3. Self-RAG : RAG avec auto-évaluation et révision
  4. Code Assistant : Génération et test de code en boucle

Architecture de référence

# Template pour agents production
class ProductionState(TypedDict):
    # Core
    messages: Annotated[list, add_messages]

    # Métadonnées
    session_id: str
    user_id: str
    created_at: str

    # Workflow
    current_step: str
    iteration_count: int

    # Résultats
    final_output: Optional[str]

    # Monitoring
    errors: List[dict]
    logs: List[dict]
    costs_usd: float

# Nœuds avec observabilité complète
# Gestion d'erreurs robuste
# Limites de sécurité
# Checkpoints pour reprise

Conclusion

LangGraph transforme la manière de construire des agents IA en offrant un contrôle total sur des workflows complexes impossibles à réaliser avec des chains classiques.

Points clés à retenir

1. Graphes > Chains pour logique complexe

  • Les chains LangChain sont linéaires et limitées
  • LangGraph permet cycles, conditions, et orchestration avancée
  • Contrôle explicite du flux d’exécution vs décisions LLM imprévisibles

2. State explicite = Debugging facilité

  • Le state typé rend le workflow transparent
  • Inspection à chaque node pour comprendre ce qui se passe
  • Checkpointing permet reprendre exactement où on s’est arrêté

3. Cycles et conditions = Agents autonomes

  • Retry logic jusqu’à succès
  • Amélioration itérative automatique
  • Validation multi-passes pour qualité optimale

4. Human-in-the-loop natif

  • Interruption et reprise avec checkpointing
  • Validation humaine dans workflows critiques
  • Collaboration homme-machine fluide

5. Production-ready dès le départ

  • Persistance intégrée avec PostgreSQL/SQLite
  • Streaming pour UX temps réel
  • Monitoring et observabilité natives

Quand utiliser LangGraph

✅ Utilisez LangGraph si :

  • Workflows avec logique de contrôle (if/else, loops)
  • Agents autonomes multi-étapes
  • Human validation requise dans le processus
  • Retry/amélioration itérative nécessaire
  • Besoin d’inspecter et debugger le state
  • Production avec haute disponibilité

❌ Restez sur Chains si :

  • Workflows simples et linéaires
  • RAG basique sans logique complexe
  • Prototypes rapides sans persistence
  • Pas de cycles ni conditions multiples

Cas d’usage transformés par LangGraph

  • 🔄 Génération de contenu avec révision automatique jusqu’à qualité suffisante
  • 🤝 Systèmes multi-agents collaboratifs (researcher + writer + editor)
  • 📊 Workflows d’analyse de données en plusieurs passes avec validation
  • 💬 Chatbots conversationnels avec mémoire longue et persistence
  • 🔍 Agents de recherche avec validation qualité et amélioration itérative
  • ⚖️ Workflows d’approval avec human-in-the-loop pour décisions critiques

Parcours d’apprentissage recommandé

1. Débutants (Semaine 1-2) :

  • Commencez par un chatbot simple avec 2-3 nodes
  • Ajoutez progressivement des conditional edges
  • Implémentez votre premier cycle (retry logic)
  • Testez le streaming des résultats

2. Intermédiaires (Semaine 3-4) :

  • Créez un workflow avec révision (exemple recherche→rédaction→review)
  • Ajoutez checkpointing pour persistence
  • Implémentez human-in-the-loop avec interruptions
  • Mesurez coûts et latence

3. Avancés (Semaine 5-8) :

  • Système multi-agents avec spécialisation
  • Orchestration complexe avec supervisor pattern
  • Implémentez retry logic et error handling robuste
  • Ajoutez monitoring complet (métriques, logs, alertes)

4. Production (Semaine 9+) :

  • API REST avec FastAPI ou LangServe
  • PostgreSQL checkpointing en production
  • CI/CD avec tests automatisés
  • Observabilité complète (Grafana, DataDog)
💡 🎯 Démarrez aujourd’hui : créez un simple graph avec 2 nodes et conditional edge. Vous découvrirez rapidement la puissance de cette approche et ne reviendrez plus aux chains linéaires.

Pour aller plus loin :

LangGraph représente l’avenir des agents IA : contrôlables, robustes, et prêts pour la production.

← Retour à la Formation LangChain | Module suivant : LangSmith - Monitoring →