LangGraph 提供了时间旅行功能，以支持这些使用场景。具体来说，你可以从之前的检查点恢复执行——要么重放相同的状态，要么修改它来探索替代方案。在所有情况下，恢复过去的执行都会在历史中产生一个新的分支。

官方文档：https://docs.langchain.com/oss/python/langgraph/use-time-travel

核心概念

什么是 Checkpoint 重放？

LangGraph 的 Checkpoint 重放，就像给你的工作流装了一台时间机器：

可以回到历史执行中的任意节点（super-step）
从该节点继续往下执行，或者修改当时的状态、走不同的分支
相当于在同一个对话 / 流程里，随时可以 “读档重来”，探索不同路径

Checkpoint 的结构

Checkpoint 本质上是每个 super-step（节点执行周期） 的完整状态快照，它包含三个核心部分：

字段	说明
`values`	当前状态的值（比如对话上下文、中间结果、用户输入等）
`next`	下一个要执行的节点（决定了流程的下一步走向）
`tasks`	任务执行状态（记录当前步骤里的任务执行情况）

重放机制的优势

LangGraph 的重放机制有两个关键特点：

智能恢复，不重复执行：已执行过的步骤结果会被直接恢复，不会重新运行，节省计算资源和时间
确定性与零成本：恢复过程是可预测的，且不会产生额外的模型调用或计算成本

Checkpoint 重放的本质，是给 LangGraph 工作流加上了状态持久化 + 分支探索能力：

它让复杂的多轮对话、多节点流程可以 “断点续传”
也支持用户随时回溯、修改中间步骤，探索不同分支，而不用从头跑一遍整个流程

重放模式

简单重放

这段代码演示了 LangGraph 中最基础的 Checkpoint 重放 能力，主要解决这些场景：

故障恢复：流程执行一半中断后，可以从断点继续
人为中断后继续：用户 / 系统主动暂停，之后从断点恢复
跳过已完成步骤：重放时不会重复执行已完成的步骤，节省成本

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


class State(dict):  # 简化状态定义
    pass


# 构建带checkpoint的graph
graph = StateGraph(State)
graph.add_node("slow_llm", lambda state: {"result": f"LLM-{uuid.uuid4().hex[:4]}"})
graph.add_edge(START, "slow_llm")
checkpointer = MemorySaver()
graph = graph.compile(checkpointer=checkpointer)


# 🚀 首次运行（生成历史）
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
print("=== 首次运行（3秒） ===")
graph.invoke({}, config)  # 模拟LLM调用，耗时3秒


# 📜 查看历史checkpoint（逆时序）
print("\n=== 历史Checkpoint ===")
states = list(graph.get_state_history(config))
for i, state in enumerate(reversed(states)):
    print(f"  #{i}: next={state.next}, id={state.config['configurable']['checkpoint_id'][:8]}...")


# 🔄 简单重放：从第1个checkpoint继续（0.1秒！）
mid_checkpoint = states[1]
config_replay = {
    "configurable": {
        "thread_id": config["configurable"]["thread_id"],
        "checkpoint_id": mid_checkpoint.config["configurable"]["checkpoint_id"]
    }
}

print("\n=== 简单重放（0.1秒） ===")
result = graph.invoke(None, config_replay)
print("结果:", result)

代码逐步解读：

导入依赖
1
2
3
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START
import uuid
- MemorySaver：LangGraph 内置的内存级状态存储，用来保存每个 checkpoint 的状态
- StateGraph/START：构建 LangGraph 工作流的核心组件
- uuid：用来生成唯一 ID，模拟 LLM 调用的结果和会话标识
定义状态类
1
2
class State(dict): # 简化状态定义
pass
- 这是一个简化的状态定义，继承自 dict，可以在里面存任意键值对（比如对话结果、中间数据等）

构建带 Checkpoint 的 Graph

# 构建带checkpoint的graph
graph = StateGraph(State)
graph.add_node("slow_llm", lambda state: {"result": f"LLM-{uuid.uuid4().hex[:4]}"})
graph.add_edge(START, "slow_llm")
checkpointer = MemorySaver()
graph = graph.compile(checkpointer=checkpointer)

add_node("slow_llm", ...)：添加一个模拟 LLM 调用的节点，每次运行会生成一个带随机 ID 的结果（比如 LLM-a1b2）
graph.compile(checkpointer=checkpointer)：编译图时绑定 MemorySaver，开启自动 checkpoint 功能
每执行一个 super-step（这里就是执行 slow_llm 节点），LangGraph 都会自动保存一个 checkpoint

首次运行（生成历史）
1
2
3
4
# 🚀 首次运行（生成历史）
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
print("=== 首次运行（3秒） ===")
graph.invoke({}, config) # 模拟LLM调用，耗时3秒
- thread_id：会话的唯一标识，LangGraph 会按 thread_id 来隔离不同会话的 checkpoint
- graph.invoke({}, config)：执行整个工作流，这里会执行 slow_llm 节点，生成第一个 checkpoint，耗时约 3 秒（模拟 LLM 调用）

查看历史 Checkpoint

# 📜 查看历史checkpoint（逆时序）
print("\n=== 历史Checkpoint ===")
states = list(graph.get_state_history(config))
for i, state in enumerate(reversed(states)):
    print(f"  #{i}: next={state.next}, id={state.config['configurable']['checkpoint_id'][:8]}...")

graph.get_state_history(config)：获取当前 thread_id 下的所有 checkpoint 历史
reversed(states)：逆序打印，方便看到从最早到最新的 checkpoint
每个 state 里包含：
- state.next：下一个要执行的节点（这里执行完 slow_llm 后 next 为空）
- checkpoint_id：每个 checkpoint 的唯一 ID

简单重放：从历史 Checkpoint 继续

# 🔄 简单重放：从第1个checkpoint继续（0.1秒！）
mid_checkpoint = states[1]
config_replay = {
    "configurable": {
        "thread_id": config["configurable"]["thread_id"],
        "checkpoint_id": mid_checkpoint.config["configurable"]["checkpoint_id"]
    }
}

print("\n=== 简单重放（0.1秒） ===")
result = graph.invoke(None, config_replay)
print("结果:", result)

mid_checkpoint = states[1]：从历史 checkpoint 列表中取一个 checkpoint（这里是倒数第二个，也就是 slow_llm 执行前的状态）
config_replay：构造重放配置，指定要恢复的 thread_id 和 checkpoint_id
graph.invoke(None, config_replay)：从指定的 checkpoint 继续执行
- 因为 slow_llm 已经在之前执行过，LangGraph 会直接恢复结果，不会重新执行 LLM 调用，耗时仅约 0.1 秒
- 最终输出的 result 和首次运行的结果是一样的（不会生成新的随机 ID）

核心效果与优势

操作	耗时	行为
首次运行	约 3 秒	执行 LLM 节点，生成 checkpoint，保存结果
重放运行	约 0.1 秒	直接恢复 checkpoint 状态，跳过重复执行，返回相同结果

这就是 LangGraph Checkpoint 重放的核心价值：

确定性：从同一个 checkpoint 重放，结果永远一致
零成本：不会重复执行昂贵的 LLM 调用或其他耗时操作
可回溯：随时可以回到历史任意节点，修改状态或探索新分支

修改后重放

这段代码演示了 LangGraph 中更进阶的 Checkpoint 能力 ——修改历史状态并重放，主要解决这些场景：

调试 “如果当时改了会怎样”：修改中间步骤的状态，看看后续流程会如何变化
Human-in-the-Loop 干预：人工介入修改状态，再让流程继续执行
A/B 测试：从同一个历史节点出发，修改不同参数，对比不同分支的结果

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START
import uuid
import time


class State(dict):  # 简化状态定义
    pass


# 构建带checkpoint的graph
graph = StateGraph(State)
# 模拟一个耗时的LLM节点，会读取状态中的override字段
def slow_llm(state):
    time.sleep(3)  # 模拟耗时
    if "override" in state:
        return {"result": f"LLM-{uuid.uuid4().hex[:4]} | 收到override: {state['override']}"}
    return {"result": f"LLM-{uuid.uuid4().hex[:4]}"}

graph.add_node("slow_llm", slow_llm)
graph.add_edge(START, "slow_llm")
checkpointer = MemorySaver()
graph = graph.compile(checkpointer=checkpointer)


# 🚀 首次运行（生成历史）
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
print("=== 首次运行（3秒） ===")
graph.invoke({}, config)  # 模拟LLM调用，耗时3秒


# 📜 查看历史checkpoint（逆时序）
print("\n=== 历史Checkpoint ===")
states = list(graph.get_state_history(config))
for i, state in enumerate(reversed(states)):
    print(f"  #{i}: next={state.next}, id={state.config['configurable']['checkpoint_id'][:8]}...")


# 🔄 简单重放：从第1个checkpoint继续（0.1秒！）
mid_checkpoint = states[1]
config_replay = {
    "configurable": {
        "thread_id": config["configurable"]["thread_id"],
        "checkpoint_id": mid_checkpoint.config["configurable"]["checkpoint_id"]
    }
}

print("\n=== 简单重放（0.1秒） ===")
result = graph.invoke(None, config_replay)
print("结果:", result)


# ⏱️ 回到历史第1个checkpoint，修改状态
print("\n=== 修改后重放 (Time Travel) ===")
selected_state = states[1]  # 选择历史状态

# 📝 修改历史状态，创建新checkpoint
new_config = graph.update_state(
    selected_state.config,
    values={"override": "人工修改的值！"}  # 修改状态
)
print("新分支checkpoint:", new_config["configurable"]["checkpoint_id"][:8])

# 🚀 从修改后的状态继续执行
branch_result = graph.invoke(None, new_config)
print("新分支结果:", branch_result)

操作	效果
回到历史 checkpoint	回到流程执行中的任意节点
`update_state()` 修改状态	生成一条新的分支时间线，不影响原流程
从新 checkpoint 执行	后续节点会基于修改后的状态运行，产生不同结果

重放模式的最佳实践

场景应用建议

场景	方法	关键 `config`/ 操作
故障恢复	简单重放最后成功点	`checkpoint_id=最后成功checkpoint`
HITL 干预（人工介入）	修改后重放	`update_state()` + 重放
调试回溯	查看历史 + 新分支	`get_state_history()` + Time Travel
生产续跑	简单重放最新点	`input=None, config=最新config`

故障恢复
- 场景：流程执行中崩溃、网络中断、系统重启
- 做法：找到崩溃前最后一个成功的 checkpoint，直接从这里重放
- 关键：通过 checkpoint_id 指定恢复点，不需要从头跑一遍，节省时间和成本
HITL（Human-in-the-Loop）人工干预
- 场景：需要人工介入修改中间状态，比如用户纠正了错误的工具调用结果、审批流程
- 做法：用 update_state() 修改历史状态，生成新 checkpoint，再从新分支继续执行
- 关键：修改状态后会创建新分支，不会影响原流程，实现 “干预式继续”
调试回溯
- 场景：想知道 “如果当时改了某个参数，流程会怎么走”，或者排查问题时回溯历史
- 做法：用 get_state_history() 查看所有历史 checkpoint，选择任意节点做 Time Travel（时间旅行），修改状态并生成新分支测试
- 关键：支持多分支对比，不用重复执行昂贵的 LLM / 工具调用
生产续跑
- 场景：服务重启、会话中断后，用户重新进入对话 / 流程
- 做法：用 input=None + 最新的 config（包含 thread_id 和最新 checkpoint_id）直接续跑
- 关键：input=None 表示不传入新输入，LangGraph 会自动从上次中断的地方继续执行

实用技巧

使用 get_state_history() 查看完整的执行历史
- 可以拿到当前 thread_id 下的所有 checkpoint，包括每个步骤的状态、next 节点、checkpoint_id
- 调试时可以清晰看到流程的每一步变化，方便定位问题
通过 update_state() 创建新的执行分支
- 这是 LangGraph 实现 “时间旅行” 的核心方法
- 修改任意历史状态后，会生成一个全新的 checkpoint_id，衍生出独立的执行分支，原流程不受影响
利用重放机制避免重复计算，提升性能
- 重放时 LangGraph 会直接恢复已执行步骤的结果，不会重新调用 LLM / 工具
- 大幅降低调试、故障恢复、多分支测试的成本和耗时
在调试时使用 Time Travel 功能测试不同决策路径
- 从同一个历史节点出发，修改不同的状态值，就能快速测试不同分支的结果
- 适合 A/B 测试、策略对比、边界条件验证

本质上是 LangGraph Checkpoint 能力的「生产级使用指南」：

用简单重放解决故障恢复和续跑问题
用修改后重放解决人工干预和多分支测试问题
用 get_state_history() 和 update_state() 实现灵活的调试与回溯
全程避免重复执行，大幅提升性能和开发效率

Jean's Blog

LangGraph之时光旅行