interrupt与HITL

发邮件、删库不能自动执行——图要在关键步暂停，等人拍板。只 resume 却不改 state，列车仍按原 next 开出去，HITL 看起来像「终端里按了个继续」。真正有用的闸门是：人写下 approve / reject，条件边据此换轨。

段末注释：HITL（human-in-the-loop，人在回路）= 人写入决策字段后，图按该字段选下一节点。

社区方案：官方 interrupt() + Command(resume=...) 收回决策，再用 add_conditional_edges 的 path_map 分支。不要只配 interrupt_before 再原路续跑。风险：interrupt() 恢复时从该节点开头重跑，闸门前不要放不可重入副作用。

---

1. 定位

维度	内容
角色	暂停 + 人工决策改路由
输入 → 输出	Command(resume="approve"|"reject") → decision → path_map → execute / reject
核心 API	interrupt()、Command(resume=...)、add_conditional_edges
依赖 LangChain	无；审批 UI 在业务层

出现背景：interrupt_before 只冻结 next；人不写决策，边不会变。

---

2. 图拓扑

节点表

节点名	职责	读 State	写 State
prepare	准备待审动作	action	status
gate	interrupt() 等人；写入决策	action	decision, status
execute	批准后执行	action	status
reject	拒绝后拦截	action, decision	status

边表

源	目标	类型	router 返回值
START	prepare	固定	—
prepare	gate	固定	—
gate	execute / reject	条件	"execute" / "reject"
execute	END	固定	—
reject	END	固定	—

path_map：{"execute": "execute", "reject": "reject"}
interrupt：gate 内调用 interrupt(...)（不是 interrupt_before=["execute"]）

---

3. invoke 生命周期

1. prepare → status="ready:send_email"
2. gate 调用 interrupt({ask, action}) → put 暂停现场，本次 invoke 返回
3. get_state：next=("gate",)，tasks 上挂中断题面
4a. Command(resume="approve") → interrupt() 返回 "approve"
    → decision="approve" → router="execute" → execute → status="done:..."
4b. Command(resume="reject") → decision="reject"
    → router="reject" → reject → status="blocked:..."

同一张图、两个 thread_id，才能并排看出路径差。

---

4. 原理

4.1 两种暂停，只有一种能把「人的话」变成边

机制	停在哪	resume 之后	能否单独改路径
compile(interrupt_before=["execute"])	进入该节点前	仍走进 next 里那个节点	否。要改边须再 update_state 写字段 + 条件边
interrupt() 写在节点里	调用处	该节点从头重跑，interrupt() 返回 resume 值	能。节点把返回值写入 decision，router 读它

本篇示例用第二种：人的 resume 就是 path_map 的输入。第一种仍适合「只想在节点边界设断点、决策另文用 update_state 注入」。

4.2 暂停 / 恢复如何打 Saver

无 Saver 时中断现场无处可放。

1. compile(checkpointer=cp) 绑 Saver（本示例不再传 interrupt_before）。
2. invoke(input, cfg)：prepare 后进入 gate，碰到 interrupt() → put，next=("gate",)，本次返回。
3. get_state(cfg) → get_tuple：看 values、next、以及 tasks 上的中断 payload（给审批 UI 展示题面）。
4. invoke(Command(resume="approve"), cfg)：再 get_tuple，gate 重入，interrupt() 得到 "approve"，写入 decision，条件边选轨后再 put。

少了第 2 步的 put → 进程一走，题面和 next 都没了。
第 4 步换了 thread_id → 走进空树，不会接着 gate。

4.3 人工反馈如何改路径

interrupt 本身不改边。链路是：

1. 人传入 Command(resume=x)，x 必须是 path_map 认识的语义（本篇 "approve" / "reject"）。
2. gate 返回 {"decision": x}。
3. router(state) 读 state["decision"]，返回 path_map 的键（"execute" / "reject"）。
4. 调度走进对应节点。

resume="approve" 却把 router 写成只看 status → 人扳了闸，道岔没动。
空 decision 必须有默认轨（本篇默认 reject），否则 path_map 对不上会报错。

4.4 interrupt ≠ 鉴权

只暂停调度并改路由；权限校验仍在 API 层。

---

5. 最小可运行示例

from typing import Literal, TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt


class State(TypedDict):
    action: str
    decision: str
    status: str


def prepare(state: State) -> dict:
    return {"status": f"ready:{state['action']}"}  # 拓扑 §2 prepare


def gate(state: State) -> dict:
    # 恢复时本函数从头执行；interrupt() 返回 Command(resume=...) 的值
    decision = interrupt({
        "ask": "approve 或 reject",
        "action": state["action"],
    })
    return {"decision": decision, "status": f"gated:{decision}"}


def execute(state: State) -> dict:
    return {"status": f"done:{state['action']}"}  # 批准轨


def reject(state: State) -> dict:
    return {"status": f"blocked:{state['action']}"}  # 拒绝轨


def route(state: State) -> Literal["execute", "reject"]:
    return "execute" if state.get("decision") == "approve" else "reject"


builder = StateGraph(State)
builder.add_node("prepare", prepare)
builder.add_node("gate", gate)
builder.add_node("execute", execute)
builder.add_node("reject", reject)
builder.add_edge(START, "prepare")
builder.add_edge("prepare", "gate")
builder.add_conditional_edges(
    "gate",
    route,
    {"execute": "execute", "reject": "reject"},  # path_map
)
builder.add_edge("execute", END)
builder.add_edge("reject", END)

graph = builder.compile(checkpointer=InMemorySaver())


def run(thread_id: str, resume_value: str) -> None:
    cfg = {"configurable": {"thread_id": thread_id}}
    graph.invoke({"action": "send_email", "decision": "", "status": ""}, cfg)
    snap = graph.get_state(cfg)
    print(thread_id, "paused", snap.next, snap.values["status"])
    out = graph.invoke(Command(resume=resume_value), cfg)
    print(thread_id, "final", out["decision"], out["status"])


run("hitl-ok", "approve")  # decision=approve → execute → done:send_email
run("hitl-no", "reject")   # decision=reject  → reject  → blocked:send_email

---

6. 执行追踪

thread	暂停时 next / status	resume	router	终态 status
hitl-ok	("gate",) / ready:send_email	"approve"	execute	done:send_email
hitl-no	同上	"reject"	reject	blocked:send_email

两条轨共享 prepare + gate；差别只在人回的那一个字符串。

---

重要配置参数

参数（API 名）	类型 / 默认值	功能说明	作用与影响	参考起点 / 常用范围	配置指导
interrupt(payload)	任意可序列化	节点内暂停并把题面 put 进 checkpoint	恢复时从节点开头重跑，返回 resume 值	审批题面 dict	闸门前勿做不可重入 IO
Command(resume=...)	"approve" / "reject"	再 invoke，值变成 interrupt() 的返回值	写进 decision 才能改边	与 path_map 语义对齐	同 thread_id
add_conditional_edges(gate, route, path_map)	router + dict	读 decision 映射下一节点	键对不上则运行时报错	{"execute","reject"}	先画边表
compile(interrupt_before=[...])	list[str]	节点前断点，不注入决策	只冻 next	调试 / 固定闸门	要改路径另写字段
compile(checkpointer=)	必填	暂停 put、resume get_tuple	无则现场不可恢复	InMemorySaver	生产换 DB
get_state().next / tasks	tuple / 任务	待跑节点与中断题面	给审批 UI	("gate",)	空 next=已结束
thread_id	str	暂停与 resume 同一把钥匙	换 ID 找不到现场	UUID	两条审批用两个 ID

---

7. 易踩坑

1. 只 resume(True) 不写决策：decision 仍空，本例 router 一律进 reject，看起来「HITL 没生效」。
2. 以为 interrupt_before 能改边：它只停在既定 next；改路径要 decision + 条件边。
3. interrupt() 前做发信/扣款：恢复会重跑节点前半段，副作用会重复。
4. 无 checkpointer：无处 put，暂停现场不可恢复。
5. resume 换了 thread_id：get_tuple 走进空树。
6. 把 interrupt 当鉴权：仍需服务端鉴权与审计日志。

---

小结

• interrupt 只停车；人写入的 decision + path_map 才扳道岔。
• Command(resume="approve"|"reject") → gate → execute / reject，两条轨终态不同。
• 暂停现场靠 Saver.put / get_tuple；审批权限在业务层。

---

参考链接

• Interrupts
• Human-in-the-loop