Puertas de validación y enrutamiento multiagente para un bot de soporte LangGraph
Añadir puertas de lógica empresarial deterministas, un bucle de validación autocorregible y un supervisor con tres subgrafos especializados a un agente LangGraph para comercio electrónico.
Imagínese un bot de soporte al cliente para una tienda en línea, llamado aquí ShopBot, desarrollado con LangGraph. Este bot busca los pedidos, verifica la elegibilidad para reembolsos, procesa los reembolsos una vez que un humano da su aprobación y mantiene un resumen de la conversación. Ya existe una versión básica de este bot en funcionamiento. Lo que se describe a continuación añade tres elementos necesarios para un bot con consecuencias reales: puertas de validación para cada salida del modelo, un bucle que corrige automáticamente al modelo y una estructura de múltiples agentes en la que un supervisor asigna tareas a especialistas. El objetivo es contar con un conjunto de códigos listo para ser sometido a revisión en entorno de producción sin problemas.
La base de partida asumida es un grafo con los módulos habituales: herramientas, un bucle al estilo ReAct, estado y una llamada a interrupt() que pausa el grafo a la espera de la aprobación humana.
La suposición implícita en el bot básico
La versión simple confía en el modelo de lenguaje en todo momento. Su agente realiza una verificación de elegibilidad y asume que el ID del pedido que proporciona fue extraído correctamente del chat. El nodo de revisión muestra ese ID del pedido a un humano, pero el revisor aprueba o rechaza una acción sin verificar si el ID existe realmente. La herramienta de reembolso funciona con los argumentos que el modelo elija. Si el modelo genera erróneamente “ORD-99X” cuando debería ser “ORD-001”, la herramienta genera un error, el agente lo detecta y podría ocurrir algo razonable. “Podría ocurrir algo razonable” es una descripción insuficiente del manejo de errores en un sistema que maneja dinero.
Los profesionales que gestionan flujos de trabajo automatizados de adquisiciones o cumplimiento lo enfatizan claramente: las fallas probabilísticas son inaceptables en esos casos. La solución no consiste en confiar más en el modelo, sino en dejar de confiar por completo en sus resultados y verificar cada resultado estructurado con Python determinista antes de que interactúe con algo externo.
Tres adiciones y cómo se integran
Compuertas de lógica empresarial se encuentran dentro de los nodos y validan los resultados del modelo antes de que se tome cualquier acción al respecto. Están hechas con Python puro, sin llamadas al modelo. Una compuerta permite que los datos pasen o los rechaza con una razón específica y estructurada. Esa es la validación determinista dentro de un sistema probabilístico.
Bucles de validación se encargan de los fallos en las compuertas de control. Un rechazo no hace que el grafo se colapse ni genera silenciosamente una respuesta incorrecta. El control pasa a un nodo de corrección que transmite la razón del fallo al modelo y solicita una salida corregida. Un contador de intentos en el estado evita bucles infinitos, y un límite máximo provoca la escalada a un humano.
Una arquitectura multiagente reemplaza el nodo de agente único por tres especialistas. Un Agente de Pedidos se ocupa de las búsquedas y las consultas de estado. Un Agente de Reembolsos se encarga de las verificaciones de elegibilidad y el procesamiento de reembolsos. Un Agente de Quejas atiende a los clientes insatisfechos y gestiona las escaladas. Un nodo supervisor examina cada mensaje recibido y lo envía al especialista adecuado, y cada especialista es un subgrafo compilado por sí mismo.
Estas piezas funcionan como un sistema único. El Agente de Reembolsos maneja una compuerta; si esta falla, puede iniciarse un bucle de validación, y el supervisor analiza cómo terminó ese bucle antes de decidir si intentarlo de nuevo, escalar el problema o abortar. Juntos, brindan al bot final la solidez que exigen los entornos empresariales.
Planifique el grafo antes de tocar el código
LangGraph valora la planificación, y cuanto más grande sea el grafo, más cierto es esto. Dibuje primero todo el sistema: cada caja debe corresponder exactamente a un nodo o función de enrutamiento, y cada flecha a un arista. El código se convierte entonces en la traducción de esa representación a Python, módulo por módulo.
La nueva estructura de archivos
El bot básico cabía en un único archivo. Con este tamaño, un solo archivo se vuelve poco práctico, por lo que el proyecto se divide en tres.
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)
Se ha decidido intencionadamente mantener las importaciones de LangGraph fuera del archivo gates. Los gates se pueden probar con pruebas unitarias habituales: no se necesita ningún modelo, ningún grafo ni ningún proceso en ejecución para confirmar que la lógica de validación es correcta. Se llama a un gate con datos de ejemplo y se verifica el resultado; eso es lo que en la práctica significa “determinista”.
Ampliando el esquema de estado
El estado necesita nuevos campos antes que nada. La versión inicial contaba con tres campos personalizados: un resumen, el nombre del cliente y una categoría de ticket. Se añaden cuatro má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
El nuevo campo clave es extracted_data. En el bot básico, el modelo generaba texto en formato libre y llamaba directamente a las herramientas, por lo que los argumentos de estas eran lo que el modelo incluía en sus llamadas. Ahora, el modelo emite primero datos estructurados, que se almacenan en extracted_data. La compuerta de control los verifica, y solo después de que pase esa verificación los datos se convierten en argumentos de la herramienta. Esto elimina cualquier ruta directa desde el modelo hacia sistemas externos.
El campo validation_errors está marcado con operator.add, por lo que acumula errores a lo largo de toda la conversación y no solo en una sola interacción. Supongamos que en la tercera interacción aparece un ID de pedido ficticio y en la séptima una razón deficiente para el reembolso: ambos errores terminan en el registro de auditoría. Esto es importante en las revisiones de cumplimiento.
Redacción de las compuertas
Una puerta es una función de validación pura. Toma los datos extraídos y devuelve una tupla: (True, cleaned_data) en caso de éxito o (False, error_reason) en caso de fallo. Esa es toda la interfaz, sin llamadas a modelos ni dependencias de gráficos. El módulo a continuación incluye una verificación del ID del pedido, una verificación de reembolso completo y una verificación de quejas, además de las constantes empresariales que utiliza.
# ============================================================
# 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
Mire qué carece el archivo: importaciones de LangGraph, LangChain o cualquier biblioteca de IA. Cualquier desarrollador del equipo puede leerlo, probarlo y modificarlo sin necesidad de conocer nada sobre la arquitectura de los agentes. Cuando un auditor pregunta qué es exactamente lo que valida una solicitud de reembolso antes de que se transfiera el dinero, este es el archivo que se debe mostrar, y las personas que no son ingenieros pueden entenderlo.
El bucle de validación
El nodo de corrección es lo que hace útiles las compuertas. Cuando una compuerta falla, la ejecución no se detiene sino que se dirige aquí. El nodo toma la razón específica del fallo del estado, crea una instrucción que indica al modelo exactamente qué salió mal y solicita una salida estructurada corregida.
# ── 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 verificación de intentos de reintentos en la parte superior es el mecanismo de seguridad más importante de todo el sistema. Sin ella, una entrada realmente defectuosa (un cliente que no proporciona un ID de pedido válido, o una configuración del modelo que sigue generando errores) seguiría funcionando indefinidamente, alternando entre el nodo de corrección y la puerta de control. Dos intentos de reintento constituyen un valor por defecto razonable para el trabajo de soporte, ya que brindan al modelo dos oportunidades de corregirse antes de que intervenga un humano. Este número puede ajustarse según el grado de tolerancia entre la latencia y la precisión.
Los tres subgrafos especializados
Cada especialista es un subgrafo compilado con sus propias herramientas, lógica de extracción, puerta de control y enrutamiento interno. El grafo supervisor padre no conoce ni le importa esa estructura interna; solo ve un nodo que recibe estado y devuelve estado.
El Agente de Pedidos
# ── 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()
Trace el bucle en este subgráfico. El nodo de la puerta de pedido envía la información de corrección en caso de fallo, y el nodo de corrección envía datos de vuelta a la puerta. Ese ciclo constituye el bucle de validación, y se ejecuta como máximo dos veces antes de que el contador genere el mensaje de escalada.
El Agente de Reembolsos
Este especialista es el más importante porque se ocupa del dinero. Ejecuta dos procesos de validación: primero la validación del ID del pedido y luego la validación del reembolso completo, y la interrupción por aprobación humana permanece dentro de este subgráfico.
# ── 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()
El agente de quejas
Se trata del especialista menos complicado. Reconoce la frustración del cliente, confirma que hay suficientes detalles para abrir un ticket y lo transfiere a un humano mediante interrupt() cuando el sentimiento es muy intenso.
# ── 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()
El grafo de supervisores
Una vez compilados los subgrafos, el grafo de supervisores resulta ser sorprendentemente simple. Solo se encarga de enrutar. Nunca realiza tareas especializadas. El principio detrás de los patrones de múltiples agentes es exactamente este: los supervisores enrutan y los especialistas ejecutan.
# ============================================================
# 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()
Qué cambios en el comportamiento
Tanto la versión básica como la empresarial son implementaciones válidas de los patrones estándar de LangGraph, pero realizan suposiciones diferentes, y esa diferencia es importante una vez que se abandona el entorno de pruebas por un sistema con consecuencias reales.
El agente básico asumía que el modelo estaba correcto. Si el modelo extraía “ORD-0O1”, con una letra O en lugar de un cero, la herramienta daría un error, el agente probablemente se recuperaría, y el cliente podría recibir una respuesta ligeramente confusa. Probablemente no hay problema.
El agente empresarial asume que el modelo a veces cometerá errores y crea una vía de recuperación antes de que ese error llegue a entornos externos. La compuerta detecta inmediatamente a “ORD-0O1” y genera un mensaje de error que indica con precisión qué fue lo incorrecto y por qué. El nodo de corrección envía ese mensaje exacto de vuelta al modelo, quien tiene una segunda oportunidad con instrucciones precisas. Si se produce otro fallo, la conversación se transfiere a una persona. Ninguna herramienta funcionó con datos defectuosos, y ningún sistema externo llegó a ver la alucinación.
Gracias a validation_errors en el estado, un ingeniero senior que revisa una sesión que dio una respuesta incorrecta dispone de un registro completo de cada fallo de la compuerta, intento de corrección y número de reintentos. El fallo puede reproducirse y comprenderse, y luego se corrige el prompt o se refuerzan las reglas de la compuerta. Sin dicho registro, depurar un sistema probabilístico implica hacer conjeturas.
La estructura de tres archivos también resulta útil para los auditores. Al preguntar qué valida un reembolso antes de que se transfiera el dinero, la respuesta es un archivo de puro Python sin arquitectura de agentes que explique nada: reglas de negocio claras y auditables en un lenguaje que cualquier desarrollador puede leer.
Glosario de las nuevas componentes
- Business logic gate: Python puro que acepta datos extraídos y devuelve
(True, cleaned_data)o(False, error_reason). Tiene su propio archivo sin importaciones de LangGraph y se ejecuta en un nodo antes de cualquier llamada a herramientas. with_structured_output(PydanticModel): hace que el modelo devuelva datos que coincidan con un esquema Pydantic y no texto libre. El esquema se verifica incluso antes de que se ejecute la compuerta, por lo que la salida estructurada sumada a la compuerta proporciona dos niveles de validación.
extracted_data: dict: el campo de estado donde se almacena la salida estructurada del modelo antes de que las compuertas la examinen. Las herramientas leen desde aquí y nunca desde la salida bruta del modelo.validation_errors: Annotated[list[str], operator.add]: el registro de auditoría. Cada fallo de una compuerta se agrega a él, y se acumula a lo largo de toda la conversación.retry_count: int: protección contra bucles infinitos. El nodo de corrección lo incrementa, la función de enrutamiento de la compuerta lo verifica, y superar el máximo valor implica pasar a un nivel superior y no realizar otro intento.correction_node: el nodo de recuperación. Lee el último error de validación, crea una instrucción de corrección específica, vuelve a invocar al modelo con una salida estructurada y redirige la solicitud a la compuerta. Los tres especialistas lo comparten.gate_status: str: el canal entre los nodos de control y las funciones de enrutamiento, con valores “pending”, “passed”, “failed”, “escalated” y “aborted”. Las funciones de enrutamiento lo leen en lugar de analizar la salida del modelo.supervisor_node: lee el estado de la conversación y devuelve el nombre de un especialista. Nunca llama a herramientas ni toma ninguna acción.graph_builder.add_node("name", compiled_subgraph). El nodo padre muestra una caja negra, mientras que el subgráfico se encarga de su propio enrutamiento, lógica de control y llamadas a herramientas.Cerrando la brecha
Las partes más difíciles del trabajo con agentes empresariales se encuentran en otros aspectos además del primer diseño gráfico: el manejo determinista de errores, la validación de estados a gran escala y una capa de orquestación que no está vinculada a un único proveedor de modelos. Aquí se encuentra la respuesta para los tres. La validación es determinista: un fallo desencadena un ciclo de corrección y no una caída del sistema. Cada decisión puede ser revisada a través del registro de auditoría. Además, nada en la capa de orquestación, ya sean puertas de control, funciones de enrutamiento, lógica de supervisión o el nodo de corrección, importa elementos específicos de algún proveedor. Cambie el proveedor configurado de ChatOpenAI a ChatAnthropic, o a cualquier otro que soporte init_chat_model, y cada puerta de control, ruta y regla de negocio seguirá funcionando exactamente como antes. El modelo es un módulo conectable, pero la arquitectura no lo es.
Leer más
- Triaje de tiquetes de soporte multiejecutor con LangGraph y Ollama — Crea un grafo de triaje de tiquetes con tres ejecutores utilizando LangGraph: un clasificador, un respondedor y un control humano, que funcionan con Ollama local o a través de una API, con solo una línea modificada.
- Construir un agente de documentación autocorregible con LangGraph y Ollama — Un pipeline de reflexión local en el que un modelo redacta las documentaciones de la API en Python, otro las critica y un tercero armoniza el resultado, conectados en un bucle de LangGraph sobre Ollama.