AB
AiBoss站
教程

用 LangGraph 构建带人工审批的退款 Agent:人工介入检查点实战

教程

用 LangGraph 构建带人工审批的退款 Agent:人工介入检查点实战

AI Agent 在读取数据、检索信息时大可以自主运行,但一旦要动钱、改客户记录或触发外部系统,就需要在动作落地前插入人工审批。本文以退款 Agent 为例,讲清人工介入检查点(human-in-the-loop)该放在哪一步、如何用 LangGraph 的 interrupt() 暂停工作流、如何用 checkpointer 保存状态并在人工审核后恢复执行,并给出从订单查询、结构化退款提案到审批路由、退款执行的完整可运行示例。

AI Agent 的响应速度很快,但当下一步动作涉及资金、客户记录或外部系统时,速度本身就是风险。人工介入检查点(human-in-the-loop checkpoint)解决的就是这个问题:让 Agent 自主完成调查、检索与推理,在它准备把建议变成真实世界动作的那一刻停下来,交给一个人确认。本文用一个退款 Agent 作为完整例子,演示如何用 LangGraph 搭建「调查 → 提案 → 暂停审批 → 执行或终止」的工作流。

什么是人工介入检查点

人工介入检查点是一段刻意插入 AI 工作流中的暂停。在到达这个点之前,Agent 可以完全自主地工作:理解请求、检索信息、调用工具、准备建议。但在执行敏感动作之前,工作流会停下,请人复核这份建议。

一个有用的复核环节应当清楚呈现四件事:

  • Agent 想做什么
  • 它为什么选择这个动作
  • 支撑这个决定的信息是什么
  • 如果批准,接下来会发生什么

检查点的位置至关重要。如果 Agent 已经发出邮件、更新了客户账户或完成了退款,再请人审批,这个复核就失去意义了。检查点必须落在执行之前。

在 LangGraph 中,这可以通过 interrupt() 实现。图会在该点暂停,借助 checkpointer 保存自身状态,等待外部决策后再继续。概念上,这种设计把两项职责分开了:Agent 决定它建议什么,应用程序决定这个建议是否被允许变成真实动作。

什么时候该让 Agent 请求审批

并非 Agent 的每个动作都需要人工审批。如果 Agent 只是读取信息、检索知识库或起草文稿,插入人工环节只会带来不必要的摩擦。但一旦 Agent 即将修改数据、联系客户、转移资金或执行其他有后果的动作,复核环节就变得有价值。

一个简单的判断方式是把低风险动作与会影响外部世界的动作分开:

  • 低风险动作 → Agent 可以继续自主执行
  • 高影响动作 → 执行前必须人工复核

在退款场景里,查询订单是只读动作,Agent 可以自主完成;而发起退款不同,它会改变财务状态,人工检查点就放在这里。每个组织都应当根据自己的政策、风险容忍度和错误动作的后果来划定这条边界。

退款 Agent 的工作流结构

在写代码之前,先理解整个退款流程如何运转。它分为四个阶段:准备、复核、决策、执行。

准备

客户用自然语言提出退款请求。LLM Agent 读取请求,判断是否需要补充信息。如果请求里带有订单号,Agent 调用 get_order 工具,取回经过核实的详情,例如商品、金额、配送状态和退款状态。结合客户请求与检索到的订单信息,Agent 生成一份结构化的退款提案。

复核

提案被送入人工审批检查点。LangGraph 不会立即执行退款,而是用 interrupt() 暂停工作流。复核者可以查看建议的退款金额、订单号和 Agent 给出的理由。

决策

复核者批准或拒绝这份提案。批准则工作流继续进入退款执行步骤;拒绝则工作流结束,退款工具不会被调用。

执行

只有被批准的请求才会到达 issue_refund 工具。该工具执行模拟退款并返回结果。

这种职责分离很重要:LLM 负责调查与推理,应用程序与人工复核者共同控制有后果的动作是否被执行。Agent 在调查阶段保持自主,而人工监督恰好落在工作流开始影响外部世界的位置。

准备工作

开始构建之前,先确认环境满足以下条件:

  • Python 3.10 或更高版本
  • 一个 OpenAI API key
  • 基本的 Python 使用经验
  • 对 LLM Agent 与工具调用有基本了解

安装所需依赖库:

pip install -U langchain langgraph langchain-openai pydantic

各库的分工如下:

  • langchain:定义工具、创建 Agent
  • langchain-openai:把 Agent 连接到 OpenAI 模型
  • langgraph:编排工作流,并在人工审批处暂停
  • pydantic:定义结构化的退款提案

运行代码前先设置 API key。macOS 或 Linux 下:

export OPENAI_API_KEY="your-api-key"

Windows PowerShell 下:

$env:OPENAI_API_KEY="your-api-key"

环境就绪后,先创建 Agent 需要检查的模拟订单数据。

操作步骤

第一步:创建模拟订单数据

先准备一个内存中的小型订单存储,让 Agent 有东西可查。

ORDERS = {
    "ORD-1024": {
        "product": "Wireless Headphones",
        "amount": 79.99,
        "status": "delivered",
        "refunded": False
    },
    "ORD-2048": {
        "product": "Mechanical Keyboard",
        "amount": 119.00,
        "status": "delivered",
        "refunded": False
    }
}

每个订单包含四个字段:

  • product:购买的商品
  • amount:可退款的金额
  • status:当前订单状态
  • refunded:该订单是否已经退过款

为了让示例容易运行,这里把数据全部放在内存里。在生产系统中,同样的信息通常来自数据库、电商平台、CRM 或内部订单 API。关键点是:Agent 不应该凭空编造订单信息,而应当通过工具获取经过核实的数据。

第二步:创建订单查询工具

接下来创建一个只读工具,让 Agent 能取回经过核实的订单信息。

from langchain.tools import tool

@tool
def get_order(order_id: str) -> dict:
    """Retrieve order details for a given order ID."""
    order = ORDERS.get(order_id)
    if not order:
        return {
            "found": False,
            "order_id": order_id
        }
    return {
        "found": True,
        "order_id": order_id,
        **order
    }

@tool 装饰器让这个函数对 LLM Agent 可用。文档字符串同样重要,模型会借助这段描述理解工具做什么、什么时候该调用它。

例如客户说「我的订单号是 ORD-1024,耳机到货时已经损坏」,Agent 能识别出订单号,并判断在给出建议前需要更多信息,于是调用:

get_order("ORD-1024")

得到返回:

{
    "found": true,
    "order_id": "ORD-1024",
    "product": "Wireless Headphones",
    "amount": 79.99,
    "status": "delivered",
    "refunded": false
}

这是 Agent 设计中的重要一环:LLM 不猜测商品、价格或订单状态,而是从受控来源检索这些信息。get_order 工具是只读的,它能查看数据,但不能修改任何东西,因此适合交给 Agent 自主使用。

第三步:定义退款提案结构

Agent 不应该返回一段之后还要解析的非结构化文字。这里用 Pydantic 定义结构化的退款提案。

from pydantic import BaseModel, Field
from typing import Literal

class RefundProposal(BaseModel):
    order_id: str = Field(
        description="The order being evaluated"
    )
    action: Literal["refund", "no_refund"] = Field(
        description="Whether a refund should be proposed"
    )
    refund_amount: float = Field(
        description="Refund amount. Use 0 if no refund is proposed."
    )
    reason: str = Field(
        description="Short explanation for the recommendation"
    )

这给了 Agent 一个可预测的输出格式,例如:

{
    "order_id": "ORD-1024",
    "action": "refund",
    "refund_amount": 79.99,
    "reason": "The customer reported receiving a damaged product."
}

结构化输出对工作流的其余部分很有帮助,因为每个值都可以直接访问:

proposal["order_id"]
proposal["refund_amount"]
proposal["action"]
proposal["reason"]

这比从自由文本里抽取字段可靠得多,也让人工审批环节更容易实现——应用程序可以把确切的订单号、建议金额和理由直接展示给复核者。

第四步:创建 LLM 并构建退款 Agent

现在创建真正由 LLM 驱动的 Agent。先初始化模型:

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="gpt-4.1-mini",
    temperature=0
)

这里使用较低的 temperature,因为工作流更希望得到可预测的响应。接着创建 Agent:

from langchain.agents import create_agent

refund_agent = create_agent(
    model=model,
    tools=[get_order],
    system_prompt="""
    You are a customer support refund agent.
    Your job is to investigate refund requests.
    If an order ID is available, use the get_order tool to verify the order before making a recommendation.
    Never invent order information.
    Use the customer request and verified order details to decide whether a refund should be proposed.
    Do not issue refunds yourself.
    Your responsibility is to investigate the request and recommend an action.
    """
)

到这里,工作流才真正具备 Agent 特性:模型可以分析客户请求、判断何时需要订单信息、调用 get_order、检查返回数据,并基于这些经过核实的信息继续推理。

一个关键细节是:Agent 只拿到只读的 get_order 工具,没有拿到退款执行工具。这种分离是刻意的——Agent 可以自由调查并给出建议,但它无法自行发起退款。

第五步:运行 Agent 调查

Agent 配置好 get_order 工具后,就可以发送真实的客户请求了。先写一个小辅助函数:

def investigate_request(customer_request: str):
    result = refund_agent.invoke({
        "messages": [
            {
                "role": "user",
                "content": customer_request
            }
        ]
    })
    return result

然后用一条退款请求试试:

customer_request = """
The headphones I received are damaged.
My order ID is ORD-1024.
Please issue a refund.
"""

investigation = investigate_request(
    customer_request
)

Agent 以自然语言接收请求,识别出订单号,判断需要更多信息,并调用 get_order 工具。对 ORD-1024 来说,工具返回:

{
    "found": true,
    "order_id": "ORD-1024",
    "product": "Wireless Headphones",
    "amount": 79.99,
    "status": "delivered",
    "refunded": false
}

Agent 随后基于这些经过核实的信息继续推理。这一点很重要:LLM 并不被期望自己知道订单金额或状态,这些细节来自受控的数据源。

第六步:把调查结果转成结构化退款提案

调查完成后,需要把 Agent 的结论收敛成第三步定义的 RefundProposal。做法是让模型以结构化输出的方式返回提案,而不是自由文本。这样后续节点可以直接读取 order_id、action、refund_amount 和 reason,无需再做文本解析。

提案生成后,工作流就进入下一个关键节点:人工审批检查点。在此之前,所有动作都是只读的;在此之后,动作将改变财务状态。

第七步:创建退款工具

退款执行工具只在审批通过后被调用。它接收订单号与金额,执行(示例中为模拟)退款并返回结果。这个工具不会交给调查用的 Agent,它只出现在审批通过后的执行节点里。

第八步:定义 LangGraph 状态

LangGraph 工作流需要一个状态结构,用来在节点之间传递信息。状态中通常包含:

  • 消息列表(客户请求与 Agent 的推理过程)
  • 结构化的退款提案
  • 人工审批的决策结果
  • 退款执行的结果

状态是工作流暂停与恢复的基础。使用 checkpointer 时,这份状态会被持久化,因此图可以在人工审批期间保持暂停,并在收到决策后从原处继续。

第九步:创建 Agent 节点

Agent 节点负责调用前面构建的 refund_agent,完成订单查询与提案生成,并把结果写回状态。这个节点是自主的,不需要人工干预。

第十步:加入人工审批检查点

这是整篇教程的核心。在提案生成之后、退款执行之前,插入一个审批节点,并在其中调用 interrupt():

from langgraph.types import interrupt

def human_approval_node(state):
    proposal = state["proposal"]
    decision = interrupt({
        "order_id": proposal["order_id"],
        "action": proposal["action"],
        "refund_amount": proposal["refund_amount"],
        "reason": proposal["reason"]
    })
    return {"decision": decision}

图会在这个点暂停,把状态交给 checkpointer 保存,然后等待外部决策。传给 interrupt() 的内容就是复核者看到的提案内容:订单号、建议动作、退款金额和理由。

第十一步:路由审批决策

审批结果需要决定工作流往哪走。用条件边根据决策值路由:

  • 决策为批准 → 进入退款执行节点
  • 决策为拒绝 → 进入拒绝节点,工作流结束,不调用退款工具

第十二步:创建退款执行与拒绝节点

执行节点调用 issue_refund 工具,把订单号与金额传进去,并把返回结果写入状态。拒绝节点不执行任何有副作用的动作,只记录拒绝结果并结束流程。

第十三步:组装完整的 LangGraph 工作流

把上述节点按顺序连起来:

  1. Agent 节点(调查 + 生成提案)
  2. 人工审批节点(interrupt() 暂停)
  3. 条件路由(批准 / 拒绝)
  4. 退款执行节点 或 拒绝节点

这个顺序保证了:任何未被批准的请求都不可能到达退款工具。

第十四步:加入 checkpointer 并编译图

要让 interrupt() 真正能暂停并恢复,必须给图配置 checkpointer。编译时把 checkpointer 传进去,状态才能在暂停期间被保存下来,并在人工给出决策后恢复执行。

第十五步:运行工作流并在审批处暂停

用一条客户请求启动工作流。图会执行 Agent 节点,完成订单查询与提案生成,然后在人工审批节点处暂停,等待外部输入。此时退款尚未发生。

第十六步:人工复核后恢复工作流

复核者查看提案后给出决策,工作流从暂停处恢复。批准则进入退款执行节点,调用 issue_refund 并返回结果;拒绝则进入拒绝节点,流程结束,退款工具不会被调用。

第十七步:端到端测试

至少覆盖两条路径:

  • 批准路径:Agent 查到订单、生成提案、暂停、批准、执行退款、返回结果
  • 拒绝路径:Agent 查到订单、生成提案、暂停、拒绝、流程结束、退款工具未被调用

同时建议测试订单不存在的情况,确认 Agent 不会编造订单信息,而是基于 found: false 的返回给出「不退款」的建议。

一个完整示例

把上面的步骤串起来,最小可运行流程如下。

准备阶段:安装依赖、设置 OPENAI_API_KEY、定义 ORDERS 字典、定义 get_order 工具、定义 RefundProposal 模型、创建 ChatOpenAI 与只带 get_order 的 refund_agent。

调查阶段:调用 investigate_request,传入客户请求。Agent 识别出 ORD-1024,调用 get_order,拿到商品为 Wireless Headphones、金额 79.99、状态 delivered、未退款。

提案阶段:把调查结果转成 RefundProposal,得到类似这样的结构:

{
    "order_id": "ORD-1024",
    "action": "refund",
    "refund_amount": 79.99,
    "reason": "The customer reported receiving a damaged product."
}

审批阶段:工作流进入人工审批节点,interrupt() 把上述提案交给复核者,图暂停。此时没有任何资金变动。

决策与执行阶段:复核者批准后,工作流恢复,进入退款执行节点,调用 issue_refund,返回模拟退款结果。若复核者拒绝,工作流进入拒绝节点并结束,issue_refund 永远不会被调用。

整个示例的价值不在于退款本身,而在于这条边界:LLM 负责调查与建议,应用程序与人工负责决定建议能否落地。

把工作流推向生产

示例中的订单存储是内存字典,退款工具是模拟实现。要投入实际使用,需要替换的部分包括:

  • 订单数据改为来自数据库、电商平台、CRM 或内部订单 API
  • 退款工具对接真实的支付或退款接口,并做好幂等处理
  • checkpointer 使用持久化后端,保证暂停期间状态不丢失
  • 审批界面把订单号、金额、理由完整呈现给复核者
  • 为审批与执行动作保留审计记录

模型名称、可用区域、配额与计费方式都可能变化,具体请以各服务官网当前信息为准。

注意事项

  • 检查点必须在执行之前。动作已经发生再请人审批,复核就失去意义。
  • Agent 不应拿到执行工具。调查用的 Agent 只挂载只读工具,退款工具只出现在审批通过后的节点里。
  • 不要让 LLM 编造订单信息。商品、金额、状态都应通过工具从受控来源检索。
  • 没有 checkpointer 就无法暂停恢复。interrupt() 依赖状态持久化。
  • 审批边界因组织而异。哪些动作需要人工复核,应结合自身政策、风险容忍度与错误后果来定。
  • 结构化输出优于自由文本。用 Pydantic 模型约束提案格式,后续节点才能可靠取值。