LangChain 记忆与会话管理全攻略:让 Agent 拥有「记忆」

⚠️ 重要:本文包含 记忆体系认知(短期 / 长期 / 工作记忆)Buffer Memory(ConversationBufferMemory / BufferWindowMemory)向量化长期记忆(VectorStoreRetrieverMemory)摘要记忆(ConversationSummaryMemory / SummaryBufferMemory)LangGraph 持久化(MemorySaver / SqliteSaver / Checkpoint)会话隔离(thread_id / user_id 多租户) 等记忆与会话管理六大核心主题。

文中涉及的相关代码示例地址:https://github.com/m12305/Langchain-LangGraph-agent — Langchain/LangGraph 学习项目
相关项目推荐:https://github.com/m12305/hello-FastAPI — FastAPI 学习项目

LLM 本质上是无状态的——每次调用都是"第一次见面"。本文从记忆的三种层次出发,一路深入到 Buffer 滑动窗口、向量化长期记忆、摘要压缩、LangGraph Checkpoint 持久化,最终落到多租户会话隔离——把"如何让 Agent 记住"这件事讲透。

一、记忆体系:为什么 Agent 需要"记忆"

1.1 LLM 是无状态的

你: "我叫小明,我养了一只猫。"
AI: "好的小明,我记住了!"

你: "我猫叫什么名字?"
AI: "???什么猫?你养猫了?"  ← 完全失忆!

原因: 每次 API 调用是独立的,LLM 本身不保存任何状态

LLM 的本质:一个函数 f(messages) → response,你不把历史传给它,它就不知道。

1.2 记忆的三种层次

📝 短期记忆 (Working Memory)
  └── 当前对话窗口内的消息,例: "刚才我说了什么?"

📚 长期记忆 (Long-term Memory)
  └── 跨会话的持久化知识,例: "三个月前我告诉过你我喜欢蓝色"

🧠 工作记忆 (Episodic Memory)
  └── 对过去对话的总结和反思,例: "根据之前的对话,你偏好简洁回答"

1.3 LangChain 的记忆方案总览

记忆类型 实现 适用场景
Buffer Memory ConversationBufferMemory 短对话(< 10 轮)
Buffer Window ConversationBufferWindowMemory 中短对话,只保留最近 K 轮
Summary Memory ConversationSummaryMemory 长对话,用模型做摘要
Summary Buffer ConversationSummaryBufferMemory 长对话,摘要 + 最近 K 轮
Vector Memory VectorStoreRetrieverMemory 跨会话,向量检索相关历史
LangGraph 持久化 MemorySaver / SqliteSaver Agent 工作流状态持久化(最推荐)

1.4 记忆选择决策树

你的场景是什么?
├── 简单问答 (单轮)           → 不需要记忆
├── 短对话 (5-10轮)           → BufferMemory / BufferWindowMemory
├── 长对话 (20+轮)            → SummaryBufferMemory (摘要 + 最近K轮)
├── 跨会话 (下次打开还记得)    → VectorStoreRetrieverMemory (嵌入检索)
├── Agent 工作流 (LangGraph)  → MemorySaver / SqliteSaver (状态持久化)
└── 生产环境 (多用户)          → PostgresSaver + thread_id 隔离

⚠️ 两个常见坑:别把记忆当数据库(结构化数据存 DB,记忆只存对话);别把短期/长期记忆混为一谈(100 轮对话全塞 buffer 会 token 爆炸)。


二、Buffer Memory:最简单的短期记忆

2.1 ConversationBufferMemory — 完整缓冲

类比:一个笔记本,每轮对话都往上记。优点是完整不丢失;缺点是 token 不断增长。

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(
    memory_key="history",       # 在 prompt 中的变量名
    return_messages=True,        # 返回 Message 对象(而非字符串)
)

memory.save_context({"input": "我叫小明,我喜欢蓝色。"}, {"output": "你好小明!"})
memory.save_context({"input": "我养了一只猫叫小花。"}, {"output": "小花一定很可爱!"})

history = memory.load_memory_variables({})["history"]
# [HumanMessage("我叫小明..."), AIMessage("你好小明!"), ...]

完整管道调用(记忆集成的核心循环):

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是助手"),
    MessagesPlaceholder(variable_name="history"),  # ← 记忆注入点
    ("human", "{input}"),
])
chain = prompt | model | StrOutputParser()

for user_input in conversations:
    history = memory.load_memory_variables({})["history"]  # 1. 加载历史
    response = chain.invoke({"input": user_input, "history": history})  # 2. 调用
    memory.save_context({"input": user_input}, {"output": response})     # 3. 保存本轮

2.2 ConversationBufferWindowMemory — 滑动窗口

类比:写日记(保留所有)vs 发朋友圈(只保留最近几次互动)。

from langchain.memory import ConversationBufferWindowMemory

memory = ConversationBufferWindowMemory(
    memory_key="history",
    return_messages=True,
    k=3,  # ← 只保留最近 3 轮
)

K 值选择:

K 值 对话轮数 Token 消耗 适用
1-2 很短 很低 简单问答
3-5 适中 中等 日常对话
10+ 较长 较高 深度对话

💡 建议:从 K=5 开始,根据任务调整。

2.3 对比总结

特性 BufferMemory BufferWindowMemory(K=3)
保留所有对话
Token 增长 线性增长 恒定(K 轮)
早期信息保留 ❌(被遗忘)
适用 短对话(<15轮) 任意长度对话

⚠️ memory_key 必须和 prompt 中的 MessagesPlaceholder(variable_name=...) 一致,否则 KeyError。


三、向量化长期记忆:跨会话的语义检索

3.1 为什么需要向量化?

类比:笔记本 vs 图书馆。Buffer 只能按时间顺序访问,向量化记忆能语义搜索——问"那个关于猫的事",能搜到 20 轮前的相关对话。

每次对话结束:
  用户消息 → embedding() → 向量 → 存入向量数据库
新对话时:
  用户消息 → embedding() → 向量 → 搜索最相似历史 → 注入 prompt 作为"记忆"

3.2 VectorStoreRetrieverMemory

from langchain.memory import VectorStoreRetrieverMemory
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

embedding_model = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(collection_name="long_term_memory", embedding_function=embedding_model)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

memory = VectorStoreRetrieverMemory(retriever=retriever, memory_key="relevant_history")

memory.save_context({"input": "我喜欢蓝色"}, {"output": "好的,我记住了。"})

# 语义搜索!新问题无需和旧问题"完全匹配"
results = memory.load_memory_variables({"input": "推荐一个适合我的颜色"})
# → 会检索到"我喜欢蓝色"的记忆(语义相似)

3.3 什么该存?什么不该存?

✅ 该存长期记忆的:
  "我叫小明" / "我对猫过敏" / "我是Python工程师" / "我在做XX项目"
❌ 不该存的:
  "今天天气怎么样?" / "1+1等于几?" / "讲个笑话吧"

💡 进阶:保存前先用 LLM 提取"值得记住的信息",只存关键内容而非整个对话,避免污染检索结果。

⚠️ :① 用支持中文的 Embedding 模型(text-embedding-3-small,别用老版 ada-002);② K 设 3~5 即可,太大引入噪音。


四、摘要记忆:用 LLM 压缩长对话

4.1 为什么需要摘要?

类比:全程录音 vs 会议纪要。1 轮对话约 100 tokens,200 轮约 20000 tokens 会超出上下文窗口;而 SummaryMemory 无论多少轮,摘要始终约 200-500 tokens。

4.2 ConversationSummaryMemory

from langchain.memory import ConversationSummaryMemory

memory = ConversationSummaryMemory(
    llm=model,             # 用于做摘要的模型
    memory_key="history",
    return_messages=False, # 摘要返回字符串,不是列表
)
memory.save_context({"input": "我叫小明,从事软件开发。"}, {"output": "你好小明!"})
print(memory.load_memory_variables({})["history"])
# "The human, 小明, is a software developer..."

⚠️ :SummaryMemory 返回的是字符串,所以 prompt 里用 {history} 直接插入,而不是 MessagesPlaceholder

4.3 ConversationSummaryBufferMemory — 最佳实践

混合策略:前 50 轮摘要成一段文字,最近 5 轮保留完整对话:

from langchain.memory import ConversationSummaryBufferMemory

memory = ConversationSummaryBufferMemory(
    llm=model,
    memory_key="history",
    max_token_limit=200,   # 关键参数: 记忆总 token 上限,超过就压缩
    return_messages=True,  # 最近消息以 Message 对象返回
)

4.4 三种记忆对比

维度 BufferMemory SummaryMemory SummaryBufferMemory
早期信息 完整保留 摘要保留 摘要保留
近期细节 完整保留 丢失! 完整保留
Token 增长 线性增长 固定 固定(上限)
额外成本 每次摘要调 LLM 每次摘要调 LLM
适用长度 短 (< 15轮) 任意 任意 ⭐

五、LangGraph 持久化:Checkpoint 机制

5.1 为什么需要?—— 游戏存档 vs 聊天记录

传统 Memory       = 聊天记录(只记说了什么,不知道 Agent 正在第几步)
LangGraph Checkpoint = 游戏存档(记录完整状态,断掉后可完美恢复)

传统 Memory 丢失了:Agent 当前步骤、工具调用历史、中间推理过程(Thought 链)。而 Checkpoint 保存的是完整的 Graph State。

5.2 MemorySaver — 内存级持久化

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState
from langchain_core.messages import HumanMessage

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "user-123"}}

# 第1次对话
graph.invoke({"messages": [HumanMessage("我叫小明")]}, config)
# 第2次对话 — 同一个 thread_id,自动加载历史状态!
result2 = graph.invoke({"messages": [HumanMessage("我叫什么?")]}, config)

5.3 SqliteSaver — 磁盘持久化

MemorySaver: ✅ 零配置  ❌ 重启丢失、不能跨进程
SqliteSaver: ✅ 持久化磁盘、重启不丢  ❌ 不适合高并发(单文件数据库)
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

conn = sqlite3.connect("agent_checkpoints.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
graph = builder.compile(checkpointer=checkpointer)  # API 完全一致!

5.4 时间旅行 — Checkpoint 的高级能力

# 回到之前的某个 checkpoint,重放 / 分支 / 调试 Agent 的决策
state_before = graph.get_state({
    "configurable": {"thread_id": "user-123", "checkpoint_id": "1ef...abc"}
})

5.5 传统 Memory vs LangGraph 持久化

维度 传统 Memory LangGraph Checkpoint
保存内容 只有对话文本 完整的 Graph State
恢复能力 只能接着聊 回到任何历史状态
多线程隔离 需手动管理 thread_id 天然隔离
持久化 BufferMemory 不持久化 SqliteSaver/PostgresSaver
推荐度 ⭐⭐ (简单场景) ⭐⭐⭐⭐⭐ (所有场景)

⚠️ :① 忘了传 thread_id → 每次都是新会话;② 生产环境别用 MemorySaver(重启丢失),换 SqliteSaver/PostgresSaver;③ thread_id 用 UUID 或组合键,避免碰撞。


六、会话范围与隔离:多租户设计

6.1 为什么需要隔离?

类比酒店房卡:每个客人有自己的房卡(thread_id),刷卡进自己的房间。没有隔离时,用户 B 问"我叫什么?“会拿到用户 A 的"我叫小明”——信息泄露

6.2 三层命名空间设计

user_id    : 标识"谁"      → 用户级别(长期记忆、用户偏好)
thread_id  : 标识"哪次对话" → 会话级别(当前对话状态)
run_id     : 标识"哪次调用" → 调用级别(单次 API 调用,自动生成)

user_123 (用户)
  ├── thread_001 (第1次对话)
  │   ├── run_a / run_b
  └── thread_002 (第2次对话)
def make_thread_id(user_id: str, session_id: str) -> str:
    return f"{user_id}/{session_id}"   # 组合键,天然隔离

config = {"configurable": {"thread_id": "user-456/chat-789", "user_id": "user-456"}}

6.3 长期记忆按 user_id 过滤

class UserAwareVectorMemory:
    """按用户隔离的长期记忆"""
    def save(self, user_id: str, text: str, metadata: dict):
        metadata["user_id"] = user_id
        self.vectorstore.add_texts([text], metadatas=[metadata])

    def load(self, user_id: str, query: str, k: int = 5):
        return self.vectorstore.similarity_search(
            query, k=k, filter={"user_id": user_id},  # ← 只检索自己的记忆
        )

6.4 生产级多租户 Agent 架构

class MultiTenantAgent:
    def chat(self, user_id: str, session_id: str, message: str):
        thread_id = f"{user_id}/{session_id}"
        config = {"configurable": {"thread_id": thread_id, "user_id": user_id}}

        long_term = self.long_term_memory.load(user_id, message, k=3)  # 1. 检索长期记忆
        enriched = HumanMessage(content=f"[用户背景]\n{long_term}\n\n[当前消息]\n{message}")

        result = self.graph.invoke({"messages": [enriched]}, config)   # 2. 短期记忆由 checkpoint 管

        memory_worthy = self._extract_memory(result)                   # 3. 提取关键信息
        if memory_worthy:
            self.long_term_memory.save(user_id, memory_worthy, {})     # 4. 存长期记忆
        return result

⚠️ :① thread_id 用自增 ID 会碰撞 → 用组合键/UUID;② 忘记清理过期会话 → checkpoint 无限累积;③ 匿名用户记忆登录后丢失 → 先存临时记忆,登录后合并。


七、全章知识地图

                    ┌─────────────────────────────────────────┐
                    │         记忆与会话管理知识体系            │
                    └─────────────────────────────────────────┘
                                       │
           ┌───────────────────────────┼───────────────────────────┐
           │                           │                           │
           ▼                           ▼                           ▼
   ┌───────────────┐         ┌───────────────┐         ┌───────────────┐
   │  短期记忆      │         │  长期记忆      │         │  会话管理      │
   │               │         │               │         │               │
   │ • BufferMemory│         │ • 向量化记忆   │         │ • thread_id   │
   │ • Window K轮  │         │ • 摘要记忆     │         │ • user_id     │
   │ • SummaryBuffer│        │ • 语义检索     │         │ • 多租户隔离   │
   │               │         │ • 记忆提取     │         │ • 会话清理     │
   └───────────────┘         └───────────────┘         └───────────────┘
           │                           │                           │
           └───────────────────────────┼───────────────────────────┘
                                       │
                    ┌──────────────────┴──────────────────┐
                    │          LangGraph Checkpoint        │
                    │                                     │
                    │  • MemorySaver (内存) / SqliteSaver   │
                    │  • 完整 State 持久化                  │
                    │  • 时间旅行 / 分支 / 重放             │
                    └─────────────────────────────────────┘

总结

模块 核心洞察
记忆体系 LLM 无状态是根源;短期/长期/工作记忆各司其职,别混为一谈
Buffer Memory 最简单短期记忆;Window 版用 K 值控制 token,是"近期细节"的答案
向量化长期记忆 从"按时间顺序"升级到"语义搜索";只存值得记住的信息,用多语言 Embedding
摘要记忆 用 LLM 压缩长对话;SummaryBufferMemory(摘要 + 最近 K 轮)是最佳实践
LangGraph 持久化 Checkpoint 保存完整 State,是 Agent 记忆的现代推荐方案
会话隔离 thread_id 隔离会话、user_id 隔离用户;组合键 + UUID 防碰撞

掌握了记忆,你的 Agent 就不再是"金鱼"了。下一站:工具调用(Tool Calling)——让 Agent 真正"做事" 🚀


本文基于"AI Agent 学习项目"第 5 阶段(记忆与会话管理)整理,覆盖 5.1 对话记忆概述5.2 Buffer Memory5.3 向量化长期记忆5.4 摘要记忆5.5 LangGraph 持久化5.6 会话范围与隔离 六个章节。

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐