LangGraph Multi-Agent Systems : superviseurs, sous-graphes et envoi

Prérequis : les exemples partent du principe que vous savez déjà créer un graphe LangGraph à un seul agent avec des outils et de la mémoire, ainsi que comment mettre en pause et reprendre une exécution à l’aide de interrupt(). Tout ce qui suit s’appuie sur ces bases, en passant de la question de savoir pourquoi on voudrait plusieurs agents à la création d’un assistant de recherche multi-agents complet.

Pourquoi diviser un agent en plusieurs

Supposons qu’un assistant à un seul agent fonctionne bien. Les utilisateurs commencent alors à lui demander de vérifier leurs e-mails, de rechercher des sujets, d’écrire des rapports et de gérer des tâches, tout au cours de la même conversation. La réponse habituelle consiste à ajouter des outils et à intégrer davantage d’instructions dans le prompt du système. Bientôt, on compte 15 outils et un prompt de 2 000 mots, ce qui entraîne une baisse progressive de la qualité. Le modèle commence à confondre les tâches, utilisant l’outil d’e-mail lors d’une tâche de recherche et l’outil de recherche lors d’une tâche d’e-mail.

C’est ce qu’on appelle la surcharge cognitive, et c’est la raison principale de l’existence des architectures multi-agents. La solution sur laquelle les ingénieurs logiciels comptent depuis des décennies est la suivante : diviser les responsabilités. Dans LangGraph, cela permet de résoudre trois problèmes distincts :

• Surcharge cognitive. Un seul modèle doit gérer trop de tâches non liées entre elles. Des agents spécialisés, chacun expert dans un domaine précis, réduisent cette pression.
• Bouchons séquentiels. Les tâches qui pourraient se dérouler en même temps s’exécutent une après l’autre. Les agents parallèles diminuent le temps de latence.
• Complexité. Personne ne peut raisonner sur un graphe de 40 nœuds. Des sous-graphe plus petits peuvent être analysés et testés un par un.

Chacun des modèles présentés ci-dessous vise l’un de ces trois problèmes.

Cinq niveaux d’architecture

Imaginez cette progression comme un escalier où chaque marche ajoute une idée supplémentaire.

Niveau 1 : superviseur. Un agent superviseur coordonne des travailleurs spécialisés. Il s’agit toujours d’un seul décideur, mais il délègue des tâches au lieu de tout faire lui-même, ce qui le distingue déjà d’un simple chatbot utilisant des outils.

Niveau 2 : parallélisation. Les tâches indépendantes sont divisées et exécutées en même temps, avant d’être fusionnées. L’idée repose sur des branches qui s’étendent et se rejoignent : plusieurs branches s’exécutent simultanément pour alimenter une étape ultérieure. Cela convient aux tâches répétitives sur plusieurs sources ou sous-tâches.

Niveau 3 : sous-graphes. Au lieu d’un seul flux de travail colossal, chaque branche devient un graphique plus petit disposant de ses propres étapes internes et de son propre schéma de sortie. Les parties complexes deviennent ainsi plus faciles à concevoir, à tester et à comprendre.

Niveau 4 : parallélisme dynamique. Avec le schéma Send, qui suit un style map-reduce, le nombre de branches n’est pas connu à l’avance. Des travailleurs sont créés en temps de exécution, un par élément, par sujet ou par analyste. Les agents ne sont pas prédéfinis, ils travaillent tous simultanément, et chacun est chargé d’une tâche spécifique.

Niveau 5 : tout combiné. Un système réaliste utilise l’ensemble des méthodes mentionnées ci-dessus, comme dans l’assistant de recherche présent à la fin.

Une façon rapide de s’en souvenir : au Niveau 1, les routes passent par un superviseur ; au Niveau 2, un ensemble fixe de travailleurs fonctionne ensemble ; au Niveau 3, chaque branche peut constituer un graphe complet ; au Niveau 4, des travailleurs sont créés sur demande.

Niveau 1 : le schéma du superviseur

C’est le modèle multi-agent le plus simple et le point de départ idéal. Un superviseur reçoit la demande de l’utilisateur et choisit quel spécialiste doit s’en occuper. Le spécialiste effectue le travail et transmet les résultats, puis le superviseur décide de la suite des opérations.

Imaginez un cabinet d’avocats. Le partenaire principal discute avec le client, comprend ses besoins et assigne des tâches aux associés juniors : un pour les contrats, un pour les litiges, un pour le respect des réglementations. L’associé effectue le travail détaillé, tandis que le partenaire principal le revue et répond au client.

Le flux se déroule comme suit :

User Input
    ↓
[Supervisor Node]  ← decides who handles what
    ↓         ↓
[Specialist A] [Specialist B]   ← do the actual work
    ↓         ↓
[Supervisor Node]  ← reviews results, responds or routes again
    ↓
Response

État

La décision de conception clé est le champ next_agent. Le superviseur y écrit et le routeur le lit.

# ── MODULE 2: STATE ─────────────────────────────────────────
from typing import TypedDict, Annotated, Literal
from langgraph.graph import MessagesState

class SupervisorState(MessagesState):
    # The supervisor writes this field to indicate who should act next.
    # "researcher", "writer", "FINISH" are the possible values.
    next_agent: str

Outils spécialisés

Chaque spécialiste dispose de ses propres outils ciblés, sans chevauchement entre eux.

# ── MODULE 3: TOOLS ─────────────────────────────────────────
from langchain_core.tools import tool

# Researcher's tools
@tool
def search_web(query: str) -> str:
    """Search the web for factual information, news, or research."""
    return f"[Web results for: {query}]"
@tool
def search_academic(query: str) -> str:
    """Search academic papers and journals for scholarly sources."""
    return f"[Academic results for: {query}]"
# Writer's tools
@tool
def format_as_report(content: str, title: str) -> str:
    """Format raw content into a structured report with headings."""
    return f"# {title}\n\n{content}"
@tool
def check_grammar(text: str) -> str:
    """Check and correct grammar in the provided text."""
    return f"[Grammar-checked]: {text}"
# Each specialist gets only its own tools
researcher_tools = [search_web, search_academic]
writer_tools = [format_as_report, check_grammar]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", temperature=0)
researcher_llm = llm.bind_tools(researcher_tools)
writer_llm     = llm.bind_tools(writer_tools)

Nœuds

Viennent ensuite le nœud superviseur et les nœuds spécialisés. La réponse en texte brut du superviseur constitue elle-même la décision de routage.

# ── MODULE 4: NODES ─────────────────────────────────────────
from langchain_core.messages import SystemMessage, HumanMessage
from langgraph.prebuilt import ToolNode

# ── Supervisor ────────────────────────────────────────────
SUPERVISOR_PROMPT = """You are a supervisor coordinating a team of specialists.
Given the conversation, decide who should act next.
Your team:
- researcher: finds information, searches web and academic sources
- writer: formats content, writes reports, checks grammar
- FINISH: use this when the task is complete and you have a final answer
Respond with ONLY the name of the next agent: researcher, writer, or FINISH."""
def supervisor_node(state: SupervisorState) -> dict:
    """Reads the conversation history and routes to the right specialist."""
    messages = [SystemMessage(content=SUPERVISOR_PROMPT)] + state["messages"]
    response = llm.invoke(messages)
    # The supervisor's plain text response IS the routing decision
    return {"next_agent": response.content.strip()}

# ── Researcher Specialist ──────────────────────────────────
def researcher_node(state: SupervisorState) -> dict:
    """Specialist that handles all research and information-gathering tasks."""
    messages = [
        SystemMessage(content="You are a research specialist. Use your search tools to find accurate, thorough information.")
    ] + state["messages"]
    response = researcher_llm.invoke(messages)
    return {"messages": [response]}

# ── Writer Specialist ──────────────────────────────────────
def writer_node(state: SupervisorState) -> dict:
    """Specialist that handles all writing, formatting, and editing tasks."""
    messages = [
        SystemMessage(content="You are a writing specialist. Use your tools to produce well-structured, polished output.")
    ] + state["messages"]
    response = writer_llm.invoke(messages)
    return {"messages": [response]}

# Tool execution nodes for each specialist
researcher_tool_node = ToolNode(researcher_tools)
writer_tool_node     = ToolNode(writer_tools)

Encadrement

# ── MODULE 5: ROUTING ────────────────────────────────────────
def route_after_supervisor(state: SupervisorState) -> Literal["researcher", "writer", "__end__"]:
    """Read the supervisor's decision and route accordingly."""
    decision = state["next_agent"]
    if decision == "FINISH":
        return "__end__"
    return decision  # "researcher" or "writer"

def route_researcher(state: SupervisorState) -> Literal["researcher_tools", "supervisor"]:
    """After researcher acts: did it call a tool, or produce a final answer?"""
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "researcher_tools"
    return "supervisor"  # Research done - report back to supervisor
def route_writer(state: SupervisorState) -> Literal["writer_tools", "supervisor"]:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "writer_tools"
    return "supervisor"

Assemblage du graphe

# ── MODULE 6: GRAPH ASSEMBLY ────────────────────────────────
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

graph_builder = StateGraph(SupervisorState)
# Register all nodes
graph_builder.add_node("supervisor",        supervisor_node)
graph_builder.add_node("researcher",        researcher_node)
graph_builder.add_node("researcher_tools",  researcher_tool_node)
graph_builder.add_node("writer",            writer_node)
graph_builder.add_node("writer_tools",      writer_tool_node)
# Entry: always start at supervisor
graph_builder.add_edge(START, "supervisor")
# Supervisor routes to specialist or ends
graph_builder.add_conditional_edges(
    "supervisor",
    route_after_supervisor,
    {"researcher": "researcher", "writer": "writer", "__end__": END}
)
# Researcher loop: tools → researcher → supervisor
graph_builder.add_conditional_edges(
    "researcher", route_researcher,
    {"researcher_tools": "researcher_tools", "supervisor": "supervisor"}
)
graph_builder.add_edge("researcher_tools", "researcher")
# Writer loop: tools → writer → supervisor
graph_builder.add_conditional_edges(
    "writer", route_writer,
    {"writer_tools": "writer_tools", "supervisor": "supervisor"}
)
graph_builder.add_edge("writer_tools", "writer")
graph = graph_builder.compile(checkpointer=MemorySaver())

Ce qui est important dans l’assemblage, c’est le flux qui revient au superviseur. Lorsqu’un spécialiste a terminé, il fait un rapport. Le superviseur examine toute la conversation puis choisit l’une de trois actions : transférer la tâche à un autre spécialiste, demander au même spécialiste de la reprendre ou terminer. C’est essentiellement cela, l’orchestration.

Niveau 2 : parallélisation avec diffusion et concentration

Le modèle du superviseur est séquentiel, avec un seul spécialiste actif à la fois. De nombreuses tâches réelles comprennent des sous-tâches indépendantes qui peuvent être exécutées en même temps, et le parallélisme permet de résoudre le problème de latence.

La cuisine est une bonne analogie. Personne ne termine la salade avant de commencer les pâtes, puis la sauce. On prépare la salade, on met de l’eau à bouillir et on commence à préparer la sauce en même temps, ce qui permet de servir le dîner plus rapidement.

Dans LangGraph, un nœud disposant de plusieurs arêtes sortantes crée une séparation parallèle. LangGraph détecte cette expansion et exécute simultanément toutes les cibles au sein de ce qu’il appelle une superétape. Les résultats sont collectés avant que tout autre nœud ne commence à fonctionner, ce qui correspond à l’entrée parallèle.

Il existe une règle stricte : lorsque plusieurs nœuds au sein d’une même superétape écrivent dans le même champ d’état, LangGraph doit savoir comment fusionner ces écritures. Sans réducteur, cela provoque une erreur. Avec operator.add, les résultats sont concaténés en toute sécurité.

# ── MODULE 2: STATE FOR PARALLEL PATTERN ────────────────────
import operator
from typing import Annotated

class ResearchState(TypedDict):
    query: str
    # Without Annotated + operator.add, parallel writes to 'context' would CRASH.
    # With it, LangGraph concatenates results from all parallel nodes.
    context: Annotated[list[str], operator.add]
    final_answer: str

Un exemple de recherche parallèle utilise deux sources :

# ── MODULE 4: PARALLEL WORKER NODES ─────────────────────────

def web_search_node(state: ResearchState) -> dict:
    """Runs in parallel with academic_search_node."""
    result = f"[WEB] Results for: {state['query']}"
    return {"context": [result]}   # Returns a LIST - operator.add will concatenate
def academic_search_node(state: ResearchState) -> dict:
    """Runs in parallel with web_search_node."""
    result = f"[ACADEMIC] Papers about: {state['query']}"
    return {"context": [result]}   # Also a LIST
def synthesize_node(state: ResearchState) -> dict:
    """Runs AFTER both parallel nodes complete (fan-in point).
    state['context'] already contains results from BOTH parallel nodes."""
    combined = "\n".join(state["context"])
    prompt = [
        SystemMessage(content="Synthesize the following research into a clear answer."),
        HumanMessage(content=f"Research:\n{combined}\n\nQuery: {state['query']}")
    ]
    response = llm.invoke(prompt)
    return {"final_answer": response.content}

# ── MODULE 6: PARALLEL GRAPH ASSEMBLY ───────────────────────
from langgraph.graph import StateGraph, START, END
graph_builder = StateGraph(ResearchState)
graph_builder.add_node("web_search",      web_search_node)
graph_builder.add_node("academic_search", academic_search_node)
graph_builder.add_node("synthesize",      synthesize_node)
# Fan-out: START branches to BOTH search nodes simultaneously
graph_builder.add_edge(START,       "web_search")
graph_builder.add_edge(START,       "academic_search")
# Fan-in: BOTH search nodes must complete before synthesize runs
graph_builder.add_edge("web_search",      "synthesize")
graph_builder.add_edge("academic_search", "synthesize")
graph_builder.add_edge("synthesize", END)
parallel_graph = graph_builder.compile()

Le fan-out consiste simplement en deux arêtes partant de la même source, START. LangGraph détecte cela et exécute simultanément les deux destinations. Le fan-in se produit également de lui-même, car la fonction synthesize n’est exécutée qu’une fois que les deux arêtes d’entrée sont satisfaites, c’est-à-dire une fois que les deux nœuds parallèles ont terminé leur traitement.

Niveau 3 : sous-graphes pour l’encapsulation

Les nœuds parallèles plats permettent de gérer des cas simples. Lorsque chaque branche est complexe, avec plusieurs étapes, ses propres outils et sa propre logique, les sous-graphes constituent un outil plus adapté. Un sous-graphe est un StateGraph entièrement compilé qui fonctionne comme un seul nœud au sein d’un graphique parent.

Pensez aux départements au sein d’une entreprise. Le PDG, qui représente le graphique parent, délègue des tâches au département Ingénierie et au département Marketing. Chaque département dispose de processus et de réunions internes, mais le PDG ne fait que transmettre des tâches et recevoir des résultats.

Le parent et le sous-graphe communiquent grâce à des clés d’état partagées. Toute clé présente à la fois dans l’état du parent et dans l’état du sous-graphe est transmise au démarrage du sous-graphe et renvoyée à sa fin.

Parent State: {topic, cleaned_data, report_A, report_B, processed_ids}
                         ↓
Subgraph A State: {cleaned_data, internal_step_1, report_A, processed_ids}
                  ← reads 'cleaned_data' from parent
                  → writes 'report_A' and 'processed_ids' back to parent

Subgraph B State: {cleaned_data, internal_step_2, report_B, processed_ids}
                  ← reads 'cleaned_data' from parent
                  → writes 'report_B' and 'processed_ids' back to parent

Les clés qui n’existent que dans le sous-graphe, comme internal_step_1 ci-dessus, restent privées, et le parent ne les voit jamais.

Voici une implémentation complète :

# ── MODULE 2: STATE SCHEMAS (Three Levels) ──────────────────

# Parent state - what the orchestrating graph tracks
class ParentState(TypedDict):
    raw_input: str
    cleaned_data: str                               # passed INTO both subgraphs
    failure_summary: str                            # written back by subgraph A
    performance_report: str                         # written back by subgraph B
    processed_ids: Annotated[list[int], operator.add]  # BOTH subgraphs write this

# Subgraph A's internal state - includes its private fields + shared keys
class FailureAnalysisState(TypedDict):
    cleaned_data: str        # shared input from parent
    failures: list[str]      # private intermediate state - parent never sees this
    failure_summary: str     # shared output to parent
    processed_ids: Annotated[list[int], operator.add]  # shared output (accumulated)

# Subgraph A's output schema - filters which keys return to parent
# Without this, ALL keys return, which can cause conflicts
class FailureAnalysisOutput(TypedDict):
    failure_summary: str     # only these two keys go back to parent
    processed_ids: Annotated[list[int], operator.add]

# Subgraph B's internal state
class PerformanceAnalysisState(TypedDict):
    cleaned_data: str
    metrics: list[str]          # private
    performance_report: str     # shared output
    processed_ids: Annotated[list[int], operator.add]

class PerformanceAnalysisOutput(TypedDict):
    performance_report: str
    processed_ids: Annotated[list[int], operator.add]

# ── SUBGRAPH A: Failure Analysis ─────────────────────────────
def extract_failures(state: FailureAnalysisState) -> dict:
    failures = [f"Error in {state['cleaned_data']}: timeout"]
    return {"failures": failures, "processed_ids": [1]}
def summarize_failures(state: FailureAnalysisState) -> dict:
    summary = f"Found {len(state['failures'])} failure(s): {state['failures']}"
    return {"failure_summary": summary}
failure_builder = StateGraph(FailureAnalysisState, output=FailureAnalysisOutput)
failure_builder.add_node("extract_failures",  extract_failures)
failure_builder.add_node("summarize_failures", summarize_failures)
failure_builder.add_edge(START, "extract_failures")
failure_builder.add_edge("extract_failures", "summarize_failures")
failure_builder.add_edge("summarize_failures", END)
failure_subgraph = failure_builder.compile()  # compiled → can be used as a node

# ── SUBGRAPH B: Performance Analysis ────────────────────────
def collect_metrics(state: PerformanceAnalysisState) -> dict:
    metrics = [f"Latency: 120ms", "Uptime: 99.8%"]
    return {"metrics": metrics, "processed_ids": [2]}
def write_report(state: PerformanceAnalysisState) -> dict:
    report = f"Performance Report:\n" + "\n".join(state["metrics"])
    return {"performance_report": report}
performance_builder = StateGraph(PerformanceAnalysisState, output=PerformanceAnalysisOutput)
performance_builder.add_node("collect_metrics", collect_metrics)
performance_builder.add_node("write_report",    write_report)
performance_builder.add_edge(START, "collect_metrics")
performance_builder.add_edge("collect_metrics", "write_report")
performance_builder.add_edge("write_report", END)
performance_subgraph = performance_builder.compile()

# ── PARENT GRAPH ─────────────────────────────────────────────
def clean_data_node(state: ParentState) -> dict:
    return {"cleaned_data": state["raw_input"].strip().lower()}
def combine_reports_node(state: ParentState) -> dict:
    combined = f"{state['failure_summary']}\n\n{state['performance_report']}"
    print(f"Final combined report:\n{combined}")
    return {}
parent_builder = StateGraph(ParentState)
parent_builder.add_node("clean_data",           clean_data_node)
parent_builder.add_node("failure_analysis",     failure_subgraph)    # subgraph as node
parent_builder.add_node("performance_analysis", performance_subgraph) # subgraph as node
parent_builder.add_node("combine_reports",      combine_reports_node)
parent_builder.add_edge(START, "clean_data")
# Fan-out to both subgraphs (parallel execution)
parent_builder.add_edge("clean_data",           "failure_analysis")
parent_builder.add_edge("clean_data",           "performance_analysis")
# Fan-in: combine only after both subgraphs complete
parent_builder.add_edge("failure_analysis",     "combine_reports")
parent_builder.add_edge("performance_analysis", "combine_reports")
parent_builder.add_edge("combine_reports", END)
parent_graph = parent_builder.compile()

Le schéma de sortie, FailureAnalysisOutput et PerformanceAnalysisOutput dans cet exemple, est la partie que de nombreux tutoriels omettent, d’où les conflits liés aux états clés. Il définit avec précision quels champs le sous-graphe expose à son parent, et tout ce qui y manque est considéré comme un état interne privé.

Niveau 4 : l’API Send et map-reduce

Les branches parallèles et les sous-graphes couvrent les cas où le nombre de branches est connu au moment de la conception. Mais que se passe-t-il si un utilisateur souhaite que 5 sujets soient étudiés aujourd’hui et 20 demain ? Il n’est pas possible de coder fixement 20 branches. C’est là que intervient le modèle map-reduce, mis en œuvre grâce à Send.

Imaginez un éditeur qui attribue des chapitres. Un rédacteur reçoit le manuscrit, le divise en chapitres et les envoie simultanément à des correcteurs différents. Qu’il s’agisse de deux ou de vingt chapitres, c’est autant de correcteurs qui sont engagés. Une fois que tous ont terminé, leurs résultats sont rassemblés pour former le livre final. Le nombre de travailleurs simultanés est déterminé au moment de l’exécution, et non lors de la conception.

Trois mots-clés sont importants :

• Send(node_name, state), importé de langgraph.constants, crée une instance nouvelle d’un nœud doté de son propre chargement d’état. Une fonction de routage qui renvoie une liste de ces nœuds fait en sorte que LangGraph les lance tous en même temps, chacun avec un état indépendant.
• operator.add sur le champ de collecte représente l’étape de réduction. Au fur et à mesure que les travailleurs terminent, leurs résultats s’accumulent dans une liste qui grandit dans l’état parent.

État du travailleur peut être un dictionnaire personnalisé qui n’a pas besoin de correspondre à l’état du parent. Un travailleur ne contient que les données relatives à sa propre partie de travail.

Voici la mise en œuvre complète de map-reduce :

# ── MODULE 2: MAP-REDUCE STATE ──────────────────────────────
from langgraph.constants import Send

class OverallState(TypedDict):
    topic: str
    subjects: list[str]                         # populated by the map step
    jokes: Annotated[list[str], operator.add]   # accumulated by the reduce step
    best_joke: str
class JokeState(TypedDict):
    """The private state each parallel worker gets.
    Notice: this does NOT need to match OverallState."""
    subject: str

# ── MODULE 4: MAP-REDUCE NODES ──────────────────────────────
def generate_subjects(state: OverallState) -> dict:
    """MAP step 1: Expands the topic into a list of subjects to process."""
    prompt = [
        SystemMessage(content="Generate a list of 3 sub-topics related to the given topic. Return as comma-separated values only."),
        HumanMessage(content=state["topic"])
    ]
    response = llm.invoke(prompt)
    subjects = [s.strip() for s in response.content.split(",")]
    return {"subjects": subjects}

def generate_joke(state: JokeState) -> dict:
    """MAP step 2: Worker node. Each instance handles exactly ONE subject.
    Receives its own isolated JokeState - not the full OverallState."""
    prompt = [HumanMessage(content=f"Write a short, funny joke about: {state['subject']}")]
    response = llm.invoke(prompt)
    # Returns a LIST - operator.add in OverallState will accumulate these
    return {"jokes": [response.content]}

def pick_best_joke(state: OverallState) -> dict:
    """REDUCE step: All jokes are now in state['jokes']. Pick the winner."""
    jokes_text = "\n".join([f"{i}. {j}" for i, j in enumerate(state["jokes"])])
    prompt = [
        SystemMessage(content="You are a comedy judge. Pick the funniest joke from the list below. Reply with ONLY the number."),
        HumanMessage(content=jokes_text)
    ]
    response = llm.invoke(prompt)
    winning_index = int(response.content.strip())
    return {"best_joke": state["jokes"][winning_index]}

# ── MODULE 5: THE SEND ROUTER ────────────────────────────────
def continue_to_jokes(state: OverallState):
    """The routing function that LAUNCHES parallel workers.

    Returning a LIST of Send objects (not a string) is what triggers map-reduce.
    Each Send creates an independent, parallel execution of 'generate_joke'
    with its own state payload. The number of parallel workers is determined
    at runtime by how many subjects exist in state.
    """
    return [
        Send("generate_joke", {"subject": subject})
        for subject in state["subjects"]
    ]

# ── MODULE 6: MAP-REDUCE GRAPH ASSEMBLY ─────────────────────
graph_builder = StateGraph(OverallState)
graph_builder.add_node("generate_subjects", generate_subjects)
graph_builder.add_node("generate_joke",     generate_joke)
graph_builder.add_node("pick_best_joke",    pick_best_joke)
graph_builder.add_edge(START, "generate_subjects")
# This conditional edge uses the Send router - it fans out dynamically
graph_builder.add_conditional_edges(
    "generate_subjects",
    continue_to_jokes,   # returns a LIST of Send objects
    ["generate_joke"]    # list of possible destination nodes (for graph validation)
)
# Fan-in: pick_best_joke runs after ALL parallel generate_joke instances complete
graph_builder.add_edge("generate_joke", "pick_best_joke")
graph_builder.add_edge("pick_best_joke", END)
map_reduce_graph = graph_builder.compile()

La magie réside dans continue_to_jokes. Il renvoie une liste d’objets Send au lieu d’une chaîne de caractères, et lorsque LangGraph reçoit une telle liste d’une fonction de routage, il lance immédiatement chacun d’eux. Trois éléments dans state["subjects"] signifient trois nœuds generate_joke parallèles, et vingt éléments signifient vingt. Le graphe s’adapte en fonction de ce que génère le système en temps réel.

Niveau 5 : un assistant de recherche multi-agents

Désormais, tout est combiné : une orchestration de type superviseur, l’approbation humaine, des sous-graphe parallèles ainsi que l’API Send, au sein d’un système de type production. Il s’agit du projet de fin de formation du cours LangGraph Academy, commenté ici.

Le fonctionnement : un utilisateur indique un sujet de recherche. Le système crée une équipe de personnages d’analystes IA, que quelqu’un peut valider ou affiner. Chaque analyste agit ensuite en tant qu’agent secondaire pour des entretiens en parallèle, interrogeant un « expert » IA à l’aide de recherches sur le Web et rédigeant une section du rapport. Enfin, un nœud de rédacteur assemble ces sections pour former le document final.

User provides topic
        ↓
[create_analysts] → LLM generates N analyst personas
        ↓
[human_feedback]  ← INTERRUPT: human approves or sends feedback
        ↓         ↑
[should_continue] → if feedback, loop back to create_analysts
        ↓
[initiate_research] → Send API: spawns N parallel interview subgraphs
        ↓↓↓ (N parallel workers running simultaneously)
[interview_subgraph × N] → each analyst interviews an "expert" via web search
        ↓
[write_report]    → LLM compiles all sections into a final report

État à trois niveaux

from pydantic import BaseModel, Field

# ── Analyst persona - a Pydantic model for structured LLM output ──
class Analyst(BaseModel):
    name: str = Field(description="The analyst's full name")
    role: str = Field(description="Their professional role (e.g. 'Financial Analyst')")
    focus: str = Field(description="Their specific analytical focus area")

    @property
    def persona(self) -> str:
        return f"Name: {self.name}\nRole: {self.role}\nFocus: {self.focus}"
class Perspectives(BaseModel):
    analysts: list[Analyst]

# ── Outer graph state ─────────────────────────────────────────
class ResearchGraphState(TypedDict):
    topic: str
    max_analysts: int
    human_analyst_feedback: str      # human writes here during HITL pause
    analysts: list[Analyst]
    sections: Annotated[list, operator.add]  # accumulated from all parallel interviews
    final_report: str

# ── Interview subgraph state ──────────────────────────────────
class InterviewState(MessagesState):
    # inherits: messages: Annotated[list[BaseMessage], add_messages]
    max_num_turns: int
    context: Annotated[list, operator.add]  # search results accumulate here
    analyst: Analyst
    interview: str       # formatted transcript
    sections: list       # completed section (sent back to outer graph)

Génération d’analystes, avec une vérification humaine

# ── ANALYST GENERATION NODE ───────────────────────────────────
ANALYST_PROMPT = """You are generating a team of {max_analysts} research analysts
for the topic: {topic}.
{feedback_section}
Each analyst should have a unique perspective and area of focus.
Generate analysts with diverse viewpoints - financial, technical, social, etc."""
def create_analysts(state: ResearchGraphState) -> dict:
    """Generates analyst personas using structured LLM output."""
    feedback = state.get("human_analyst_feedback", "")
    feedback_section = (
        f"User feedback to incorporate:\n{feedback}" if feedback
        else "No feedback yet - generate your best initial set."
    )

    prompt = ANALYST_PROMPT.format(
        max_analysts=state["max_analysts"],
        topic=state["topic"],
        feedback_section=feedback_section
    )

    # with_structured_output: forces LLM to return a Pydantic model
    # instead of free-form text - ensures clean, usable data
    structured_llm = llm.with_structured_output(Perspectives)
    result = structured_llm.invoke([SystemMessage(content=prompt)])
    return {"analysts": result.analysts}

def human_feedback_node(state: ResearchGraphState) -> dict:
    """No-op node. Its only purpose is to be a pause point.
    The graph is compiled with interrupt_before=['human_feedback'].
    See Part 3 for the full HITL pattern."""
    pass  # Does nothing - the pause happens BEFORE this node via interrupt_before

def should_continue_to_research(state: ResearchGraphState) -> Literal["create_analysts", "initiate_research"]:
    """After human feedback: loop back to refine, or proceed to interviews."""
    if state.get("human_analyst_feedback"):
        return "create_analysts"
    return "initiate_research"

Le sous-graphe d’interview

Chaque analyste fonctionne en tant que son propre sous-graphe, un petit agent ReAct qui interroge un « expert » (un autre modèle LLM jouant ce rôle) et effectue des recherches sur le web.

from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode

@tool
def web_search_tool(query: str) -> str:
    """Search the web for information on a topic."""
    return f"[Search results for '{query}']"  # Replace with real search API
interview_tools = [web_search_tool]
interview_llm = llm.bind_tools(interview_tools)

def generate_question(state: InterviewState) -> dict:
    """The analyst node: generates the next interview question."""
    analyst = state["analyst"]
    messages = [
        SystemMessage(content=(
            f"You are {analyst.name}, a {analyst.role} focused on {analyst.focus}. "
            f"You are interviewing an expert about the research topic. "
            f"Ask insightful questions that align with your analytical focus. "
            f"When you have enough information, say 'Thank you, that is all I needed.'"
        ))
    ] + state["messages"]
    response = interview_llm.invoke(messages)
    return {"messages": [response]}

def generate_answer(state: InterviewState) -> dict:
    """The expert node: answers the analyst's question, using web search."""
    messages = [
        SystemMessage(content=(
            "You are an expert being interviewed. Answer thoroughly and factually. "
            "Use the web search tool when you need current data."
        ))
    ] + state["messages"]
    response = interview_llm.invoke(messages)
    return {"messages": [response]}

def save_interview(state: InterviewState) -> dict:
    """Formats the full Q&A exchange into a clean transcript string."""
    transcript = []
    for msg in state["messages"]:
        if hasattr(msg, "content"):
            role = "Analyst" if type(msg).__name__ == "HumanMessage" else "Expert"
            transcript.append(f"{role}: {msg.content}")
    return {"interview": "\n\n".join(transcript)}

def write_section(state: InterviewState) -> dict:
    """The final node: uses the interview transcript to write a report section."""
    analyst = state["analyst"]
    prompt = [
        SystemMessage(content=(
            f"Based on the following interview conducted by {analyst.name} ({analyst.role}), "
            f"write a structured report section focused on {analyst.focus}. "
            f"Be concise, factual, and cite specific points from the interview."
        )),
        HumanMessage(content=state["interview"])
    ]
    response = llm.invoke(prompt)
    # sections is the key that Send passes back to the outer graph
    return {"sections": [response.content]}

def route_interview(state: InterviewState) -> Literal["generate_answer", "save_interview"]:
    """Continue interviewing OR wrap up when analyst is satisfied or max turns hit."""
    last_message = state["messages"][-1]
    if (
        "thank you, that is all" in last_message.content.lower()
        or len(state["messages"]) >= state.get("max_num_turns", 6) * 2
    ):
        return "save_interview"
    # Check if analyst made a tool call (web search)
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "generate_answer"
    return "generate_answer"

# ── INTERVIEW SUBGRAPH ASSEMBLY ───────────────────────────────
interview_builder = StateGraph(InterviewState)
interview_tool_node = ToolNode(interview_tools)
interview_builder.add_node("generate_question", generate_question)
interview_builder.add_node("generate_answer",   generate_answer)
interview_builder.add_node("interview_tools",   interview_tool_node)
interview_builder.add_node("save_interview",    save_interview)
interview_builder.add_node("write_section",     write_section)
interview_builder.add_edge(START, "generate_question")
interview_builder.add_conditional_edges(
    "generate_question",
    route_interview,
    {"generate_answer": "generate_answer", "save_interview": "save_interview"}
)
interview_builder.add_conditional_edges(
    "generate_answer",
    lambda s: "interview_tools" if (hasattr(s["messages"][-1], "tool_calls") and s["messages"][-1].tool_calls) else "generate_question",
    {"interview_tools": "interview_tools", "generate_question": "generate_question"}
)
interview_builder.add_edge("interview_tools", "generate_answer")
interview_builder.add_edge("save_interview",  "write_section")
interview_builder.add_edge("write_section",   END)
interview_graph = interview_builder.compile()

Le graphique d’orchestration externe

# ── SEND ROUTER: Spawn one interview per analyst ─────────────
def initiate_research(state: ResearchGraphState):
    """Launches N parallel interview subgraphs via Send — one per analyst.
    This is the map-reduce Send pattern from Level 4, applied to subgraphs."""
    return [
        Send("interview", {
            "analyst": analyst,
            "messages": [HumanMessage(content=f"Research topic: {state['topic']}")],
            "context": [],
            "max_num_turns": 3,
            "interview": "",
            "sections": []
        })
        for analyst in state["analysts"]
    ]
# ── REPORT COMPILATION NODE ───────────────────────────────────
def write_report(state: ResearchGraphState) -> dict:
    """Fan-in point: all sections have arrived from parallel interviews.
    The LLM compiles them into a final report."""
    sections_text = "\n\n---\n\n".join(state["sections"])
    prompt = [
        SystemMessage(content=(
            "You are a senior editor. Compile the following report sections written by "
            "different analysts into one cohesive, well-structured final report. "
            "Add a proper introduction and conclusion. Do not repeat content."
        )),
        HumanMessage(content=f"Topic: {state['topic']}\n\nSections:\n{sections_text}")
    ]
    response = llm.invoke(prompt)
    return {"final_report": response.content}

# ── OUTER GRAPH ASSEMBLY ─────────────────────────────────────
outer_builder = StateGraph(ResearchGraphState)
outer_builder.add_node("create_analysts",   create_analysts)
outer_builder.add_node("human_feedback",    human_feedback_node)
outer_builder.add_node("interview",         interview_graph)   # subgraph as node
outer_builder.add_node("write_report",      write_report)
outer_builder.add_edge(START, "create_analysts")
outer_builder.add_edge("create_analysts", "human_feedback")
outer_builder.add_conditional_edges(
    "human_feedback",
    should_continue_to_research,
    {"create_analysts": "create_analysts", "initiate_research": "initiate_research"}
)
# initiate_research is a ROUTING FUNCTION (returns Send objects), not a node
outer_builder.add_conditional_edges(
    "human_feedback",
    initiate_research,   # returns list of Send objects
    ["interview"]        # validation: lists possible destinations
)
outer_builder.add_edge("interview",    "write_report")
outer_builder.add_edge("write_report", END)
research_graph = outer_builder.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["human_feedback"]  # HITL pause for analyst approval
)

Exécuter l’ensemble

if __name__ == "__main__":
    config = {"configurable": {"thread_id": "research-001"}}

    # Step 1: Generate analysts
    result = research_graph.invoke(
        {
            "topic": "The impact of AI on the future of software engineering",
            "max_analysts": 3,
            "human_analyst_feedback": "",
            "analysts": [],
            "sections": [],
            "final_report": ""
        },
        config=config
    )

    # Step 2: HITL — graph paused at human_feedback node
    snapshot = research_graph.get_state(config)
    analysts = snapshot.values["analysts"]

    print("Generated analysts:")
    for a in analysts:
        print(f"  - {a.name} ({a.role}): {a.focus}")

    feedback = input("\nProvide feedback to refine analysts (or press Enter to approve): ")

    if feedback.strip():
        # Human has feedback — inject it and let supervisor regenerate
        research_graph.update_state(
            config,
            {"human_analyst_feedback": feedback},
            as_node="human_feedback"
        )
    else:
        # Approved — clear feedback and proceed to interviews
        research_graph.update_state(
            config,
            {"human_analyst_feedback": None},
            as_node="human_feedback"
        )

    # Step 3: Resume — N parallel interviews launch via Send, then report is compiled
    print("\nRunning parallel interviews...")
    final_result = research_graph.invoke(None, config=config)

    print("\n" + "="*60)
    print("FINAL RESEARCH REPORT")
    print("="*60)
    print(final_result["final_report"])

Choix d’un schéma

• Un seul agent qui se confond entre les tâches nécessite le schéma de superviseur. Le superviseur dirige les tâches tandis que les spécialistes les exécutent.
• Des tâches indépendantes et lentes exigent des structures de type fan-out et fan-in. Exécutez-les simultanément et ajoutez des réducteurs operator.add pour les champs partagés.
• Des branches parallèles elles-mêmes complexes nécessitent des sous-graphes. Compilez chaque branche en tant que graphique distinct, intégrez-le en tant que nœud et utilisez des schémas de sortie pour contrôler ce qui est renvoyé au nœud parent.
• Un nombre d’agents déterminé à l’exécution requiert l’utilisation de Send. Un routeur génère une liste d’objets Send, et chacun d’eux crée un agent indépendant avec son propre état.
• Un véritable système de production utilise tout ce qui a été mentionné ci-dessus. L’assistant de niveau 5 combine tous les modèles car une tâche réelle en a besoin de tous.

Référence des mots-clés

Structure de supervision. supervisor_node est l’orchestrateur. Il lit l’état, écrit la valeur de next_agent et ne réalise jamais de tâches spécialisées. specialist_node gère une tâche spécifique à l’aide de ses propres outils et de son propre prompt, avant de renvoyer l’information au superviseur. next_agent: str est le champ d’état conventionnel utilisé pour le routage : le superviseur y écrit un nom que le routeur lit ensuite.

Parallélisme. Le fan-out correspond à plusieurs appels à add_edge depuis un même nœud source, que LangGraph exécute en parallèle. Le fan-in correspond à plusieurs arêtes menant vers un même nœud de destination, qui attend la fin du traitement de chaque source. Annotated[list, operator.add] est obligatoire pour tout champ vers lequel plusieurs nœuds parallèles écrivent des données, car sans cela les écritures parallèles provoquent une erreur ; operator.add étant le réducteur habituel, il concatène les listes provenant des nœuds concurrents.

Subgraphes. StateGraph(InternalState, output=OutputSchema) est la version à deux arguments, où le deuxième argument filtre les clés qui sont renvoyées au parent. Un schéma de sortie est un TypedDict ne contenant que les clés à exposer, tandis que tout le reste reste privé. Les clés superposées servent de canal entre le parent et l’enfant, car toute clé présente dans les deux schémas est partagée automatiquement. subgraph.compile() transforme le sous-graphique en une fonction appelable, qui est ensuite transmise à parent.add_node("name", compiled_subgraph). L’appel de graph.get_graph(xray=1).draw_mermaid() affiche la structure interne du sous-graphique dans le diagramme.

Map-reduce. Send(node_name, state_dict) correspond à une invocation d’un travailleur parallèle, et son chargement n’a pas besoin de correspondre à l’état parent. Le retour de [Send(...), Send(...)] depuis un routeur lance toutes ces invocations. Lors de l’utilisation de Send, le troisième argument de add_conditional_edges est une liste, et non un dictionnaire, contenant les noms de destination valides, utilisée uniquement pour la validation du graphe. Enfin, with_structured_output(PydanticModel) est utilisé après une appel à un modèle pour forcer la génération de JSON conforme à un schéma Pydantic. C’est la méthode habituelle pour obtenir des décisions fiables du superviseur ainsi que des listes d’analystes.

La composition est la compétence clé

Passer d’un agent à plusieurs n’est pas vraiment un saut qualitatif. Il s’agit simplement des mêmes éléments de base appliqués à plusieurs reprises et reliés entre eux. Chaque agent dans un système multi-agents constitue un graphe, un nœud ou un sous-graphe, et les primitives restent inchangées. Ce qui compte, c’est savoir comment les combiner.

Le chemin suivi était intentionnel : un superviseur, suivi de nœuds parallèles simples, de sous-graphes parallèles pour les branches complexes, des travailleurs dynamiques pilotés par Send, et enfin l’ensemble assemblé. Si une étape semblait naturelle, c’est que la conception avait réussi son objectif, car chaque niveau s’appuie de manière claire sur le précédent.

L’assistant de recherche de niveau 5 correspond presque à ce qu’implique une base de code multi-agents en environnement de production : des boucles d’approbation humaine, des spécialistes travaillant en parallèle, des sorties de modèles structurées, une génération dynamique d’agents et une étape de synthèse finale. Associés à une structure de projet canonique, à une gestion de la mémoire et à des mécanismes de sécurité impliquant l’intervention humaine, ces modèles couvrent l’architecture sous-jacente de la plupart des applications LangGraph réelles.