Langchain模块Memory常用组件

 2025/12/19 

会话缓冲记忆ConversationBufferMemory

将预设的聊天记录保存起来，为之后的聊天提供上下文信息，同时也可以将实时聊天信息保存起来

会话缓冲窗口记忆ConversationBufferWindowMemory

记录最近k次的聊天交互信息

会话缓冲记忆ConversationBufferMemory

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv

load_dotenv()
llm = ChatDeepSeek(model="deepseek-chat",streaming=True)
memory = ConversationBufferMemory(return_messages=True)

# 历史聊天信息，正常情况下不会这样使用
memory.save_context({"input": "你叫啥名字"},{"output": "我叫ShengWang"})

# 直接使用到链中（常用）
conversation = ConversationChain(
    llm=llm,
    memory=memory,  # 将记忆对象传递给链
    verbose=True
)

response = conversation.predict(input="你还记得你叫什么吗？")
print(f"AI: {response}\n")
#
response = conversation.predict(input="师傅你是做什么工作的")
print(f"AI: {response}\n")

print("=== 自动保存的对话历史 ===")
print(memory.buffer)

会话窗口缓冲记忆ConversationBufferWindowMemory

from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationChain
from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv

load_dotenv()
llm = ChatDeepSeek(model="deepseek-chat",streaming=True)

# 设置k=2，save三次
memory = ConversationBufferWindowMemory(k=2)
memory.save_context({"input": "你叫啥名字"},{"output": "我叫ShengWang"})
memory.save_context({"input": "hello"},{"output": "hello"})
memory.save_context({"input": "hello"},{"output": "hello"})

conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

response = conversation.predict(input="你还记得你叫什么吗？")
print(f"AI: {response}\n")

# AI: 当然记得！我是DeepSeek，由深度求索公司创造的AI助手。很高兴为您提供帮助！😊

会话摘要记忆Conversation summary memory

将对话进行总结，将总结存储在内存中。只返回摘要，极端节省token，对于长对话来说非常实用，但丢失细节

from langchain.memory import ConversationSummaryMemory,ChatMessageHistory
from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv
load_dotenv()
llm = ChatDeepSeek(model="deepseek-chat")
memory = ConversationSummaryMemory(
    llm=llm,
    memory_key="history",
    return_messages=True
)
memory.save_context(
    {"input": "你好，我叫张三，我来自北京"},
    {"output": "你好张三！很高兴认识你。北京是个很棒的城市。"}
)
memory.save_context(
    {"input": "我今年25岁，是一名软件工程师"},
    {"output": "真不错！25岁的软件工程师，很有前途的职业。"}
)
memory.save_context(
    {"input": "我喜欢打篮球和旅游"},
    {"output": "运动加旅游是很健康的生活方式！"}
)

# 查看摘要
print(memory.buffer)

# 使用新对话
from langchain.chains import ConversationChain
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)
response = conversation.predict(input="你还记得关于我的哪些信息？")
print(f"AI: {response}")

当然记得！你叫张三，今年25岁，是一位来自北京的软件工程师。你之前提到过自己喜欢打篮球和旅行，我觉得这是一个非常棒的生活方式——既能锻炼身体，又能开阔眼界。北京这座城市充满活力，而软件工程又是一个充满前景的领域，相信你在这条路上一定有很多有趣的故事吧？

会话摘要记忆缓冲ConversationSummaryBufferMemory（略）

是会话摘要记忆的升级版，同样用法

保留最近的N条原始对话 + 将早期对话压缩为摘要，既保留近期细节，又压缩早期内容

max_token_limit 控制总token数，max_len（可选）控制保留的对话轮数

但需要支持token计算的模型，比如openAI。而deepseek不支持

向量存储后端记忆Vector store-backend memory

将对话历史存储在向量数据库中，通过语义相似度检索来获取相关记忆的方法。它与传统记忆方法完全不同，更像是一个”长期记忆”或”知识库”。

from langchain.memory import VectorStoreRetrieverMemory
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import ConversationChain
from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv
import os

load_dotenv()

# 初始化模型和embedding
llm = ChatDeepSeek(model="deepseek-chat")
embeddings = OpenAIEmbeddings()  # 需要OpenAI API key
# 或者使用其他embedding模型，如HuggingFaceEmbeddings

# 创建向量存储
vectorstore = Chroma(
    embedding_function=embeddings,
    persist_directory="./vector_db"  # 持久化存储
)

# 创建向量记忆
memory = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever(
        search_kwargs={"k": 3}  # 返回最相似的3条
    ),
    memory_key="chat_history",
    return_docs=True  # 返回文档对象
)

print("=== 向量存储记忆示例 ===")

# 添加历史对话（可以来自不同会话）
historical_conversations = [
    "用户: 我叫张三，来自北京\n助手: 你好张三，北京是个很棒的城市！",
    "用户: 我今年30岁，是一名软件工程师\n助手: 30岁的软件工程师，经验丰富啊。",
    "用户: 我喜欢打篮球和旅游\n助手: 运动加旅游是很好的生活方式。",
    "用户: 我养了一只叫Tom的猫\n助手: Tom一定很可爱！",
    "用户: 最近在学习Python编程\n助手: Python是很好的入门语言。",
    "用户: 想转行做数据科学家\n助手: 数据科学家需要统计学和编程技能。",
    "用户: 我经常去健身房锻炼\n助手: 健身对身体和心理健康都很重要。",
    "用户: 去年去了日本旅行\n助手: 日本的文化和风景都很迷人。",
    "用户: 我的妻子是医生\n助手: 医生是很崇高的职业。",
    "用户: 我们计划明年要孩子\n助手: 家庭计划很重要，祝你们幸福。",
]

print("添加历史对话到向量库...")
for conv in historical_conversations:
    memory.save_context(
        {"input": conv.split("\n")[0].replace("用户: ", "")},
        {"output": conv.split("\n")[1].replace("助手: ", "")}
    )

# 测试检索
print("\n测试语义检索:")
test_queries = [
    "关于我的个人信息",
    "我的兴趣爱好",
    "我的家庭情况",
    "我的职业发展",
]

for query in test_queries:
    print(f"\n查询: {query}")
    docs = memory.retriever.invoke(query)
    for i, doc in enumerate(docs, 1):
        print(f"  相关记忆 {i}: {doc.page_content[:80]}...")

  相关记忆 1: input: 我喜欢打篮球和旅游
output: 运动加旅游是很好的生活方式。...
  相关记忆 2: input: 我经常去健身房锻炼
output: 健身对身体和心理健康都很重要。...
  相关记忆 3: input: 我的妻子是医生
output: 医生是很崇高的职业。...

查询: 我的兴趣爱好
  相关记忆 1: input: 我养了一只叫Tom的猫
output: Tom一定很可爱！...
  相关记忆 2: input: 我的妻子是医生
output: 医生是很崇高的职业。...
  相关记忆 3: input: 我喜欢打篮球和旅游
output: 运动加旅游是很好的生活方式。...

查询: 我的家庭情况
  相关记忆 1: input: 我们计划明年要孩子
output: 家庭计划很重要，祝你们幸福。...
  相关记忆 2: input: 我的妻子是医生
output: 医生是很崇高的职业。...
  相关记忆 3: input: 我喜欢打篮球和旅游
output: 运动加旅游是很好的生活方式。...

查询: 我的职业发展
  相关记忆 1: input: 我养了一只叫Tom的猫
output: Tom一定很可爱！...
  相关记忆 2: input: 我的妻子是医生
output: 医生是很崇高的职业。...
  相关记忆 3: input: 我喜欢打篮球和旅游
output: 运动加旅游是很好的生活方式。...