第 13 章 时间旅行

本章目标：在第 12 章的检查点机制之上，掌握回放（replay）与分叉（fork）两种时间旅行形态。前者"重看历史"，后者"改写历史"——它们是调试复杂工作流和做 A/B 对比实验的核心武器。

13.1 调试工作流的老大难

评估系统上线后收到投诉：某份简历被打了个离谱的低分。复盘时你面对一串问题：

• 当时模型到底输出了什么？中间状态长什么样？
• 如果当时的评分 prompt 换一个版本，结果会不同吗？
• 如果在"解析完成"那个时刻插一步人工校验，后续会怎么走？

传统应用的调试手段——断点、日志、单步——对图工作流都嫌笨拙，因为图执行的"时刻"不是代码行号，而是检查点。LangGraph 的答案优雅得近乎奢侈：既然每个超步都有快照，那就让时间回到任意快照。

时间旅行两种形态，先建立总览：

两种形态都不改写历史（第 12 章末尾的规则：回退产生新分支）——时间旅行是安全的，原始记录永远在。

13.2 共同的地基：定位历史时刻

两种形态的第一步相同：从检查点历史里挑一个时刻。第 12 章的 get_state_history 返回的每条记录自带 checkpoint_id：

from typing import Annotated, TypedDict
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

class ScreenState(TypedDict):
    resume_text: str
    score: Annotated[int, operator.add]
    log: Annotated[list[str], operator.add]

def extract(state: ScreenState) -> dict:
    return {"log": ["extract 完成"]}

def evaluate(state: ScreenState) -> dict:
    return {"score": 42, "log": ["evaluate 完成"]}

builder = StateGraph(ScreenState)
builder.add_node("extract", extract)
builder.add_node("evaluate", evaluate)
builder.add_edge(START, "extract")
builder.add_edge("extract", "evaluate")
builder.add_edge("evaluate", END)

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

config = {"configurable": {"thread_id": "debug-001"}}
graph.invoke({"resume_text": "李思远 ..."}, config)

# 拉出全部历史时刻
history = list(graph.get_state_history(config))
for s in history:
    print(s.config["configurable"]["checkpoint_id"], s.next)
# 每个 checkpoint_id 形如 1ef4f797-8b5f-6bc8-8001-8b417b28a1d3 的 UUID

挑时刻的实用技巧：get_state_history 按时间倒序返回（最新在前）。想找"evaluate 节点执行前"的时刻，认准 s.next == ('evaluate',) 的那条——"next 是 evaluate"意味着这是 evaluate 即将启动前的快照。

13.3 replay：原样重跑

replay 的诉求：不改任何东西，把某时刻之后的执行再看一遍。方法是拿那个历史 config 再 stream 一遍（输入传 None，表示"不注入新输入"）：

# 选中"evaluate 执行前"的快照
target = next(s for s in history if s.next == ("evaluate",))

# replay：从该时刻重跑后续节点
for chunk in graph.stream(None, target.config, stream_mode="updates"):
    print(chunk)
# {'evaluate': {'score': 42, 'log': ['evaluate 完成']}}

replay 的第一用途是取证复盘：逐事件观察"当时每一步做了什么"。配合 stream_mode="values"，能看到从那个时刻起的完整状态演进。

它的第二用途藏在"重跑"两个字里：当时因故障中断的运行，现在恢复执行。第 12 章的断点续跑，本质就是 replay 的工程化包装——从最新检查点 replay 剩余部分。注意措辞的精确性：replay 重跑的是该检查点之后的节点，已完成的节点（如本例的 extract）不会重新执行。

有个细节容易忽略：replay 用的仍是当前进程里的图代码。如果你 replay 时已经改了 evaluate 的实现，重跑的是新代码。想看"当时的代码跑出的当时的结果"，翻原始历史检查点的 values 即可——那才是历史原貌。

13.4 fork：从历史时刻另起一路

fork 与 replay 的操作几乎一样，意图完全不同：我要修改这个时刻的状态/输入，看看另一条时间线。

方法一：在历史 config 上叠加新输入 invoke：

# 同样从"evaluate 执行前"分叉，但这次换一份简历
for chunk in graph.stream(
    {"resume_text": "王敏：金融系统后端"},   # fork 的新输入（会合并进该时刻状态）
    target.config,
    stream_mode="updates",
):
    print(chunk)
# {'evaluate': {'score': 42, 'log': ['evaluate 完成']}}

方法二：先用 update_state 直接改写某时刻的状态，再从那里续跑。注意 update_state 会产生一个新的检查点并返回指向它的 config——续跑要用返回值，而不是原来的 target.config（后者仍指向改写前的旧检查点）：

# 在目标时刻上手动改状态：预置一个分数
forked_config = graph.update_state(target.config, {"score": 20})

# 查看分叉后的状态：分数已变，且记录了这次手工更新
snap = graph.get_state(forked_config)
print(snap.values["score"])   # 20（首次写入直接生效）

# 从改写后的时刻继续执行
for chunk in graph.stream(None, forked_config, stream_mode="updates"):
    print(chunk)

update_state 相当于"插入一个虚拟节点的补丁"：按各字段的 reducer 合并进指定时刻，产生一个新的检查点。fork 的典型实验就长这样：

"如果当时 extract 解析出的技能多两项，评分会差多少？"——update_state 改 skills，从 evaluate 前的时刻 fork，对比两条时间线的最终 score。

A/B 对比、回归测试（"图结构升级后，同一批历史输入的结论是否漂移"）、反事实分析——都建立在这套机制上。

13.5 一个完整的取证实验

把两种形态串成一个真实工作流：发现某次评估分数异常，先 replay 定位问题，再 fork 验证猜想：

config_bad = {"configurable": {"thread_id": "case-042"}}

# ...（假设该线程已经跑完一次评估，分数异常低）

# 第一步：replay 取证——从 evaluate 前的时刻重看一遍
history = list(graph.get_state_history(config_bad))
before_eval = next(s for s in history if s.next == ("evaluate",))
for chunk in graph.stream(None, before_eval.config, stream_mode="values"):
    print(chunk)          # 观察状态演进，发现 evaluate 输出异常

# 第二步：fork 验证——手工修正解析结果，看评分是否恢复
forked = graph.update_state(before_eval.config, {
    "log": ["手工修正：补充技能字段"],
})
for chunk in graph.stream(None, forked, stream_mode="values"):
    print(chunk)          # 对比两条时间线的 score

取证 → 猜想 → fork 验证 → 修复代码，整个闭环不碰生产输入、不改历史记录。这就是时间旅行给 LLM 工作流带来的调试范式——对"当时的决策"做受控实验。

常见坑与排错

1. 把 get_state_history 当正序列表用。它按时间倒序返回（最新在前）。history[0] 是最新时刻不是最初时刻；要找最早时刻取 history[-1]，要找特定节点前找 s.next 匹配项。

2. replay 时期望已完成节点重跑。replay 从选中检查点的 next 开始执行，该检查点之前完成的节点不会重跑。想从头跑，选 next == ('extract',)（即第一个节点将执行）的检查点，或干脆新开线程。

3. fork 忘了输入传 None。从历史 config stream 时传了完整新输入（如 {"resume_text": ...}），会与该时刻的历史状态合并——有时正是你要的（方法一），有时是意外污染。想"纯粹改几个字段"，用 update_state 而不是塞输入。

4. update_state 对 reducer 字段的语义想当然。update_state 的补丁同样走 reducer：对 operator.add 的 score 写 20 是"加 20"不是"设为 20"。想设置绝对值，要么算好差值，要么用 as_node 配合无 reducer 字段。这一坑是第 3 章规则在时间旅行场景的重演。

5. 时间旅行之后接着用原 config 操作。fork 后线程的"最新检查点"变成了新分支。原 config（不带 checkpoint_id 的裸 thread_id）再 invoke 会从新分支最新处继续——如果你期望的是原时间线，会得到意外结果。取证实验做完，明确下一步要用哪个 config。

6. 在 InMemorySaver 上演示时间旅行后直接上生产。开发时一切正常，生产换 Postgres 后行为不变（协议一致），但性能特征不同：长线程的检查点历史查询会随长度变慢。高频时间旅行操作的大线程，注意存储端的索引与清理策略。

7. 以为时间旅行能恢复"模型的随机性"。replay 重跑 evaluate 时，如果节点内部模型 temperature > 0，两次结果可能不同——检查点保存的是状态不是模型权重与采样种子。要可复现，节点内固定 temperature 或固定随机种子。

动手练习

1. 给 13.2 的图增加 report 节点，完整跑一个线程后：找到"evaluate 执行前"的检查点，replay 一遍，对比两次执行的 log 字段是否一致。

2. fork 实验：从"evaluate 执行前"分叉，用 update_state 给 score 预置 30 分（注意 reducer 语义，想清楚写多少），再续跑到 END。最终 score 与原时间线差多少？

3. 综合：构造一个三节点图（parse → evaluate → report），故意让 evaluate 节点在简历包含"赌博"关键词时返回异常低的分数。用 replay 定位该行为，用 fork 验证"如果 parse 阶段过滤掉该关键词，评分正常"。全程不修改原始线程的历史记录，最后用 get_state_history 确认历史未被改写。

本章速查