Validierungstore und Mehr-Agenten-Routing für einen LangGraph-Support-Bot
Fügen Sie deterministische Geschäftslogik-Tore, einen selbstkorrigierenden Validierungszyklus sowie einen Supervisor mit drei spezialisierten Untergraphen zu einem E-Commerce-LangGraph-Agenten hinzu.
Stellen Sie sich einen Kundenservice-Bot für einen Online-Shop vor, hier ShopBot genannt, der auf LangGraph basiert. Er sucht Bestellungen nach, prüft die Berechtigung für Rückerstattungen, bearbeitet diese nach der Freigabe durch einen Menschen und hält die Konversation zusammengefasst bereit. Eine grundlegende Version dieses Bots funktioniert bereits. Im Folgenden werden drei Elemente hinzugefügt, die ein Bot mit tatsächlichen Auswirkungen benötigt: Validierungsmechanismen für jede Modellausgabe, eine Schleife, die das Modell automatisch korrigiert, sowie eine mehragentige Architektur, in der ein Supervisor Aufgaben an Spezialisten weiterleitet. Das Ziel ist ein Codebasis, den man ohne Zögern einer Produktivprüfung unterziehen könnte.
Die zugrunde liegende Basis ist ein Graph mit den üblichen Modulen: Werkzeuge, eine ReAct-ähnliche Schleife, Zustände sowie ein interrupt()-Aufruf, der den Graphen zur menschlichen Freigabe pausiert.
Die stillschweigende Annahme im grundlegenden Bot
Die einfache Version verlässt sich überall auf das Sprachmodell. Ihr Agent ruft eine Eignungsprüfung auf und ist überzeugt, dass die übergebene Bestellnummer korrekt aus dem Chat extrahiert wurde. Der Überprüfungsmodul zeigt diese Bestellnummer einem Menschen an, doch der Prüfer genehmigt oder lehnt eine Aktion ab, ohne zu überprüfen, ob die Nummer tatsächlich existiert. Das Rückerstattungstool wird mit den Argumenten ausgeführt, die das Modell ausgewählt hat. Wenn das Modell fälschlicherweise „ORD-99X“ generiert, obwohl eigentlich „ORD-001“ gemeint war, tritt ein Fehler auf, der Agent sieht diesen Fehler, und es könnte etwas Vernünftiges geschehen. „Es könnte etwas Vernünftiges geschehen“ ist eine schwache Beschreibung der Fehlerbehandlung in einem System, das Geld bewegt.
Praktiker, die automatisierte Beschaffungs- oder Konformitätsworkflows betreiben, betonen unmissverständlich: Wahrscheinlichkeitsbasierte Fehler sind dort nicht akzeptabel. Die Lösung besteht nicht darin, dem Modell mehr zu vertrauen. Vielmehr muss man dem Ergebnis des Modells ganz entsagen und jedes strukturierte Ergebnis vor dem Kontakt mit externen Systemen mit deterministischem Python überprüfen.
Drei Ergänzungen und wie sie zusammenpassen
Business-Logik-Türen befinden sich innerhalb der Knoten und validieren das Modellergebnis, bevor darauf reagiert wird. Es handelt sich dabei um reines Python ohne Aufrufe des Modells. Eine Logik-Tür lässt die Daten entweder durch oder lehnt sie mit einem spezifischen, strukturierten Grund ab. Das ist eine deterministische Validierung innerhalb eines probabilistischen Systems.
Validierungszyklen kümmern sich um Fehler bei den Schaltstellen. Eine Ablehnung führt weder zum Absturz des Graphen noch zu einer stillschweigenden Bereitstellung einer falschen Antwort. Der Kontrollfluss wechselt zu einem Korrekturknoten, der den Fehlergrund an das Modell weiterleitet und eine korrigierte Ausgabe anfordert. Ein Wiederholungszähler im Zustandsmanagement verhindert endlose Schleifen, und eine Obergrenze löst eine Eskalation an einen Menschen aus.
Eine Mehr-Agenten-Architektur ersetzt den einzelnen Agentenknoten durch drei Spezialisten. Ein Bestellagent kümmert sich um Abfragen und Statusanfragen. Ein Rückerstattungsagent übernimmt die Überprüfung der Berechtigung und den Prozess der Rückerstattung. Ein Beschwerdemanager kümmert sich um unzufriedene Kunden und Eskalationen. Ein Überwachungsknoten prüft jede eingehende Nachricht und leitet sie an den richtigen Spezialisten weiter, wobei jeder Spezialist ein selbstständiger, kompiliierter Untergraph ist.
Diese Komponenten arbeiten als ein System zusammen. Der Refund Agent enthält eine Schleuse; eine fehlerhafte Schleuse kann einen Validierungszyklus auslösen, und der Supervisor prüft, wie dieser Zyklus endete, bevor er entscheidet, ob er es erneut versucht, die Situation weiterverfolgt oder abbricht. Zusammen verleihen sie dem fertigen Bot die Robustheit, die eine Unternehmensumgebung erfordert.
Planen Sie das Diagramm, bevor Sie Code anfassen
LangGraph belohnt Planung – und je größer das Diagramm ist, desto mehr gilt das. Skizzieren Sie zunächst das gesamte System. Jedes Feld sollte genau einem Knoten oder einer Routenfunktion entsprechen und jede Pfeilspitze genau einer Kante. Der Code wird anschließend als Übersetzung dieses Diagramms in Python, Modul für Modul, erstellt.
Die neue Dateistruktur
Der grundlegende Bot passte in eine einzige Datei. Bei dieser Größe wird eine Datei jedoch unhandlich, weshalb das Projekt in drei Teile aufgeteilt wird.
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)
Es wird absichtlich darauf geachtet, dass die LangGraph-Importe nicht im Gates-Datei enthalten sind. So können die Gates mit gewöhnlichen Unit-Tests getestet werden: Es ist weder ein Modell, noch ein Graph und noch ein laufender Prozess erforderlich, um zu überprüfen, ob die Validierungslogik korrekt ist. Man ruft ein Gate mit Beispieldaten auf und überprüft das Ergebnis – das ist im Grunde das, was „deterministisch“ in der Praxis bedeutet.
Erweiterung des Zustandsschemas
Bevor etwas anderes hinzugefügt wird, benötigt der Zustand neue Felder. Die Ausgangsversion enthielt drei benutzerdefinierte Felder: eine Zusammenfassung, den Namen des Kunden und eine Ticketkategorie. Nun werden vier weitere hinzugefügt.
# ============================================================
# 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
Das wichtigste neue Feld ist extracted_data. Im grundlegenden Bot erzeugte das Modell freiformigen Text und rief die Tools direkt auf, wodurch die Argumente der Tools genau das waren, was das Modell in seine Aufrufe einfügte. Nun gibt das Modell zunächst strukturierte Daten aus, die in extracted_data gespeichert werden. Die Schleuse überprüft diese Daten, und erst nachdem die Überprüfung erfolgreich war, werden die Daten zu Argumenten für die Tools. Dadurch wird jede direkte Verbindung vom Modell zu externen Systemen beseitigt.
Das Feld validation_errors ist mit operator.add markiert, wodurch Fehler über den gesamten Dialog hinweg akkumuliert werden und nicht nur für einen einzelnen Schritt. Angenommen, in Schritt drei tritt eine erfundene Bestellnummer auf und in Schritt sieben ein unzureichender Rückgabegrund: Beide Fehler landen im Prüfprotokoll. Das ist bei Compliance-Überprüfungen von Bedeutung.
Die Schleusen schreiben
Eine Gate-Funktion ist eine reine Validierungsfunktion. Sie nimmt extrahierte Daten entgegen und gibt ein Tupel zurück: (True, cleaned_data) bei Erfolg oder (False, error_reason) bei Misserfolg. Das ist die gesamte Schnittstelle – ohne Aufrufe von Modellen und ohne Abhängigkeiten von Graphen. Der untenstehende Modul enthält eine Überprüfung der Bestellnummer, eine Überprüfung auf vollständige Rückerstattung sowie eine Überprüfung von Beschwerden, zusätzlich zu den dabei verwendeten Geschäftskonstanten.
# ============================================================
# 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
Betrachten Sie, was dem Datei fehlt: Importe aus LangGraph, LangChain oder irgendeiner AI-Bibliothek. Jeder Entwickler im Team kann sie lesen, testen und ändern, ohne etwas über die Agent-Architektur zu wissen. Wenn ein Prüfer fragt, was genau eine Rückerstattungsanfrage vor dem Geldtransfer validiert, ist dies die Datei, die gezeigt werden sollte – und auch Nicht-Entwickler können sie verstehen.
Der Validierungszyklus
Der Korrekturknoten ist es, der Tore nützlich macht. Wenn ein Tor fehlschlägt, stoppt die Ausführung nicht, sondern wird hier weitergeleitet. Der Knoten nimmt den spezifischen Fehlergrund aus dem Zustand, erstellt eine Anweisung, die dem Modell genau mitteilt, was schiefgelaufen ist, und bittet um eine korrigierte, strukturierte Ausgabe.
# ── 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)}"],
}
Die Wiederholungsprüfung ganz oben ist das wichtigste Sicherheitsmechanismus im gesamten System. Ohne sie würde eine wirklich fehlerhafte Eingabe (ein Kunde, der keine gültige Bestellnummer liefert, oder eine Modellkonfiguration, die ständig Fehlinformationen erzeugt) endlos im Kreis laufen und zwischen dem Korrekturknoten und dem Schaltkreis hin- und herwechseln. Zwei Wiederholungsversuche bilden einen sinnvollen Standard für die Support-Arbeit, da sie dem Modell zwei Chancen geben, sich selbst zu korrigieren, bevor ein Mensch eingreift. Die Anzahl kann je nach Toleranzgrenze zwischen Latenz und Genauigkeit angepasst werden.
Die drei spezialisierten Untergraphen
Jeder Spezialist ist ein kompiliierter Untergraph mit eigenen Werkzeugen, Extraktionslogik, Schaltkreisen und internen Routings. Der übergeordnete Überwachungsgraph kennt diese interne Struktur weder noch kümmert er sich darum. Er sieht lediglich einen Knoten, der Zustände empfängt und zurückgibt.
Der Bestellagent
# ── 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()
Verfolgen Sie den Kreislauf in diesem Untergraphen. Der Order-Gate-Node gibt bei Fehlern Daten an den Korrektur-Node weiter, und dieser gibt die Daten wieder an den Gate-Node zurück. Dieser Kreislauf ist der Validierungszyklus, und er läuft höchstens zweimal, bevor der Zähler die Eskalierungsmitteilung erzeugt.
Der Rückerstattungsagent
Dieser Spezialist ist am wichtigsten, weil er mit Geld umgeht. Er führt zwei Gate-Verfahren durch – zunächst die Validierung der Bestellnummer und anschließend die vollständige Rückerstattungsvalidierung – wobei die Unterbrechung durch menschliche Genehmigung innerhalb dieses Untergraphen bleibt.
# ── 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()
Der Beschwerdemanager
Das ist der am wenigsten komplizierte Spezialist. Er erkennt die Frustration des Kunden an, bestätigt, dass genügend Details vorliegen, um ein Ticket zu erstellen, und leitet es bei stark negativer Stimmung über interrupt() an einen Menschen weiter.
# ── 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()
Das Überwachungsgraph
Sobald die Untergraphen erstellt sind, erweist sich das Überwachungsgraph als äußerst einfach. Es leitet nur den Datenfluss weiter. Es übernimmt niemals spezialisierte Aufgaben. Das Prinzip hinter Mehr-Agenten-Strukturen besteht genau darin: Die Überwacher leiten und die Spezialisten führen aus.
# ============================================================
# 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()
Welche Verhaltensänderungen?
Sowohl die Basis- als auch die Enterprise-Version sind gültige Implementierungen der Standard-LangGraph-Muster, doch sie gehen von unterschiedlichen Annahmen aus, und dieser Unterschied wird relevant, sobald man den Sandbox für ein System mit tatsächlichen Konsequenzen verlässt.
Der Basis-Agent ging davon aus, dass das Modell richtig arbeitet. Wenn das Modell „ORD-0O1“ extrahierte, wobei ein Buchstabe O anstelle einer Null stand, würde das Tool einen Fehler melden, der Agent würde sich wahrscheinlich erholen, und der Kunde könnte eine etwas verwirrende Antwort erhalten. Wahrscheinlich kein Problem.
Der Unternehmensagent geht davon aus, dass das Modell manchmal falsch liegen kann, und erstellt daher einen Wiederherstellungsweg, bevor dieser Fehler nach außen dringen kann. Die Schleuse erfasst sofort „ORD-0O1“ und gibt eine Fehlermeldung aus, in der genau beschrieben wird, was falsch war und warum. Der Korrekturknoten leitet diese genaue Meldung zurück an das Modell, sodass dieses mit einer präzisen Anweisung einen zweiten Versuch unternehmen kann. Sollte es erneut zu einem Fehler kommen, wird das Gespräch an einen Menschen weitergeleitet. Kein Tool arbeitete mit fehlerhaften Daten, und kein externes System sah jemals die Halluzination.
Dank der validation_errors im Zustandsverlauf hat ein erfahrener Ingenieur, der eine Sitzung mit falscher Antwort überprüft, einen vollständigen Protokollauszug aller Schleusenfehler, Korrekturversuche sowie der Anzahl der Wiederholungsversuche. Der Fehler kann reproduziert und verstanden werden, woraufhin entweder die Anfrage angepasst oder die Schleuse verschärft wird. Ohne ein solches Protokoll bedeutet das Debuggen eines probabilistischen Systems nur Raten.
Die dreifach-Datei-Struktur bringt auch den Prüfern Vorteile. Auf die Frage, was eine Rückerstattung vor dem Geldtransfer validiert, lautet die Antwort eine reine Python-Datei ohne Agentenarchitektur, die zu erklären wäre: klare, prüfbare Geschäftsregeln in einer Sprache, die jeder Entwickler lesen kann.
Glossar der neuen Komponenten
- Geschäftslogik-Tor: reines Python, das extrahierte Daten entgegennimmt und entweder
(True, cleaned_data)oder(False, error_reason)zurückgibt. Es verfügt über seine eigene Datei ohne LangGraph-Importe und wird in einem Knoten vor jedem Toolaufruf ausgeführt. with_structured_output(PydanticModel): sorgt dafür, dass das Modell Daten zurückgibt, die einem Pydantic-Schema entsprechen, und nicht freien Text. Das Schema wird bereits vor dem Ausführen des Tores überprüft, sodass strukturierte Ausgabe zusammen mit dem Tor zwei Ebenen der Validierung bieten.
extracted_data: dict: Das Zustandsfeld, in dem die strukturierte Ausgabe des Modells gespeichert wird, bevor die Gate-Elemente sie prüfen. Tools lesen von hier aus und niemals direkt von der rohen Ausgabe des Modells.validation_errors: Annotated[list[str], operator.add]: Die Auditrückverfolgung. Jeder Fehler bei der Prüfung fügt sich hier hinzu, wodurch sich die Fehler über das gesamte Gespräch hinweg ansammeln.retry_count: int: Schutz vor endlosen Schleifen. Der Korrekturknoten erhöht diese Zahl, die Routing-Funktion der Gate-Elemente überprüft sie, und bei Überschreiten des Maximalwerts wird keine weitere An attempt unternommen, sondern die Situation eskaliert.correction_node: Der Wiederherstellungsknoten. Er liest den neuesten Validierungsfehler, erstellt eine spezifische Korrekturanweisung, ruft das Modell erneut mit der strukturierten Ausgabe auf und leitet den Prozess zurück an das Gate-Element. Alle drei Spezialisten teilen sich diesen Knoten.gate_status: str: Der Kanal zwischen Gate-Node und Routing-Funktionen, wobei die Werte „pending“, „passed“, „failed“, „escalated“ und „aborted“ gelten. Die Routing-Funktionen lesen diesen Wert anstelle der Analyse der Modellausgabe.supervisor_node: Lies den Gesprächszustand ein und gibt den Namen eines Spezialisten zurück. Er ruft niemals Tools auf und handelt nicht selbst.graph_builder.add_node("name", compiled_subgraph) hinzugefügt. Der übergeordnete Node sieht eine schwarze Box, während das Untergraphen seine eigene Routing-Logik sowie Toolaufrufe verarbeitet.Die Lücke schließen
Die schwierigen Aspekte der Arbeit mit Enterprise-Agenten liegen nicht im ersten Graphenlayout, sondern in der deterministischen Fehlerbehandlung, der Validierung des Zustands in großem Maßstab sowie in einer Orchestrierungsschicht, die nicht an einen bestimmten Modellanbieter gebunden ist. Für all diese Aspekte gibt es hier eine Lösung. Die Validierung ist deterministisch – ein Fehler löst einen Korrekturzyklus aus und nicht einen Absturz. Jede Entscheidung kann über die Audit-Trail überprüft werden. Zudem importiert die Orchestrierungsschicht, egal ob es sich um Gateways, Routing-Funktionen, Überwachungslogik oder Korrekturknoten handelt, nichts, was spezifisch für einen Anbieter ist. Ändern Sie den konfigurierten Anbieter von ChatOpenAI auf ChatAnthropic oder auf jeden anderen Anbieter, den init_chat_model unterstützt – jede Gateway-, Routing- und Geschäftsregel funktioniert weiterhin genauso wie zuvor. Das Modell ist ein einbaufähiges Modul, die Architektur hingegen nicht.
Verwandte Artikel
- Multi-Agenten-Support-Ticket-Triage mit LangGraph und Ollama — Erstellen Sie ein Ticket-Triage-Graph mit drei Agenten mithilfe von LangGraph: ein Klassifikator, ein Antwortgeber und eine menschliche Überprüfungsebene, die lokal auf Ollama oder über eine API mit nur einer geänderten Zeile läuft.
- Erstellung eines selbstkorrigierenden Dokumentations-Agenten mit LangGraph und Ollama — Ein lokaler Reflexionsprozess, bei dem ein Modell Python-API-Dokumente entwirft, ein zweites sie kritisiert und ein drittes das Ergebnis abgleicht, wobei alles als LangGraph-Loop auf Ollama verbunden ist.