LangGraph 系列 9：智能体的短期记忆与长期记忆

AgentState负责描述一次 Agent 运行中有哪些状态，以及节点和 Tool 应该如何读取、更新这些状态。

但是，只有 State 还不等于拥有记忆。如果没有配置持久化组件，一次 invoke() 结束后，下一次调用不会自动恢复上一次的消息。进程退出后，保存在内存里的状态也会一起消失。

这一篇继续解决两个问题：

• 同一个会话怎样记住前面说过的话。
• 同一个用户怎样在不同会话中共享长期偏好。

本文使用 LangGraph Checkpointer 保存短期记忆，使用 Store 保存长期记忆，并通过独立 PostgreSQL 容器验证重启恢复。这里不使用 Embedding、pgvector 或语义检索，长期记忆只是结构化数据的精确读写。

1. State、记忆与持久化的分别

先把三个容易混淆的概念分开。

• AgentState 定义 Agent 当前有哪些状态。例如 messages 保存当前对话中的用户消息、工具调用和模型回答。它解决的是“运行时数据长什么样”。

• 记忆 表示 Agent 可以在后续运行中继续使用以前的信息。它解决的是“哪些过去的信息需要再次取出来”。

• 持久化 负责把这些信息保存到进程之外。它解决的是“程序或数据库重启后，数据是否还在”。

LangGraph 中的短期记忆通常以 Thread 为边界，通过 Checkpointer 保存 AgentState 的 checkpoint；长期记忆通常以用户或业务对象为边界，通过 Store 保存独立的 key-value 数据。

2. 短期记忆与长期记忆

短期记忆主要服务于当前会话。它保存同一个 Thread 中的消息和状态，让 Agent 能够理解“我刚才说了什么”。

长期记忆不应该被某一个 Thread 限制。例如用户在线程 A 中告诉 Agent 自己喜欢周末去西湖散步，之后新建线程 B，Agent 仍然可以读取这项偏好。

这两类记忆使用不同的定位方式：

• Checkpointer 使用 thread_id 找到某个会话的 State。
• Store 使用 namespace + key 找到某个用户或业务对象的数据。

下图先展示短期记忆的保存与恢复过程。thread_id 只是 Checkpointer 定位某个 Thread 最新 checkpoint 的查询条件，不是记忆内容本身。相同 thread_id 会先恢复已有 State，再合并本轮输入；不同 thread_id 则从独立 State 开始。

InMemorySaver 与 PostgresSaver 的逻辑流程相同，区别在于保存位置和生命周期：前者只保存在当前 Python 进程中，后者将 checkpoint 持久化到 PostgreSQL。

3. 准备独立 PostgreSQL

为了不影响本机已有数据库，本例单独启动一个 PostgreSQL 16.9 容器，只监听 127.0.0.1:25432。

项目目录如下：

langgraph/p09_agent_memory/
├── 01_in_memory_short_term.py
├── 02_setup_postgres_memory.py
├── 03_postgres_short_term_write.py
├── 04_postgres_short_term_resume.py
├── 05_postgres_store_basic.py
├── 06_postgres_long_term_write.py
├── 07_postgres_long_term_read.py
├── 08_inspect_postgres_memory.py
├── docker-compose.yml
├── requirements-lock.txt
├── requirements.txt
└── README.md

docker-compose.yml 的完整内容如下：

services:
  postgres:
    image: postgres:16.9
    container_name: llm-learning-p09-postgres
    restart: unless-stopped
    environment:
      POSTGRES_DB: langgraph_memory
      POSTGRES_USER: langgraph
      POSTGRES_PASSWORD: langgraph_demo_1234
    ports:
      - "127.0.0.1:25432:5432"
    volumes:
      - p09_postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U langgraph -d langgraph_memory"]
      interval: 2s
      timeout: 3s
      retries: 30

volumes:
  p09_postgres_data:
    name: llm-learning-p09-postgres-data

这里使用的是本地测试密码，只允许本机访问。真实项目不能把生产数据库密码直接写在 Compose 文件或源码中。

从 llm_learning 根目录启动数据库：

docker compose \
  -f langgraph/p09_agent_memory/docker-compose.yml \
  up -d

docker compose \
  -f langgraph/p09_agent_memory/docker-compose.yml \
  ps

停止容器但保留数据卷：

docker compose \
  -f langgraph/p09_agent_memory/docker-compose.yml \
  down

只有执行下面的命令才会同时删除演示数据：

docker compose \
  -f langgraph/p09_agent_memory/docker-compose.yml \
  down -v

4. 安装 Checkpointer 与 Store 依赖

本例继续使用 .venv_langgraph，只增加 PostgreSQL Checkpointer、psycopg 和连接池相关依赖。具体依赖由锁文件统一管理。

安装锁定依赖：

cd /path/to/llm-learning
source .venv_langgraph/bin/activate

python -m pip install \
  -r langgraph/p09_agent_memory/requirements-lock.txt
python -m pip check

5. 启动本地 Qwen3

短期记忆和长期记忆示例都通过本地 Qwen3 完成真实对话。先在 llm_learning 根目录启动模型服务：

source .venv_tool_server/bin/activate

"$VIRTUAL_ENV/bin/python" -m mlx_lm server \
  --model Qwen3-14B-AWQ-4bit-MLX \
  --host 127.0.0.1 \
  --port 18080 \
  --prompt-cache-size 0 \
  --chat-template-args '{"enable_thinking": false}'

另开终端检查服务：

curl --noproxy '*' http://127.0.0.1:18080/v1/models

只要返回 HTTP 200 和非空模型列表，就可以继续运行代码。文章不保留接口中的动态创建时间。

6. 短期记忆:InMemorySaver

先不连接 PostgreSQL，使用 InMemorySaver 理解 thread_id 的作用。

InMemorySaver 会根据 configurable.thread_id 保存 checkpoint。相同 thread_id 会恢复以前的 State，不同 thread_id 使用不同状态。

完整代码如下：

"""使用 InMemorySaver 演示同一线程的短期记忆和不同线程的隔离。"""
import httpx
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver

# 第 1 步：创建模型。
model = ChatOpenAI(
    model="Qwen3-14B-AWQ-4bit-MLX",
    base_url="http://127.0.0.1:18080/v1",
    api_key="not-needed",
    temperature=0,
    max_tokens=128,
    # 禁止读取系统代理，确保请求直接发送到本机 Qwen3。
    http_client=httpx.Client(trust_env=False),
)

# 第 2 步：创建内存 Checkpointer，并交给 Agent。
# InMemorySaver 只把线程状态保存在当前 Python 进程的内存中。
checkpointer = InMemorySaver()
agent = create_agent(
    model=model,
    tools=[],
    checkpointer=checkpointer,
    system_prompt=(
        "你是一个中文助手。只根据当前线程的历史消息回答。"
        "如果历史消息中没有用户姓名，只回答：不知道。"
    ),
)

# 第 3 步：为两段互相隔离的对话准备不同 thread_id。
same_thread = {"configurable": {"thread_id": "p09-memory-a"}}
other_thread = {"configurable": {"thread_id": "p09-memory-b"}}

# 第一轮把名字写入 p09-memory-a 的 State。
first = agent.invoke(
    {"messages": [{"role": "user", "content": "请记住：我叫小明。只回复：已记住。"}]},
    config=same_thread,
)
print(f"第一轮回答：{first['messages'][-1].content}")

# 相同 thread_id 会先恢复上一轮 State，因此能够回答姓名。
second = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么名字？只回答名字。"}]},
    config=same_thread,
)
print(f"同一线程回答：{second['messages'][-1].content}")

# 不同 thread_id 拥有独立 State，不能看到 p09-memory-a 的消息。
third = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么名字？只回答名字。"}]},
    config=other_thread,
)
print(f"另一线程回答：{third['messages'][-1].content}")

运行代码：

python langgraph/p09_agent_memory/01_in_memory_short_term.py

一次真实输出如下：

第一轮回答：已记住。
同一线程回答：小明
另一线程回答：不知道。

这说明 Thread 是短期记忆的隔离边界。不过，InMemorySaver 的数据只存在于当前 Python 进程中，脚本结束后无法再次恢复。

7. 初始化 PostgreSQL 数据表

需要跨进程和重启恢复时，可以把 Checkpointer 换成 PostgresSaver。长期记忆使用的 PostgresStore 也需要自己的数据表。

7.1 初始化代码

"""初始化 PostgresSaver 和 PostgresStore 所需的数据表。"""

import psycopg
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore

DB_URI = (
    "postgresql://langgraph:langgraph_demo_1234@127.0.0.1:25432/langgraph_memory?sslmode=disable"
)

# 第 1 步：初始化 Checkpointer 数据表。
# Checkpointer 和 Store 使用不同的数据表，因此需要分别执行 setup()。
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()

# 第 2 步：初始化长期记忆 Store 数据表。
with PostgresStore.from_conn_string(DB_URI) as store:
    store.setup()

# 第 3 步：查询初始化后的表名。
# 只展示稳定的表名，不输出迁移时间或其他动态字段。
with psycopg.connect(DB_URI) as connection:
    with connection.cursor() as cursor:
        cursor.execute(
            """
            SELECT table_name
            FROM information_schema.tables
            WHERE table_schema = 'public'
            ORDER BY table_name
            """
        )
        table_names = [row[0] for row in cursor.fetchall()]

print("初始化完成，public Schema 中的数据表：")
for table_name in table_names:
    print(f"- {table_name}")

连续运行两次：

python langgraph/p09_agent_memory/02_setup_postgres_memory.py
python langgraph/p09_agent_memory/02_setup_postgres_memory.py

两次都得到相同结果：

初始化完成，public Schema 中的数据表：
- checkpoint_blobs
- checkpoint_migrations
- checkpoint_writes
- checkpoints
- store
- store_migrations

setup() 用于创建表和执行缺少的迁移，可以重复调用。它不会为了重新初始化而清空已经保存的 checkpoint 和 Store 数据。

7.2 表结构分析

-- public.checkpoint_blobs definition

-- Drop table

-- DROP TABLE public.checkpoint_blobs;

CREATE TABLE public.checkpoint_blobs (
	thread_id text NOT NULL,
	checkpoint_ns text DEFAULT ''::text NOT NULL,
	channel text NOT NULL,
	"version" text NOT NULL,
	"type" text NOT NULL,
	"blob" bytea NULL,
	CONSTRAINT checkpoint_blobs_pkey PRIMARY KEY (thread_id, checkpoint_ns, channel, version)
);
CREATE INDEX checkpoint_blobs_thread_id_idx ON public.checkpoint_blobs USING btree (thread_id);

-- public.checkpoints definition

-- Drop table

-- DROP TABLE public.checkpoints;

CREATE TABLE public.checkpoints (
	thread_id text NOT NULL,
	checkpoint_ns text DEFAULT ''::text NOT NULL,
	checkpoint_id text NOT NULL,
	parent_checkpoint_id text NULL,
	"type" text NULL,
	"checkpoint" jsonb NOT NULL,
	metadata jsonb DEFAULT '{}'::jsonb NOT NULL,
	CONSTRAINT checkpoints_pkey PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
);
CREATE INDEX checkpoints_thread_id_idx ON public.checkpoints USING btree (thread_id);

-- public.checkpoint_migrations definition

-- Drop table

-- DROP TABLE public.checkpoint_migrations;

CREATE TABLE public.checkpoint_migrations (
	v int4 NOT NULL,
	CONSTRAINT checkpoint_migrations_pkey PRIMARY KEY (v)
);

-- public.checkpoint_writes definition

-- Drop table

-- DROP TABLE public.checkpoint_writes;

CREATE TABLE public.checkpoint_writes (
	thread_id text NOT NULL,
	checkpoint_ns text DEFAULT ''::text NOT NULL,
	checkpoint_id text NOT NULL,
	task_id text NOT NULL,
	idx int4 NOT NULL,
	channel text NOT NULL,
	"type" text NULL,
	"blob" bytea NOT NULL,
	task_path text DEFAULT ''::text NOT NULL,
	CONSTRAINT checkpoint_writes_pkey PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
);
CREATE INDEX checkpoint_writes_thread_id_idx ON public.checkpoint_writes USING btree (thread_id);


-- public.store definition

-- Drop table

-- DROP TABLE public.store;

CREATE TABLE public.store (
	prefix text NOT NULL,
	"key" text NOT NULL,
	value jsonb NOT NULL,
	created_at timestamptz DEFAULT CURRENT_TIMESTAMP NULL,
	updated_at timestamptz DEFAULT CURRENT_TIMESTAMP NULL,
	expires_at timestamptz NULL,
	ttl_minutes int4 NULL,
	CONSTRAINT store_pkey PRIMARY KEY (prefix, key)
);
CREATE INDEX idx_store_expires_at ON public.store USING btree (expires_at) WHERE (expires_at IS NOT NULL);
CREATE INDEX store_prefix_idx ON public.store USING btree (prefix text_pattern_ops);

-- public.store_migrations definition

-- Drop table

-- DROP TABLE public.store_migrations;

CREATE TABLE public.store_migrations (
	v int4 NOT NULL,
	CONSTRAINT store_migrations_pkey PRIMARY KEY (v)
);

这 6 张表可以分成两组：

• checkpoints、checkpoint_blobs、checkpoint_writes 和 checkpoint_migrations 由 PostgresSaver 使用，负责短期记忆。
• store 和 store_migrations 由 PostgresStore 使用，负责长期记忆。

两组表彼此独立。短期记忆以 thread_id 为主要隔离边界，长期记忆以 namespace + key 为定位方式。PostgresSaver.setup() 和 PostgresStore.setup() 也分别维护自己的迁移记录，因此两者都需要执行。

7.2.1 checkpoints：主表

checkpoints 的一行表示某个 Thread 在某个执行时刻生成的一份 checkpoint。它保存 checkpoint 的结构、元数据以及与前一个 checkpoint 的关系，是恢复 Thread 状态时首先查询的表。

列	作用
thread_id	Thread 的业务标识，对应调用 Agent 时传入的 configurable.thread_id。相同 thread_id 的 checkpoint 属于同一段短期记忆。
checkpoint_ns	checkpoint 命名空间，用来隔离同一 Thread 内不同图或子图的 checkpoint。根图通常使用默认值 ''。它和 thread_id 一起确定 checkpoint 所属的逻辑空间。
checkpoint_id	checkpoint 在当前 Thread 和命名空间内的唯一标识，由 LangGraph 生成。一次 Thread 执行会产生多个 checkpoint，因此不能只用 thread_id 定位某一个具体状态。
parent_checkpoint_id	当前 checkpoint 的父 checkpoint 标识。第一份 checkpoint 没有父节点，因此可以为 NULL。LangGraph 利用它形成状态演进链，并支持恢复、回放和分支。
type	表结构保留的类型字段。当前版本的 PostgresSaver 保存 checkpoint 主体时通常不写这个字段，所以一般为 NULL；它不是 Agent State 的业务类型。
checkpoint	checkpoint 的主体，使用 jsonb 保存。里面通常包含 checkpoint 格式版本、ID、时间、各 channel 的版本号、节点已经观察到的版本等信息。可以直接放入 JSON 的简单 channel 值也可能内联在这里；需要序列化的复杂值则保存在 checkpoint_blobs。
metadata	checkpoint 的附加运行信息，使用 jsonb 保存，例如产生 checkpoint 的来源、执行步数以及本轮节点写入摘要。list() 的 metadata 过滤最终也会作用在这一列上。

主键是 (thread_id, checkpoint_ns, checkpoint_id)。这表示不同 Thread 或不同命名空间可以拥有各自独立的 checkpoint ID 空间。checkpoints_thread_id_idx 用于加速按 Thread 查询或删除数据。

7.2.2 checkpoint_blobs：状态值表

LangGraph 的 State 在内部由多个 channel 组成，例如 Agent 常见的 messages 就是一个 channel。checkpoint_blobs 保存不能直接内联到 checkpoints.checkpoint 中的 channel 状态快照。

列	作用
thread_id	channel 状态所属的 Thread。
checkpoint_ns	channel 状态所属的 checkpoint 命名空间，含义与 checkpoints.checkpoint_ns 相同。
channel	State channel 的名称，例如 messages，也可能是 LangGraph 使用的内部 channel。
version	该 channel 值的版本号。它是 channel 级别的版本，不是数据库迁移版本，也不是 checkpoint_id。checkpoints.checkpoint 中的 channel_versions 会记录每个 channel 应读取哪个版本。
type	序列化类型标签。LangGraph 反序列化 blob 时会同时使用 type 和 blob，因此不能只读取二进制内容来判断原始 Python 类型。特殊值 empty 表示这个版本没有实际状态值。
blob	channel 值序列化后的二进制内容。该列允许为 NULL，因为当 type='empty' 时只需要记录“值为空”这一语义，不需要保存二进制数据。

主键是 (thread_id, checkpoint_ns, channel, version)，其中没有 checkpoint_id。这是有意的设计：checkpoint 通过 channel_versions 引用 channel 的具体版本；如果某个 channel 在相邻 checkpoint 之间没有变化，两份 checkpoint 可以引用同一个 blob，避免重复保存完整状态。

checkpoint_blobs_thread_id_idx 用于加速按 Thread 查询和清理 blob。

7.2.3 checkpoint_writes：节点中间写入表

checkpoint_writes 保存某个 checkpoint 上由节点任务产生、但尚未合并为下一份完整 checkpoint 的写入，也称为 pending writes。它让 LangGraph 在失败、重试、中断恢复或并行任务场景中保留已经完成的节点结果，避免恢复时无条件重复执行所有工作。

列	作用
thread_id	写入所属的 Thread。
checkpoint_ns	写入所属的 checkpoint 命名空间。
checkpoint_id	写入基于哪一份 checkpoint 产生，用来和 checkpoints 主表进行逻辑关联。
task_id	产生这次写入的任务标识。一次图执行中可能有多个节点任务，task_id 用于区分它们。
idx	同一个任务内各项写入的位置编号，用于维持稳定顺序并实现幂等保存。部分内部 channel 会使用预定义编号，因此它不应被当作整个图的全局执行步数。
channel	本项写入的目标 channel。除了业务 State channel，还可能出现任务调度、错误或中断等 LangGraph 内部 channel。
type	blob 的序列化类型标签，作用与 checkpoint_blobs.type 相同。表结构允许它为 NULL；正常通过 LangGraph 写入时会记录实际的序列化类型。
blob	本项写入值序列化后的二进制内容。恢复 pending writes 时，LangGraph 使用 type + blob 还原原始值。
task_path	任务在图或子图中的执行路径。根路径默认是 ''，LangGraph 可用它对任务写入进行稳定排序，并区分嵌套或并行任务。

主键是 (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)。它保证同一任务的同一项写入不会被重复插入。checkpoint_writes_thread_id_idx 用于加速按 Thread 查询和清理中间写入。

7.2.4 checkpoint_migrations：迁移记录表

checkpoint_migrations 不保存任何 Agent 记忆，只记录 PostgresSaver 已经执行过哪些数据库迁移。

列	作用
v	迁移版本号，也是该表的主键。setup() 会读取最大的 v，只继续执行后续尚未应用的迁移。

因此，重复调用 PostgresSaver.setup() 不会重复建表或清空 checkpoint。这个表应由 LangGraph 管理，不应该把其中的版本号当作应用版本手工修改。

7.2.5 store：长期记忆数据表

store 是 PostgresStore 的主体表。一行表示一个由 namespace + key 唯一定位的长期记忆项。它不依赖 thread_id，所以同一个用户可以在多个 Thread 中读取同一项长期记忆。

列	作用
prefix	Store 的 namespace。Python 中的 namespace 是字符串元组，例如 ("users", "user-001")；PostgresStore 会将其编码为 users.user-001 这样的点分字符串存入 prefix。它负责划分用户、应用或业务领域。
key	namespace 内某一项记忆的唯一键，例如 profile 或 preferences。
value	记忆正文，使用 jsonb 保存。调用 store.put(namespace, key, value) 时传入的结构化数据就写入这里。
created_at	该记忆项首次创建的时间。更新同一个 (prefix, key) 时会保留原创建时间。
updated_at	最近一次写入该记忆项的时间。发生 upsert 更新时会改为当前时间；未使用向量检索时，search() 默认也会按它从新到旧排序。
expires_at	该记忆项的实际过期时间。未设置 TTL 时为 NULL。设置 TTL 时通常为写入时间加上 TTL 时长；启用 TTL 刷新后，读取或搜索也可以延后它。
ttl_minutes	TTL 的时长，单位为分钟。它不仅记录最初的过期策略，也用于刷新 TTL 时重新计算 expires_at。未设置 TTL 时为 NULL。

主键是 (prefix, key)。因此，相同 key 可以出现在不同 namespace 中，而对相同 (prefix, key) 再次调用 put() 会更新原有记录。

store_prefix_idx 使用 text_pattern_ops，主要加速按 namespace 前缀进行的查询；idx_store_expires_at 是只包含非空 expires_at 的部分索引，用于加速清理过期数据。需要注意，写入 expires_at 不等于数据库会自动删除记录：应用还需要调用 TTL 清理方法，或启动 PostgresStore 的 TTL sweeper。

本文没有配置向量索引，因此只有 store 主表。如果为 PostgresStore 配置 Embedding 和 pgvector，setup() 还会创建 store_vectors、vector_migrations 等相关表。

7.2.6 store_migrations：迁移记录表

store_migrations 与 checkpoint_migrations 的职责相同，但它只服务于 PostgresStore。

列	作用
v	Store 数据库迁移的版本号，也是该表的主键。PostgresStore.setup() 根据它判断接下来还需要执行哪些迁移。

两个 migration 表不能合并，因为 Checkpointer 和 Store 是两个独立组件，各自拥有不同的表结构和升级节奏。

7.2.7 表之间的关系

这套表结构没有声明外键，表之间由 LangGraph 在应用层通过复合字段建立逻辑关联：

1. checkpoints 根据 thread_id + checkpoint_ns 保存一个 Thread 的 checkpoint 链，parent_checkpoint_id 指向链中的上一份 checkpoint。
2. checkpoint 中的 channel_versions 根据 thread_id + checkpoint_ns + channel + version 到 checkpoint_blobs 读取复杂 channel 值。
3. LangGraph 根据 thread_id + checkpoint_ns + checkpoint_id 从 checkpoint_writes 取回该 checkpoint 上的 pending writes。
4. store 完全独立于 checkpoint 表，只通过 prefix + key 定位长期记忆。

因此，短期记忆不是简单地把完整对话 JSON 重复插入一张表，而是由 checkpoint 结构、按版本保存的 channel 值和节点中间写入共同组成。PostgresSaver.delete_thread(thread_id) 也会分别删除 checkpoints、checkpoint_blobs 和 checkpoint_writes 中的数据，而不会影响独立的长期记忆 store。

8. 短期记忆: PostgreSQL

8.1 写入短期记忆

写入脚本使用固定线程 p09-short-001。为了让示例可以重复执行，代码只删除当前演示线程，不会清空数据库中的其他会话。

"""使用 PostgresSaver 写入第一轮短期记忆。"""

import httpx
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = (
    "postgresql://langgraph:langgraph_demo_1234@127.0.0.1:25432/langgraph_memory?sslmode=disable"
)
THREAD_ID = "p09-short-001"

# 第 1 步：创建模型。
model = ChatOpenAI(
    model="Qwen3-14B-AWQ-4bit-MLX",
    base_url="http://127.0.0.1:18080/v1",
    api_key="not-needed",
    temperature=0,
    max_tokens=128,
    http_client=httpx.Client(trust_env=False),
)

# 第 2 步：打开 PostgreSQL Checkpointer。
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # 只清理当前示例线程，保证脚本重复执行时不会累积旧消息。
    checkpointer.delete_thread(THREAD_ID)

    # 第 3 步：创建 Agent，并把第一轮对话保存到固定 thread_id。
    agent = create_agent(
        model=model,
        tools=[],
        checkpointer=checkpointer,
        system_prompt=(
            "你是一个中文助手。只根据当前线程的历史消息回答。"
            "如果历史消息中没有用户姓名，只回答：不知道。"
        ),
    )
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "请记住：我叫小明。只回复：已记住。"}]},
        config={"configurable": {"thread_id": THREAD_ID}},
    )

print(f"已写入线程：{THREAD_ID}")
print(f"模型回答：{result['messages'][-1].content}")

运行结果：

已写入线程：p09-short-001
模型回答：已记住。

Agent 执行完成时，PostgresSaver 已经把该 Thread 的 State checkpoint 保存到 PostgreSQL。

8.2 恢复短期记忆

先停止再启动 PostgreSQL：

docker compose \
  -f langgraph/p09_agent_memory/docker-compose.yml \
  down

docker compose \
  -f langgraph/p09_agent_memory/docker-compose.yml \
  up -d

Compose 的命名数据卷仍然存在，因此数据库重启不会删除记忆。

在新的 Python 进程中运行恢复脚本：

"""在新进程中恢复 PostgreSQL 短期记忆，并验证线程隔离。"""

import httpx
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = (
    "postgresql://langgraph:langgraph_demo_1234@127.0.0.1:25432/langgraph_memory?sslmode=disable"
)
SAVED_THREAD_ID = "p09-short-001"
OTHER_THREAD_ID = "p09-short-other"

# 第 1 步：创建与写入脚本相同的模型。
model = ChatOpenAI(
    model="Qwen3-14B-AWQ-4bit-MLX",
    base_url="http://127.0.0.1:18080/v1",
    api_key="not-needed",
    temperature=0,
    max_tokens=128,
    http_client=httpx.Client(trust_env=False),
)

# 第 2 步：重新连接 PostgreSQL。这里是一个新的 Python 进程。
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # 清理对照线程，避免它受到以前测试数据的影响。
    checkpointer.delete_thread(OTHER_THREAD_ID)

    agent = create_agent(
        model=model,
        tools=[],
        checkpointer=checkpointer,
        system_prompt=(
            "你是一个中文助手。只根据当前线程的历史消息回答。"
            "如果历史消息中没有用户姓名，只回答：不知道。"
        ),
    )

    # 第 3 步：固定 thread_id 能恢复原消息，另一个 thread_id 则读不到。
    resumed = agent.invoke(
        {"messages": [{"role": "user", "content": "我叫什么名字？只回答名字。"}]},
        config={"configurable": {"thread_id": SAVED_THREAD_ID}},
    )
    isolated = agent.invoke(
        {"messages": [{"role": "user", "content": "我叫什么名字？只回答名字。"}]},
        config={"configurable": {"thread_id": OTHER_THREAD_ID}},
    )

print(f"恢复线程回答：{resumed['messages'][-1].content}")
print(f"对照线程回答：{isolated['messages'][-1].content}")

一次真实输出如下：

恢复线程回答：小明
对照线程回答：不知道。

写入与读取发生在两个独立 Python 进程中，并且中间重启过 PostgreSQL。相同 thread_id 恢复了历史，不同 thread_id 仍然保持隔离。

9. 长期记忆: PostgreSQL

Checkpointer 保存的是 AgentState 的 checkpoint。长期偏好不应该依赖某个 Thread，因此需要使用 Store。

Store 中的一条记录由三部分定位：

• namespace：一组分层名称，用于隔离用户和业务范围。
• key：namespace 中某条数据的稳定名称。
• value：真正保存的字典数据。

下图中线程 A 和线程 B 的 thread_id 不同，但两次调用都通过 Runtime Context 提供 user-001，因此会定位到同一个 namespace 和 key。user-002 使用独立 namespace，不能读到 user-001 的偏好。

Store 不会自动保存聊天内容。只有 Tool 或节点显式调用 put()、get()、search() 或 delete() 时，长期记忆才会发生变化；thread_id 也不参与 Store 的定位。

9.1 操作长期记忆CRUD

长期记忆的增删改查操作

"""直接使用 PostgresStore 演示长期记忆的基础 CRUD 接口。"""

from langgraph.store.postgres import PostgresStore

DB_URI = (
    "postgresql://langgraph:langgraph_demo_1234@127.0.0.1:25432/"
    "langgraph_memory?sslmode=disable"
)
NAMESPACE = ("p09", "store-api", "user-001")
KEY = "preferences"


with PostgresStore.from_conn_string(DB_URI) as store:
    # 清理固定 key，保证每次都从相同初始状态开始。
    store.delete(NAMESPACE, KEY)

    # 第 1 步：put() 新增数据，get() 按 namespace 和 key 读取。
    store.put(NAMESPACE, KEY, {"activity": "周末去公园散步"})
    created = store.get(NAMESPACE, KEY)
    if created is None:
        print("新增后：没有找到记录")
    else:
        print(f"新增后：{created.value}")

    # 第 2 步：相同 namespace 和 key 再次 put 会更新原有数据。
    store.put(NAMESPACE, KEY, {"activity": "周末去西湖散步"})
    updated = store.get(NAMESPACE, KEY)
    if updated is None:
        print("更新后：没有找到记录")
    else:
        print(f"更新后：{updated.value}")

    # 第 3 步：没有配置向量索引时，search() 只列出 namespace 下的记录。
    items = store.search(NAMESPACE)
    print(f"namespace 中的记录数：{len(items)}")

    # 第 4 步：delete() 删除指定记录。
    store.delete(NAMESPACE, KEY)
    deleted = store.get(NAMESPACE, KEY)
    print(f"删除后：{deleted}")

运行结果：

新增后：{'activity': '周末去公园散步'}
更新后：{'activity': '周末去西湖散步'}
namespace 中的记录数：1
删除后：None

这里没有为 Store 配置向量索引，所以 search(NAMESPACE) 只是列出该 namespace 下的数据，不是基于 Embedding 的语义搜索。

9.2 写入长期偏好

真正交给 Agent 使用时，用户身份通过 Runtime Context 传入。Tool 从 runtime.context 读取 user_id，再使用 runtime.store 写入偏好。

本例统一使用下面的长期记忆位置：

namespace = ("users", user_id, "memories")
key = "preferences"

完整写入代码如下：

"""在线程 A 中通过 Tool 把用户偏好写入 PostgresStore。"""

from typing_extensions import TypedDict

import httpx
from langchain.agents import create_agent
from langchain.tools import ToolRuntime, tool
from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.utils.function_calling import convert_to_openai_tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore


DB_URI = (
    "postgresql://langgraph:langgraph_demo_1234@127.0.0.1:25432/"
    "langgraph_memory?sslmode=disable"
)
USER_ID = "user-001"
MEMORY_KEY = "preferences"
THREAD_IDS = ["p09-long-a", "p09-long-b", "p09-long-other"]


class UserContext(TypedDict):
    """定义每次 Agent 运行必须传入的用户身份。"""

    user_id: str


def memory_namespace(user_id: str) -> tuple[str, str, str]:
    """为每个用户生成相互隔离的长期记忆 namespace。"""

    # 不同 user_id 会生成不同路径，从而实现用户数据隔离。
    return ("users", user_id, "memories")


@tool
def save_user_preference(
    preference: str,
    runtime: ToolRuntime[UserContext],
) -> str:
    """保存当前用户的休闲偏好。"""

    if runtime.context is None:
        raise ValueError("必须通过 context 传入 user_id")

    # runtime.store 和 runtime.context 都由 Agent 自动注入。
    user_id = runtime.context["user_id"]
    runtime.store.put(
        memory_namespace(user_id),
        MEMORY_KEY,
        {"preference": preference},
    )
    return f"已保存偏好：{preference}"


# 第 1 步：创建模型。
model = ChatOpenAI(
    model="Qwen3-14B-AWQ-4bit-MLX",
    base_url="http://127.0.0.1:18080/v1",
    api_key="not-needed",
    temperature=0,
    max_tokens=192,
    http_client=httpx.Client(trust_env=False),
)


# 第 2 步：同时打开短期记忆 Checkpointer 和长期记忆 Store。
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    with PostgresStore.from_conn_string(DB_URI) as store:
        # 清理本示例使用的线程和长期记忆，保证结果可重复。
        for thread_id in THREAD_IDS:
            checkpointer.delete_thread(thread_id)
        store.delete(memory_namespace(USER_ID), MEMORY_KEY)

        # 第 3 步：将保存偏好的工具和 Store 一起交给 Agent。
        agent = create_agent(
            model=model,
            tools=[save_user_preference],
            checkpointer=checkpointer,
            store=store,
            context_schema=UserContext,
            system_prompt=(
                "你是一个用户偏好助手。"
            ),
        )
        result = agent.invoke(
            {
                "messages": [
                    {
                        "role": "user",
                        "content": "请记住：我喜欢周末去西湖散步。",
                    }
                ]
            },
            config={"configurable": {"thread_id": "p09-long-a"}},
            context={"user_id": USER_ID},
        )

        # 第 4 步：绕过模型直接读取 Store，确认工具确实写入了数据。
        saved_item = store.get(memory_namespace(USER_ID), MEMORY_KEY)


# 第 5 步：runtime 和 user_id 由框架注入，模型只需要生成 preference。
tool_schema = convert_to_openai_tool(save_user_preference)
visible_parameters = tool_schema["function"]["parameters"]["properties"]
print(f"模型可见参数：{list(visible_parameters)}")

for message in result["messages"]:
    if isinstance(message, AIMessage) and message.tool_calls:
        call = message.tool_calls[0]
        print(f"工具调用：{call['name']}，参数={call['args']}")
    elif isinstance(message, ToolMessage):
        print(f"工具结果：{message.content}")

if saved_item is None:
    print("Store 数据：没有找到记录")
else:
    print(f"Store 数据：{saved_item.value}")
print(f"最终回答：{result['messages'][-1].content}")

一次真实输出如下：

模型可见参数：['preference']
工具调用：save_user_preference，参数={'preference': '周末去西湖散步'}
工具结果：已保存偏好：周末去西湖散步
Store 数据：{'preference': '周末去西湖散步'}
最终回答：好的，已记住您喜欢在周末去西湖散步。如果您需要关于西湖的更多信息或建议，随时告诉我！

runtime 和 user_id 都不会出现在模型看到的 Tool Schema 中。模型只负责提取 preference，用户身份由应用通过 Context 提供，避免模型自行编造或修改用户 ID。

9.3 读取长期记忆

写入完成后，再次重启 PostgreSQL，然后在新的 Python 进程中使用线程 B 读取数据。

读取 Tool 根据 user_id 生成相同 namespace。只要用户相同，即使 thread_id 不同，也可以取回同一份长期记忆。

"""在线程 B 中读取同一用户的长期记忆，并验证用户隔离。"""

from typing_extensions import TypedDict

import httpx
from langchain.agents import create_agent
from langchain.tools import ToolRuntime, tool
from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.utils.function_calling import convert_to_openai_tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore


DB_URI = (
    "postgresql://langgraph:langgraph_demo_1234@127.0.0.1:25432/langgraph_memory?sslmode=disable"
)
MEMORY_KEY = "preferences"


class UserContext(TypedDict):
    """定义每次 Agent 运行必须传入的用户身份。"""

    user_id: str


def memory_namespace(user_id: str) -> tuple[str, str, str]:
    """为每个用户生成相互隔离的长期记忆 namespace。"""

    # 读取时必须使用与写入脚本完全相同的 namespace 规则。
    return ("users", user_id, "memories")


@tool
def get_user_preference(runtime: ToolRuntime[UserContext]) -> str:
    """读取当前用户已经保存的休闲偏好。"""

    if runtime.context is None:
        raise ValueError("必须通过 context 传入 user_id")

    # 工具根据当前 Context 选择 namespace，模型不能自行指定 user_id。
    user_id = runtime.context["user_id"]
    item = runtime.store.get(memory_namespace(user_id), MEMORY_KEY)
    if item is None:
        return "没有保存过偏好"
    return str(item.value["preference"])


# 第 1 步：创建模型。
model = ChatOpenAI(
    model="Qwen3-14B-AWQ-4bit-MLX",
    base_url="http://127.0.0.1:18080/v1",
    api_key="not-needed",
    temperature=0,
    max_tokens=192,
    http_client=httpx.Client(trust_env=False),
)


def print_result(label: str, result: dict) -> None:
    """打印稳定的工具信息和最终回答，不输出动态 Tool Call ID。"""

    print(label)
    for message in result["messages"]:
        if isinstance(message, AIMessage) and message.tool_calls:
            print(f"工具调用：{message.tool_calls[0]['name']}")
        elif isinstance(message, ToolMessage):
            print(f"工具结果：{message.content}")
    print(f"最终回答：{result['messages'][-1].content}\n")


# 第 2 步：打开同一个 PostgreSQL 中的 Checkpointer 和 Store。
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    with PostgresStore.from_conn_string(DB_URI) as store:
        # 清理两个读取线程，但不删除 user-001 在线程 A 保存的 Store 数据。
        checkpointer.delete_thread("p09-long-b")
        checkpointer.delete_thread("p09-long-other")
        store.delete(memory_namespace("user-002"), MEMORY_KEY)

        agent = create_agent(
            model=model,
            tools=[get_user_preference],
            checkpointer=checkpointer,
            store=store,
            context_schema=UserContext,
            system_prompt=(
                "你是一个用户偏好助手。"
            ),
        )

        # 第 3 步：新线程仍可读取同一用户的长期记忆。
        same_user = agent.invoke(
            {"messages": [{"role": "user", "content": "你还记得我的休闲偏好吗？"}]},
            config={"configurable": {"thread_id": "p09-long-b"}},
            context={"user_id": "user-001"},
        )
        # 第 4 步：另一个用户使用不同 namespace，读取不到 user-001 的记忆。
        other_user = agent.invoke(
            {"messages": [{"role": "user", "content": "你还记得我的休闲偏好吗？"}]},
            config={"configurable": {"thread_id": "p09-long-other"}},
            context={"user_id": "user-002"},
        )


tool_schema = convert_to_openai_tool(get_user_preference)
visible_parameters = tool_schema["function"]["parameters"]["properties"]
print(f"模型可见参数：{visible_parameters}\n")
print_result("同一用户的新线程：", same_user)
print_result("另一用户的线程：", other_user)

一次真实输出如下：

模型可见参数：{}

同一用户的新线程：
工具调用：get_user_preference
工具结果：周末去西湖散步
最终回答：是的，我记得你喜欢周末去西湖散步。如果你需要关于西湖的更多信息或者有什么其他需求，随时告诉我！

另一用户的线程：
工具调用：get_user_preference
工具结果：没有保存过偏好
最终回答：看来还没有保存过您的休闲偏好呢！您可以告诉我您喜欢的休闲活动，比如阅读、运动、看电影等，这样我就能更好地为您推荐相关内容了。您愿意分享一下吗？

这是一次真实模型输出，模型每次生成的措辞可能略有不同，但下面三个行为应保持稳定：

• user-001 在线程 B 中能够读取线程 A 保存的偏好。
• user-002 无法读取 user-001 的 namespace。
• 模型可见参数为空，user_id 和 runtime 没有暴露给模型。

10. 查看真实记忆数据

为了确认记忆确实进入数据库，可以查看数据表记录数、演示 Thread 和长期记忆。代码不打印 checkpoint UUID、二进制内容或时间戳。

"""检查 PostgreSQL 中 Checkpointer 和 Store 的表与演示数据。"""

import psycopg

DB_URI = (
    "postgresql://langgraph:langgraph_demo_1234@127.0.0.1:25432/langgraph_memory?sslmode=disable"
)

# 表名来自 PostgreSQL Checkpointer 的迁移定义。
TABLES = [
    "checkpoint_migrations",
    "checkpoints",
    "checkpoint_blobs",
    "checkpoint_writes",
    "store_migrations",
    "store",
]

# 第 1 步：连接数据库并查看六张核心表的记录数。
with psycopg.connect(DB_URI) as connection:
    with connection.cursor() as cursor:
        print("数据表记录数：")
        for table_name in TABLES:
            # 表名来自上面的固定白名单，不接收外部输入。
            cursor.execute(f'SELECT COUNT(*) FROM "{table_name}"')
            count = cursor.fetchone()[0]
            print(f"- {table_name}: {count}")

        # 第 2 步：只查询本系列创建的 p09 演示线程。
        cursor.execute(
            """
            SELECT DISTINCT thread_id
            FROM checkpoints
            WHERE thread_id LIKE 'p09-%'
            ORDER BY thread_id
            """
        )
        thread_ids = [row[0] for row in cursor.fetchall()]

        # 第 3 步：查询 users namespace 下的长期记忆。
        cursor.execute(
            """
            SELECT prefix, key, value
            FROM store
            WHERE prefix LIKE 'users.%'
            ORDER BY prefix, key
            """
        )
        memories = cursor.fetchall()

print(f"演示线程：{thread_ids}")
print("长期记忆：")
for prefix, key, value in memories:
    print(f"- namespace={prefix}, key={key}, value={value}")

本次实测输出：

数据表记录数：
- checkpoint_migrations: 10
- checkpoints: 24
- checkpoint_blobs: 30
- checkpoint_writes: 30
- store_migrations: 4
- store: 1
演示线程：['p09-long-a', 'p09-long-b', 'p09-long-other', 'p09-short-001', 'p09-short-other']
长期记忆：
- namespace=users.user-001.memories, key=preferences, value={'preference': '周末去西湖散步'}

checkpoint 记录数会受到模型执行步骤和重复测试次数影响，不应该把 24 或 30 当作固定结果。真正需要检查的是目标 Thread 存在、Store 中存在正确的 namespace、key 和 value。

在 PostgreSQL Checkpointer 的数据库表中，概念上的 namespace 存储在 store.prefix 列中。代码使用 PostgresStore API 时仍然传入 tuple namespace，不应该因为底层列名是 prefix 就改变应用层写法。

下图把运行时调用链与数据库中的两类存储放在一起。PostgresSaver 和 PostgresStore 可以连接同一个 PostgreSQL 实例，但它们使用不同的 API 和表组。

Checkpointer 会在图执行步骤完成后自动保存 checkpoint；Store 只会在业务代码显式调用时读写数据。两类数据不会自动互相复制，也不能使用其中一套 API 查询另一套数据。

11. 怎么选择

短期记忆和长期记忆都可以保存在 PostgreSQL 中，但“短期”和“长期”描述的不是数据在数据库中保存多久，而是数据的作用范围。即使 checkpoint 一直没有删除，它仍然是某个 Thread 的短期记忆；Store 中的数据即使配置了 TTL，只要用于跨 Thread 共享，仍然属于长期记忆。

11.1 两类记忆的区别

短期记忆由 Checkpointer 管理，解决的是“怎样继续当前会话”。它按照 thread_id 保存图的 checkpoint，包括消息、AgentState、channel 版本、节点中间写入和执行位置等信息。再次使用相同 thread_id 时，LangGraph 会自动恢复已有 State，因此它除了让 Agent 记住前文，还能支持失败重试、中断恢复和状态回放。

长期记忆由 Store 管理，解决的是“怎样在不同会话中继续使用某项信息”。它按照 namespace + key 保存独立的结构化数据，例如用户资料、偏好、账户设置或业务规则。Store 不依赖 thread_id，只要使用相同的 namespace 和 key，就可以在线程 A 写入、在线程 B 读取。

对比项	短期记忆：Checkpointer	长期记忆：Store
主要目的	延续和恢复当前会话的运行状态	保存跨会话使用的用户或业务信息
数据边界	thread_id	namespace + key
典型内容	messages、AgentState、checkpoint、pending writes	用户资料、偏好、规则、摘要
写入方式	LangGraph 在图执行过程中自动保存	业务代码或 Tool 显式调用 put()
读取方式	使用相同 thread_id 时自动恢复	业务代码或 Tool 显式调用 get()、search()
是否跨 Thread	默认不跨 Thread	可以跨 Thread
典型能力	多轮对话、重试、回放、中断恢复	用户画像、个性化、跨会话共享、业务检索

get_state_history() 返回的是同一 Thread 的 checkpoint 历史，用于查看 State 怎样随执行过程变化。它不会查询 Store 中的长期记忆。长期记忆必须使用 Store 的 get()、put()、search() 和 delete() 等接口。

11.2 根据数据的使用范围选择

选择时可以先判断数据以后需要在哪里使用。

如果数据只用于继续当前 Thread，应当交给 Checkpointer。例如：

• 当前对话中的用户消息和模型回答。
• Tool Call、ToolMessage 和其他 AgentState。
• 图运行到哪个节点以及接下来执行什么。
• interrupt 之后需要恢复的执行现场。

如果数据需要脱离当前 Thread，在新 Thread 中继续使用，应当写入 Store。例如：

• 用户姓名、语言和输出格式偏好。
• 用户明确要求 Agent 记住的事项。
• 跨会话复用的对话摘要。
• 账户设置、业务规则或应用配置。
• 需要按照用户、组织或业务对象检索的数据。

还可以使用下面三个问题快速判断：

1. 换一个 thread_id 后仍然需要读取吗？需要就使用 Store。
2. 需要恢复图的执行状态、节点结果或消息历史吗？需要就使用 Checkpointer。
3. 两种需求是否同时存在？同时存在就同时配置 Checkpointer 和 Store。

如果应用是一次性的无状态调用，既不需要继续会话，也不需要跨会话保存数据，那么两种组件都可以不配置。

11.3 不要把全部聊天都写入长期记忆

配置 Store 后，LangGraph 不会自动把每一句对话写成长期记忆。应用必须明确决定保存什么、使用什么 namespace 和 key，以及何时更新或删除。

例如“你好”“谢谢”“换一种表达”等临时消息只对当前对话有意义，保存在短期记忆中即可。如果把完整聊天无差别地复制到 Store，不仅会造成数据膨胀，还会让后续检索混入大量无关信息。

本文由 save_user_preference Tool 显式调用 runtime.store.put()，只保存用户要求记住的休闲偏好。这样既保留了 Thread 中完整的对话过程，又将真正需要跨 Thread 使用的信息提取为长期记忆。

11.4 实际项目通常同时使用

短期记忆和长期记忆不是互斥方案。一个需要连续对话和用户个性化的 Agent，通常会同时配置 Checkpointer 和 Store：

agent = create_agent(
    model=model,
    tools=tools,
    checkpointer=checkpointer,
    store=store,
    context_schema=UserContext,
)

此时一次调用的数据流可以概括为：

1. Checkpointer 根据 thread_id 恢复当前会话的 State。
2. Agent 根据 user_id 生成 namespace，并从 Store 读取与当前任务有关的长期记忆。
3. Agent 执行过程中产生的新消息和状态继续由 Checkpointer 自动保存。
4. 只有值得跨会话复用的信息，才由 Tool 或业务代码整理后写入 Store。

因此，最实用的选择原则是：会话现场交给 Checkpointer，跨会话知识交给 Store。

12. 小结

短期记忆和长期记忆解决的是不同问题。

Checkpointer 让同一个 Thread 在多次调用之间恢复 AgentState。InMemorySaver 适合进程内开发测试，PostgresSaver 可以跨进程和数据库重启恢复。

Store 用于保存独立于 Thread 的业务数据。本文使用 user_id 生成 namespace，在线程 A 写入偏好，再在线程 B 读取，并验证另一个用户不能读取该数据。

用户身份通过 Runtime Context 传入，执行中的消息和状态保存在 AgentState，线程内历史交给 Checkpointer，跨线程偏好交给 Store。把这几种机制分开，Agent 的数据边界才会清楚。