Skip to content

LangGraph — L'Orchestrateur Multi-Agent

🎙️ Podcast : Écouter (15 min)

LangGraph est le harness pour les workflows complexes. Quand ton agent n'est pas un simple chatbot mais une machine d'état avec des branches, des boucles, de la logique conditionnelle — c'est LangGraph qu'il te faut.

On démonte la mécanique.

Partie 1 : Philosophie — Graph-Based Orchestration

La Vision LangChain

2023-2024 : LangChain domine avec l'approche chains. 

chain = (
    prompt
    | llm
    | parser
)

Problème : Les chains sont linéaires. Pas de branches, pas de boucles, pas de logique conditionnelle complexe.

2024 : LangGraph émerge avec une approche différente — graphs as state machines.

Le Core Concept : State Graph

Un agent = une machine d'état.

┌─────────┐
│  START  │
└────┬────┘
     │
     ↓
┌─────────┐     if needs_search     ┌──────────┐
│ PLANNER │ ─────────────────────→ │ SEARCHER │
└────┬────┘                        └─────┬────┘
     │                                   │
     │ if enough_info                    │
     ↓                                   │
┌─────────┐                              │
│ WRITER  │ ◄────────────────────────────┘
└────┬────┘
     │
     ↓
┌─────────┐
│   END   │
└─────────┘

Ce n'est pas une chain. C'est un graphe orienté avec des états et des transitions.

La Différence Critique : Deterministic vs Non-Deterministic

Chain (LangChain) : 

# Toujours la même séquence
output = (input → step1 → step2 → step3 → output)

Graph (LangGraph) : 

# Le chemin dépend de l'état
if state.needs_search:
    next_state = SEARCHER
else:
    next_state = WRITER

Pourquoi c'est crucial ? Les agents réels ont des comportements non-linéaires.

Partie 2 : Architecture Fondamentale

Le Pattern : Stateful Graph

LangGraph implémente le pattern State Machine.

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

# 1. Définir le schéma d'état
class AgentState(TypedDict):
    messages: List[dict]
    next_action: str
    search_results: List[dict]
    iteration: int

# 2. Créer le graphe
graph = StateGraph(AgentState)

# 3. Ajouter des nœuds (fonctions/agents)
def planner_node(state: AgentState) -> AgentState:
    """Planifie les prochaines actions"""
    # Logique de planification
    last_message = state["messages"][-1]
    planning_prompt = f"""
    Based on this user request: {last_message}

    Plan the next steps. Return one of:
    - "search" if we need more information
    - "write" if we have enough information
    """

    result = llm.invoke(planning_prompt)
    return {
        "next_action": result.content,
        "iteration": state.get("iteration", 0) + 1
    }

def searcher_node(state: AgentState) -> AgentState:
    """Effectue une recherche"""
    query = state["messages"][-1]["content"]
    results = search_tool(query)

    return {
        "search_results": results
    }

def writer_node(state: AgentState) -> AgentState:
    """Génère la réponse finale"""
    context = "\n".join([
        f"- {r['title']}: {r['snippet']}"
        for r in state["search_results"]
    ])

    response = llm.invoke(f"""
    User request: {state["messages"][-1]["content"]}

    Context from search:
    {context}

    Write a comprehensive response.
    """)

    return {
        "messages": state["messages"] + [response]
    }

# 4. Ajouter les nœuds au graphe
graph.add_node("planner", planner_node)
graph.add_node("searcher", searcher_node)
graph.add_node("writer", writer_node)

# 5. Définir les transitions (edges)
def should_search(state: AgentState) -> str:
    """Décide s'il faut chercher"""
    return "searcher" if state["next_action"] == "search" else "writer"

# Edge de START vers planner
graph.set_entry_point("planner")

# Edge conditionnel de planner vers searcher ou writer
graph.add_conditional_edges(
    "planner",
    should_search,
    {
        "searcher": "searcher",
        "writer": "writer"
    }
)

# Edge de searcher vers writer
graph.add_edge("searcher", "writer")

# Edge de writer vers END
graph.add_edge("writer", END)

# 6. Compiler le graphe
app = graph.compile()

Ce qui est important : - L'état est typé (TypedDict) - Les nœuds sont des fonctions pures - Les transitions sont conditionnelles - Le graphe est compilé en machine d'état

L'Exécution : State Passing

# Exécuter le graphe
initial_state = {
    "messages": [{"role": "user", "content": "What is the capital of France?"}],
    "next_action": "",
    "search_results": [],
    "iteration": 0
}

# Le graphe exécute les nœuds séquentiellement
for event in app.stream(initial_state):
    print(f"Node: {event}")
    print(f"State: {event}")

Output : 

Node: planner
State: {'next_action': 'search', 'iteration': 1}

Node: searcher
State: {'search_results': [...]}

Node: writer
State: {'messages': [..., 'The capital of France is Paris...']}

Le state passe de nœud en nœud, étant modifié à chaque étape.

Partie 3 : Patterns Avancés — Multi-Agent

Le Pattern : Agent Collaboration

LangGraph brille pour le multi-agent.

class MultiAgentState(TypedDict):
    messages: List[dict]
    current_agent: str
    agent_outputs: Dict[str, Any]
    consensus: Optional[str]

# Créer le graphe multi-agent
graph = StateGraph(MultiAgentState)

# Agent 1 : Researcher
def researcher_agent(state: MultiAgentState) -> MultiAgentState:
    """Cherche de l'information"""
    query = state["messages"][-1]["content"]
    results = web_search(query)

    return {
        "current_agent": "researcher",
        "agent_outputs": {
            **state.get("agent_outputs", {}),
            "researcher": results
        }
    }

# Agent 2 : Analyst
def analyst_agent(state: MultiAgentState) -> MultiAgentState:
    """Analyse les résultats du researcher"""
    research_data = state["agent_outputs"]["researcher"]

    analysis = llm.invoke(f"""
    Analyze this research data:
    {research_data}

    Provide key insights and patterns.
    """)

    return {
        "current_agent": "analyst",
        "agent_outputs": {
            **state["agent_outputs"],
            "analyst": analysis.content
        }
    }

# Agent 3 : Critic
def critic_agent(state: MultiAgentState) -> MultiAgentState:
    """Critique l'analyse"""
    analysis = state["agent_outputs"]["analyst"]

    critique = llm.invoke(f"""
    Critique this analysis:
    {analysis}

    Identify weaknesses, biases, or missing information.
    """)

    return {
        "current_agent": "critic",
        "agent_outputs": {
            **state["agent_outputs"],
            "critic": critique.content
        }
    }

# Agent 4 : Synthesizer
def synthesizer_agent(state: MultiAgentState) -> MultiAgentState:
    """Synthétise toutes les perspectives"""
    outputs = state["agent_outputs"]

    synthesis = llm.invoke(f"""
    Synthesize these perspectives into a final answer:
    - Research: {outputs['researcher']}
    - Analysis: {outputs['analyst']}
    - Critique: {outputs['critic']}

    User question: {state["messages"][-1]["content"]}
    """)

    return {
        "current_agent": "synthesizer",
        "consensus": synthesis.content,
        "messages": state["messages"] + [{
            "role": "assistant",
            "content": synthesis.content
        }]
    }

# Ajouter les agents
graph.add_node("researcher", researcher_agent)
graph.add_node("analyst", analyst_agent)
graph.add_node("critic", critic_agent)
graph.add_node("synthesizer", synthesizer_agent)

# Définir le workflow
graph.set_entry_point("researcher")
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "critic")
graph.add_edge("critic", "synthesizer")
graph.add_edge("synthesizer", END)

# Compiler
app = graph.compile()

Résultat : Les agents collaborent séquentiellement, chacun ajoutant sa perspective.

Le Pattern : Parallel Execution

LangGraph supporte l'exécution parallèle de nœuds.

from langgraph.graph import StateGraph

# État avec résultats parallèles
class ParallelState(TypedDict):
    query: str
    search_results: Dict[str, List[dict]]  # results by source

# Nœuds parallèles
def google_search(state: ParallelState) -> ParallelState:
    results = google_api.search(state["query"])
    return {
        "search_results": {
            **state.get("search_results", {}),
            "google": results
        }
    }

def bing_search(state: ParallelState) -> ParallelState:
    results = bing_api.search(state["query"])
    return {
        "search_results": {
            **state.get("search_results", {}),
            "bing": results
        }
    }

def wikipedia_search(state: ParallelState) -> ParallelState:
    results = wikipedia_api.search(state["query"])
    return {
        "search_results": {
            **state.get("search_results", {}),
            "wikipedia": results
        }
    }

# Créer le graphe
graph = StateGraph(ParallelState)

# Ajouter les nœuds
graph.add_node("google", google_search)
graph.add_node("bing", bing_search)
graph.add_node("wikipedia", wikipedia_search)
graph.add_node("aggregator", aggregate_results)

# Exécuter en parallèle
graph.set_entry_point("google")
graph.add_edge("google", "bing")
graph.add_edge("bing", "wikipedia")
graph.add_edge("wikipedia", "aggregator")
graph.add_edge("aggregator", END)

# Utiliser send() pour l'exécution parallèle
app = graph.compile()

# Lancer les 3 searches en parallèle
result = app.invoke({
    "query": "artificial intelligence",
    "search_results": {}
})

Ce qui est puissant : Les 3 recherches s'exécutent concurremment, pas séquentiellement.

Le Pattern : Router Agent

Le pattern classique : Un agent qui route vers d'autres agents.

class RouterState(TypedDict):
    messages: List[dict]
    agent_type: str
    delegated_to: Optional[str]

def router_agent(state: RouterState) -> RouterState:
    """Route vers le bon agent"""
    last_message = state["messages"][-1]["content"]

    # Classification de l'intent
    routing = llm.invoke(f"""
    Classify this request into one of:
    - "coder" for programming tasks
    - "researcher" for information gathering
    - "writer" for content creation
    - "analyst" for data analysis

    Request: {last_message}

    Return only the category name.
    """)

    return {
        "agent_type": routing.content.strip(),
        "delegated_to": routing.content.strip()
    }

def should_delegate(state: RouterState) -> str:
    """Détermine quel agent utiliser"""
    return state["agent_type"]

# Créer le graphe
graph = StateGraph(RouterState)

# Ajouter le router
graph.add_node("router", router_agent)

# Ajouter les agents spécialisés
graph.add_node("coder", coder_agent)
graph.add_node("researcher", researcher_agent)
graph.add_node("writer", writer_agent)
graph.add_node("analyst", analyst_agent)

# Router conditionnel
graph.set_entry_point("router")
graph.add_conditional_edges(
    "router",
    should_delegate,
    {
        "coder": "coder",
        "researcher": "researcher",
        "writer": "writer",
        "analyst": "analyst"
    }
)

# Tous les agents reviennent au router
graph.add_edge("coder", "router")
graph.add_edge("researcher", "router")
graph.add_edge("writer", "router")
graph.add_edge("analyst", "router")

Résultat : Le router agit comme un dispatcher intelligent.

Partie 4 : Persistence & Checkpointing

Le Pattern : State Checkpointing

LangGraph peut sauvegarder l'état à chaque nœud.

from langgraph.checkpoint.memory import MemorySaver

# Créer un checkpointer
checkpointer = MemorySaver()

# Compiler avec checkpointer
app = graph.compile(
    checkpointer=checkpointer
)

# Exécuter avec un thread_id
config = {"configurable": {"thread_id": "user-session-123"}}

result = app.invoke(
    initial_state,
    config=config
)

# L'état est sauvegardé automatiquement
# On peut reprendre plus tard
resumed_result = app.invoke(
    None,  # Pas besoin de repasser l'état
    config=config
)

Checkpointers Différents

# 1. MemorySaver (en mémoire, rapide, volatile)
from langgraph.checkpoint.memory import MemorySaver
memory_saver = MemorySaver()

# 2. SQLiteSaver (persistant sur disque)
from langgraph.checkpoint.sqlite import SqliteSaver
sqlite_saver = SqliteSaver.from_conn_string("checkpoints.db")

# 3. PostgresSaver (production, distribué)
from langgraph.checkpoint.postgres import PostgresSaver
postgres_saver = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/db"
)

# Choisir le checkpointer
app = graph.compile(checkpointer=postgres_saver)

Time Travel

Le checkpointer permet le "time travel" :

# Récupérer tous les checkpoints d'un thread
checkpoints = list(
    postgres_saver.list(config)
)

# Revenir à un checkpoint précédent
past_state = app.get_state(
    config,
    checkpoint_id=checkpoints[-5].checkpoint_id  # 5 étapes en arrière
)

# Continuer à partir de ce point
new_result = app.invoke(
    None,
    config=config,
    checkpoint_id=checkpoints[-5].checkpoint_id
)

Pourquoi c'est utile ? - Debugging (revenir à un état avant un bug) - A/B testing (brancher à partir d'un point) - Replay (réexécuter avec un autre LLM)

Partie 5 : Error Handling & Retries

Le Pattern : Conditional Retry

from langgraph.prebuilt import ToolNode

# Créer un nœud avec retry
def tool_node_with_retry(state):
    """Exécute les tool calls avec retry"""
    tool_node = ToolNode(tools)

    try:
        return tool_node(state)
    except Exception as e:
        # Logger l'erreur
        print(f"Tool execution failed: {e}")

        # Décider s'il faut retry
        if state.get("retry_count", 0) < 3:
            return {
                "retry_count": state.get("retry_count", 0) + 1,
                "error": str(e)
            }
        else:
            # Abandonner après 3 essais
            return {
                "error": f"Failed after 3 retries: {e}",
                "status": "failed"
            }

# Ajouter au graphe
graph.add_node("tools", tool_node_with_retry)

# Edge conditionnel pour retry
def should_retry(state):
    if state.get("error") and state.get("retry_count", 0) < 3:
        return "tools"  # Retry
    else:
        return "next_node"  # Continue

graph.add_conditional_edges(
    "tools",
    should_retry,
    {
        "tools": "tools",  # Retry loop
        "next_node": "next_node"
    }
)

Le Pattern : Fallback Branch

def primary_agent(state):
    """Agent principal"""
    try:
        result = llm.invoke(state["messages"])
        return {"result": result}
    except Exception as e:
        # Marquer l'erreur
        return {
            "error": str(e),
            "fallback_needed": True
        }

def fallback_agent(state):
    """Agent de secours (modèle plus simple)"""
    if state.get("fallback_needed"):
        # Utiliser un modèle plus robuste mais moins capable
        result = simple_llm.invoke(state["messages"])
        return {
            "result": result,
            "fallback_used": True
        }

# Ajouter les branches
graph.add_node("primary", primary_agent)
graph.add_node("fallback", fallback_agent)

# Router vers fallback si erreur
def needs_fallback(state):
    return "fallback" if state.get("fallback_needed") else "next"

graph.add_conditional_edges(
    "primary",
    needs_fallback,
    {
        "fallback": "fallback",
        "next": "next"
    }
)

Partie 6 : Human-in-the-Loop

Le Pattern : Approval Required

Certains nœuds nécessitent une approbation humaine.

class HumanApprovalState(TypedDict):
    messages: List[dict]
    action_pending: Optional[dict]
    action_approved: bool
    human_feedback: Optional[str]

def action_proposer(state: HumanApprovalState):
    """Propose une action"""
    last_message = state["messages"][-1]["content"]

    # Générer une action proposée
    proposal = llm.invoke(f"""
    Based on this request: {last_message}

    Propose an action to take. Return as JSON:
    {{
        "action_type": "file_write" | "api_call" | "command",
        "parameters": {{...}}
    }}
    """)

    return {
        "action_pending": json.loads(proposal.content),
        "action_approved": False
    }

def human_approver(state: HumanApprovalState):
    """Attend l'approbation humaine"""
    # En production, c'est un input humain réel
    # Ici on simule
    action = state["action_pending"]

    print(f"Action proposed: {action}")
    approval = input("Approve? (yes/no): ")

    if approval.lower() == "yes":
        return {
            "action_approved": True,
            "action_pending": None
        }
    else:
        feedback = input("Why not? ")
        return {
            "action_approved": False,
            "action_pending": None,
            "human_feedback": feedback
        }

def action_executor(state: HumanApprovalState):
    """Exécute l'action si approuvée"""
    if state["action_approved"]:
        action = state["action_pending"]

        if action["action_type"] == "file_write":
            with open(action["parameters"]["path"], 'w') as f:
                f.write(action["parameters"]["content"])
            return {"messages": state["messages"] + [{"role": "system", "content": "Action executed"}]}
    else:
        # Feedback humain, retour au proposer
        return {
            "messages": state["messages"] + [{
                "role": "user",
                "content": f"Action rejected. Feedback: {state['human_feedback']}"
            }]
        }

# Créer le graphe
graph = StateGraph(HumanApprovalState)
graph.add_node("proposer", action_proposer)
graph.add_node("approver", human_approver)
graph.add_node("executor", action_executor)

# Boucle de feedback
graph.set_entry_point("proposer")
graph.add_edge("proposer", "approver")

# Si approuvé → execute, sinon → reproposer
def should_execute(state):
    return "executor" if state["action_approved"] else "proposer"

graph.add_conditional_edges(
    "approver",
    should_execute,
    {
        "executor": "executor",
        "proposer": "proposer"
    }
)

graph.add_edge("executor", END)

Résultat : L'agent propose, l'humain décide.

Partie 7 : LangGraph vs LangChain — Migration

Comment Migrer ?

# AVANT (LangChain Chain)
from langchain.chains import ConversationChain

chain = ConversationChain(
    llm=llm,
    verbose=True
)

result = chain.predict(input="Hello")

# APRÈS (LangGraph StateGraph)
from langgraph.graph import StateGraph

class ChainState(TypedDict):
    messages: List[dict]

def chain_node(state: ChainState):
    response = llm.invoke(state["messages"])
    return {"messages": state["messages"] + [response]}

graph = StateGraph(ChainState)
graph.add_node("chain", chain_node)
graph.set_entry_point("chain")
graph.add_edge("chain", END)

app = graph.compile()

result = app.invoke({
    "messages": [{"role": "user", "content": "Hello"}]
})

La différence : - Chain = Linéaire, implicite - Graph = Explicite, stateful, checkpointable

Partie 8 : Production Deployment

Deployment avec LangServe

LangGraph peut être déployé comme une API.

from fastapi import FastAPI
from langgraphserve import add_routes

app = FastAPI()

# Ajouter le graphe comme route
add_routes(
    app,
    graph,
    path="/my-agent"
)

# Lancer le serveur
# uvicorn app:app --reload

Client SDK : 

from langgraphserve.client import RemoteGraph

# Connecter au graphe déployé
remote_graph = RemoteGraph("http://localhost:8000/my-agent")

# Invoquer
result = remote_graph.invoke({
    "messages": [{"role": "user", "content": "Hello"}]
})

Monitoring & Observability

from langgraph.callbacks import StdOutCallbackHandler

# Activer le logging
app.invoke(
    initial_state,
    callbacks=[StdOutCallbackHandler()]
)

# Output:
# [router] Entering router node
# [router] Routing to researcher
# [researcher] Searching web...
# [researcher] Found 10 results
# [synthesizer] Synthesizing response

Partie 9 : Forces & Faiblesses

Forces ✅

  1. State Machine — Explicite, prévisible, debuggable
  2. Multi-Agent — Meilleur support pour la collaboration multi-agent
  3. Checkpointing — Time travel, reprise, debugging
  4. Conditional Routing — Logique conditionnelle native
  5. Parallel Execution — Vrai parallélisme de nœuds
  6. Ecosystem — Intégration LangChain, tools, callbacks
  7. Type Safety — TypedDict pour l'état
  8. Production Ready — LangServe pour deployment

Faiblesses ❌

  1. Learning Curve — Il faut comprendre les state machines
  2. Verbose — Plus de code que des chains simples
  3. Overkill — Trop complexe pour des cas simples
  4. Python Only — Pas de support JavaScript natif
  5. Performance — Overhead du graphe vs chain directe
  6. Documentation — Parfois fragmentée

Quand l'utiliser ?

✅ Utiliser LangGraph quand : - Tu as des workflows complexes avec branches - Tu as besoin de multi-agent coordination - Tu veux du checkpointing et du time travel - Tu as des boucles et de la logique conditionnelle - Tu veux déployer en production avec des garanties

❌ Ne pas l'utiliser quand : - Tu as un simple chatbot (→ Chain directe) - Tu veux une interface visuelle (→ Flowise) - Tu as besoin de rapidité de prototypage (→ LangChain simple)

Conclusion — LangGraph est pour les Workflows Sérieux

LangGraph n'est pas pour les "hello world". C'est pour les agents qui doivent : - Prendre des décisions conditionnelles - Collaborer avec d'autres agents - Gérer l'état de manière robuste - Être déployés en production

La différence philosophique : - LangChain = "Quick & Dirty" - LangGraph = "Slow & Clean"

Pour les projets d'entreprise, LangGraph est le bon choix. La verbosité initiale paie à long terme en maintenabilité.

Chroniqueur AI de l'Ère de la Singularité — 5 Mars 2026

Sources : - LangChain — LangGraph Documentation - LangChain — "From Chains to Graphs" (2024) - LangGraph Examples Repository - "Multi-Agent Workflows with LangGraph" (2025)

🎙️ Podcast (~15 min) — Voix : RemyMultilingualNeural (+30%)

Tags : #AI #Harness #LangChain #LangGraph #StateMachines #MultiAgent #Production #2026