Assistant de recherche multi-agents : superviseur, recherche parallèle et synthèse

« Fais-moi une synthèse sur l’état de l’art des bases vectorielles en 2026. » Une seule question, mais qui en cache dix : quels acteurs, quelles performances, quels compromis, quelles tendances ? Un agent monolithique s’y noie. La bonne réponse est architecturale : décomposer, investiguer en parallèle, puis synthétiser.

Ce cas construit cet assistant de bout en bout. On y mobilise les patterns du deep-dive orchestration — superviseur, fan-out par Send, human-in-the-loop — et on trace tout avec LangFuse.

L’architecture

Trois rôles, un graphe. Un planificateur (le superviseur) découpe la question. Un essaim de chercheurs (le même agent ReAct, instancié N fois) traite chaque sous-question en parallèle. Un rédacteur assemble les trouvailles en rapport.

Le cœur du flux est un map-reduce : on disperse une branche par sous-question (map), chaque chercheur accumule ses trouvailles dans l’état partagé via un reducer, puis le rédacteur agrège (reduce).

L’état partagé

On commence par modéliser ce qui circule. Le champ findings est accumulé en parallèle : il lui faut un reducer.

from typing import Annotated
from typing_extensions import TypedDict
import operator

class Finding(TypedDict):
    question: str
    resume: str
    sources: list[str]

class Etat(TypedDict):
    sujet: str                                   # la question de recherche initiale
    sous_questions: list[str]                    # produites par le planificateur
    findings: Annotated[list[Finding], operator.add]   # accumulées EN PARALLÈLE
    rapport: str                                 # la synthèse finale

Construire l’assistant, étape par étape

from langchain_core.tools import tool

@tool
def recherche_web(requete: str) -> str:
    """Recherche sur le web. À utiliser pour toute question factuelle nécessitant
    des informations à jour ou des sources externes."""
    resultats = moteur.search(requete, max_results=5)
    return "\n\n".join(f"[{r.url}]\n{r.extrait}" for r in resultats)

outils = [recherche_web]

from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model

modele = init_chat_model("openai:gpt-4o")

class Plan(BaseModel):
    sous_questions: list[str] = Field(
        description="3 à 6 sous-questions ciblées et complémentaires couvrant le sujet"
    )

def planifier(etat: Etat) -> dict:
    plan = modele.with_structured_output(Plan).invoke(
        f"Décompose cette question de recherche en sous-questions précises, "
        f"non redondantes et couvrant l'ensemble du sujet :\n{etat['sujet']}"
    )
    return {"sous_questions": plan.sous_questions}

from langgraph.types import Send

def disperser(etat: Etat) -> list[Send]:
    return [Send("chercheur", {"question": q}) for q in etat["sous_questions"]]

from langgraph.prebuilt import create_react_agent

chercheur_agent = create_react_agent(modele, outils)

def chercheur(payload: dict) -> dict:
    question = payload["question"]
    res = chercheur_agent.invoke({
        "messages": [{"role": "user", "content":
            f"Recherche et synthétise une réponse sourcée à : {question}"}]
    })
    texte = res["messages"][-1].content
    sources = extraire_urls(res["messages"])         # collecte les URLs citées par les outils
    return {"findings": [{"question": question, "resume": texte, "sources": sources}]}

def rediger(etat: Etat) -> dict:
    blocs = "\n\n".join(
        f"### {f['question']}\n{f['resume']}\nSources : {', '.join(f['sources'])}"
        for f in etat["findings"]
    )
    rapport = modele.invoke(
        f"Rédige un rapport de synthèse clair et structuré sur « {etat['sujet']} », "
        f"à partir de ces recherches. Cite les sources.\n\n{blocs}"
    )
    return {"rapport": rapport.content}

from langgraph.graph import StateGraph, START, END

g = StateGraph(Etat)
g.add_node("planifier", planifier)
g.add_node("chercheur", chercheur)
g.add_node("rediger", rediger)

g.add_edge(START, "planifier")
g.add_conditional_edges("planifier", disperser, ["chercheur"])
g.add_edge("chercheur", "rediger")
g.add_edge("rediger", END)

assistant = g.compile()

Insérer une validation humaine du plan

Lancer dix recherches web coûte du temps et des tokens. Avant de disperser, on montre le plan à un humain. C’est le interrupt() du deep-dive checkpointing, employé ici comme point de contrôle métier.

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver

def valider_plan(etat: Etat) -> Command:
    decision = interrupt({
        "plan": etat["sous_questions"],
        "demande": "Valider ou amender ce plan de recherche ?",
    })
    if isinstance(decision, list):                # l'humain a fourni un plan amendé
        return Command(update={"sous_questions": decision})
    return Command()                              # plan accepté tel quel

g.add_node("valider_plan", valider_plan)
g.add_edge("planifier", "valider_plan")
g.add_conditional_edges("valider_plan", disperser, ["chercheur"])
assistant = g.compile(checkpointer=InMemorySaver())   # le HITL exige un checkpointer

À l’exécution, le graphe se gèle sur l’interruption ; on présente le plan, on reçoit la validation, puis on reprend avec Command(resume=...). La recherche coûteuse ne démarre qu’après le feu vert humain.

Instrumenter et lire la trace

On branche LangFuse — un simple callback, sans toucher la logique — et on regroupe toute l’exécution sous une session.

from langfuse.langchain import CallbackHandler

handler = CallbackHandler()

resultat = assistant.invoke(
    {"sujet": "État de l'art des bases vectorielles en 2026"},
    config={
        "callbacks": [handler],
        "configurable": {"thread_id": "recherche-001"},
        "metadata": {"langfuse_session_id": "recherche-001", "langfuse_tags": ["multi-agent"]},
        "recursion_limit": 40,
    },
)
print(resultat["rapport"])

La trace révèle l’architecture en un coup d’œil : la planification, puis les chercheurs imbriqués et parallèles, puis la rédaction — chacun avec son coût et sa latence.

Aller plus loin

Cette base se prête à des extensions naturelles, chacune adossée à un deep-dive :

• Mémoire — mémoriser les rapports déjà produits dans un Store pour ne pas re-chercher l’identique.
• Qualité — ajouter une boucle évaluateur-optimiseur qui fait critiquer le rapport et le réviser avant livraison.
• Coût — router les chercheurs vers un modèle plus petit, ne réservant le grand modèle qu’à la synthèse finale.
• UX — streamer la progression (« 3/6 sous-questions investiguées… ») via des événements custom.

Bonnes pratiques & pièges

Ce qu’il faut retenir

• Une question ouverte se traite par décomposition + parallélisme + synthèse, pas par un agent monolithique.
• Le planificateur produit un plan structuré ; Send réalise le fan-out, un reducer le fan-in.
• Chaque chercheur est un agent ReAct complet, encapsulé en un nœud — composition par sous-agents.
• Le human-in-the-loop valide le plan avant la dépense ; le checkpointer le rend possible.
• La trace LangFuse rend l’architecture — et son coût parallèle — lisible.

Ce système investigue largement. Le prochain cas resserre la focale sur la fiabilité d’une seule réponse : un RAG qui se relit, se corrige et refuse d’halluciner — le RAG agentique correctif.