Chuoer47's Blog

Appearance

人机交互（Human-in-the-Loop） ​

人机交互（Human-in-the-Loop，HITL）是 LangGraph 的杀手级特性。它允许 Agent 在执行过程中暂停，等待人类审批、修改或提供额外信息，然后恢复执行。这在生产环境中至关重要——你不会希望 AI 在没有人类确认的情况下发送邮件、执行代码或修改数据库。

为什么需要 HITL ​

AI Agent 的典型风险场景：

Agent: "我找到了所有符合条件的用户，准备删除他们的账号。"
       ↓
  [没有 HITL] → 直接执行，无法撤回
  [有 HITL]   → 暂停，等待人类确认

HITL 的核心价值：

场景说明
安全审批删除数据、发送邮件、执行代码前确认
质量控制AI 生成的内容需要人工审核
信息补充AI 需要人类提供缺失信息
纠错人类发现 AI 走偏时及时纠正
合规法律/政策要求人工审核某些操作

三种 HITL 机制 ​

LangGraph 提供三种不同粒度的人机交互方式：

1. interrupt_before — 节点执行前暂停 ​

在某个节点执行之前暂停，等待人类确认：

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

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("dangerous_action", delete_node)
graph.add_edge(START, "agent")
graph.add_edge("agent", "dangerous_action")
graph.add_edge("dangerous_action", END)

# 在 dangerous_action 执行前暂停
app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["dangerous_action"],  # ← 关键
)

2. interrupt_after — 节点执行后暂停 ​

在某个节点执行之后暂停，检查结果：

python
# 在 agent 节点执行后暂停（查看 AI 的输出再决定是否继续）
app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_after=["agent"],  # ← 关键
)

3. interrupt() 函数 — 动态暂停（推荐） ​

在节点内部根据条件动态决定是否暂停：

python
from langgraph.types import interrupt

def agent_node(state: AgentState) -> dict:
    response = llm.invoke(state["messages"])

    # 检查是否有工具调用需要审批
    if response.tool_calls:
        for tc in response.tool_calls:
            if tc["name"] in ["delete_file", "send_email", "execute_code"]:
                # 动态暂停，等待人类审批
                human_decision = interrupt({
                    "type": "approval",
                    "action": tc["name"],
                    "args": tc["args"],
                    "message": f"Agent 想要执行 {tc['name']}，是否允许？",
                })

                if human_decision == "deny":
                    # 人类拒绝，取消工具调用
                    response.tool_calls = [
                        t for t in response.tool_calls if t != tc
                    ]

    return {"messages": [response]}

interrupt() 详解 ​

interrupt() 是 LangGraph 最灵活的 HITL 机制。它的工作方式：

  1. 暂停：执行到 interrupt() 时，图的执行暂停
  2. 保存状态：Checkpointer 自动保存当前状态
  3. 返回给调用者：调用者收到暂停信号和 interrupt() 传入的值
  4. 人类操作：人类查看信息、做出决策
  5. 恢复执行：调用者传入人类的决策，图从暂停点恢复

interrupt() 的返回值 ​

python
def my_node(state: AgentState) -> dict:
    # interrupt() 的参数会返回给调用者（作为暂停的原因）
    human_input = interrupt("请输入你的名字：")

    # 当图恢复执行时，human_input 就是人类传入的值
    return {"messages": [("assistant", f"你好，{human_input}！")]}

与调用者交互 ​

python
# 1. 运行图（会暂停）
config = {"configurable": {"thread_id": "1"}}
result = app.invoke({"messages": [("user", "你好")]}, config=config)

# 2. 检查是否暂停
state = app.get_state(config)
if state.next and "__interrupt__" in str(state.next):
    # 获取 interrupt 的值
    interrupt_value = state.tasks[0].interrupts[0].value
    print(interrupt_value)  # "请输入你的名字："

    # 3. 人类输入后恢复
    app.invoke(Command(resume="张三"), config=config)

Command 对象 ​

Command 是 LangGraph 用于恢复中断的特殊对象：

python
from langgraph.types import Command

# 恢复执行，传入人类的输入
app.invoke(Command(resume="人类的回答"), config=config)

# 恢复执行，传入复杂对象
app.invoke(Command(resume={
    "approved": True,
    "reason": "确认执行",
    "modifications": {"limit": 100},
}), config=config)

完整 HITL 工作流 ​

下面是一个完整的审批工作流示例：

python
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 状态
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    approved_actions: list

# 需要审批的工具
@tool
def delete_user(user_id: str) -> str:
    """删除用户账号"""
    return f"用户 {user_id} 已删除"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件"""
    return f"邮件已发送到 {to}"

# 工具定义（带审批）
DANGEROUS_TOOLS = {"delete_user", "send_email"}

# Agent 节点
def agent_node(state: AgentState) -> dict:
    llm = ChatOpenAI(model="gpt-4o").bind_tools([delete_user, send_email])
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 审批节点
def approval_node(state: AgentState) -> dict:
    last_message = state["messages"][-1]
    if not last_message.tool_calls:
        return {}

    approved_calls = []
    for tc in last_message.tool_calls:
        if tc["name"] in DANGEROUS_TOOLS:
            # 需要审批
            decision = interrupt({
                "type": "tool_approval",
                "tool": tc["name"],
                "args": tc["args"],
                "message": f"Agent 想要调用 `{tc['name']}`，参数：{tc['args']}。是否允许？",
            })
            if decision.get("approved"):
                approved_calls.append(tc)
        else:
            # 安全工具，直接执行
            approved_calls.append(tc)

    # 更新消息中的 tool_calls（只保留已审批的）
    return {
        "messages": [last_message.model_copy(update={"tool_calls": approved_calls})],
        "approved_actions": [tc["name"] for tc in approved_calls],
    }

# 工具执行节点
def tool_node(state: AgentState) -> dict:
    last_message = state["messages"][-1]
    results = []
    tools_map = {"delete_user": delete_user, "send_email": send_email}

    for tc in last_message.tool_calls:
        tool = tools_map.get(tc["name"])
        if tool:
            result = tool.invoke(tc["args"])
            results.append({"role": "tool", "content": result, "tool_call_id": tc["id"]})

    return {"messages": results}

# 路由
def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "approval"
    return END

# 构建图
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("approval", approval_node)
graph.add_node("tools", tool_node)

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {
    "approval": "approval",
    END: END,
})
graph.add_edge("approval", "tools")
graph.add_edge("tools", "agent")

# 编译
app = graph.compile(checkpointer=MemorySaver())

# 使用
config = {"configurable": {"thread_id": "demo"}}

# 发送请求
result = app.invoke({
    "messages": [("user", "帮我删除用户 12345")]
}, config=config)

# 检查是否需要审批
state = app.get_state(config)
if state.next:
    interrupt_info = state.tasks[0].interrupts[0].value
    print(interrupt_info["message"])
    # "Agent 想要调用 `delete_user`，参数：{'user_id': '12345'}。是否允许？"

    # 人类审批
    app.invoke(Command(resume={"approved": True}), config=config)

使用 create_react_agent 的 HITL ​

create_react_agent 也支持 HITL：

python
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver

agent = create_react_agent(
    model=llm,
    tools=[delete_user, send_email],
    checkpointer=MemorySaver(),
    interrupt_before=["tools"],  # 在工具执行前暂停
)

# 运行
config = {"configurable": {"thread_id": "1"}}
result = agent.invoke({"messages": [("user", "删除用户123")]}, config=config)

# 检查暂停
state = agent.get_state(config)
print(state.next)  # ('tools',) → 在 tools 节点前暂停

# 人类确认后恢复
agent.invoke(Command(resume=True), config=config)  # 简单的确认模式

多轮 HITL ​

复杂的审批流程可能需要多轮交互：

python
def complex_approval(state: AgentState) -> dict:
    last_message = state["messages"][-1]

    for tc in last_message.tool_calls:
        # 第一轮：确认操作
        confirm = interrupt({
            "step": "confirm",
            "message": f"确认执行 {tc['name']}？",
        })
        if not confirm.get("approved"):
            continue

        # 第二轮：确认参数
        params_ok = interrupt({
            "step": "verify_params",
            "message": f"参数确认：{tc['args']}",
        })
        if not params_ok.get("approved"):
            # 允许修改参数
            new_args = params_ok.get("modified_args", tc["args"])
            tc["args"] = new_args

    return {"messages": [last_message]}

HITL 与流式输出 ​

结合流式输出，可以实时显示 Agent 的思考过程：

python
config = {"configurable": {"thread_id": "stream-demo"}}

# 流式执行
for event in app.stream(input_data, config=config):
    for node, output in event.items():
        print(f"[{node}]", output)

        # 检查是否中断
        if node == "__interrupt__":
            print("需要人类介入！")
            # 人类操作后继续...

最佳实践 ​

  1. 明确中断点：不要在每个节点都加中断，只在关键操作前加
  2. 提供足够信息：interrupt() 的参数要包含足够的信息供人类决策
  3. 处理拒绝：人类拒绝后要有合理的降级逻辑
  4. 超时处理：生产环境要考虑中断超时
  5. 日志记录：记录所有人工决策，便于审计
python
# ✅ 好的实践
decision = interrupt({
    "action": "delete_user",
    "target": "user_12345",
    "reason": "用户请求注销",
    "risk_level": "high",
    "irreversible": True,
    "estimated_impact": "该用户的3个订单将被取消",
})

# ❌ 不好的实践
decision = interrupt("确认？")  # 信息太少

小结 ​

机制说明适用场景
interrupt_before节点执行前暂停已知的危险节点
interrupt_after节点执行后暂停审查 AI 输出
interrupt()动态暂停复杂的条件审批
Command(resume=...)恢复执行传入人类决策

下一篇 ​

→ 多 Agent 协作：Supervisor、Swarm、子图嵌套、Agent 通信模式。