LangGraph源码学习路线：第 19 篇 Time Travel、update_state 与 Debug Replay源码解剖#

1
Checkpoint / Functional API
2
  ↓
3
Time Travel / update_state / Replay
4
  ↓
5
综合复盘与工程判断
6
  ↓
7
工程化 Agent 能力

1
config = {"configurable": {"thread_id": "trip-001"}}
2

3
answer = graph.invoke(
4
    {"messages": [("user", "帮我规划大阪 4 天游")], "budget": "medium"},
5
    config=config,
6
)
7

8
current = graph.get_state(config)
9
history = list(graph.get_state_history(config))
10
before_bad_tool = history[2]
11

12
fork_config = graph.update_state(
13
    before_bad_tool.config,
14
    {"budget": "high", "warnings": ["用户后来确认预算更高"]},
15
)
16

17
for event in graph.stream(None, config=fork_config, stream_mode="debug"):
18
    print(event)

1
def time_travel_debug(graph, base_config, patch):
2
    latest_snapshot = graph.get_state(base_config)
3
    history = list(graph.get_state_history(base_config))
4

5
    target = choose_snapshot(
6
        history,
7
        predicate=lambda s: "search" in s.metadata.get("source", ""),
8
    )
9

10
    fork_config = graph.update_state(
11
        target.config,
12
        values=patch,
13
        as_node="human_fix",
14
    )
15

16
    debug_events = []
17
    for event in graph.stream(None, config=fork_config, stream_mode="debug"):
18
        debug_events.append(event)
19

20
    final_snapshot = graph.get_state(fork_config)
21
    return final_snapshot, debug_events

1
def replay_from_checkpoint(graph, fork_config):
2
    snapshot = graph.get_state(fork_config)
3

4
    if snapshot.next:
5
        return graph.invoke(None, config=fork_config)
6

7
    if not snapshot.next and needs_new_input(snapshot):
8
        return graph.invoke({"messages": [("user", "继续优化")]}, config=fork_config)
9

10
    return snapshot.values

1
langgraph/
2
├── graph/
3
├── pregel/
4
├── types.py
5
└── func/
6

7
langchain_core/
8
├── runnables/
9
├── tools/
10
├── retrievers.py
11
└── messages/

1
1. 先读官方概念文档，确认公共契约。
2
2. 再读公开 API 或装饰器入口。
3
3. 顺着构建期对象进入 runtime。
4
4. 追踪一次 invoke / stream / tool call / task call。
5
5. 单独检查异常、重试、恢复和扩展点。
6
6. 最后回到旅行规划助手做工程判断。

1
Public API / Decorator / Tool Protocol
2
  ↓
3
Runtime wrapper
4
  ↓
5
State / Config / Context
6
  ↓
7
Storage / Model / Tool / Graph Runtime

1
公共 API
2
  负责：给业务代码稳定入口。
3
  不负责：暴露所有内部调度细节。
4

5
运行时协议
6
  负责：让状态、配置、工具、模型和持久化协作。
7
  不负责：替业务判断所有策略。
8

9
扩展接口
10
  负责：允许业务插入可维护的定制逻辑。
11
  不负责：保证错误扩展仍然安全。

1
先找公开入口
2
  ↓
3
确认输入输出类型
4
  ↓
5
沿调用方追到核心实现
6
  ↓
7
记录状态与对象变化
8
  ↓
9
检查分支、异常和结束条件
10
  ↓
11
回到设计目的

1
def compile_graph_with_persistence(builder, checkpointer):
2
    graph = builder.compile(checkpointer=checkpointer)
3

4
    graph.get_state = lambda config: read_latest_snapshot(graph, config)
5
    graph.get_state_history = lambda config, **kw: iter_snapshots(graph, config, **kw)
6
    graph.update_state = lambda config, values, **kw: write_state_update(
7
        graph=graph,
8
        config=config,
9
        values=values,
10
        **kw,
11
    )
12
    return graph

1
def build_state_snapshot(checkpoint_tuple, graph):
2
    checkpoint = checkpoint_tuple.checkpoint
3
    config = checkpoint_tuple.config
4
    metadata = checkpoint_tuple.metadata
5

6
    values = graph.channels_to_values(checkpoint["channel_values"])
7
    next_tasks = graph.compute_next_tasks(checkpoint)
8
    tasks = graph.restore_task_metadata(checkpoint)
9

10
    return StateSnapshot(
11
        values=values,
12
        next=tuple(task.name for task in next_tasks),
13
        config=config,
14
        metadata=metadata,
15
        tasks=tasks,
16
        created_at=checkpoint.get("ts"),
17
        parent_config=checkpoint_tuple.parent_config,
18
    )

1
def make_checkpoint_config(thread_id, checkpoint_id=None, namespace=None):
2
    configurable = {"thread_id": thread_id}
3
    if checkpoint_id is not None:
4
        configurable["checkpoint_id"] = checkpoint_id
5
    if namespace is not None:
6
        configurable["checkpoint_ns"] = namespace
7

8
    return {"configurable": configurable}

1
def get_state(graph, config):
2
    runtime_config = ensure_config(config)
3
    checkpoint_tuple = graph.checkpointer.get_tuple(runtime_config)
4

5
    if checkpoint_tuple is None:
6
        return StateSnapshot(values={}, next=(), config=runtime_config, metadata={}, tasks=())
7

8
    return build_state_snapshot(checkpoint_tuple, graph)

1
def get_state_history(graph, config, limit=None, before=None):
2
    runtime_config = ensure_config(config)
3

4
    for checkpoint_tuple in graph.checkpointer.list(
5
        runtime_config,
6
        limit=limit,
7
        before=before,
8
    ):
9
        yield build_state_snapshot(checkpoint_tuple, graph)

1
def update_state(graph, config, values, as_node=None):
2
    snapshot = graph.get_state(config)
3

4
    if as_node is None:
5
        as_node = infer_update_source(snapshot)
6

7
    writes = graph.map_values_to_channel_writes(values, as_node=as_node)
8
    new_checkpoint = graph.apply_writes_to_checkpoint(
9
        base_config=snapshot.config,
10
        base_values=snapshot.values,
11
        writes=writes,
12
        metadata={"source": "update_state", "as_node": as_node},
13
    )
14

15
    return graph.checkpointer.put(snapshot.config, new_checkpoint, metadata=new_checkpoint.metadata)

1
def invoke_from_checkpoint(graph, input_value, config):
2
    snapshot = graph.get_state(config)
3
    runtime = PregelLoop(graph=graph, input=input_value, checkpoint=snapshot, config=config)
4

5
    while runtime.has_next_task():
6
        task = runtime.next_task()
7
        writes = runtime.run_task(task)
8
        runtime.apply_writes(writes)
9
        runtime.checkpoint_after_step()
10

11
    return runtime.output()

1
def validate_update_state(snapshot, values_patch):
2
    allowed = {}
3
    rejected = {}
4

5
    for key, value in values_patch.items():
6
        if key not in snapshot.values:
7
            rejected[key] = "not_declared_in_state_schema"
8
        elif key in {"next", "tasks", "config", "metadata"}:
9
            rejected[key] = "runtime_field_not_state_value"
10
        else:
11
            allowed[key] = value
12

13
    return allowed, rejected

1
def reason_about_replay(snapshot, external_system):
2
    replay_changes = {
3
        "langgraph_state": "can_branch_from_checkpoint",
4
        "checkpoint_history": "append_new_branch",
5
        "debug_events": "emit_again_for_new_run",
6
    }
7
    not_changed = {
8
        "database_rows": external_system.requires_manual_rollback,
9
        "sent_emails": "already_sent",
10
        "payment_requests": "must_use_business_idempotency",
11
    }
12
    return replay_changes, not_changed

1
def decide_replay_path(snapshot, new_input):
2
    if snapshot.next:
3
        return "continue_pending_tasks"
4
    if new_input is not None:
5
        return "start_next_turn_from_snapshot"
6
    return "inspect_only_no_execution"

1
def load_snapshot_or_fail(graph, config):
2
    if graph.checkpointer is None:
3
        raise ValueError("time travel requires a checkpointer")
4

5
    snapshot = graph.get_state(config)
6
    if snapshot is None:
7
        raise LookupError("checkpoint not found for config")
8

9
    return snapshot

1
def debug_replay_workflow(graph, thread_id):
2
    base_config = {"configurable": {"thread_id": thread_id}}
3
    history = list(graph.get_state_history(base_config, limit=20))
4
    target = select_snapshot(history, reason="before_wrong_tool_result")
5

6
    fork_config = graph.update_state(
7
        target.config,
8
        {"selected_docs": [], "warnings": ["removed stale search result"]},
9
        as_node="human_debugger",
10
    )
11

12
    events = list(graph.stream(None, config=fork_config, stream_mode="debug"))
13
    final = graph.get_state(fork_config)
14
    return {"fork_config": fork_config, "events": events, "final": final}

1
@entrypoint(checkpointer=checkpointer)
2
def support_workflow(request: str) -> str:
3
    docs = retrieve_docs(request).result()
4
    draft = write_answer(docs).result()
5
    return draft
6

7
config = {"configurable": {"thread_id": "case-001"}}
8
history = list(support_workflow.get_state_history(config))
9
fork = support_workflow.update_state(
10
    history[1].config,
11
    {"retrieve_docs": [{"title": "corrected doc"}]},
12
)
13
answer = support_workflow.invoke(None, config=fork)

1
def correlate_trace_and_checkpoint(trace_run, graph):
2
    thread_id = trace_run.metadata["thread_id"]
3
    failed_node = trace_run.error.metadata.get("node")
4

5
    config = {"configurable": {"thread_id": thread_id}}
6
    history = list(graph.get_state_history(config))
7

8
    candidate = find_snapshot_before_node(history, failed_node)
9
    return candidate.config

1
性能：额外抽象会增加序列化、检索、trace 和存储成本。
2
可靠性：可恢复机制要求状态 schema、幂等工具和错误分类清楚。
3
安全：metadata、state、tool args、retrieved docs 都可能包含敏感信息。
4
可观测性：至少记录 thread_id、node/tool、策略命中、错误和耗时。

1
公开 API 通常只负责归一化和装配；真正的运行语义在 runtime、协议对象和扩展点之间流转。

1
state、store、Document、metadata、trace、prompt messages 是不同协议；只有经过明确注入的内容才进入模型上下文。

1
可恢复异常可以转成 retry、fallback、interrupt；不可恢复异常必须上抛并进入 trace。

1
StateGraph.compile(checkpointer=...)
2
  ↓
3
checkpoint saver 绑定 graph runtime
4
  ↓
5
get_state / get_state_history / update_state
6
  ↓
7
StateSnapshot + checkpoint config 坐标系

1
graph.invoke(input, thread_id)
2
  ↓
3
每步写入 checkpoint history
4
  ↓
5
get_state_history 定位 StateSnapshot
6
  ↓
7
update_state(snapshot.config, values)
8
  ↓
9
生成 fork config
10
  ↓
11
invoke/stream 从分支继续并输出 debug events

1
可改 → state schema 中的业务字段，并遵守 reducer
2
不可改 → next / tasks / config / metadata 等运行时坐标
3
可重放 → checkpoint 分支之后的后续调度
4
不可回滚 → 外部数据库、邮件、支付等副作用
5
缺失 history → 明确报错，不应静默当作空状态

1
综合复盘：旅行规划助手完整执行链路与框架工程判断

1
这项能力在旅行规划助手中应该放在哪一层？
2
它和前一篇能力如何协作？
3
哪些边界需要通过源码证据确认？