时间旅行

LangGraph 的 Checkpoint 机制支持"时间旅行"——回放历史执行步骤，或从某个 checkpoint 分支出去探索不同的路径。这是调试和实验的强大工具。

概述

时间旅行利用 Checkpoint 实现了两种能力：

• Replay（回放）：从历史的某个 checkpoint 重新执行
• Fork（分支）：从历史 checkpoint 修改状态后，探索不同的路径

Replay：重放执行

从某个历史 checkpoint 重新执行，之前已完成的节点不会重跑：

from langgraph.graph import StateGraph, START
from langgraph.checkpoint.memory import InMemorySaver
from typing_extensions import TypedDict
from langchain_core.utils.uuid import uuid7

class State(TypedDict):
    topic: str
    joke: str

def generate_topic(state: State):
    return {"topic": "洗衣机里的袜子"}

def write_joke(state: State):
    return {"joke": f"为什么{state['topic']}会消失? 它们私奔了!"}

checkpointer = InMemorySaver()
graph = (
    StateGraph(State)
    .add_node("generate_topic", generate_topic)
    .add_node("write_joke", write_joke)
    .add_edge(START, "generate_topic")
    .add_edge("generate_topic", "write_joke")
    .compile(checkpointer=checkpointer)
)

# 1. 执行
config = {"configurable": {"thread_id": str(uuid7())}}
result = graph.invoke({}, config)
print(result["joke"])  # 为什么洗衣机里的袜子会消失? 它们私奔了!

# 2. 查找 write_joke 之前的 checkpoint
history = list(graph.get_state_history(config))
before_joke = next(s for s in history if s.next == ("write_joke",))

# 3. 从该 checkpoint 重放
replay_result = graph.invoke(None, before_joke.config)
# generate_topic 不会重跑，write_joke 会重跑
print(replay_result["joke"])

注意事项

• 重放会重新执行节点，不是读缓存
• LLM 调用和 API 请求会再次触发，可能返回不同结果
• 从最终 checkpoint（无 next 节点）重放是空操作

Fork：分支探索

从历史 checkpoint 修改状态，创造一条新的路线：

# 找到 write_joke 之前的 checkpoint
history = list(graph.get_state_history(config))
before_joke = next(s for s in history if s.next == ("write_joke",))

# Fork：修改主题为"鸡"
fork_config = graph.update_state(
    before_joke.config,
    values={"topic": "鸡"},
)

# 从 fork 点继续
fork_result = graph.invoke(None, fork_config)
print(fork_result["joke"])  # 关于鸡的笑话

使用 as_node 指定节点

# 把状态更新伪装成 generate_topic 产生的
# 执行将从 generate_topic 的后继节点继续
fork_config = graph.update_state(
    before_joke.config,
    values={"topic": "鸡"},
    as_node="generate_topic",
)

需要显式指定 as_node 的场景：

• 并行分支：多个节点在同一 step 更新了 state
• 无执行历史：在新线程上设置初始状态
• 跳过节点：让图认为某个节点已经执行过了

与 Interrupt 的交互

如果图使用了 Interrupt，时间旅行时会重新触发中断：

from langgraph.types import interrupt, Command

class State(TypedDict):
    value: list[str]

def ask_human(state: State):
    answer = interrupt("你叫什么名字?")
    return {"value": [f"你好, {answer}!"]}

def final_step(state: State):
    return {"value": ["完成"]}

graph = (
    StateGraph(State)
    .add_node("ask_human", ask_human)
    .add_node("final_step", final_step)
    .add_edge(START, "ask_human")
    .add_edge("ask_human", "final_step")
    .compile(checkpointer=InMemorySaver())
)

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

# 第一次执行：遇到 interrupt 暂停
graph.invoke({"value": []}, config)
# 恢复
graph.invoke(Command(resume="小明"), config)

# 回放到 ask_human 之前
history = list(graph.get_state_history(config))
before_ask = [s for s in history if s.next == ("ask_human",)][0]

# 重放：ask_human 会再次触发 interrupt
graph.invoke(None, before_ask.config)
# 需要重新提供 Command(resume=...)

常见用例

1. 调试：回放到出错的步骤，观察 state
2. 分支测试：修改输入看不同路径
3. 人工修正：如果 LLM 给出了错误结果，可以回退人工修正
4. A/B 测试：同一个 checkpoint 分支出去对比不同策略

参考

• 官方文档：Use time-travel
• 官方文档：Persistence