为自主代理构建“人工干预”审批门控：LangGraph 实践指南-青云TOP-AI综合资源站平台|青云聚合API大模型调用平台|全网AI资源导航平台

📢 转载信息

原文链接：https://machinelearningmastery.com/building-a-human-in-the-loop-approval-gate-for-autonomous-agents/

原文作者：Iván Palomares Carrascosa

在本文中，您将学习如何在 LangGraph 中实现由状态管理的中间中断，以便智能体工作流可以在恢复执行前暂停以获得人工审批。

我们将涵盖的主题包括：

什么是状态管理中断，为什么它们在代理式 AI 系统中很重要。
如何定义一个具有共享代理状态和可执行节点的简单 LangGraph 工作流。
如何暂停执行，通过人工审批更新保存的状态，并恢复工作流。

继续阅读以获取所有信息。

为自主代理构建“人工干预”审批门控
图片由编辑提供

引言

在代理式 AI 系统中，当智能体的执行管道被有意停止时，我们就称之为状态管理中断。就像保存的电子游戏一样，暂停的智能体的“状态”——其活动变量、上下文、记忆和计划的动作——会被持久地保存，智能体处于睡眠或等待状态，直到外部触发器恢复其执行。

由于高度自主、基于智能体的 AI 应用的进步，状态管理中断的重要性日益增长，这主要有几个原因。它们不仅作为有效的安全护栏，可以在高风险环境中从本不可逆转的动作中恢复，而且还可以实现人工干预审批和纠正。人类监督者可以在执行基于不正确响应的动作之前，重新配置暂停的智能体的状态，并防止不良后果。

LangGraph，一个用于构建有状态大型语言模型 (LLM) 应用程序的开源库，支持具有人工干预机制和状态管理中断的基于智能体的[工作流](https://machinelearningmastery.com/what-are-workflows/)，从而提高了对错误的鲁棒性。

本文将所有这些元素整合在一起，并逐步展示如何使用 LangGraph 在 Python 中通过人工干预方法实现状态管理中断。虽然下面定义的大多数示例进程旨在由智能体自动执行，但我们也 [将展示](https://machinelearningmastery.com/how-to-build-a-human-in-the-loop-machine-learning-system/) 如何在需要人工审查的关键点停止工作流，然后恢复执行。

分步指南

首先，我们 pip install langgraph 并为这个实际示例进行必要的导入：

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

注意到导入的类之一名为 StateGraph。LangGraph 使用状态图来[建模](https://machinelearningmastery.com/category/modeling/) 涉及智能体的[循环](https://machinelearningmastery.com/how-to-create-a-while-loop-in-python/)、复杂工作流。存在状态，代表系统的共享内存（又名数据负载），以及节点，代表定义用于更新此状态的执行逻辑的动作。状态和节点都需要被显式定义和[检查点](https://machinelearningmastery.com/how-to-save-and-load-machine-learning-models/)。现在让我们这样做。

class AgentState(TypedDict):
    draft: str
    approved: bool
    sent: bool

智能体状态的结构与 Python 字典类似，因为它继承自 TypedDict。状态充当我们的“保存文件”，因为它在节点之间传递。

关于节点，我们将定义两个，每个节点代表一个动作：起草电子邮件和发送电子邮件。

def draft_node(state: AgentState):
    print("[Agent]: Drafting the email...")
    # The agent builds a draft and updates the state
    return {"draft": "Hello! Your server update is ready to be deployed.", "approved": False, "sent": False}

def send_node(state: AgentState):
    print(f"[Agent]: Waking back up! Checking approval status...")
    if state.get("approved"):
        print("[System]: SENDING EMAIL ->", state["draft"])
        return {"sent": True}
    else:
        print("[System]: Draft was rejected. Email aborted.")
        return {"sent": False}

draft_node() 函数模拟了智能体起草电子邮件的动作。要使智能体执行实际操作，您需要将模拟行为的 print() 语句替换为执行它的实际指令。这里要注意的关键细节是函数返回的对象：一个字典，其字段与我们之前定义的智能体状态类中的字段匹配。

同时，send_node() 函数模拟了发送电子邮件的动作。但有一个[陷阱](https://machinelearningmastery.com/how-to-avoid-common-machine-learning-pitfalls/)：用于人工干预机制的核心逻辑[位于](https://machinelearningmastery.com/how-to-implement-a-decision-tree-algorithm-for-classification/)此处，特别是在对 approved 状态的检查中。只有当 approved 字段被设置为 True——由人类（如我们将看到的）或通过模拟的人工干预——电子邮件才会被真正发送。同样，为了简单起见，通过简单的 print() 语句模拟了动作，将重点放在状态管理中断机制上。

还需要什么？智能体工作流由具有多个连接状态的图描述。让我们定义一个简单的、线性的动作序列，如下所示：

workflow = StateGraph(AgentState)

# Adding action nodes
workflow.add_node("draft_message", draft_node)
workflow.add_node("send_message", send_node)

# Connecting nodes through edges: Start -> Draft -> Send -> End
workflow.set_entry_point("draft_message")
workflow.add_edge("draft_message", "send_message")
workflow.add_edge("send_message", END)

为了实现保存智能体状态的数据库式机制，并在智能体即将发送消息时引入状态管理中断，我们使用以下代码：

# MemorySaver is like our "database" for saving states
memory = MemorySaver()

# THIS IS A KEY PART OF OUR PROGRAM: telling the agent to pause before sending
app = workflow.compile(
    checkpointer=memory,
    interrupt_before=["send_message"]
)

现在是真正的[行动](https://machinelearningmastery.com/how-to-turn-off-warnings-in-python/)。我们将执行上面定义的动作图。请注意，下面使用了一个线程 ID，以便内存可以跨执行跟踪工作流状态。

config = {"configurable": {"thread_id": "demo-thread-1"}}
initial_state = {"draft": "", "approved": False, "sent": False}

print("\n--- RUNNING INITIAL GRAPH ---")
# The graph will run 'draft_node', then hit the breakpoint and pause.
for event in app.stream(initial_state, config):
    pass

接下来是人工干预的时刻，在此期间流程暂停，并通过将 approved 设置为 True 来模拟人工审批：

print("\n--- GRAPH PAUSED ---")
current_state = app.get_state(config)
print(f"Next node to execute: {current_state.next}") # Should show 'send_message'
print(f"Current Draft: '{current_state.values['draft']}'")

# Simulating a human reviewing and approving the email draft
print("\n [Human]: Reviewing draft... Looks good. Approving!")

# IMPORTANT: the state is updated with the human's decision
app.update_state(config, {"approved": True})

这会恢复图并完成执行。

print("\n--- RESUMING GRAPH ---")
# We pass 'None', as the input tells the graph to just resume where it left off
for event in app.stream(None, config):
    pass

print("\n--- FINAL STATE ---")
print(app.get_state(config).values)

此模拟工作流打印的总体输出应如下所示：


--- RUNNING INITIAL GRAPH ---
[Agent]: Drafting the email...

--- GRAPH PAUSED ---
Next node to execute: ('send_message',)
Current Draft: 'Hello! Your server update is ready to be deployed.'

 [Human]: Reviewing draft... Looks good. Approving!

--- RESUMING GRAPH ---
[Agent]: Waking back up! Checking approval status...
[System]: SENDING EMAIL -> Hello! Your server update is ready to be deployed.

--- FINAL STATE ---
{'draft': 'Hello! Your server update is ready to be deployed.', 'approved': True, 'sent': True}

总结

本文通过引入人工干预机制，说明了如何在基于智能体的工作流中实现状态管理中断——在不希望完全自主的情况下，这是关键高风险场景中的一项重要能力。我们使用了 LangGraph，一个用于构建智能体驱动的 LLM 应用程序的强大库，来模拟受这些规则约束的工作流。

🚀 想要体验更好更全面的AI调用？

欢迎使用青云聚合API，约为官网价格的十分之一，支持300+全球最新模型，以及全球各种生图生视频模型，无需翻墙高速稳定，文档丰富，小白也可以简单操作。

目录CONTENT

为自主代理构建“人工干预”审批门控：LangGraph 实践指南

引言

分步指南

总结

评论区