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

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

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

环境准备

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

pipinstalllangchain langchain-openai langchain-community chromadb faiss-cpu tiktoken# 如果需要 SQLite checkpointerpipinstalllanggraph-checkpoint-sqlite

创建.env文件：

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

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

importosfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAI 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）

保存全部历史，最直接。

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

2. 会话窗口记忆（ConversationBufferWindowMemory）

只保留最近 k 轮对话。

fromlangchain.memoryimportConversationBufferWindowMemory 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 将历史压缩为摘要。

fromlangchain.memoryimportConversationSummaryMemory 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）

窗口 + 摘要的组合。

fromlangchain.memoryimportConversationSummaryBufferMemory memory=ConversationSummaryBufferMemory(llm=llm,max_token_limit=100)conversation=ConversationChain(llm=llm,memory=memory,verbose=False)foriinrange(5):conversation.predict(input=f"第{i}条消息：这是测试内容。")print(memory.load_memory_variables({})["history"])# 超出 token 限制的历史会被压缩为摘要

5. 实体记忆（ConversationEntityMemory）

抽取并存储实体信息。

fromlangchain.memoryimportConversationEntityMemory 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 和嵌入模型实现语义检索。

fromlangchain.memoryimportVectorStoreRetrieverMemoryfromlangchain.embeddingsimportHuggingFaceEmbeddingsfromlangchain.vectorstoresimportFAISS# 初始化嵌入模型（首次使用会下载）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（内存版）

fromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.prebuiltimportcreate_react_agentfromlangchain_core.messagesimportHumanMessage# 创建 checkpointercheckpointer=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（持久化）

importsqlite3fromlanggraph.checkpoint.sqliteimportSqliteSaver 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）

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

fromlangchain_core.messagesimporttrim_messagesfromlangchain_core.promptsimportChatPromptTemplate# 构造长消息列表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 节点函数中）defdelete_old_messages(state):# 只保留最后 3 条消息state["messages"]=state["messages"][-3:]returnstate

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

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

fromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimportSummarizationMiddlewarefromlanggraph.checkpoint.memoryimportInMemorySaver 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)# 此时旧消息已被总结，但模型仍能回答“程序员”

总结