LangChain 与 LangGraph 记忆管理策略完整代码演示

本文将用多个独立、可运行的代码示例，逐一演示 LangChain 中短期记忆、长期记忆以及LangGraph 的 Checkpointer 会话记忆，并覆盖修剪、删除、总结等高级记忆管理策略。所有示例均基于阿里云百炼 Qwen 模型（兼容 OpenAI API）。

---

环境准备

在运行任何代码前，请安装依赖并设置 .env 文件。

pip install langchain langchain-openai langchain-community chromadb faiss-cpu tiktoken
# 如果需要 SQLite checkpointer
pip install langgraph-checkpoint-sqlite

创建 .env 文件：

DASHSCOPE_API_KEY=你的API Key
DASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1

通用模型初始化（后续所有示例共用）：

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

llm = ChatOpenAI(
    model="qwen-plus",
    temperature=0.7,
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_BASE_URL"),
    model_kwargs={"extra_body": {"enable_thinking": False}}
)

---

一、短期记忆（Short-Term Memory）

1. 会话缓冲记忆（ConversationBufferMemory）

保存全部历史，最直接。

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

memory = ConversationBufferMemory()
conversation = ConversationChain(llm=llm, memory=memory, verbose=False)

conversation.predict(input="我叫小明，我喜欢编程。")
response = conversation.predict(input="你还记得我叫什么名字吗？")
print(response)
# 输出会正确回答“小明”

2. 会话窗口记忆（ConversationBufferWindowMemory）

只保留最近 k 轮对话。

from langchain.memory import ConversationBufferWindowMemory

memory = ConversationBufferWindowMemory(k=2)
conversation = ConversationChain(llm=llm, memory=memory, verbose=False)

conversation.predict(input="我叫小明。")
conversation.predict(input="我喜欢编程。")
conversation.predict(input="你还记得我的名字吗？")  # 窗口只保留后两轮，可能忘记名字
print(conversation.predict(input="我叫什么？"))

3. 会话摘要记忆（ConversationSummaryMemory）

利用 LLM 将历史压缩为摘要。

from langchain.memory import ConversationSummaryMemory

memory = ConversationSummaryMemory(llm=llm)
conversation = ConversationChain(llm=llm, memory=memory, verbose=False)

conversation.predict(input="我叫小明，我是程序员。")
conversation.predict(input="我喜欢 Python 和 Go。")
response = conversation.predict(input="请回顾一下我的信息。")
print(response)
# 模型会基于摘要回答，能看到姓名和爱好

4. 会话摘要缓冲记忆（ConversationSummaryBufferMemory）

窗口 + 摘要的组合。

from langchain.memory import ConversationSummaryBufferMemory

memory = ConversationSummaryBufferMemory(llm=llm, max_token_limit=100)
conversation = ConversationChain(llm=llm, memory=memory, verbose=False)

for i in range(5):
    conversation.predict(input=f"第{i}条消息：这是测试内容。")
print(memory.load_memory_variables({})["history"])
# 超出 token 限制的历史会被压缩为摘要

5. 实体记忆（ConversationEntityMemory）

抽取并存储实体信息。

from langchain.memory import ConversationEntityMemory

memory = ConversationEntityMemory(llm=llm)
memory.save_context({"input": "小明是程序员，喜欢 Go 语言。"}, {"output": "已记录。"})
memory.save_context({"input": "小红是设计师，喜欢画画。"}, {"output": "好的。"})

print("实体存储：", memory.entity_store.store)
# 输出: {'小明': '小明是程序员，喜欢 Go 语言。', '小红': '小红是设计师，喜欢画画。'}

---

二、长期记忆（Long-Term Memory）

向量存储记忆（VectorStoreRetrieverMemory）

使用 FAISS 和嵌入模型实现语义检索。

from langchain.memory import VectorStoreRetrieverMemory
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import FAISS

# 初始化嵌入模型（首次使用会下载）
embedding_model = HuggingFaceEmbeddings(model_name="paraphrase-multilingual-MiniLM-L12-v2")

# 创建空向量库（需先有一条文本初始化）
vectorstore = FAISS.from_texts(["初始化占位"], embedding_model)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})
memory = VectorStoreRetrieverMemory(retriever=retriever)

# 保存记忆
memory.save_context({"input": "我喜欢科幻电影"}, {"output": "记住了"})
memory.save_context({"input": "最近在学 LangChain"}, {"output": "好的"})

# 检索相关记忆
related = memory.load_memory_variables({"input": "推荐一部电影"})["history"]
print(related)
# 输出会包含“我喜欢科幻电影”

---

三、LangGraph Checkpointer 会话记忆

1. 使用 InMemorySaver（内存版）

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent
from langchain_core.messages import HumanMessage

# 创建 checkpointer
checkpointer = InMemorySaver()

# 创建 Agent（需定义工具，这里用空工具列表）
agent = create_react_agent(llm, tools=[], checkpointer=checkpointer)

config = {"configurable": {"thread_id": "user_123"}}

# 第一轮
agent.invoke({"messages": [HumanMessage(content="我叫小明，我喜欢编程。")]}, config)
# 第二轮，依赖记忆
response = agent.invoke({"messages": [HumanMessage(content="我叫什么名字？")]}, config)
print(response["messages"][-1].content)  # 会回答“小明”

2. 使用 SqliteSaver（持久化）

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

conn = sqlite3.connect("checkpoints.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
checkpointer.setup()

agent = create_react_agent(llm, tools=[], checkpointer=checkpointer)
config = {"configurable": {"thread_id": "persistent_user"}}

agent.invoke({"messages": [HumanMessage(content="我的邮箱是 test@example.com")]}, config)
# 重启程序后，仍可恢复记忆
response = agent.invoke({"messages": [HumanMessage(content="我的邮箱是什么？")]}, config)
print(response["messages"][-1].content)

---

四、高级记忆管理策略（修剪 / 删除 / 总结）

1. 修剪消息（trim_messages）

发送给模型前截断，不改变原始状态。

from langchain_core.messages import trim_messages
from langchain_core.prompts import ChatPromptTemplate

# 构造长消息列表
messages = [
    HumanMessage(content="1"), AIMessage(content="a"),
    HumanMessage(content="2"), AIMessage(content="b"),
    HumanMessage(content="3"), AIMessage(content="c"),
]

trimmer = trim_messages(strategy="last", max_tokens=50, token_counter=llm.get_num_tokens)
trimmed = trimmer.invoke(messages)
print(f"原始 {len(messages)} 条，修剪后 {len(trimmed)} 条")

2. 删除消息（删除 State 中的消息）

直接操作 State（如在 LangGraph 节点中）。

# 伪代码示例（在 LangGraph 节点函数中）
def delete_old_messages(state):
    # 只保留最后 3 条消息
    state["messages"] = state["messages"][-3:]
    return state

3. 总结消息中间件（SummarizationMiddleware）

官方推荐方式，自动将旧消息压缩为摘要。

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
middleware = SummarizationMiddleware(
    model=llm,
    trigger=("messages", 4),   # 消息超过4条时触发总结
    keep=("messages", 2)       # 保留最近2条原始消息
)

agent = create_agent(
    model=llm,
    middleware=[middleware],
    checkpointer=checkpointer
)

config = {"configurable": {"thread_id": "summary_demo"}}
agent.invoke({"messages": [HumanMessage(content="我叫小明。")]}, config)
agent.invoke({"messages": [HumanMessage(content="我是程序员。")]}, config)
agent.invoke({"messages": [HumanMessage(content="我喜欢 Go。")]}, config)
agent.invoke({"messages": [HumanMessage(content="你还记得我的职业吗？")]}, config)
# 此时旧消息已被总结，但模型仍能回答“程序员”

---

总结

记忆类型	策略	适用场景
短期记忆	Buffer / Window / Summary / Entity	单次会话、有限上下文
长期记忆	VectorStoreRetrieverMemory	跨会话知识库、个性化推荐
会话持久化	Checkpointer (InMemory/SQLite/Postgres)	多轮对话、断点续聊
记忆压缩	Trim / Delete / SummarizationMiddleware	防止上下文溢出、优化成本