在之前的章节中，我们一直致力于让 Agent 更自主（Autonomous）。但在现实的商业世界里配资炒股大全，“完全自主”往往意味着“完全失控”。
想象一下，你写了一个帮客户自动退款的 Agent。
用户：“我不想要了，退钱。”
Agent：“好的，已为您发起 100 万元的退款。”
如果缺乏人类的监管，这种幻觉（Hallucination）会导致灾难性的后果。
Human in the loop
Human-in-the-loop (HITL) 是一种设计模式，它允许 Agent 在执行关键步骤前“暂停”，等待人类的批准、修改或拒绝，然后再继续。这不仅是为了安全，更是为了让 AI 从“黑盒”变成“白盒”。
14.1 控制的频谱：从 HIL 到 HOTL
并不是所有的 Agent 都需要同样程度的人工干预。在设计架构时，我们需要根据风险等级选择不同的模式：
Human-in-the-loop (HITL)：强依赖。Agent 每做一步关键决策，都必须经过人确认。例如：自动转账系统。
Human-on-the-loop (HOTL)：弱监督。Agent 自主运行，人看着大屏幕。如果人发现不对劲，可以随时按下“紧急停止”按钮。例如：自动驾驶汽车。
Human-out-of-the-loop (HOOTL)：全自动。人只看结果。例如：推荐系统。
图14.1展示了这三种模式的架构差异。
图14.1 人类介入的三种控制级别
14.2 核心机制：interrupt_before
LangGraph 实现 HITL 的核心 API 是 interrupt_before 和 interrupt_after。
它的工作原理类似于编程时的 Debug 断点。当图运行到指定节点（例如 tools 节点）之前，会强制挂起（Suspend），并将当前状态保存到 Checkpoint 中。此时程序退出，等待下一次唤醒。
14.3 实战：构建“敏感邮件发送助手”
我们将利用 本地 Qwen3:8b 构建一个帮写邮件的 Agent。
风险点：Agent 可能会写出带有攻击性或错误的邮件并发出去。
解决方案：在调用 send_email 工具之前，暂停，让人类审查邮件内容。如果不满意，人类可以直接修改草稿。
14.3.1 定义工具与图结构
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver
import operator
# --- 1. 定义工具 ---
@tool
def send_email(recipient: str, content: str):
"""发送邮件给指定收件人"""
# 在真实场景中，这里会调用 SMTP 服务
return f" 邮件已发送给 {recipient}，内容长度: {len(content)}"
# --- 2. 定义图节点 ---
# 状态定义 (同第12章)
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], operator.add]
# 配置本地 Qwen3
llm = ChatOpenAI(
base_url="http://localhost:11434/v1",
api_key="ollama",
model="qwen3:8b",
temperature=0
)
tools = [send_email]
llm_with_tools = llm.bind_tools(tools)
def agent_node(state: AgentState):
return {"messages": [llm_with_tools.invoke(state["messages"])]}
tool_node = ToolNode(tools)
# --- 3. 构建图 (关键点：添加中断) ---
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("action", tool_node) # 我们把工具节点命名为 action
workflow.set_entry_point("agent")
# 简单的条件边逻辑
def should_continue(state):
if state["messages"][-1].tool_calls:
return "action"
return END
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("action", "agent")
# ★★★ 关键点：设置断点 ★★★
# 意思是：在进入 'action' 节点之前，暂停！
app = workflow.compile(
checkpointer=MemorySaver,
interrupt_before=["action"]
)
14.3.2 运行流程：审查与批准
现在我们来模拟一次完整的“人机交互”过程。
# 初始化配置
thread_config = {"configurable": {"thread_id": "email_task_1"}}
# === 阶段 1: AI 起草邮件 ===
print(" 用户指令: 给老板写封邮件请假。")
input_1 = {"messages": [HumanMessage(content="给老板写封邮件，说我明天病假。")]}
# 运行图 (它会在 action 之前停下来)
for event in app.stream(input_1, config=thread_config):
pass
print("\n⏸️ 系统暂停：等待人类审批...")
# === 阶段 2: 人类审查 ===
# 获取当前状态
snapshot = app.get_state(thread_config)
ai_msg = snapshot.values["messages"][-1]
# 提取 Agent 想要执行的动作
if ai_msg.tool_calls:
tool_call = ai_msg.tool_calls[0]
print(f"️ AI 申请执行: {tool_call['name']}")
print(f" 邮件草稿内容:\n{tool_call['args']['content']}")
# === 阶段 3: 人类决策 (Approve or Reject) ===
user_decision = input(">> 是否批准发送？(y/n/edit): ")
if user_decision == "y":
print(" 批准！继续执行...")
# 传入 None 表示“什么都不改，继续往下跑”
# 图会从断点处恢复，进入 'action' 节点执行发送
for event in app.stream(None, config=thread_config):
for k, v in event.items:
if "messages" in v:
print(f"系统反馈: {v['messages'][-1].content}")
elif user_decision == "edit":
print("✏️ 进入编辑模式...")
new_content = input(">> 请输入新的邮件内容: ")
# 修改状态 (Time Travel)
# 我们构造一个新的 ToolCall，替换掉旧的
new_tool_call = tool_call.copy
new_tool_call['args']['content'] = new_content
from langchain_core.messages import AIMessage
new_msg = AIMessage(content=ai_msg.content, tool_calls=[new_tool_call], id=ai_msg.id)
# 更新状态
app.update_state(thread_config, {"messages": [new_msg]}, as_node="agent")
print(" 修改已应用，继续执行...")
for event in app.stream(None, config=thread_config):
print("邮件已发送 (修改版).")
else:
print(" 拒绝发送。任务终止。")
# 这里可以选择不继续 stream，或者 update_state 注入一条“用户拒绝”的消息让 Agent 重新生成
执行结果如下：
执行后会出现一个确认界面：
图14.2 确认界面
如果同意（输入y）结果如下：
用户指令: 给老板写封邮件请假。
⏸️ 系统暂停：等待人类审批...
️ AI 申请执行: send_email
邮件草稿内容:
您好，因明天身体不适，需请假一天，望批准。感谢理解！
>> 是否批准发送？(y/n/edit): y
批准！继续执行...
系统反馈: 邮件已发送给 老板，内容长度: 26
系统反馈: 邮件已成功发送给老板，内容简洁明了地说明了请假原因。如果需要进一步补充病情说明或提供医疗证明，可后续补充说明。
如果不同意（输入n）结果如下
用户指令: 给老板写封邮件请假。
⏸️ 系统暂停：等待人类审批...
️ AI 申请执行: send_email
邮件草稿内容:
您好，因明天身体不适，需请假一天，望批准。感谢理解！
>> 是否批准发送？(y/n/edit): n
拒绝发送。任务终止。
14.4 设计模式：升级汇报 (Escalation)
除了审批，HITL 的另一种常见模式是 Escalation（升级汇报）。
当 Agent 对自己的回答**置信度（Confidence）**较低，或者遇到无法处理的异常时，它应该主动“举手”请求人类介入。
在 LangGraph 中，这可以通过在 Prompt 中加入逻辑来实现：
System Prompt: 如果你无法确定问题的答案，或者是涉及法律建议的问题，请调用 ask_human 工具。
图14.3 展示了这种升级机制。
图14.3 升级汇报模式：不懂就问
本章小结
本章我们探讨了 Agent 安全性的基石——人在回路（Human-in-the-loop）。
通过 LangGraph 的 interrupt_before 和 stream(None) 机制，我们把原本黑盒的 AI 执行流切开了一个口子，让人类可以随时向里面“注视”并“干预”。
对于高风险操作（金钱、数据删除），使用 Approval（审批） 模式。
对于低置信度任务，使用 Escalation（升级） 模式。
在下一章，我们将结合前面的所有知识（记忆、工具、HITL），构建一个能够真正根据用户画像进行“个性化”服务的智能系统。
思考与面试角
Q1：interrupt_after 和 interrupt_before 有什么区别？
参考答案：
interrupt_before=["action"]：在进入节点之前暂停。常用于审批（Approve），比如在执行“发送邮件”工具前停下，让你检查参数。
interrupt_after=["action"]：在节点执行之后暂停。常用于结果验收（Review），比如工具已经把 SQL 查出来了，停下来让人看看结果对不对，如果不对可以回滚或修改。
Q2：在 Web API（如 FastAPI）中如何实现 HITL？请求不能一直挂起等待用户输入啊？
参考答案：这是一个很好的工程问题。在 Web 服务中，我们不能用 input 阻塞线程。
正确的做法是：
用户发起请求 -> 后端运行 app.stream -> 遇到断点暂停 -> 后端保存 Checkpoint -> 返回前端 202 Accepted（附带一个 thread_id）。
用户在前端界面点击“批准” -> 前端发起新的 POST 请求（带上 thread_id 和批准指令 resume） -> 后端恢复 app.stream(None, config) 继续运行。
这就是异步人机交互的标准模式。
Q3：如果人类一直不批准，Checkpoint 会过期吗？
参考答案：这取决于 Checkpointer 的后端实现。如果是 MemorySaver，重启就没了。如果是 SqliteSaver 或 PostgresSaver，数据是持久化的，理论上可以永久保存。但在业务逻辑上，通常会设置一个定时任务（TTL）配资炒股大全，如果超过 24 小时未批准，则自动标记任务为“已过期/已拒绝”并清理状态。
华林优配提示：文章来自网络，不代表本站观点。
