🧠 9.6 记忆管理与会话状态持久化

“没有记忆的 Agent 每一秒都在经历失忆，而没有节制的记忆则会迅速撑爆上下文并耗尽你的钱包。”
在 LangChain 1.4 中，旧时代的 ConversationBufferMemory 早已废弃，取而代之的是 LangGraph Checkpointer + thread_id 的线程级短期记忆、trim_messages 窗口裁剪，以及 Store / LangMem 的跨会话长期记忆。

💡 先看一个直观例子：随身便签白板与智能历史剪报

1. 记忆系统的两大核心难题

难题 1：多租户会话隔离（张冠李戴）：用户 A 告诉 AI“我叫 Alice”，同一时间用户 B 进来聊天，AI 绝不能把 A 的隐私透露给 B。
难题 2：上下文窗口溢出（Token 爆炸）：聊了 50 轮后，历史对话几万字，每次提问都全量带上所有历史，不仅模型响应极慢，费用呈指数级飙升，还会超出模型的 Context Window 导致崩溃。

2. 生活比喻：办公桌抽屉与智能历史剪报员

Checkpointer + thread_id（1.4 现代方案）：就像给每个来访客户（thread_id）在档案柜里分配一个专属抽屉，LangGraph 自动把每一次对话状态写入抽屉，下次凭同一把钥匙（thread_id）原样取出继续聊；
RunnableWithMessageHistory（经典 LCEL 方案）：同样是“抽屉 + 钥匙”，只是需要自己用 session_store 字典手动管理抽屉（链级场景仍可用）；
trim_messages：就像一位贴身剪报秘书。当档案厚度超过限制（Token 预算）时，秘书会自动撕掉最早无关紧要的闲聊，但严格保留最顶层的**核心原则（SystemMessage）**以及最近几轮的关键问答，确保大脑始终高效清爽！

💻 核心实操：多用户记忆管理与 Token 窗口裁剪

1. 🆕 LangChain 1.4 现代方案：LangGraph Checkpointer（短期记忆/线程级记忆）

LangChain 1.4 的记忆推荐做法是给 create_agent 配一个 checkpointer（LangGraph 持久化检查点），并在每次调用时带上 thread_id（相当于旧版的 session_id）。框架会在每次调用后自动把完整消息状态“落盘”，下次同 thread_id 调用时自动加载历史——你不用再手写 session_store 字典。

python

# code/s06_memory_and_trimming.py
from langgraph.checkpoint.memory import MemorySaver   # MemorySaver 与 InMemorySaver 是同一个类的两个名字，官方文档示例使用 InMemorySaver
from langchain.agents import create_agent
from s01_model_io import get_chat_model

# 1. 创建检查点（开发用内存版；生产用 SqliteSaver / PostgresSaver）
checkpointer = MemorySaver()

# 2. 创建带记忆的 Agent
agent = create_agent(
    model=get_chat_model(),
    tools=[],
    checkpointer=checkpointer,
)

# 3. 用 thread_id 区分不同用户的多轮会话
config_alice = {"configurable": {"thread_id": "user_alice"}}
config_bob   = {"configurable": {"thread_id": "user_bob"}}

agent.invoke({"messages": [("user", "你好！我最喜欢的编程语言是 Python，生日是 10 月 1 日。")]}, config=config_alice)
agent.invoke({"messages": [("user", "你好！我是 Bob，我主要写 C++。")]}, config=config_bob)

# 4. 检验记忆隔离：Alice 的隐私不会被 Bob 串台
res = agent.invoke({"messages": [("user", "还记得我最喜欢的语言和生日吗？")]}, config=config_alice)
print(res["messages"][-1].content)  # 正确回答：Python / 10月1日

🧰 Checkpointer 选型（记忆持久化的“后端数据库”）：
实现导入路径适用场景
MemorySaver（旧名 InMemorySaver 已废弃） langgraph.checkpoint.memory 开发/测试，进程退出即清空
SqliteSaver langgraph.checkpoint.sqlite 单机/桌面应用，落盘文件持久化
PostgresSaver langgraph.checkpoint.postgres 生产环境，支持高并发多进程共享

实现	导入路径	适用场景
`MemorySaver`（旧名 `InMemorySaver` 已废弃）	`langgraph.checkpoint.memory`	开发/测试，进程退出即清空
`SqliteSaver`	`langgraph.checkpoint.sqlite`	单机/桌面应用，落盘文件持久化
`PostgresSaver`	`langgraph.checkpoint.postgres`	生产环境，支持高并发多进程共享

2. 经典 LCEL 方案：RunnableWithMessageHistory（链级仍可用）

如果你暂时不用 Agent，而是在普通 LCEL 链上挂记忆，RunnableWithMessageHistory 依然在 langchain-core 中可用，通过 session_id 在配置中切换会话：

python

# code/s06_memory_and_trimming.py
from typing import Dict
from langchain_core.chat_history import InMemoryChatMessageHistory, BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from s01_model_io import get_chat_model

# 1. 声明多 Session 会话存储容器
session_store: Dict[str, InMemoryChatMessageHistory] = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    """按 session_id 提取或创建历史记录"""
    if session_id not in session_store:
        session_store[session_id] = InMemoryChatMessageHistory()
    return session_store[session_id]

# 2. 组装基础对话链
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名贴心的智能生活管家。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{question}")
])
llm = get_chat_model()
chain = prompt | llm | StrOutputParser()

# 3. 包装为带记忆能力的 Runnable
chain_with_memory = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="question",
    history_messages_key="history"
)

# 4. 分别与不同用户进行多轮对话
config_alice = {"configurable": {"session_id": "user_alice"}}
config_bob = {"configurable": {"session_id": "user_bob"}}

chain_with_memory.invoke({"question": "你好！我最喜欢的颜色是蓝色。"}, config=config_alice)
chain_with_memory.invoke({"question": "你好！我叫 Bob，我最喜欢红色。"}, config=config_bob)

# 检验 Alice 的记忆是否被 Bob 污染
res = chain_with_memory.invoke({"question": "还记得我最喜欢的颜色吗？"}, config=config_alice)
print(res) # 正确回答：蓝色

3. trim_messages 智能滑动窗口裁剪

当对话轮数很多时，使用 trim_messages 在输入模型前对上下文进行精准瘦身：

python

from langchain_core.messages import trim_messages, SystemMessage, HumanMessage, AIMessage

messages = [
    SystemMessage(content="你是核心安全系统。"),
    HumanMessage(content="第 1 轮：服务器状态如何？"),
    AIMessage(content="一切正常。"),
    HumanMessage(content="第 2 轮：CPU 负载多少？"),
    AIMessage(content="当前 30%。"),
    HumanMessage(content="最新提问：我刚才第 1 轮问了什么？")
]

# 智能裁剪策略配置
trimmed = trim_messages(
    messages,
    max_tokens=60,         # 最大 Token 或字符预算
    token_counter=len,     # 计数器（生产中可用 tiktoken）
    strategy="last",       # 保留最新的对话内容
    start_on="human",      # 保证第一条有效消息始于 human
    include_system=True    # 必须保留最顶层的 System 核心指令
)

print(f"裁剪后保留消息数: {len(trimmed)}")

🆕 1.1+ 进阶：SummarizationMiddleware 一键自动摘要。上面 trim_messages 是"硬裁剪"（丢掉旧消息），而官方中间件 SummarizationMiddleware 会在对话超长时自动把历史提炼成摘要，配合 9.7 的中间件机制一行装配进 create_agent 即可。

SummarizationMiddleware 1.4 参数精读（本地实测 `langchain==1.4.0`）

把它想象成一位会议纪要秘书：会上讨论到第 100 页 PPT 时（触发点），她会把前 80 页浓缩成一页纪要（摘要），只留下最后 20 页原文（keep）；就算她临时请假（摘要生成失败），会议记录也一页不丢（保留原始历史，绝不用编造的摘要顶替）。

python

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model=llm,
    tools=[search_weather],
    middleware=[
        SummarizationMiddleware(
            model=llm,                              # 负责"写纪要"的模型（可与主模型不同，用便宜的小模型省钱）
            trigger=("tokens", 3000),               # 触发点：三种触发模式（详见下表），也支持列表 = OR 多条件
            keep=("messages", 20),                  # 摘要后保留最近的 20 条原始消息（默认值）
            trim_tokens_to_summarize=4000,          # 单次送给摘要模型的原文 token 上限（默认 4000）
            # token_counter=count_tokens_approximately,  # 自定义计数器，默认近似计数（零依赖、不装 tokenizer）
        ),
    ],
)

trigger / keep 的三种量法（元组第一项是"单位"，第二项是"阈值"）：

写法	触发/保留条件	生活比喻
`("messages", 50)`	消息条数达到 50	会议开了 50 页发言记录
`("tokens", 4000)`	Token 数达到 4000	发言稿写了 4000 字
`("fraction", 0.8)`	占模型上下文窗口的 80%	会议记录本写满了八成

两个 1.4 版值得注意的行为细节（源自官方文档与源码 docstring）：

AND 组合触发：trigger 传 TriggerClause 或含 op="and" 的组合子句，可表达"且超过 80% 上下文"这类复合条件；直接传列表则是 OR（任一命中即触发）。
摘要失败 ≠ 丢历史：摘要生成这类瞬时错误（限流 429 / 超时）会由 Runnable.with_retry 在进程内自动重试（共 3 次）；重试后仍失败则保留原始历史继续运行——宁可"记满"也不拿编造的摘要糊弄上下文（1.4 修复：此前失败会丢历史）。
摘要时会自动保持 AI/Tool 消息成对（不会把工具调用和它的结果拆散），且摘要采用固定的四段式上下文提取格式（SESSION INTENT / SUMMARY / ARTIFACTS / NEXT STEPS）。

4. 记忆三数据源全景：Runtime Context / State / Store

LangChain 1.4 官方把 Agent 能读写的"数据来源"统一归纳为三类（记忆的本质，就是这三类数据的读写策略）：

数据源	别名	作用域	典型例子
Runtime Context	静态配置	单次会话（运行期固定）	`user_id`、API Key、数据库连接、权限等级、环境标识
State	短期记忆	单次会话（持续追加）	当前消息列表、已上传文件、认证状态、工具结果
Store	长期记忆	跨会话（永久留存）	用户偏好、画像、历史数据、知识沉淀

前两节讲的 Checkpointer + thread_id 正是 State；而本节的 Store 是跨会话长期记忆。二者之上还有一层 Runtime Context，用 create_agent(..., context_schema=...) 声明，配合 9.11 的自定义中间件在 request.runtime.context 里读取。

5. 长期记忆（跨会话）：Store 与 namespace 分层

短期记忆只在单条会话内生效；当我们要跨会话记住"用户张三喜欢黑色、预算 5000 元"这类长期画像时，就要用 LangGraph Store（或官方 LangMem SDK 做自动记忆提取）。Store 用元组命名空间组织数据，支持像文件目录一样分层存取：

python

# code/s06_memory_and_trimming.py —— 跨会话长期记忆：namespace 分层 + 按 user_id 隔离
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# 用元组 namespace 分层："users" 层下的 "zhangsan" 用户画像
store.put(("users", "zhangsan"), "preferences", {"color": "black", "budget": 5000})
store.put(("users", "zhangsan"), "purchase_history", {"last": "机械键盘"})
store.put(("users", "lisi"), "preferences", {"color": "white"})

# 按 (namespace, key) 精确读回，天然隔离不同用户 / 不同维度
item = store.get(("users", "zhangsan"), "preferences")
print(item.value)  # {'color': 'black', 'budget': 5000}

# 也可以按 namespace 前缀批量搜索（如拉取某用户全部档案）
items = store.search(("users", "zhangsan"))

把 Store 挂进 create_agent（create_agent(..., store=store, context_schema=UserContext)）后，自定义中间件就能通过 request.runtime.store 在每次模型调用前读取长期画像、动态注入系统提示——这正是 9.10 上下文工程的核心动作。

记忆能力矩阵速记：短期记忆 = State（Checkpointer + thread_id）；长期记忆 = Store + namespace + user_id；运行期静态配置 = Runtime Context。

📚 权威官方资料直达

🔗 LangGraph 记忆官方指南（Checkpointer / Store）：LangGraph Memory
🔗 长期记忆与 Store 官方指南：LangGraph Store (Long-term Memory)
🔗 上下文工程三数据源（Runtime Context / State / Store）：Context Engineering in Agents
🔗 会话历史官方指南（经典 LCEL 方案）：How to add memory to chat models
🔗 上下文裁剪官方文档：How to trim messages
🔗 SummarizationMiddleware 参数全表（1.4）：Built-in Middleware · Summarization

会话标识与用户标识不能混用

thread_id 用于定位一段会话，user_id 用于区分长期记忆的归属。两者混用可能让不同用户读到彼此的内容，或让同一用户的多个任务互相污染。生产环境还应把租户、用户和应用范围纳入 namespace，并在读取时再次校验权限。

记忆需要完整生命周期：写入前判断是否值得长期保存，读取时考虑时效，允许更正和删除，过期后按规则清理。Checkpointer 与 Store 中若含敏感数据，也要设置加密、访问日志和保留期限。

🎯 本节小结与思考

核心收获：掌握了 1.4 现代记忆方案 —— LangGraph Checkpointer + thread_id 线程级短期记忆、trim_messages Token 预算裁剪、SummarizationMiddleware 自动摘要（1.4 的 trigger 三量法 / AND 组合 / 摘要失败保留历史）、Store/LangMem 跨会话长期记忆，以及经典 LCEL 链的 RunnableWithMessageHistory 兜底方案；并从 Runtime Context / State / Store 三数据源视角理解了记忆的本质。
下一步探索：大模型调用过程中，我们如何实时监控每一轮的 Token 消耗与网络耗时？遇到敏感数据如何自动脱敏拦截？下一节我们学习 9.7 Callbacks 回调与可观测性中间件。

🧠 9.6 记忆管理与会话状态持久化 ​

💡 先看一个直观例子：随身便签白板与智能历史剪报 ​

1. 记忆系统的两大核心难题 ​

2. 生活比喻：办公桌抽屉与智能历史剪报员 ​

💻 核心实操：多用户记忆管理与 Token 窗口裁剪 ​

1. 🆕 LangChain 1.4 现代方案：LangGraph Checkpointer（短期记忆/线程级记忆） ​

2. 经典 LCEL 方案：RunnableWithMessageHistory（链级仍可用） ​

3. trim_messages 智能滑动窗口裁剪 ​

SummarizationMiddleware 1.4 参数精读（本地实测 langchain==1.4.0） ​

4. 记忆三数据源全景：Runtime Context / State / Store ​

5. 长期记忆（跨会话）：Store 与 namespace 分层 ​

📚 权威官方资料直达 ​

会话标识与用户标识不能混用 ​

🎯 本节小结与思考 ​

🧠 9.6 记忆管理与会话状态持久化

💡 先看一个直观例子：随身便签白板与智能历史剪报

1. 记忆系统的两大核心难题

2. 生活比喻：办公桌抽屉与智能历史剪报员

💻 核心实操：多用户记忆管理与 Token 窗口裁剪

1. 🆕 LangChain 1.4 现代方案：LangGraph Checkpointer（短期记忆/线程级记忆）

2. 经典 LCEL 方案：RunnableWithMessageHistory（链级仍可用）

3. trim_messages 智能滑动窗口裁剪

SummarizationMiddleware 1.4 参数精读（本地实测 `langchain==1.4.0`）

4. 记忆三数据源全景：Runtime Context / State / Store

5. 长期记忆（跨会话）：Store 与 namespace 分层

📚 权威官方资料直达

会话标识与用户标识不能混用

🎯 本节小结与思考