基于本文回答

播面 播面

刷题像听歌，多听自然懂
0
评论
请从左侧选择一个分类

LangGraph 是如何支持Human-in-the-loop（人机协同）的？

知识点图片

LangGraph 是目前 AI Agent 开发中最强大的框架之一，它之所以强大，很大程度上归功于其原生且优雅的 Human-in-the-loop (HITL，人机协同) 支持。

LangGraph 的底层逻辑是一个状态机（State Machine）。通过结合“状态（State）”和“检查点（Checkpointer）”，LangGraph 可以随时暂停图的执行、等待人类介入，然后再从断点处恢复。

以下是 LangGraph 支持 HITL 的核心机制和具体工作流程：

1. 核心底层机制：检查点与线程（Checkpointer & Threads）

要实现人机协同，系统必须具备“记忆”能力，以便在暂停后恢复。

  • Thread ID（线程 ID）： 每次执行时，LangGraph 都会分配一个 Thread ID。这就好比一个对话会话。
  • Checkpointer（检查点）： LangGraph 在执行完图中的每一个节点（Node）后，都会将当前的总状态（State）保存到数据库中（可以是内存 MemorySaver，也可以是 SQLite/Postgres）。
  • 作用： 因为状态被持久化了，程序完全可以退出运行，等人类第二天来看，看完后再通过 Thread ID 唤醒并继续执行。

2. 设置断点（Breakpoints）

LangGraph 允许你在图的执行路径上显式地设置断点。主要有两种方式：

  • 静态断点 (interrupt_before / interrupt_after)：
    在编译图（compile）的时候，你可以指定在运行某个节点之前或之后暂停。
    • 场景： 比如有一个节点叫 execute_sql（执行数据库操作），你可以设置 interrupt_before=["execute_sql"]。当图运行到这里时，会自动抛出中断信号并停止。
  • 动态中断 (interrupt() 函数 - 新特性)：
    LangGraph 新版本引入了 types.interrupt。你可以在节点的 Python 函数内部根据逻辑动态触发中断。
    • 场景： 只有当购买金额 > 1000 元时，才触发 interrupt("需要主管审批")。

3. 人类介入的三种核心操作

当图运行到断点暂停后，人类可以执行以下三种主要操作：

A. 审批通过 (Approve / Continue)

人类查看当前状态（比如 AI 生成的邮件草稿），觉得没问题，直接让程序继续运行。

  • 操作： 只需要带着相同的 Thread ID，传入 None 再次调用执行命令即可。
    python 
    app.invoke(None, config={"configurable": {"thread_id": "1"}})

B. 修改状态 (Edit / Correct)

人类发现 AI 生成的内容有瑕疵，或者想强行改变 AI 的决策。LangGraph 允许人类像一个节点一样，直接修改当前状态。

  • 操作： 使用 update_state 方法。
    python 
    # 人类修改了 AI 准备发送的邮件内容
app.update_state(
    config={"configurable": {"thread_id": "1"}},
    {"email_body": "这是人类修改后的最终邮件内容。"}
)
# 然后继续执行
app.invoke(None, config)

C. 提供额外输入 (Provide Input)

AI 在执行过程中发现信息不足（比如需要用户输入验证码，或者需要用户补充说明）。

  • 操作： 通过动态中断向外部请求信息，人类将信息作为 Command(resume=...) 或者状态更新传回图中，图接收到新信息后继续运转。

4. 常见的人机协同应用场景

  1. 高风险操作拦截（Tool Approval）：
    在 Agent 调用危险工具（如：发送邮件、修改数据库、执行本地 Python 代码、执行自动交易）之前，必须由人类点击“同意”。
  2. 内容审核与编辑（Review & Edit）：
    Agent 负责写研报的初稿，写完后暂停。人类接管，修改其中的错别字和不准确的数据。修改完成后点击继续，Agent 拿着修改后的状态去执行下一步（如生成 PDF 或排版）。
  3. 多轮问询（Clarification）：
    当 Agent 发现用户的意图不明确时，主动挂起，向前端发送一个表单或问题，等用户填完表单（状态更新）后，再继续处理。

5. 简单的代码概念演示

以下是一个典型的“拦截并修改”的 LangGraph 工作流代码缩影：

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

# 1. 定义状态和图
builder = StateGraph(MyState)
builder.add_node("agent_draft", draft_email_node)
builder.add_node("send_email", send_email_node)

builder.add_edge(START, "agent_draft")
builder.add_edge("agent_draft", "send_email")
builder.add_edge("send_email", END)

# 2. 核心：添加记忆，并设置在发邮件前中断
memory = MemorySaver()
app = builder.compile(
    checkpointer=memory,
    interrupt_before=["send_email"]  # <--- 在这里设置断点
)

config = {"configurable": {"thread_id": "thread_123"}}

# 3. 第一次运行：执行完 agent_draft 后会自动暂停
app.invoke({"task": "给老板写一封请假信"}, config)

# 4. 人类介入：查看当前状态
current_state = app.get_state(config)
print("AI 草稿是:", current_state.values["email_draft"])

# 5. 人类介入：修改状态（人类觉得 AI 语气不够诚恳，自己改了）
app.update_state(config, {"email_draft": "老板我真的生病了，请假一天。"})

# 6. 继续运行：传入 None，图会从 send_email 节点继续执行
app.invoke(None, config)

总结

LangGraph 对 Human-in-the-loop 的支持不是通过复杂的 workaround 实现的，而是由其底层架构（状态持久化 + 有向图图论）原生决定的。它把 AI 的执行看作可以被随时冻结和解冻的“时间胶囊”，让人类可以无缝地穿插在 Agent 的思考和行动链路中，极大地提升了 Agent 在生产环境中的安全性和可用性。

00:00
00:00