人机交互 HITL

Human-in-the-Loop（人机交互）指在图执行过程中暂停，让人来审批、编辑状态、确认工具调用，然后再继续。这是把 AI 工作流从"全自动"变成"可控"的关键能力。

一、HITL 的典型场景

场景	例子
审批	发邮件、转账、删数据前要人确认
编辑状态	人工修正检索结果、改写用户输入后再跑
确认工具调用	LLM 要调"删除文件"工具前先问人
纠错	跑到一半发现错了，人工修正后继续

实现 HITL 的前提是有 Checkpointer（见 持久化与检查点），因为暂停后状态得存得住，人工处理后才能恢复。

二、interrupt_before / interrupt_after：在指定节点暂停

compile 时可以指定在某些节点之前/之后自动暂停：

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

# ...定义状态、节点...
graph = StateGraph(State)
# graph.add_node(...)

app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["send_email"],  # 执行到 send_email 节点前暂停
)

• interrupt_before=["X"]：执行到 X 之前暂停，等人工确认后再继续执行 X。
• interrupt_after=["X"]：执行完 X 之后暂停。

当 invoke 执行到暂停点时，会立即返回（不会执行该节点），状态停在暂停点。此时 get_state 的 .next 就是那个待执行的节点。

三、update_state：人工修改状态后继续

暂停后，你可以用 update_state 修改状态，然后再 invoke(None) 继续：

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

# 第一次 invoke：跑到 send_email 前暂停
app.invoke({"draft": "初稿内容"}, config=config)

# 看看现在停在哪
state = app.get_state(config)
print(state.next)  # ('send_email',)

# 人工修改草稿
app.update_state(config, {"draft": "人工修改后的内容"})

# 继续（传 None 表示不输入新内容，从暂停点继续）
app.invoke(None, config=config)

update_state 接收一个部分更新字典，行为和节点返回值类似——会按 reducer 合并。

四、interrupt() 函数：节点内主动暂停

除了在 compile 时指定暂停点，还可以在节点内部用 interrupt() 主动暂停，并向人类征求输入。interrupt(value) 的 value 会作为暂停信息暴露给外部，interrupt 的返回值会注入回状态。

from langgraph.types import interrupt

def review_node(state):
    # 把要审批的内容抛出去，等人类反馈
    feedback = interrupt({"to_review": state["draft"]})
    # feedback 是人类 resume 时传入的值
    return {"draft": state["draft"] + f" [审过:{feedback}]"}

恢复执行时，通过 Command(resume=...) 把人类输入传进去：

from langgraph.types import Command

# 第一次：会暂停在 review_node 的 interrupt 处
app.invoke({"draft": "初稿"}, config=config)

# 人工决定通过，传回 "approved"
app.invoke(Command(resume="approved"), config=config)

五、完整示例：工具调用前人工审批

下面这个例子：LLM 要调用工具前暂停，人工审批后才执行工具，否则跳过。

from typing import TypedDict, Annotated
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver

@tool
def send_email(to: str, body: str) -> str:
    """发送邮件给某人。"""
    return f"已发送给 {to}：{body}"

tools = [send_email]
tool_node = ToolNode(tools)

# 假 LLM：第一轮要发邮件，第二轮看到结果就结束
def fake_llm(state):
    msgs = state["messages"]
    has_result = any(isinstance(m, ToolMessage) for m in msgs)
    if not has_result:
        return {"messages": [AIMessage(content="", tool_calls=[
            {"name": "send_email", "args": {"to": "boss@x.com", "body": "涨薪申请"}, "id": "c1"}])]}
    return {"messages": [AIMessage(content="邮件已发送，流程结束")]}

graph = StateGraph(MessagesState)
graph.add_node("llm", fake_llm)
graph.add_node("tools", tool_node)
graph.add_edge(START, "llm")
graph.add_conditional_edges("llm", tools_condition)
graph.add_edge("tools", "llm")

# 关键：在 tools 节点前暂停，等人工审批
app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["tools"],
)

config = {"configurable": {"thread_id": "approval-1"}}

# 1. 第一次运行：LLM 决定发邮件，但 tools 前暂停
r1 = app.invoke({"messages": [HumanMessage(content="帮我给老板发涨薪邮件")]}, config=config)
print("暂停在:", app.get_state(config).next)  # ('tools',)

# 2. 人工查看 LLM 想调什么
state = app.get_state(config)
last_ai = [m for m in state.values["messages"] if isinstance(m, AIMessage)][-1]
print("LLM 想调用:", last_ai.tool_calls)

# 3. 人工批准，继续执行 tools
app.invoke(None, config=config)

# 4. tools 执行完回到 llm，给出最终回复
final = app.get_state(config)
print("最终:", final.values["messages"][-1].content)

流程图：

flowchart LR
    START([START]) --> LLM[LLM]
    LLM -->|tool_calls| PAUSE([暂停：等人工审批])
    PAUSE -->|批准| TN[ToolNode]
    PAUSE -->|拒绝| END([END])
    TN --> LLM
    LLM -->|无 tool_calls| END

如果想"拒绝"，可以用 update_state 把那条带 tool_calls 的 AIMessage 改成不带 tool_calls 的，或塞一条人类拒绝消息，再 invoke(None) 让 LLM 重新决策。

六、interrupt() 与 Command 配合

interrupt 适合"节点跑到一半要问人"。Command 不仅能恢复执行，还能携带人类输入：

from langgraph.types import interrupt, Command

def grade_node(state):
    # 让人类打分
    score = interrupt("请给草稿打分 0-10")
    return {"score": float(score)}

# 恢复时
app.invoke(Command(resume="8"), config=config)
# interrupt("...") 的返回值就是 "8"，被写进 state["score"]

Command 还能用于动态跳转节点（Command(goto="node_x")），在人机交互里很灵活。

七、常见踩坑

踩坑 1：忘记用 checkpointer

HITL 依赖 checkpointer 保存暂停点状态。没传 checkpointer，interrupt_before 不会生效，或者暂停后状态丢失无法恢复。

踩坑 2：中断后直接再次 invoke 原输入

暂停后应该用 invoke(None, config) 或 invoke(Command(resume=...), config) 继续，而不是又 invoke({"messages": ...})。后者会重新开始一轮新执行，行为错乱。

踩坑 3：没看 get_state 的 next 就瞎继续

中断后先 get_state 看 .next，确认确实停在你预期的节点。如果停在意外节点（比如条件边走偏了），先排查再继续。

踩坑 4：update_state 改错字段

update_state 也走 reducer。改 messages 时记得用 id 覆盖或追加语义，别误覆盖整段历史。详见 多轮对话状态管理。

踩坑 5：interrupt() 在非 checkpointer 图里用

interrupt() 必须配合 checkpointer，否则没有地方存暂停上下文。报错信息通常会提示需要 checkpointer。

八、小结

• HITL 让 AI 工作流"可控"：审批、编辑、确认、纠错。
• interrupt_before/after 在 compile 时指定自动暂停点。
• interrupt() + Command(resume=...) 在节点内主动暂停并接收人类输入。
• 中断后用 get_state 看停在哪、update_state 人工改状态、invoke(None) 或 invoke(Command(...)) 继续。
• 一切的前提是有 checkpointer。

下一篇 流式输出 Streaming 讲怎么把执行过程实时吐给前端。