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，保持前面文章的 LangChain 和 LangGraph 版本不变，只增加 PostgreSQL Checkpointer 相关依赖。

langchain==1.2.13
langchain-core==1.2.22
langchain-openai==1.1.12
langgraph==1.1.3
openai==2.30.0
httpx==0.28.1
langgraph-checkpoint-postgres==3.0.5
psycopg[binary]==3.3.3
psycopg-pool==3.3.0

安装锁定依赖：

cd source/_posts/llm_learning
source .venv_langgraph/bin/activate

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

实测 pip check 输出：

No broken requirements found.

5. 启动本地 Qwen3

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

source .venv_tool_server/bin/activate

"$VIRTUAL_ENV/bin/python" -m mlx_lm server \
  --model model \
  --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


model = ChatOpenAI(
    model="default_model",
    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),
)


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


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 也需要自己的数据表。

两套组件必须分别执行一次 setup()：

"""初始化 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"
)


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

with PostgresStore.from_conn_string(DB_URI) as store:
    store.setup()


# 只展示稳定的表名，不输出迁移时间或其他动态字段。
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 数据。

8. 使用 PostgresSaver 写入短期记忆

写入脚本使用固定线程 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"


model = ChatOpenAI(
    model="default_model",
    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),
)


with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # 只清理当前示例线程，保证脚本重复执行时不会累积旧消息。
    checkpointer.delete_thread(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。

9. 重启数据库后恢复短期记忆

先停止再启动 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"


model = ChatOpenAI(
    model="default_model",
    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),
)


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=(
            "你是一个中文助手。只根据当前线程的历史消息回答。"
            "如果历史消息中没有用户姓名，只回答：不知道。"
        ),
    )

    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 仍然保持隔离。

10. PostgresStore 的 namespace、key 和 value

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 的定位。

先直接调用 PostgresStore 的基础接口：

"""直接使用 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)

    store.put(NAMESPACE, KEY, {"activity": "周末去公园散步"})
    created = store.get(NAMESPACE, KEY)
    print(f"新增后：{created.value if created else None}")

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

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

    store.delete(NAMESPACE, KEY)
    deleted = store.get(NAMESPACE, KEY)
    print(f"删除后：{deleted}")

运行结果：

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

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

11. 在线程 A 写入长期偏好

真正交给 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。"""

    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")

    user_id = runtime.context["user_id"]
    runtime.store.put(
        memory_namespace(user_id),
        MEMORY_KEY,
        {"preference": preference},
    )
    return f"已保存偏好：{preference}"


model = ChatOpenAI(
    model="default_model",
    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),
)


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)

        agent = create_agent(
            model=model,
            tools=[save_user_preference],
            checkpointer=checkpointer,
            store=store,
            context_schema=UserContext,
            system_prompt=(
                "你是一个用户偏好助手。"
                "用户要求记住休闲偏好时，必须调用 save_user_preference 工具。"
            ),
        )
        result = agent.invoke(
            {
                "messages": [
                    {
                        "role": "user",
                        "content": "请记住：我喜欢周末去西湖散步。",
                    }
                ]
            },
            config={"configurable": {"thread_id": "p09-long-a"}},
            context={"user_id": USER_ID},
        )

        saved_item = store.get(memory_namespace(USER_ID), MEMORY_KEY)


# 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}")

print(f"Store 数据：{saved_item.value if saved_item else None}")
print(f"最终回答：{result['messages'][-1].content}")

一次真实输出如下：

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

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

12. 在线程 B 读取同一用户的偏好

写入完成后，再次重启 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。"""

    return ("users", user_id, "memories")


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

    if runtime.context is None:
        raise ValueError("必须通过 context 传入 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"])


model = ChatOpenAI(
    model="default_model",
    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")


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=(
                "你是一个用户偏好助手。"
                "用户询问休闲偏好时，必须调用 get_user_preference 工具，"
                "并根据工具结果回答。"
            ),
        )

        same_user = agent.invoke(
            {"messages": [{"role": "user", "content": "你还记得我的休闲偏好吗？"}]},
            config={"configurable": {"thread_id": "p09-long-b"}},
            context={"user_id": "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 没有暴露给模型。

13. 查看 PostgreSQL 中的真实数据

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

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

import psycopg


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

# 表名来自 langgraph-checkpoint-postgres==3.0.5 的迁移定义。
TABLES = [
    "checkpoint_migrations",
    "checkpoints",
    "checkpoint_blobs",
    "checkpoint_writes",
    "store_migrations",
    "store",
]


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}")

        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()]

        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。

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

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

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

14. 三个最容易混淆的边界

14.1 Checkpointer 不等于 Store

PostgresSaver 保存某个 Thread 的 State checkpoint。它使同一会话能够恢复消息和状态。

PostgresStore 保存独立于 Thread 的业务数据。它适合用户偏好、账户设置和其他需要跨线程读取的信息。

两者是独立接口，可以只配置其中一个，也可以像本文一样同时配置：

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

14.2 get_state_history() 不是长期记忆查询

get_state_history() 返回的是同一 Thread 的 checkpoint 历史，用于检查 State 怎样随执行过程变化。它不会读取 PostgresStore 中的用户偏好。

长期记忆必须使用 Store 的 get()、put()、search() 和 delete() 等接口。

14.3 Store 不会自动保存全部聊天

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

本文由 save_user_preference Tool 显式调用 runtime.store.put()，只保存用户要求记住的休闲偏好。这样比把完整聊天无差别写入长期存储更容易控制数据范围。

15. 常见问题

15.1 相同用户却读不到短期记忆

短期记忆按 thread_id 隔离，不按 user_id 隔离。读取时必须使用写入时相同的 thread_id。

15.2 新线程读不到长期记忆

检查写入和读取是否使用完全相同的 namespace 与 key：

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

tuple 中任意一段不同，都会进入另一个命名空间。

15.3 忘记传递 Runtime Context

本文的 Tool 通过 runtime.context[“user_id”] 确定用户身份，调用 Agent 时必须传入：

context={"user_id": "user-001"}

用户身份不应该让模型从提示词中猜测，也不应该作为可变 AgentState 让模型随意修改。

15.4 忘记执行 setup()

第一次使用 PostgreSQL Checkpointer 和 Store 前，必须分别执行 setup()。否则会因为数据表不存在而失败。

重复执行 setup() 是安全的。它会检查迁移状态，不会删除已有记忆。

15.5 误以为必须安装 pgvector

本文只使用普通 key-value Store，没有配置向量索引，不需要 pgvector、Embedding 或向量数据库。

15.6 数据库重启后数据丢失

确认 Compose 使用了命名数据卷，并且停止时执行的是 down，不是 down -v。后者会删除数据卷。

15.7 本地请求受到代理影响

先使用下面的命令检查 Qwen3：

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

示例中的 httpx.Client(trust_env=False) 用于避免 VPN 或系统代理转发本机模型请求。

16. 小结

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

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

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

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

对象或机制	主要保存内容	定位方式	生命周期与范围	是否自动持久化
Runtime Context	用户 ID、租户和权限等调用前已确定的数据	调用时传入 context	单次运行，按调用隔离	否
AgentState	消息和 Agent 执行过程中变化的数据	当前 Graph 运行	单次运行；可由 Checkpointer 保存	否
thread_id	某个会话的稳定标识	configurable.thread_id	同一 Thread 的多次调用	它是定位键，不是存储组件
InMemorySaver	Thread 的 State checkpoint	thread_id	当前 Python 进程	是，但只在进程内
PostgresSaver	Thread 的 State checkpoint	thread_id	跨进程、跨数据库重启	是
PostgresStore	用户偏好等跨线程业务数据	namespace + key	可跨 Thread，范围由 namespace 决定	是
namespace	长期记忆的分层业务范围	例如 (“users”, user_id, “memories”)	通常按用户、组织或业务域隔离	它是定位键，不是存储组件

LangGraph 系列前九篇到这里完成了 Agent 主线。下一篇开始进入 Workflow 主线，继续学习 State、Reducer、固定路由、条件分支、循环与人工介入。