LangChain-06-Memory（记忆）

dijk

655人浏览 · 2026-04-13 17:39:02

dijk · 2026-04-13 17:39:02 发布

LangChain之Memory（记忆）

Memory概述与上下文管理

什么是Memory？

在LangChain中，Memory（记忆）组件负责在对话或交互过程中存储、检索和管理信息。它使AI能够"记住"之前的交互内容，从而实现更连贯、更有上下文感知能力的对话。

为什么需要Memory？

大型语言模型（LLM）本身是无状态的，它们不会自动记住之前的对话内容。每次请求都是独立的，模型无法访问之前的交互历史。Memory组件解决了这个问题，通过：

维护对话历史：存储用户和AI之间的交互记录
提供上下文：将相关历史信息注入到当前提示中
实现连续对话：使AI能够引用之前讨论过的内容
个性化体验：记住用户偏好和特定信息

Memory的核心功能

存储机制：如何保存对话历史
检索策略：如何从历史中提取相关信息
格式化输出：如何将记忆内容整合到提示中
容量管理：如何处理长对话和限制记忆大小

Memory的工作原理

LangChain的Memory组件通常通过以下方式工作：

对话后保存：每次交互完成后，将对话内容存储起来
提示前加载：在生成新响应前，从记忆中检索相关内容
上下文注入：将检索到的内容整合到当前提示中
格式化处理：将记忆内容转换为模型可理解的格式

ConversationBufferMemory

概述

ConversationBufferMemory是最简单的Memory类型，它将整个对话历史以原始形式存储在缓冲区中。它保存所有交互的完整记录，并在需要时将它们提供给模型。

特点

简单直接：无需复杂的处理逻辑
完整记录：保存所有对话内容，不进行任何筛选或压缩
线性增长：记忆大小随对话长度线性增加
精确上下文：提供完整的对话历史

实现方式

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

# 创建记忆实例
memory = ConversationBufferMemory()

# 创建对话链
llm = OpenAI(temperature=0)
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 进行对话
conversation.predict(input="你好，我想了解人工智能的发展历史。")
conversation.predict(input="能详细介绍一下深度学习吗？")

# 查看记忆内容
print(memory.buffer)

内部结构

ConversationBufferMemory内部维护两个主要组件：

chat_memory：存储消息历史的对象
buffer：格式化后的对话字符串

适用场景

短期对话：适合不需要长期记忆的应用
简单应用：不需要复杂记忆管理的场景
原型开发：快速构建和测试对话系统
完整上下文：需要完整对话历史的场景

限制与缺点

内存消耗：随着对话增长，内存使用量线性增加
Token限制：可能超出模型的上下文窗口限制
处理效率：长对话会导致处理时间增加
信息冗余：可能包含大量不相关的旧信息

ConversationBufferWindowMemory

概述

ConversationBufferWindowMemory是ConversationBufferMemory的变体，它只保留最近的K轮对话，而不是整个对话历史。这通过滑动窗口机制控制记忆大小。

特点

固定窗口：只保留最近的指定数量的交互
内存可控：通过窗口大小限制内存使用
上下文新鲜：专注于最近的对话内容
自动淘汰：自动移除超出窗口的旧对话

实现方式

from langchain.memory import ConversationBufferWindowMemory
from langchain.llms import OpenAI
from langchain.chains import ConversationChain

# 创建窗口记忆实例，保留最近3轮对话
memory = ConversationBufferWindowMemory(k=3)

# 创建对话链
llm = OpenAI(temperature=0)
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 进行多轮对话
conversation.predict(input="你好，我想了解机器学习。")
conversation.predict(input="什么是监督学习？")
conversation.predict(input="能解释一下决策树吗？")
conversation.predict(input="随机森林和决策树有什么区别？")
conversation.predict(input="深度学习和机器学习的关系是什么？")

# 查看记忆内容（只有最近3轮）
print(memory.buffer)

窗口机制

窗口记忆的工作原理：

初始化：设置窗口大小K
添加新对话：将新交互添加到记忆中
检查容量：如果超过K轮，移除最早的交互
保持最新：始终保留最近的K轮对话

适用场景

长期对话：需要处理长对话但限制上下文大小
资源受限：内存或Token使用受限的环境
关注近期：更关注最近对话内容的应用
流式对话：连续对话但不需要完整历史的场景

参数配置

k：窗口大小，保留的对话轮数
return_messages：是否返回消息对象而非字符串

限制与缺点

信息丢失：超出窗口的对话内容永久丢失
上下文断裂：可能丢失重要的长期上下文
窗口选择：选择合适的窗口大小可能需要实验
话题跳跃：可能无法理解跨越窗口的话题关联

ConversationSummaryMemory

概述

ConversationSummaryMemory使用LLM来总结对话历史，而不是存储原始对话内容。它定期生成对话摘要，并将摘要作为记忆内容，从而大大减少Token使用量。

特点

摘要压缩：将对话内容压缩为摘要形式
Token高效：显著减少Token使用量
动态更新：随着对话进行更新摘要
语义保持：保留对话的核心语义信息

实现方式

from langchain.memory import ConversationSummaryMemory
from langchain.llms import OpenAI
from langchain.chains import ConversationChain

# 创建摘要记忆实例
memory = ConversationSummaryMemory(llm=OpenAI(temperature=0))

# 创建对话链
conversation = ConversationChain(
    llm=OpenAI(temperature=0),
    memory=memory,
    verbose=True
)

# 进行多轮对话
conversation.predict(input="我想了解量子计算的基本原理。")
conversation.predict(input="量子比特和经典比特有什么区别？")
conversation.predict(input="量子纠缠是如何工作的？")

# 查看摘要内容
print(memory.buffer)

摘要生成机制

摘要记忆的工作流程：

初始摘要：开始时为空或使用初始提示
新对话添加：将新交互添加到当前摘要中
摘要更新：使用LLM重新生成包含新内容的摘要
摘要存储：保存更新后的摘要作为记忆

摘要提示工程

摘要质量依赖于提示设计：

from langchain.prompts import PromptTemplate

# 自定义摘要提示
summary_prompt = PromptTemplate(
    input_variables=["summary", "new_lines"],
    template="Progressively summarize the following lines of conversation.\n\nCurrent summary:\n{summary}\n\nNew lines of conversation:\n{new_lines}\n\nNew summary:"
)

memory = ConversationSummaryMemory(
    llm=OpenAI(temperature=0),
    prompt=summary_prompt
)

适用场景

长对话：需要处理大量交互但受Token限制
主题持久：对话围绕特定主题展开
信息压缩：需要保留核心信息但减少存储
成本敏感：希望减少API调用成本的应用

限制与缺点

信息丢失：摘要可能丢失细节和微妙之处
摘要质量：依赖于LLM的摘要能力
更新成本：每次更新需要调用LLM
累积误差：多次摘要可能导致信息失真

ConversationSummaryBufferMemory

概述

ConversationSummaryBufferMemory结合了摘要记忆和缓冲区记忆的优点。它保留最近的对话原始内容，同时对较早的对话进行摘要，实现了平衡的记忆管理。

特点

混合策略：结合摘要和缓冲区两种方法
双重存储：近期对话保持原始形式，早期对话存储为摘要
智能切换：根据对话轮数决定何时开始摘要
灵活平衡：在详细上下文和Token效率间取得平衡

实现方式

from langchain.memory import ConversationSummaryBufferMemory
from langchain.llms import OpenAI
from langchain.chains import ConversationChain

# 创建摘要缓冲区记忆实例
# max_token_limit: 触发摘要的最大Token数
memory = ConversationSummaryBufferMemory(
    llm=OpenAI(temperature=0),
    max_token_limit=100
)

# 创建对话链
conversation = ConversationChain(
    llm=OpenAI(temperature=0),
    memory=memory,
    verbose=True
)

# 进行多轮对话
conversation.predict(input="我想了解区块链技术。")
conversation.predict(input="什么是去中心化？")
conversation.predict(input="智能合约是如何工作的？")
conversation.predict(input="区块链有哪些实际应用？")

# 查看记忆内容
print("Buffer:", memory.chat_memory.messages)
print("Summary:", memory.summary)

工作机制

摘要缓冲区记忆的工作流程：

初始阶段：对话开始时，所有内容存储在缓冲区中
容量监控：监控缓冲区大小（Token数）
摘要触发：当超过阈值时，将部分内容摘要化
内容转移：将摘要内容移至摘要部分，保留最新内容在缓冲区
动态平衡：根据需要重复此过程

参数配置

max_token_limit：触发摘要的最大Token数
llm：用于生成摘要的语言模型
prompt：自定义摘要提示模板

适用场景

中等长度对话：比简单缓冲区长但不需要完整摘要
平衡需求：既需要近期细节又需要长期上下文
资源优化：希望在质量和效率间取得平衡
动态对话：对话长度不确定的应用

限制与缺点

复杂度增加：比单一策略更复杂
阈值选择：需要合理设置Token阈值
切换点：摘要和缓冲区的切换可能不自然
管理开销：需要管理两种不同类型的内容

ConversationTokenBufferMemory

概述

ConversationTokenBufferMemory基于Token数量管理对话历史，而不是基于交互轮数。它保留足够的内容以保持在指定的Token限制内，确保不会超出模型的上下文窗口。

特点

Token精确：基于Token数量而非交互次数
自适应裁剪：根据Token限制智能裁剪内容
上下文完整：尽可能保留完整的对话上下文
模型兼容：确保记忆内容符合模型限制

实现方式

from langchain.memory import ConversationTokenBufferMemory
from langchain.llms import OpenAI
from langchain.chains import ConversationChain

# 创建基于Token的记忆实例
memory = ConversationTokenBufferMemory(
    llm=OpenAI(temperature=0),
    max_token_limit=200  # 最大Token数
)

# 创建对话链
conversation = ConversationChain(
    llm=OpenAI(temperature=0),
    memory=memory,
    verbose=True
)

# 进行多轮对话
conversation.predict(input="我想了解可再生能源技术。")
conversation.predict(input="太阳能和风能有什么区别？")
conversation.predict(input="氢能源的发展前景如何？")

# 查看记忆内容
print(memory.buffer)

Token管理机制

Token缓冲区记忆的工作原理：

Token计算：计算当前对话内容的Token数量
容量检查：比较当前Token数与限制
内容裁剪：如果超限，从最早的内容开始裁剪
动态平衡：确保总Token数不超过限制
完整性保持：尽可能保持对话的连续性

Token计算方式

from langchain.memory import ConversationTokenBufferMemory
from langchain.llms import OpenAI

# 使用不同的Token计算方法
memory = ConversationTokenBufferMemory(
    llm=OpenAI(temperature=0),
    max_token_limit=200,
    # 可选：自定义Token计数器
    # tokenizer=None  # 使用默认的LLM tokenizer
)

# 检查当前Token使用量
token_count = memory.llm.get_num_tokens(memory.buffer)
print(f"Current token count: {token_count}")

适用场景

严格限制：需要精确控制Token使用的场景
多模型支持：使用不同上下文大小的模型
API优化：希望最大化利用API Token限制
成本控制：需要严格控制API调用成本

限制与缺点

对话断裂：裁剪可能导致对话上下文不完整
Token计算：Token计算可能增加处理开销
内容丢失：可能丢失重要的早期信息
模型依赖：Token计算依赖于特定的tokenizer

ConversationKGMemory

概述

ConversationKGMemory（Knowledge Graph Memory）使用知识图谱来表示和存储对话中的实体及其关系。它从对话中提取实体和关系，构建知识图谱，并基于此提供上下文。

特点

结构化表示：将对话内容转换为结构化知识图谱
实体关系：识别和存储实体间的语义关系
上下文推理：基于图谱进行推理和关联
知识积累：随着对话不断丰富知识图谱

实现方式

from langchain.memory import ConversationKGMemory
from langchain.llms import OpenAI
from langchain.chains import ConversationChain

# 创建知识图谱记忆实例
memory = ConversationKGMemory(llm=OpenAI(temperature=0))

# 创建对话链
conversation = ConversationChain(
    llm=OpenAI(temperature=0),
    memory=memory,
    verbose=True
)

# 进行包含实体和关系的对话
conversation.predict(input="张三是AI公司的数据科学家。")
conversation.predict(input="他负责开发机器学习模型。")
conversation.predict(input="他的团队有5名工程师。")

# 查看知识图谱内容
print("Knowledge Graph:", memory.knowledge_graph)

# 获取当前上下文
print("Current entities:", memory.get_current_entities())

知识图谱构建

知识图谱记忆的工作流程：

实体识别：从对话中识别关键实体
关系提取：分析实体间的关系
图谱构建：将实体和关系添加到知识图谱中
上下文生成：基于图谱生成对话上下文
图谱更新：随着新对话不断更新图谱

实体和关系提取

from langchain.prompts import PromptTemplate

# 自定义实体和关系提取提示
entity_extraction_prompt = PromptTemplate(
    input_variables=["chat_history", "input"],
    template="""Extract entities and their relationships from the following conversation.
    
    Conversation History:
    {chat_history}
    
    New Input:
    {input}
    
    Extract entities and relationships in format: (entity1)-[relation]->(entity2)
    """
)

memory = ConversationKGMemory(
    llm=OpenAI(temperature=0),
    prompt=entity_extraction_prompt
)

适用场景

关系密集：对话包含大量实体和关系的场景
知识积累：需要长期积累和关联知识的应用
复杂推理：需要理解实体间复杂关系的任务
专业领域：特定领域的专业对话系统

限制与缺点

提取准确性：依赖LLM的实体和关系提取能力
图谱复杂度：可能产生复杂且难以管理的图谱
上下文生成：从图谱生成自然上下文可能不自然
计算开销：图谱构建和查询可能增加计算成本

VectorStoreRetrieverMemory

概述

VectorStoreRetrieverMemory使用向量存储和检索机制来管理对话记忆。它将对话内容转换为向量，存储在向量数据库中，并基于语义相似度检索相关内容。

特点

语义检索：基于语义相似度而非时间顺序检索
向量存储：使用高效的向量数据库存储对话内容
可扩展性：能够处理大量对话历史
灵活检索：支持多种检索策略和参数

实现方式

from langchain.memory import VectorStoreRetrieverMemory
from langchain.llms import OpenAI
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import ConversationChain

# 创建向量存储
embedding_model = OpenAIEmbeddings()
vectorstore = Chroma(embedding_function=embedding_model)

# 创建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# 创建向量存储记忆
memory = VectorStoreRetrieverMemory(
    retriever=retriever,
    # 当没有检索到相关内容时返回的消息
    memory_key="relevant_history",
    input_key="input"
)

# 创建对话链
conversation = ConversationChain(
    llm=OpenAI(temperature=0),
    memory=memory,
    verbose=True
)

# 进行多轮对话
conversation.predict(input="我想了解机器学习中的过拟合问题。")
conversation.predict(input="如何防止过拟合？")
conversation.predict(input="什么是正则化？")

# 测试记忆检索
print("Retrieved context:", memory.load_memory_variables({"input": "过拟合"}))

向量化与检索机制

向量存储记忆的工作流程：

内容向量化：将对话内容转换为向量表示
向量存储：将向量存储在向量数据库中
相似度检索：基于当前输入检索相似的历史内容
上下文组装：将检索到的内容组装为上下文
动态更新：新对话内容自动添加到向量存储

检索策略配置

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

# 使用不同的向量存储和检索策略
vectorstore = FAISS.from_texts(
    ["示例文本1", "示例文本2"],
    embedding_model
)

# 配置检索参数
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={
        "k": 5,
        "score_threshold": 0.5
    }
)

memory = VectorStoreRetrieverMemory(
    retriever=retriever,
    # 自定义检索后的处理
    return_docs=True
)

适用场景

长对话历史：需要处理大量历史对话的场景
语义关联：需要基于语义而非时间检索内容
多话题对话：对话涉及多个不同话题的情况
知识密集：需要从大量历史中检索特定知识

限制与缺点

上下文断裂：可能丢失对话的时间顺序和连续性
检索质量：依赖向量化和相似度计算的质量
存储需求：需要额外的向量存储资源
计算成本：向量化和检索可能增加计算开销

Memory选择指南

选择考虑因素

选择合适的Memory类型需要考虑以下因素：

对话长度
- 短对话：ConversationBufferMemory
- 中等长度：ConversationBufferWindowMemory或ConversationSummaryBufferMemory
- 长对话：ConversationSummaryMemory或VectorStoreRetrieverMemory
上下文需求
- 完整上下文：ConversationBufferMemory
- 近期上下文：ConversationBufferWindowMemory
- 语义上下文：VectorStoreRetrieverMemory
- 关系上下文：ConversationKGMemory
资源限制
- 内存限制：ConversationBufferWindowMemory或ConversationTokenBufferMemory
- Token限制：ConversationSummaryMemory或ConversationTokenBufferMemory
- 计算资源：避免VectorStoreRetrieverMemory或ConversationKGMemory
应用场景
- 简单聊天：ConversationBufferMemory
- 客服系统：ConversationSummaryBufferMemory
- 知识问答：VectorStoreRetrieverMemory
- 专业咨询：ConversationKGMemory

决策矩阵

场景	推荐Memory	替代方案
简单对话，短期交互	ConversationBufferMemory	ConversationBufferWindowMemory
长期对话，资源受限	ConversationSummaryMemory	ConversationTokenBufferMemory
需要近期细节和长期摘要	ConversationSummaryBufferMemory	自定义混合方案
多话题，语义检索	VectorStoreRetrieverMemory	ConversationKGMemory
实体关系密集	ConversationKGMemory	VectorStoreRetrieverMemory
严格Token限制	ConversationTokenBufferMemory	ConversationSummaryMemory

组合策略

在某些复杂场景中，可以组合使用多种Memory类型：

from langchain.memory import CombinedMemory
from langchain.memory import ConversationBufferMemory
from langchain.memory import ConversationSummaryMemory

# 创建组合记忆
conv_memory = ConversationBufferMemory(memory_key="chat_history")
summary_memory = ConversationSummaryMemory(llm=OpenAI(), memory_key="summary")

# 组合记忆
combined_memory = CombinedMemory(
    memories=[conv_memory, summary_memory],
    input_key="input"
)

# 在链中使用组合记忆
conversation = ConversationChain(
    llm=OpenAI(temperature=0),
    memory=combined_memory,
    verbose=True
)

最佳实践与注意事项

性能优化

定期清理

# 定期清理过期记忆
if len(memory.chat_memory.messages) > 100:
    memory.chat_memory.messages = memory.chat_memory.messages[-50:]

异步处理

# 使用异步记忆更新
from langchain.memory import AsyncMemory

async def update_memory_async(memory, input, output):
    await memory.save_context({"input": input}, {"output": output})

批量处理

# 批量保存对话历史
def save_conversation_batch(memory, conversations):
    for input_text, output_text in conversations:
        memory.save_context({"input": input_text}, {"output": output_text})

错误处理

记忆容量管理

# 防止记忆溢出
try:
    memory.save_context({"input": user_input}, {"output": ai_response})
except Exception as e:
    # 处理记忆溢出错误
    print(f"Memory error: {e}")
    # 清理或重置记忆
    memory.clear()

检索失败处理

# 处理检索失败
try:
    context = memory.load_memory_variables({"input": query})
except Exception as e:
    print(f"Retrieval error: {e}")
    context = {"history": "无法检索相关历史记录"}

安全与隐私

敏感信息过滤

# 过滤敏感信息
def filter_sensitive_info(text):
    # 实现敏感信息过滤逻辑
    return filtered_text

# 在保存前过滤
memory.save_context(
    {"input": filter_sensitive_info(user_input)},
    {"output": filter_sensitive_info(ai_response)}
)

记忆加密

# 加密敏感记忆内容
from cryptography.fernet import Fernet

class EncryptedMemory(ConversationBufferMemory):
    def __init__(self, key, **kwargs):
        super().__init__(**kwargs)
        self.cipher = Fernet(key)
    
    def save_context(self, inputs, outputs):
        # 加密后保存
        encrypted_inputs = {k: self.cipher.encrypt(v.encode()) for k, v in inputs.items()}
        encrypted_outputs = {k: self.cipher.encrypt(v.encode()) for k, v in outputs.items()}
        super().save_context(encrypted_inputs, encrypted_outputs)