使用LangGraph构建Human-in-the-loop智能体：Python实现带人工审批的退款流程

关键词：LangGraph, LangChain, 状态机, SOP, Human-in-the-loop

设想一下，当你用 AutoGen 这类多智能体协作框架打造了一个客服系统并上线后，老板却可能找上门来问：“昨天有个用户申请退款，你的 AI 没核对订单状态就直接同意了？那订单金额可是10万块，居然没经过我就批准了？”

这暴露了一个核心痛点：像 AutoGen 这种“多智能体自由讨论”的模式，非常适用于创意型任务，例如撰写研究报告或生成代码。但在处理业务型任务，如退款、审批、报销时，企业需要的是严谨的SOP（标准作业程序），绝不允许AI自由发挥或绕过关键审核环节。

那么，解决方案是什么？我们需要将 AI 从“自由讨论室”拉到“工厂流水线”上，对其行为进行精确的状态和流程控制。这正是 LangGraph 的使命——构建有状态、可控的图应用。

1. 为什么你需要 LangGraph？一个生动的类比

理解 LangGraph 的价值，我们可以用一个简单的类比：

• AutoGen 就像一场头脑风暴会议。参与者们畅所欲言，最终也可能产出不错的结果，但整个过程充满不确定性，更适合需要发散思维的创意工作。
• LangGraph 则像一个高度标准化的厨房流水线，例如制作汉堡的流程：
  1. 工位 1 (意图识别)：服务员查看订单，判断是制作“香辣鸡腿堡”还是“牛肉堡”（决定流程走向哪个分支）。
  2. 工位 2 (规则检查)：称重员检查肉饼重量是否达标（例如，检查退款金额是否超过1000元？）。
  3. 工位 3 (人工审批)：质检员发现异常（如肉饼颜色不对），立即按下暂停键，呼叫店长（人类）前来确认。只有店长说“行”，流水线才能继续；若说“不行”，则直接废弃当前产品。这就是 Human-in-the-loop (人在回路) 的精髓。
  4. 工位 4 (执行)：打包员封装产品，交付给顾客。

LangGraph 的核心价值在于，它让你能够用代码清晰地定义这张“流程图”，并强制 AI 严格按照预设的路径执行。最关键的是，它原生支持 Human-in-the-loop —— 在流程的关键决策点，系统会暂停并等待人类介入，只有获得人类批准后，AI 才能继续后续操作。

2. LangGraph 核心概念提炼

在 LangGraph 的世界里，你需要掌握几个基本概念：

1. State (状态) 📦：可以理解为流水线上流动的“包裹”。所有节点（工位）都读取和修改这个包裹里的数据，例如 order_id（订单号）、status（状态）等。
2. Node (节点) 👷：流水线上的“工位”。它可以是一个LLM调用、一个普通的 Python 函数，也可以是一次 API 请求。
3. Edge (边) 🔗：连接各个工位的“传送带”。
   • Conditional Edge (条件边) 🔀：一个智能分拣机。根据包裹里的信息（例如 amount > 1000），决定将其传送到“经理审批工位”还是“自动批准工位”。
4. START & END 🏁：整个流程图的入口和出口。

3. 实战项目：构建带人工审批的智能客服退款流水线

我们将实现一个严谨的退款处理系统，确保大额退款必须经过人工审核。

第一步：安装依赖

首先，通过 pip 安装必要的库。

pip install langgraph langchain-openai langchain-core

• langgraph：本次的核心库，负责定义和管理整个状态图（流水线）。
• langchain-core：LangChain 的基础组件，定义了如 BaseMessage（消息格式）等通用标准。
• langchain-openai：连接 OpenAI 模型的桥梁，如果节点需要调用 GPT 进行意图识别等，会用到它。

第二步：流程设计

我们的退款审批流程图如下所示，清晰地展示了从用户输入到最终决策的完整路径，特别是金额超过阈值时触发的人工审批环节。

第三步：核心代码拆解

1. 定义状态（State）

状态定义了在流程中传递的数据结构。

from typing import TypedDict, Annotated, List, Optional
import operator
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add] # 聊天记录 (使用 operator.add 自动追加新消息)
    order_id: Optional[str]        # 订单号
    amount: Optional[float]        # 订单金额
    # 状态机核心字段：
    # pending: 初始状态
    # waiting_approval: 等待经理审批 (金额 > 1000)
    # approved: 已批准 (金额 < 1000 或 经理同意)
    # rejected: 已拒绝 (经理拒绝)
    refund_status: str

2. 定义节点（Node）

节点是执行具体工作的函数。

def node_process_refund(state: AgentState):
    """
    退款规则检查节点
    输入: state['order_id']
    输出: 更新 state['amount'] 和 state['refund_status']
    """
    # 模拟查询数据库
    def get_order_amount(order_id):
        # 假设所有订单都是 1200 元 (大于 1000，触发审批)
        return 1200.0

    amount = get_order_amount(state['order_id'])

    # 规则引擎逻辑
    if amount > 1000:
        print("⚠️ 金额 > 1000, 需要经理审批")
        # 返回的部分数据会自动 merge 到 State 中
        return {"amount": amount, "refund_status": "waiting_approval"}
    else:
        print("✅ 金额 < 1000, 自动批准")
        return {"amount": amount, "refund_status": "approved"}

def node_human_approval(state: AgentState):
    """
    人工审批节点
    注意：这里其实是个空函数，因为我们会在进入这个节点前使用 interrupt_before 暂停。
    当人工 update_state 后，会直接跳过这个节点的执行，进入下一个节点。
    """
    pass

3. 构建图并设置人工介入点

这是实现 Human-in-the-loop 的关键。interrupt_before 参数告诉 LangGraph：在进入指定节点前暂停执行，并将当前状态保存起来，等待外部指令。

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

# 构建图
workflow = StateGraph(AgentState)

# 1. 添加节点 (Add Nodes)
workflow.add_node("process", node_process_refund)
workflow.add_node("human_approval", node_human_approval)

# 2. 设置入口 (Set Entry Point)
workflow.set_entry_point("process")

# 3. 添加条件边 (Conditional Edges)
# 相当于流水线上的“分拣机”
def route_process(state):
    # 如果状态是 waiting_approval，就传送到 human_approval 工位
    if state['refund_status'] == "waiting_approval":
        return "human_approval"
    # 否则直接结束 (END)
    return END

# 将 process 节点连接到分拣逻辑
workflow.add_conditional_edges("process", route_process)

# 4. 添加普通边 (Edges)
# 经理审批完，直接结束
workflow.add_edge("human_approval", END)

# 5. 编译图 (Compile)
# memory: 用于保存图的运行状态 (Checkpointer)
memory = MemorySaver()

app = workflow.compile(
    checkpointer=memory,
    # interrupt_before: 在进入这些节点之前，强制暂停！
    # 这是实现 Human-in-the-loop 的关键
    interrupt_before=["human_approval"]
)

4. 运行与人工交互

当流程因 interrupt_before 暂停时，程序会挂起。此时可以等待人类（如经理）做出决策，然后恢复执行。

# 1. 配置 (Config)
# thread_id: 每个对话线程的唯一ID，用于恢复状态
config = {"configurable": {"thread_id": "thread-1"}}

# 2. 初始运行 (Run)
print("🚀 系统启动...")
for event in app.stream({"order_id": "BIG888"}, config=config):
    # stream 会逐步输出每个节点的执行结果
    pass

# 3. 检查断点 (Check Interrupt)
state = app.get_state(config)
if state.next and "human_approval" in state.next:
    # 发现下个节点是 human_approval，说明被暂停了！
    print(f"🛑 触发风控！等待经理审批... (金额: {state.values['amount']})")

    user_input = input("经理是否批准？(yes/no): ")

    if user_input == "yes":
        # 4. 人工干预 (Human Intervention)
        print("✅ 经理已批准，更新状态为 approved")
        app.update_state(config, {"refund_status": "approved"})
    else:
        print("❌ 经理已拒绝，更新状态为 rejected")
        app.update_state(config, {"refund_status": "rejected"})

    # 5. 恢复执行 (Resume)
    print("▶️ 继续执行后续流程...")
    for event in app.stream(None, config=config):
        pass

第四步：运行结果展示

执行上述代码，你将看到如下交互过程，完美演示了人工介入的流程：

🚀 系统启动...
⚠️ 金额 > 1000, 需要经理审批
🛑 触发风控！等待经理审批... (金额: 1200.0)
经理是否批准？(yes/no): yes
✅ 经理已批准，更新状态为 approved
▶️ 继续执行后续流程...

看，这就是 Human-in-the-loop 的魅力所在。无论 AI 运行得多快，在预设的关键节点上它都会停下来等待人类的最终裁决。这种“暂停 -> 人工判断 -> 修改状态 -> 继续执行”的能力，是传统自动化脚本难以实现的，也是构建负责任 AI (Responsible AI) 应用的重要基石。

总结与展望

简单来说，AutoGen 像擅长发散思维的“文科生”，而 LangGraph 则是逻辑严谨的“理科生”。在真实的、对流程和风控有严格要求的商业系统中（如 ERP、CRM、工单系统），LangGraph 提供的确定性和可控性，往往是技术决策者更看重的“定海神针”。

通过本文的实战，你不仅学会了如何使用 Python 和 LangGraph 构建一个带有人工审批环节的智能体流程，更重要的是理解了状态图（State Graph）如何将复杂的业务逻辑变得清晰、可控且易于维护。这种模式可以轻松扩展到客服路由、内容审核、复杂单据处理等多种企业级应用场景。

如果你想深入了解 LangChain 生态或其他 AI 工程化实践，欢迎在云栈社区与更多开发者交流讨论。在真正的 AI 应用落地过程中，除了功能实现，如何监控和评估 AI 的稳定性与性能同样至关重要，这便进入了 LLMOps 的领域。