Créer un agent de planification dans LangGraph : planifier, exécuter, réplanifier, faire un rapport
Un agent de planification LangGraph à quatre nœuds pour la recherche financière : des plans structurés, un registre d’outils conscient du nom d’espace, des limites de replanification et un exemple d’exécution traçé.
Sources: Your AI Agent Isn’t Smart. Here’s How to Build One That Actually Thinks
La plupart des agents ne sont que des chaînes de prompts déguisés
Voici un fait dérangeant au sujet des agents d’IA : beaucoup ne sont rien de plus que des chaînes de prompts embellis sous un faux air d’agents autonomes. La recette est familière : un prompt système, quelques outils et éventuellement un boucle ReAct, et le résultat est qualifié d’agent. Ça fonctionne lors des démonstrations et dans les cas simples. Mais dès qu’un outil renvoie une erreur, que les données semblent anormales ou que la requête nécessite plus de trois étapes, tout s’effondre.
L’alternative est un agent de planification. Il élabore un plan de recherche complet avant d’utiliser tout outil, le met en œuvre étape par étape, et lorsque les résultats intermédiaires contredisent les attentes, il réécrit ce qui reste. La démonstration concerne un système de recherche financière. Donnez-lui une demande en langue simple comme « Comparez les valorisations de NVDA et AAPL et dites-moi laquelle semble la meilleure pour l’instant », et il élaborera un plan en plusieurs étapes, exécutera chaque étape à l’aide d’outils de données, se corrige automatiquement en cas de problème et rédigera un rapport de type analyste. Il repose sur LangGraph, fonctionne avec des modèles cloud via Groq ou des modèles locaux via Ollama, et tient dans environ quatre fichiers.
Pourquoi ReAct pur ne suffit pas
ReAct (Reason-Act-Observe) est le modèle de fonctionnement à partir duquel la plupart des agents commencent. Le LLM réfléchit à son prochain mouvement, invoque une outil, lit les résultats, réfléchit à nouveau et invoque l’outil suivant. Il est populaire, simple et adapté aux tâches de petite envergure.
La faiblesse majeure réside dans l’avidité, car un agent ReAct ne planifie que un pas à l’avance. Si l’on demande des informations sur cinq actions selon trois indicateurs ainsi que des nouvelles récentes, il commence par exemple par le ratio P/E de NVDA, et le premier résultat détourne son attention. Un prototype a consommé sept appels d’outils pour chaque indicateur concernant Tesla avant de se souvenir de la tâche consistant à comparer plusieurs actions ; son rapport final se composait d’un résumé des résultats financiers de Tesla ainsi que d’une seule phrase concernant les autres actions.
La capacité du modèle n’est pas la cause ici. C’est l’architecture. Lorsqu’une seule étape d’inférence doit choisir ce qu’il faut rechercher, comment le faire et quoi en faire des résultats, aucune de ces décisions ne reçoit l’attention nécessaire. Un agent de planification sépare la réflexion de l’action : il planifie d’abord, exécute ensuite, puis vérifie si le plan reste valable. Chaque nœud gère une seule tâche, et elles s’exécutent les unes après les autres.
Quatre nœuds et un cycle
L’agent de planification est un graphe orienté composé de quatre nœuds, à l’image d’un organigramme où chaque case gère une tâche et transmet l’état suivant.
START --> Planner --> Executor <--> Replanner --> Reporter --> END
- Le Planificateur prend la requête de l’utilisateur et élabore un plan de recherche structuré, étape par étape. Chaque étape indique l’outil à utiliser, les arguments ainsi que la raison de son choix.
- Le Exécutant prend en charge l’étape suivante, lance l’outil spécifié et enregistre le résultat dans un carnet de notes.
- Le Replanificateur examine tout ce qui a été collecté jusqu’à présent après chaque exécution. Si un outil a commis une erreur ou que les données révèlent quelque chose d’inattendu, il réécrit les étapes restantes. Sinon, il laisse le plan inchangé.
- Le Rapporteur transforme toutes les entrées du carnet de notes en un rapport d’analyse final, accompagné de chiffres concrets et de recommandations.
Ce qui guide la conception, c’est le boucle Exécutant-Réplanificateur. Les deux nœuds s’alternent jusqu’à ce que le plan soit finalisé, et cette alternance produit un comportement d’autocorrection, car après chaque action, l’agent peut reconsidérer sa décision.
État : la mémoire partagée
Dans LangGraph, l’état représente tout, à savoir la mémoire partagée transmise entre les nœuds. Définition complète :
import operator
from typing import Annotated, TypedDict
from pydantic import BaseModel, Field
class StrategyState(TypedDict, total=False):
"""Global state that flows through the LangGraph nodes."""
query: str
plan: list[dict]
scratchpad: Annotated[list[dict], operator.add]
current_step: int
final_report: str
replan_count: int
Dans cet état, query conserve la requête originale de l’utilisateur sans modification, plan représente la liste ordonnée des étapes, scratchpad rassemble tous les résultats des outils, current_step fait référence au plan, et replan_count assure le respect du plafond de sécurité MAX_REPLANS.
L’annotation Annotated[list[dict], operator.add] sur le scratchpad est importante. Son reducteur, operator.add, indique à LangGraph d’ajouter de nouvelles entrées tout en laissant les existantes en place. Lorsque l’Executor renvoie {"scratchpad": [new_entry]}, LangGraph l’ajoute à la liste actuelle, de sorte que les nœuds n’ont jamais à gérer manuellement des listes et qu’aucun bug de mutation ne survient.
Plans structurés et décisions
The Planner ne restitue jamais de texte brut. Son résultat est un modèle Pydantic, de sorte que le LLM doit générer des plans sous une forme que le code peut parser :
AVAILABLE_TOOLS_TEXT = """
- get_metrics(ticker, metric?): Return stock metrics. 'metric' is optional
(P/E, EPS, Revenue, Market Cap, Sector).
- search_news(ticker): Return recent news headlines for a ticker.
- compare_metrics(tickers: list, metric): Compare one metric across
multiple tickers.
"""
class PlanStep(BaseModel):
"""A single executable step inside an analysis plan."""
step_id: int = Field(description="Sequential step number")
tool: str = Field(
description=f"Tool to use. Must be one of:\n{AVAILABLE_TOOLS_TEXT}"
)
args: dict = Field(description="Arguments for the tool call")
purpose: str = Field(description="Why this step is needed")
class Plan(BaseModel):
"""The full plan generated by the planner node."""
goal: str = Field(description="The overall analysis goal")
steps: list[PlanStep] = Field(
description="Ordered list of steps to execute"
)
Le champ purpose s’avère plus utile que prévu. Il indique au Replanner pourquoi une étape existe, et non seulement ce qu’elle fait. En lisant « Comparer les ratios P/E pour identifier la valorisation relative » au lieu d’un simple « call compare_metrics », le Replanner peut juger beaucoup mieux s’il convient de conserver, de modifier ou d’omettre cette étape.
Le Replanner dispose de son propre schéma :
class ReplanDecision(BaseModel):
"""Output of the replanner node."""
reasoning: str = Field(
description="Analysis of current progress and findings"
)
should_replan: bool = Field(
description="Whether the plan needs modification"
)
updated_steps: list[PlanStep] = Field(
default_factory=list,
description="Remaining steps if replan is needed. Empty if no changes.",
)
En faisant de should_replan une valeur booléenne, on force le LLM à se décider d’une manière ou d’une autre : le plan change-t-il ou non ? Laissez de la marge de manœuvre au LLM dans les sorties structurées, et il agira différemment à chaque fois.
Le système d’outils
Trois outils de recherche financière fonctionnent avec des données simulées réalistes. En environnement de production, on utiliserait des appels API réels, mais les données simulées présentent un grand avantage lors du développement : une débogage déterministe. Lorsque l’agent se comporte de manière étrange, il devient immédiatement clair si le problème vient de sa logique ou d’une API.
Un modèle de conception qui mérite plus d’attention est le registry d’outils conscient des noms de domaine, utilisé plutôt que de fournir à chaque nœud tous les outils possibles :
from langchain_core.tools import BaseTool
from typing import Iterable, Mapping, Any
class ToolRegistry:
"""Namespace-aware container for LangChain tools."""
def __init__(self) -> None:
self._tools_by_toolset: dict[str, dict[str, BaseTool]] = {}
def add_tools(self, toolset: str, tools: Iterable[BaseTool]) -> None:
bucket = self._tools_by_toolset.setdefault(toolset, {})
bucket.update({t.name: t for t in tools})
def get_tools(self, toolset: str) -> tuple[BaseTool, ...]:
return tuple(self._tools_by_toolset.get(toolset, {}).values())
def invoke(
self, toolset: str, tool_name: str, tool_args: Mapping[str, Any]
) -> Any:
t = self._tools_by_toolset.get(toolset, {}).get(tool_name)
if t is None:
raise ValueError(
f"Unknown tool '{tool_name}' in toolset '{toolset}'"
)
return t.invoke(dict(tool_args))
L’enregistrement a lieu dans les noms d’espace "planner", "executor" et "replanner". Tous les outils sont visibles pour le Planner afin qu’il puisse les utiliser pour planifier, mais seul l’Executor peut les appeler. Même si une erreur dans la commande du Replanner générait malgré tout une demande d’appel d’outil, le registre la refuserait car ce nom d’espace ne dispose pas de droits d’exécution. Il s’agit là du contrôle d’accès basé sur les rôles pour les agents, et la séparation des responsabilités agit également comme un mécanisme de sécurité qui empêche les modifications indésirables d’état pendant la planification.
Le nœud Planner
def planner_node(state: StrategyState) -> dict:
"""Create a step-by-step research plan using structured output."""
planner = model.with_structured_output(Plan)
prompt = PLAN_PROMPT.format(
available_tools=AVAILABLE_TOOLS_TEXT,
ticker_choices=ticker_choices_text(),
metric_choices=metric_choices_text(),
query=state["query"],
)
plan: Plan = planner.invoke(prompt)
steps = [s.model_dump() for s in plan.steps]
return {"plan": steps, "current_step": 0}
model.with_structured_output(Plan) fait en sorte que le LLM réponde avec un objet JSON compatible avec Pydantic, et tout résultat incorrect déclenche simplement une nouvelle tentative. Cela surpasse de loin l’analyse de texte brut à l’aide de regex.
Règles de contrôle dans le prompt du plan
PLAN_PROMPT = """\
You are a financial research planner. Given a user's analysis request,
create a step-by-step research plan using the available tools.
Available tools:
{available_tools}
Rules:
- Use only the tools listed above.
- Every plan step must be executable with one of those tools.
- When a tool accepts 'ticker' or 'tickers', use only these exact values:
{ticker_choices}
- When a tool accepts 'metric', use one of these exact values:
{metric_choices}
- There are no other tools available. Final synthesis is handled separately.
Create an efficient plan. Group related lookups. Aim for 4-8 steps.
User request: {query}"""
Deux détails se détachent. L’instruction énumère les valeurs de paramètre valides, ce qui permet à l’LLM de ne pas devoir deviner quels titres ou indicateurs existent. Cela pourrait être le stratagème le plus efficace contre les hallucinations pour les agents appelant des outils. Sans cela, le modèle planifie avec assurance des étapes comme get_metrics(ticker="BITCOIN", metric="vibes") avant de sembler surpris lorsque celles-ci échouent.
L’autre détail est la ligne "Il n’y a pas d’autres outils disponibles." Lorsqu’elle manque, les LLM inventent des étapes telles que analyze_sentiment(ticker="NVDA") ou generate_report(findings=...), des outils qui n’existent pas.
L’instruction « viser 4 à 8 étapes » est délibérément souple. Des limites strictes telles que « exactement 5 étapes » ont eu des effets contraires, en ajoutant des étapes redondantes aux requêtes simples et en surchargeant les étapes pour les requêtes complexes. Une fourchette permet au modèle de s’adapter, et avec des directives plutôt que des ordres impératifs, il prend généralement la bonne décision.
Le nœud Executor
L’Executor est intentionnellement le nœud le plus simple : il prend l’étape suivante, exécute l’outil via le registre et stocke le résultat.
MAX_STEPS = 12 # Safety limit on total steps
def executor_node(state: StrategyState) -> dict:
"""Execute the next pending step from the plan."""
plan = state.get("plan", [])
current_step = state.get("current_step", 0)
if current_step >= len(plan):
return {}
if current_step >= MAX_STEPS:
return {"current_step": len(plan)}
step = plan[current_step]
tool_name = step["tool"]
tool_args = step["args"]
try:
result = str(
TOOL_REGISTRY.invoke(
AgentName.EXECUTOR.value, tool_name, tool_args
)
)
status = "Error" if result.startswith("Error:") else "Success"
except Exception as exc:
result = f"Error: {exc}"
status = "Error"
entry = {
"step": current_step + 1,
"tool": tool_name,
"args": tool_args,
"result": result,
"status": status,
}
return {
"scratchpad": [entry],
"current_step": current_step + 1,
}
MAX_STEPS = 12 sert de limite stricte. Si le Replanner continue d’ajouter des étapes ou se bloque, l’agent ne peut pas consommer indéfiniment des crédits API. Une fois cette limite atteinte, l’Executor définit current_step à len(plan), et le flux passe au Reporter avec les données collectées jusqu’alors.
Les échecs sont reconnus selon une convention : status = "Error" si result.startswith("Error:") else "Success". Chacun des trois outils préfixe ses messages d’erreur par "Error:", permettant ainsi à l’Executor de marquer un échec sans avoir à analyser du JSON ni gérer des exceptions spécifiques. Le Replanner lit ce statut pour décider s’il faut réessayer ou ajuster le plan.
Le nœud Replanner
Ce nœud met en évidence la différence entre un agent de planification et un simple orchestrateur. Après chaque exécution d’outil, il évalue toutes les preuves et se demande si le plan restant reste pertinent :
MAX_REPLANS = 2 # Prevent infinite replanning
def replanner_node(state: StrategyState) -> dict:
"""Review progress and optionally modify the remaining plan."""
plan = state.get("plan", [])
current_step = state.get("current_step", 0)
replan_count = state.get("replan_count", 0)
scratchpad = state.get("scratchpad", [])
remaining = plan[current_step:]
if len(remaining) = MAX_REPLANS:
return {}
scratchpad_text = "\n".join(
f"Step {e['step']}: {format_tool_call(e['tool'], e['args'])} "
f"-> [{e['status']}] {e['result'][:150]}..."
for e in scratchpad
)
remaining_text = "\n".join(
f"Step {s['step_id']}: {format_tool_call(s['tool'], s['args'])} "
f"- {s['purpose']}"
for s in remaining
)
replanner = model.with_structured_output(ReplanDecision)
prompt = REPLAN_PROMPT.format(
goal=state["query"],
scratchpad=scratchpad_text,
remaining_steps=remaining_text,
)
decision: ReplanDecision = replanner.invoke(prompt)
if decision.should_replan and decision.updated_steps:
new_steps = plan[:current_step] + [
s.model_dump() for s in decision.updated_steps
]
return {"plan": new_steps, "replan_count": replan_count + 1}
return {"replan_count": replan_count + 1}
Regardez e['result'][:150]. Le Replanner affiche chaque résultat d’outil tronqué à 150 caractères, car il a besoin de l’essentiel et non du payload JSON complet, tandis que le Reporter reçoit par la suite les données non tronquées. Les tokens sont conservés là où les détails ne comptent pas et utilisés là où ils sont importants.
Lorsqu’un replan a lieu, plan[:current_step] + [s.model_dump() for s in decision.updated_steps] conserve les étapes terminées et ne remplace que celles qui ne le sont pas, laissant l’historique inchangé.
L’invite au replan
REPLAN_PROMPT = """\
You are a financial research planner reviewing progress on a research task.
Original goal: {goal}
Completed steps and findings so far:
{scratchpad}
Remaining steps in the plan:
{remaining_steps}
Based on the findings so far, should the remaining plan change?
If an expected tool failed or revealed something unexpected, add a step
to investigate.
If a step is now redundant, remove it.
Use only the available executable tools already shown in the plan.
Do not add recommendation, summary, or report-writing steps."""
La dernière ligne joue un rôle essentiel. En supprimant « Ne pas ajouter de recommandations, de résumés ou d’étapes de rédaction de rapport », Replanner ajoute une étape de synthèse à chaque plan, même si c’est le Reporter qui est chargé de cette tâche. Les prompts doivent donc délimiter clairement les frontières entre les nœuds.
Une limite de MAX_REPLANS = 2 s’est avérée idéale. Une première version utilisait six tours de replanification pour une tâche en trois étapes, le LLM trouvant des « améliorations » qui ne changeaient rien. Deux tours suffisent pour une véritable autocorrection, comme gérer un outil défaillant ou supprimer une étape redondante, sans risque de paralysie analytique. À un moment donné, il faut indiquer à l’agent d’arrêter de réfléchir et de terminer le travail.
Le rapporteur reçoit tout le contenu du carnet de notes non tronqué et rédige un rapport structuré en Markdown. Il appelle model.invoke(prompt).content, ce qui fournit du texte brut et non une sortie structurée, permettant ainsi à l’LLM de présenter l’analyse comme il le juge approprié. Une seule instruction suffit pour obtenir de meilleurs résultats : « Soyez précis, citez des chiffres issus des données. » Sans elle, les rapports restent vagues, se contentant d’affirmer que « NVDA présente de solides fondamentaux » sans mentionner les vraies valeurs de P/E ni les chiffres de chiffre d’affaires.
Le montage dans LangGraph
Ici, les quatre nœuds se rencontrent :
from langgraph.graph import StateGraph, END
from enum import Enum
class AgentName(Enum):
PLANNER = "planner"
EXECUTOR = "executor"
REPLANNER = "replanner"
REPORT = "report"
def build_graph():
"""Build and compile the LangGraph planning-agent workflow."""
workflow = StateGraph(StrategyState)
workflow.add_node(AgentName.PLANNER.value, planner_node)
workflow.add_node(AgentName.EXECUTOR.value, executor_node)
workflow.add_node(AgentName.REPLANNER.value, replanner_node)
workflow.add_node(AgentName.REPORT.value, report_node)
workflow.set_entry_point(AgentName.PLANNER.value)
workflow.add_edge(AgentName.PLANNER.value, AgentName.EXECUTOR.value)
workflow.add_edge(AgentName.EXECUTOR.value, AgentName.REPLANNER.value)
workflow.add_conditional_edges(
AgentName.REPLANNER.value,
should_continue_execution,
{
AgentName.EXECUTOR.value: AgentName.EXECUTOR.value,
AgentName.REPORT.value: AgentName.REPORT.value,
},
)
workflow.add_edge(AgentName.REPORT.value, END)
return workflow.compile()
def should_continue_execution(state: StrategyState) -> str:
"""Return the next node name after re-planning."""
if state.get("current_step", 0) >= len(state.get("plan", [])):
return AgentName.REPORT.value
return AgentName.EXECUTOR.value
Le graphique ressemble presque à de l’anglais. Le Planner passe au Executor, l’Executor au Replanner, et le Replanner se dirige vers l’Executor s’il reste des étapes à effectuer ou vers le Reporter une fois tout terminé. Le Reporter se rend à END.
L’enum AgentName semble mineur mais est utile. Des littéraux de chaîne simples tels que "planner" dispersés dans le code incitent aux fautes de frappe, et "plannr" créerait silencieusement un nœud non connecté. Avec l’enum, l’erreur apparaît au moment de l’importation.
Le routage vérifie une seule condition : toutes les étapes sont-elles terminées ? Lorsque current_step >= len(plan), le contrôle passe au Reporter, sinon il revient à l’Executor. La boucle s’arrête car current_step augmente à chaque exécution, et MAX_STEPS = 12 limite le nombre d’étapes, même lorsque le Replanner en ajoute.
Une requête, traçée
La demande de exemple demande à l’agent d’examiner NVDA et AMD, de présenter leurs ratios P/E ainsi que les dernières nouvelles côte à côte, puis de choisir l’un des deux.
Planification. Le planificateur génère cinq étapes : des indicateurs complets pour NVDA, des indicateurs complets pour AMD, une comparaison directe des ratios P/E, une recherche de nouvelles concernant NVDA et une recherche de nouvelles concernant AMD. Les profils individuels sont présentés en premier pour obtenir une vue d’ensemble, suivis de la comparaison directe, et les nouvelles en dernier car elles constituent un contexte supplémentaire et non des données de valorisation essentielles.
Exécution. Les étapes 1 et 2 renvoient des données JSON structurées, et à chaque fois le réplanificateur confirme que les données sont correctes et que le plan reste inchangé. L’étape 3 renvoie une comparaison des valeurs des ratios P/E :
{
"metric": "P/E",
"values": {
"NVDA": 58.3,
"AMD": 102.5
}
}
Les deux valeurs sont présentes et valides, donc rien n’est à replanifier. Les étapes 4 et 5 rapportent chacune trois articles de presse. Une fois l’étape 5 terminée, current_step vaut len(plan), et le routage transfère le contrôle au Reporteur.
Rapport. Ayant reçu cinq entrées complètes dans le carnet de notes, le Reporteur note des chiffres concrets : par exemple, un ratio P/E de 58,3 pour NVDA contre 102,5 pour AMD indique que NVDA présente une meilleure valeur relative, malgré un capital boursier supérieur d’environ 2,8 billions de dollars. Une exécution complète prend environ 8 à 10 secondes sur Groq et 30 à 40 secondes sur Ollama local en utilisant un modèle de 4 milliards de paramètres.
Mises à jour les plus importantes
La version actuelle utilise des données fictives, mais l’architecture est prête pour le déploiement en production. Trois mises à jour sont particulièrement importantes :
- Données financières réelles. Remplacez les recherches dans le dictionnaire par des appels à Yahoo Finance ou Alpha Vantage. Les signatures restent inchangées et seuls les corps des fonctions évoluent, grâce au motif de décorateur
@tool; ainsi, le Planificateur et l’Exécuteur ne savent jamais si les chiffres proviennent d’un fichier JSON fictif ou d’une API REST. - Mémoire persistante. Actuellement, le espace de travail temporaire n’est utilisé que pour une seule exécution. Avec un stockage vectoriel comme ChromaDB ou Pinecone, l’agent pourrait consulter des analyses antérieures, répondre à des questions telles que « Comment NVDA se compare-t-il à ce que nous avons vu le mois dernier ? », devenant ainsi un analyste évolutif plutôt qu’un outil sans état.
Ce que la structure ne peut pas fournir
Le cycle planifier-exécuter-replan fait son travail. Une sortie structurée maintient l’agent sur la bonne voie, les registres d’outils évitent le chaos, et des limites de sécurité empêchent des dépenses incontrôlées. La conception tient bon et le code reste ordonné.
Même ainsi, l’intelligence du système dans son ensemble repose sur le LLM qui se trouve en dessous de cette structure. Un modèle plus puissant élabore de meilleurs plans, détecte davantage d’erreurs lors du réplanification et rédige des rapports plus précis, tandis qu’un modèle plus faible suit scrupuleusement la structure et fournit une analyse médiocre. Le graphe ne peut pas rendre le modèle plus intelligent ; il se contente de lui fournir un cadre amélioré pour utiliser l’intelligence qu’il possède déjà.
La couche de gestion de projet est en grande partie résolue : l’état, l’orchestration des outils, le flux d’exécution et l’autocorrection suivent tous des schémas clairs. Le jugement, en revanche, reste une question ouverte. Quand le réplanificateur doit-il vraiment réplanifier, et quand doit-il rester silencieux ? Quand un chiffre est-il anormal et quand est-il simplement surprenant ? Actuellement, ces décisions sont intégrées dans des prompts, dans l’espoir que le LLM les résolve, ce qui fonctionne plus souvent que l’on ne s’y attendrait, mais moins souvent que l’on le souhaiterait.
Lectures complémentaires
- DS-STAR expliqué : l’agent de science des données de Google et les raisons de son succès — Une présentation détaillée de l’agent de science des données DS-STAR de Google : ses sept modules, le cycle de rapport DS-STAR+, les prompts publiés, les résultats d’ablation et les limites actuelles.
- Un agent AI DevOps en Python : LangChain, Claude et des outils Bash sécurisés — Créez un agent capable d’exécuter des commandes Bash, de lire des journaux et de interroger Kubernetes : des règles de sécurité en premier, puis des outils, un cycle ReAct, une mémoire, un enveloppe FastAPI ainsi que des solutions pour les pannes fréquentes.