5.2 短期记忆持久化与使用#
LangGraph 没有为短期记忆单独设计一套存储系统，而是直接复用了我们在第4.9节中介绍过的状态机制（State）。Agent 在一次对话中需要记住的所有内容——包括对话历史、用户上传的文件、检索到的文档片段、中间生成的结果——都可以作为字段放进 State 里。换句话说，State 本身就是短期记忆的载体，不需要额外引入任何新概念。
5.2.1 记忆的组织结构#
如果仅将 Agent 的全局状态存在内存里是不够的，因为一旦程序重启或连接中断数据就丢失了。LangGraph 通过 Checkpointer 机制来解决这个问题，每执行完一个节点当前的 State 快照就会被自动保存到数据库中。这意味着一次对话哪怕中途中断，下次重新连接时也可以从断点处完整恢复，之前的所有上下文都不会丢失。
在 LangGraph 中，所有的状态信息都是通过 StateSnapshot 这个类来进行管理的，全局状态每更新一次就会生成一个新的 StateSnapshot 类对象，所有类对象将会按时间顺序放到一个列表中。
具体地，StateSnapshot 类的定义如下：
1 class StateSnapshot(NamedTuple):
2 values: dict[str, Any] | Any
3 next: tuple[str, ...]
4 config: RunnableConfig
5 metadata: CheckpointMetadata | None
6 created_at: str | None
7 parent_config: RunnableConfig | None
8 tasks: tuple[PregelTask, ...]
9 interrupts: tuple[Interrupt, ...]在上述代码中，第1行 StateSnapshot 是 LangGraph 中定义了一个用于表示状态快照的类，通过其中的成员变量来记录图在某一个超步执行结束后的完整运行状态。第2行 values 表示截止到当前时候所有超步运行后产生的结果值。第3行 next 表示下一步需要执行的节点名称，并且因为一个超步内可以包含多个节点所以类型是 tuple。第4行config 表示程序想要读取这个快照的时候需要使用到的信息，形式为 {'configurable': {'thread_id': '1', , 'checkpoint_id': '1f14ab...'}}。第5行 metadata 表示一些元数据信息，例如 {'source': 'input', 'step': -1, 'parents': {}, 'k': 20, 'top_n': 3} 。第6行表示快照的创建时间戳。第7行 parent_config 表示父快照对应的 config 信息，可以理解为当前快照是从哪个快照演化来的。第8行 tasks 表示当前超步中真正要执行的任务，它与 next 的却别在于 next 记录的是节点名称而 tasks 记录的是节点中具体的任务，它是某一轮超步中要执行的最小任务单元。第9行 interrupts 表示当前等待解决的中断是 Human-in-the-loop 的核心，例如人工审批、用户交互等。
5.2.2 记忆在内存中存储#
基于第4.10节中的实现代码，只需要如下两行语句便能够将每一次更新后的全局状态信息以 StateSnapshot 的形式保存下来：
1 checkpointer = InMemorySaver()
2 graph = workflow.compile(checkpointer=checkpointer)进一步，可以通过如下代码输出保存后的记忆：
print(list(agent.get_state_history(config))[::-1])回忆我们在第4.10节中所搭建的 RAG Agent，整个流程一共历经了5次状态的改变，因此上述输出结果将会是一个包含有5个 StateSnapshot 类对象的列表，核心信息如下（ interrupts 字段均为空所以没有展示）：
[StateSnapshot(
values={'messages': []},
next=('__start__',),
config={'configurable': {'thread_id': '1', , 'checkpoint_id': '1f14ab4f-8ad9-66c6-bfff'}},
metadata={'source': 'input', 'step': -1, 'parents': {}, 'k': 20, 'top_n': 3},
created_at='2026-05-10T08:07:35.189060+00:00',
parent_config=None,
tasks=(PregelTask(id='1471b648-87a9', name='__start__', path=('__pregel_pull', '__start__'), result={'messages': [HumanMessage(content='郭靖和杨康是什么关系？', ....., id='b5e274f0-f729-4def')]}),)),
StateSnapshot(
values={'messages': [HumanMessage(content='郭靖和杨康是什么关系？', ....., id='b5e274f0-f729-4def')]},
next=('decide',),
config={'configurable': {'thread_id': '1', , 'checkpoint_id': '1f14ab4f-8ada-6a9e-8000'}},
metadata={'source': 'loop', 'step': 0, 'parents': {}, 'k': 20, 'top_n': 3},
created_at='2026-05-10T08:07:35.189565+00:00',
parent_config={'configurable': {'thread_id': '1', , 'checkpoint_id': '1f14ab4f-8ad9-66c6-bfff'}},
tasks=(PregelTask(id='961989dc-de9b', name='decide', path=('__pregel_pull', 'decide'), result={'messages': [AIMessage(content='', , id='lc_run--019e06a0-b2ee-7772', tool_calls=[{'name': 'retrieve_context', 'args': {'query': '郭靖和杨康的关系'}, 'id': 'call_f564a34ed6', 'type': 'tool_call'}], )]}),)),
StateSnapshot(
values={'messages': [HumanMessage(content='郭靖和杨康是什么关系？', ....., id='b5e274f0-f729-4def'), AIMessage(content='', , id='lc_run--019e06a0-b2ee-7772', tool_calls=[{'name': 'retrieve_context', 'args': {'query': '郭靖和杨康的关系'}, 'id': 'call_f564a34ed6', 'type': 'tool_call'}], )]},
next=('my_retrieve',),
config={'configurable': {'thread_id': '1', , 'checkpoint_id': '1f14ab4f-984c-6ff6-8001'}},
metadata={'source': 'loop', 'step': 1, 'parents': {}, 'k': 20, 'top_n': 3},
created_at='2026-05-10T08:07:36.599493+00:00',
parent_config={'configurable': {'thread_id': '1', , 'checkpoint_id': '1f14ab4f-8ada-6a9e-8000'}}, tasks=(PregelTask(id='6bc95c95-f0bb', name='my_retrieve', path=('__pregel_pull', 'my_retrieve'), result={'messages': [ToolMessage(content='杨康边哭边说，涕泪滂沱，......直送到店门之外。', name='retrieve_context', id='61b416d2-aa18-4c5b', tool_call_id='call_f564a34ed6', artifact=.....)]}),)),
StateSnapshot(
values={'messages': [HumanMessage(content='郭靖和杨康是什么关系？', ....., id='b5e274f0-f729-4def'), AIMessage(content='', , id='lc_run--019e06a0-b2ee-7772', tool_calls=[{'name': 'retrieve_context', 'args': {'query': '郭靖和杨康的关系'}, 'id': 'call_f564a34ed6', 'type': 'tool_call'}], ), ToolMessage(content='杨康边哭边说，涕泪滂沱，......直送到店门之外。', name='retrieve_context', id='61b416d2-aa18-4c5b', tool_call_id='call_f564a34ed6', artifact=.....)]},
next=('generate_answer',),
config={'configurable': {'thread_id': '1', , 'checkpoint_id': '1f14ab4f-a798-64a6-8002'}},
metadata={'source': 'loop', 'step': 2, 'parents': {}, 'k': 20, 'top_n': 3},
created_at='2026-05-10T08:07:38.203222+00:00',
parent_config={'configurable': {'thread_id': '1', , 'checkpoint_id': '1f14ab4f-984c-6ff6-8001'}},
tasks=(PregelTask(id='440bc686-4eec', name='generate_answer', path=('__pregel_pull', 'generate_answer'), result={'messages': [AIMessage(content='郭靖和杨康是结义兄弟。两人在郭啸天的灵前对拜八拜，结为兄弟，郭靖先出世一个月，为兄，杨康为弟。', , id='lc_run--019e06a0-bddc-7023', tool_calls=[],)]}),)),
StateSnapshot(
values={'messages': [HumanMessage(content='郭靖和杨康是什么关系？', ....., id='b5e274f0-f729-4def'), AIMessage(content='', , id='lc_run--019e06a0-b2ee-7772', tool_calls=[{'name': 'retrieve_context', 'args': {'query': '郭靖和杨康的关系'}, 'id': 'call_f564a34ed6', 'type': 'tool_call'}], ), ToolMessage(content='杨康边哭边说，涕泪滂沱，......直送到店门之外。', name='retrieve_context', id='61b416d2-aa18-4c5b', tool_call_id='call_f564a34ed6', artifact=.....), AIMessage(content='郭靖和杨康是结义兄弟。两人在郭啸天的灵前对拜八拜，结为兄弟，郭靖先出世一个月，为兄，杨康为弟。', , id='lc_run--019e06a0-bddc-7023', tool_calls=[], )]},
next=(),
config={'configurable': {'thread_id': '1', , 'checkpoint_id': '1f14ab4f-b770-62de-8003'}}, metadata={'source': 'loop', 'step': 3, 'parents': {}, 'k': 20, 'top_n': 3},
created_at='2026-05-10T08:07:39.864505+00:00',
parent_config={'configurable': {'thread_id': '1', , 'checkpoint_id': '1f14ab4f-a798-64a6-8002'}}, tasks=())]在上述输出结果中，从上到下依次是由远及近的状态快照内容。同时，根据输出结果可以发现：① values 字段中存放了到截止到当前，所有的 Message，例如最后一个 StateSnapshot 中的 values 就保存了所有的各类 Message 信息；② 相邻两个状态直接通过 config 字段中的 checkpoint_id 进行关联；③最后一个状态快照中将会包含先前所有快照的结果。
以上完整示例代码可参见 Code/Chapter05/C01_short_memory_store.py 文件。
5.2.3 记忆持久化到数据库#
到此，我们只是得到了保存在内存中的状态快照，真正要达到持久化的目的必须要保存到硬盘中。具体地，LangGraph 提供了3种方式来持久化存储记忆，Postgres、MongoDB 和 Redis 。这里我们以 Postgres 数据库为例来进行介绍。
首先，需要安装完成 Postgres 数据库，这里各位可以自行查找相关资料安装。然后通过如下命令安装对应 Python 驱动等。
pip install -U "psycopg[binary,pool]" langgraph langgraph-checkpoint-postgres进一步，需要对先前代码做如下改造：
1 def get_workflow(checkpointer: PostgresSaver):
2 workflow = StateGraph(MessagesState)
3 ......
4 return workflow.compile(checkpointer=checkpointer)
5
6 if __name__ == '__main__':
7 with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
8 checkpointer.setup()
9 agent = get_workflow(checkpointer)
10 vector_store = get_vector_store()
11 config = RunnableConfig(configurable={"vector_store": vector_store, "k": 20,
12 "top_n": 3, "thread_id": "2"})
13 input = {"messages": convert_to_messages([
14 {"role": "user", "content": "郭靖和杨康是什么关系？",}])}
15 for event in agent.stream(input, config=config):
16 for node, update in event.items():
17 update["messages"][-1].pretty_print()在上述代码中，第7行 PostgresSaver.from_conn_string() 是上下文管理器需覆盖整个运行阶段，其中 DB_URI 表示 Postgres 的数据库连接串，格式为 postgresql://用户名:密码@服务器IP:端口/数据库名?sslmode=disable。第8行 setup() 的作用是自动创建 LangGraph 持久化记忆时所需要用到的4张内部表（checkpoint_blobs、checkpoint_migrations、checkpoint_writes和checkpoints），第一次运行时需要加上这一行，否则后面运行时会报：relation does not exist 。第12行中的 thread_id 用于指定本轮会话的 ID，它就是用来告诉 Checkpointer 这次调用属于哪一个对话会话，应该读取和保存哪一份状态快照，不同会话应区分开。
上述代码运行结束以后将会在数据库中看到新建立的4张表以及对应的持久化内容。注意，数据库中只会持久化最后一个 StateSnapshot 状态快照。以上完整代码可参见 Code/Chapter05/C02_short_memory_store_pg.py 文件。
5.2.4 读取记忆进行对话#
当记忆被持久化到数据库以后可以将其再次加载到会话中进行使用。具体地，可以通过如下方式打印对应 THREAD_ID 在数据库中已有的记忆信息：
1 def print_memory(agent, config: RunnableConfig) -> None:
2 snapshot = agent.get_state(config)
3 messages = snapshot.values.get("messages", [])
4 print(f"读取到 thread_id={THREAD_ID} 的短期记忆，共 {len(messages)} 条消息：")
5 for index, message in enumerate(messages, start=1):
6 print(f"{index:02d}. {message.type}: {_message_preview(message)}")
7
8 if __name__ == '__main__':
9 with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
10 agent = get_workflow(checkpointer)
11 vector_store = get_vector_store()
12 config = RunnableConfig(configurable={
13 "vector_store": vector_store, "k": 20,
14 "top_n": 3, "thread_id": THREAD_ID})
15 print("========== Step 1: 读取数据库中的短期记忆 ==========")
16 print_memory(agent, config)在上述代码中，1~6行是读取从数据库中加载的历史记忆并格式化输出。第9行开始则是数据库连接的上下文管理器，用于 LangGraph 后端载入记忆。
在上述代码中运行结束以后将会输出类似如下信息：
读取到 thread_id=2 的短期记忆，共 4 条消息：
01. human: 郭靖和杨康是什么关系？
02. ai:
03. tool: 杨康边哭边说，涕泪滂沱，断断续续地道：“我是郭靖的结义兄弟，郭大哥给人用这铁枪的枪头刺死了。那奸贼是宋朝军官，料来是受了宰相史弥远的指使。” 拖雷兄妹听到那通蒙古语的军官传译出来，都似焦雷轰顶，做声不得。哲别、博尔忽都和郭靖情谊甚深，四人登时捶胸大哭。 杨康又说起郭靖在宝应杀退金兵、相救拖雷等人之事。拖雷等更无怀疑，细询郭靖的死状，仇人是谁。杨康说道害死郭靖的是大宋指挥使段天德，他知道此人的所在，...
04. ai: 郭靖和杨康是结义兄弟。两人在郭啸天灵前对拜八拜，结为兄弟，郭靖先出世一个月，为兄，杨康为弟。可以发现，输出的历史记忆就是上面最后一个 StateSnapshot 中的 values 值。
进一步，可以通过如下方式基于历史记忆继续对话：
1 if __name__ == '__main__':
2 with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
3 agent = get_workflow(checkpointer)
4 ......
5
6 print("\n========== Step 2: 基于历史记忆继续追问 ==========")
7 follow_up_input = {
8 "messages": convert_to_messages([{ "role": "user",
9 "content": "在上个问题中我问的问题是什么？告诉我",}])}
10
11 for event in agent.stream(follow_up_input, config=config):
12 for node, update in event.items():
13 print(f"#### 节点 {node} 处理完毕")
14 update["messages"][-1].pretty_print()
15 print("\n")
16
17 print("========== Step 3: 查看追加后的完整记忆 ==========")
18 print_memory(agent, config)上述代码运行结束后将会得到类似如下输出：
========== Step 2: 基于历史记忆继续追问 ==========
#### 节点 decide 处理完毕
================================== Ai Message ==================================
你上一个问题问的是：“郭靖和杨康是什么关系？”
========== Step 3: 查看追加后的完整记忆 ==========
读取到 thread_id=2 的短期记忆，共 6 条消息：
01. human: 郭靖和杨康是什么关系？
02. ai:
03. tool: 杨康边哭边说，涕泪滂沱，断断续续地道：“我是郭靖的结义兄弟，郭大哥给人用这铁枪的枪头刺死了。那奸贼是宋朝军官，料来是受了宰相史弥远的指使。” 拖雷兄妹听到那通蒙古语的军官传译出来，都似焦雷轰顶，做声不得。哲别、博尔忽都和郭靖情谊甚深，四人登时捶胸大哭。 杨康又说起郭靖在宝应杀退金兵、相救拖雷等人之事。拖雷等更无怀疑，细询郭靖的死状，仇人是谁。杨康说道害死郭靖的是大宋指挥使段天德，他知道此人的所在，...
04. ai: 郭靖和杨康是结义兄弟。两人在郭啸天灵前对拜八拜，结为兄弟，郭靖先出世一个月，为兄，杨康为弟。
05. human: 在上个问题中我问的问题是什么？告诉我
06. ai: 你上一个问题问的是：“郭靖和杨康是什么关系？”从输出结果可以看出，在 with PostgresSaver.from_conn_string() 数据库连接上下文环境管理器的包裹下，所有会话内容都将同步被持久化到数据库中。以上完整示例代码可参见 Code/Chapter05/C03_short_memory_use.py 文件。
在多轮会话过程中如果还想观察每次喂给大模型的具体内容，还可以在 RunnableConfig 中加入 callbacks=[ConsoleCallbackHandler()] 进行输出。
例如对于上面的多轮交互过程来说，当用户再次询问“郭靖比杨康先出生几个月？”这个问题时，输入到大模型中的内容其实是如下样式：
"prompts": ["Human: 郭靖和杨康是什么关系？\nAI: [{'name': 'retrieve_context', 'args': {'query': '郭靖和杨康的关系'}, 'id': 'call_51a94d40fcbd410d930577', 'type': 'tool_call'}]\nTool: 杨康边哭边说，涕泪滂沱，断断续续地道：“我是郭靖的结义兄弟，郭大哥给人用这铁枪的枪头刺死了。那奸贼是宋朝军官，料来是受了宰相史弥远的指使。”\n拖雷兄妹听到那通蒙古语的军官传译出来，......\nAI: 郭靖和杨康是结义兄弟。两人在郭啸天灵前对拜八拜，结为兄弟，郭靖先出世一个月，为兄，杨康为弟。\nHuman: 在上个问题中我问的问题是什么？告诉我\nAI: 你上一个问题问的是：“郭靖和杨康是什么关系？”\nHuman: 郭靖比杨康先出生几个月？"]输出结果为：
根据提供的文本，郭靖比杨康先出生**一个月**。原文依据：“两人叙起年纪，郭靖先出世一个月，两人在郭啸天灵前对拜了八拜，结为兄弟。”换句话说，就是粗暴地将之前的所有历史对话信息都给一股脑的输入到了大模型中，而这显然也带来了巨大的弊端以上完整输出日志内容可参见 Code/Chapter05/C03_short_memory_use_output_log.txt 文件。
到此，对于短期记忆的持久化与加载使用就介绍完了。在下一节内容中我们将继续介绍短期记忆的管理。
引用#
[1] https://docs.langchain.com/oss/python/langgraph/add-memory