Langgraph持久化基础概念

LangGraph 是构建多智能体 / 大语言模型工作流的核心框架，而持久化是它的关键能力之一。

什么是持久化

LangGraph 的持久化机制通过检查点器（checkpointer）实现。当您使用检查点器编译图时，它会在每个超步（super-step）保存图状态的检查点。这些检查点保存到线程中，可以在图执行后访问。

我们拆成几个关键术语来讲：

检查点器（Checkpointer）

这是实现持久化的核心组件，你可以把它理解成游戏里的 “存档系统”。
超步（Super-step）

LangGraph 图执行的一个基本单元，指的是图从开始执行到下一次遇到用户输入或分支决策的一段完整流程。每走完一个超步，检查点器就会自动保存当前的所有状态。
图状态（State）

包括对话历史、工具调用结果、中间变量、分支判断的上下文等所有信息。
线程（Thread）

相当于一个独立的会话 / 对话 ID，所有的检查点都会和这个线程绑定，方便后续随时调用。

一句话总结：

持久化就是让 LangGraph 工作流可以像游戏一样 “随时存档、随时读档”，每一步的状态都被完整保存下来。

核心价值

持久化不是一个 “锦上添花” 的功能，而是 LangGraph 很多高级特性的基础。它支持以下四大核心能力：

功能	通俗解释	实际应用场景
人机交互	人工检查、中断和批准图步骤	比如 AI 要调用一个高风险工具（如转账、发送邮件），工作流会自动暂停，等待人工确认后再继续执行。
记忆	在交互之间保留状态	用户中途退出对话，下次再进来时，工作流能直接从上次的状态继续，不用重新输入上下文。
时间旅行	重放之前的图执行	可以回到之前任意一个检查点，修改参数后重新执行，用来调试、复现问题或尝试不同分支。
容错	错误恢复和故障转移	如果工作流中途因为网络、服务崩溃而中断，可以从最近的检查点恢复执行，而不是从头再来。

LangGraph 构建的是有状态的、多步骤的、可中断的复杂工作流，而不是一次性的 API 调用。

没有持久化，一旦流程中断，所有的上下文都会丢失，用户体验极差。
没有持久化，就无法实现人工介入、会话记忆、问题调试这些企业级应用必备的功能。

检查点器就是 LangGraph 实现这些能力的基石，它让 “一次性的对话” 变成了 “可持续的会话”，让 “黑盒的执行” 变成了 “可观测、可干预、可恢复的流程”。

线程（Threads）

在 LangGraph 里，线程（Thread） 是持久化机制的核心载体。

核心定义

Thread_id 是 checkpointer 的唯一标识符，包含一系列运行的累积状态。

你可以把 thread_id 理解成「会话 ID」或「存档槽位」。
同一个图（Workflow），可以用不同的 thread_id 跑多个独立的对话 / 流程，互不干扰。
每个 thread_id 里，会保存这个流程从开始到现在的所有检查点（Checkpoint）和状态历史。

代码示例

thread_example.py

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

# ------------------------------
# 1. 定义状态模式
# ------------------------------
class State(TypedDict):
    foo: str
    bar: Annotated[list[str], add]  # 使用 add 操作符实现列表追加更新

# ------------------------------
# 2. 定义节点函数
# ------------------------------
def node_a(state: State):
    print("--- 执行 node_a ---")
    return {"foo": "a", "bar": ["a"]}

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

# ------------------------------
# 3. 构建工作流图
# ------------------------------
workflow = StateGraph(State)
workflow.add_node("node_a", node_a)
workflow.add_node("node_b", node_b)

# 设置边：START → node_a → node_b → END
workflow.add_edge(START, "node_a")
workflow.add_edge("node_a", "node_b")
workflow.add_edge("node_b", END)

# ------------------------------
# 4. 配置检查点器并编译图
# ------------------------------
checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

# ------------------------------
# 5. 运行示例
# ------------------------------
if __name__ == "__main__":
    # 必须指定 thread_id，LangGraph 才会启用持久化
    config: RunnableConfig = {"configurable": {"thread_id": "1"}}

    # 第一次调用：传入初始状态
    print("===== 第一次调用 =====")
    result = graph.invoke({"foo": "", "bar": []}, config)
    print(f"最终状态: {result}")
    # 输出: {'foo': 'b', 'bar': ['a', 'b']}

    # 第二次调用：使用同一个 thread_id，状态会被保留
    print("\n===== 第二次调用（同一个 thread_id） =====")
    result2 = graph.invoke({"foo": "override_test", "bar": ["c"]}, config)
    print(f"最终状态: {result2}")
    # 输出: {'foo': 'b', 'bar': ['a', 'b', 'c', 'a', 'b']}
    # 解释：bar 会追加，foo 会被 node_b 覆盖

    # 第三次调用：使用不同的 thread_id，状态相互隔离
    print("\n===== 第三次调用（不同 thread_id） =====")
    config_new: RunnableConfig = {"configurable": {"thread_id": "2"}}
    result3 = graph.invoke({"foo": "new_thread", "bar": ["x"]}, config_new)
    print(f"最终状态: {result3}")
    # 输出: {'foo': 'b', 'bar': ['x', 'a', 'b']}

代码逐行解读：

导入依赖

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

StateGraph：LangGraph 的核心，用来构建状态机工作流。
InMemorySaver：内存型检查点器，用来把状态保存在内存里（生产环境会用数据库 / Redis 等）。
RunnableConfig：LangChain/LangGraph 的配置对象，用来传递 thread_id 等参数。
TypedDict：用来定义状态（State）的结构。
add：列表合并操作符，配合 Annotated 实现状态的追加更新。

定义状态模式（State Schema）
1
2
3
class State(TypedDict):
foo: str
bar: Annotated[list[str], add]
- State 是图的状态结构，每次节点执行都会读取 / 更新这个状态。
- foo: str：普通字符串字段，节点更新时会直接覆盖旧值。
- bar: Annotated[list[str], add]：特殊标记，表示这个列表会用 add 操作追加内容，而不是覆盖。
  - 比如节点 A 返回 ["a"]，节点 B 返回 ["b"]，最终 bar 会变成 ["a", "b"]。
定义节点函数
1
2
3
4
5
def node_a(state: State):
return {"foo": "a", "bar": ["a"]}

def node_b(state: State):
return {"foo": "b", "bar": ["b"]}
- 两个节点，分别叫 node_a 和 node_b。
- node_a 执行后：
  - foo 会被设为 "a"
  - bar 会追加 "a"
- node_b 执行后：
  - foo 会被设为 "b"
  - bar 会追加 "b"

构建工作流图

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

构建了一个线性流程：START → node_a → node_b → END。
执行顺序是固定的：先跑 node_a，再跑 node_b。

配置检查点器并编译图
1
2
checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)
- checkpointer = InMemorySaver()：创建一个内存型检查点器。
- graph = workflow.compile(checkpointer=checkpointer)：编译图时绑定检查点器，开启持久化能力。
- 关键：没有绑定 checkpointer，就无法使用 thread_id 和状态持久化。
运行图（核心：指定 thread_id）
1
2
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
result = graph.invoke({"foo": "", "bar": []}, config)
- config 里的 thread_id: "1" 是关键：
  - 它告诉 LangGraph：这次执行属于 thread_id=1 这个会话 / 存档。
  - 执行过程中，每个超步（super-step）的状态都会被保存到这个 thread_id 下。
- 初始状态：{"foo": "", "bar": []}
- 执行结果会是：
  - foo: "b"（被 node_b 覆盖）
  - bar: ["a", "b"]（两次追加的结果）

线程是 LangGraph 持久化机制的核心概念。每个线程代表一个独立的工作流执行实例，通过唯一的 thread_id 进行标识。线程内部保存了完整的执行历史，包括所有检查点，这使得我们可以实现时间旅行、状态恢复等高级功能。

这段话可以拆解为：

线程 = 独立执行实例

不同的thread_id就是不同的会话，它们的状态互不干扰。比如用户 A 和用户 B 同时使用你的对话机器人，就可以用不同的thread_id区分。
线程 = 完整执行历史

线程里保存了从开始到现在的所有检查点，相当于一份 “完整的操作日志 + 存档”。
支撑高级功能

正是因为线程里保存了所有历史，LangGraph 才能实现：
- 状态恢复（从上次中断的地方继续）
- 时间旅行（回到之前的检查点重跑）
- 多会话并行（同一个图服务多个用户）

检查点（Checkpoints）

核心概念

检查点是线程在特定时间点的状态快照。

可以把它理解成：

线程（Thread）是一个完整的 “游戏存档”
检查点就是存档里的每一个关键时间点快照
每个检查点都完整保存了当时的状态、下一步要做什么、任务信息，让你可以随时回到这里

检查点的关键属性解读

checkpoint_properties = {
    "config": "与此检查点关联的配置",
    "metadata": "元数据",
    "values": "此时的状态通道值",
    "next": "图中要执行的下一个节点元组",
    "tasks": "包含下一个任务信息的PregelTask对象元组"
}

属性	通俗解释	实际作用
`config`	关联的配置	包含 `thread_id` 等信息，用来标识这个检查点属于哪个会话
`metadata`	元数据	额外的描述信息，比如执行时间、触发节点、耗时等，用于调试和日志
`values`	状态通道值	最重要的字段，保存了当前所有状态变量的值（比如之前例子里的 `foo` 和 `bar`）
`next`	下一个节点元组	告诉 LangGraph 执行完这个检查点后，下一个要跑的节点是谁
`tasks`	下一个任务信息	包含了 `next` 节点对应的执行任务对象，用于调度和恢复执行

一句话总结：检查点 = 「当前状态 + 下一步计划 + 配置信息」，LangGraph 靠这几样东西实现中断、恢复和时间旅行。

检查点示例执行

这里用的就是我们上一个例子里的线性图：START → node_a → node_b → END

执行后会生成 4 个检查点，我们按顺序看：

空检查点（初始状态）
- 状态：空（还没输入任何数据）
- next：START 节点
- 这是线程刚创建时的初始检查点，相当于游戏的 “空存档”。
用户输入后的检查点
- 状态：{"foo": "", "bar": []}
- next：node_a
- 这是你调用 graph.invoke() 传入初始状态后生成的检查点，流程准备开始执行 node_a。
node_a 执行后的检查点
- 状态：{"foo": "a", "bar": ["a"]}
- next：node_b
- node_a 执行完成，状态被更新，流程准备进入 node_b。
node_b 执行后的检查点
- 状态：{"foo": "b", "bar": ["a", "b"]}
- next：无（已经到 END）
- 整个流程执行完成，生成最终的检查点，流程结束。

为什么一个简单的线性流程，会生成 4 个检查点？

每一次状态变更、每一次节点执行前后，LangGraph 都会生成一个检查点。
这些检查点就是 LangGraph 实现以下功能的基础：
- 中断与恢复：可以在任意一个检查点暂停，之后从这里继续执行。
- 时间旅行：可以回到任意一个检查点，修改状态后重新执行。
- 调试与复现：可以查看每一步的状态变化，定位问题出在哪个节点。

代码示例

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig
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):
    print("--- 执行 node_a ---")
    return {"foo": "a", "bar": ["a"]}

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

# 构建图
workflow = StateGraph(State)
workflow.add_node("node_a", node_a)
workflow.add_node("node_b", 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)

if __name__ == "__main__":
    # 配置 thread_id
    config: RunnableConfig = {"configurable": {"thread_id": "1"}}

    # 执行图
    print("===== 执行图 =====")
    result = graph.invoke({"foo": "", "bar": []}, config)
    print(f"最终状态: {result}")

    # 查看该线程下的所有检查点
    print("\n===== 查看所有检查点 =====")
    all_checkpoints = list(checkpointer.list(config))
    print(f"共生成 {len(all_checkpoints)} 个检查点\n")

    for i, checkpoint in enumerate(all_checkpoints):
        print(f"检查点 {i+1}:")
        print(f"  config: {checkpoint.config}")
        print(f"  next: {checkpoint.next}")
        print(f"  values: {checkpoint.values}")
        print("-" * 50)

运行上面的代码，你会看到和图里完全对应的 4 个检查点：

初始空检查点
用户输入后的检查点（next 是 node_a）
node_a 执行后的检查点（next 是 node_b）
node_b 执行后的检查点（next 为空）

你可以通过 checkpointer.list(config) 查看这些快照，也可以通过 checkpointer.get(config)回到任意一个检查点，实现 “时间旅行”。

状态管理

获取当前状态

这部分展示了 LangGraph 如何获取状态快照，分为两种场景：获取最新状态和获取特定历史检查点状态。

获取最新状态快照
1
2
config = {"configurable": {"thread_id": "1"}}
current_state = graph.get_state(config)
- 作用：获取 thread_id="1" 这个线程当前的最新状态快照。
- 关键：只需要 thread_id，LangGraph 会自动帮你找到该线程的最新检查点。
- 返回的 current_state 包含：当前的 values（状态值）、next（下一个节点）、metadata 等信息。
获取特定检查点的状态
1
2
3
4
5
6
7
specific_config = {
"configurable": {
"thread_id": "1",
"checkpoint_id": "1ef663ba-28fe-6528-8002-5a559208592"
}
}
historical_state = graph.get_state(specific_config)
- 作用：获取 thread_id="1" 线程中，指定 checkpoint_id 对应的历史状态快照。
- 关键：除了 thread_id，还需要 checkpoint_id（每个检查点的唯一 ID）。
- 这就是 LangGraph 「时间旅行」能力的基础：你可以随时回到任意一个历史检查点查看当时的状态。

获取状态历史

如何查看一个线程的所有历史检查点，完整回溯执行过程。

config = {"configurable": {"thread_id": "1"}}
history = list(graph.get_state_history(config))

# 历史记录按时间倒序排列（最新在前）
for i, snapshot in enumerate(history):
    print(f"检查点 {i}:")
    print(f"值: {snapshot.values}")
    print(f"下一个节点: {snapshot.next}")
    print(f"步骤: {snapshot.metadata.get('step', 'N/A')}")
    print("---")

graph.get_state_history(config)
- 传入 thread_id，就能拿到该线程的所有检查点历史。
- 返回的结果是一个迭代器，用 list() 可以转为列表。
排序方式：历史记录按时间倒序排列，最新的检查点在列表最前面。
每个 snapshot 的信息：
- snapshot.values：该检查点对应的状态值（比如 foo 和 bar 的值）。
- snapshot.next：该检查点执行完成后，下一个要执行的节点。
- snapshot.metadata.get('step', 'N/A')：检查点的步骤信息，用于调试。

一句话总结：这段代码可以帮你完整复盘整个流程的每一步状态变化，相当于查看一份「执行日志 + 状态快照」。

重放执行与状态更新

LangGraph 支持从特定检查点重放执行，以及直接更新状态值。状态更新采用部分更新机制，节点只更新它返回的字段，未返回的字段保持原值不变。

这部分是状态管理的核心价值，分为两个关键能力：

1. 从特定检查点重放执行（时间旅行）

你可以用 checkpoint_id 回到任意一个历史检查点，然后重新执行后续流程。
比如：你可以回到 node_a 执行前的状态，修改 foo 的值，然后重新执行 node_a 和 node_b，看看不同输入会产生什么结果。
应用场景：调试流程、复现问题、A/B 测试不同分支逻辑。

2. 直接更新状态值（部分更新机制）

LangGraph 的状态更新是部分更新：节点只需要返回它要修改的字段，未返回的字段会保持原值不变。
比如你的状态有 foo 和 bar，节点只返回 {"bar": ["new_value"]}，那么 foo 的值不会被改变，只有 bar 会被追加更新。
你也可以通过代码直接修改状态，比如在任意检查点手动修改 values，然后继续执行流程。

代码示例

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

# ------------------------------
# 1. 定义状态和节点
# ------------------------------
class State(TypedDict):
    foo: str
    bar: Annotated[list[str], add]

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

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

# ------------------------------
# 2. 构建并编译图
# ------------------------------
workflow = StateGraph(State)
workflow.add_node("node_a", node_a)
workflow.add_node("node_b", 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)

if __name__ == "__main__":
    config: RunnableConfig = {"configurable": {"thread_id": "1"}}

    # 执行一次流程，生成检查点
    print("===== 执行流程 =====")
    graph.invoke({"foo": "", "bar": []}, config)

    # ------------------------------
    # 3. 获取当前最新状态
    # ------------------------------
    print("\n===== 获取当前最新状态 =====")
    current_state = graph.get_state(config)
    print(f"当前状态值: {current_state.values}")
    print(f"下一个节点: {current_state.next}")
    print(f"检查点ID: {current_state.config['configurable']['checkpoint_id']}")

    # ------------------------------
    # 4. 获取状态历史
    # ------------------------------
    print("\n===== 获取状态历史 =====")
    history = list(graph.get_state_history(config))
    print(f"共 {len(history)} 个检查点\n")

    for i, snapshot in enumerate(history):
        print(f"检查点 {i}:")
        print(f"  值: {snapshot.values}")
        print(f"  下一个节点: {snapshot.next}")
        print(f"  检查点ID: {snapshot.config['configurable']['checkpoint_id']}")
        print("  ---")

    # ------------------------------
    # 5. 时间旅行：回到指定检查点重放
    # ------------------------------
    print("\n===== 时间旅行：回到node_a执行前的检查点 =====")
    # 取倒数第2个检查点（node_a执行前）
    replay_config = history[-2].config
    # 修改状态值
    replay_config["values"] = {"foo": "modified", "bar": ["custom"]}
    # 从该检查点继续执行
    replay_result = graph.invoke(None, replay_config)
    print(f"重放执行结果: {replay_result}")

核心知识点总结

功能	代码	作用
获取最新状态	`graph.get_state(config)`	查看当前线程的最新状态
获取特定检查点状态	`graph.get_state(specific_config)`	查看指定 `checkpoint_id` 的历史状态
获取所有状态历史	`graph.get_state_history(config)`	回溯线程的所有执行快照
时间旅行重放	`graph.invoke(None, replay_config)`	从历史检查点重新执行流程
部分状态更新	节点只返回修改的字段	未修改的字段保持原值不变

Langgraph状态管理机制

核心概念：什么是 LangGraph 的部分更新机制？

LangGraph 的状态更新采用部分更新机制，这是它状态管理的关键特性：

节点执行时，不需要返回完整的状态对象，只需要返回它想修改的字段即可。
节点未返回的字段，会保持原值不变，不会被覆盖或清空。
搭配「状态还原器（Reducer）」，还能实现追加、合并等高级更新逻辑。

这就像你修改文档，只需要提交修改的部分，而不是每次都上传整个文档，既高效又安全。

状态更新规则

节点只更新它返回的字段

节点的返回值，只会修改状态中对应的字段，不会影响其他字段。

未返回的字段保持原值不变

节点没有返回的字段，会保留执行前的值，不会被清空或修改。

这就是为什么上面的例子中，messages 字段不会被覆盖。

还原器只对明确更新的字段生效

只有节点返回的字段，才会触发该字段对应的还原器（比如 add）。

节点未返回的字段，还原器不会被调用，自然也不会改变原值。

代码示例

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

# ------------------------------
# 1. 定义状态结构
# ------------------------------
class ConversationState(TypedDict):
    messages: Annotated[list, add]  # 有还原器的字段：追加更新
    user_profile: dict              # 无还原器的字段：直接覆盖

# ------------------------------
# 2. 定义节点函数
# ------------------------------
def add_message(state: ConversationState):
    """添加一条消息，只更新 messages 字段"""
    return {"messages": ["你好，我是LangGraph助手"]}

def update_user_profile(state: ConversationState):
    """更新用户配置，只更新 user_profile 字段"""
    print("当前消息历史:", state["messages"])
    return {"user_profile": {"preferences_saved": True, "theme": "dark"}}

# ------------------------------
# 3. 构建并编译图
# ------------------------------
workflow = StateGraph(ConversationState)
workflow.add_node("add_message", add_message)
workflow.add_node("update_user_profile", update_user_profile)

workflow.add_edge(START, "add_message")
workflow.add_edge("add_message", "update_user_profile")
workflow.add_edge("update_user_profile", END)

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

# ------------------------------
# 4. 运行示例
# ------------------------------
if __name__ == "__main__":
    config: RunnableConfig = {"configurable": {"thread_id": "1"}}
    
    # 初始状态：有消息和用户配置
    initial_state = {
        "messages": ["用户：你好"],
        "user_profile": {"preferences_saved": False}
    }

    print("===== 初始状态 =====")
    print(initial_state)

    # 执行流程
    print("\n===== 执行流程 =====")
    final_state = graph.invoke(initial_state, config)

    print("\n===== 最终状态 =====")
    print(final_state)

关键代码解读

关键依赖倒入
1
2
from typing import Annotated, TypedDict
from operator import add
- Annotated：用来给状态字段附加「还原器（Reducer）」信息。
- add：列表追加操作符，作为 messages 字段的还原器，实现消息的追加更新。
定义状态结构 ConversationState
1
2
3
class ConversationState(TypedDict):
messages: Annotated[list, add] # 有还原器的字段
user_profile: dict # 无还原器的字段
这里定义了两个字段，它们的更新规则不同：

| 字段 | 类型 | 还原器 | 作用 | 更新规则 |
| :——————- | :——- | :——- | :—————- | :———————————————- |
| messages | list | add | 保存对话历史 | 节点返回的列表会追加到原列表后面 |
| user_profile | dict | 无 | 保存用户信息 | 节点返回的字典会直接覆盖原字典 |

定义节点函数 update_user_profile

def update_user_profile(state: ConversationState, config: RunnableConfig):
    # 只更新 user_profile，不返回 messages
    return {"user_profile": {"preferences_saved": True}}
    # messages 字段保持不变，仍然可以访问完整历史

关键：这个节点只返回了 user_profile 字段，没有返回 messages 字段。
执行效果：
- user_profile：会被节点返回的 {"preferences_saved": True} 直接覆盖更新。
- messages：因为节点没有返回，所以保持执行前的原值不变，对话历史不会丢失。

存储（Store） — 跨县城共享

核心概念：什么是 Store？

存储（Store）用于跨线程共享数据，支持命名空间和语义搜索功能。

简单来说，Store 是 LangGraph 提供的长期、跨会话的数据存储层：

之前讲的「检查点 / 线程状态」是临时的、单会话的；Store 则是持久的、跨会话的。
它的核心作用是：让同一个用户的不同对话（不同 thread_id），都能访问到共享的记忆 / 数据。
支持两大关键特性：命名空间隔离 和 语义搜索。

代码示例

from langgraph.store.memory import InMemoryStore
import uuid

# ------------------------------
# 1. 创建 Store 实例
# ------------------------------
store = InMemoryStore()

# ------------------------------
# 2. 定义命名空间（按用户+数据类型隔离）
# ------------------------------
user_id = "user_123"
namespace_memories = (user_id, "memories")
namespace_preferences = (user_id, "preferences")

# ------------------------------
# 3. 存储多条数据
# ------------------------------
# 存储用户记忆
memory_id_1 = str(uuid.uuid4())
memory_1 = {"content": "I like pizza", "type": "food_preference"}
store.put(namespace_memories, memory_id_1, memory_1)

memory_id_2 = str(uuid.uuid4())
memory_2 = {"content": "I hate spicy food", "type": "food_preference"}
store.put(namespace_memories, memory_id_2, memory_2)

# 存储用户偏好
pref_id_1 = str(uuid.uuid4())
pref_1 = {"theme": "dark", "notifications": True}
store.put(namespace_preferences, pref_id_1, pref_1)

# ------------------------------
# 4. 搜索数据
# ------------------------------
print("===== 搜索用户的所有记忆 =====")
memories = store.search(namespace_memories)
for i, item in enumerate(memories):
    print(f"记忆 {i+1}: {item.dict()}")

print("\n===== 搜索用户的偏好设置 =====")
preferences = store.search(namespace_preferences)
print(f"偏好设置: {preferences[0].dict()}")

# ------------------------------
# 5. 获取单条数据（通过ID）
# ------------------------------
print("\n===== 获取指定ID的记忆 =====")
specific_memory = store.get(namespace_memories, memory_id_1)
print(f"指定记忆: {specific_memory.dict()}")

关键代码详解：

关键导入依赖
1
2
from langgraph.store.memory import InMemoryStore
import uuid
- InMemoryStore：内存型 Store 实现，适合开发 / 测试使用（生产环境可以用 Redis、PostgreSQL 等持久化实现）。
- uuid：用来生成唯一的 memory_id，方便后续检索。
创建存储实例
1
in_memory_store = InMemoryStore()
创建一个内存存储实例，所有数据都会存在这里（程序重启后会丢失）。
定义命名空间（Namespace）
1
2
user_id = "1"
namespace_for_memory = (user_id, "memories")
- 命名空间（Namespace）：Store 用来隔离数据的关键机制，格式是一个元组。
- 这里 (user_id, "memories") 的含义：
  - 第一层 user_id：按用户隔离，不同用户的数据互不干扰。
  - 第二层 "memories"：按数据类型隔离，比如你还可以定义 (user_id, "preferences")、(user_id, "orders") 等。
- 有了命名空间，你可以轻松实现：「用户 1 的记忆」和「用户 2 的记忆」完全隔离，互不干扰。
存储数据（put）
1
2
3
memory_id = str(uuid.uuid4())
memory = {"food_preference": "I like pizza"}
in_memory_store.put(namespace_for_memory, memory_id, memory)
- memory_id：这条数据的唯一 ID，用 UUID 生成保证不重复。
- memory：要存储的数据，支持字典、JSON 等结构化数据。
- store.put(namespace, id, data)：把数据存入指定的命名空间下。
- 执行后，这条用户 1 的食物偏好数据，就被永久保存在 Store 里了。
搜索数据（search）
1
2
memories = in_memory_store.search(namespace_for_memory)
latest_memory = memories[-1].dict()
- store.search(namespace)：从指定命名空间中搜索所有数据，默认按时间倒序返回。
- memories[-1]：取最新的一条数据。
- .dict()：把返回的结果对象转为字典，方便读取。
- 执行后，你就能拿到用户 1 之前存储的食物偏好数据，即使是在不同的 thread_id 里。

语义搜索配置

存储支持基于向量嵌入的语义搜索，可以基于含义而非精确匹配来查找数据。

这是 Store 的高级能力，也是 LangGraph 实现「长期记忆」的核心：

你可以为存储的数据配置向量嵌入（Embedding），让 Store 支持语义搜索。
比如用户之前存了「我喜欢吃披萨」，当你搜索「用户喜欢什么食物」时，Store 可以通过向量相似度找到这条记忆，而不需要精确匹配关键词。
这对于构建带长期记忆的对话机器人、个人助手非常有用。

核心价值

跨会话记忆：同一个用户，多次对话都能访问到自己的历史偏好和记忆。
数据隔离：通过命名空间，轻松实现用户、数据类型的隔离，避免混乱。
语义搜索：结合向量嵌入，实现自然语言的记忆检索，是智能助手的核心能力。
灵活扩展：从内存到 Redis、PostgreSQL 都支持，适配开发到生产的全流程。

`Checkpointer` 和 `Store`对比

特性	Checkpointer（检查点器）	Store（存储）
数据范围	线程内状态持久化	跨线程数据共享
通俗理解	只管「单会话内部」的状态，比如某一次对话的流程快照	负责「跨会话 / 跨线程」的全局数据，比如用户所有对话都能访问的记忆
—-	—-	—-
数据组织	按执行步骤组织的检查点序列	按命名空间组织的键值对
通俗理解	像电影的分镜存档，按执行顺序排成一条时间线	像数据库表，按用户 / 类型分命名空间，存独立的键值对
—-	—-	—-
使用场景	会话记忆、执行历史、状态恢复	用户偏好、知识库、共享数据
通俗理解	解决「这次对话中断了怎么办」「回到上一步重新来」的问题	解决「用户上次对话说过的信息，这次对话怎么记住」的问题
—-	—-	—-
数据模型	完整的状态快照	独立的键值对条目
通俗理解	每次都存整个状态对象，包含所有字段	存的是独立的、结构化的数据条目，和会话状态无关
—-	—-	—-
访问方式	按线程 ID 和时间顺序访问	按命名空间和键访问
通俗理解	必须指定 `thread_id`，再按时间顺序取检查点	按 `(user_id, "memories")` 这种命名空间 + key 直接存取
—-	—-	—-
典型用例	对话历史、工作流状态	用户配置、产品目录
通俗理解	比如：对话过程中人工审批、错误后恢复流程、回溯对话步骤	比如：记住用户的食物偏好、保存知识库文档、存储系统配置

核心区别一句话总结

Checkpointer：是「单会话的时间线存档系统」，只服务于当前 thread_id，记录流程每一步的状态快照，用来做中断恢复、时间旅行。
Store：是「跨会话的全局数据仓库」，和 thread_id 无关，用来存用户长期记忆、共享配置等，让不同对话都能访问。

实际应用场景举例

1. Checkpointer 典型场景

对话中断恢复：用户聊到一半退出，下次进来可以从上次的状态继续。
人工审批流程：AI 生成内容后暂停，等人工确认后再继续执行。
流程调试：回到任意一个检查点，修改状态后重新执行，排查问题。

2. Store 典型场景

长期用户记忆：用户第一次对话说「我喜欢吃披萨」，第二次对话时机器人能记住这个偏好。
共享知识库：把产品信息、文档存到 Store 里，所有对话都能调用。
用户配置管理：保存用户的主题设置、通知偏好，跨设备 / 跨会话都生效。

两者配合使用的最佳实践

在实际项目中，Checkpointer 和 Store 通常是搭配使用的：

用 Checkpointer 保存当前对话的流程状态，实现中断恢复、人工介入。
用 Store 保存用户的长期记忆和配置，实现跨会话的个性化体验。
对话节点执行时，从 Store 读取用户偏好，结合当前 Checkpointer 里的会话状态，生成个性化回复。

代码示例

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langchain_core.runnables import RunnableConfig
from typing_extensions import TypedDict
from typing import Annotated
from operator import add
import uuid

# 定义状态
class State(TypedDict):
    messages: Annotated[list, add]

# 1. 初始化 Checkpointer 和 Store
checkpointer = InMemorySaver()
store = InMemoryStore()

# 2. 定义节点：从 Store 读取用户记忆并生成回复
def chat_node(state: State, config: RunnableConfig):
    # 获取用户ID和线程ID
    user_id = config["configurable"]["user_id"]
    thread_id = config["configurable"]["thread_id"]

    # 从 Store 读取用户记忆
    namespace = (user_id, "memories")
    memories = store.search(namespace)
    user_preference = memories[-1].dict()["content"] if memories else "无"

    # 生成回复
    reply = f"你之前说过：{user_preference}。有什么可以帮你的吗？"
    return {"messages": [reply]}

# 3. 构建图
workflow = StateGraph(State)
workflow.add_node("chat", chat_node)
workflow.add_edge(START, "chat")
workflow.add_edge("chat", END)

# 编译图，绑定 Checkpointer
graph = workflow.compile(checkpointer=checkpointer, store=store)

if __name__ == "__main__":
    user_id = "user_123"
    # 先把用户偏好存入 Store（可以在之前的对话中完成）
    store.put((user_id, "memories"), str(uuid.uuid4()), {"content": "我喜欢吃披萨"})

    # 对话1：thread_id=1
    print("===== 对话1 =====")
    config1 = {"configurable": {"thread_id": "1", "user_id": user_id}}
    result1 = graph.invoke({"messages": ["你好"]}, config1)
    print(result1["messages"][-1])

    # 对话2：thread_id=2（同一个用户，不同会话）
    print("\n===== 对话2 =====")
    config2 = {"configurable": {"thread_id": "2", "user_id": user_id}}
    result2 = graph.invoke({"messages": "今天吃什么好？"}, config2)
    print(result2["messages"][-1])

Checkpointer 解决的是「会话内的流程问题」：状态恢复、中断、回溯。
Store 解决的是「跨会话的共享数据问题」：用户记忆、全局配置。
两者结合，才能构建出既稳定可靠、又具备长期记忆能力的企业级 LangGraph 应用。

Checkpointer的存储选择

三种存储对比

1. 内存检查点器 `InMemorySaver`

from langgraph.checkpoint.memory import InMemorySaver

# 适用于开发和测试环境
checkpointer = InMemorySaver()

核心特点：数据存在应用进程的内存中。

✅ 优点：

配置零依赖，直接就能用
读写速度极快，没有额外的 IO 开销

❌ 缺点：

进程重启、服务崩溃后，所有检查点数据都会丢失
不支持多实例部署，只能单进程使用

适用场景：仅用于开发、单元测试、本地调试，不适合生产环境。

2.SQLite 检查点器 `SqliteSaver`

from langgraph.checkpoint.sqlite import SqliteSaver

# 使用SQLite数据库
checkpointer = SqliteSaver("checkpoints.db")

核心特点：数据持久化到本地 SQLite 文件中。

✅ 优点：

数据会写入磁盘，进程重启后不会丢失
无需额外安装数据库服务，一个文件搞定
轻量级，部署成本极低

❌ 缺点：

只能单机部署，无法多实例共享数据
并发写入性能有限，高并发场景下容易成为瓶颈

适用场景：中小型单机应用、个人项目、轻量服务，比如本地运行的对话机器人、单实例部署的服务。

3.Redis 检查点器 `RedisSaver`

from langgraph.checkpoint.redis import RedisSaver

# 使用Redis数据库
checkpointer = RedisSaver("redis://localhost:6379")

核心特点：数据存储在 Redis 中，支持网络访问。

✅ 优点：

高性能读写，支持高并发场景
支持分布式部署，多个服务实例可以共享同一个检查点数据
数据持久化（Redis 开启持久化后），服务重启不丢失

❌ 缺点：

需要额外部署和维护 Redis 服务
有一定的运维成本和资源开销

适用场景：高并发生产环境、分布式服务、多实例部署，比如企业级对话平台、SaaS 服务。

官方建议

开发测试：用 InMemorySaver，简单快速，不依赖外部服务。
小型应用：用 SqliteSaver，兼顾持久化和低部署成本。
生产环境：推荐 RedisSaver 或 PostgreSQL 等成熟数据库，保证性能和稳定性。
云部署：优先考虑云数据库服务（如云 Redis、云 PostgreSQL），减少运维成本。

场景	推荐实现	关键考量
本地调试 / 单元测试	`InMemorySaver`	开发效率优先，不需要持久化
单机轻量服务 / 个人项目	`SqliteSaver`	持久化 + 零额外依赖
生产级 SaaS / 多实例服务	`RedisSaver` / `PostgreSQLSaver`	高并发、分布式、数据可靠性

Langgraph持久化基础概念

什么是持久化

核心价值

线程（Threads）

核心定义

代码示例

检查点（Checkpoints）

核心概念

检查点的关键属性解读

检查点示例执行

代码示例

状态管理

获取当前状态

获取状态历史

重放执行与状态更新

1. 从特定检查点重放执行（时间旅行）

2. 直接更新状态值（部分更新机制）

代码示例

核心知识点总结

Langgraph状态管理机制

核心概念：什么是 LangGraph 的部分更新机制？

状态更新规则

代码示例

存储（Store） — 跨县城共享

核心概念：什么是 Store？

代码示例

语义搜索配置

核心价值

Checkpointer 和 Store对比

实际应用场景举例

1. Checkpointer 典型场景

2. Store 典型场景

两者配合使用的最佳实践

代码示例

Checkpointer的存储选择

三种存储对比

1. 内存检查点器 InMemorySaver

2.SQLite 检查点器 SqliteSaver

3.Redis 检查点器 RedisSaver

官方建议

`Checkpointer` 和 `Store`对比

1. 内存检查点器 `InMemorySaver`

2.SQLite 检查点器 `SqliteSaver`

3.Redis 检查点器 `RedisSaver`