时间旅行与状态回溯

线上 Agent 第 5 步调错工具——需要回到第 3 步 checkpoint 重放或对比 state 差异。get_state_history 经 Saver 的 list() 列出该 thread+ns 的快照；指定 checkpoint_id 再 get_tuple 可从任意点 fork。

图 1　history 走 list（新→旧）；带 checkpoint_id 再 invoke 是 fork（对应 §4）

1. 定位

维度	内容
角色	checkpoint 历史 inspect 与回溯
输入 → 输出	`get_state_history(config)` → 快照迭代器
核心 API	`get_state`、`get_state_history`、`update_state`
依赖 LangChain	无

2. 图拓扑

节点表

节点名	职责	写 State
`a`	第一步	`step`
`b`	第二步	`step`

边表

源	目标	类型
START	a	固定
a	b	固定
b	END	固定

3. invoke 生命周期

invoke 一次 → 每 superstep put 一枚 checkpoint
get_state(cfg) → Saver.get_tuple(cfg)（无 id = 最新）
get_state_history(cfg) → Saver.list(cfg)，新→旧
带 checkpoint_id 的 invoke → get_tuple 定点加载，再 put 新 id（fork）

4. 原理

4.1 图如何向 Saver 要历史

compile(checkpointer=cp) 之后：

一次 invoke 在每个 superstep 结束 put 一枚完整 checkpoint（另有一枚 input / __start__）。线性图里「一拍一个节点」，看起来像每个 node 一张；同拍并行的多个节点只共享一枚。
get_state(cfg) → cp.get_tuple(cfg)：无 checkpoint_id 取最新，包成 StateSnapshot。
get_state_history(cfg) → cp.list(cfg)，按 id 降序，每条都是 StateSnapshot，不是节点增量日志。
时间旅行：把某条的 checkpoint_id 放进 config 再 invoke。get_tuple 加载该快照的 完整 channels，随后 put 的新 id 以它为 parent——fork，旧链仍在。

少了 checkpointer → get_state / history 报 No checkpointer set。
把 history 当成旧→新 → 第一条不是「最初」。

4.2 `StateSnapshot` 里有什么

get_state_history 产出的每条都是同一结构（与 get_state 相同）：

字段	是不是「完整 state」	含义
`values`	是	该 superstep 结束后的全量 channels（已按 reducer 合并）。线性 `a→b` 在 b 之后是 `{step: 2}`，不是只含 b 的增量。
`next`	否	下一拍要跑的节点；`()` 表示已到 END
`config`	否	`thread_id` / `checkpoint_ns` / `checkpoint_id`
`metadata`	否	`source`（`input` / `loop` / `update`）、`step`、`writes`（本拍节点的 partial）
`created_at`	否	落盘时间
`parent_config`	否	上一枚 checkpoint 的 config
`tasks`	否	本拍待执行任务；`subgraphs=True` 时才带上子图快照

节点跑完时还会 put_writes（容错用的 pending writes）。这些不是 history 里的独立条目；时间旅行只能站在 superstep 的完整快照上，不能卡在「同拍里 A 完了、B 还没跑」的中间。

图 2　history 每张是完整 values；本拍增量在 metadata.writes（对应 §4.2）

4.3 与 LangSmith：不是同一件事

Smith = 这条走过的路 + 路上花了多少资源（实际调度的节点 span、latency / token / 费用）；没走到的条件边不会出现在 trace 里。
history = 每一站的整车货物，并能从某站再开出去（完整 values + checkpoint_id fork）。

	`get_state_history`	LangSmith 追踪	LangSmith Studio / Agent Server
存什么	checkpointer 里的 `StateSnapshot`	每次 invoke 的 span 树（节点 I/O、latency、token）	服务端同一套 checkpoint，经 HTTP 暴露
要不要 Saver	必须	不必	平台自带
完整 `values`	有	一般没有（是这次调用的输入输出，不是全量 channels 链）	有（`threads.get_history` ≈ 远程 `get_state_history`）
时间旅行 / fork	`checkpoint_id` + `invoke` / `update_state`	不能从 trace 恢复图	Studio 里可以对 thread 做 replay / fork
无网	本地 Saver 可用	不上报	不可用

把 Smith 当成「history 的云版」会漏掉：没配 checkpointer 时 trace 仍在，但不能 get_state / 续跑 / 改 decision 再走另一条边。故障复现要 trace（看见当时调了什么）+ history（回到那一拍的 state）。

5. 最小可运行示例

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    step: int


def node_a(_: State) -> dict:
    return {"step": 1}


def node_b(state: State) -> dict:
    return {"step": state["step"] + 1}


builder = StateGraph(State)
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", END)

cp = InMemorySaver()
graph = builder.compile(checkpointer=cp)
cfg = {"configurable": {"thread_id": "tt-1"}}
graph.invoke({"step": 0}, cfg)

history = list(graph.get_state_history(cfg))
print(len(history))  # 线性图通常 ≥4：input / 待跑 a / a 后 / b 后
for snap in history:
    print(
        snap.metadata.get("source"),
        snap.metadata.get("step"),
        snap.values,
        snap.next,
        snap.metadata.get("writes"),
    )

6. 执行追踪

顺序（新→旧）	`values`（完整）	`next`	`metadata.writes`（本拍增量）
最新 `loop`	`{step: 2}`	`()`	`{b: {step: 2}}`
a 后	`{step: 1}`	`('b',)`	`{a: {step: 1}}`
input 后	`{step: 0}`	`('a',)`	无 / None
更早 `__start__`	缺省 / 空	`('__start__',)`	input 写入

重要配置参数

参数（API 名）	类型 / 默认值	功能说明	作用与影响	参考起点 / 常用范围	配置指导
`get_state_history(config)`	iterator	内部 `Saver.list`	只列该 thread+ns	调试	顺序新→旧
`get_state(config)`	`StateSnapshot`	内部 `get_tuple`	无 id = 最新 ≈ history[0]	运行时	读 `.config`
`checkpoint_id`	str，可选	定点 `get_tuple`	再 invoke 会 fork	从 history 拷	与 thread_id 同传
`compile(checkpointer=)`	必填	才有 list/get_tuple	无则 history 报错	InMemorySaver	生产换 DB
`list(..., filter/before/limit)`	Saver API	按 metadata / 更旧于某 id 截断	少扫	大 thread	先 limit
`update_state`	新 put	人工改态再续	不删旧链	HITL	暂停态用

7. 易踩坑

无 checkpointer 调 history：No checkpointer set。
假设 history 顺序为旧→新：官方 新→旧。
把 history 当成「每个 node 的增量」：values 是合并后的全量；增量在 metadata.writes。
并行同拍只找「A 刚跑完」：history 没有这张中间照，只有 superstep 整拍。
fork 后 thread_id 管理混乱：同 ID 会继续往该 thread 叠新 id；要分叉会话就换 ID 并自己记来源。

小结

get_state_history = 该 thread+ns 上每枚 superstep 完整 StateSnapshot（values 是全量 state）。
本拍节点写了什么看 metadata.writes；并行节点不会各占一条 history。
checkpoint_id 再 invoke 是 fork；顺序 新→旧。