基于本文回答

播面 播面

刷题像听歌，多听自然懂
0
评论
请从左侧选择一个分类

LangGraph 的 Time Travel（时间旅行 / 回滚）

知识点图片

在 LangGraph 中，“时间旅行”（Time Travel）是其最核心且最具特色的高级功能之一。它允许你查看 Agent（智能体）过去的执行状态，回滚到任意一个历史节点，甚至修改当时的状态并从那里重新开始执行（类似创建平行宇宙/分支）。

这个功能非常适合用于调试（Debugging）、人工干预（Human-in-the-loop, HITL）以及探索不同的执行分支。

以下是关于 LangGraph 时间旅行的详细解析和实战指南：

1. 时间旅行的前提条件：检查点（Checkpointer）

要实现时间旅行，Graph 必须拥有“记忆”。在 LangGraph 中，这是通过在编译时传入 checkpointer 来实现的。每一次节点执行完毕，Checkpointer 都会保存当前状态的快照。

  • Thread ID (thread_id): 代表一次完整的会话或任务流。
  • Checkpoint ID (checkpoint_id): 代表任务流中某一个具体步骤的“快照版本号”。

2. 时间旅行的两大核心操作

操作 A：回放（Replay）

回到过去某个时间点，并从那里重新开始执行。不修改当时的状态，只是单纯地重新跑一遍后续的流程（或者测试后续节点代码修改后的效果）。

操作 B：修改并继续（Fork / Update State）

回到过去某个时间点，手动修改那个时候的 State（例如：纠正 LLM 产生的一个幻觉错误，或者修改用户的输入），然后以修改后的状态为基础，继续执行。这会创建一个全新的 checkpoint_id 历史分支。

3. 代码实战演示

为了让你直观理解，我们构建一个极其简单的计算图：

  • 初始值 val = 1
  • Node 1 (Add): val + 1
  • Node 2 (Multiply): val * 2

第一步：构建图并正常运行

python 
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

# 1. 定义状态
class State(TypedDict):
    val: int

# 2. 定义节点
def add_node(state: State):
    return {"val": state["val"] + 1}

def multiply_node(state: State):
    return {"val": state["val"] * 2}

# 3. 构建图
builder = StateGraph(State)
builder.add_node("add", add_node)
builder.add_node("multiply", multiply_node)
builder.add_edge(START, "add")
builder.add_edge("add", "multiply")
builder.add_edge("multiply", END)

# 4. 必须加入 Checkpointer！
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

# 5. 正常运行
config = {"configurable": {"thread_id": "test_thread_1"}}
initial_state = {"val": 1}

print("=== 正常运行 ===")
for event in graph.stream(initial_state, config):
    print(event)
# 预期输出:
# {'add': {'val': 2}}
# {'multiply': {'val': 4}}

第二步：获取历史记录 (Get History)

现在，图执行完毕了，最后的值是 4。我们来看看它的历史“快照”。

python 
print("\n=== 查看历史状态 ===")
# 获取该 Thread 的所有历史记录 (倒序)
history = list(graph.get_state_history(config))

for h in history:
    print(f"Step: {h.next}, State: {h.values}, Checkpoint_ID: {h.config['configurable']['checkpoint_id']}")

# 输出大概是这样的：
# Step: (), State: {'val': 4}, Checkpoint_ID: <id_3>  (结束状态)
# Step: ('multiply',), State: {'val': 2}, Checkpoint_ID: <id_2> (add 执行完，准备执行 multiply)
# Step: ('add',), State: {'val': 1}, Checkpoint_ID: <id_1> (START 执行完，准备执行 add)

第三步：时间旅行 - 回放 (Replay)

假设我们想回到 add 节点刚刚执行完，multiply 节点还没执行的时候（即状态值为 2 的那个快照），重新往下跑。

python 
# 找到我们想回到的历史状态 (例如倒数第二个状态)
past_state = history[1] 
past_config = past_state.config # 包含了当时的 thread_id 和 checkpoint_id

print("\n=== 时间旅行：回放 (Replay) ===")
# 直接使用过去的 config 调用 graph.stream 或 graph.invoke
for event in graph.stream(None, past_config):
    print(event)
# 预期输出:
# {'multiply': {'val': 4}}  (它直接从过去的快照处继续运行了)

第四步：时间旅行 - 修改状态并分叉 (Fork/Update)

这是最强大的功能。假设觉得 add 节点算出的 val=2 错了，我作为一个“人类监督员”，想回到过去，把那个时刻的 val 强行改成 10，然后再让图继续运行。

python 
print("\n=== 时间旅行：修改并继续 (Update & Fork) ===")

# 1. 依然使用我们刚才获取的 past_config (停留在 val=2 的那个时刻)
# 2. 使用 graph.update_state 修改状态
# 注意：这不会覆盖原历史，而是会创建一个新的 checkpoint_id (平行宇宙)

graph.update_state(
    past_config, 
    {"val": 10}, # 将过去的值强制修改为 10
)

# 获取更新后的最新 config
new_config = {"configurable": {"thread_id": "test_thread_1"}}

# 3. 让图继续运行
for event in graph.stream(None, new_config):
    print(event)
    
# 预期输出:
# {'multiply': {'val': 20}} (10 * 2 = 20，图顺着被修改的历史继续跑了！)

4. 关键 API 总结

  1. graph.get_state_history(config): 查看某个 thread_id 的所有历史快照。
  2. graph.invoke(None, past_config): 回放。传入包含旧 checkpoint_id 的 config，图会从那个时刻继续运行。
  3. graph.update_state(config, state_update, as_node="node_name"): 修改状态。
    • config: 你想要修改的那个时刻的配置。
    • state_update: 你想要覆盖/更新的数据。
    • as_node (可选): 假装这个更新是由哪个节点执行的（这会影响图接下来决定走哪条边）。

5. 常见应用场景 (Use Cases)

  • 纠正 LLM 的错误 (Human-in-the-loop): Agent 在收集资料写报告。它写完草稿后停下来等待。人类审查后发现第一段有幻觉。人类通过 update_state 修改第一段的内容，然后让 Agent 继续润色。
  • 探索不同路径 (What-If Analysis): 在做决策树流时，回到一个分支点，强行修改用户的意图（比如从“想要退款”改成“想要换货”），看看下游的 Agent 会作何反应，用于测试系统的鲁棒性。
  • 断点续传/重试机制: 某个 API 调用失败导致 Graph 崩溃。修复外部 API 后，直接找到崩溃前的 checkpoint_id 继续运行，而不需要从头开始消耗 Token。
00:00
00:00