Acheminement, diffusion, ReAct, critique et approbation : cinq modèles LangGraph
Apprenez cinq modèles de flux de travail agents dans LangGraph, allant des routeurs et des boucles ReAct aux portes d’évaluation et à l’approbation humaine, ainsi que les contraintes nécessaires pour chacun d’eux en environnement de production.
Des instructions plus longues ne résolvent que rarement les problèmes liés aux fonctionnalités d’IA peu fiables. Lorsqu’un système doit naviguer, écrire du code, effectuer des vérifications de conformité ou modifier du texte destiné aux clients, une seule appel à un modèle non déterministe est insuffisante. Une structure bien organisée aide : le modèle prend des décisions là où un jugement est nécessaire, tandis que le code gère le routage, les boucles et l’arrêt du processus. Ci-dessous sont présentés cinq modèles de ce type, chacun accompagné d’un exemple exécutable en LangGraph en Python, ainsi que des précautions à prendre avant mise en production.
Pourquoi un graphe convient aux workflows d’agents
Les programmes conventionnels s’exécutent de manière linéaire. Les agents, en revanche, ont besoin de boucles, de branches conditionnelles et d’un état persistant : si le code généré échoue à un test, le système doit enregistrer l’erreur, revenir en arrière et réessayer.
LangGraph modélise cela sous forme de graphe orienté :
- Nœuds : fonctions Python qui effectuent une seule tâche, comme une requête SQL ou un appel à un modèle.
Pour en savoir plus sur ces primitives, consultez LangGraph en pratique : état, nœuds et arêtes.
Pattern 1 : le routeur
┌───> [Specialized Coding Agent] ───> [END]
[START] ──> [Router]
└───> [General Knowledge Agent] ───> [END]
Utilisez-le pour réduire la latence et les coûts, ou pour associer des intentions à des outils spécialisés.
coding ou general, et add_conditional_edges mappe cette étiquette sur un nœud de traitement. route_decision recourt à general si le modèle renvoie quelque chose d’inattendu.
from typing import TypedDict, Literal
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
# 1. Define the shared state
class RouterState(TypedDict):
query: str
route: str
response: str
# Use a fast, cost-effective model for classification
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 2. Define the Nodes
def classify_query(state: RouterState):
prompt = f"""Classify the following user query into one of two categories: 'coding' or 'general'.
Respond with exactly one word, either 'coding' or 'general'.
Query: {state['query']}"""
response = model.invoke([HumanMessage(content=prompt)])
classification = response.content.strip().lower()
return {"route": classification}
def handle_coding(state: RouterState):
return {"response": "Executing advanced syntax processing and code compilation logic..."}
def handle_general(state: RouterState):
return {"response": "Processing casual conversation or general knowledge search..."}
# 3. Define Conditional Routing Logic
def route_decision(state: RouterState) -> Literal["coding", "general"]:
return state["route"] if state["route"] in ["coding", "general"] else "general"
# 4. Construct the Graph
workflow = StateGraph(RouterState)
workflow.add_node("classifier", classify_query)
workflow.add_node("coding_agent", handle_coding)
workflow.add_node("general_agent", handle_general)
workflow.add_edge(START, "classifier")
workflow.add_conditional_edges("classifier", route_decision, {
"coding": "coding_agent",
"general": "general_agent"
})
workflow.add_edge("coding_agent", END)
workflow.add_edge("general_agent", END)
# Compile and Run
app = workflow.compile()
result = app.invoke({"query": "How do I implement a binary search tree in Python?"})
print(f"Route Taken: {result['route']}\nResponse: {result['response']}")
Pattern 2 : orchestrateur et travailleurs
┌───> [Worker A: Section 1] ───┐
[START] ──> [Orchestrator] ├───> [Worker B: Section 2] ───┼───> [Synthesizer] ───> [END]
└───> [Worker C: Section 3] ───┘
L’orchestrateur demande une liste JSON contenant deux sous-sujets ; le nœud workers écrit un paragraphe pour chacun d’eux, et le synthétiseur les combine.
import json
from typing import TypedDict, List
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
class OrchestratorState(TypedDict):
topic: str
tasks: List[str]
worker_outputs: List[str]
final_report: str
model = ChatOpenAI(model="gpt-4o", temperature=0.2)
def orchestrator_plan(state: OrchestratorState):
prompt = f"Create a JSON list of exactly two sub-topics needed to write a comprehensive guide about: {state['topic']}. Return ONLY a valid JSON list of strings."
response = model.invoke([HumanMessage(content=prompt)])
tasks = json.loads(response.content.strip())
return {"tasks": tasks, "worker_outputs": []}
def worker_execute(state: OrchestratorState):
outputs = []
for task in state["tasks"]:
prompt = f"Write a brief, highly technical paragraph explaining: {task}"
response = model.invoke([HumanMessage(content=prompt)])
outputs.append(response.content)
return {"worker_outputs": outputs}
def synthesize_report(state: OrchestratorState):
combined_context = "\n\n".join(state["worker_outputs"])
prompt = f"Combine the following sections into a cohesive newsletter update regarding {state['topic']}:\n\n{combined_context}"
response = model.invoke([HumanMessage(content=prompt)])
return {"final_report": response.content}
# Graph Construction
orchestrator_flow = StateGraph(OrchestratorState)
orchestrator_flow.add_node("orchestrator", orchestrator_plan)
orchestrator_flow.add_node("workers", worker_execute)
orchestrator_flow.add_node("synthesizer", synthesize_report)
orchestrator_flow.add_edge(START, "orchestrator")
orchestrator_flow.add_edge("orchestrator", "workers")
orchestrator_flow.add_edge("workers", "synthesizer")
orchestrator_flow.add_edge("synthesizer", END)
app = orchestrator_flow.compile()
output = app.invoke({"topic": "Quantum Computing Security Implications"})
print(output["final_report"])
Deux précautions à prendre. Ce nœud travailleur parcourt les tâches séquentiellement, donc rien ne s’exécute en parallèle ; l’API Send de LangGraph peut assigner un travailleur par tâche, en collectant les résultats via un réducteur d’état. De plus, les modèles enveloppent parfois le JSON dans des balises Markdown, il convient donc de valider le plan à l’aide d’une sortie structurée plutôt que de se fier à json.loads sur du texte brut.
Pattern 3 : ReAct, raisonnement et action en boucle
ReAct alterne entre raisonnement et actions : le modèle évalue la situation, appelle un outil tel qu’une recherche ou une requête de base de données, observe le résultat et s’arrête dès qu’il peut répondre.
┌────────────────────────┐
▼ │
[START] ──> [Reasoner (Thought)] ───> (Should Call Tool?) ───> [Tool Executor (Act)]
│
└─ (Has Final Answer) ──> [END]
Ce modèle convient aux agents de recherche, d’assistance et de débogage, où les données nécessaires ne peuvent pas être prédites.
Le raisonneur demande ACTION: call_stock_api ou FINAL: ..., y compris la dernière observation. L’outil renvoie une cotation simulée, et le retour vers reasoner clôt la boucle. should_continue limite ce processus à trois itérations.
from typing import TypedDict, Literal
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
class ReActState(TypedDict):
user_input: str
agent_thought: str
tool_output: str
final_answer: str
loop_count: int
model = ChatOpenAI(model="gpt-4o", temperature=0)
def reason(state: ReActState):
loop_count = state.get("loop_count", 0) + 1
tool_context = f"\nTool Observation: {state.get('tool_output', '')}" if loop_count > 1 else ""
prompt = f"""You are a ReAct agent. Your goal is to find the current stock price of AAPL.
Current Loop: {loop_count} {tool_context}
Decide your next step. You must respond in one of two ways:
1. If you need data, say: 'ACTION: call_stock_api'
2. If you have the data, provide the answer starting with: 'FINAL: [your answer]'
User Request: {state['user_input']}"""
response = model.invoke([HumanMessage(content=prompt)]).content.strip()
if "FINAL:" in response:
return {"final_answer": response.replace("FINAL:", "").strip(), "loop_count": loop_count, "agent_thought": "done"}
else:
return {"agent_thought": "call_tool", "loop_count": loop_count}
def call_tool(state: ReActState):
print("-> System: Executing external stock database API call...")
mock_api_result = "$185.40 USD (Up 1.2% today)"
return {"tool_output": mock_api_result}
def should_continue(state: ReActState) -> Literal["call_tool", "end"]:
# Hard loop-break guardrail to prevent infinite execution loops
if state["agent_thought"] == "call_tool" and state["loop_count"] < 3:
return "call_tool"
return "end"
react_flow = StateGraph(ReActState)
react_flow.add_node("reasoner", reason)
react_flow.add_node("tool_executor", call_tool)
react_flow.add_edge(START, "reasoner")
react_flow.add_conditional_edges("reasoner", should_continue, {
"call_tool": "tool_executor",
"end": END
})
react_flow.add_edge("tool_executor", "reasoner")
app = react_flow.compile()
result = app.invoke({"user_input": "What is the market status of Apple right now?", "loop_count": 0})
print(f"\nFinal Agent Resolution:\n{result['final_answer']}")
Si la limite est atteinte avant d’obtenir une réponse FINAL:, final_answer n’est jamais défini et le dernier print provoque une erreur KeyError ; il faut donc gérer ce cas. Les systèmes réels utilisent généralement des appels d’outils natifs plutôt que des marqueurs de chaîne. Pour l’équivalent en TypeScript, consultez les boucles agentes limitées pour l’utilisation d’outils LLM.
Pattern 4 : évaluateur et optimiseur
┌───> [Optimizer (Generate/Refine)] ───> [Evaluator (Critique)]
│ │
└──────────────── (If Rejected) ────────────────┼───> [Approved] ───> [END
Cela convient à la rédaction, à la génération de code, à la traduction ainsi qu’à des règles strictes de qualité ou réglementaires. Un modèle moins coûteux, avec une température de 0,7, rédige un slogan en intégrant les retours obtenus lors des passes suivantes. L’évaluateur, avec une température de 0, vérifie que le texte contient les mots future ou smart et répond dans un format fixe ACCEPTED / FEEDBACK ; routing_gate renvoie ensuite les rejets.
from typing import TypedDict, Literal
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
class EvaluationState(TypedDict):
task: str
draft: str
feedback: str
accepted: bool
iterations: int
generator_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
evaluator_llm = ChatOpenAI(model="gpt-4o", temperature=0)
def generate_draft(state: EvaluationState):
iterations = state.get("iterations", 0) + 1
feedback_context = f"\nPrevious Feedback to incorporate: {state.get('feedback', '')}" if iterations > 1 else ""
prompt = f"""Write a catchy, 3-sentence marketing slogan for: '{state['task']}'.
{feedback_context}
Provide ONLY the slogan."""
response = generator_llm.invoke([HumanMessage(content=prompt)]).content.strip()
return {"draft": response, "iterations": iterations}
def evaluate_draft(state: EvaluationState):
prompt = f"""Review the following marketing slogan for the product '{state['task']}':
Slogan: "{state['draft']}"
CRITERIA: The slogan must include the exact word 'future' or 'smart'.
Respond in EXACTLY the following format:
ACCEPTED: True or False
FEEDBACK: [If rejected, explain what needs fixing. If accepted, leave blank.]"""
response = evaluator_llm.invoke([HumanMessage(content=prompt)]).content.strip()
accepted = "ACCEPTED: True" in response
feedback = response.split("FEEDBACK:")[-1].strip() if not accepted else ""
return {"accepted": accepted, "feedback": feedback}
def routing_gate(state: EvaluationState) -> Literal["refine", "approve"]:
if state["accepted"] or state["iterations"] >= 3:
return "approve"
return "refine"
eval_flow = StateGraph(EvaluationState)
eval_flow.add_node("generator", generate_draft)
eval_flow.add_node("evaluator", evaluate_draft)
eval_flow.add_edge(START, "generator")
eval_flow.add_edge("generator", "evaluator")
eval_flow.add_conditional_edges("evaluator", routing_gate, {
"refine": "generator",
"approve": END
})
app = eval_flow.compile()
result = app.invoke({"task": "Eco-friendly Electric Skateboards", "iterations": 0})
print(f"Final Slogan: {result['draft']}\nTotal Iterations: {result['iterations']}")
Le système approuve également après trois itérations, quel que soit le résultat, ce qui peut entraîner un rejet ; il convient donc de conserver le flag accepted avec le résultat. Une règle basée sur des mots-clés comme celle-ci est moins coûteuse et plus fiable lorsqu’elle est intégrée directement dans le code.
Pattern 5 : l’humain dans la boucle
Pour des opérations à risque telles que la suppression de tables, l’utilisation d’argent ou l’envoi d’e-mails aux clients, le mécanisme de point de contrôle permet au graphe de s’arrêter avant un nœud sensible, de conserver l’état en mémoire et d’attendre l’approbation.
[START] ──> [Stager] ──> ⛔ (State Saved to DB / Graph Pauses)
│
[DevOps Manager Clicks "Approve"]
│
▼
[Executor (Run Production Deploy)] ──> [END]
Utilisez-le pour les migrations, les déploiements, les paiements ou l’envoi massif d’e-mails.
stager prépare une commande et executor la exécute. La compilation avec un point de contrôle et interrupt_before=["executor"] arrête le processus après la phase de préparation. Les exécutions sont identifiées par thread_id ; ainsi, get_state affiche les valeurs enregistrées ainsi que l’étape en attente ('executor',) ; update_state enregistre l’approbation et invoke(None, config) reprend l’exécution.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
class DeploymentState(TypedDict):
command: str
approved: bool
execution_log: str
# 1. Initialize thread checkpoint memory saver
memory = MemorySaver()
def stage_deployment(state: DeploymentState):
print("-> System: Staging server deployment commands...")
return {"command": "sudo systemctl restart production_api"}
def execute_deployment(state: DeploymentState):
print("-> System: Execution approved. Running command on production servers...")
return {"execution_log": f"Successfully executed: {state['command']}"}
hitl_flow = StateGraph(DeploymentState)
hitl_flow.add_node("stager", stage_deployment)
hitl_flow.add_node("executor", execute_deployment)
hitl_flow.add_edge(START, "stager")
hitl_flow.add_edge("stager", "executor")
hitl_flow.add_edge("executor", END)
# CRITICAL: Define the interrupt checkpoint before the executor node runs
app = hitl_flow.compile(checkpointer=memory, interrupt_before=["executor"])
# --- SIMULATING THE ACTIVE DEPLOYMENT WORKFLOW ---
config = {"configurable": {"thread_id": "prod_deploy_001"}}
# 1. Kick off the graph execution
initial_state = app.invoke({"command": "", "approved": False}, config)
# Verify the graph successfully halted its progress
print(f"\n[Current Graph State]: {app.get_state(config).values}")
print(f"[Next Pending Steps]: {app.get_state(config).next}") # Next step will say: ('executor',)
print("\n--- Halting Execution. Waiting for DevOps Manager Review... ---\n")
# 2. Simulate Human Reviewing the State and Updating with Approval
app.update_state(config, {"approved": True}, as_node="stager")
# 3. Resume execution thread seamlessly from the exact checkpoint
final_output = app.invoke(None, config)
print(f"[Final System Output]: {final_output['execution_log']}")
Trois mises en garde. MemorySaver ne fonctionne qu’en mémoire ; les pauses qui persistent après un redémarrage nécessitent un point de contrôle basé sur une base de données. executor ne vérifie jamais l’état approved, il faut donc ajouter une vérification ou un mécanisme conditionnel pour interrompre l’exécution en cas de refus. Les versions plus récentes de LangGraph proposent également une fonction interrupt() ; consultez la documentation actuelle pour connaître l’approche recommandée.
Choix du modèle
La fiabilité provient de l’adaptation de la structure au problème, et non de modèles plus gros ou de prompts plus longs :
- Router : de nombreux types de requêtes avec des besoins différents en termes de coût ou de compétences.
- Orchestrator et workers : une tâche importante divisée en parties indépendantes.
- ReAct : les informations nécessaires ne peuvent être obtenues qu’en temps d’exécution.
Les modèles se composent de parties : un routeur peut envoyer une tâche à un agent ReAct dont l’action finale attend approbation. Appliquez toujours trois principes de sécurité : fixez une limite à chaque boucle, validez la sortie du modèle dont le code dépend, et enregistrez si un résultat a été approuvé ou si les tentatives se sont épuisées. Ainsi, le modèle n’a pas besoin d’être correct dès la première fois, car le flux de travail lui permet de router, de tester, de réessayer et de faire appel aux humains.
Lectures complémentaires
- Construire un agent de recherche ReAct dans LangGraph : Cerveau, Mains, Routageur — Apprenez à mettre en œuvre le cycle raisonner-agir-observer de ReAct en tant que sous-graphique LangGraph, avec une réflexion forcée, des budgets d’itération et une recherche par dispersion-collecte en parallèle.
- Construire un agent IA de zéro : Patterns, ReAct et LangGraph — Découvrez les concepts fondamentaux des agents IA — planification, utilisation d’outils, réflexion et le pattern ReAct — ainsi que le rôle de LangChain et LangGraph dans la création manuelle d’un tel agent.