中断与恢复

长流程跑到一半被打断（异常、人工暂停、进程重启），如何从断点接着跑而不是从头来？本篇结合检查点讲清楚"断点续跑"和"状态回溯"。

一、为什么能恢复

回顾 持久化与检查点：每步执行后状态被存进 checkpointer。所以"打断"只是停止了执行循环，状态已经安全保存在检查点里。用相同的 thread_id 再次调用，框架会自动从上次停下的地方继续。

flowchart LR
    A[执行] -->|每步存快照| CP[(Checkpointer)]
    A -->|中断/异常| STOP[停止]
    STOP -->|同 thread_id 重启| B[恢复]
    B -->|加载最新快照| CP
    B --> C[从断点继续]

二、执行被打断后状态在哪

无论什么原因停止，最后状态都在检查点里。几种典型中断：

中断原因	状态情况
异常崩溃	停在异常节点之前那步
interrupt_before/after	停在指定节点前/后，.next 指向待执行节点
interrupt() 主动暂停	停在调用 interrupt 的节点内
进程重启	状态在持久化 checkpointer（Sqlite/Postgres）里，重启后可恢复

前提是用了持久化 checkpointer。MemorySaver 进程重启就丢，没法恢复。

三、用相同 thread_id 重新 invoke 即可继续

最简单的恢复：直接再调一次 invoke。

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, START, END
import sqlite3

conn = sqlite3.connect("resume.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)

# 假装这是个会抛错的长流程
class State(TypedDict):
    step_done: int
    result: str

def step1(state):
    return {"step_done": 1}
def step2(state):
    # 假装这一步会抛错
    raise RuntimeError("故意失败")
def step3(state):
    return {"result": "完成"}

g = StateGraph(State)
g.add_node("step1", step1)
g.add_node("step2", step2)
g.add_node("step3", step3)
g.add_edge(START, "step1")
g.add_edge("step1", "step2")
g.add_edge("step2", "step3")
g.add_edge("step3", END)
app = g.compile(checkpointer=checkpointer)

cfg = {"configurable": {"thread_id": "job-1"}}

try:
    app.invoke({"step_done": 0, "result": ""}, config=cfg)
except Exception as e:
    print("中断了:", e)

此时 step1 已完成并存检查点，step2 抛错中断。看历史确认：

state = app.get_state(cfg)
print(state.values)   # {'step_done': 1, 'result': ''}
print(state.next)     # ('step2',)  —— 下一步要执行 step2

修复 step2 的逻辑后（假设改成不抛错），用相同 thread_id 继续：

# 传 None 表示"不输入新内容，从当前停下的地方继续"
app.invoke(None, config=cfg)

step2、step3 会继续执行完。这就是断点续跑。

四、get_state_history：查看所有检查点

get_state_history 返回这个会话的所有快照，从新到旧：

for snap in app.get_state_history(cfg):
    print(snap.config["configurable"]["checkpoint_id"],
          "| next:", snap.next,
          "| step_done:", snap.values.get("step_done"))

输出大致：

最新id  | next: ()         | step_done: 3   # 全部跑完
...     | next: ('step3',) | step_done: 2
...     | next: ('step2',) | step_done: 1   # step2 抛错前
...     | next: ('step1',) | step_done: 0   # 最初

每个快照都有一个 checkpoint_id，可以精确回到任意一个。

五、回溯到任意历史点继续

拿到某个历史快照的 config，用它作为"起点"继续跑，就从那一刻重放：

history = list(app.get_state_history(cfg))
target = history[2]   # 比如回到 step2 还没执行那个点

# 用它的 config 继续（state 参数 None 表示不输入新内容）
app.invoke(None, config=target.config)

这样会从 step2 那一刻重新执行 step2、step3。如果 step2 逻辑已修复，这次就能跑通。

注意：回溯继续会生成新的检查点分支，原来的后续历史还在。这像 git 的分支——不会丢历史，只是另开一条线。

六、完整示例：模拟中断并恢复

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

class State(TypedDict):
    counter: int
    log: str

attempt = {"n": 0}   # 用 dict 模拟可变状态，控制中断

def work(state):
    attempt["n"] += 1
    if attempt["n"] == 2:
        # 第二次执行到这步时假装崩了
        raise RuntimeError("模拟中断")
    return {"counter": state["counter"] + 1,
            "log": state["log"] + f"->step{state['counter']+1}"}

g = StateGraph(State)
g.add_node("work", work)
g.add_edge(START, "work")
g.add_edge("work", END)
app = g.compile(checkpointer=MemorySaver())
cfg = {"configurable": {"thread_id": "recover-demo"}}

# 第一次：崩了
try:
    app.invoke({"counter": 0, "log": ""}, config=cfg)
except Exception as e:
    print("第1次中断:", e)

# 看 history，确认存了"counter=1"那步
for snap in app.get_state_history(cfg):
    print("hist counter=", snap.values.get("counter"), "next=", snap.next)
    break

# 第二次：相同 thread_id 继续（这次不崩了）
attempt["n"] = 0  # 重置触发条件，让它正常跑
result = app.invoke(None, config=cfg)
print("恢复后:", result["counter"], result["log"])

运行流程：

1. 第一次 invoke：counter 0→1，存检查点；再执行一次 work（counter 1→2 前抛错）中断。
2. 第二次 invoke(None)：从 counter=1 那步继续，正常完成。

七、与 HITL 的关系

中断与恢复是 HITL（见 人机交互 HITL）的底层机制。HITL 的"暂停等人工"本质就是：执行到中断点停下，人工处理后用 invoke(None) 或 invoke(Command(resume=...)) 恢复。两者共用同一套检查点机制。

区别：

• HITL 是主动中断（interrupt_before/interrupt()），通常 .next 指向待执行节点。
• 异常恢复是被动中断，.next 指向抛错的那个节点。

八、常见踩坑

踩坑 1：恢复时传了原输入而不是 None

# ❌ 又传了一次初始输入，会被当成新的一轮执行
app.invoke({"counter": 0, "log": ""}, config=cfg)

# ✅ 续跑要传 None
app.invoke(None, config=cfg)

如果是 interrupt() 暂停的，要传 Command(resume=人类输入) 而不是 None。

踩坑 2：thread_id 不一致

中断时用的 thread_id="job-1"，恢复时写成 "job1"，框架找不到历史，从头开始。恢复必须用完全相同的 thread_id。

踩坑 3：用 MemorySaver 期望跨进程恢复

MemorySaver 存在进程内存里，进程一退就没了。要跨进程/重启恢复，必须用 SqliteSaver 或 PostgresSaver。

踩坑 4：回溯后以为原历史被覆盖

回溯继续是开新分支，原历史还在。调试时 get_state_history 会看到分叉，别误以为数据丢了。

踩坑 5：恢复后节点又抛同样的错

如果没真正修复导致中断的 bug，恢复后会在同一个节点再次抛错。恢复前先确认问题已解决，或加重试兜底（见 错误处理与重试）。

九、小结

• 中断后状态安全存在检查点里，用相同 thread_id + invoke(None) 即可续跑。
• get_state 看当前停在哪，get_state_history 看所有历史快照。
• 回溯到任意历史点的 config 继续执行，会开新分支不丢原历史。
• 跨进程恢复必须用持久化 checkpointer（Sqlite/Postgres）。
• HITL 的"暂停等人工"和异常恢复共用同一套机制。