持久化机制与记忆

可恢复执行

把一次任务执行过程中的关键进度、状态、结果保存到可靠存储中，使任务可以在中断、失败、等待外部输入后继续执行，它关注的部署数据长期保存本身，而是解决 “执行过程能否恢复” 的问题，即系统能否记住任务执行到了哪里、当前状态是什么、接下来应该继续执行什么

普通执行：
开始 -> A -> B -> C
如果执行到 B 后程序挂了，重启后可能只能从 A 重新开始。

可恢复执行：
开始 -> A   保存检查点
     -> B   保存检查点
     -> C
如果执行到 B 后挂了，恢复时可以从已保存的位置继续。

启用可恢复执行

基于内存

在 Jupyter 场景下，只要 Jupyter 内核没有被重启，并且 InMemorySaver 实例没有被重新创建，内存中的检查点数据就会继续存在
如果重新执行完整代码，导致 checkpointer = InMemorySaver 被重新执行，那么之前保存在内存中的检查点也会被清空

python

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import MessagesState
from langgraph.checkpoint.memory import InMemorySaver
from langchain_deepseek import ChatDeepSeek
from langchain.messages import HumanMessage

from dotenv import load_dotenv
load_dotenv(override=True)

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)

class OverAllState(MessagesState):
    output: str

def llm_node(state: OverAllState) -> OverAllState:
    messages = state["messages"]
    res = model.invoke(messages)
    return {
        "messages": [res]
    }

def output_node(state: OverAllState) -> OverAllState:
    return {
        "output": state["messages"][-1].content
    }

builder = StateGraph(state_schema=OverAllState)
builder.add_node("llm_node", llm_node)
builder.add_node("output_node", output_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", "output_node")
builder.add_edge("output_node", END)

# 定义并在编译时传递 Checkpointer
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 定义配置对象
config = {"configurable": {"thread_id": "chapter_6_6-2-2"}}
# 调用时传递
graph.invoke({"messages": [HumanMessage("你好，我是老王")]}, config=config)
graph.invoke({"messages": [HumanMessage("从现在开始，你是小王")]}, config=config)
res = graph.invoke({"messages": [HumanMessage("我是谁？你是谁？")]}, config=config)
print(res["output"])

print('=' * 30, '-> 完整消息列表 <-', '=' * 30)
for msg in res["messages"]:
    msg.pretty_print()

基于 pgsql

和 InMemorySaver 不同，PostgresSaver 会将检查点保存到 PostgreSQL 数据库中，只要数据库中的记录没有被删除，即使 Python 程序结束、连接对象重建，历史检查点也仍然存在
setup 方法会创建 checkpoint 相关的表，包括 checkpoints、checkpoint_writes、checkpoint_blobs、checkpoint_migrations

python

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import MessagesState
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_deepseek import ChatDeepSeek
from langchain.messages import HumanMessage

from dotenv import load_dotenv
load_dotenv(override=True)

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)

class OverAllState(MessagesState):
    output: str

def llm_node(state: OverAllState) -> OverAllState:
    messages = state["messages"]
    res = model.invoke(messages)
    return {
        "messages": [res]
    }

def output_node(state: OverAllState) -> OverAllState:
    return {
        "output": state["messages"][-1].content
    }

builder = StateGraph(state_schema=OverAllState)
builder.add_node("llm_node", llm_node)
builder.add_node("output_node", output_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", "output_node")
builder.add_edge("output_node", END)

# 定义并在编译时传递 Checkpointer
DB_URL = "postgresql://langgraph_user:123456@localhost:5432/langgraph_db?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    # 示例中为了方便演示直接调用 setup()
    # 实际项目中通常建议把数据库初始化/迁移作为独立步骤处理
    checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)

    # 定义配置对象
    config = {"configurable": {"thread_id": "chapter_6_6.2.4"}}
    # 调用时传递
    graph.invoke({"messages": [HumanMessage("你好，我是老王")]}, config=config)
    graph.invoke({"messages": [HumanMessage("从现在开始，你是小王")]}, config=config)
    res = graph.invoke({"messages": [HumanMessage("我是谁？你是谁？")]}, config=config)
    print(res["output"])

    print('=' * 30, '-> 完整消息列表 <-', '=' * 30)
    for msg in res["messages"]:
        msg.pretty_print()

checkpoint 相关表

基本介绍

LangGraph 通过 thread_id + checkpoint_ns + checkpoint_id 联合定位和恢复状态

表名	用途
checkpoints	存储完整的检查点快照（状态、元数据等）
checkpoint_writes	存储每个检查点对应的通道写入记录
checkpoint_blobs	存储大型二进制数据（如图片、文件等，如果状态中包含的话）
checkpoint_migrations	记录数据库迁移版本，LangGraph 内部使用，用于管理表结构变更

checkpoints

主键为 (thread_id, checkpoint_ns, checkpoint_id) 联合主键
另有 thread_id 单列索引，加速按线程筛选查询

字段	类型	说明
thread_id	text	线程 ID，对应 config 中的 thread_id，用于区分不同对话
checkpoint_ns	text	检查点命名空间，默认空字符串，用于子图等场景隔离不同层级的检查点
checkpoint_id	text	检查点唯一 ID，由 LangGraph 内部生成
parent_checkpoint_id	text	父检查点 ID，构建检查点链，用于 Time Travel 回退
type	text	检查点类型
checkpoint	jsonb	检查点核心数据（状态快照），以 JSON 格式存储
metadata	jsonb	检查点元数据（创建时间、来源、步骤序号等），预设为 '{}'::jsonb

writes

主键为 (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) 联合主键，确保同一任务下的写入顺序唯一

字段	类型	说明
thread_id	text	线程 ID
checkpoint_ns	text	命名空间，与 checkpoints 表对应
checkpoint_id	text	所属检查点 ID
task_id	text	产生写入的节点任务 ID
idx	integer	写入顺序索引
channel	text	写入的目标通道名（如 messages、__pregel_tasks）
type	text	写入数据类型
blob	bytea	写入的二进制数据
task_path	text	任务路径，用于追踪节点在子图中的层级关系

blobs

主键为 (thread_id, checkpoint_ns, channel, version) 联合主键
与 checkpoint_writes 不同，checkpoint_blobs 以 (channel, version) 维度管理数据版本，不直接关联某个具体的 checkpoint_id，允许多个检查点共享同一通道的大型二进制数据

字段	类型	说明
thread_id	text	线程 ID
checkpoint_ns	text	命名空间
channel	text	通道名
version	text	数据版本标识，同一通道可以有多个版本
type	text	数据类型
blob	bytea	二进制数据本体

migrations

单列主键，仅一条记录，setup 执行时自动检查并更新
该表由 LangGraph 内部管理，用户无需手动操作，未来 LangGraph 版本升级时，若表结构有变更，setup 会根据此版本号自动执行对应的迁移 SQL

字段	类型	说明
v	integer	当前数据库迁移版本号

持久化模式

基本介绍

启用 Checkpointer 后，LangGraph 会在执行过程中保存检查点
检查点写入越及时，系统在异常中断后可恢复的状态越完整，容灾能力越强，但更及时的持久化通常也意味着更多中间写入，或额外的阻塞式等待，带来额外的性能开销和响应延迟（从调用者的角度考虑）

三大模式

（1）exit：退出模式，只在计算图正常结束、异常退出、被中断（如 Human-In-The-Loop 中断）时保存检查点，不能处理中途进程崩溃的场景，这种模式响应性能开销最小，但容灾能力最弱
（2）async：异步模式，默认模式，检查点在后台异步写入，它会在每个超步结束后写入完整检查点（主检查点），并在图中任务执行完毕后记录中间结果，和 exit 模式相比，增加了性能开销，但写入操作发生在后台，不会引入明显的响应延迟，同时提升了容灾能力
（3）sync：同步模式，和异步模式唯一的区别在于，LangGraph 会在进入下一个超步之前等待当前主检查点的写入任务完成，它的容灾能力最强，但在 async 模式的基础上增加了响应延迟

模式	写入时机	性能开销	响应延迟	容灾能力
exit	运行退出时写入	低	最低	最弱
async	每个超步末尾写入主检查点，任务完成后写入中间结果，后台异步写入	高	较高	较强
sync	和 async 区别在于，进入下一个超步之前等待当前主检查点的写入任务完成	高	最高	最强

代码示例

持久化模式影响的是检查点写入时机和故障恢复能力，通常不会影响正常情况下的业务输出，也就是说，同一张图在正常执行完成时，使用 exit、async 或 sync，最终返回结果通常是一样的，真正的区别主要体现在异常中断或进程崩溃后恢复执行时
durability 控制的是检查点写入策略，不改变图本身的执行逻辑

python

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import MessagesState
from langgraph.checkpoint.memory import InMemorySaver
from langchain_deepseek import ChatDeepSeek
from langchain.messages import HumanMessage

from dotenv import load_dotenv
load_dotenv(override=True)

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)

class OverAllState(MessagesState):
    output: str

def llm_node(state: OverAllState) -> OverAllState:
    messages = state["messages"]
    res = model.invoke(messages)
    return {
        "messages": [res]
    }

def output_node(state: OverAllState) -> OverAllState:
    return {
        "output": state["messages"][-1].content
    }

builder = StateGraph(state_schema=OverAllState)
builder.add_node("llm_node", llm_node)
builder.add_node("output_node", output_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", "output_node")
builder.add_edge("output_node", END)

# 定义并在编译时传递 Checkpointer
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 定义配置对象
config = {"configurable": {"thread_id": "chapter_tmp"}}
# 调用时传递
res=graph.invoke(
    {"messages": [HumanMessage("你好")]},
    config=config,
    durability="async" # sync / exit
)
print(res["output"])

print('=' * 30, '-> 完整消息列表 <-', '=' * 30)
for msg in res["messages"]:
    msg.pretty_print()

from IPython.display import display
display(graph)

查看历史检查点

基本介绍

LangGraph 在配置检查点存储器后，会把同一个 thread_id 下的执行过程保存为一组检查点，通过这些检查点，我们可以查看图运行的中间状态，也可以为后续的检查点回溯（重放、分叉）和失败恢复做准备
本节主要介绍两个常用方法：
（1）graph.get_state_history(config)：查看指定会话的完整历史检查点
（2）graph.get_state(config)：查看指定会话的最新检查点，或者查看某个指定 checkpoint_id 对应的检查点

完整历史检查点列表

python

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek

from loguru import logger
from dotenv import load_dotenv
load_dotenv(override=True)

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)

class OverAllState(TypedDict):
    topic: str
    poem: str
    joke: str
    final_output: str

class InputState(TypedDict):
    topic: str

class OutputState(TypedDict):
    final_output: str

def node_poem(state: InputState) -> OverAllState:
    logger.info(f"node_poem 已执行")
    topic = state["topic"]
    poem = model.invoke([HumanMessage(f"写一首关于 {topic} 的七言绝句")]).content

    return {
        "poem": poem
    }

def node_joke(state: InputState) -> OverAllState:
    logger.info(f"node_joke 已执行")
    topic = state["topic"]
    joke = model.invoke([HumanMessage(f"写一个关于 {topic} 的笑话")]).content

    return {
        "joke": joke
    }

def node_output(state: OverAllState) -> OutputState:
    logger.info("node_output 已执行")
    topic = state["topic"]
    poem = state["poem"]
    joke = state["joke"]
    final_output = f"关于 {topic} 的七言绝句：\n{poem}\n笑话：\n{joke}"

    return {
        "final_output": final_output
    }

builder = StateGraph(state_schema=OverAllState, input_schema=InputState, output_schema=OutputState)
builder.add_node("node_poem", node_poem)
builder.add_node("node_joke", node_joke)
builder.add_node("node_output", node_output)
builder.add_edge(START, "node_poem")
builder.add_edge(START, "node_joke")
builder.add_edge("node_poem", "node_output")
builder.add_edge("node_joke", "node_output")
builder.add_edge("node_output", END)

checkpointer = InMemorySaver()
config = {"configurable": {"thread_id": "123"}}
graph = builder.compile(checkpointer=checkpointer)
res = graph.invoke({"topic": "猫咪"}, config=config)

from IPython.display import display
display(graph)

print('=' * 30, '-> 运行结果 <-', '=' * 30)
print("res: {}", res)

print('=' * 30, '-> 历史检查点列表 <-', '=' * 30)
history_checkpoints = list(graph.get_state_history(config=config))
print(history_checkpoints)

根据 ID 查询指定检查点

python

# 获取检查点列表
print('=' * 30, '-> 历史检查点列表 <-', '=' * 30)
history_checkpoints = list(graph.get_state_history(config=config))
print(history_checkpoints)

# 获取指定某一个节点的位置
before_router_checkpoint = next(h for h in history_checkpoints if h.next == ('router_node',))

# 查询指定检查点
print('=' * 30, '-> 根据ID查看特定检查点 <-', '=' * 30)
checkpointe_id = history_checkpoints[3].config["configurable"]["checkpoint_id"]
print(f"checkpointe_id: {checkpointe_id}")

target_config = {
    "configurable": {
        "thread_id": config["configurable"]["thread_id"],
        "checkpoint_id": checkpointe_id
    }
}

specific_history_checkpoint = graph.get_state(config=target_config)
print(specific_history_checkpoint)

StateSnapshot

get_state_history 返回的是一个历史检查点迭代器，将其转换为列表后，可以看到一组 StateSnapshot 对象

StateSnapshot 是检查点的开发者视图，字段含义如下：

values：当前检查点的状态值
next：从该检查点继续执行时，下一步将要执行的节点，即归属于下一个超步的节点
config：当前检查点配置。常见结构如下

json

{
  "configurable": {
    "thread_id": "...",
    "checkpoint_ns": "...",
    "checkpoint_id": "..."
  }
}

其中

configurable：用于记录检查点的可配置信息，可能包含用户自定义字段，它一定包含以下三个字段：
- thread_id：会话唯一标识
- checkpoint_ns：检查点命名空间。根图的命名空间通常为空字符串 ""；子图会使用非空命名空间。
- checkpoint_id：检查点唯一标识
这三个字段可以唯一标识一条检查点记录，在大多数检查点存储器实现中，写入检查点后端时都会将它们作为检查点的唯一键
metadata：检查点元数据，本节只需要关注 step 字段，后者是当前检查点对应的超步编号
created_at：检查点创建时间
parent_config：父检查点、即上一个检查点的配置
tasks：当前检查点关联的待执行任务信息，元素类型通常是 PregelTask。
tasks 通常和 next 对应，表示从当前检查点继续执行时，下一步将要运行的任务。
需要注意的是，tasks 中还可能包含这些任务已经成功（result）或失败（error）的任务记录，这是在下一个超步执行过程中记录的中间结果。在失败恢复场景中，可以避免重复执行已成功节点。
interrupts：当前图的中断信息，中断机制详见下文

Time Travel

基本介绍

所谓检查点回溯，是指基于某个历史检查点，重新执行后续流程，或者在该检查点基础上修改状态并创建新的执行分支
检查点回溯有两种形式，根据是否更改历史状态区分
（1）Replay：检查点重放，回到某个历史检查点，沿着原先的执行路径重新执行后续节点
（2）Fork：检查点分叉，回到某个历史检查点，修改状态，从该位置创建一条新的执行分支
共同点：检查点之前的节点不会重新执行，检查点之后的节点会重新执行
区别：Replay 不修改历史状态，Fork 会基于历史检查点应用新的状态更新，并创建新的检查点分支

Replay

失败恢复通常基于最新检查点继续运行，配置中只包含 thread_id，不包含 checkpoint_id
Replay 则是显式传入某个历史检查点的配置，配置中包含 checkpoint_id，LangGraph 会据此从该历史检查点开始重放后续步骤
首先通过查看历史检查点列表找到需要恢复内容的相关参数，再执行代码进行恢复

python

with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    #5. 第一次使用PostgresSaver作为检查点 需要调用方法 setup()
    #checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)

    config = {
        "configurable":{
            "thread_id":"chapter03-08"
        }
    }
    # 获取到检查点历史
    history_checkpoints = list(graph.get_state_history(config=config))

    new_checkpoint = None
    #next = ('node_poem', 'node_joke')
    for checkpoint in history_checkpoints:
        if checkpoint.next == ('node_poem', 'node_joke'):
            new_checkpoint = checkpoint
            break

    # 如果想要实现replay的效果 状态填写None  config填写为之前某一个检查点的config
    res = graph.invoke(None,config=new_checkpoint.config)
    print(res)

Fork

Fork 依赖状态图的 update_state 方法，该方法可以理解为不执行指定节点的节点函数本身，而是把传入的 values 当作该节点已经产生的输出，执行该节点对应的 writers，应用状态写入和可能存在的路由写入，然后生成一个新的检查点
（1）config：历史检查点的配置信息，通常包含 thread_id、checkpoint_ns 和 checkpoint_id，LangGraph 会据此检索历史检查点
（2）values：要应用到该检查点上的状态更新
（3）as_node：指定这次状态更新应当被视为哪个节点产生的输出，LangGraph 会按照该节点的写入逻辑，将 values 写入状态通道，并据此决定后续从哪些节点继续执行，基于分叉检查点运行时，并不是重新执行 as_node 本身，而是从它的后继节点继续推进，如 START 所属超步编号为 0，则恢复时从编号为 1 的超步开始运行
Fork 有两种实现方式（1）只改状态（2）改状态且跳过某些节点，如果想跳过某个节点，可以直接指定该节点为 as_node 并通过 values 指定指定该节点的输出
注意：update_state 返回的是新检查点的配置，不是新状态本身，它不会修改原来的历史检查点，而是基于某个历史检查点创建一个新的检查点，这个新检查点相当于一条新的执行分支

python

before_router_checkpoint = next(h for h in history_checkpoints if h.next == ('router_node',))

# 方式一：只改状态
change_input = graph.update_state(
    config=before_router_checkpoint.config,
    values={"user_input":"帮我写一个荷花的笑话"},
    as_node=START
)
print(change_input)
graph.get_state(change_input)
res = graph.invoke(None,config=change_input)
print(res)

# 方式二：改状态且跳过某些节点
# update_state调用的时候,基于as_node实现 先执行路由逻辑在完成values的赋值 => 没法正确实现路由的
skip_router_config = graph.update_state(
    config=before_router_checkpoint.config,
    values={"topic":"狸花猫","mode":"笑话"},
    as_node="router_node"
)

print(skip_router_config)

graph.get_state(skip_router_config)
graph.invoke(None,config=skip_router_config)

长期记忆

数据准备

长期记忆存储器在编译图时通过 store 参数传递，图节点中可以通过 Runtime 对象访问
（1）namespace 使用元组是硬性约束，天然表达层级结构，本例使用两层命名空间 ("users", "Alice")，清晰区分了领域（users）和实体（用户名），而具体存储什么类型的数据则由 key 参数表达（如 "preferences"）
（2）(*USERS_NS, "Alice") 的写法表示对已有元组进行解包，再拼接新的命名空间片段
（3）PostgresStore 将长期记忆持久化到 PostgreSQL 中，程序重启后数据不会丢失，适合生产环境，本节使用和 6.2.4 节、6.5.4.1 节相同的数据库实例
（4）如果希望支持语义检索，需要在 Store 中配置索引和 embedding 函数，未配置索引时，search 只能按照命名空间和过滤条件检索

python

from typing import Final, Tuple
from langgraph.store.postgres import PostgresStore

DB_URL = "postgresql://langgraph_user:123456@localhost:5432/langgraph_db?sslmode=disable"
with PostgresStore.from_conn_string(DB_URL) as store:
    # setup() 创建长期记忆所需的表结构，幂等操作
    store.setup()

    # 命名空间 - 层级结构：("users", "用户名")
    # Final 的作用是告诉静态类型检查器，该变量不应该被重新赋值，但 python 运行时不会阻止重新赋值
    USERS_NS: Final[Tuple[str]] = ("users", )
    PREFERENCES_KEY: Final[str] = "preferences"

    # 三层: (领域, 用户实体)
    # 每个 key 存储该用户的不同类型数据
    namespace1 = (*USERS_NS, "Alice")
    value1 = {
        "course": "计算机组成原理",
        "sports": "跑步",
        "food": "紫光园奶皮子酸奶"
    }

    namespace2 = (*USERS_NS, "Bob")
    value2 = {
        "course": "数字电路与模拟电路",
        "sports": "跑步",
        "food": "奶皮子糖葫芦"
    }

    namespace3 = (*USERS_NS, "Black")
    value3 = {
        "course": "数字电路与模拟电路",
        "sports": "羽毛球",
        "food": "紫光园奶皮子酸奶"
    }

    store.put(namespace1, PREFERENCES_KEY, value1)
    store.put(namespace2, PREFERENCES_KEY, value2)
    store.put(namespace3, PREFERENCES_KEY, value3)

    for item in store.search(USERS_NS):
        print(item)

记忆访问

长期记忆存储器可以通过节点函数中的 runtime.store 访问
第一次调用从长期记忆中读取数据，并写入短期记忆；后续同一会话线程内优先复用短期记忆，避免重复查询长期记忆

python

from typing import Literal
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore
from langgraph.graph.message import MessagesState
from langgraph.runtime import Runtime
from langchain.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage
from langchain_deepseek import ChatDeepSeek

from loguru import logger
from dotenv import load_dotenv
load_dotenv(override=True)

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)

class OverAllState(MessagesState):
    username: str
    user_input: str
    output: str
    preferences: dict[str, str] # 用户偏好

def check_preferences_node(state: OverAllState, runtime: Runtime) -> OverAllState:
    # 从短期记忆中获取用户名
    username = state["username"]
    # 获取长期记忆存储器 store
    store = runtime.store

    # 根据用户信息拼接命名空间，查询用户偏好
    namespace = (*USERS_NS, username)
    key = PREFERENCES_KEY
    item = store.get(namespace, key)

    if not item:
        logger.warning("长期记忆中没有 {} 的偏好数据", username)
        return {}

    logger.info("长期记忆中保存的用户偏好: {}", item.value)
    return {
        "preferences": item.value
    }

# 路由函数，如果状态中没有用户偏好，则从长期记忆中查询，否则直接执行模型节点
def router(state: OverAllState) -> Literal["check_preferences_node", "llm_node"]:
    if not state.get("preferences"):
        logger.info("需要从长期记忆中查询用户偏好")
        return "check_preferences_node"
    logger.info("用户偏好已存在，不必查询")
    return "llm_node"

def llm_node(state: OverAllState) -> OverAllState:
    # 长期记忆中未必有用户偏好
    preferences = state.get("preferences", {})
    user_input = state["user_input"]
    human_prompt = (f"这是用户的偏好: \n{preferences}\n"
              f"这是用户的需求: \n{user_input}\n")
    system_prompt = "请根据用户偏好解决用户需求"

    # 获取历史消息列表，如果历史为空则初始化系统提示词
    messages: list[SystemMessage | HumanMessage | AIMessage | ToolMessage] = [
        SystemMessage(content=system_prompt)
    ] if not state.get("messages", []) else state["messages"]

    model_response = model.invoke(messages + [HumanMessage(content=human_prompt)])
    output = model_response.content

    return {
        "messages": messages + [HumanMessage(content=human_prompt), model_response],
        "output": output
    }

builder = StateGraph(state_schema=OverAllState)
builder.add_node("check_preferences_node", check_preferences_node)
builder.add_node("llm_node", llm_node)
builder.add_conditional_edges(START, router, path_map=["check_preferences_node", "llm_node"])
builder.add_edge("check_preferences_node", "llm_node")
builder.add_edge("llm_node", END)

DB_URL = "postgresql://langgraph_user:123456@localhost:5432/langgraph_db?sslmode=disable"
# 编译时传递短期记忆和长期记忆存储器（均使用 PostgreSQL）
with PostgresSaver.from_conn_string(DB_URL) as checkpointer, \
     PostgresStore.from_conn_string(DB_URL) as store:
    checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer, store=store)
    from IPython.display import display
    display(graph)

    config = {"configurable": {"thread_id": "345"}}
    res = graph.invoke({"username": "Alice", "user_input": "我有点无聊，和我聊聊天吧"}, config=config)

    print('=' * 30, '->  <-', '=' * 30)
    print(res)

    # 第二次调用：复用同一个 thread_id，观察短期记忆缓存效果
    new_res = graph.invoke({"user_input": "推荐一下酸奶"}, config=config)
    new_messages = new_res.pop("messages")
    print('=' * 30, '->  <-', '=' * 30)
    print(new_res)

运行时上下文

基本介绍

在某些场景下，我们希望在调用图时传入一些仅对当次调用有效的信息，比如当前登录用户、请求来源、调用方标识等，这些信息不适合放入图状态（图状态会被持久化并在同一会话中跨调用共享），而应该通过运行时上下文（Runtime Context）传递
运行时上下文的特点是：仅对本次调用生效，不会被持久化，也不会在同一会话的下一次调用中自动恢复
运行时上下文的使用方式
（1）初始化状态图时，使用 context_schema 定义上下文类型
（2）调用图时通过 context 参数传入上下文对象
（3）节点或路由函数中通过 runtime.context 访问上下文

代码示例

python

from dataclasses import dataclass
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import MessagesState
from langgraph.runtime import Runtime
from langchain.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek

from loguru import logger
from dotenv import load_dotenv
load_dotenv(override=True)

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={
        "thinking": {
            "type": "disabled"
        }
    }
)

# 1. 定义运行时上下文类型
@dataclass
class UserContext:
    username: str
    membership_level: str  # "普通用户" / "VIP"

# 2. 定义图状态
class OverAllState(MessagesState):
    user_input: str
    output: str

# 3. 定义节点：通过 runtime.context 访问运行时上下文
def llm_node(state: OverAllState, runtime: Runtime[UserContext]) -> OverAllState:
    runtime_context = runtime.context

    if runtime_context:
        username = runtime_context.username
        level = runtime_context.membership_level
        logger.info(f"当前用户: {username}, 会员等级: {level}")

        if level == "VIP":
            system_prompt = f"你是高级客服助理。当前VIP用户是{username}，请使用尊称'您'，语气热情周到，回复末尾加上'🎖️VIP专属服务'。"
        else:
            system_prompt = f"你是普通客服助理。当前用户是{username}，请友好简洁地回复。"
    else:
        logger.warning("运行时上下文为空，使用默认风格")
        system_prompt = "你是客服助理，请友好简洁地回复。"

    user_input = state["user_input"]
    messages = state.get("messages", [])
    response = model.invoke(
        [HumanMessage(content=system_prompt)] +
        messages +
        [HumanMessage(content=user_input)]
    )

    return {
        "messages": [response],
        "output": response.content
    }

# 4. 构建图，传入 context_schema
builder = StateGraph(state_schema=OverAllState, context_schema=UserContext)
builder.add_node("llm_node", llm_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", END)

graph = builder.compile()

# === 第一次调用：传入运行时上下文（VIP用户） ===
print("=" * 30, "第一次调用：VIP用户", "=" * 30)
config = {"configurable": {"thread_id": "demo-7.3"}}
res = graph.invoke(
    {"user_input": "你好，帮我查一下最近有什么优惠活动"},
    config=config,
    context=UserContext(username="Alice", membership_level="VIP")
)

print(f"output: {res['output']}")
print()

# === 第二次调用：不传运行时上下文 ===
print("=" * 30, "第二次调用：不传上下文", "=" * 30)
res2 = graph.invoke(
    {"user_input": "再帮我看看有没有新品"},
    config=config
)

print(f"output: {res2['output']}")

from IPython.display import display
display(graph)

持久化机制与记忆

可恢复执行 ​

启用可恢复执行 ​

基于内存 ​

在 Jupyter 场景下，只要 Jupyter 内核没有被重启，并且 InMemorySaver 实例没有被重新创建，内存中的检查点数据就会继续存在 ​

如果重新执行完整代码，导致 checkpointer = InMemorySaver 被重新执行，那么之前保存在内存中的检查点也会被清空 ​

基于 pgsql ​

和 InMemorySaver 不同，PostgresSaver 会将检查点保存到 PostgreSQL 数据库中，只要数据库中的记录没有被删除，即使 Python 程序结束、连接对象重建，历史检查点也仍然存在 ​

setup 方法会创建 checkpoint 相关的表，包括 checkpoints、checkpoint_writes、checkpoint_blobs、checkpoint_migrations ​

checkpoint 相关表 ​

基本介绍 ​

LangGraph 通过 thread_id + checkpoint_ns + checkpoint_id 联合定位和恢复状态 ​

checkpoints ​

主键为 (thread_id, checkpoint_ns, checkpoint_id) 联合主键 ​

另有 thread_id 单列索引，加速按线程筛选查询 ​

writes ​

主键为 (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) 联合主键，确保同一任务下的写入顺序唯一 ​

blobs ​

主键为 (thread_id, checkpoint_ns, channel, version) 联合主键 ​

与 checkpoint_writes 不同，checkpoint_blobs 以 (channel, version) 维度管理数据版本，不直接关联某个具体的 checkpoint_id，允许多个检查点共享同一通道的大型二进制数据 ​

migrations ​

单列主键，仅一条记录，setup 执行时自动检查并更新 ​

该表由 LangGraph 内部管理，用户无需手动操作，未来 LangGraph 版本升级时，若表结构有变更，setup 会根据此版本号自动执行对应的迁移 SQL ​

持久化模式 ​

基本介绍 ​

启用 Checkpointer 后，LangGraph 会在执行过程中保存检查点 ​

检查点写入越及时，系统在异常中断后可恢复的状态越完整，容灾能力越强，但更及时的持久化通常也意味着更多中间写入，或额外的阻塞式等待，带来额外的性能开销和响应延迟（从调用者的角度考虑） ​

三大模式 ​

（1）exit：退出模式，只在计算图正常结束、异常退出、被中断（如 Human-In-The-Loop 中断）时保存检查点，不能处理中途进程崩溃的场景，这种模式响应性能开销最小，但容灾能力最弱 ​

（3）sync：同步模式，和异步模式唯一的区别在于，LangGraph 会在进入下一个超步之前等待当前主检查点的写入任务完成，它的容灾能力最强，但在 async 模式的基础上增加了响应延迟 ​

代码示例 ​

durability 控制的是检查点写入策略，不改变图本身的执行逻辑 ​

查看历史检查点 ​

基本介绍 ​

LangGraph 在配置检查点存储器后，会把同一个 thread_id 下的执行过程保存为一组检查点，通过这些检查点，我们可以查看图运行的中间状态，也可以为后续的检查点回溯（重放、分叉）和失败恢复做准备 ​

本节主要介绍两个常用方法： ​

（1）graph.get_state_history(config)：查看指定会话的完整历史检查点 ​

（2）graph.get_state(config)：查看指定会话的最新检查点，或者查看某个指定 checkpoint_id 对应的检查点 ​

完整历史检查点列表 ​

最新检查点 ​

根据 ID 查询指定检查点 ​

StateSnapshot ​

Time Travel ​

基本介绍 ​

所谓检查点回溯，是指基于某个历史检查点，重新执行后续流程，或者在该检查点基础上修改状态并创建新的执行分支 ​

检查点回溯有两种形式，根据是否更改历史状态区分 ​

（1）Replay：检查点重放，回到某个历史检查点，沿着原先的执行路径重新执行后续节点 ​

（2）Fork：检查点分叉，回到某个历史检查点，修改状态，从该位置创建一条新的执行分支 ​

共同点：检查点之前的节点不会重新执行，检查点之后的节点会重新执行 ​

区别：Replay 不修改历史状态，Fork 会基于历史检查点应用新的状态更新，并创建新的检查点分支 ​

Replay ​

失败恢复通常基于最新检查点继续运行，配置中只包含 thread_id，不包含 checkpoint_id ​

Replay 则是显式传入某个历史检查点的配置，配置中包含 checkpoint_id，LangGraph 会据此从该历史检查点开始重放后续步骤 ​

首先通过查看历史检查点列表找到需要恢复内容的相关参数，再执行代码进行恢复 ​

Fork ​

（1）config：历史检查点的配置信息，通常包含 thread_id、checkpoint_ns 和 checkpoint_id，LangGraph 会据此检索历史检查点 ​

（2）values：要应用到该检查点上的状态更新 ​

Fork 有两种实现方式（1）只改状态（2）改状态且跳过某些节点，如果想跳过某个节点，可以直接指定该节点为 as_node 并通过 values 指定指定该节点的输出 ​

注意：update_state 返回的是新检查点的配置，不是新状态本身，它不会修改原来的历史检查点，而是基于某个历史检查点创建一个新的检查点，这个新检查点相当于一条新的执行分支 ​

长期记忆 ​

数据准备 ​

长期记忆存储器在编译图时通过 store 参数传递，图节点中可以通过 Runtime 对象访问 ​

（1）namespace 使用元组是硬性约束，天然表达层级结构，本例使用两层命名空间 ("users", "Alice")，清晰区分了领域（users）和实体（用户名），而具体存储什么类型的数据则由 key 参数表达（如 "preferences"） ​

（2）(*USERS_NS, "Alice") 的写法表示对已有元组进行解包，再拼接新的命名空间片段 ​

（3）PostgresStore 将长期记忆持久化到 PostgreSQL 中，程序重启后数据不会丢失，适合生产环境，本节使用和 6.2.4 节、6.5.4.1 节 相同的数据库实例 ​

（4）如果希望支持语义检索，需要在 Store 中配置索引和 embedding 函数，未配置索引时，search 只能按照命名空间和过滤条件检索 ​

记忆访问 ​

长期记忆存储器可以通过节点函数中的 runtime.store 访问 ​

第一次调用从长期记忆中读取数据，并写入短期记忆；后续同一会话线程内优先复用短期记忆，避免重复查询长期记忆 ​

运行时上下文 ​

基本介绍 ​

运行时上下文的特点是：仅对本次调用生效，不会被持久化，也不会在同一会话的下一次调用中自动恢复 ​

运行时上下文的使用方式 ​

（1）初始化状态图时，使用 context_schema 定义上下文类型 ​

（2）调用图时通过 context 参数传入上下文对象 ​

（3）节点或路由函数中通过 runtime.context 访问上下文 ​

代码示例 ​

可恢复执行

启用可恢复执行

基于内存

在 Jupyter 场景下，只要 Jupyter 内核没有被重启，并且 InMemorySaver 实例没有被重新创建，内存中的检查点数据就会继续存在

如果重新执行完整代码，导致 checkpointer = InMemorySaver 被重新执行，那么之前保存在内存中的检查点也会被清空

基于 pgsql

和 InMemorySaver 不同，PostgresSaver 会将检查点保存到 PostgreSQL 数据库中，只要数据库中的记录没有被删除，即使 Python 程序结束、连接对象重建，历史检查点也仍然存在

setup 方法会创建 checkpoint 相关的表，包括 checkpoints、checkpoint_writes、checkpoint_blobs、checkpoint_migrations

checkpoint 相关表

基本介绍

LangGraph 通过 thread_id + checkpoint_ns + checkpoint_id 联合定位和恢复状态

checkpoints

主键为 (thread_id, checkpoint_ns, checkpoint_id) 联合主键

另有 thread_id 单列索引，加速按线程筛选查询

writes

主键为 (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) 联合主键，确保同一任务下的写入顺序唯一

blobs

主键为 (thread_id, checkpoint_ns, channel, version) 联合主键

与 checkpoint_writes 不同，checkpoint_blobs 以 (channel, version) 维度管理数据版本，不直接关联某个具体的 checkpoint_id，允许多个检查点共享同一通道的大型二进制数据

migrations

单列主键，仅一条记录，setup 执行时自动检查并更新

该表由 LangGraph 内部管理，用户无需手动操作，未来 LangGraph 版本升级时，若表结构有变更，setup 会根据此版本号自动执行对应的迁移 SQL

持久化模式

基本介绍

启用 Checkpointer 后，LangGraph 会在执行过程中保存检查点

检查点写入越及时，系统在异常中断后可恢复的状态越完整，容灾能力越强，但更及时的持久化通常也意味着更多中间写入，或额外的阻塞式等待，带来额外的性能开销和响应延迟（从调用者的角度考虑）

三大模式

（1）exit：退出模式，只在计算图正常结束、异常退出、被中断（如 Human-In-The-Loop 中断）时保存检查点，不能处理中途进程崩溃的场景，这种模式响应性能开销最小，但容灾能力最弱

（3）sync：同步模式，和异步模式唯一的区别在于，LangGraph 会在进入下一个超步之前等待当前主检查点的写入任务完成，它的容灾能力最强，但在 async 模式的基础上增加了响应延迟

代码示例

durability 控制的是检查点写入策略，不改变图本身的执行逻辑

查看历史检查点

基本介绍

LangGraph 在配置检查点存储器后，会把同一个 thread_id 下的执行过程保存为一组检查点，通过这些检查点，我们可以查看图运行的中间状态，也可以为后续的检查点回溯（重放、分叉）和失败恢复做准备

本节主要介绍两个常用方法：

（1）graph.get_state_history(config)：查看指定会话的完整历史检查点

（2）graph.get_state(config)：查看指定会话的最新检查点，或者查看某个指定 checkpoint_id 对应的检查点

完整历史检查点列表

最新检查点

根据 ID 查询指定检查点

StateSnapshot

Time Travel

基本介绍

所谓检查点回溯，是指基于某个历史检查点，重新执行后续流程，或者在该检查点基础上修改状态并创建新的执行分支

检查点回溯有两种形式，根据是否更改历史状态区分

（1）Replay：检查点重放，回到某个历史检查点，沿着原先的执行路径重新执行后续节点

（2）Fork：检查点分叉，回到某个历史检查点，修改状态，从该位置创建一条新的执行分支

共同点：检查点之前的节点不会重新执行，检查点之后的节点会重新执行

区别：Replay 不修改历史状态，Fork 会基于历史检查点应用新的状态更新，并创建新的检查点分支

Replay

失败恢复通常基于最新检查点继续运行，配置中只包含 thread_id，不包含 checkpoint_id

Replay 则是显式传入某个历史检查点的配置，配置中包含 checkpoint_id，LangGraph 会据此从该历史检查点开始重放后续步骤

首先通过查看历史检查点列表找到需要恢复内容的相关参数，再执行代码进行恢复

Fork

（1）config：历史检查点的配置信息，通常包含 thread_id、checkpoint_ns 和 checkpoint_id，LangGraph 会据此检索历史检查点

（2）values：要应用到该检查点上的状态更新

Fork 有两种实现方式（1）只改状态（2）改状态且跳过某些节点，如果想跳过某个节点，可以直接指定该节点为 as_node 并通过 values 指定指定该节点的输出

注意：update_state 返回的是新检查点的配置，不是新状态本身，它不会修改原来的历史检查点，而是基于某个历史检查点创建一个新的检查点，这个新检查点相当于一条新的执行分支

长期记忆

数据准备

长期记忆存储器在编译图时通过 store 参数传递，图节点中可以通过 Runtime 对象访问

（1）namespace 使用元组是硬性约束，天然表达层级结构，本例使用两层命名空间 ("users", "Alice")，清晰区分了领域（users）和实体（用户名），而具体存储什么类型的数据则由 key 参数表达（如 "preferences"）

（2）(*USERS_NS, "Alice") 的写法表示对已有元组进行解包，再拼接新的命名空间片段

（3）PostgresStore 将长期记忆持久化到 PostgreSQL 中，程序重启后数据不会丢失，适合生产环境，本节使用和 6.2.4 节、6.5.4.1 节相同的数据库实例

（4）如果希望支持语义检索，需要在 Store 中配置索引和 embedding 函数，未配置索引时，search 只能按照命名空间和过滤条件检索

记忆访问

长期记忆存储器可以通过节点函数中的 runtime.store 访问

第一次调用从长期记忆中读取数据，并写入短期记忆；后续同一会话线程内优先复用短期记忆，避免重复查询长期记忆

运行时上下文

基本介绍

运行时上下文的特点是：仅对本次调用生效，不会被持久化，也不会在同一会话的下一次调用中自动恢复

运行时上下文的使用方式

（1）初始化状态图时，使用 context_schema 定义上下文类型

（2）调用图时通过 context 参数传入上下文对象

（3）节点或路由函数中通过 runtime.context 访问上下文

代码示例