LangGraph Agent 记忆系统实战：从对话上下文到长期知识库

一、Agent 为什么会"失忆"？

你跟 Agent 说"我叫张三"，它回了句"你好张三"。过两秒你再问"我叫什么"——它说不知道。

第1轮：用户:"我叫张三" -> Agent:"你好张三!"
第2轮：用户:"我叫什么名字?" -> Agent:"抱歉,我不知道你叫什么名字。"

这不是 Bug。LangChain 的 Chain 模式下，每次 invoke 都是独立的"感知→推理→行动"循环，上一轮的对话不会自动带到下一轮。Agent 天生没记忆。

要让它"记住"东西，你得手动装两套记忆系统：

• 短期记忆（Checkpointer）：记住"这轮对话里说了什么"
• 长期记忆（Store）：记住"这个用户是谁"，哪怕下次开个新对话也认得

---

二、两套记忆系统，别搞混了

2.1 一张表说清楚

	短期记忆	长期记忆
核心问题	“刚才我们聊了什么？”	“这个用户上周告诉我什么？”
作用范围	单一会话（thread）内	跨会话、跨 thread、永久
存储内容	消息历史、中间状态、工具调用结果	用户画像、偏好、事实、经验规则
LangGraph API	checkpointer 参数	store 参数（BaseStore）
隔离维度	thread_id	namespace（命名空间元组）
典型后端	InMemory / SQLite / Postgres / Redis	InMemoryStore / PostgresStore / MongoDB
类比人类	工作记忆（当下聊的内容）	长期知识（记得客户名字和喜好）

关键洞察：短期记忆管"对话连续性"，长期记忆管"用户认知"。生产级 Agent 通常两者都开。

2.2 认知科学视角：三种长期记忆

LangGraph 借鉴认知科学，把长期记忆分成三类：

记忆类型	存什么	人类例子	Agent 例子
语义记忆	事实、偏好、知识	我学过的东西	用户喜欢 Python、住在上海
情景记忆	经历、交互案例	我做过的事	上次解决 X 问题用了 Y 方法
程序记忆	规则、指令、工作流	本能或技能	Agent 自我更新的系统提示词

短期记忆（Checkpointer）也能临时承载这三类信息，但一旦 thread_id 更换或进程重启就没了。只有写入 Store 的长期记忆才能保证"下次见面还记得"。

2.3 代码对比：同一个场景，两种结局

场景：用户 Alice 周一说"我叫 Alice，我喜欢 Python"。周三她用新对话问"我上次说我喜欢什么？"

只用 Checkpointer — 周三失忆

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import HumanMessage

checkpointer = InMemorySaver()

builder = StateGraph(MessagesState)
builder.add_node("agent", lambda state: {
    "messages": [HumanMessage(content=f"收到: {state['messages'][-1].content}")]
})
builder.add_edge(START, "agent")
builder.add_edge("agent", END)

app = builder.compile(checkpointer=checkpointer)

# 周一
config_mon = {"configurable": {"thread_id": "alice-monday"}}
app.invoke(
    {"messages": [HumanMessage(content="我叫 Alice，我喜欢 Python")]},
    config=config_mon
)

# 周三（换了 thread_id）
config_wed = {"configurable": {"thread_id": "alice-wednesday"}}
result = app.invoke(
    {"messages": [HumanMessage(content="我上次说我喜欢什么？")]},
    config=config_wed
)
# Agent："抱歉，我不知道你在说什么。"
# thread_id 变了，checkpointer 加载不到周一的历史

加上 Store — 周三仍然记得

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
import re

checkpointer = InMemorySaver()
store = InMemoryStore()  # 长期记忆
llm = ChatOpenAI(model="gpt-4o-mini")

def agent_node(state: MessagesState, *, store):
    user_id = "alice"
    # 从 Store 读取长期记忆
    memories = store.search(namespace=("users", user_id))
    memory_text = "\n".join(f"- {m.key}: {m.value}" for m in memories)
    system_msg = SystemMessage(content=f"用户已知信息：\n{memory_text}")
    response = llm.invoke([system_msg] + state["messages"])
    return {"messages": [response]}

def save_memory_node(state: MessagesState, *, store):
    user_id = "alice"
    last_msg = state["messages"][-1].content
    # 从回复中提取 [MEMORY: key=value] 写入 Store
    for key, value in re.findall(r"\[MEMORY:\s*(\w+)=(.+?)\]", last_msg):
        store.put(("users", user_id), key, {"fact": value.strip()})
    return {"messages": []}

builder = StateGraph(MessagesState)
builder.add_node("agent", agent_node)
builder.add_node("save_memory", save_memory_node)
builder.add_edge(START, "agent")
builder.add_edge("agent", "save_memory")
builder.add_edge("save_memory", END)

# 关键：同时传入 checkpointer 和 store
app = builder.compile(checkpointer=checkpointer, store=store)

# 周一（thread_id = session-1）
config1 = {"configurable": {"thread_id": "session-1", "user_id": "alice"}}
app.invoke(
    {"messages": [HumanMessage(
        content="我叫 Alice，我喜欢 Python [MEMORY: name=Alice] [MEMORY: language=Python]")]},
    config=config1
)

# 周三（全新 thread_id = session-2）
config2 = {"configurable": {"thread_id": "session-2", "user_id": "alice"}}
result = app.invoke(
    {"messages": [HumanMessage(content="我上次说我喜欢什么？")]},
    config=config2
)
# Agent："你上次说你喜欢 Python。"
# thread_id 变了（短期记忆清空），但 Store 里的长期记忆还在

一眼看出核心区别：

• checkpointer 按 thread_id 隔离，换 thread 就清空
• store 按 namespace 隔离（如 ("users", "alice")），换 thread 不影响

2.4 决策树：什么时候用哪个？

需要记住"本轮对话"的上下文？
  ├─ 是 → 用 Checkpointer
  │       └─ 对话太长导致 Token 爆炸？
  │           ├─ 是 → 加 SummarizationMiddleware 或滑动窗口
  │           └─ 否 → 直接用
  │
  └─ 需要记住"这个用户是谁"？
      ├─ 是 → 用 Store
      │       └─ 需要语义搜索（模糊匹配）？
      │           ├─ 是 → Store 配置 embedding index
      │           └─ 否 → 普通 KV Store 即可
      └─ 否 → 只用 Checkpointer 够了

生产建议：绝大多数生产级 Agent 应该同时启用两者。

---

三、短期记忆：Checkpointer

3.1 原理：自动存档

Checkpointer 在每个节点执行后自动保存当前图的完整状态（messages、tool_calls、intermediate_steps 等所有通道数据）。下次调用时自动加载历史，拼到新输入前面。

调用结束 → 自动存档（Checkpoint）
调用开始 → 自动读档，拼接到新输入
效果：LLM 看到的消息 = 历史消息 + 本次新消息

3.2 thread_id：会话隔离

同时服务多个用户时，用 thread_id 隔离对话历史：

thread_id: "user_张三" → [消息1, 消息2, ...]
thread_id: "user_李四" → [消息A, 消息B, ...]

3.3 四种存储后端

同一套代码，无缝切换持久化方案：

后端	导入路径	持久性	适用场景	并发
InMemorySaver	langgraph.checkpoint.memory	进程结束即丢	开发/测试	单进程
SqliteSaver	langgraph.checkpoint.sqlite	本地文件	单机原型	单进程
PostgresSaver	langgraph.checkpoint.postgres	完整 ACID	生产/分布式	高并发
RedisSaver	社区包 langgraph-checkpoint-redis	依赖 RDB/AOF	短会话/缓存	高并发

A. InMemorySaver（开发环境）

from langgraph.checkpoint.memory import InMemorySaver

memory = InMemorySaver()
app = builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke(
    {"messages": [("user", "今天北京天气怎么样？")]},
    config=config
)
# 同一个 thread_id 再调用 → 有记忆
result2 = app.invoke(
    {"messages": [("user", "我刚才问你什么了？")]},
    config=config
)

InMemorySaver 是 MemorySaver 的别名，新代码统一用 InMemorySaver。

B. SqliteSaver（单机原型）

pip install langgraph-checkpoint-sqlite>=3.0.1

from langgraph.checkpoint.sqlite import SqliteSaver

# 同步
with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    app = builder.compile(checkpointer=checkpointer)
    result = app.invoke(
        {"messages": [("user", "你好")]},
        config={"configurable": {"thread_id": "session-001"}}
    )

# 异步（推荐）
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

async with AsyncSqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    app = builder.compile(checkpointer=checkpointer)
    result = await app.ainvoke(
        {"messages": [("user", "你好")]},
        config={"configurable": {"thread_id": "session-001"}}
    )

C. PostgresSaver（生产环境）

pip install langgraph-checkpoint-postgres>=3.1.0

import asyncio
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from psycopg.rows import dict_row

async def main():
    conn = "postgresql://user:pass@localhost:5432/agentdb"

    async with await AsyncPostgresSaver.from_conn_string(
        conn,
        conn_kwargs={
            "autocommit": True,
            "row_factory": dict_row,
            "prepare_threshold": None,  # 用 PgBouncer 必须设
        }
    ) as checkpointer:
        await checkpointer.setup()  # 首次必须调用，建表
        app = builder.compile(checkpointer=checkpointer)

        result = await app.ainvoke(
            {"messages": [("user", "帮我查北京天气")]},
            config={"configurable": {"thread_id": "user-alice-001"}}
        )

asyncio.run(main())

D. RedisSaver（分布式缓存）

pip install langgraph-checkpoint-redis>=1.0.2

from langgraph.checkpoint.redis import RedisSaver

with RedisSaver.from_conn_string("redis://localhost:6379/0") as checkpointer:
    app = builder.compile(checkpointer=checkpointer)
    result = app.invoke(
        {"messages": [("user", "你好")]},
        config={"configurable": {"thread_id": "session-redis-001"}}
    )

数据库表结构（SQLite / PostgreSQL 共用）

Checkpointer 自动创建三张表。SQLite 用 BLOB，PostgreSQL 用 BYTEA，结构一致：

-- 状态快照（每个超级步骤一行）
CREATE TABLE checkpoints (
    thread_id TEXT NOT NULL,
    checkpoint_ns TEXT NOT NULL DEFAULT '',
    checkpoint_id TEXT NOT NULL,
    parent_checkpoint_id TEXT,
    type TEXT,
    checkpoint BLOB,     -- PostgreSQL: BYTEA。msgpack 序列化的状态数据
    metadata BLOB,       -- PostgreSQL: BYTEA
    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
);

-- 任务写入的通道数据
CREATE TABLE checkpoint_writes (
    thread_id TEXT NOT NULL,
    checkpoint_ns TEXT NOT NULL DEFAULT '',
    checkpoint_id TEXT NOT NULL,
    task_id TEXT NOT NULL,
    idx INTEGER NOT NULL,
    channel TEXT NOT NULL,
    type TEXT,
    value BLOB,          -- PostgreSQL: BYTEA
    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
);

-- 大体积二进制数据
CREATE TABLE checkpoint_blobs (
    thread_id TEXT NOT NULL,
    checkpoint_ns TEXT NOT NULL DEFAULT '',
    checkpoint_id TEXT NOT NULL,
    channel TEXT NOT NULL,
    version TEXT NOT NULL,
    type TEXT,
    value BLOB,          -- PostgreSQL: BYTEA
    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, channel, version)
);

存储空间估算：

• checkpoints 行数 ≈ (超级步骤数 + 1) × 命名空间数
• 对于追加型通道（如 messages），存储空间与轮次呈二次方增长——对话翻倍，存储约翻四倍

---

四、长期记忆：Store

Store 是 LangGraph 的长期记忆存储抽象接口，用来存放跨会话的数据。
不受 thread 限制，不同会话可以读取同一份记忆。

• 保存 Agent 的长期记忆：用户偏好、历史经验、事实、知识库片段。
• 通过 namespace（命名空间）做数据隔离，一般用 (user_id, xxx)，实现按用户隔离记忆。
• PostgresStore 实现还可以叠加向量检索，做语义记忆召回。

Checkpointer 绑定 thread_id，只存这一次会话的运行快照，会话之间互相隔离。

4.1 核心概念

Store 就是"长期知识库"，

Store 通过 BaseStore 接口提供四个核心操作：

操作	方法	说明
写入	put(namespace, key, value)	存储或更新一条记忆
精确读取	get(namespace, key)	按命名空间 + 键读取
语义搜索	search(namespace, query)	按相似度搜索（需配置 embedding）
删除	delete(namespace, key)	删除一条记忆

命名空间（namespace） 是字符串元组，类似文件路径，用于隔离不同用户/类别的记忆：

("users", "alice", "preferences")   # Alice 的偏好
("users", "bob", "facts")           # Bob 的事实
("global", "company_policy")        # 全局公司政策

4.2 后端选择

后端	导入路径	语义搜索	适用场景
InMemoryStore	langgraph.store.memory	需配置 index	本地开发
PostgresStore	langgraph.store.postgres	需 pgvector	生产环境
MongoDB Store	langchain-mongodb	需 Atlas Vector Search	云原生

4.3 代码实战

基础版：InMemoryStore

from langgraph.store.memory import InMemoryStore

store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": "openai:text-embedding-3-small",
    }
)

# 写入
store.put(("users", "alice"), "preference", {"language": "Python", "framework": "FastAPI"})

# 精确读取
item = store.get(("users", "alice"), "preference")
print(item.value)  # {'language': 'Python', 'framework': 'FastAPI'}

# 语义搜索
results = store.search(("users", "alice"), query="她喜欢什么编程语言？")
for r in results:
    print(r.key, r.value)

生产版：PostgresStore

from langgraph.store.postgres import PostgresStore

store = PostgresStore.from_conn_string(
    "postgresql://user:pass@localhost:5432/agentdb",
    index={"dims": 1536, "embed": "openai:text-embedding-3-small"}
)
store.setup()  # 建表

# API 与 InMemoryStore 完全一致
store.put(("users", "alice"), "profile", {"name": "Alice", "role": "工程师"})
memories = store.search(("users", "alice"), query="用户的职业是什么？")

与 Agent 集成

from langgraph.prebuilt import create_react_agent
from langgraph.store.memory import InMemoryStore
from langgraph.checkpoint.memory import InMemorySaver
from langchain.chat_models import init_chat_model
from langmem import create_manage_memory_tool, create_search_memory_tool

store = InMemoryStore(index={"dims": 1536, "embed": "openai:text-embedding-3-small"})

# LangMem SDK 提供的记忆管理工具
manage_memory = create_manage_memory_tool(namespace=("users", "alice"))
search_memory = create_search_memory_tool(namespace=("users", "alice"))

# 同时传入 checkpointer + store
agent = create_react_agent(
    model=init_chat_model("gpt-4o-mini"),
    tools=[manage_memory, search_memory],
    store=store,
    checkpointer=InMemorySaver(),
)
# Agent 现在同时拥有：
# 1. Checkpointer：记住本轮对话上下文
# 2. Store：记住 Alice 的长期偏好和事实

4.4 存储结构

与 Checkpointer 的 checkpoints 表不同，Store 存的是结构化 JSON + 可选向量：

-- PostgresStore 表结构（简化）
CREATE TABLE store (
    prefix TEXT NOT NULL,       -- namespace 序列化
    key TEXT NOT NULL,
    value JSONB,
    created_at TIMESTAMP,
    updated_at TIMESTAMP,
    PRIMARY KEY (prefix, key)
);

CREATE TABLE store_vectors (
    prefix TEXT NOT NULL,
    key TEXT NOT NULL,
    embedding VECTOR(1536),     -- pgvector 扩展
    text TEXT,
    PRIMARY KEY (prefix, key)
);

一句话区分：Checkpointer 存的是完整的图状态快照（msgpack 二进制），Store 存的是结构化 JSON 文档 + 向量 embedding。

---

五、从开发到生产：三阶段配置

阶段一：开发（InMemorySaver + InMemoryStore）

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langgraph.graph import StateGraph, MessagesState

checkpointer = InMemorySaver()
store = InMemoryStore()

app = builder.compile(checkpointer=checkpointer, store=store)
config = {"configurable": {"thread_id": "test-123", "user_id": "alice"}}

进程重启全丢，但代码最简单，适合本地调试。

阶段二：原型（SqliteSaver + InMemoryStore）

from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("prototype.db") as checkpointer:
    store = InMemoryStore()
    app = builder.compile(checkpointer=checkpointer, store=store)

短期记忆落盘（重启不丢），长期记忆仍在内存。适合单机小工具。

阶段三：生产（AsyncPostgresSaver + PostgresStore）

import asyncio
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.store.postgres import PostgresStore
from psycopg.rows import dict_row

async def main():
    conn = "postgresql://agent_user:secret@postgres.internal:5432/agent_db"

    async with await AsyncPostgresSaver.from_conn_string(
        conn,
        conn_kwargs={"autocommit": True, "row_factory": dict_row, "prepare_threshold": None}
    ) as checkpointer:
        await checkpointer.setup()

        store = PostgresStore.from_conn_string(
            conn,
            index={"dims": 1536, "embed": "openai:text-embedding-3-small"}
        )
        store.setup()

        app = builder.compile(checkpointer=checkpointer, store=store)

        # Alice 的对话
        alice_cfg = {"configurable": {"thread_id": "user-alice-001", "user_id": "alice"}}
        r1 = await app.ainvoke(
            {"messages": [HumanMessage(content="帮我订一张明天去上海的机票")]},
            config=alice_cfg
        )

        # 几小时后，Alice 开新会话
        r2 = await app.ainvoke(
            {"messages": [HumanMessage(content="我的机票订好了吗？")]},
            config={"configurable": {"thread_id": "user-alice-002", "user_id": "alice"}}
            # thread_id 变了，但 user_id 相同 → Store 里的长期记忆还在
        )

asyncio.run(main())

短期、长期记忆全部落盘到 PostgreSQL，支持高并发。这是生产标配。

---

六、时间旅行与状态回退

Checkpointer 不只把状态存下来让你"记得对话"，它把每一次执行后的完整状态快照都留了底。所以你随时能回看"当时 Agent 走到哪了"，也能像打游戏存读档一样退回上一步或从某个存档点分叉。

这就是所谓"时间旅行（Time Travel）

1. 查看历史：get_state_history

config = {"configurable": {"thread_id": "user-alice-001"}}
history = list(app.get_state_history(config))

干什么用：把这个 thread 里所有历史 checkpoint 倒序拉出来（最新的在 index 0）。

每个 checkpoint 对象有三个关键字段：

• metadata['step'] —— 第几步（超级步骤序号）
• metadata['source'] —— 这步是咋来的（loop 正常循环 / update 手动改的 / branch 分叉的 等）
• values —— 当时的完整状态（含 messages 等所有通道数据）

for cp in history:
    print(f"步骤: {cp.metadata['step']}  节点: {cp.metadata['source']}  消息数: {len(cp.values.get('messages', []))}")

典型场景：审计 Agent 每一步干了啥、调试"为什么第三步开始答非所问"。

---

2. 回退：update_state

if len(history) >= 2:
    previous = history[1]   # 0=最新, 1=上一个状态
    app.update_state(config, previous.values)

干什么用：把当前 thread 的状态直接覆盖成某个历史快照——相当于"撤销"最后几步。

• history[1] 就是"上一个 checkpoint"，覆盖后 Agent 下次调用就从那一步重新开始。
• 这不是删历史，而是在历史顶端压入一个新节点（source 标记为 update），原历史还在。

典型场景：Agent 最后一步工具调用写错了，不想重跑整段，直接退回上一步重来。

---

3. 崩溃恢复：invoke(None, config)

result = app.invoke(None, config=config)   # 传 None = 从存档恢复

干什么用：进程崩了、服务重启了，传 None 作为输入，LangGraph 会自动加载该 thread 最后一个 checkpoint，从断点接着跑。

前提：checkpointer 用的是持久化后端（SQLite/Postgres/Redis），内存版重启就丢了没得恢复。

典型场景：长任务跑到一半进程挂了，用户再发一条消息时无缝续上，不用从头来。

4. 进阶：从过去某个点分叉（Branch）

update_state 不仅能回退，还能改了状态后沿着新路径往下走——这就是"分叉出新路径"。

# 退回第 3 步的状态，但手动改点东西
app.update_state(config, {"messages": [HumanMessage("换个问法重新来")]}, as_node="agent")
# 之后正常 invoke，会从第 3 步之后用新输入继续，原第 4、5 步历史保留不受影响

核心区别：

操作	效果
get_state_history	只读，看历史
update_state	写，覆盖/修正当前状态（可回退、可分支）
invoke(None, ...)	重启后从断点续跑

一句话：存读档 + 悔棋 + 分叉试错，全靠 Checkpointer 留的底。

七、避坑指南

7.1 版本安全（重要）

2025-2026 年 Checkpointer 层曝出多个高危 CVE：

CVE	影响组件	风险	最低安全版本
CVE-2025-67644	SQLite Checkpointer	SQL 注入 → RCE	langgraph-checkpoint-sqlite >= 3.0.1
CVE-2026-28277	核心 msgpack 反序列化	远程代码执行	langgraph >= 1.0.10
CVE-2026-27022	Redis Checkpointer	查询注入	langgraph-checkpoint-redis >= 1.0.2
CVE-2026-71433	Postgres/SQLite 命名空间	跨租户数据泄露	langgraph-checkpoint >= 4.0.1

所有后端都必须升级核心 langgraph 包——msgpack 解码器在共享核心里。

7.2 常见错误速查

错误	原因	解决
CheckpointerConnectionError	数据库连接失败	检查连接字符串，确保服务已启动
对话历史丢失	用了随机 thread_id	用确定性 ID（如 user-{user_id}）
消息无限增长	没配消息压缩	加 SummarizationMiddleware 或定期清理
SQLite 并发错误	多进程同时写	升级到 PostgreSQL
跨线程记忆泄露	thread_id / namespace 未隔离	每个用户用唯一 thread_id 和 namespace
Store 搜索无结果	没配 embedding index	创建时传 index={"dims": 1536, "embed": "..."}

7.3 存储空间优化

# 策略 1：PostgreSQL 定期清理旧 checkpoint
# DELETE FROM checkpoints
#   WHERE metadata->>'timestamp' < NOW() - INTERVAL '30 days';

# 策略 2：Redis 利用原生 TTL 自动过期（最省心）

# 策略 3：SummarizationMiddleware 压缩长对话
from langchain.agents.middleware import SummarizationMiddleware
from langchain_openai import ChatOpenAI

summary = SummarizationMiddleware(
    model=ChatOpenAI(model="gpt-4o-mini"),
    trigger=("messages", 100)  # 消息达 100 条时触发压缩
)

注意：SummarizationMiddleware 的导入路径在不同 LangChain 版本可能不同。如果上述路径报错，尝试 from langgraph.graph.middleware import SummarizationMiddleware 或查阅当前版本文档。

---

八、总结

Checkpointer（短期记忆）让 Agent 记住"刚才聊了什么"，绑定 thread_id，存到 checkpoints 表；Store（长期记忆）让 Agent 记住"这个用户是谁"，绑定 namespace，存到向量/KV 库。生产 Agent 两者缺一不可：Checkpointer 保对话连贯，Store 保用户认知持久。

---