LangGraph Persistence 持久化

本文最后更新于 2026年7月27日

引言

持久化（短期/长期记忆）是大模型应用中十分重要的功能，langgraph有两种数据持久化的实现：

checkpoint

保存图的执行过程快照，使图可以暂停，恢复，回溯，类似游戏玩到一半暂存。langgraph具有内置的持久化层，可以将图状态保存为检查点（checkpointer），当使用检查点记录器编译图时，系统会在执行的每一步保存图状态快照，并按thread_id进行组织，实现人机回环（human-in-the-loop），本次对话记忆，时光旅行调试以及容错执行。
Memory store

保存的是业务数据，基于namespace+key，长期记忆使得智能体能够跨会话，跨用户记住一些信息（用户画像，偏好，知识）。

	checkpointer	Memory store
保存内容	整个图的快照，State	自定义的JSON数据
保存时机	超级步 (super-step) 完成	手动put
持久化类	XxxxSaver	XxxxStore

https://docs.langchain.org.cn/oss/python/langgraph/persistence

简而言之，Checkpointer保存图状态实现本次对话的短期记忆，主要目的还是保存业务数据，以thread_id为作用域。

而Memory store专门用于实现针对不同用户特点，偏好，知识等固有属性的长期记忆，与某个State/thread_id解耦。

1.checkpoint

每次调用.stream()/.invoke()，langgraph都会维护一个状态State，如果没有checkpoint，调用结束后就丢掉了，如果启用了checkpoint，就会将State保存到存储中，按照thread_id持久化，下次继续调用时，可以恢复State

例如在这个图中

import operator
from typing import TypedDict, Annotated

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import START, END
from langgraph.graph import StateGraph

class DemoState(TypedDict):
    msg: Annotated[list, operator.add]
    step_count: Annotated[int, operator.add]



def step_one(state: DemoState):
    print('=>执行步骤1')
    return {
        'msg': ['执行步骤1'],
        'step_count': 1,
    }

def step_two(state: DemoState):
    print('=>执行步骤2')
    return {
        'msg': ['执行步骤2'],
        'step_count': 1,
    }

def step_three(state: DemoState):
    print('=>执行步骤3')
    return {
        'msg': ['执行步骤3'],
        'step_count': 1,
    }

if __name__ == '__main__':
    graph = StateGraph(state_schema=DemoState)
    graph.add_node('step_one', step_one)
    graph.add_node('step_two', step_two)
    graph.add_node('step_three', step_three)

    graph.add_edge(START, 'step_one')
    graph.add_edge('step_one', 'step_two')
    graph.add_edge('step_two', 'step_three')
    graph.add_edge('step_three', END)

例：指定checkpointer并传入thread_id，使用InMemorySaver暂存在内存中

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

cfg = {
    'configurable': {'thread_id': 'task_001'}
}

input = {
    'msg': ['开始'],
    'step_count': 0,
}

res = app.invoke(input=input, config=cfg)
print(res)

=>执行步骤1
=>执行步骤2
=>执行步骤3
{'msg': ['开始', '执行步骤1', '执行步骤2', '执行步骤3'], 'step_count': 3}

例：读取最新快照/下一个要执行的节点，由于图一下子执行完了，所以下个节点是空的()

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

cfg = {
    'configurable': {'thread_id': 'task_001'}
}

input = {
    'msg': ['开始'],
    'step_count': 0,
}

app.invoke(input=input, config=cfg)

state = app.get_state(cfg)

print('最新状态值', state.values)
print('下一个要执行的节点', state.next)

=>执行步骤1
=>执行步骤2
=>执行步骤3
最新状态值 {'msg': ['开始', '执行步骤1', '执行步骤2', '执行步骤3'], 'step_count': 3}
下一个要执行的节点 ()

例：状态历史记录，倒序，栈结构

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

cfg = {
    'configurable': {'thread_id': 'task_001'}
}

input = {
    'msg': ['开始'],
    'step_count': 0,
}

app.invoke(input=input, config=cfg)

history = app.get_state_history(cfg)

for point in history:
    print('最新状态值', point.values)
    print('下一个要执行的节点', point.next)
    print(100 * '*')

=>执行步骤1
=>执行步骤2
=>执行步骤3
最新状态值 {'msg': ['开始', '执行步骤1', '执行步骤2', '执行步骤3'], 'step_count': 3}
下一个要执行的节点 ()
****************************************************************************************************
最新状态值 {'msg': ['开始', '执行步骤1', '执行步骤2'], 'step_count': 2}
下一个要执行的节点 ('step_three',)
****************************************************************************************************
最新状态值 {'msg': ['开始', '执行步骤1'], 'step_count': 1}
下一个要执行的节点 ('step_two',)
****************************************************************************************************
最新状态值 {'msg': ['开始'], 'step_count': 0}
下一个要执行的节点 ('step_one',)
****************************************************************************************************
最新状态值 {'msg': [], 'step_count': 0}
下一个要执行的节点 ('__start__',)
****************************************************************************************************

例：time travel时间旅行，可以回到历史上某个点，从这个点开始重新执行一次，因为get_state_history是倒序，因此第二个是step_one()，但是是step_one()执行完的状态，继续向下执行不包含step_one()

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

cfg = {
    'configurable': {'thread_id': 'task_001'}
}

input = {
    'msg': ['开始'],
    'step_count': 0,
}

app.invoke(input=input, config=cfg) #已经执行了一次

history = app.get_state_history(cfg)

lst = list(history)

target = lst[2]
#print(target)

travel_result = app.invoke(None, target.config)
print('*************')
print(travel_result)

=>执行步骤1
=>执行步骤2
=>执行步骤3
=>执行步骤2
=>执行步骤3
*************
{'msg': ['开始', '执行步骤1', '执行步骤2', '执行步骤3'], 'step_count': 3}

时间旅行到某个节点时，还可以修改State，通过.update_state()

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

cfg = {
    'configurable': {'thread_id': 'task_001'}
}

input = {
    'msg': ['开始'],
    'step_count': 0,
}

app.invoke(input=input, config=cfg)  # 已经执行了一次

history = app.get_state_history(cfg)

lst = list(history)

target = lst[2]
# print(target)

new_config = app.update_state(target.config, values={"step_count": 120})

travel_result = app.invoke(None, new_config)
print('*************')
print(travel_result)

=>执行步骤1
=>执行步骤2
=>执行步骤3
=>执行步骤2
=>执行步骤3
*************
{'msg': ['开始', '执行步骤1', '执行步骤2', '执行步骤3'], 'step_count': 123}

例：断点续连，InMemorySaver只能将数据保存进内存，运行结束就消失，如果中断再发起续连，首先要换成持久化数据的数据库，以Sqlite为例

pip install langgraph-checkpoint-sqlite

import operator
from typing import TypedDict, Annotated
import sqlite3

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.constants import START, END
from langgraph.graph import StateGraph

class DemoState(TypedDict):
    msg: Annotated[list, operator.add]
    step_count: Annotated[int, operator.add]


def step_one(state: DemoState):
    print('=>执行步骤1')
    return {
        'msg': ['执行步骤1'],
        'step_count': 1,
    }

def step_two(state: DemoState):
    print('=>执行步骤2')
    return {
        'msg': ['执行步骤2'],
        'step_count': 1,
    }

def step_three(state: DemoState):
    print('=>执行步骤3')
    return {
        'msg': ['执行步骤3'],
        'step_count': 1,
    }

if __name__ == '__main__':
    graph = StateGraph(state_schema=DemoState)
    graph.add_node('step_one', step_one)
    graph.add_node('step_two', step_two)
    graph.add_node('step_three', step_three)

    graph.add_edge(START, 'step_one')
    graph.add_edge('step_one', 'step_two')
    graph.add_edge('step_two', 'step_three')
    graph.add_edge('step_three', END)

    conn = sqlite3.connect(database='test.db', check_same_thread=False)

    app = graph.compile(checkpointer=SqliteSaver(conn=conn))

    # 每次执行，sqlite都会插入数据
    cfg = {
        'configurable': {'thread_id': 'task_10001'}
    }

    input = {
        'msg': ['开始'],
        'step_count': 0,
    }

    res = app.invoke(input=input, config=cfg)

    print(res)

    conn.close()

=>执行步骤1
=>执行步骤2
=>执行步骤3
{'msg': ['开始', '执行步骤1', '执行步骤2', '执行步骤3'], 'step_count': 3}

重新启动一个进程，再次执行这个程序，就会发现，两次的会话记录是连在一起的，前提是thread_id必须是同一个

=>执行步骤1
=>执行步骤2
=>执行步骤3
{'msg': ['开始', '执行步骤1', '执行步骤2', '执行步骤3', '开始', '执行步骤1', '执行步骤2', '执行步骤3'], 'step_count': 6}

总结：checkpoint适合短期记忆，记住一轮对话的场景，例如客服系统，如果长期存储一些信息需要长期记忆。

2.Memory store

Memory store实现长期记忆，是按照user_id/namespace来做隔离区分的，通过store属性来设置使用长期记忆

user_id可以通过各种方式设置，只要能传入节点

例：相同用户，不同thread_id，获取同一用户画像， store=InMemoryStore()设置长期记忆保存在内存，生产环境必须换成持久化数据库，store: BaseStore将store对象注入节点，节点内自由调用API进行存取。

import operator
from typing import TypedDict, Annotated

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore


class DemoState(TypedDict):
    msg: str
    user_id: str


def node1(state: DemoState, store: BaseStore):
    if state['msg'] == '第一次执行':
        print(state['msg'])

        namespace = ('user_id', state['user_id'])
        store.put(namespace, 'profile', {'name': 'lzj', 'employee': 'teacher'})

    return state

def node2(state: DemoState, store: BaseStore):
    if state['msg'] == '第二次执行':
        print(state['msg'])

        namespace = ('user_id', state['user_id'])
        profile = store.get(namespace, 'profile')
        print('获取用户画像 ', profile)

    return state

if __name__ == '__main__':
    graph = StateGraph(state_schema=DemoState)
    graph.add_node('node1', node1)
    graph.add_node('node2', node2)

    graph.add_edge(START, 'node1')
    graph.add_edge('node1', 'node2')
    graph.add_edge('node2', END)

    app = graph.compile(
        checkpointer=InMemorySaver(),
        store=InMemoryStore()
    )

    app.invoke(
        input= {
            'msg': '第一次执行',
            'user_id': 'lzj'
        },
        config= {
            'configurable': {
                'thread_id': 'task_1'
            }
        }
    )

    app.invoke(
        input={
            'msg': '第二次执行',
            'user_id': 'lzj'
        },
        config={
            'configurable': {
                'thread_id': 'task_2'
            }
        }
    )

第一次执行
第二次执行
获取用户画像  Item(namespace=['user_id', 'lzj'], key='profile', value={'name': 'lzj', 'employee': 'teacher'}, created_at='2026-07-26T05:42:17.537858+00:00', updated_at='2026-07-26T05:42:17.537859+00:00')

"如果文章对您有帮助，可以请作者喝杯咖啡吗？"