LangGraph 实战：给 AI 智能体加上「人工审批」断点

🔧 LangGraph 实战：给 AI 智能体加上「人工审批」断点，让工具调用更安全可控

💡 一句话总结：用 interrupt_before 在工具执行前插入人工确认环节，构建可审计、可干预的 Agent 工作流。

---

🎯 为什么需要「断点审批」？

在真实业务场景中，让 AI 直接调用工具（如数据库写入、支付接口、文件删除）风险极高。
我们希望在 关键操作前 插入一个人工确认环节：

用户提问 → AI 分析 → 决定调用工具 → ⚠️ 暂停等待审批 → 用户确认 → 执行工具 → 返回结果

这正是 LangGraph 的 interrupt_before 机制的用武之地。今天我们就用一段完整的代码，拆解如何实现这个「安全断点」。

---

🧱 代码结构全景图

graph LR
    Start([开始]) --> Assistant[🤖 Assistant 节点]
    Assistant -->|需要工具？ | Condition{tools_condition}
    Condition -->|是 | Tools[🛠️ Tools 节点]
    Condition -->|否 | End([结束])
    Tools -->|执行后 | Assistant
    style Tools fill:#ff9f43,stroke:#333
    style Assistant fill:#54a0ff,stroke:#333

💡 图中橙色节点 Tools 就是我们设置断点的位置：interrupt_before=["tools"]

---

🔍 逐段解析：关键代码拆解

1️⃣ 环境变量 & 模型初始化

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")

_set_env("OPENAI_API_KEY")
_set_env("OPENAI_BASE_URL")  # 支持代理地址

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="deepseek-chat")

✅ 最佳实践：敏感信息绝不硬编码，通过 getpass 交互式输入更安全。

---

2️⃣ 定义工具函数 & 绑定模型

def multiply(a: int, b: int) -> int:
    """乘法函数"""
    return a * b

# ... add, divide 同理

tools = [add, multiply, divide]
llm_with_tools = llm.bind_tools(tools)  # 🔑 关键：让模型知道有哪些工具可用

📌 LangChain 机制：bind_tools 会将函数签名转换为 JSON Schema，供模型推理时参考。

---

3️⃣ 构建 LangGraph 工作流

from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition

builder = StateGraph(MessagesState)

# 添加节点
builder.add_node("assistant", assistant)  # AI 决策节点
builder.add_node("tools", ToolNode(tools))  # 工具执行节点

# 定义流程边
builder.add_edge(START, "assistant")
builder.add_conditional_edges("assistant", tools_condition)  # 自动路由
builder.add_edge("tools", "assistant")  # 工具执行后返回 AI 继续推理

🧠 核心概念：

• MessagesState：自动管理消息历史的状态容器
• tools_condition：根据 AI 输出是否含工具调用，自动决定下一步走向
• ToolNode：预构建节点，自动解析工具调用并执行

---

4️⃣ ⚡ 关键一步：设置断点 + 内存检查点

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()  # 持久化对话状态

graph = builder.compile(
    interrupt_before=["tools"],  # 🔥 在执行 tools 前暂停！
    checkpointer=memory          # 支持多轮对话状态恢复
)

🎯 interrupt_before=["tools"] 的作用：

• 执行流到达 tools 节点前自动暂停
• 返回当前状态，等待外部输入（如用户审批）
• 传入 None 继续执行，实现「断点续传」

---

5️⃣ 执行流程：带人工审批的交互

initial_input = {"messages": HumanMessage(content="Multiply 2 and 3")}
thread = {"configurable": {"thread_id": "2"}}  # 唯一会话标识

# 第一阶段：执行到断点
for event in graph.stream(initial_input, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()

# 等待用户决策
user_approval = input("你是否要调用工具？(yes/no): ")

if user_approval.lower() == "yes":
    # 第二阶段：继续执行
    for event in graph.stream(None, thread, stream_mode="values"):
        event['messages'][-1].pretty_print()
else:
    print("操作已被用户取消。")

🔄 执行流示意：

[用户输入] 
   ↓
[AI 分析 → 决定调用 multiply(2,3)] 
   ↓
[⚠️ 断点暂停] → 等待 input() 
   ↓
[用户 yes] → [执行 multiply] → [AI 整合结果] → [返回最终答案]

---

🧠 核心知识点总结

概念	作用	关键代码
bind_tools	让 LLM 知道可用工具及参数规范	llm.bind_tools(tools)
ToolNode	自动解析并执行工具调用	ToolNode(tools)
tools_condition	根据消息内容自动路由	add_conditional_edges(..., tools_condition)
interrupt_before	在指定节点前插入断点	compile(interrupt_before=["tools"])
MemorySaver	持久化对话状态，支持断点续传	checkpointer=MemorySaver()
stream_mode="values"	流式获取每一步状态变化	graph.stream(..., stream_mode="values")

---

🚀 扩展思考：还能怎么用？

✅ 场景 1：敏感操作二次确认

# 在调用支付/删除接口前中断
graph = builder.compile(interrupt_before=["payment_tool", "delete_tool"])

✅ 场景 2：多步审批工作流

# 不同工具设置不同审批策略
if tool_name in HIGH_RISK_TOOLS:
    require_approval = True

✅ 场景 3：可视化审批界面

将 input() 替换为 Web 接口，前端展示 AI 决策依据 + 工具参数，用户点击「批准/拒绝」按钮回调继续执行。

---

🎨 视觉建议（配图思路）

根据您对技术插画的偏好，推荐以下配图方案：

1. 封面图：3D 等距风格，中央为 LangGraph 状态机，发光节点通过数据流连接，tools 节点前有一个醒目的「暂停/审批」图标，配色采用深蓝+电光青+紫色点缀。

2. 流程图：扁平矢量风格，用不同颜色区分 assistant（蓝色）、tools（橙色）、END（灰色），断点处用⚡闪电或🔒锁形图标强调。

3. 代码高亮区：深色背景 + 语法高亮，关键行（如 interrupt_before）用渐变光晕突出。

4. 概念隐喻图：将 Agent 比作驾驶员，工具比作引擎，断点审批比作「红灯检查站」，增强记忆点。

---