Portes de validation et routage multi-agents pour un bot d’assistance LangGraph.
Ajouter des portes de logique métier déterministes, une boucle de validation auto-corrigée ainsi qu’un superviseur doté de trois sous-graphes spécialisés à un agent LangGraph pour le e-commerce.
Imaginez un bot de support client pour une boutique en ligne, appelé ici ShopBot, construit sur LangGraph. Il recherche les commandes, vérifie l’éligibilité aux remboursements, traite les remboursements après validation par un humain, et conserve un résumé de la conversation. Une version de base de ce bot fonctionne déjà. Ce qui suit ajoute trois éléments nécessaires à un bot ayant des conséquences réelles : des vérifications pour chaque sortie du modèle, une boucle qui corrige automatiquement le modèle, et une architecture multi-agents dans laquelle un superviseur assigne des tâches aux spécialistes. L’objectif est de disposer d’une base de code que l’on pourrait soumettre à une revue en production sans hésitation.
La base de référence supposée est un graphe comprenant les modules habituels : des outils, une boucle de type ReAct, un état, ainsi qu’une appel à interrupt() qui pause le graphe en attendant l’approbation humaine.
L’hypothèse implicite dans le bot de base
La version simple fait confiance au modèle de langage en toutes circonstances. Son agent effectue une vérification de conformité et suppose que l’ID de commande qu’il transmet a été correctement extrait du chat. Le nœud d’évaluation montre cet ID de commande à un humain, mais l’examinateur approuve ou rejette une action sans vérifier si l’ID existe réellement. Outil de remboursement : il fonctionne avec les arguments choisis par le modèle. Si le modèle génère faussement « ORD-99X » au lieu de « ORD-001 », l’outil commet une erreur, l’agent la détecte, et une solution raisonnable peut être trouvée. « Une solution raisonnable peut être trouvée » est une description insuffisante du traitement des erreurs dans un système qui gère de l’argent.
Les professionnels qui gèrent des flux de travail automatisés de passation de commandes ou de conformité insistent fortement : les échecs probabilistes ne sont pas acceptables dans ce contexte. La solution n’est pas de faire plus confiance au modèle, mais d’arrêter complètement de faire confiance à ses résultats et de vérifier chaque résultat structuré avec du Python déterministe avant qu’il ne touche quoi que ce soit à l’extérieur.
Trois ajouts et leur interaction
Les portes de logique métier se trouvent à l’intérieur des nœuds et valident les résultats du modèle avant que quoi que ce soit ne s’y attaque. Il s’agit de code Python simple sans appel à de modèles. Une porte autorise soit les données à passer, soit les rejette en indiquant une raison spécifique et structurée. Il s’agit donc d’une validation déterministe au sein d’un système probabiliste.
Les boucles de validation gèrent les défaillances des écluses. Un rejet ne fait pas planter le graphe et ne produit pas non plus silencieusement une réponse erronée. Le contrôle passe à un nœud de correction qui transmet la raison de la défaillance au modèle et demande une sortie corrigée. Un compteur de tentatives dans l’état empêche les boucles infinies, tandis qu’une limite stricte déclenche une escalade vers un humain.
Une architecture multi-agents remplace le nœud d’agent unique par trois spécialistes. Un agent de commande s’occupe des recherches et des demandes d’état. Un agent de remboursement s’occupe des vérifications de critères d’éligibilité et du traitement des remboursements. Un agent de plaintes gère les clients mécontents et les escalades. Un nœud de superviseur examine chaque message reçu et le renvoie au bon spécialiste, et chaque spécialiste constitue un sous-graphe compilé indépendant.
Ces éléments fonctionnent comme un seul système. L’agent de remboursement gère une porte de contrôle ; une porte défaillante peut déclencher une boucle de validation, et le superviseur examine la manière dont cette boucle s’est terminée avant de décider de réessayer, d’escalader le problème ou d’abandonner. Ensemble, ils confèrent au bot final la robustesse exigée par un environnement d’entreprise.
Planifiez le graphe avant de toucher au code
LangGraph encourage la planification, et plus le graphe est vaste, plus cela est vrai. Dessinez d’abord l’ensemble du système : chaque case doit correspondre à un seul nœud ou fonction de routage, et chaque flèche à un seul arête. Le code devient alors une traduction de ce schéma en Python, module par module.
La nouvelle structure des fichiers
Le bot de base tenait dans un seul fichier. À cette échelle, un seul fichier devient difficile à gérer, c’est pourquoi le projet est divisé en trois fichiers.
shopbot_enterprise/
├── gates.py ← Business Logic Gates (pure Python, no LangGraph imports)
├── agents.py ← All three specialist subgraphs
└── shopbot.py ← Supervisor graph + entrypoint (unchanged module structure)
Il est intentionnel de ne pas inclure les importations de LangGraph dans le fichier gates. Les gates deviennent ainsi testables avec des tests unitaires ordinaires : aucun modèle, aucun graphe et aucun processus en cours n’est nécessaire pour vérifier que la logique de validation est correcte. On appelle un gate avec des données d’échantillon et on vérifie le résultat, c’est ce que signifie concrètement le terme « déterministe ».
Étendre le schéma d’état
Le schéma d’état a besoin de nouveaux champs avant toute autre chose. La version de base comprenait trois champs personnalisés : un résumé, le nom du client et une catégorie de ticket. Quatre champs supplémentaires sont ajoutés.
# ============================================================
# MODULE 2: STATE (Enterprise Version)
# ============================================================
from typing import TypedDict, Annotated
from langgraph.graph import MessagesState
import operator
class SupportState(MessagesState):
# From the original ShopBot - unchanged
summary: str
customer_name: str
ticket_category: str
# NEW: The structured data the LLM extracted from the conversation.
# Gates validate this field. Tools use this field. The LLM never
# writes to external systems directly - it writes here first.
extracted_data: dict
# NEW: The current status of the gate decision for this turn.
# Values: "pending" | "passed" | "failed" | "escalated" | "aborted"
# The supervisor reads this to decide what to do after a gate runs.
gate_status: str
# NEW: A log of every gate failure in this conversation.
# Uses operator.add so entries accumulate rather than replace.
# This is your audit trail. Every compliance system needs one.
validation_errors: Annotated[list[str], operator.add]
# NEW: How many correction attempts have fired in this turn.
# The hard cap is 2. After 2 failed corrections, escalate to human.
# Without this, a broken LLM output creates an infinite loop.
retry_count: int
Le nouveau champ clé est extracted_data. Dans le bot de base, le modèle générait du texte libre et appelait directement les outils, de sorte que les arguments des outils correspondaient à ce que le modèle indiquait dans ses appels. Désormais, le modèle émet d’abord des données structurées qui sont stockées dans extracted_data. Le mécanisme de contrôle les vérifie, et ce n’est qu’après son approbation que ces données deviennent des arguments d’outil. Cela élimine tout lien direct entre le modèle et les systèmes externes.
Le champ validation_errors est marqué avec operator.add, ce qui permet à ses erreurs de s’accumuler tout au long de la conversation entière et non seulement sur une seule étape. Supposons qu’un ID de commande fictif apparaisse à l’étape trois et qu’une raison de remboursement insuffisante apparaisse à l’étape sept : toutes deux finissent dans le journal d’audit. Cela est important pour les audits de conformité.
Rédaction des mécanismes de contrôle
Une porte est une fonction de validation pure. Elle prend des données extraites et renvoie une tupule : (True, cleaned_data) en cas de succès ou (False, error_reason) en cas d’échec. Telle est l’ensemble de l’interface, sans appels à de modèles ni dépendances graphiques. Le module ci-dessous comprend une vérification de l’ID de commande, une vérification du remboursement complet et une vérification des plaintes, ainsi que les constantes métier qu’ils utilisent.
# ============================================================
# gates.py — Business Logic Gates
# Pure Python. No LangGraph. No LLM calls.
# Each gate returns (bool, data_or_error_string).
# ============================================================
import re
# The same fake database as the original ShopBot.
# In production this would be a database query.
ORDERS_DB = {
"ORD-001": {"customer": "Alex", "product": "Wireless Headphones",
"status": "Delivered", "amount": 89.99,
"delivery_date": "2025-06-10"},
"ORD-002": {"customer": "Sam", "product": "Phone Case",
"status": "In Transit", "amount": 14.99,
"delivery_date": "Expected 2025-06-18"},
"ORD-003": {"customer": "Jordan", "product": "Laptop Stand",
"status": "Processing", "amount": 45.00,
"delivery_date": "Expected 2025-06-20"},
}
MAX_REFUND_AMOUNT = 500.00 # Business rule: auto-refunds above this need director sign-off
MIN_REASON_LENGTH = 10 # Business rule: refund reasons must be substantive
def validate_order_id(extracted_data: dict) -> tuple[bool, dict | str]:
"""
Gate 1: Order ID Validation
Checks that the order ID the LLM extracted actually exists in our system
and follows the correct format. This is the most basic hallucination check
in a support workflow. LLMs frequently confuse order IDs, especially when
a customer has mentioned multiple orders or used an informal format like
"order one" instead of "ORD-001".
Why this matters: if the agent calls a tool with a hallucinated order ID,
the tool returns an error, the agent sees the error, and may hallucinate
a recovery that looks plausible but is wrong. Catching it here, before
the tool call, is the correct place to stop the failure.
"""
order_id = extracted_data.get("order_id", "").strip().upper()
# Check 1: Format validation - must match ORD-XXX pattern
if not re.match(r"^ORD-\d{3}quot;, order_id):
return (
False,
f"Order ID '{order_id}' does not match the required format ORD-XXX "
f"(example: ORD-001). The LLM may have extracted the wrong value. "
f"Ask the customer to confirm their exact order number."
)
# Check 2: Existence validation - must exist in our system
if order_id not in ORDERS_DB:
return (
False,
f"Order ID '{order_id}' passed format validation but does not exist "
f"in the orders database. This is likely a hallucination. "
f"Ask the customer to provide the exact order ID from their confirmation email."
)
# Gate passed: return cleaned data with the normalised order ID
cleaned = {**extracted_data, "order_id": order_id}
return True, cleaned
def validate_refund_request(extracted_data: dict) -> tuple[bool, dict | str]:
"""
Gate 2: Refund Request Validation
A compound gate that checks every business rule that must be true
before a refund can be processed. Runs after validate_order_id passes,
so we can assume order_id is valid at this point.
Business rules checked:
1. The order must exist and have Delivered status
2. The refund amount must be below the auto-approval threshold
3. The reason must be substantive (not empty or one word)
Each failed rule produces a specific error message that gets fed back
to the LLM as a correction instruction. Vague errors produce vague
corrections. Specific errors produce specific corrections.
"""
order_id = extracted_data.get("order_id", "").strip().upper()
reason = extracted_data.get("reason", "").strip()
order = ORDERS_DB.get(order_id, {})
errors = []
# Rule 1: Order must be in Delivered status
if order.get("status") != "Delivered":
errors.append(
f"Order {order_id} has status '{order.get('status')}'. "
f"Refunds can only be processed for Delivered orders. "
f"Inform the customer and do not attempt to process a refund."
)
# Rule 2: Amount must be within auto-approval limit
amount = order.get("amount", 0)
if amount > MAX_REFUND_AMOUNT:
errors.append(
f"Refund amount ${amount:.2f} exceeds the auto-approval limit "
f"of ${MAX_REFUND_AMOUNT:.2f}. This requires director-level sign-off. "
f"Escalate to a senior agent - do not process automatically."
)
# Rule 3: Reason must be substantive
if len(reason) < MIN_REASON_LENGTH:
errors.append(
f"Refund reason '{reason}' is too short (minimum {MIN_REASON_LENGTH} characters). "
f"Ask the customer to provide a more detailed reason for their return request."
)
if errors:
# Return all errors joined - the correction node will feed this
# exact string back to the LLM as a correction instruction
return False, " | ".join(errors)
return True, extracted_data
def validate_complaint(extracted_data: dict) -> tuple[bool, dict | str]:
"""
Gate 3: Complaint Validation
Ensures the complaints agent has enough structured data to work with
before attempting any escalation or response. Complaints without
a clear issue category cannot be routed correctly in a real ticketing system.
"""
issue_summary = extracted_data.get("issue_summary", "").strip()
sentiment = extracted_data.get("sentiment", "").strip().lower()
valid_sentiments = {"frustrated", "angry", "neutral", "disappointed"}
if len(issue_summary) < 20:
return (
False,
f"Complaint issue summary is too vague: '{issue_summary}'. "
f"Ask the customer to describe their problem in more detail before escalating."
)
if sentiment not in valid_sentiments:
return (
False,
f"Sentiment '{sentiment}' is not a recognised value. "
f"Valid values are: {', '.join(valid_sentiments)}. "
f"Re-classify the customer's sentiment and try again."
)
return True, extracted_data
Regardez ce que ce fichier manque : des importations depuis LangGraph, LangChain ou toute autre bibliothèque d’IA. N’importe quel développeur de l’équipe peut le lire, le tester et le modifier sans avoir la moindre connaissance en matière d’architecture d’agent. Lorsqu’un auditeur demande ce qui valide précisément une demande de remboursement avant que l’argent ne soit transféré, c’est ce fichier qu’il faut montrer, et les non-ingénieurs peuvent le comprendre.
Le cycle de validation
C’est le nœud de correction qui rend les portes utiles. Lorsqu’une porte échoue, l’exécution ne s’arrête pas mais est redirigée vers ce nœud. Celui-ci récupère la raison spécifique de l’échec dans l’état, crée une instruction indiquant au modèle exactement ce qui ne va pas, et demande une sortie structurée corrigée.
# ── agents.py (partial) ─────────────────────────────────────
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from pydantic import BaseModel, Field
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# ── Pydantic Models (Structured LLM Output) ──────────────────
# These are the schemas that define what the LLM must return.
# with_structured_output() enforces these - if the LLM returns
# something that doesn't match, Pydantic raises a validation error
# before the output even reaches a gate.
class OrderExtraction(BaseModel):
order_id: str = Field(description="The order ID in the format ORD-XXX, e.g. ORD-001")
query_type: str = Field(description="What the customer wants to know: 'status', 'delivery_date', 'product', or 'general'")
class RefundExtraction(BaseModel):
order_id: str = Field(description="The order ID in the format ORD-XXX")
reason: str = Field(description="The customer's stated reason for wanting a refund, in their own words")
class ComplaintExtraction(BaseModel):
issue_summary: str = Field(description="A one-sentence summary of the customer's complaint")
sentiment: str = Field(description="The customer's emotional state: 'frustrated', 'angry', 'neutral', or 'disappointed'")
# ── The Correction Node ───────────────────────────────────────
def correction_node(state) -> dict:
"""
The validation loop's core. When a Business Logic Gate fails, this
node receives control. It reads the most recent gate failure from
state, builds a precise correction instruction for the LLM, and
asks for a new extraction attempt.
This node is shared across all three specialist agents. The
ticket_category in state tells it which Pydantic model to use
for the re-extraction, so the same node handles order corrections,
refund corrections, and complaint corrections without duplicating logic.
Critical design decision: the correction instruction goes into the
messages list as a HumanMessage, not a SystemMessage. This makes it
part of the visible conversation history. If the LLM makes the same
mistake twice, the history shows both the first failure and the second
correction attempt, which helps diagnose systematic prompt problems.
"""
# Read the most recent validation error from the audit log
errors = state.get("validation_errors", [])
last_error = errors[-1] if errors else "Unknown validation error."
retry_count = state.get("retry_count", 0)
# Hard stop: after 2 retries, do not attempt again - escalate
if retry_count >= 2:
return {
"gate_status": "escalated",
"messages": [
AIMessage(
content=(
f"I was unable to process your request correctly after {retry_count} attempts. "
f"I'm escalating this to a senior support agent who will contact you directly. "
f"I apologise for the inconvenience."
)
)
]
}
# Build the correction instruction - be maximally specific
correction_instruction = (
f"Your previous response contained a validation error that must be corrected "
f"before we can proceed. The exact error was:\n\n"
f"{last_error}\n\n"
f"Please re-read the conversation carefully and provide a corrected response "
f"that addresses this specific error. Do not repeat the same mistake."
)
category = state.get("ticket_category", "order_inquiry")
# Choose the right structured output model based on what kind of
# correction is needed
if category == "refund_request":
structured_llm = llm.with_structured_output(RefundExtraction)
elif category == "complaint":
structured_llm = llm.with_structured_output(ComplaintExtraction)
else:
structured_llm = llm.with_structured_output(OrderExtraction)
messages = (
[SystemMessage(content="You are a customer support assistant. Extract the requested information accurately.")]
+ state["messages"]
+ [HumanMessage(content=correction_instruction)]
)
try:
corrected = structured_llm.invoke(messages)
return {
"extracted_data": corrected.model_dump(),
"gate_status": "pending",
"retry_count": retry_count + 1,
"messages": [HumanMessage(content=correction_instruction)],
}
except Exception as e:
# Structured output itself failed (Pydantic validation error)
# Count this as a retry and surface the raw error
return {
"gate_status": "failed",
"retry_count": retry_count + 1,
"validation_errors": [f"Structured output failed: {str(e)}"],
}
La vérification de tentative de réessai en haut est le mécanisme de sécurité le plus important de tout le système. Sans elle, une entrée véritablement défectueuse (un client qui ne fournit pas un ID de commande valide, ou une configuration de modèle qui continue de produire des erreurs) tournerait indéfiniment, alternant entre le nœud de correction et le portail de validation. Deux tentatives de réessai constituent une valeur par défaut raisonnable pour le travail d’assistance, car elles donnent au modèle deux chances de se corriger avant l’intervention d’un humain. Ce nombre peut être ajusté en fonction du niveau de latence tolérable par rapport à la précision.
Les trois sous-graphes spécialisés
Chaque spécialiste est un sous-graphique compilé disposant de ses propres outils, de sa propre logique d’extraction, de son portail de validation et de son routage interne. Le graphique superviseur parent ne connaît pas cette structure interne et s’en moque ; il ne voit qu’un nœud qui reçoit un état et en retourne un autre.
L’Agent de commande
# ── agents.py (continued) ────────────────────────────────────
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from gates import validate_order_id, ORDERS_DB
@tool
def lookup_order(order_id: str) -> str:
"""Look up the details of a customer's order by order ID.
Use this when the customer wants to know their order status,
product name, or expected delivery date.
"""
order = ORDERS_DB.get(order_id.upper())
if not order:
return f"No order found with ID '{order_id}'."
return (
f"Order {order_id.upper()}: {order['product']} | "
f"Status: {order['status']} | "
f"Amount: ${order['amount']:.2f} | "
f"Delivery: {order['delivery_date']}"
)
order_tools = [lookup_order]
order_tool_node = ToolNode(order_tools)
order_llm = llm.bind_tools(order_tools)
def order_extract_node(state) -> dict:
"""
Step 1 for the Order Agent: extract the order ID from the conversation
using structured output. The result goes into state['extracted_data'].
The gate runs next. The tool call only happens if the gate passes.
"""
structured_llm = llm.with_structured_output(OrderExtraction)
messages = (
[SystemMessage(content="Extract the order ID and query type from the customer's message.")]
+ state["messages"]
)
try:
result = structured_llm.invoke(messages)
return {
"extracted_data": result.model_dump(),
"gate_status": "pending",
"ticket_category": "order_inquiry",
}
except Exception as e:
return {
"gate_status": "failed",
"validation_errors": [f"Extraction failed: {str(e)}"],
"ticket_category": "order_inquiry",
}
def order_gate_node(state) -> dict:
"""
Step 2: Run the Business Logic Gate on the extracted data.
Either passes gate_status to 'passed' and lets the tool run,
or sets it to 'failed' and logs the reason to validation_errors.
"""
extracted = state.get("extracted_data", {})
passed, result = validate_order_id(extracted)
if passed:
return {"gate_status": "passed", "extracted_data": result}
return {
"gate_status": "failed",
"validation_errors": [result], # operator.add accumulates this
}
def order_agent_node(state) -> dict:
"""
Step 3 (only runs after gate passes): call the LLM with tools bound.
Now the LLM uses the validated order_id from extracted_data rather
than whatever it decided to extract ad hoc.
"""
extracted = state.get("extracted_data", {})
order_id = extracted.get("order_id", "")
messages = (
[SystemMessage(content=(
"You are the Order Agent for ShopBot. Your only job is to look up "
"order information and report it clearly to the customer. "
f"The validated order ID is {order_id}. Use lookup_order to get the details."
))]
+ state["messages"]
)
response = order_llm.invoke(messages)
return {"messages": [response]}
def route_order_gate(state) -> str:
status = state.get("gate_status", "pending")
if status == "passed":
return "order_agent_node"
if state.get("retry_count", 0) >= 2:
return "order_agent_node" # correction_node already added the escalation message
return "correction_node"
def route_order_agent(state) -> str:
last = state["messages"][-1]
if hasattr(last, "tool_calls") and last.tool_calls:
return "order_tools"
return "__end__"
# ── Order subgraph assembly ──────────────────────────────────
order_builder = StateGraph(state_schema=type(None)) # Uses parent state directly
order_builder = StateGraph(dict)
order_builder.add_node("order_extract_node", order_extract_node)
order_builder.add_node("order_gate_node", order_gate_node)
order_builder.add_node("correction_node", correction_node)
order_builder.add_node("order_agent_node", order_agent_node)
order_builder.add_node("order_tools", order_tool_node)
order_builder.add_edge(START, "order_extract_node")
order_builder.add_edge("order_extract_node", "order_gate_node")
order_builder.add_conditional_edges(
"order_gate_node",
route_order_gate,
{"order_agent_node": "order_agent_node", "correction_node": "correction_node"}
)
order_builder.add_edge("correction_node", "order_gate_node")
order_builder.add_conditional_edges(
"order_agent_node",
route_order_agent,
{"order_tools": "order_tools", "__end__": END}
)
order_builder.add_edge("order_tools", "order_agent_node")
order_agent = order_builder.compile()
Tracez le boucle dans ce sous-graphe. Le nœud de porte d’ordre alimente le nœud de correction en cas d’échec, et le nœud de correction renvoie une information vers la porte. Ce cycle constitue la boucle de validation, qui s’exécute au plus deux fois avant que le compteur ne génère le message d’escalade.
L’agent de remboursement
Cet spécialiste est particulièrement important car il gère l’argent. Il met en œuvre deux portes, d’abord la validation de l’ID de commande, puis la validation complète du remboursement, et l’interruption liée à l’approbation humaine reste à l’intérieur de ce sous-graphe.
# ── agents.py (continued) ────────────────────────────────────
from gates import validate_order_id, validate_refund_request
from langgraph.types import interrupt, Command
from langchain_core.messages import ToolMessage
from datetime import datetime
@tool
def check_refund_eligibility(order_id: str) -> str:
"""Check whether an order is eligible for a refund.
An order is eligible only if its status is Delivered.
"""
order = ORDERS_DB.get(order_id.upper())
if not order:
return f"Cannot check refund: order '{order_id}' not found."
if order["status"] == "Delivered":
return (
f"Order {order_id.upper()} IS eligible for a refund. "
f"Amount: ${order['amount']:.2f}."
)
return (
f"Order {order_id.upper()} is NOT eligible. "
f"Status: {order['status']}."
)
@tool
def process_refund(order_id: str, reason: str) -> str:
"""Process a refund for a delivered order.
Only call this after human approval has been confirmed.
"""
order = ORDERS_DB.get(order_id.upper())
if not order:
return f"Refund failed: order '{order_id}' not found."
ref = f"REF-{order_id.upper()}-{datetime.now().strftime('%H%M%S')}"
return (
f"Refund APPROVED and PROCESSED. Reference: {ref}. "
f"${order['amount']:.2f} will be returned within 3 to 5 business days."
)
refund_tools = [check_refund_eligibility, process_refund]
refund_tool_node = ToolNode(refund_tools)
refund_llm = llm.bind_tools(refund_tools)
def refund_extract_node(state) -> dict:
structured_llm = llm.with_structured_output(RefundExtraction)
messages = (
[SystemMessage(content="Extract the order ID and refund reason from the customer's message.")]
+ state["messages"]
)
try:
result = structured_llm.invoke(messages)
return {
"extracted_data": result.model_dump(),
"gate_status": "pending",
"ticket_category": "refund_request",
}
except Exception as e:
return {
"gate_status": "failed",
"validation_errors": [f"Refund extraction failed: {str(e)}"],
"ticket_category": "refund_request",
}
def refund_gate_node(state) -> dict:
"""
Compound gate: runs both order ID validation and full refund validation.
Both must pass for execution to continue.
"""
extracted = state.get("extracted_data", {})
# Gate 1: Order ID
id_passed, id_result = validate_order_id(extracted)
if not id_passed:
return {"gate_status": "failed", "validation_errors": [id_result]}
# Gate 2: Full refund rules (eligibility, amount, reason)
ref_passed, ref_result = validate_refund_request(id_result)
if not ref_passed:
return {"gate_status": "failed", "validation_errors": [ref_result]}
return {"gate_status": "passed", "extracted_data": ref_result}
def refund_review_node(state) -> dict:
"""
The human approval gate, carried over from the original ShopBot
and now living inside the Refund Agent subgraph.
interrupt() pauses the entire graph here. The parent graph's
checkpointer saves the state. A human reviews the details.
Command(resume=...) restarts from this exact line.
"""
last_message = state["messages"][-1]
refund_tool_call = None
if hasattr(last_message, "tool_calls"):
for tc in last_message.tool_calls:
if "refund" in tc["name"].lower():
refund_tool_call = tc
break
extracted = state.get("extracted_data", {})
human_decision = interrupt({
"message": "Refund approval required",
"order_id": extracted.get("order_id", "Unknown"),
"reason": extracted.get("reason", "Not provided"),
"tool_being_called": refund_tool_call["name"] if refund_tool_call else "process_refund",
"arguments": refund_tool_call["args"] if refund_tool_call else extracted,
"summary": state.get("summary", "No summary yet"),
"options": ["approve", "reject", "escalate"],
})
if human_decision == "approve":
return {}
if human_decision in ("reject", "escalate"):
label = "declined" if human_decision == "reject" else "escalated to a senior agent"
return {
"messages": [
ToolMessage(
content=f"Refund request was {label} by the support team.",
tool_call_id=refund_tool_call["id"] if refund_tool_call else "unknown",
)
]
}
return {}
def refund_agent_node(state) -> dict:
extracted = state.get("extracted_data", {})
messages = (
[SystemMessage(content=(
"You are the Refund Agent for ShopBot. Your job is to check refund "
"eligibility and process approved refunds. "
f"Validated order ID: {extracted.get('order_id', '')}. "
f"Customer reason: {extracted.get('reason', '')}. "
"Always check eligibility first, then process if eligible and approved."
))]
+ state["messages"]
)
response = refund_llm.invoke(messages)
return {"messages": [response]}
def route_refund_gate(state) -> str:
status = state.get("gate_status", "pending")
if status == "passed":
return "refund_agent_node"
if state.get("retry_count", 0) >= 2:
return "refund_agent_node"
return "correction_node"
def route_refund_agent(state) -> str:
last = state["messages"][-1]
if hasattr(last, "tool_calls") and last.tool_calls:
tool_names = [tc["name"] for tc in last.tool_calls]
if "process_refund" in tool_names:
return "refund_review_node"
return "refund_tools"
return "__end__"
def route_after_review(state) -> str:
last = state["messages"][-1]
if isinstance(last, ToolMessage):
return "refund_agent_node"
return "refund_tools"
# ── Refund subgraph assembly ─────────────────────────────────
refund_builder = StateGraph(dict)
refund_builder.add_node("refund_extract_node", refund_extract_node)
refund_builder.add_node("refund_gate_node", refund_gate_node)
refund_builder.add_node("correction_node", correction_node)
refund_builder.add_node("refund_agent_node", refund_agent_node)
refund_builder.add_node("refund_review_node", refund_review_node)
refund_builder.add_node("refund_tools", refund_tool_node)
refund_builder.add_edge(START, "refund_extract_node")
refund_builder.add_edge("refund_extract_node", "refund_gate_node")
refund_builder.add_conditional_edges(
"refund_gate_node",
route_refund_gate,
{"refund_agent_node": "refund_agent_node", "correction_node": "correction_node"}
)
refund_builder.add_edge("correction_node", "refund_gate_node")
refund_builder.add_conditional_edges(
"refund_agent_node",
route_refund_agent,
{
"refund_review_node": "refund_review_node",
"refund_tools": "refund_tools",
"__end__": END,
}
)
refund_builder.add_conditional_edges(
"refund_review_node",
route_after_review,
{"refund_tools": "refund_tools", "refund_agent_node": "refund_agent_node"}
)
refund_builder.add_edge("refund_tools", "refund_agent_node")
refund_agent = refund_builder.compile()
L’agent chargé des réclamations
C’est le spécialiste le moins complexe. Il prend acte de la frustration du client, confirme qu’il y a suffisamment de détails pour ouvrir un ticket, et le transmet à un humain via interrupt() lorsque l’émotion exprimée est forte.
# ── agents.py (continued) ────────────────────────────────────
from gates import validate_complaint
complaints_llm = llm # Complaints agent doesn't need tool-calling
def complaints_extract_node(state) -> dict:
structured_llm = llm.with_structured_output(ComplaintExtraction)
messages = (
[SystemMessage(content=(
"Extract the complaint summary and the customer's emotional sentiment "
"from the conversation. Sentiment must be one of: frustrated, angry, "
"neutral, disappointed."
))]
+ state["messages"]
)
try:
result = structured_llm.invoke(messages)
return {
"extracted_data": result.model_dump(),
"gate_status": "pending",
"ticket_category": "complaint",
}
except Exception as e:
return {
"gate_status": "failed",
"validation_errors": [f"Complaint extraction failed: {str(e)}"],
"ticket_category": "complaint",
}
def complaints_gate_node(state) -> dict:
extracted = state.get("extracted_data", {})
passed, result = validate_complaint(extracted)
if passed:
return {"gate_status": "passed", "extracted_data": result}
return {"gate_status": "failed", "validation_errors": [result]}
def complaints_respond_node(state) -> dict:
"""
For high-sentiment complaints (frustrated, angry), trigger a human
escalation via interrupt(). For lower-sentiment complaints, the LLM
responds empathetically and the turn ends.
This mirrors the Part 3 pattern but applied conditionally based on
the validated sentiment field, not a keyword scan of free-form text.
The gate already confirmed the sentiment is a valid value, so this
routing decision is deterministic.
"""
extracted = state.get("extracted_data", {})
sentiment = extracted.get("sentiment", "neutral")
if sentiment in ("frustrated", "angry"):
human_decision = interrupt({
"message": "High-sentiment complaint requires human review",
"issue_summary": extracted.get("issue_summary", ""),
"sentiment": sentiment,
"options": ["escalate_to_senior", "provide_compensation", "close_ticket"],
})
if human_decision == "escalate_to_senior":
response_text = (
"I completely understand your frustration, and I want to make sure "
"you receive the best possible support. I am escalating your case to "
"a senior support agent who will contact you within 2 business hours. "
"Your ticket has been flagged as priority."
)
elif human_decision == "provide_compensation":
response_text = (
"I sincerely apologise for the experience you have had. As a gesture "
"of goodwill, our team will be in touch within 24 hours with a resolution "
"that includes appropriate compensation for the inconvenience caused."
)
else:
response_text = (
"Thank you for bringing this to our attention. Your feedback has been "
"recorded and will help us improve our service."
)
from langchain_core.messages import AIMessage
return {"messages": [AIMessage(content=response_text)]}
# Lower-sentiment complaints: LLM responds empathetically
messages = (
[SystemMessage(content=(
"You are the Complaints Agent for ShopBot. The customer has a complaint. "
"Respond with genuine empathy, acknowledge their specific issue, and "
"explain what steps will be taken. Be warm and professional."
))]
+ state["messages"]
)
response = complaints_llm.invoke(messages)
return {"messages": [response]}
def route_complaints_gate(state) -> str:
status = state.get("gate_status", "pending")
if status == "passed":
return "complaints_respond_node"
if state.get("retry_count", 0) >= 2:
return "complaints_respond_node"
return "correction_node"
# ── Complaints subgraph assembly ─────────────────────────────
complaints_builder = StateGraph(dict)
complaints_builder.add_node("complaints_extract_node", complaints_extract_node)
complaints_builder.add_node("complaints_gate_node", complaints_gate_node)
complaints_builder.add_node("correction_node", correction_node)
complaints_builder.add_node("complaints_respond_node", complaints_respond_node)
complaints_builder.add_edge(START, "complaints_extract_node")
complaints_builder.add_edge("complaints_extract_node", "complaints_gate_node")
complaints_builder.add_conditional_edges(
"complaints_gate_node",
route_complaints_gate,
{
"complaints_respond_node": "complaints_respond_node",
"correction_node": "correction_node",
}
)
complaints_builder.add_edge("correction_node", "complaints_gate_node")
complaints_builder.add_edge("complaints_respond_node", END)
complaints_agent = complaints_builder.compile()
Le graphe des superviseurs
Une fois les sous-graphes compilés, le graphe des superviseurs s’avère remarquablement simple. Il se contente de diriger les flux. Il ne réalise jamais de tâches spécialisées. Le principe à l’origine des modèles multi-agents est précisément celui-ci : les superviseurs dirigent et les spécialistes exécutent.
# ============================================================
# shopbot.py — Supervisor Graph (Enterprise Version)
# ============================================================
# ── MODULE 1: IMPORTS & CONFIGURATION ───────────────────────
import os
from typing import Literal
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, RemoveMessage, AIMessage
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command
# Import the compiled subgraphs from agents.py
from agents import order_agent, refund_agent, complaints_agent
from state import SupportState # The extended state from Module 2 above
load_dotenv()
# The supervisor uses a plain LLM - no tools, just routing decisions
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# ── MODULE 4: SUPERVISOR NODE ────────────────────────────────
SUPERVISOR_SYSTEM = """You are the ShopBot Supervisor. Your only job is to read
the customer's message and decide which specialist should handle it.
Available specialists:
order_agent - handles order status, delivery date, and product questions
refund_agent - handles refund requests and return eligibility
complaints_agent - handles complaints, frustration, and negative feedback
FINISH - use this only when the conversation is fully resolved
Respond with exactly one word: the name of the specialist, or FINISH."""
def supervisor_node(state: SupportState) -> dict:
"""
Reads the full conversation and routes to the correct specialist.
Returns a next_agent string that the routing function reads.
Uses plain LLM with no tools - the supervisor decides, it never acts.
This separation keeps the routing logic deterministic and auditable.
"""
messages = [SystemMessage(content=SUPERVISOR_SYSTEM)] + state["messages"]
response = llm.invoke(messages)
decision = response.content.strip().lower()
# Normalise the response in case the LLM adds punctuation or casing
if "refund" in decision:
next_agent = "refund_agent"
elif "complaint" in decision or "frustrated" in decision:
next_agent = "complaints_agent"
elif "order" in decision:
next_agent = "order_agent"
elif "finish" in decision:
next_agent = "FINISH"
else:
next_agent = "order_agent" # Safe default for ambiguous routing
return {"next_agent": next_agent}
# ── MODULE 4 CONTINUED: SUMMARIZE NODE ───────────────────────
# Carried over from the original ShopBot without changes.
# The enterprise version still needs conversation compression.
def summarize_node(state: SupportState) -> dict:
existing_summary = state.get("summary", "")
instruction = (
f"Current summary:\n{existing_summary}\n\nExtend with new messages. "
"Under 5 sentences. Include customer name, issue, orders mentioned, actions taken."
if existing_summary
else "Summarise this support conversation in under 5 sentences."
)
messages = state["messages"] + [HumanMessage(content=instruction)]
response = llm.invoke(messages)
to_delete = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
return {"summary": response.content, "messages": to_delete}
# ── MODULE 5: ROUTING ────────────────────────────────────────
def route_supervisor(state: SupportState) -> Literal[
"order_agent", "refund_agent", "complaints_agent", "summarize_node", "__end__"
]:
"""
Reads next_agent from state and routes accordingly.
Also checks whether the conversation is long enough to need summarisation.
"""
next_agent = state.get("next_agent", "order_agent")
if next_agent == "FINISH":
if len(state["messages"]) > 6:
return "summarize_node"
return "__end__"
return next_agent
# ── MODULE 6: GRAPH ASSEMBLY ─────────────────────────────────
graph_builder = StateGraph(SupportState)
graph_builder.add_node("supervisor", supervisor_node)
graph_builder.add_node("order_agent", order_agent) # compiled subgraph
graph_builder.add_node("refund_agent", refund_agent) # compiled subgraph
graph_builder.add_node("complaints_agent", complaints_agent) # compiled subgraph
graph_builder.add_node("summarize_node", summarize_node)
graph_builder.add_edge(START, "supervisor")
graph_builder.add_conditional_edges(
"supervisor",
route_supervisor,
{
"order_agent": "order_agent",
"refund_agent": "refund_agent",
"complaints_agent": "complaints_agent",
"summarize_node": "summarize_node",
"__end__": END,
}
)
# After each specialist finishes, always return to supervisor.
# The supervisor re-evaluates and decides: another specialist, or FINISH.
graph_builder.add_edge("order_agent", "supervisor")
graph_builder.add_edge("refund_agent", "supervisor")
graph_builder.add_edge("complaints_agent", "supervisor")
graph_builder.add_edge("summarize_node", END)
memory = MemorySaver()
shopbot = graph_builder.compile(checkpointer=memory)
# ── MODULE 7: ENTRYPOINT ──────────────────────────────────────
def run_shopbot():
print("=" * 60)
print(" ShopBot Enterprise - Multi-Agent Customer Support")
print("=" * 60)
print("Type your message. Type 'exit' to quit. Type 'audit' to")
print("see the validation error log for this session.\n")
config = {"configurable": {"thread_id": "enterprise-session-001"}}
while True:
user_input = input("You: ").strip()
if not user_input:
continue
if user_input.lower() == "exit":
print("ShopBot: Thank you for contacting ShopBot. Have a great day!")
break
if user_input.lower() == "audit":
snapshot = shopbot.get_state(config)
errors = snapshot.values.get("validation_errors", [])
print(f"\n[AUDIT LOG] {len(errors)} validation event(s) this session:")
for i, err in enumerate(errors, 1):
print(f" {i}. {err}")
print()
continue
result = shopbot.invoke(
{"messages": [HumanMessage(content=user_input)]},
config=config,
)
while "__interrupt__" in result:
interrupt_data = result["__interrupt__"][0].value
print("\n" + "=" * 60)
print(" HUMAN REVIEW REQUIRED")
print("=" * 60)
for key, value in interrupt_data.items():
if key != "options":
print(f" {key.replace('_', ' ').title()}: {value}")
options = interrupt_data.get("options", [])
print(f"\n Options: {' | '.join(f'[{o[0]}] {o}' for o in options)}")
raw = input(" Your decision: ").strip().lower()
chosen = next((o for o in options if o.startswith(raw)), options[-1])
result = shopbot.invoke(Command(resume=chosen), config=config)
last = result["messages"][-1]
print(f"\nShopBot: {last.content}\n")
if __name__ == "__main__":
run_shopbot()
Quelles modifications du comportement
Tant la version de base que la version entreprise constituent des implémentations valables des modèles standard LangGraph, mais elles reposent sur des hypothèses différentes, et cette différence devient importante dès que l’on quitte l’environnement de test pour un système ayant des conséquences réelles.
L’agent de base partait du principe que le modèle avait raison. Si le modèle extrayait « ORD-0O1 », avec une lettre O à la place d’un zéro, l’outil produirait une erreur, l’agent se remettrait probablement en marche, et le client pourrait recevoir une réponse légèrement confuse. Cela devrait généralement être acceptable.
L’agent d’entreprise part du principe que le modèle peut parfois se tromper et met en place une voie de récupération avant que cette erreur n’affecte des éléments externes. Le contrôleur intercepte immédiatement « ORD-0O1 » et génère un message d’erreur indiquant précisément ce qui n’allait pas et pourquoi. Le nœud de correction renvoie ce message exact au modèle, qui peut ainsi tenter à nouveau en suivant des instructions précises. En cas d’échec répété, la conversation est transférée à un humain. Aucun outil n’a fonctionné avec des données erronées, et aucun système externe n’a jamais vu cette hallucination.
Grâce à validation_errors dans l’état, un ingénieur senior qui examine une session ayant donné une réponse incorrecte dispose d’un journal complet de chaque échec de contrôle, de chaque tentative de correction et du nombre de réessais. L’échec peut être reproduit et compris, ce qui permet soit de corriger la demande initiale, soit de renforcer les contrôles. Sans un tel journal, déboguer un système probabiliste revient à deviner.
La structure à trois fichiers présente également des avantages pour les auditeurs. Lorsqu’on demande ce qui valide un remboursement avant que l’argent ne soit transféré, la réponse est un fichier en Python pur, sans architecture d’agent à expliquer : des règles métier claires et auditables, rédigées dans un langage que tout développeur peut comprendre.
Glossaire des nouveaux éléments
- Porte de logique métier : du Python simple qui accepte des données extraites et renvoie soit
(True, cleaned_data)soit(False, error_reason). Il dispose de son propre fichier, sans importations de LangGraph, et s’exécute dans un nœud avant toute appel d’outil. with_structured_output(PydanticModel): fait en sorte que le modèle renvoie des données conformes à un schéma Pydantic et non du texte libre. Le schéma est vérifié avant même que la porte ne s’exécute, de sorte que la sortie structurée, associée à une porte de validation, offre deux niveaux de contrôle.
extracted_data: dict : le champ d’état où se trouve la sortie structurée du modèle avant que les portes de contrôle ne l’examinent. Les outils lisent à cet endroit et jamais dans la sortie brute du modèle.validation_errors: Annotated[list[str], operator.add] : le journal d’audit. Chaque échec d’une porte de contrôle y est ajouté, et il s’accumule au cours de toute la conversation.retry_count: int : mécanisme de protection contre les boucles infinies. Le nœud de correction l’incrémente, la fonction de routage de la porte de contrôle le vérifie, et dépasser la valeur maximale entraîne une escalade sans nouvelle tentative.correction_node : le nœud de récupération. Il lit la dernière erreur de validation, crée une instruction de correction spécifique, réappelle le modèle avec une sortie structurée et renvoie vers la porte de contrôle. Les trois spécialistes y ont accès.gate_status: str : le canal entre les nœuds de contrôle et les fonctions de routage, avec les valeurs « pending », « passed », « failed », « escalated » et « aborted ». Les fonctions de routage le lisent au lieu d’analyser la sortie du modèle.supervisor_node : lit l’état de la conversation et renvoie le nom d’un spécialiste. Il n’appelle jamais d’outils et n’agit jamais.graph_builder.add_node("name", compiled_subgraph). Le nœud parent perçoit une boîte noire, tandis que le sous-graphe gère lui-même son routage, sa logique de contrôle et les appels aux outils.Compléter les lacunes
Les aspects difficiles du travail des agents d’entreprise se trouvent ailleurs que dans le premier layout graphique : la gestion déterministe des erreurs, la validation des états à grande échelle, ainsi qu’une couche d’orchestration qui n’est pas liée à un seul fournisseur de modèles. Toutes ces questions trouvent une réponse ici. La validation est déterministe : une panne déclenche un cycle de correction plutôt qu’une panne totale. Chaque décision peut être examinée grâce au journal d’audit. De plus, rien dans la couche d’orchestration – que ce soient les points de contrôle, les fonctions de routage, la logique de supervision ou le nœud de correction – ne fait appel à des éléments spécifiques à un fournisseur. Modifiez le fournisseur configuré de ChatOpenAI en ChatAnthropic, ou en tout autre fournisseur pris en charge par init_chat_model, et chaque point de contrôle, chaque route et chaque règle métier fonctionneront exactement comme avant. Le modèle est un module interchangeable, mais pas l’architecture.
Lectures complémentaires
- Tri des demandes de support multijoueur avec LangGraph et Ollama — Créez un graphe de tri des demandes utilisant trois agents avec LangGraph : un classificateur, un répondant et un contrôleur humain, fonctionnant sur Ollama local ou via une API, avec une seule ligne modifiée.
- Création d’un agent de documentation auto-correction avec LangGraph et Ollama — Un pipeline de réflexion local où un modèle rédige les documents API en Python, un deuxième les critique et un troisième concilie les résultats, connectés en boucle LangGraph sur Ollama.