LangGraph客服机器人的验证门与多智能体路由机制
向电子商务LangGraph智能体中添加确定性的业务逻辑门、自纠正的验证循环以及包含三个专业子图的监督器。
想象一下，有一个基于LangGraph开发的在线商店客服机器人，这里称之为ShopBot。它能查询订单信息、判断退款资格，在人工确认后处理退款，并对对话内容进行总结。该机器人的基础版本已经可以运行。接下来的内容将介绍具备实际应用价值的机器人所需的三个功能：对每个模型输出的结果进行验证、能够自动修正模型的循环机制，以及由监督者将任务分配给各专业人员的多智能体架构。最终目标是打造出一个可以直接提交给生产环境审查而无需担心的代码库。
所采用的基准模型是一个包含常规组件的图结构：工具模块、ReAct风格的循环机制、状态管理，以及用于暂停图执行以等待人工确认的interrupt()函数。
基础机器人中的隐性假设
简单版本在所有情况下都依赖语言模型。其代理会进行资格检查，并相信从聊天中提取的订单编号是正确的。审核节点会将该订单编号展示给人工审核员，但审核员仅负责批准或拒绝操作，而不会验证该编号是否存在。退款工具则按照模型选择的参数来运行。如果模型错误地生成了“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 的功能：无需模型、无需图结构，也无需运行任何进程，即可确认验证逻辑是否正确。只需用示例数据调用 gates 并对结果进行断言，这便是实践中“确定性”的含义。
扩展状态模式
在其他任何改动之前，首先需要为状态添加新字段。最初的版本包含三个自定义字段：摘要、客户姓名以及工单类别，现在又增加了四个字段。
# ============================================================
# 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标记，因此它会在整个对话过程中持续累积，而不仅仅局限于某一次交互。假设在第三次交互中出现了一个虚构的订单编号，而在第七次交互中出现了不合理的退款理由：这两种错误都会被记录到审计日志中。这在合规审查中非常重要。
编写审核机制
门函数是一种纯粹的验证函数。它接收提取到的数据，并返回一个元组：成功时返回 (True, cleaned_data)，失败时返回 (False, error_reason)。这就是其全部接口，无需调用模型，也没有图结构依赖。下面的模块包含了订单编号校验、全额退款校验以及投诉校验功能，同时还包含了它们所使用的业务常量。
# ============================================================
# 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)}"],
}
顶部的重试检查是整个系统中最重要的安全机制。如果没有它，那些真正出问题的输入（比如客户不提供有效的订单编号，或是模型配置持续产生错误结果）将会无限循环，在校正节点与入口节点之间反复切换。对于支持工作而言，两次重试是一个合理的默认值，因为这给了模型在人工介入之前自我修正两次的机会。该次数可根据对延迟与准确度的容忍程度进行调整。
三个专业子图
每个专业子图都是经过编译的独立结构，拥有自己的工具、提取逻辑、入口节点以及内部路由机制。上层管理图既不了解也不关心这些内部结构，它看到的只是接收状态并返回状态的节点。
订单处理代理
# ── 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代替了数字0，工具会报错，智能体或许能够自行恢复，而客户收到的回复可能会略显混乱。不过通常也没什么大问题。
企业代理假定模型有时会出错，因此会在错误影响到外部系统之前就建立相应的恢复路径。检测机制能立即捕获“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 上运行，仅需修改一行代码即可实现。
- 使用 LangGraph 和 Ollama 构建具备自我修正功能的文档智能体 —— 建立一个本地反馈流程：第一个模型负责起草 Python API 文档，第二个模型对其进行评审，第三个模型整合结果，所有环节通过 LangGraph 在 Ollama 上形成循环结构。