Langgraph logo Langgraph
登录
10. 持久化与时间旅行

10.持久化与时间旅行

LangGraph 的持久化机制是其最具特色的功能之一。不同于传统的对话系统只在内存中维护当前状态，LangGraph 会在图的每个执行步骤自动保存状态快照。这种设计不仅实现了故障恢复和人工介入，更重要的是让我们能够像操作版本控制系统一样，随时回到任意历史状态，探索不同的执行路径。

检查点快照原理

检查点是 LangGraph 持久化系统的核心概念。每当图执行完一个节点，检查点器（checkpointer）就会捕获当前状态的完整快照，并将其保存到持久化存储中。这个过程对开发者完全透明，无需手动干预。

检查点的数据结构

一个检查点对应一个 StateSnapshot 对象，包含五个关键字段：

  • values：当前时刻所有状态通道的值
  • next：接下来要执行的节点名称元组
  • config：与此检查点关联的配置信息，包含 thread_id 和 checkpoint_id
  • metadata：元数据，记录来源、写入内容和执行步骤
  • parent_config：父检查点的配置，形成历史链条

让我们看一个具体的例子。假设有一个简单的图，包含两个节点 node_a 和 node_b，状态包含 foo 和 bar 两个字段：

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

class State(TypedDict):
    foo: str
    bar: Annotated[list[str], add]

def node_a(state: State):
    return {"foo": "a", "bar": ["a"]}

def node_b(state: State):
    return {"foo": "b", "bar": ["b"]}

workflow = StateGraph(State)
workflow.add_node(node_a)
workflow.add_node(node_b)
workflow.add_edge(START, "node_a")
workflow.add_edge("node_a", "node_b")
workflow.add_edge("node_b", END)

checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": ""}, config)

这段代码创建了一个简单的线性工作流。node_a 将 foo 设为 "a" 并在 bar 列表中添加 "a"，node_b 将 foo 设为 "b" 并在 bar 列表中添加 "b"。关键点在于 bar 字段使用了 add 作为 reducer，这意味着新值会与旧值合并，而不是完全替换。

执行完成后，系统会生成四个检查点：

  1. 初始空检查点，next 为 ('__start__',)
  2. 包含用户输入 {'foo': '', 'bar': []} 的检查点，next 为 ('node_a',)
  3. 包含 node_a 输出 {'foo': 'a', 'bar': ['a']} 的检查点，next 为 ('node_b',)
  4. 包含 node_b 输出 {'foo': 'b', 'bar': ['a', 'b']} 的检查点，next 为空

每个检查点都精确记录了图在特定时刻的完整状态，包括数据值和下一步执行计划。这种细粒度的快照机制为后续的时间旅行功能奠定了基础。

检查点的存储机制

检查点器负责将快照持久化到存储后端。LangGraph 提供了多种实现：

  • InMemorySaver：内存存储，适合开发和测试
  • SqliteSaver：SQLite 数据库，适合本地工作流
  • PostgresSaver：PostgreSQL 数据库，生产环境首选

检查点器接口定义了四个核心方法：

# 存储检查点
checkpointer.put(config, checkpoint, metadata, {})

# 获取特定检查点
checkpointer.get_tuple(config)

# 列出检查点历史
list(checkpointer.list(config))

# 删除线程所有检查点
checkpointer.delete_thread(thread_id)

异步执行时，对应的方法为 aput、aget_tuple、alist 和 adelete_thread。这种设计确保了无论使用同步还是异步方式执行图，都能正确持久化状态。

检查点存储时需要进行序列化。默认使用 JsonPlusSerializer，支持 LangChain 和 LangGraph 原语、日期时间、枚举等多种类型。对于特殊对象（如 Pandas DataFrame），可以启用 pickle_fallback 选项。生产环境中，还可以通过 EncryptedSerializer 对存储的数据进行加密，只需设置 LANGGRAPH_AES_KEY 环境变量即可。

线程状态生命周期

线程是 LangGraph 中隔离不同执行上下文的机制。每个线程对应一个独立的 thread_id，拥有自己的检查点历史。这种设计使得多租户应用或并行对话变得简单自然。

线程的创建与使用

要使用持久化功能，必须在执行图时提供线程 ID：

config = {"configurable": {"thread_id": "1"}}
result = graph.invoke(input_data, config)

线程 ID 可以是任意字符串，通常使用用户 ID、会话 ID 或其他业务标识。同一个线程 ID 的多次调用会共享状态历史，而不同线程 ID 则完全隔离。

让我们看一个对话机器人的例子：

from langchain_core.messages import HumanMessage, AIMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START
from typing_extensions import TypedDict
from typing import Annotated
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]

def chatbot(state: State):
    # 简化的响应逻辑
    last_message = state["messages"][-1]
    if "name" in last_message.content.lower():
        response = f"Your name is {state['messages'][0].content.split()[-1]}"
    else:
        response = "Hello! How can I help?"
    return {"messages": [AIMessage(content=response)]}

memory = InMemorySaver()
graph = StateGraph(State).add_node("chatbot", chatbot).add_edge(START, "chatbot").compile(checkpointer=memory)

# 第一次对话
config1 = {"configurable": {"thread_id": "user_123"}}
graph.invoke({"messages": [HumanMessage(content="My name is Alice")]}, config1)

# 第二次对话，同一线程
graph.invoke({"messages": [HumanMessage(content="What's my name?")]}, config1)
# 输出包含 "Your name is Alice"

# 新线程
config2 = {"configurable": {"thread_id": "user_456"}}
graph.invoke({"messages": [HumanMessage(content="What's my name?")]}, config2)
# 输出不包含名字信息，因为这是新线程

这个例子展示了线程的核心价值：状态隔离。user_123 线程记住了用户名字，而 user_456 线程没有之前的上下文。每个线程的状态历史独立保存，互不影响。

状态历史的遍历

LangGraph 提供了查看完整状态历史的能力：

# 获取最新状态快照
latest_snapshot = graph.get_state(config)

# 获取完整历史，按时间倒序排列
history = list(graph.get_state_history(config))

for snapshot in history:
    print(f"Step: {snapshot.metadata['step']}")
    print(f"Next nodes: {snapshot.next}")
    print(f"Values: {snapshot.values}")
    print("-" * 40)

get_state_history 返回一个生成器，按时间倒序产出所有检查点。每个快照包含完整的执行信息：步骤编号、下一步计划、状态值、元数据等。通过遍历历史，可以清晰看到图执行的完整轨迹。

状态历史在调试时特别有用。当图的行为不符合预期时，可以逐检查点检查状态变化，定位问题所在。结合 LangSmith 的可视化工具，这种细粒度的历史追踪能力让复杂代理系统的调试变得可控。

状态的更新与分支

除了查看历史，LangGraph 还允许直接修改历史状态，创建新的执行分支：

# 获取某个历史状态
history = list(graph.get_state_history(config))
target_state = history[2]  # 选择第三个检查点

# 修改状态值
new_config = graph.update_state(
    target_state.config,
    values={"messages": [HumanMessage(content="Modified message")]}
)

# 从新状态继续执行
graph.invoke(None, new_config)

update_state 方法会在指定检查点的基础上创建一个新的检查点，保留历史但修改指定值。这相当于在版本控制系统中创建了一个新分支。后续执行会从这个新分支继续，不影响原始历史。

这种能力在人工介入场景中尤为重要。例如，当代理执行出错时，人类审核者可以回到出错前的状态，修正输入或中间结果，然后让代理从修正后的状态继续执行，避免从头开始。

时间旅行状态重放

时间旅行是 LangGraph 最具魅力的特性。它允许将图的状态回退到任意历史检查点，然后重新执行后续步骤。这在调试、实验和交互式应用中价值巨大。

基本重放机制

重放操作非常简单：只需在调用图时提供目标检查点的 ID：

# 获取历史状态
history = list(graph.get_state_history(config))
target_checkpoint = history[3]  # 选择第四个检查点

# 从该检查点重放
graph.invoke(None, target_checkpoint.config)

当 invoke 的第一个参数为 None 且配置中包含 checkpoint_id 时，LangGraph 会自动重放该检查点之前的所有步骤，然后继续执行后续节点。重放过程不会重新执行已经完成的操作，而是直接从检查点加载状态，确保高效且一致。

让我们看一个更完整的例子，模拟一个研究助理代理：

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

class State(TypedDict):
    topic: NotRequired[str]
    research_notes: NotRequired[str]
    summary: NotRequired[str]

def research_topic(state: State):
    # 模拟研究过程
    return {"research_notes": f"Research on {state.get('topic', 'unknown')}: ..."}

def write_summary(state: State):
    # 模拟总结过程
    return {"summary": f"Summary of {state['research_notes']}"}

# 构建图
builder = StateGraph(State)
builder.add_node("research", research_topic)
builder.add_node("summarize", write_summary)
builder.add_edge(START, "research")
builder.add_edge("research", "summarize")
builder.add_edge("summarize", END)

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

# 第一次执行
config = {"configurable": {"thread_id": "research_1"}}
result = graph.invoke({"topic": "AI agents"}, config)

# 查看历史
history = list(graph.get_state_history(config))
for i, state in enumerate(history):
    print(f"{i}: Step {state.metadata['step']}, Next: {state.next}")

# 输出类似：
# 0: Step 2, Next: ()
# 1: Step 1, Next: ('summarize',)
# 2: Step 0, Next: ('research',)
# 3: Step -1, Next: ('__start__',)

# 回到研究阶段之后、总结阶段之前的状态
research_completed_state = history[1]

# 修改研究笔记
new_config = graph.update_state(
    research_completed_state.config,
    values={"research_notes": "Modified research content"}
)

# 从修改后的状态继续总结
result = graph.invoke(None, new_config)

这个例子展示了时间旅行的完整流程：执行图、查看历史、选择检查点、修改状态、继续执行。通过这种方式，可以探索不同的执行路径，而无需从头开始。

实际应用场景

时间旅行在多种场景下都有实际价值：

调试复杂代理：当代理做出错误决策时，可以回到决策点之前，检查当时的上下文信息，理解错误原因。甚至可以修改某些状态值，观察不同输入下的决策结果，快速验证修复方案。

交互式应用：在对话系统中，用户可能想修改之前的某个问题或回答。通过时间旅行，可以回到对话的任意点，修改消息后让对话继续，系统会自动应用新的上下文重新生成后续回复。

实验与优化：对于非确定性系统（如使用 LLM 的代理），可以多次重放同一段历史，观察不同随机种子或模型参数下的行为差异，帮助优化系统配置。

教学演示：在展示代理工作原理时，可以逐步重放执行过程，让学生清晰看到每个步骤的状态变化，加深理解。

历史状态分支更新

时间旅行不仅限于重放，更重要的是能够基于历史状态创建新的分支。这种分支能力让 LangGraph 具备了类似 Git 的版本控制特性。

分支的创建与管理

每次调用 update_state 或从非最新检查点继续执行时，都会创建一个新的分支：

# 获取历史状态
history = list(graph.get_state_history(config))
original_state = history[2]

# 创建第一个分支：修改主题
branch1_config = graph.update_state(
    original_state.config,
    values={"topic": "machine learning"}
)

# 创建第二个分支：修改另一个字段
branch2_config = graph.update_state(
    original_state.config,
    values={"priority": "high"}
)

# 两个分支独立执行
result1 = graph.invoke(None, branch1_config)
result2 = graph.invoke(None, branch2_config)

每个分支都有自己的检查点历史，从分叉点开始与主历史分离。通过 thread_id 和 checkpoint_id 的组合，可以精确访问任意分支的任意状态。

在云平台的应用

LangGraph Platform 提供了完整的 API 支持时间旅行和分支管理：

from langgraph_sdk import get_client

client = get_client(url="https://your-deployment.com")

# 创建线程并执行
thread = await client.threads.create()
await client.runs.wait(
    thread["thread_id"],
    "agent",
    input={"topic": "research"}
)

# 获取历史状态
states = await client.threads.get_history(thread["thread_id"])
selected_state = states[2]

# 修改状态并继续
new_config = await client.threads.update_state(
    thread["thread_id"],
    {"topic": "modified topic"},
    checkpoint_id=selected_state["checkpoint_id"]
)

# 从修改后的状态继续
await client.runs.wait(
    thread["thread_id"],
    "agent",
    input=None,
    checkpoint_id=new_config["checkpoint_id"]
)

云平台还提供了 Studio 可视化工具，可以图形化查看线程历史、选择检查点、修改状态并创建分支。这种交互式调试方式极大提升了开发效率。

分支策略与最佳实践

在实际应用中，分支管理需要遵循一些最佳实践：

命名规范：为不同目的的线程和检查点使用清晰的命名规则。例如，user_123_main 表示主对话线程，user_123_experiment_1 表示实验分支。

清理策略：分支过多会占用存储空间。可以设置 TTL（Time-To-Live）自动清理旧分支，或在应用层定期删除不再需要的分支。

状态兼容性：修改图结构（添加或删除节点）时，要确保与旧检查点兼容。LangGraph 支持大部分拓扑变更，但删除或重命名节点可能导致从旧检查点恢复时出错。

并发控制：多个用户或进程同时操作同一线程时，需要注意并发冲突。LangGraph 提供了乐观锁机制，当检测到冲突时会抛出异常，应用层需要妥善处理。

总结

LangGraph 的持久化与时间旅行功能为构建可靠、可调试的代理系统提供了坚实基础。检查点机制自动保存每个执行步骤的完整状态，线程系统隔离不同执行上下文，时间旅行让我们能够回到任意历史点并探索不同路径。

这些能力不仅提升了开发效率，更重要的是开启了新的应用可能性。从故障恢复到人工介入，从交互式对话到实验优化，持久化与时间旅行让代理系统变得更加灵活和可控。

掌握这些概念后，可以进一步探索 LangGraph Platform 的高级功能，如分布式部署、语义搜索集成、认证授权等，构建真正生产级的代理应用。

上一章

流式与实时交互

 下一章

LangGraph 平台功能