Фільтры праверкі і шматагентная маршрутызацыя для бота падтрымкі на LangGraph
Дадаём да агента LangGraph для e-commerce дэтэрмінаваныя фільтры бізнес-логікі, самавыпраўляльны цыкл праверкі і супервізара з трыма спецыялізаванымі падграфамі.
Уявіце сабот для падкінання кліентаў у інтэрнэт-магазіне, які тут называецца ShopBot і створаны на базе LangGraph. Ён шукае замовленні, пераканальваецца ў правам на вярненне грошаў, апрацоўвае ўярненні пасля пацверджэння чалавекам і падтрымвае падсумованы варыянт дыялогу. Базовая версія такога сабота вялікі час вучыцца працаваць. У наступным разе дадаюцца тры элементы, якія неабходны саботу, які мае рэальныя наследкі: механізмы паўтарной пераканальвання для кожнага рэзультата модэлю, цыкл, які автаматычна выправляе помылкі модэлю, і структура з калькамі спецыялістаў, у якой керуючы аддзеўае заведаменні спецыялістам. Мета — стварыць кодавую базу, якую можна было б без жадных абавясцей практычна адразу вікорыстоўваць.
Прыйнятым базовым прыкладам ёсць граф з звычнымі модулямі: інструментамі, цыклам у стылі ReAct, станам і вызывам interrupt(), які паўзіць працу графа для пацверджэння чалавекам.
Тыхая прыпуск у базовым сабоце
Простая версія павераецца модэллю мовы ў всіх аспектах. Іе агент выкананае пераконтрацею праваможнасці і павераецца тым, што ID замовлення, які ён перадае, быў правільна выкараны з чату. Вузол адзінакоўвання паказвае гэты ID чалавеку, але адзінакоўвач затверджвае або адхоўчвае дзеянне, не перакантрачычы, чыі існуе гэты ID. Адзінакоўвач вярнення грошаў працюе з тымі аргументамі, якія выбраўа модэль. Калі модэль стварыць халюцынацію і запісае „ORD-99X“, калі мела б быць „ORD-001“, адзінакоўвач даўць памылку, агент убачыць гэтую памылку, і можа выйсці ўсё правільна. Фраза „можа выйсці ўсё правільна“ — гэта слабкі апіс праблемы адзінакоўвання памылак у сістэме, якая пераказвае грошы.
Эксперты, які керуюць автаматызаванымі працэсамі закупкаў чыста або супакоўнення, чытка адзначаюць: там некалічымыя парадоксы не ўзімкі. Рашэння не ў тым, каб больш давераць модэлю. Яго — зовсім перэстаць давераць у ўсіх вынікі, і пераканацца ў кожным структураваным выніку за дапамогою дэтэрміністычнага Python прытымкі, прычым яны не трапляюцца з усім зовнішнім.
Тры дадатковыя элементы і як яны взаіднаяюць
Бізнес-логічныя вораты знаходзяцца ўнутрь вузлаў і пераканаюць вынікі модэлю прытымкі, прычым нічога не дзейнае з імі. Гэта звычны Python без жадных вызоў модэлю. Вораты альбо прымаюць даны, альбо адхоўляюць іх з конкрэтным, структураваным адказам. Гэта дэтэрміністычная пераканаць унутрь парадоксальнай сістэмы.
Цыклы парабярання раслічваюцься з неудачамі на выходах. Адмова не спрычынае зламу графа і не дае некорэктнай адпаведзі. Контроль пераходзіць да вузла карэкцыі, які перадае прычыну неудачы модэлю і просіць або правільны выхід. Лічылка прабаў у стане запобегае бесканцовым цыклам, а жорсткі ліміт спрычынае перадачу справы чалавеку.
Архітэктура з калькамі заменяе вузел адного агента на трох спецыялістаў. Агент з ардэрамі займаецца пошукамі і запытамі пра статус. Агент з вярненням грошаў адпавядае за перакананне ў адпаведнасці і апрацоўку вярнення. Агент з працэўкамі карыстаецца ся з незадоволенымі кліентамі і справамі, якія трэба перадаць вышэй. Вузел надзорчыка пераглядае кожную прыйшоўшую паведамленне і перадае яе праваму спецыялісту, пры чым кожны спецыяліст ёсць скомпіляваныя падграфамі сама сабе.
Этыя элементы працуюць як адна сістэма. Агент на адзмену грошаў несе з сабой «вораты», а неякшыя «вораты» могу запусціць цыкл паўтарной перапрацоўкі, а надзірнік аналізуе, як заканчыўся гэты цыкл, прычым выбірае, чы хацець прабаваць знову, падняць проблему на вышэйшы ранг чы абарыць процэс. Разам яны надаюць готоваму боту стойкасці, якой трэбуе корпаратыўнае сераўернае сэрвіса.
Спланаваце граф прычым прыступаеце да коду
LangGraph адзначае важнасць планавання, і чым большы граф, тым больш гэта правда. Спачатку накрэсліце весь сістэма. Кожны блакет павінен адпавядаць толькі аднаму вузлу чы функцыі маршрутацыі, а кожны стрэлкі — толькі аднай грані. Код стане тады перакладам гэтага пляна ў Python, модуль за модулем.
Новая структура файлаў
Базовы бот могаў поместыцца ў адны файл. Пры такой велічыне адны файл стае незручным, таму праект дзеліцца на тры часткі.
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)
Намеравана ўсунення імпорта LangGraph з файла gates. Функцыі gates стаюць практычна тэставаныя за дапамою звычных тэстаў на елементы: для парадактнасці логікі верыфікацыі не трэба ні модэля, ні графа, ні працюючага процэса. Вы вызываеце функцыю gate з прыкладным дадзенням і пераканальваетэсь у выніку, і гэта і ўсё, што на практыцы значыць „дэтэрміністычнасць“.
Пашырэнне схемы стану
Перш за ўсё стану патрэбны новыя полья. Пачатковая версія мела тры спецыяльныя полья: апূরкненне, імя кліента і катэгорыю квітанцыі. Да іх дадаўшы ўсьмо чатыры.
# ============================================================
# 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
Ключовым новым полем ёсць extracted_data. У базовым боце модэль ствараў тэкст свабоднага фармата і безпасерэдзьва вызывалаў інструменты, таму аргументы інструментаў былі тымі, што модэль укладвала ў своіы вызывы. Цяпер модэль спачатку выдае структураваныя даны, якія падаюць у extracted_data. Сістэма перакрыцча іх пераглядае, і толькі пасля таго, калі перакрыцча прабачае, гэтыя даны стаюць аргументамі інструмента. Гэта усунеў любы безпасерэдні шлях ад модэлі да званачных сістэм.
Поле validation_errors анотаванае operator.add, таму яна накапліваецца працэсе всіх размов, а не толькі працэсе адной партыі. Падазроўцай, у трэцій партыі з’яўляецца вымышлены ID замовлення, а ў сьпявай партыі — паслабыя празыгрышы для вярнення грошаў: обаі прычыны патрапляюць у журнал аўдыту. Гэта мае значэнне пад час пераглядоў са адпаведнасцю.
Стварэнне сістэм перакрыцча
Вората — гэта чыстая функцыя параболізацыі. Яна прыме выкараныя даны і вяртае тупл: (True, cleaned_data) у разе успеху або (False, error_reason) у разе неудачы. Гэта ўсё, што ўключае інтэрфейс — без вызоў модэляў і без залежнасцей ад графа. Модуль ніжэй уключае перакананне ID замовлення, перакананне аб повнай вярнуццы грошаў і перакананне аб прытварэнні, а таксама сталярныя значэнні, якія ў яму вікорыстоўваюцца.
# ============================================================
# 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
Паглядзіце, чаго не хапае гэты файл: імпортаў з LangGraph, LangChain чыя-небудзь іншай бібліятэкі AI. Кожны распрацоўшчык у командзе можа чытаць, тэставаць і зменяць яго, не ведаючы ніч пра архітэктуру агента. Калі аўдытар спытае, што самэўсьвядома паўтарае запіт на вярненне грошаў пры перадачы сумы, самэўсьвядома трэба паказаць гэты файл, і людзі, якія не ўжо ў галузі інжынерыі, зможуць яго зрозумець.
Цыкл паўтарэння
Самэўсьвядома корыстныя ўсё тыя „вораты“ — гэта ўзначальваеецца за дапамогою вузла карэкцыі. Калі „вораты“ не працуюць, выкананне не зупіняецца, а пераходзіць сюды. Вузел берае конкрэтную прычыну абякання з стану, стварае інструкцыю, якая чытальна паведамляе модэлю, што самэўсьвядома паўтарылася, і просіць паўтарэнную структураваную выходную інфармацыю.
# ── 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)}"],
}
Перакантрольванне пасля прабавок, якое знаходзіцца ў верхней часты, ўсё ж такі ўсё важлівейшы механізм безпекі ў цэлай сістэме. Без яго справжня несправнае даныя (кліент, які не можа запростаць правільны ID замовлення, або конфігурацыя модэлю, якая продовжае ствараць халюцынаціі), будзе працаваць без канца, перэключаючыся між вузлам для корэкцыі і шлюзам. Две прабавкі ўтвараюць разумны стандарт для працы падтрымкі, адколі ў модэль даецца два шансы сама сабе выправіцца прычым перш чым усе будзе перадана людзю. Колькісць прабавак можна налаштаваць залежна ад таго, насколькі допускаецца затрымка проты точнасці.
Тры спецыялістскія падграфы
Кожны спецыяліст — гэта скомпільаваны падграф з своімі інструментамі, логікай выкарыстоўвання даных, шлюзам і внутранім маршрутызаваннем. Галоўны надзорчы граф нічога не ведае і не цікавіцца гэтай внутранней структурой. Ён бачыць толькі вузел, які прыме стан і вяртае стан.
Агент замовленняў
# ── 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()
Адаптавайце цікл у гэтым падграфе. Узел контролю замовлення падае сігнал працэўніку з корэкцыі у разе абякання, а працэўнік з корэкцыі падае сігнал назад у узел контролю. Гэты цыкл являе сабою цыкл паўтаральнай перапрацоўкі, і ён выконваецца ў максымум два разы, прычым перад тым, як лічыльнік створыць паведамленне пра падышчу проблемы.
Агент з вярнення грошаў
Гэты спецыяліст мае вялікое значэнне, таму што ён керуе грошамі. Ён выконвае два крокі: спачатку перакананне ідэнтыфікатара замовлення, а потым перакананне аб повнам вярненні грошаў, прычым працэс людскай апраўды застаецца ў межах гэтага падграфа.
# ── 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()
Агент з прыткіх
Гэта самы ўжо просты спецыяліст. Ён выразае разумянне дзіўнення кліента, паўтарае, што ёсць достатнько дакладнасці, каб створыць заявку, і перадае справу чалавеку за дапамогою interrupt(), калі настроў кліента ўжо дужа негатыўны.
# ── 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()
Граф каналізатора
Пасля складання падграфаў граф каналізатора выявляецца ўсёлякі простым. Ён толькі каналізуе. Ён ніколі не выконвае спецыялістскіх задач. Прынцып, які стоіць за патэранамі з калькамі агентаў, якраз такі: каналізаторы каналізуюць, а спецыялісты выконваюць.
# ============================================================
# 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()
Які змяненні ў повадзе
Як базовая, так і корпоратыўная версіі ўсунулі правільныя рэалізацыі стандартных шаблонаў LangGraph, але яны выконваюць разныя прыпускі, і гэта разліка стае значным, калі вы выйдзеце за межы сандбокса і перэйдзеце да сістэмы з рэальнымі наследкамі.
Базовы агент прыпускаў, што модэль правы. Калі модэль выкарыстоўвала значэнне „ORD-0O1“, дзе літэра O заменяла нуль, інструмент даў бы памылку, агент, верагодна, зможа бы выкарыстаць альтэрнатыўны падход, а кліент мог бы пацярпець трохі заплутаны адказ. Верагодна, гэта нічога страшнага.
Агент адміністрацыі прыймае, што мадэль іноды можа памыліцца, і стварае шлях для вярнення да стану нормы, прычыму гэтая памылка не дася да якога-небудзь зовнішньага элемента. Сістэма адразу ж перапрымае сігнал „ORD-0O1“ і выдае паведамленне пра памылку, у яком чыткаа паказана, што самэўсь памылкай і чаму. Вузел корэкцыі перадае гэтае паведамленне назад модэлі, і модэль мае другую спробу з точнымі інструкцыямі. Калі памылка паўтарыцца, дыялог перадаецца чалавеку. Жадныя інструменты не працавалі з некорэктнымі дадзеннямі, і жадная зовнішняя сістэма не бачыла гэтай галюцынацыі.
Завязанае з станам validation_errors магчыма старшаму інжынеру, які пераглядае сесію, у якой была дана некорэктная адпаведзь, мець полны лог усіх памылак сістэмы, спроб корэкцыі і колькісці паўтароў. Такую памылку можна воспроізвесці і зрозумець, пасля чаго або практычна пасправляецца запіт, або ж сістэма стае болей строгай. Без такога логу дэбаггін верагатывочнай сістэмы значыць толькі здагадкі.
Структура з трыма файламі таксама ўзгодна для аўдытораў. Калі запытаюць, што паўніста падтверджуе можлівасць вярнення грошаў пры ўсуненні ўсіх рызыкаў, адпаведзенне — адзін файл на чыстай мове Python без якой-лібо архітэктуры агентаў, якая трэба было б поясняць: чыстыя, падлягаюцы аўдыту бізнес-правіла, напісаныя мовай, яку можа прачытаць кожны распрацоўшчык.
Глосарый новых элементаў
- Бізнес-логічны брамнік: просты код на Python, які прымеў выкаранае даныя і вярне або
(True, cleaned_data), або(False, error_reason). У яго ёсць свой файл без імпортаў з LangGraph, і ён выконваецца як окрэмы вузол раней, чым выкананы які-лібо іншы інструмент. with_structured_output(PydanticModel): змушвае мадэль вяртаць даныя, якія адпавядаюць схеме Pydantic, а не просты тэкст. Схема пераканальваецца раней, чым запускаецца брамнік, таму структураваны выхід у поўнасці разам з брамнікам стварае два роўні падтверджэння.
extracted_data: dict: поле стану, дзе зберагаецца структураваны выхід модэля прычымо да таго, як яго аналізуюць шлюзы. Адпрацоўчыкі чытаюць тут, а не з первіснага выхіднага даныя модэля.validation_errors: Annotated[list[str], operator.add]: летак аўдыту. Кожна неудача шлюза дадаецца да яго, і ён накапліваецца працэсе всіх размов.retry_count: int: захоп працэса ад бесканцовых цыклів. Узел корэкцыі збільшае яго, функцыя направлення шлюза перакантролюе яго, а перасягнуць максімум означае неабходнасць пераводу справы на вышэйшы ранг і абмову ад новых спроб.correction_node: узел вяснавання. Ён чытае апошню памятку пра адмову ад верыфікацыі, стварае конкрэтную інструкцыю па корэкцыі, знову запрашвае модэль з структураваным выхідным даным і направляе яго назад да шлюза. Усі трое спецыялістаў выкарыстоўваюць яго.gate_status: str: канал між вузламі-брамамі і функціям маршрутавання, пры якім значэнняя можа быць "pending", "passed", "failed", "escalated" і "aborted". Функцыі маршрутавання чытаюць гэты параметр замест таго, каб аналізаваць выходны даныя модэлю.supervisor_node: чытае статус размовы і вяртае імя спецыяліста. Ён ніколі не вызывае інструменты і не выконвае жадных дзеянь.graph_builder.add_node("name", compiled_subgraph). Абяцелы вузел бачыць чорную скрынку, тады калі падграф сам адпаведальны за свае маршрутаванне, логіку брам і вызывы інструментаў.Закрыцце прыемасці
Складныя аспекты работы агента падпрыемства заключаюцца не толькі у пачатковай структуры графа: дэтэрміністычна обработка памылак, перакананне стану на вялікіх масштабах і шар оркестрацыі, які не прыкрэплён да адного прадавца модэляў. На гэтыя тры аспекты ёсць адпаведныя адказы. Перакананне стану ўраджанае — адказка выклікае цыкл карэкцыі, а не збой. Кожна рашэнне можа быць пераглянута праз лёгкі для адстэплення журнал. У шарэ оркестрацыі, незалежна ад таго, чы то вораты, функцыі маршрутызацыі, логіка кантролера чы вузел карэкцыі, нічога спецыфічнага для прадавца не імпортуецца. Змяніце настроўанага прадавца з ChatOpenAI на ChatAnthropic чы на будзь-што іншае, што падтрымлівае init_chat_model, і кожны ворат, маршрут і бізнес-правілл будуць працаваць так сама, як і раней. Модэль ёсць модулем, які можна падключыць, а архітэктура — ні.
Спадні матэрыялы
- Трыяж тыкетаў падтрымкі з удзелам калькольнікаў з LangGraph і Ollama — Створыце граф трыяжа тыкетаў з трох калькольнікаў за дапамогай LangGraph: класыфікатор, адпаведальны за адпаведзь і людзкі адпаведальны, які працуюць на локальным Ollama або праз API, з адной змененай лінійкай коду.
- Створыце агента для документацыі, які сам сабе карэктуе, з LangGraph і Ollama — Локальны практык для перагляду, дзе адны модэль прабывае напісаць документацыю Python API, другі яе крытыкуе, а трэці супрацоўвае з рэзультатам, усе гэта з’еднанае ў цыкл LangGraph на Ollama.