LangGraph - workflows d'agents basés sur des graphes

La chaîne des leçons précédentes va toujours dans un seul sens : chercheur, analyste, rédacteur, fin. Mais que faire si le rédacteur doit renvoyer le travail au chercheur, ou si un agent doit demander l'accord d'un humain avant d'acheter des billets ? Un vrai sentier à travers la savane a des demi-tours et des bifurcations.

LangGraph est un framework pour construire des applications d'IA complexes et avec état, basées sur des graphes. Il permet de créer des cycles, des bifurcations et des flux de contrôle avancés pour les agents. Vois-le comme une carte des sentiers : les nœuds sont des campements, les arêtes des chemins, et l'état est le sac à dos que tu portes de campement en campement.

Qu'est-ce que LangGraph ?

LangGraph étend les capacités de LangChain avec :

• State Management - la gestion de l'état entre les étapes
• Cycles - la possibilité de créer des boucles (contrairement à un DAG)
• Branching - des bifurcations conditionnelles du flux
• Human-in-the-loop - des points d'arrêt pour l'interaction avec un humain

Le schéma montre un graphe d'agent typique avec des outils, dans lequel le flux peut revenir à l'agent plusieurs fois :

1┌─────────────────────────────────────────────────────────────┐
2│                     LangGraph Flow                          │
3├─────────────────────────────────────────────────────────────┤
4│                                                             │
5│     ┌──────────┐                                            │
6│     │  START   │                                            │
7│     └────┬─────┘                                            │
8│          │                                                   │
9│          ▼                                                   │
10│     ┌──────────┐     ┌──────────────┐                       │
11│     │  Agent   │────▶│   Condition  │                       │
12│     │  Node    │     │   (Router)   │                       │
13│     └──────────┘     └──────┬───────┘                       │
14│          ▲                  │                               │
15│          │        ┌─────────┼─────────┐                     │
16│          │        ▼         ▼         ▼                     │
17│          │   ┌────────┐ ┌────────┐ ┌────────┐              │
18│          │   │ Tool A │ │ Tool B │ │  END   │              │
19│          │   └───┬────┘ └───┬────┘ └────────┘              │
20│          │       │          │                               │
21│          └───────┴──────────┘                               │
22│              (cycle back)                                    │
23│                                                             │
24└─────────────────────────────────────────────────────────────┘

L'agent décide, le routeur choisit le chemin vers un outil ou vers la fin, et après l'utilisation d'un outil, le flux revient à l'agent.

Installation et bases

L'élément le plus important de LangGraph, c'est l'état. On le définit comme un TypedDict, un dictionnaire dont les types des clés sont décrits :

1# pip install langgraph langchain-openai
2
3from langgraph.graph import StateGraph, END
4from langgraph.checkpoint.memory import MemorySaver
5from typing import TypedDict, Annotated, Literal
6from langchain_openai import ChatOpenAI
7from langchain_core.messages import HumanMessage, AIMessage
8import operator
9
10# Définition du State - élément clé de LangGraph
11class AgentState(TypedDict):
12    messages: Annotated[list, operator.add]  # Liste de messages
13    current_step: str
14    iteration_count: int
15
16# Modèle LLM
17llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

Annotated[list, operator.add] indique à LangGraph comment fusionner les mises à jour : les nouveaux messages sont ajoutés à la liste au lieu de la remplacer. Les clés sans annotation sont écrasées. Pour le chat, il existe aussi un réducteur tout prêt, add_messages, dans le module langgraph.graph.message.

Graphe d'agent de base

Construire un graphe suit toujours le même ordre : définition de l'état, StateGraph, nœuds (add_node), arêtes (add_edge) et compilation (compile). On commence par l'état et le nœud de l'agent :

1from langgraph.graph import StateGraph, START, END
2
3# Définition du State
4class State(TypedDict):
5    messages: Annotated[list, operator.add]
6    next_action: str
7
8# Fonctions de nœud
9def agent_node(state: State) -> State:
10    """Nœud agent - prend les décisions."""
11    messages = state["messages"]
12
13    response = llm.invoke(messages)
14
15    return {
16        "messages": [response],
17        "next_action": "tool" if response.tool_calls else "end"
18    }

Un nœud est une fonction ordinaire : elle reçoit l'état et renvoie un dictionnaire avec les clés qu'elle veut modifier. Pour que le modèle renvoie des tool_calls, il faut lui attacher les outils avec llm.bind_tools(tools), sinon la liste sera toujours vide.

Le deuxième nœud exécute les outils, et le routeur choisit l'étape suivante :

1def tool_node(state: State) -> State:
2    """Nœud outil - exécute les actions."""
3    last_message = state["messages"][-1]
4
5    # Exécute les appels d'outils
6    tool_results = []
7    for tool_call in last_message.tool_calls:
8        result = execute_tool(tool_call)
9        tool_results.append(result)
10
11    return {"messages": tool_results, "next_action": "agent"}
12
13def router(state: State) -> Literal["tool", "end"]:
14    """Router - décide de l'étape suivante."""
15    return state["next_action"]

execute_tool est l'emplacement pour ta propre logique. En pratique, tu remplaceras ce nœud par le ToolNode tout prêt de langgraph.prebuilt.

Maintenant, on relie tout dans un graphe :

1# Construis le graphe
2workflow = StateGraph(State)
3
4# Ajoute les nœuds
5workflow.add_node("agent", agent_node)
6workflow.add_node("tool", tool_node)
7
8# Ajoute les arêtes
9workflow.add_edge(START, "agent")
10workflow.add_conditional_edges(
11    "agent",
12    router,
13    {
14        "tool": "tool",
15        "end": END
16    }
17)
18workflow.add_edge("tool", "agent")  # Boucle de retour vers l'agent
19
20# Compile le graphe
21app = workflow.compile()
22
23# Lance
24result = app.invoke({
25    "messages": [HumanMessage(content="What is the weather in the Serengeti?")],
26    "next_action": ""
27})

add_conditional_edges demande au routeur un nom de chemin et le traduit en nœud grâce au dictionnaire. L'arête de "tool" vers "agent" crée le cycle.

Agent ReAct avec LangGraph

Ce modèle est si populaire qu'il existe une version toute prête. D'abord les outils avec le décorateur @tool, que tu connais depuis la leçon sur les systèmes multi-agents :

1from langchain_core.tools import tool
2from langgraph.prebuilt import create_react_agent
3
4# Définitions des outils
5@tool
6def get_weather(location: str) -> str:
7    """Gets weather for a Safari location."""
8    weather_data = {
9        "Serengeti": "28C, sunny",
10        "Masai Mara": "25C, cloudy",
11        "Kruger": "30C, hot"
12    }
13    return weather_data.get(location, f"No data for {location}")
14
15@tool
16def search_animals(query: str) -> str:
17    """Searches for information about Safari animals."""
18    return f"Information about {query}: A fascinating animal of the African savanna!"
19
20@tool
21def calculate_safari_cost(days: int, people: int) -> str:
22    """Calculates Safari cost."""
23    cost_per_day = 200
24    total = days * people * cost_per_day
25    return f"Safari cost: {total} USD ({days} days, {people} people)"

Chaque outil a une docstring, que le modèle lit comme description, et des types de paramètres qui lui disent quoi fournir.

create_react_agent construit tout le graphe agent-outils en une seule ligne :

1# Crée un agent ReAct
2tools = [get_weather, search_animals, calculate_safari_cost]
3react_agent = create_react_agent(llm, tools)
4
5# Lance
6result = react_agent.invoke({
7    "messages": [HumanMessage(content="Calculate the cost of a 5-day Safari for 4 people")]
8})
9print(result["messages"][-1].content)

Le modèle appellera lui-même calculate_safari_cost avec days=5 et people=4. Attention : dans LangGraph 1.0, create_react_agent est marqué deprecated, et le nouveau code doit utiliser create_agent du paquet langchain.agents.

Multi-agent avec LangGraph

Un graphe convient parfaitement à une équipe d'agents. Chaque agent est un nœud, et le champ current_agent de l'état indique qui passe ensuite :

1from langgraph.graph import StateGraph, START, END
2from typing import Literal
3
4class MultiAgentState(TypedDict):
5    messages: Annotated[list, operator.add]
6    current_agent: str
7    task_complete: bool
8
9# Agents spécialistes
10def researcher_agent(state: MultiAgentState) -> MultiAgentState:
11    """Agent de recherche - rassemble des informations."""
12    system_prompt = "You are a Safari researcher. Gather information about animals and locations."
13
14    response = llm.invoke([
15        {"role": "system", "content": system_prompt},
16        *state["messages"]
17    ])
18
19    return {
20        "messages": [AIMessage(content=f"[Researcher]: {response.content}")],
21        "current_agent": "analyzer"
22    }

Le chercheur ajoute sa réponse à messages et désigne l'analyste comme suivant.

L'analyste et le rédacteur fonctionnent de la même façon, seul le rédacteur termine avec le drapeau task_complete :

1def analyzer_agent(state: MultiAgentState) -> MultiAgentState:
2    """Agent analytique - analyse les données."""
3    system_prompt = "You are an analyst. Analyze gathered information and draw conclusions."
4
5    response = llm.invoke([
6        {"role": "system", "content": system_prompt},
7        *state["messages"]
8    ])
9
10    return {
11        "messages": [AIMessage(content=f"[Analyzer]: {response.content}")],
12        "current_agent": "writer"
13    }
14
15def writer_agent(state: MultiAgentState) -> MultiAgentState:
16    """Agent rédacteur - crée le rapport final."""
17    system_prompt = "You are a writer. Create an engaging report based on the analysis."
18
19    response = llm.invoke([
20        {"role": "system", "content": system_prompt},
21        *state["messages"]
22    ])
23
24    return {
25        "messages": [AIMessage(content=f"[Writer]: {response.content}")],
26        "current_agent": "end",
27        "task_complete": True
28    }

Chaque agent voit tout l'historique des messages, car messages grandit à chaque nœud.

Il reste le routeur et le graphe :

1def router(state: MultiAgentState) -> Literal["analyzer", "writer", "end"]:
2    """Router entre les agents."""
3    return state["current_agent"]
4
5# Construis le graphe multi-agents
6multi_agent_graph = StateGraph(MultiAgentState)
7
8multi_agent_graph.add_node("researcher", researcher_agent)
9multi_agent_graph.add_node("analyzer", analyzer_agent)
10multi_agent_graph.add_node("writer", writer_agent)
11
12multi_agent_graph.add_edge(START, "researcher")
13multi_agent_graph.add_conditional_edges(
14    "researcher",
15    router,
16    {"analyzer": "analyzer"}
17)
18multi_agent_graph.add_conditional_edges(
19    "analyzer",
20    router,
21    {"writer": "writer"}
22)
23multi_agent_graph.add_conditional_edges(
24    "writer",
25    router,
26    {"end": END}
27)
28
29multi_agent_app = multi_agent_graph.compile()
30
31# Lance le pipeline
32result = multi_agent_app.invoke({
33    "messages": [HumanMessage(content="Prepare a report about lions in the Serengeti")],
34    "current_agent": "researcher",
35    "task_complete": False
36})

L'ordre est linéaire, mais le routeur permet d'ajouter facilement un retour vers le chercheur.

Checkpointing et persistance

Par défaut, le graphe oublie tout après invoke. Un checkpointer enregistre l'état après chaque étape, ce qui permet de poursuivre une conversation :

1from langgraph.checkpoint.memory import MemorySaver
2from langgraph.checkpoint.sqlite import SqliteSaver  # pip install langgraph-checkpoint-sqlite
3import sqlite3
4
5# Checkpoint en mémoire (in-memory)
6memory_checkpointer = MemorySaver()
7
8# Checkpoint SQLite (persistant)
9sqlite_checkpointer = SqliteSaver(sqlite3.connect("safari_agent.db", check_same_thread=False))
10
11# Compile avec checkpointer
12app_with_memory = workflow.compile(checkpointer=memory_checkpointer)

MemorySaver garde l'état en mémoire et le perd au redémarrage. SqliteSaver, du paquet langgraph-checkpoint-sqlite, l'enregistre dans un fichier de base SQLite.

Les conversations se distinguent par le thread_id de la configuration :

1# Lance avec thread_id pour la continuation
2config = {"configurable": {"thread_id": "safari-session-1"}}
3
4# Premier message
5result1 = app_with_memory.invoke(
6    {"messages": [HumanMessage(content="Tell me about lions")]},
7    config
8)
9
10# Continue la conversation (même thread_id)
11result2 = app_with_memory.invoke(
12    {"messages": [HumanMessage(content="What about their hunting?")]},
13    config
14)
15
16# L'historique est préservé !
17print(result2["messages"])

Le deuxième appel avec le même thread_id voit le premier message, donc le modèle sait que "their" désigne les lions.

Human-in-the-loop

Certaines décisions, un agent ne devrait pas les prendre seul. D'abord l'état, les nœuds et le routeur :

1from langgraph.graph import StateGraph, START, END
2from langgraph.checkpoint.memory import MemorySaver
3
4class HumanLoopState(TypedDict):
5    messages: Annotated[list, operator.add]
6    needs_approval: bool
7    approved: bool
8
9def agent_node(state: HumanLoopState) -> HumanLoopState:
10    """L'agent propose une action."""
11    response = llm.invoke(state["messages"])
12
13    # Vérifie si l'action nécessite une approbation
14    needs_approval = "delete" in response.content.lower() or "buy" in response.content.lower()
15
16    return {
17        "messages": [response],
18        "needs_approval": needs_approval
19    }
20
21def human_approval_node(state: HumanLoopState) -> HumanLoopState:
22    """Point d'approbation humaine."""
23    # Ce nœud interrompt l'exécution
24    # L'utilisateur doit approuver avant de continuer
25    return {"approved": state.get("approved", False)}
26
27def execute_node(state: HumanLoopState) -> HumanLoopState:
28    """Exécute l'action approuvée."""
29    return {"messages": [AIMessage(content="Action executed!")]}
30
31def approval_router(state: HumanLoopState) -> Literal["human", "execute", "end"]:
32    """Router vérifiant si une approbation est nécessaire."""
33    if state.get("needs_approval") and not state.get("approved"):
34        return "human"
35    elif state.get("approved"):
36        return "execute"
37    return "end"

Le nœud human ne calcule rien, c'est seulement un point d'arrêt. Il renvoie une clé inchangée, car renvoyer tout l'état ajouterait une deuxième fois tous les messages via le réducteur operator.add.

On compile le graphe avec un checkpointer et interrupt_before :

1# Graphe avec human-in-the-loop
2human_loop_graph = StateGraph(HumanLoopState)
3
4human_loop_graph.add_node("agent", agent_node)
5human_loop_graph.add_node("human", human_approval_node)
6human_loop_graph.add_node("execute", execute_node)
7
8human_loop_graph.add_edge(START, "agent")
9human_loop_graph.add_conditional_edges(
10    "agent",
11    approval_router,
12    {"human": "human", "execute": "execute", "end": END}
13)
14human_loop_graph.add_edge("human", "agent")  # Après approbation, retour à l'agent
15human_loop_graph.add_edge("execute", END)
16
17# Compile avec interrupt_before pour le nœud human
18app = human_loop_graph.compile(
19    checkpointer=MemorySaver(),
20    interrupt_before=["human"]  # Arrête avant le nœud human
21)

Le routeur renvoie "end", c'est pourquoi le dictionnaire des chemins traduit ce nom en END. Sans lui, LangGraph lève une erreur de nœud inconnu end.

L'exécution s'arrête avant le nœud human, puis on approuve et on reprend :

1# Lance - s'arrêtera avant le nœud human
2config = {"configurable": {"thread_id": "approval-1"}}
3result = app.invoke(
4    {"messages": [HumanMessage(content="Delete all files")]},
5    config
6)
7
8# Vérifie l'état
9state = app.get_state(config)
10print(f"Needs approval: {state.values.get('needs_approval')}")
11
12# L'utilisateur approuve
13app.update_state(config, {"approved": True})
14
15# Poursuit l'exécution
16final_result = app.invoke(None, config)

update_state enregistre l'accord, et invoke(None, config) relance le graphe, tandis que le routeur envoie le flux vers execute. Dans les versions récentes, la fonction interrupt() appelée dans un nœud est plus pratique.

Branching et exécution parallèle

Deux branches indépendantes peuvent tourner en même temps. D'abord les fonctions des nœuds :

1from langgraph.graph import StateGraph, START, END
2from typing import Literal
3
4class ParallelState(TypedDict):
5    query: str
6    weather_result: str
7    animal_result: str
8    combined_result: str
9
10def weather_branch(state: ParallelState) -> ParallelState:
11    """Récupère la météo en parallèle."""
12    # Appel d'API simulé
13    return {"weather_result": "Weather: 28C, sunny"}
14
15def animal_branch(state: ParallelState) -> ParallelState:
16    """Récupère les infos animales en parallèle."""
17    return {"animal_result": "Animals: Lions, elephants, giraffes"}
18
19def combine_results(state: ParallelState) -> ParallelState:
20    """Combine les résultats des branches parallèles."""
21    combined = f"{state['weather_result']}. {state['animal_result']}"
22    return {"combined_result": combined}

Chaque branche écrit une clé d'état différente, il n'y a donc pas de conflit.

Deux arêtes depuis START séparent le flux :

1# Graphe avec exécution parallèle
2parallel_graph = StateGraph(ParallelState)
3
4parallel_graph.add_node("weather", weather_branch)
5parallel_graph.add_node("animals", animal_branch)
6parallel_graph.add_node("combine", combine_results)
7
8# Arêtes parallèles
9parallel_graph.add_edge(START, "weather")
10parallel_graph.add_edge(START, "animals")
11parallel_graph.add_edge("weather", "combine")
12parallel_graph.add_edge("animals", "combine")
13parallel_graph.add_edge("combine", END)
14
15parallel_app = parallel_graph.compile()
16
17result = parallel_app.invoke({"query": "Safari info"})
18print(result["combined_result"])

weather et animals s'exécutent dans la même étape, et combine ne démarre que lorsque les deux ont fini.

Supervisor Pattern

Dans ce modèle, un agent gère tous les autres. Le superviseur demande au modèle qui doit travailler :

1from langgraph.graph import StateGraph, START, END
2from typing import Literal
3
4class SupervisorState(TypedDict):
5    messages: Annotated[list, operator.add]
6    next_worker: str
7    completed_workers: list[str]
8
9def supervisor_node(state: SupervisorState) -> SupervisorState:
10    """Le superviseur décide quel worker doit agir."""
11    system_prompt = """You are a supervisor managing a team:
12    - researcher: gathers information
13    - analyst: analyzes data
14    - writer: creates reports
15
16    Based on the task and results so far, choose the next worker.
17    Reply with ONLY the worker name or 'FINISH' if the task is complete."""
18
19    completed = state.get("completed_workers", [])
20
21    response = llm.invoke([
22        {"role": "system", "content": system_prompt},
23        {"role": "user", "content": f"Completed: {completed}\nTask: {state['messages'][-1].content}"}
24    ])
25
26    next_worker = response.content.strip().lower()
27
28    return {"next_worker": next_worker}

Le superviseur renvoie seulement le nom du worker suivant.

Une fabrique crée les nœuds des workers, et le routeur traduit la décision :

1def worker_node(worker_name: str):
2    """Fabrique de nœuds worker."""
3    def node(state: SupervisorState) -> SupervisorState:
4        prompts = {
5            "researcher": "You are a researcher. Gather information.",
6            "analyst": "You are an analyst. Analyze the data.",
7            "writer": "You are a writer. Write the report."
8        }
9
10        response = llm.invoke([
11            {"role": "system", "content": prompts[worker_name]},
12            *state["messages"]
13        ])
14
15        completed = state.get("completed_workers", []) + [worker_name]
16
17        return {
18            "messages": [AIMessage(content=f"[{worker_name}]: {response.content}")],
19            "completed_workers": completed
20        }
21    return node
22
23def supervisor_router(state: SupervisorState) -> Literal["researcher", "analyst", "writer", "end"]:
24    """Router du superviseur."""
25    next_worker = state.get("next_worker", "").lower()
26    if next_worker == "finish" or next_worker == "end":
27        return "end"
28    return next_worker

Si le modèle répond quelque chose qui n'est pas dans la liste, le routeur renvoie un nom inconnu et le graphe plante, donc en production impose le format de la réponse.

Chaque worker rend la main au superviseur :

1# Construis le graphe supervisor
2supervisor_graph = StateGraph(SupervisorState)
3
4supervisor_graph.add_node("supervisor", supervisor_node)
5supervisor_graph.add_node("researcher", worker_node("researcher"))
6supervisor_graph.add_node("analyst", worker_node("analyst"))
7supervisor_graph.add_node("writer", worker_node("writer"))
8
9supervisor_graph.add_edge(START, "supervisor")
10
11# Toutes les arêtes reviennent au superviseur
12for worker in ["researcher", "analyst", "writer"]:
13    supervisor_graph.add_edge(worker, "supervisor")
14
15supervisor_graph.add_conditional_edges(
16    "supervisor",
17    supervisor_router,
18    {
19        "researcher": "researcher",
20        "analyst": "analyst",
21        "writer": "writer",
22        "end": END
23    }
24)
25
26supervisor_app = supervisor_graph.compile()
27
28# Lance
29result = supervisor_app.invoke({
30    "messages": [HumanMessage(content="Prepare a complete report on Safari in the Serengeti")],
31    "completed_workers": []
32})

Je te conseille de commencer par un graphe linéaire et d'ajouter les cycles et le superviseur quand tu as vraiment besoin de décisions en cours de route.

LangGraph est un outil puissant pour construire des flux d'IA complexes. Dans la prochaine leçon, tu vas découvrir les subagents et les hiérarchies d'agents !

Retiens : LangGraph, c'est une carte des sentiers sur laquelle on a le droit de faire demi-tour, de se séparer et d'attendre la décision du guide.