Build ShopBot: Ein LangGraph-Support-Agent mit Speicher, Werkzeugen und Genehmigung für Rückerstattungen
Erstellen Sie einen Kundenservice-Agenten im LangGraph-Modul Schritt für Schritt: Zustände, Tools, Knoten, Routing, rollierende Zusammenfassungen sowie eine Unterbrechung, die auf die Freigabe der Rückerstattung wartet.
Quellen: Your First Real LangGraph Project: Building a Customer Support Agent.
Dieses Projekt integriert drei Konzepte von LangGraph in einen einzigen Agenten: eine Grapharchitektur, ein Gesprächsmemory sowie eine Genehmigung durch einen Menschen. Der Agent ist ein Kundenservice-Bot namens ShopBot für einen fiktiven Online-Shop. Kunden können Fragen zu Bestellungen stellen, Rückerstattungen beantragen und Beschwerden weiterleiten. Der Bot merkt sich das Gespräch, sucht mit echten Tools nach Bestelldaten und hält vor der Ausführung einer Rückerstattung eine Genehmigung durch einen Menschen ab. Der Code wird Modul für Modul entwickelt, wobei jedes Teil erläutert wird, und Vorkenntnisse von LangGraph sind nicht erforderlich.
Planen Sie zunächst das Graph
Ein häufiger Anfängerfehler besteht darin, eine leere Datei zu öffnen und direkt zu tippen. LangGraph belohnt es, den Graphen zunächst auf Papier oder im Kopf zu zeichnen. Auf Papier ist ShopBot einfach strukturiert: Der Agentenschritt, der Tool-Runner, die Rückerstattungsprüfung und der Zusammenfasser bilden die Knoten, zwei Pfade bestimmen den Ablauf, ein einzelner interrupt() veranlasst eine Pause zur Genehmigung, und ein Auslöser startet die Zusammenfassung. Der Code überträgt lediglich dieses Bild.
Einrichtung
Erstellen Sie eine Datei mit dem Namen shopbot.py oder öffnen Sie ein Colab-Notebook. Installieren Sie die Abhängigkeiten:
pip install langgraph langchain langchain-openai langgraph-checkpoint-sqlite python-dotenv
Erstellen Sie anschließend eine .env-Datei im selben Verzeichnis, um den Schlüssel zu speichern:
OPENAI_API_KEY=your-key-here
Modul 1: Importe und Konfiguration
Jeder Import für die gesamte Datei wird ganz oben platziert, es werden keine Importe mitten in der Datei vorgenommen. Die Konfiguration befindet sich ebenfalls hier.
# ============================================================
# MODULE 1: IMPORTS & CONFIGURATION
# ============================================================
import os
import sqlite3
from typing import Annotated, Literal
from datetime import datetime
from dotenv import load_dotenv
# LangChain - the AI layer
from langchain_openai import ChatOpenAI
from langchain_core.messages import (
HumanMessage,
AIMessage,
SystemMessage,
BaseMessage,
RemoveMessage,
)
from langchain_core.tools import tool
# LangGraph - the graph layer
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
load_dotenv()
# ── The LLM ─────────────────────────────────────────────────
# temperature=0 means deterministic - the agent behaves
# consistently, which is what you want for a support bot.
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
Die gewählte Modellversion ist gpt-4o-mini. Es ist leistungsfähig genug für Unterstützungsarbeiten, günstig genug, um es häufig nutzen zu können, und schnell genug, sodass die Nutzer keine Verzögerungen bemerken. Ein Produktionsystem könnte bei komplexeren Aufgaben auf gpt-4o wechseln, doch mit kleinen Schritten anzufangen ist sinnvoll.
Modul 2: Zustand
Der Zustand ist das Arbeitsgedächtnis des Agents, weshalb alles, was über das Graph strömt, darin gespeichert werden muss. Bestimmen Sie, was der Agent in jeder Phase wissen muss.
# ============================================================
# MODULE 2: STATE
# ============================================================
class SupportState(MessagesState):
# MessagesState already gives us:
# messages: Annotated[list[BaseMessage], add_messages]
# We add three more fields for our specific needs:
# The running summary of the conversation (Part 2 pattern).
# Starts empty. Gets written by summarize_node when conversation gets long.
summary: str
# The name of the customer, extracted early in the conversation.
# Used to personalise every response. Starts empty.
customer_name: str
# Tracks the current ticket category, set by the agent.
# Helps the human reviewer understand context during escalation.
# Values: "order_inquiry" | "refund_request" | "complaint" | "general"
ticket_category: str
Es gibt drei zusätzliche Felder neben der Nachrichtenhistorie. summary enthält eine komprimierte Aufzeichnung früherer Nachrichtenwechsel: Sobald ein Gespräch sechs Nachrichten überschreitet, fasst der Agent die Historie in dieses Feld zusammen und löscht die alten Rohnachrichten, damit die Tokenkosten konstant bleiben. customer_name ist ein praktisches Feld für fast jeden Support-Agenten. Mit dem Namen im Speicher kann der Agent die Nutzer persönlich begrüßen, ohne dass jedes Knotenpunkt das Chat-Verlauf erneut durchlesen muss, um ihn zu finden. ticket_category ermöglicht die bedingte Weiterleitung, da der Agent diesen Wert festlegt und eine Weiterleitungsfunktion ihn ausliest, um zu entscheiden, ob der Weg zur menschlichen Freigabe ausgelöst werden soll.
Modul 3: Tools
Tools sind die Hände des Agenten. Ohne sie kann er nur sprechen, mit ihnen kann er Informationen suchen und Aktionen ausführen. ShopBot verfügt über drei solcher Tools, unterstützt durch eine fingierte In-Memory-Datenbank:
# ============================================================
# MODULE 3: TOOLS
# ============================================================
# ── Fake Database ────────────────────────────────────────────
# In a real project, these would be database queries or API calls.
# For learning purposes, we use a simple Python dictionary.
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",
},
}
@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 provides an order number and wants
to know the status, product name, or delivery date of their order.
Args:
order_id: The order ID string, e.g. 'ORD-001'
Returns:
A formatted string with full order details, or an error message
if the order is not found.
"""
order = ORDERS_DB.get(order_id.upper())
if not order:
return f"No order found with ID '{order_id}'. Please double-check the order number."
return (
f"Order {order_id.upper()}: {order['product']} | "
f"Status: {order['status']} | "
f"Amount: ${order['amount']:.2f} | "
f"Delivery: {order['delivery_date']}"
)
@tool
def check_refund_eligibility(order_id: str) -> str:
"""Check whether an order is eligible for a refund.
Use this BEFORE processing any refund request. An order is eligible
for a refund only if its status is 'Delivered'. Orders Fthat are
'In Transit' or 'Processing' cannot be refunded yet.
Args:
order_id: The order ID string, e.g. 'ORD-001'
Returns:
A string stating whether the order is eligible and why.
"""
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"Product: {order['product']}, Amount: ${order['amount']:.2f}. "
f"Proceed to refund processing."
)
else:
return (
f"Order {order_id.upper()} is NOT eligible for a refund yet. "
f"Current status: {order['status']}. Refunds are only available "
f"for delivered orders."
)
@tool
def process_refund(order_id: str, reason: str) -> str:
"""Process a refund for a delivered order.
IMPORTANT: This tool actually issues the refund. It should only be
called AFTER human approval has been obtained. Never call this tool
without prior confirmation.
Args:
order_id: The order ID to refund
reason: The customer's stated reason for the refund
Returns:
A confirmation string with the refund reference number.
"""
order = ORDERS_DB.get(order_id.upper())
if not order:
return f"Refund failed: order '{order_id}' not found."
# In a real system, this would hit your payments API.
refund_ref = f"REF-{order_id.upper()}-{datetime.now().strftime('%H%M%S')}"
return (
f"Refund APPROVED and PROCESSED. Reference: {refund_ref}. "
f"${order['amount']:.2f} will be returned to the original payment method "
f"within 3–5 business days. Reason logged: '{reason}'."
)
# ── Collect tools and bind to LLM ───────────────────────────
# All three tools in one list.
tools = [lookup_order, check_refund_eligibility, process_refund]
# llm_with_tools = the LLM that KNOWS about the tools and can decide to call them.
# This is what we use inside agent_node.
llm_with_tools = llm.bind_tools(tools)
# tool_node = the pre-built node that EXECUTES whatever tool the LLM chose.
# This is what we register in Module 6.
tool_node = ToolNode(tools)
Betrachten Sie genau die Dokumentation von process_refund – darin steht, dass das Tool tatsächlich die Rückerstattung ausführt und nur nach menschlicher Freigabe aufgerufen werden sollte. Das Modell liest die Dokumentationen und verwendet sie bei der Entscheidung, was zu tun ist. Schreiben Sie sie, als würden Sie einem klugen Praktikanten, der Ihre Geschäftsregeln noch nicht kennt, Informationen geben, denn genau das geschieht in Wirklichkeit.
Modul 4: Knoten
Die Knoten erledigen die Arbeit. Jeder erhält den Zustand, führt eine Aktion aus und gibt ein Dictionary mit den Aktualisierungen zurück – nichts weiter. ShopBot verfügt über drei von Ihnen zu schreibende Knoten sowie den im vorherigen Modul bereits vorhandenen tool_node. Dazu gehören der Systemprompt, der Hauptagentenschritt, der Rückerstattungsüberprüfungsschritt mit interrupt() sowie der Zusammenfasser:
# ============================================================
# MODULE 4: NODES
# ============================================================
# ── The System Prompt ────────────────────────────────────────
# Written once, used in every call to the LLM from agent_node.
# This is the personality and rulebook of your agent.
SYSTEM_PROMPT = """You are ShopBot, a friendly and professional customer support \
agent for an e-commerce store.
Your capabilities:
- Look up order details using the lookup_order tool
- Check if an order qualifies for a refund using check_refund_eligibility
- Process approved refunds using the process_refund tool
Your rules:
- Always greet the customer by name once you know it
- Always check refund eligibility BEFORE attempting to process a refund
- For refund requests, set ticket_category to "refund_request" in your reasoning
- Be empathetic, clear, and concise
- If you cannot help, offer to escalate to a human agent
Important: The process_refund tool requires prior human approval. Do not call it \
unless the conversation shows that a human has already approved the refund."""
# ── Node 1: agent_node ──────────────────────────────────────
def agent_node(state: SupportState) -> dict:
"""The brain of the operation. Reads state, calls the LLM, and decides
whether to use a tool, give a final answer, or do something else.
This node handles two cases:
1. Normal conversation - just call the LLM and respond
2. Long conversation - if a summary exists, prepend it so the LLM
has context without seeing all the raw messages
"""
# Part 2 pattern: check for an existing summary
summary = state.get("summary", "")
if summary:
# Build context: system prompt + compressed history + recent messages
system_with_summary = SystemMessage(
content=f"{SYSTEM_PROMPT}\n\nSummary of conversation so far:\n{summary}"
)
messages_to_send = [system_with_summary] + state["messages"]
else:
# No summary yet - full history is short enough to send as-is
system_msg = SystemMessage(content=SYSTEM_PROMPT)
messages_to_send = [system_msg] + state["messages"]
# Call the LLM. It sees tools and can choose to call one.
response = llm_with_tools.invoke(messages_to_send)
# Detect ticket category from the response for routing purposes.
# A smarter version would have the LLM explicitly set this -
# for now, we scan for keywords.
content_lower = response.content.lower() if response.content else ""
updates: dict = {"messages": [response]}
if "refund" in content_lower or (
hasattr(response, "tool_calls")
and any("refund" in str(tc).lower() for tc in (response.tool_calls or []))
):
updates["ticket_category"] = "refund_request"
return updates
# ── Node 2: review_refund ───────────────────────────────────
def review_refund(state: SupportState) -> dict:
"""The human approval gate. Pauses execution, shows the pending refund
details to a human agent, and waits for their decision.
This implements the Part 3 interrupt() pattern. Execution stops here
until someone calls graph.invoke(Command(resume=...), config).
Three outcomes the human can choose:
- "approve" → let the refund tool call proceed unchanged
- "reject" → cancel the refund, send a message to the customer
- "escalate" → hand the entire ticket to a human support agent
"""
last_message = state["messages"][-1]
# Find the refund-related tool call in the last AI message.
# We look for process_refund specifically - the "real action" tool.
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
# Surface the context to the human reviewer via interrupt().
# Everything in this dict is what the human sees before deciding.
human_decision = interrupt({
"message": "⚠️ Refund approval required",
"customer_name": state.get("customer_name", "Unknown"),
"tool_being_called": refund_tool_call["name"] if refund_tool_call else "refund tool",
"arguments": refund_tool_call["args"] if refund_tool_call else {},
"conversation_summary": state.get("summary", "No summary yet"),
"options": ["approve", "reject", "escalate"],
})
# ── Handle the human's decision ─────────────────────────
if human_decision == "approve":
# Do nothing to state - let tool_node execute the tool call as-is
return {}
elif human_decision == "reject":
# Cancel the tool call. The LLM will see a ToolMessage explaining why,
# and generate a polite response to the customer.
from langchain_core.messages import ToolMessage
return {
"messages": [
ToolMessage(
content=(
"Refund request was reviewed and declined by our support team. "
"Please inform the customer politely and offer alternatives."
),
tool_call_id=refund_tool_call["id"] if refund_tool_call else "unknown",
)
]
}
elif human_decision == "escalate":
# Signal escalation - in a real system you'd open a ticket,
# ping Slack, or transfer to a live agent queue.
from langchain_core.messages import ToolMessage
return {
"messages": [
ToolMessage(
content=(
"This ticket has been escalated to a senior support agent. "
"Inform the customer that a human agent will contact them "
"within 2 business hours."
),
tool_call_id=refund_tool_call["id"] if refund_tool_call else "unknown",
)
]
}
# Fallback - treat as approve
return {}
# ── Node 3: summarize_node ──────────────────────────────────
def summarize_node(state: SupportState) -> dict:
"""Triggered when the conversation exceeds 6 messages. Compresses the
full message history into a short summary, then deletes old raw messages.
This is the rolling summary pattern from Part 2. The summary grows
richer turn by turn. Token costs stay nearly flat no matter how long
the conversation runs.
"""
existing_summary = state.get("summary", "")
if existing_summary:
# Extend the existing summary with new messages
summary_instruction = (
f"Current summary:\n{existing_summary}\n\n"
"Extend this summary with the new messages above. "
"Keep it under 5 sentences. Focus on: the customer's name, "
"their issue, any orders mentioned, and what actions were taken."
)
else:
# First time summarising
summary_instruction = (
"Summarise this customer support conversation in under 5 sentences. "
"Include: the customer's name (if mentioned), their issue, "
"any order numbers discussed, and what actions were taken so far."
)
messages = state["messages"] + [HumanMessage(content=summary_instruction)]
response = llm.invoke(messages) # Plain llm, no tools needed here
# Delete all but the 2 most recent messages.
# The summary now holds everything that was in the deleted messages.
messages_to_delete = [
RemoveMessage(id=m.id) for m in state["messages"][:-2]
]
return {
"summary": response.content,
"messages": messages_to_delete,
}
Modul 5: Kanten und Routing
Routing-Funktionen entscheiden, wohin es als Nächstes geht. Sie prüfen den Zustand und geben einen String zurück; sie ändern den Zustand niemals, da das die Aufgabe der Knoten ist. Es sind zwei erforderlich:
# ============================================================
# MODULE 5: EDGES & ROUTING
# ============================================================
def route_after_agent(state: SupportState) -> Literal[
"review_refund", "tools", "summarize_node", "__end__"
]:
"""Called after agent_node runs. Decides what happens next.
Four possible routes:
1. The LLM wants to call process_refund → must go through human review first
2. The LLM wants to call any other tool → go directly to tool_node
3. The LLM gave a plain text answer AND the conversation is long → summarise
4. The LLM gave a plain text answer and conversation is short → we're done
"""
last_message = state["messages"][-1]
has_tool_calls = hasattr(last_message, "tool_calls") and bool(last_message.tool_calls)
if has_tool_calls:
# Check if ANY of the tool calls is the sensitive process_refund tool
tool_names = [tc["name"] for tc in last_message.tool_calls]
if "process_refund" in tool_names:
return "review_refund" # → Pause for human approval first
return "tools" # → Safe tool, run it directly
# No tool call - the LLM gave a plain response.
# Check if the conversation is long enough to need summarisation.
if len(state["messages"]) > 6:
return "summarize_node"
return "__end__" # → Conversation turn is complete
def route_after_review(state: SupportState) -> Literal["tools", "agent_node"]:
"""Called after review_refund runs (i.e., after the human has decided).
Two routes:
1. Human approved or escalated → run the tool (tool_node handles the call)
2. Human rejected → the review node already added a ToolMessage cancelling
the tool call, so skip tool_node and go back to agent_node to respond
"""
last_message = state["messages"][-1]
# If the last message is a ToolMessage, the review node cancelled the call.
# Go back to agent_node so it can generate a customer-facing response.
from langchain_core.messages import ToolMessage
if isinstance(last_message, ToolMessage):
return "agent_node"
# Otherwise, the review node returned {} (approved) - proceed to tools.
return "tools"
Modul 6: Zusammenbau des Graphen
Sämtliche Komponenten sind nun vorhanden, und der Zusammenbau folgt stets fünf Schritten: Initialisierung, Registrierung der Knoten, Festlegung des Eingangspunkts, Hinzufügen von Kanten und Kompilierung.
# ============================================================
# MODULE 6: GRAPH ASSEMBLY
# ============================================================
# ── Step 1: Initialize ──────────────────────────────────────
graph_builder = StateGraph(SupportState)
# ── Step 2: Register All Nodes ──────────────────────────────
# Format: add_node("string_name", function)
# The string name is what you use in every edge definition below.
graph_builder.add_node("agent_node", agent_node)
graph_builder.add_node("tools", tool_node) # Pre-built from Module 3
graph_builder.add_node("review_refund", review_refund)
graph_builder.add_node("summarize_node", summarize_node)
# ── Step 3: Set Entry Point ─────────────────────────────────
# The first node that runs when a user sends a message.
graph_builder.add_edge(START, "agent_node")
# ── Step 4: Wire the Edges ──────────────────────────────────
# After agent_node: conditional - depends on what the LLM decided
graph_builder.add_conditional_edges(
"agent_node", # Source
route_after_agent, # Router function from Module 5
{
"review_refund": "review_refund", # Refund tool → human review first
"tools": "tools", # Other tools → run directly
"summarize_node": "summarize_node", # Long conversation → summarise
"__end__": END, # Plain answer → done
}
)
# After review_refund: conditional - depends on human's decision
graph_builder.add_conditional_edges(
"review_refund",
route_after_review,
{
"tools": "tools", # Approved → execute the tool
"agent_node": "agent_node", # Rejected → back to agent to respond
}
)
# After tools run: always go back to agent_node
# (the ReAct loop - agent sees tool result, decides what to do next)
graph_builder.add_edge("tools", "agent_node")
# After summarization: conversation turn is done
graph_builder.add_edge("summarize_node", END)
# ── Step 5: Compile ─────────────────────────────────────────
# Using MemorySaver for development.
# For production, swap this one line to SqliteSaver or PostgresSaver.
memory = MemorySaver()
shopbot = graph_builder.compile(checkpointer=memory)
In Colab oder Jupyter können Sie das Diagramm als Grafik darstellen:
from IPython.display import display, Image
from langchain_core.runnables.graph import MermaidDrawMethod
display(Image(
shopbot.get_graph().draw_mermaid_png(
draw_method=MermaidDrawMethod.API
)
))
Die Anzeige eines echten Flussdiagramms dessen, was man erstellt hat, gehört zu den besten Funktionen von LangGraph für Lernende. Wenn das Diagramm falsch aussieht, sind auch die Kanten falsch.
Modul 7: Eingangspunkt
Hier läuft der Agent tatsächlich. In einem Skript ist es der if __name__ == „__main__“-Block, und in einem API-Server ist es der Route-Handler – das Muster ist jedoch in beiden Fällen dasselbe.
# ============================================================
# MODULE 7: ENTRYPOINT
# ============================================================
def run_shopbot():
"""
Interactive command-line session with ShopBot.
Demonstrates: multi-turn conversation, tool use, and human-in-the-loop.
"""
print("=" * 55)
print(" ShopBot - Customer Support Agent")
print(" Powered by LangGraph")
print("=" * 55)
print("Type your message below. Type 'exit' to quit.")
print("Type 'state' to inspect what ShopBot currently remembers.\n")
# One config per session.
# thread_id is the session key - same ID = same memory thread.
# Change the ID to start a completely fresh conversation.
config = {"configurable": {"thread_id": "customer-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 support. Have a great day!")
break
# ── Debug: inspect current state ────────────────────
if user_input.lower() == "state":
snapshot = shopbot.get_state(config)
print("\n[DEBUG] Current State:")
print(f" Messages in state : {len(snapshot.values.get('messages', []))}")
print(f" Customer name : {snapshot.values.get('customer_name', '(not set)')}")
print(f" Ticket category : {snapshot.values.get('ticket_category', '(not set)')}")
print(f" Summary : {snapshot.values.get('summary', '(none yet)')}")
print(f" Next node(s) : {snapshot.next}\n")
continue
# ── Normal message: invoke the graph ─────────────────
result = shopbot.invoke(
{"messages": [HumanMessage(content=user_input)]},
config=config,
)
# ── Check if graph paused for human approval ─────────
# This is how you detect that interrupt() was called inside review_refund.
while "__interrupt__" in result:
interrupt_data = result["__interrupt__"][0].value
print("\n" + "=" * 55)
print(" HUMAN APPROVAL REQUIRED")
print("=" * 55)
print(f" Customer : {interrupt_data.get('customer_name', 'Unknown')}")
print(f" Action : {interrupt_data.get('tool_being_called', 'refund')}")
print(f" Arguments : {interrupt_data.get('arguments', {})}")
print(f" Context : {interrupt_data.get('conversation_summary', 'N/A')}")
print("=" * 55)
print("Options: [a] Approve [r] Reject [e] Escalate")
human_choice = input("Your decision: ").strip().lower()
if human_choice == "a":
resume_value = "approve"
elif human_choice == "r":
resume_value = "reject"
elif human_choice == "e":
resume_value = "escalate"
else:
print("Invalid choice. Defaulting to reject.")
resume_value = "reject"
# Resume the graph with the human's decision.
# Command(resume=...) answers the pending interrupt() call.
result = shopbot.invoke(
Command(resume=resume_value),
config=config,
)
# ── Print the agent's final response ─────────────────
last_message = result["messages"][-1]
print(f"\nShopBot: {last_message.content}\n")
if __name__ == "__main__":
run_shopbot()
Eine echte Sitzung
Ein Beispielgespräch zeigt, was Sie erwarten können, wenn Sie es ausführen:
=======================================================
ShopBot — Customer Support Agent
Powered by LangGraph
=======================================================
Type your message below. Type 'exit' to quit.
Type 'state' to inspect what ShopBot currently remembers.
You: Hi, my name is Alex and I need help with order ORD-001
ShopBot: Hi Alex! I'd be happy to help you with order ORD-001.
I can see your order:
- Product: Wireless Headphones
- Status: Delivered on 2025-06-10
- Amount: $89.99
What can I help you with regarding this order?
You: I want to return it and get a refund
ShopBot: I've checked your order ORD-001 and it's eligible for a refund
since it's been delivered. Let me process that for you right away.
=======================================================
HUMAN APPROVAL REQUIRED
=======================================================
Customer : Alex
Action : process_refund
Arguments : {'order_id': 'ORD-001', 'reason': 'Customer requested return'}
Context : No summary yet
=======================================================
Options: [a] Approve [r] Reject [e] Escalate
Your decision: a
ShopBot: Great news, Alex! Your refund has been successfully processed.
Refund Reference: REF-ORD-001-143022
Amount: $89.99 will be returned to your original payment method within
3–5 business days.
Is there anything else I can help you with?
You: state
[DEBUG] Current State:
Messages in state : 6
Customer name : (not set)
Ticket category : refund_request
Summary : (none yet)
Next node(s) : ()
You: exit
ShopBot: Thank you for contacting support. Have a great day!
Was das Projekt zeigt
Die sieben-Modul-Anordnung zeigt die Architektur von Anfang bis Ende: StateGraph, add_node, add_conditional_edges und compile, alles in der richtigen Reihenfolge. Was das Speichern angeht, gibt es das summary-Feld, die Funktion summarize_node sowie das rollierende Zusammenfassungsmuster, bei dem neue Nachrichten komprimiert und alte gelöscht werden, dazu einen MemorySaver-Checkpointer, der die Historie über mehrere Schritte hinweg aufrechterhält. Für den menschlichen Eingriff pausiert der Knoten review_refund mit interrupt(), der Eingangspunkt setzt mit Command(resume=...) fort, und eine Schleife while "__interrupt__" in result kümmert sich um mehrere Genehmigungen in einem einzigen Schritt und bietet die Optionen zustimmen, ablehnen oder weiterleiten.
Mehrere Elemente stammen von Werkzeugen. Reale Werkzeuge werden mit @tool deklariert, der vorbereitete ToolNode führt sie aus, und bind_tools verknüpft sie mit dem Modell. Der ReAct-Loop wechselt zwischen Werkzeugen, Agentenknoten und wieder Werkzeugen usw. Die Routenverwaltung behandelt Werkzeuge unterschiedlich: Sicherere Werkzeuge gelangen direkt zu tool_node, während sensible Werkzeuge zunächst durch review_refund geleitet werden.
Welcher nächste Schritt?
Das Projekt ist absichtlich vollständig, aber minimal gehalten. Drei Erweiterungen, zuerst die einfachsten:
- Bessere Kategorisierung. Derzeit wird
ticket_categorydurch das Scannen von Schlüsselwörtern imagent_nodeermittelt. Eine sauberere Methode besteht darin, das Modell aufzufordern, die Anfrage explizit mit strukturiertem Ausgabeformat zu kategorisieren – schauen Sie sich daherwith_structured_output()an. - SQLite-Persistenz. Ersetzen Sie
MemorySaver()durchSqliteSaver– das ist eine einzeilige Änderung in Modul 6 – und fügen Sie eine Startanfrage hinzu, die abfragt, ob es sich um dieselbe Sitzung oder eine neue handelt. Dadurch bleiben die Gespräche auch nach Neustarts erhalten. - Ein Eskalierungsweg für Beschwerden. Fügen Sie einen vierten Knoten,
escalate_node, hinzu, der ausgelöst wird, wennticket_category == „complaint“und die Stimmung des Kunden„FRUSTRATED“ist. Er ruftinterrupt()auf, um einen menschlichen Agenten zu alarmieren, und verknüpft sich anschließend überroute_after_agentals neue Abzweigung.
Kurzfassung
Neu in diesem Projekt sind einige Muster. ToolNode(tools) ist ein vorgefertigter Knoten, der die in der letzten Nachricht des Modells enthaltenen Tool-Aufrufe ausführt. llm.bind_tools(tools) verbindet die Tool-Liste mit dem Modell, sodass dieses anschließend ein Tool anfordern kann, indem es statt reinem Text ein tool_calls-Feld zurückgibt. @tool wandelt eine Python-Funktion in etwas um, das das Modell aufrufen kann, wobei die Dokumentation als Grundlage dient, um zu entscheiden, wann sie verwendet werden soll. Die Überprüfung hasattr(message, „tool_calls“) and message.tool_calls ist die übliche Methode, um festzustellen, ob die letzte Nachricht einen Tool-Aufruf enthält, wobei hasattr zuerst aufgerufen werden sollte, bevor auf das Attribut zugegriffen wird. Die Schleife while „__interrupt__“ in result kümmert sich um einen oder mehrere ausstehende Unterbrechungen, da jede Command(resume=...) eine solche beantwortet und möglicherweise die nächste auslöst. Schließlich zeigt shopbot.get_state(config) zu jedem Zeitpunkt den vollständigen aktuellen Zustand an: .values liefert das Zustandsdict und .next angibt, was als Nächstes ausgeführt werden würde – was für das Debuggen von unschätzbarem Wert ist.
Der Unterschied zwischen dem Verständnis eines Konzepts und dem Schreiben von Code dafür existiert tatsächlich, und ein kleines Projekt, das in den Kopf passt, aber alle verfügbaren Werkzeuge einsetzt, hilft, diese Lücke zu schließen. Hier haben Sie einen Überprüfungspunkt, echte Werkzeuge, eine Freigabestufe sowie einen Auslöser für die Zusammenfassung, die alle gemeinsam in einem Agenten arbeiten. Bauen Sie es, brechen Sie es absichtlich, beheben Sie es und erweitern Sie es anschließend.
Verwandte Literatur
- Validierungsmechanismen und Mehr-Agenten-Routing für einen LangGraph-Support-Bot — Fügen Sie einem LangGraph-Agenten für den E-Commerce deterministische Geschäftslogik-Regeln, einen selbstkorrigierenden Validierungszyklus sowie einen Supervisor mit drei spezialisierten Subgraphen hinzu.
- Mehr-Agenten-Support-Ticket-Triage mit LangGraph und Ollama — Erstellen Sie mit LangGraph ein dreifach-Agenten-System zur Ticket-Triage: einen Klassifikator, einen Antwortgeber und einen menschlichen Überprüfer, das auf lokalem Ollama oder über eine API mit nur einer geänderten Zeile läuft.