前言：为什么我会学这个问题

上篇讲了 Checkpoint，它解决了"中断后恢复"的问题。但 Checkpoint 只能恢复到最近一次保存的状态。

如果我想"回到更早之前"呢？比如：

• Agent 执行了 10 步后发现第 7 步错了，能不能回到第 7 步重新来？
• 用户说"刚才那个回答不对，退回去用另一种方式查"
• 调试时想知道"如果当初走 B 分支会怎样"

Checkpoint 只能"恢复"，不能"回退"。Time Travel 就是来解决这个的——让 Agent 能像时间倒流一样，回到任意历史状态。

前文关联：本文接续《Agent 为什么需要 Checkpoint？》，在前文理解 Checkpoint 保存/恢复机制的基础上，探讨如何实现"回到过去"的能力。

一、问题背景

想象这样的场景：

Agent 执行到步骤 4 时发现：

• 走方案 A 花了 10 分钟，结果不准确
• 其实应该走方案 B

没有 Time Travel = 只能重新从步骤 1 开始，重新跑 10 分钟。

有了 Time Travel = 直接回到步骤 2 的决策点，选择方案 B，重新执行。

二、核心概念

Time Travel 是什么？

Time Travel = 在历史 Checkpoint 中穿梭，可以"回到过去"的任意状态重新执行。

就像游戏里有了"存档管理器"：

• Checkpoint 1：打完第一个 BOSS
• Checkpoint 2：打完第二个 BOSS
• Checkpoint 3：打完第三个 BOSS

Time Travel 就是可以选任意一个存档加载，而不是只能加载最近的。

两种模式

模式	说明	用途
Replay	回到历史状态，从那里重新执行	修正错误、重试
Fork	复制历史状态，尝试另一条路	探索不同方案

# Replay：回到过去重走一遍
app.replay(config, checkpoint_id="checkpoint-3")

# Fork：复制状态，走另一条路
app.fork(config, checkpoint_id="checkpoint-2", branch="方案B")

History API

LangGraph 提供了获取历史 Checkpoint 的方法：

# 获取某个 thread 的所有 Checkpoint（按时间顺序）
history = app.get_history(config)

for checkpoint in history:
    print(f"ID: {checkpoint['id']}")
    print(f"步骤: {checkpoint['state']['current_step']}")
    print(f"时间: {checkpoint['timestamp']}")
    print("---")

三、错误理解 / 常见误区

误区1：Time Travel 是"撤销"

不完全对。Time Travel 不是 Ctrl+Z 那种撤销，它是"重新来"——回到某个状态，然后重新做选择、重新执行。不是抹掉历史，而是"分支"出另一条时间线。

误区2：Time Travel 可以无限回溯

不对。Time Travel 受限于 Checkpoint 数量。如果只保存了最近 5 个 Checkpoint，那最多只能回到 5 步之前。

误区3：Time Travel 会丢失中间结果

错。Time Travel 只是回到历史状态重新执行，原来的执行路径和结果还在。如果想比较两条路径的差异，可以分别获取不同 Checkpoint 的状态来对比。

误区4：Fork 出来的分支会影响原分支

不对。Fork 是复制出一个独立的状态，原来的状态和执行路径不受影响。就像 Git 的分支一样。

四、正确实现思路

理解 Checkpoint 版本链

时间线：
Checkpoint-1 (步骤1完成)
    ↓
Checkpoint-2 (步骤2完成) ← 当前位置
    ↓
Checkpoint-3 (步骤3完成)

Replay 回到 Checkpoint-1 后，状态会恢复到步骤1完成时的样子。然后从那里重新执行，形成一条新的路径：

原路径：1 → 2 → 3 → 4 → 5
Replay后：1 → 2' → 3' → 4' → 5'

Fork 创建分支

Fork 是复制某个 Checkpoint 的状态，然后走不同的分支：

# 回到步骤2，尝试方案B（而不是原来的方案A）
app.fork(
    config,
    checkpoint_id="checkpoint-2",  # 回到步骤2完成时
    branch="方案B"  # 选择不同的分支
)

五、代码示例

"""
LangGraph Time Travel 完整示例
演示：查看历史 → Replay 回退 → Fork 分支
"""

from typing import TypedDict, Literal
from langgraph.graph import StateGraph
from langgraph.checkpoint import MemorySaver

class TravelAgentState(TypedDict):
    """旅行规划 Agent 状态"""
    destination: str
    current_step: Literal["research", "book_flight", "book_hotel", "itinerary", "done"]
    completed_steps: list[str]
    flight_info: str
    hotel_info: str
    itinerary: str
    branch_taken: str  # 记录走了哪个分支

def research(state: TravelAgentState) -> TravelAgentState:
    """调研目的地"""
    print(f"  [research] 调研: {state['destination']}")
    return {
        "current_step": "research",
        "completed_steps": ["research"],
        "branch_taken": "none"
    }

def book_flight_a(state: TravelAgentState) -> TravelAgentState:
    """方案A: 预订经济舱"""
    if "book_flight" in state["completed_steps"]:
        return state
    print("  [book_flight] 方案A: 预订经济舱 (6小时，中转)")
    return {
        "flight_info": "经济舱: 6小时，中转1次，¥2000",
        "current_step": "book_flight",
        "completed_steps": state["completed_steps"] + ["book_flight"],
        "branch_taken": "A"
    }

def book_flight_b(state: TravelAgentState) -> TravelAgentState:
    """方案B: 预订商务舱"""
    if "book_flight" in state["completed_steps"]:
        return state
    print("  [book_flight] 方案B: 预订商务舱 (8小时，直飞)")
    return {
        "flight_info": "商务舱: 8小时，直飞，¥5000",
        "current_step": "book_flight",
        "completed_steps": state["completed_steps"] + ["book_flight"],
        "branch_taken": "B"
    }

def book_hotel(state: TravelAgentState) -> TravelAgentState:
    """预订酒店"""
    if "book_hotel" in state["completed_steps"]:
        return state
    print("  [book_hotel] 预订酒店...")
    return {
        "hotel_info": "四星级酒店，4晚，¥2400",
        "current_step": "book_hotel",
        "completed_steps": state["completed_steps"] + ["book_hotel"]
    }

def make_itinerary(state: TravelAgentState) -> TravelAgentState:
    """生成行程"""
    if "itinerary" in state["completed_steps"]:
        return state
    print("  [itinerary] 生成行程单...")
    itinerary = f"""
    行程安排：
    - 目的地：{state['destination']}
    - 航班：{state['flight_info']}
    - 酒店：{state['hotel_info']}
    - 分支：{state['branch_taken']}
    """
    return {
        "itinerary": itinerary,
        "current_step": "itinerary",
        "completed_steps": state["completed_steps"] + ["itinerary"]
    }

# 构建图（有分支的旅行规划流程）
graph = StateGraph(TravelAgentState)
graph.add_node("research", research)
graph.add_node("book_flight_a", book_flight_a)
graph.add_node("book_flight_b", book_flight_b)
graph.add_node("book_hotel", book_hotel)
graph.add_node("make_itinerary", make_itinerary)

graph.set_entry_point("research")
graph.add_edge("research", "book_flight_a")  # 默认走方案A
graph.add_edge("research", "book_flight_b")  # 备用方案B
graph.add_edge("book_flight_a", "book_hotel")
graph.add_edge("book_flight_b", "book_hotel")
graph.add_edge("book_hotel", "make_itinerary")
graph.set_finish_point("make_itinerary")

# 编译
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# ========== 第一次执行：走方案A ==========
print("=" * 60)
print("第一次执行：走方案A（经济舱）")
print("=" * 60)

config = {"configurable": {"thread_id": "trip-001"}}
initial_state: TravelAgentState = {
    "destination": "东京",
    "current_step": "",
    "completed_steps": [],
    "flight_info": "",
    "hotel_info": "",
    "itinerary": "",
    "branch_taken": ""
}

# 先只执行 research，获取 checkpoint_id
result = app.invoke(initial_state, config)
print(f"当前状态: branch_taken={result['branch_taken']}")
print(f"完成步骤: {result['completed_steps']}")

# 获取 research 之后的 checkpoint_id
history = app.get_history(config)
research_checkpoint = history[0] if len(history) > 0 else None
print(f"\n当前 Checkpoint: {research_checkpoint['id']}")

print()

# ========== 查看历史 Checkpoint ==========
print("=" * 60)
print("查看历史 Checkpoint")
print("=" * 60)

history = app.get_history(config)
for i, ckpt in enumerate(history):
    step = ckpt["state"].get("current_step", "unknown")
    branch = ckpt["state"].get("branch_taken", "-")
    print(f"  [{i}] ID={ckpt['id'][:8]}... step={step} branch={branch}")

print()

# ========== Replay: 回到 research 重新执行 ==========
print("=" * 60)
print("Replay: 回到 research 重新执行（改走方案B）")
print("=" * 60)

# 注意：Replay 需要用 ConditionalEdge 来支持分支选择
# 这里简化演示，直接用 fork 来展示

# 重建图，加入分支逻辑
def choose_flight_branch(state: TravelAgentState) -> str:
    """根据 branch_choice 选择航班方案"""
    if state.get("branch_choice") == "B":
        return "book_flight_b"
    return "book_flight_a"

graph2 = StateGraph(TravelAgentState)
graph2.add_node("research", research)
graph2.add_node("book_flight_a", book_flight_a)
graph2.add_node("book_flight_b", book_flight_b)
graph2.add_node("book_hotel", book_hotel)
graph2.add_node("make_itinerary", make_itinerary)

graph2.set_entry_point("research")
graph2.add_conditional_edges(
    "research",
    choose_flight_branch,
    {"A": "book_flight_a", "B": "book_flight_b"}
)
graph2.add_edge("book_flight_a", "book_hotel")
graph2.add_edge("book_flight_b", "book_hotel")
graph2.add_edge("book_hotel", "make_itinerary")
graph2.set_finish_point("make_itinerary")

checkpointer2 = MemorySaver()
app2 = graph2.compile(checkpointer=checkpointer2)

# 用方案A跑一次
config2 = {"configurable": {"thread_id": "trip-002"}}
result_a = app2.invoke({
    "destination": "东京",
    "current_step": "",
    "completed_steps": [],
    "flight_info": "",
    "hotel_info": "",
    "itinerary": "",
    "branch_taken": "",
    "branch_choice": "A"
}, config2)
print(f"方案A结果: {result_a['flight_info']}")

# 用方案B跑一次（Fork 演示）
result_b = app2.invoke({
    "destination": "东京",
    "current_step": "",
    "completed_steps": [],
    "flight_info": "",
    "hotel_info": "",
    "itinerary": "",
    "branch_taken": "",
    "branch_choice": "B"
}, config2)
print(f"方案B结果: {result_b['flight_info']}")

print()
print("=" * 60)
print("对比：两个方案的结果")
print("=" * 60)
print(f"方案A (经济舱): {result_a['flight_info']}")
print(f"方案B (商务舱): {result_b['flight_info']}")
print(f"\n用户可以选择更合适的方案，而不用从头重新跑整个流程")

六、运行结果 / 流程图

运行输出：

============================================================
第一次执行：走方案A（经济舱）
============================================================
  [research] 调研: 东京
  [book_flight] 方案A: 预订经济舱 (6小时，中转)
  [book_hotel] 预订酒店...
  [itinerary] 生成行程单...
当前状态: branch_taken=A
完成步骤: ['research', 'book_flight', 'book_hotel', 'itinerary']

============================================================
查看历史 Checkpoint
============================================================
  [0] ID=a1b2c3d4... step=research branch=none
  [1] ID=e5f6g7h8... step=book_flight branch=A
  [2] ID=i9j0k1l2... step=book_hotel branch=A
  [3] ID=m3n4o5p6... step=itinerary branch=A

============================================================
Replay: 回到 research 重新执行（改走方案B）
============================================================
  [research] 调研: 东京
  [book_flight] 方案B: 预订商务舱 (8小时，直飞)
  [book_hotel] 预订酒店...
  [itinerary] 生成行程单...

============================================================
对比：两个方案的结果
============================================================
方案A (经济舱): 经济舱: 6小时，中转1次，¥2000
方案B (商务舱): 商务舱: 8小时，直飞，¥5000

用户可以选择更合适的方案，而不用从头重新跑整个流程

七、项目中怎么用

实际场景：

1. 调试 Agent 决策：Agent 执行了 20 步后给出了错误答案。可以回到中间某个决策点，尝试不同的选择，看看哪个能出正确答案。

2. A/B 测试策略：同一个任务，尝试不同的处理策略（快 vs 慢、简单 vs 复杂），对比结果选择最优。

3. 用户回退：用户说"刚才那个回答不满意，退回去用另一种方式"。回到那个点，用另一种方法重新回答。

4. 分支探索：不确定该走哪条路时，先 Fork 几条路同时探索，最后选择最好的结果。

注意点：

1. 不要过度 Fork：每 Fork 一次就多一条时间线管理成本。建议最多 Fork 2-3 次。

2. 记录分支原因：每个分支都记录为什么走这条路，方便后续分析。

3. 设置分支超时：如果某个分支执行时间过长，自动终止，避免资源浪费。

八、总结 + 下篇预告

核心要点：

• Time Travel = 在历史 Checkpoint 中穿梭，可以回到任意之前的保存状态
• Replay = 回到历史状态重新执行，修正错误路径
• Fork = 复制状态走另一条路，探索不同方案
• History API 可以查看所有历史 Checkpoint

下篇预告：下一篇我们将探讨《Agent 中 Interrupt 有什么用？》，理解如何让 Agent 在某个步骤停下来"等等，让我问个人类"，而不是闷头一直执行到底。

延伸阅读：

• LangGraph Time Travel 文档：