第 13 期 | Human in the Loop (HITL)：审批节点的建立
本期副标题：主编写好大纲准备动笔前，使用 interrupt_before 等待人类点击"同意"再执行。
各位极客们，欢迎回到《LangGraph 多智能体专家课》。我是你们的老朋友，那个在代码堆里摸爬滚打十年的 AI 架构师。
在前面的 12 期里，我们的「AI 万能内容创作机构 (AI Content Agency)」已经初具规模。我们有了运筹帷幄的 Planner（策划），有了博览群书的 Researcher（研究员），还有了奋笔疾书的 Writer（主笔）。看着它们在代码里自动流转，你是不是有一种当上赛博资本家的快感？
但是，现实往往会给你一记响亮的耳光。
昨天，我的 Planner 脑洞大开，给一篇关于“量子计算”的文章定了一个大纲，里面竟然有一章叫“如何用量子计算炒股并实现财富自由”。然后 Writer 毫不犹豫地顺着这个离谱的大纲写了 3000 字的废话。结果？不仅浪费了大量的 Token（那可都是白花花的银子啊！），还产出了一堆不可用的垃圾。
这说明什么？完全脱缰的 AI 是危险的，也是昂贵的。 在关键的决策节点，我们需要引入人类的智慧来进行把控。这就是我们今天的主题：Human in the Loop (HITL，人类在环)。
今天，我们将为我们的 Agency 引入一位真正的“主编”——也就是屏幕前的你。在 Planner 写好大纲、Writer 准备动笔之前，系统必须停下来，等你大喊一声“准奏”，流程才能继续。
🎯 本期学习目标
通过本期实战，你将把以下技能收入囊中：
- 理解 HITL 的底层逻辑：为什么图状态（State）的持久化是实现人类在环的先决条件？
- 掌握
interrupt_before的魔法：如何在 LangGraph 编译阶段精准设置断点，让工作流在悬崖边刹车。 - 熟练运用 Thread ID (线程 ID)：学习如何唤醒一个沉睡（暂停）的图，并让它沿着之前的状态继续狂奔。
- 重构 Agency 工作流：在 Planner 和 Writer 之间建立坚不可摧的“人类审批墙”。
📖 原理解析
在传统的代码逻辑中，程序要么运行，要么结束。如果你想让程序中途停下来等你，通常只能用 input() 阻塞线程，但这在现代的异步网络服务（比如 Web API）中是绝对行不通的。
LangGraph 是如何优雅地解决这个问题的呢？答案是：状态快照 (State Snapshot) + 检查点机制 (Checkpointer)。
想象一下你在打单机游戏（比如《黑神话：悟空》）。你打完了第一个 Boss（Planner 生成了大纲），前面就是更难的 Boss（Writer 开始写长文）。为了防止翻车，你会怎么做？存档（Save Game）！
LangGraph 的 HITL 就是这个逻辑：
- 当图运行到你设置的断点（比如
interrupt_before=["writer"]）时。 - LangGraph 会利用
Checkpointer（比如MemorySaver或数据库）把当前的所有状态（State）拍一张快照并存下来。 - 然后，图的执行直接结束（挂起），释放计算资源。
- 等人类在前端 UI 看了大纲，点击了“同意”按钮后。
- 我们带着当时的
thread_id（存档槽位）重新调用图，LangGraph 会读取快照，从断点处无缝恢复执行。
老规矩，一图胜千言。来看看我们今天重构后的 Agency 工作流：
stateDiagram-v2
%% 定义样式
classDef aiNode fill:#e1f5fe,stroke:#0288d1,stroke-width:2px,color:#000
classDef humanNode fill:#fff3e0,stroke:#f57c00,stroke-width:2px,stroke-dasharray: 5 5,color:#000
classDef stateNode fill:#e8f5e9,stroke:#388e3c,stroke-width:2px,color:#000
%% 节点定义
Start((START))
Planner[Planner Agent\n(生成内容大纲)]:::aiNode
Checkpoint[(Checkpointer\n保存状态快照)]:::stateNode
Human{人类审批\n(HITL)}:::humanNode
Writer[Writer Agent\n(根据大纲撰写正文)]:::aiNode
End((END))
%% 流程连接
Start --> Planner : 接收用户主题
Planner --> Checkpoint : 大纲生成完毕
Checkpoint --> Human : interrupt_before='Writer'
Human --> Writer : 人类点击"同意"\n(传入 thread_id 继续)
Human --> Planner : 人类点击"驳回"\n(修改状态,重新生成)
Writer --> End : 文章撰写完毕敲黑板！这张图里最核心的不是那几个 Agent，而是那个虚线框的人类节点和 Checkpointer。没有 Checkpointer，图就失忆了，你就算同意了，它也不知道之前的大纲是什么。
💻 实战代码演练
废话不多说，Show me the code。今天我们将用 Python 配合 LangGraph 最新的 API 来实现这个功能。为了让大家看清本质，我将使用 Mock（模拟）的 LLM 调用，把注意力全部集中在图的控制流上。
前置准备：请确保你安装了
langgraph。pip install langgraph
1. 定义状态与节点
首先，我们定义 Agency 的状态，以及 Planner 和 Writer 两个节点。
import time
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
# ==========================================
# 1. 定义状态 (State) - 我们的"游戏存档"数据结构
# ==========================================
class AgencyState(TypedDict):
topic: str # 用户的初始主题
outline: Optional[str] # Planner 生成的大纲
article: Optional[str] # Writer 生成的正文
approval_status: str # 人类审批状态：'pending', 'approved', 'rejected'
# ==========================================
# 2. 定义节点 (Nodes) - 我们的打工人
# ==========================================
def planner_node(state: AgencyState) -> dict:
"""
Planner 节点：负责根据主题生成大纲
[EN] Planner Node: Responsible for generating an outline based on the topic.
"""
print("\n[Planner 👨💼] 正在疯狂构思大纲中...")
time.sleep(1) # 模拟思考时间
topic = state["topic"]
# 模拟 LLM 生成的大纲
mock_outline = f"""
主题：《{topic}》
1. 什么是{topic}？
2. {topic}的核心原理
3. {topic}的未来展望
"""
print("[Planner 👨💼] 大纲生成完毕！")
# 返回更新的状态
return {
"outline": mock_outline,
"approval_status": "pending" # 设置为待审批
}
def writer_node(state: AgencyState) -> dict:
"""
Writer 节点：负责根据大纲撰写正文
[EN] Writer Node: Responsible for writing the article based on the outline.
"""
print("\n[Writer ✍️] 收到主编的通过指令，开始根据大纲码字...")
time.sleep(1)
outline = state["outline"]
# 模拟 LLM 根据大纲写文章
mock_article = f"这是关于大纲【{outline.strip()}】的详细正文...\n(此处省略一万字)"
print("[Writer ✍️] 文章撰写完毕！可以下班了！")
return {"article": mock_article}
2. 构建带断点的计算图
这是本期的高光时刻。看仔细了，我们是如何把 MemorySaver 和 interrupt_before 结合起来的。
# ==========================================
# 3. 构建图 (Graph Construction)
# ==========================================
builder = StateGraph(AgencyState)
# 添加节点
builder.add_node("planner", planner_node)
builder.add_node("writer", writer_node)
# 定义边（工作流走向）
builder.add_edge(START, "planner")
builder.add_edge("planner", "writer") # 注意：这里虽然直接连向writer，但我们会在编译时打断它
builder.add_edge("writer", END)
# 核心步骤 1：实例化一个 Checkpointer (内存版，生产环境可用 PostgresSaver)
memory = MemorySaver()
# 核心步骤 2：编译图，并设置断点！
# interrupt_before=["writer"] 意味着：在执行 writer 节点【之前】，暂停图的执行。
graph = builder.compile(
checkpointer=memory,
interrupt_before=["writer"]
)
print("✅ Agency 工作流编译成功，已开启 HITL 模式。")
3. 模拟实际运行：从暂停到恢复
现在，让我们模拟一个真实的业务场景：用户提交需求 -> 图暂停 -> 终端提示用户确认 -> 用户输入同意 -> 图继续执行。
# ==========================================
# 4. 模拟运行 (Simulation Loop)
# ==========================================
def run_agency():
# 必须配置 thread_id，这是图用来识别"存档"的唯一凭证
# [EN] thread_id is crucial. It's the unique identifier for the "save slot".
thread_config = {"configurable": {"thread_id": "agency_task_001"}}
initial_state = {
"topic": "LangGraph 多智能体架构",
"outline": None,
"article": None,
"approval_status": "none"
}
print("\n🚀 [系统] 启动 Agency，接收到新任务...")
# 第一阶段运行：它会在 writer 节点前自动停下
for event in graph.stream(initial_state, config=thread_config):
for node_name, node_state in event.items():
pass # 节点内部已经打印了信息，这里不做额外处理
# 此时，图已经暂停了。我们可以检查图的当前状态
current_state = graph.get_state(thread_config)
# next 属性会告诉你，如果恢复运行，下一个将要执行的节点是什么
# 如果 next 有值，说明图被 interrupt 了
if current_state.next:
print(f"\n⏸️ [系统] 工作流已暂停。下一个将要执行的节点是: {current_state.next}")
print(f"📄 [系统] 当前生成的大纲内容如下：\n{current_state.values.get('outline')}")
# 模拟人类审批过程
user_input = input("👨⚖️ [人类主编] 大纲如上，是否同意让 Writer 开始撰写？(输入 'y' 同意，其他退出): ")
if user_input.strip().lower() == 'y':
print("\n👨⚖️ [人类主编] 审批通过！放行！")
# 核心步骤 3：恢复执行！
# 传入 None 作为输入状态，表示"不需要修改当前状态，直接顺着往下走"
# 必须传入相同的 thread_config 才能找到之前的存档
for event in graph.stream(None, config=thread_config):
pass
else:
print("\n👨⚖️ [人类主编] 审批驳回！任务终止。")
else:
print("\n✅ [系统] 任务已全部执行完毕。")
# 运行我们的测试
if __name__ == "__main__":
run_agency()
运行效果展示
当你运行这段代码时，你的终端会发生这样的交互：
✅ Agency 工作流编译成功，已开启 HITL 模式。
🚀 [系统] 启动 Agency，接收到新任务...
[Planner 👨💼] 正在疯狂构思大纲中...
[Planner 👨💼] 大纲生成完毕！
⏸️ [系统] 工作流已暂停。下一个将要执行的节点是: ('writer',)
📄 [系统] 当前生成的大纲内容如下：
主题：《LangGraph 多智能体架构》
1. 什么是LangGraph 多智能体架构？
2. LangGraph 多智能体架构的核心原理
3. LangGraph 多智能体架构的未来展望
👨⚖️ [人类主编] 大纲如上，是否同意让 Writer 开始撰写？(输入 'y' 同意，其他退出): y
👨⚖️ [人类主编] 审批通过！放行！
[Writer ✍️] 收到主编的通过指令，开始根据大纲码字...
[Writer ✍️] 文章撰写完毕！可以下班了！
看到了吗？图在 writer 执行前硬生生地刹车了！ 整个上下文（大纲内容）被完美保存在了 thread_id: agency_task_001 中。当你输入 y 并再次调用 stream(None, config) 时，它直接从断点处苏醒，完成了剩下的工作。
坑与避坑指南
在引入 HITL 时，有很多初学者（甚至一些老鸟）会踩进深坑。作为导师，我必须在这里给你们拉响警报：
💣 坑一：忘了加 Checkpointer，图直接"失忆"
症状：设置了 interrupt_before，但是图一跑完第一阶段，你想继续时，它报错说找不到状态，或者从头开始跑。
诊断：你没有在 compile() 里传入 checkpointer。
避坑：HITL 必须依赖状态持久化。没有 MemorySaver（或 Postgres/Redis Saver），LangGraph 根本不知道图暂停在哪里。记住公式：HITL = Checkpointer + interrupt_before/after。
💣 坑二：Thread ID 混乱导致"串线"
症状：张三点击了同意，结果李四的文章被发布了。
诊断：在恢复图的时候，传入了错误的 thread_id。
避坑：在实际的 Web 后端开发中（比如使用 FastAPI），你必须将 thread_id 与你的数据库中的 Task ID 或 User ID 强绑定。图挂起时，前端轮询或等待；用户点击按钮后，前端把 Task ID 传回后端，后端拼装成 thread_config 去唤醒图。永远不要在生产环境写死 thread_id！
💣 坑三：传入新的状态覆盖了旧状态，导致逻辑崩盘
症状：恢复图执行时，传入了初始状态，导致已经生成的大纲不见了。
诊断：在第二阶段调用时写成了 graph.stream(initial_state, config)。
避坑：如果你只是想让图单纯地继续执行，不要传入任何新状态，传 None！即 graph.stream(None, config)。LangGraph 看到 None 就知道：“哦，老板没新指示，我接着干活就行。”
(进阶剧透：如果你想在暂停期间修改大纲，你可以通过 graph.update_state() 来实现，这个高级操作我们会在下一期“状态的时间旅行”中详细剖析！)
📝 本期小结
今天，我们给狂奔的 AI 加上了缰绳。我们学习了：
- HITL 的本质：基于 Checkpointer 的状态快照与挂起。
- 断点设置：通过
interrupt_before=["writer"]建立人类审批墙。 - 图的唤醒：通过相同的
thread_id和stream(None)让工作流继续。
在我们的 AI Content Agency 中，引入“主编审批”节点是一个质的飞跃。它意味着我们从“盲盒开奖”式的 AI 生成，正式迈向了**“人机协同 (Human-AI Collaboration)”**的工业级工作流。
课后思考题： 如果人类主编看了大纲，觉得不满意，不想直接退出，而是希望打回给 Planner 重新写，我们的图和代码应该怎么改？
发挥你们的极客精神去尝试一下吧！我们第 14 期见，届时我将带你解锁 LangGraph 中最酷炫的功能之一：图状态的修改与时间旅行 (Time Travel)。下课！