深入 `VectorStoreRetrieverMemory`：构建一个具备“长久遗忘曲线”的无限容量对话记忆系统

各位技术同仁，下午好！

今天，我们齐聚一堂，共同深入探讨一个在大型语言模型（LLM）应用开发中至关重要，且充满挑战的议题——构建具备“长久遗忘曲线”的无限容量对话记忆系统。在与LLM交互的过程中，我们都曾遇到过模型“健忘”的问题，它无法记住稍早的对话细节，导致交互中断，体验不连贯。传统的记忆机制，如固定窗口记忆，虽然简单有效，但其容量的限制，使其在处理复杂、长时间的对话场景时显得力不从心。

我们将聚焦于LangChain框架中的 VectorStoreRetrieverMemory，并在此基础上，巧妙地融入“遗忘”机制，以期达到既能无限扩展记忆容量，又能模拟人类记忆的衰减特性，从而构建出更加智能、高效且符合直觉的对话系统。

一、 对话记忆的挑战与机遇

在探讨具体实现之前，我们首先需要理解对话记忆在LLM应用中的核心价值与面临的挑战。

1.1 对话记忆的重要性

一个没有记忆的对话系统，就像一个每次见面都需要重新介绍自己的陌生人。在人机交互中，记忆赋予了系统以下关键能力：

• 上下文理解： 能够理解用户当前表达的深层含义，因为它知道之前的对话背景。
• 连贯性与一致性： 维持对话的逻辑连贯性，避免重复提问或提供不一致的信息。
• 个性化体验： 记住用户的偏好、历史行为和特定信息，从而提供定制化的服务。
• 复杂任务处理： 支持用户在多次交互中逐步完成复杂任务，无需一次性提供所有信息。

1.2 传统记忆机制的局限

LangChain提供了多种内置记忆类型，其中最常用的是：

• ConversationBufferMemory: 简单地存储所有对话历史。
• ConversationBufferWindowMemory: 只存储最近N轮对话。

记忆类型	优点	缺点	适用场景
ConversationBufferMemory	简单易用，保留所有历史	内存无限增长，超出LLM上下文窗口限制，效率低	短对话，演示
ConversationBufferWindowMemory	控制内存大小，避免超出LLM上下文窗口	无法记住N轮之前的关键信息，"短期失忆"严重	对话轮次有限，对历史依赖不高的场景

这些传统机制，尤其是在面对需要长时间跟踪上下文、或者用户与系统存在海量交互历史的场景时，显得捉襟见肘。我们亟需一种能够突破固定窗口限制，同时又能智能管理信息，避免“记忆过载”的解决方案。

1.3 无限容量与“遗忘曲线”的愿景

我们理想中的记忆系统，应该具备以下特性：

• 无限容量： 理论上可以存储用户与系统的所有历史交互，不因对话轮次增多而受限。
• 智能检索： 在需要时，能够高效地从海量历史中检索出与当前对话最相关的信息。
• 长久遗忘曲线： 能够模拟人类记忆的特性——近期发生的事情记得清楚，久远的事情逐渐模糊、遗忘，但重要的、频繁提及的事件仍能被记住。这有助于减轻存储和检索负担，并提升相关性。

这正是 VectorStoreRetrieverMemory 结合自定义遗忘策略所能带来的机遇。

二、 核心概念解析：VectorStore、Retriever与Memory

在深入 VectorStoreRetrieverMemory 之前，我们必须先理解其构成基石：VectorStore、Retriever 和 Memory。

2.1 VectorStore：语义记忆的基石

VectorStore（向量数据库）是存储和检索高维向量数据的专门数据库。在LLM语境下，这些向量通常是文本的语义嵌入（Embeddings）。

• Embedding（嵌入）： 是一种将文本（词、短语、句子或文档）转换为固定长度数值向量的技术。这些向量捕捉了文本的语义信息，使得语义相似的文本在向量空间中距离更近。
• 语义相似度搜索： 通过计算查询文本的Embedding与数据库中所有Embedding的距离（如余弦相似度），找出语义上最相关的文本。

常用 VectorStore 示例：

VectorStore 类型	描述	特点
FAISS	Facebook AI Similarity Search，本地向量库	内存驻留，快速，轻量级，适合小型至中型应用
Chroma	开源嵌入数据库，支持本地和客户端/服务器模式	易用，功能丰富，适合本地开发和中小型部署
Pinecone	云原生向量数据库	高扩展性，高性能，适合大规模生产环境
Weaviate	开源向量数据库，支持语义搜索和图谱查询	功能强大，灵活，支持复杂数据模型

我们将在本文中主要使用FAISS进行演示，因为它易于设置和本地运行。

2.2 Retriever：从记忆中提取相关信息

Retriever（检索器）的职责是从 VectorStore 或其他数据源中，根据用户查询或当前上下文，提取出最相关的信息片段（documents）。

• 工作原理： Retriever 接收一个查询（通常是当前的用户输入或对话的总结），将其转换为Embedding，然后在 VectorStore 中执行相似度搜索，返回Top-K个最相关的文档。
• 检索策略： 除了简单的相似度搜索，Retriever 还可以采用更复杂的策略，如MMR（Maximal Marginal Relevance），以返回多样化的结果，避免检索到的文档过于相似。

2.3 Memory：LLM应用的记忆模块

在LangChain中，Memory 模块负责管理和维护LLM与用户之间的对话历史。它提供了一致的接口来：

• 保存（Save）： 将新的用户输入和AI响应添加到记忆中。
• 加载（Load）： 从记忆中检索出历史对话，并以特定格式（通常是字符串或消息列表）提供给LLM作为上下文。

VectorStoreRetrieverMemory 正是将 VectorStore 和 Retriever 的能力，巧妙地集成到 Memory 接口中，从而实现了基于语义检索的对话记忆。

三、 VectorStoreRetrieverMemory 的工作原理

VectorStoreRetrieverMemory 的核心思想是：将每一次对话（用户输入和AI响应）作为一个独立的“记忆单元”，将其转换为Embedding并存储在 VectorStore 中。当需要回忆历史时，它不会简单地加载所有历史，而是根据当前的对话内容，语义地检索出最相关的历史片段，作为上下文提供给LLM。

3.1 存储机制

每次对话结束后，VectorStoreRetrieverMemory 会将用户输入和AI响应拼接成一个文本块，然后：

1. 使用预设的Embedding模型将其转换为向量。
2. 将该向量及其对应的文本块（以及可选的元数据）存储到底层的 VectorStore 中。

3.2 检索机制

当LLM需要上下文时，VectorStoreRetrieverMemory 会执行以下步骤：

1. 获取当前的最新用户输入。
2. 将最新用户输入转换为Embedding。
3. 使用 VectorStore 执行相似度搜索，找出与当前输入语义上最相关的N个历史记忆单元。
4. 将这些检索到的历史记忆单元格式化，并与当前输入一起作为上下文提供给LLM。

通过这种方式，即使记忆库中有数百万条历史记录，LLM也只会看到最相关的少数几条，从而有效解决了LLM上下文窗口限制的问题，并实现了“无限容量”的愿景。

四、 构建基础记忆系统

现在，让我们通过代码来构建一个基于 VectorStoreRetrieverMemory 的基础对话记忆系统。

4.1 环境准备与库安装

首先，我们需要安装必要的库：LangChain、OpenAI（用于Embedding模型和LLM）、FAISS（作为本地VectorStore）和Tiktoken（用于token计算）。

pip install langchain openai faiss-cpu tiktoken

4.2 初始化组件

我们将需要一个Embedding模型、一个LLM模型以及一个VectorStore。

import os
import time
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import VectorStoreRetrieverMemory
from langchain_core.output_parsers import StrOutputParser

# 设置OpenAI API密钥
# 建议通过环境变量设置，这里为了演示直接写入，实际应用中请避免
os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# 1. 初始化Embedding模型
# 我们使用OpenAI的text-embedding-ada-002模型
embeddings_model = OpenAIEmbeddings(model="text-embedding-ada-002")

# 2. 初始化LLM模型
# 使用GPT-3.5-turbo作为对话模型
llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7)

# 3. 初始化VectorStore（这里使用FAISS作为示例）
# 最初，VectorStore是空的，我们将通过对话逐步填充
# 注意：FAISS需要一个`embeddings_model`来处理文本
vectorstore = FAISS.from_texts(["初始化记忆，等待对话。"], embeddings_model)
# 移除初始化的占位符，因为我们希望从干净的记忆开始
# 实际上，VectorStoreRetrieverMemory在第一次保存时会自动添加，这里只是为了初始化FAISS对象
vectorstore.delete(vectorstore.index_to_docstore_id.keys())

print("核心组件初始化完成：Embedding模型、LLM模型、FAISS VectorStore。")

4.3 实例化 VectorStoreRetrieverMemory

现在，我们可以创建 VectorStoreRetrieverMemory 实例。重要的是要配置 retriever_kwargs，它允许我们指定检索器如何从 VectorStore 中获取文档，例如 k 值（要检索的文档数量）。

# 4. 实例化VectorStoreRetrieverMemory
# 我们需要一个检索器，它知道如何从vectorstore中获取文档
# 这里使用vectorstore.as_retriever()来创建一个默认的检索器
# retriever_kwargs可以配置检索器的行为，例如检索多少个文档
memory = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3}), # 检索最近3个最相关的记忆片段
    memory_key="chat_history", # 在prompt中用于引用历史对话的键
    return_messages=True # 返回消息对象列表而不是单个字符串
)

print("VectorStoreRetrieverMemory 初始化完成。")

4.4 集成到对话链并进行交互

为了演示，我们将构建一个简单的对话链，将LLM、Prompt和我们刚才创建的Memory连接起来。

# 5. 定义对话Prompt模板
prompt = ChatPromptTemplate.from_messages(
    [
        SystemMessage("你是一个友好的AI助手，擅长根据历史对话提供帮助。"),
        MessagesPlaceholder(variable_name="chat_history"), # 记忆将注入到这里
        HumanMessage(content="{input}") # 当前用户输入
    ]
)

# 6. 构建一个简单的Runnable链
# input -> load_memory -> prompt -> llm -> save_memory -> output
# 使用RunnablePassthrough将当前输入传递给LLM
# 并通过.assign()来加载和保存记忆
conversation_chain = (
    RunnablePassthrough.assign(
        chat_history=lambda x: memory.load_memory_variables(x)["chat_history"]
    )
    | prompt
    | llm
    | StrOutputParser()
)

# 7. 定义一个封装函数来运行对话并保存记忆
def chat_with_memory(user_input: str, current_memory: VectorStoreRetrieverMemory):
    print(f"n用户: {user_input}")

    # 调用链获取AI响应
    ai_response = conversation_chain.invoke({"input": user_input})
    print(f"AI: {ai_response}")

    # 保存当前轮次的对话到记忆中
    # VectorStoreRetrieverMemory会自动处理Embedding并存储到VectorStore
    current_memory.save_context({"input": user_input}, {"output": ai_response})
    return ai_response

print("n开始对话...")

# 进行几轮对话
chat_with_memory("你好，我叫张三。", memory)
chat_with_memory("我喜欢编程，尤其是Python。", memory)
chat_with_memory("你觉得Python有哪些流行的框架？", memory)
chat_with_memory("我之前跟你说过我的名字叫什么？", memory)
chat_with_memory("我最喜欢的编程语言是什么？", memory)

# 额外测试，看是否能记住更早的信息
chat_with_memory("你还记得我最喜欢的编程语言对应的框架有哪些吗？", memory)

print("n对话结束。")

# 检查VectorStore中的实际文档数量
print(f"VectorStore中存储的记忆片段数量: {len(vectorstore.docstore._dict)}")

代码解释：

• 我们通过 memory.load_memory_variables(x)["chat_history"] 从 VectorStoreRetrieverMemory 加载历史记录。这一步会自动触发语义检索。
• memory.save_context({"input": user_input}, {"output": ai_response}) 在每一轮对话结束后，将当前的用户输入和AI响应保存到 VectorStore 中。VectorStoreRetrieverMemory 会自动将这些文本转换为Embedding并存储。
• MessagesPlaceholder(variable_name="chat_history") 是LangChain Prompt模板中的一个占位符，它会在运行时被实际加载的 chat_history 消息列表填充。

通过以上代码，我们已经构建了一个基础的、具备语义检索能力的无限容量记忆系统。它能够根据当前输入，从历史中智能地提取相关信息，避免了传统窗口记忆的局限。

五、 引入“长久遗忘曲线”

尽管我们的系统现在能够存储无限量的历史并进行智能检索，但它仍然缺乏一个关键特性：遗忘。一个真正智能的系统，应该像人类一样，能够区分信息的“新旧”和“重要性”，让不重要的、过时的信息逐渐淡出，甚至被清除，以降低存储和检索的成本，并提升整体效率。

5.1 遗忘的必要性

• 资源优化： 长期积累的海量记忆会增加 VectorStore 的存储空间，并可能减慢检索速度。
• 相关性提升： 过时的信息往往与当前对话的相关性较低，甚至可能引入噪音，影响LLM的判断。
• 模拟人类认知： 模拟遗忘机制，使得系统行为更符合人类直觉。

5.2 实现遗忘策略

我们将主要探讨两种遗忘策略，并重点实现基于时间衰减的遗忘。

a) 基于时间衰减的遗忘 (Time-Decay Forgetting)

这是最直观的遗忘方式，即信息越久远，其重要性或可检索性越低。我们可以通过在存储记忆时添加时间戳元数据来实现。

实现思路：

1. 存储时间戳： 在将对话片段存储到 VectorStore 时，为其附加一个 timestamp 元数据。
2. 检索时衰减： 在检索阶段，可以：
   • 过滤： 直接过滤掉超过某个时间阈值的记忆片段。
   • 评分调整： 根据时间衰减函数调整检索到的记忆片段的相似度分数，使近期记忆得分更高。
3. 定期清理： 设置一个后台任务，定期扫描 VectorStore，物理删除那些超过设定“保质期”的记忆片段。

b) 基于重要性/使用频率的遗忘 (Importance/Frequency-Based Forgetting)

这种策略更复杂，它认为频繁被检索或被系统标注为“重要”的记忆，应该被保留更长时间。

实现思路：

1. 重要性评分： 每次存储记忆时，可以由LLM对记忆内容进行评分，或根据关键词密度、情感分析等方法赋予初始重要性分数。
2. 使用频率计数： 每次记忆被检索到并使用时，增加其“访问计数”或“重要性权重”。
3. 混合清理： 清理时，优先删除那些重要性评分低且长时间未被访问的记忆。

为了演示，我们将侧重于实现“基于时间衰减的遗忘”和“定期清理”机制。

5.3 代码实践：时间衰减遗忘与记忆管理器

我们将创建一个 ForgettingVectorStoreRetrieverMemory 类，它继承自 VectorStoreRetrieverMemory，并增强其保存和加载逻辑，以支持时间戳和过滤。同时，我们还将实现一个 MemoryManager 类，用于定期清理过时记忆。

# 5.3.1 定义一个增强的VectorStoreRetrieverMemory，支持时间戳
class ForgettingVectorStoreRetrieverMemory(VectorStoreRetrieverMemory):
    forget_after_seconds: Optional[int] = None # 记忆保留时长（秒）
    retriever_k: int = 3 # 检索的记忆片段数量

    def __init__(self, **kwargs: Any):
        super().__init__(**kwargs)
        # 确保retriever_kwargs中的k值与retriever_k一致
        self.retriever = self.vectorstore.as_retriever(search_kwargs={"k": self.retriever_k})

    def _get_current_timestamp(self) -> float:
        """获取当前UTC时间戳"""
        return datetime.utcnow().timestamp()

    def _get_relevant_documents(self, input: str) -> List[Any]:
        """
        重写_get_relevant_documents方法，在检索时考虑时间衰减。
        这里我们直接过滤掉过期的记忆。
        更高级的策略可以是对过期记忆进行分数衰减而非直接过滤。
        """
        # 调用父类的检索方法获取初步相关文档
        all_relevant_docs = self.retriever.get_relevant_documents(input)

        if self.forget_after_seconds is None:
            return all_relevant_docs

        # 计算过期时间戳
        expiration_timestamp = self._get_current_timestamp() - self.forget_after_seconds

        # 过滤掉过期的文档
        filtered_docs = []
        for doc in all_relevant_docs:
            if doc.metadata and "timestamp" in doc.metadata:
                if doc.metadata["timestamp"] >= expiration_timestamp:
                    filtered_docs.append(doc)
            else:
                # 如果没有时间戳，默认保留（或者根据业务需求决定是否过滤）
                filtered_docs.append(doc)

        # 如果过滤后文档不足k个，可以考虑放宽条件或发出警告
        # 目前就直接返回过滤后的结果
        return filtered_docs

    def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
        """
        重写save_context方法，在保存记忆时添加时间戳元数据。
        """
        # 构建要存储的文本和元数据
        text = self.get_buffer_string(inputs, outputs)
        metadata = {"timestamp": self._get_current_timestamp()}

        # 将文本和元数据添加到VectorStore
        self.vectorstore.add_texts([text], metadatas=[metadata])

# 5.3.2 实现一个记忆管理器，用于定期清理过时记忆
class MemoryManager:
    def __init__(self, vectorstore: FAISS, forget_after_seconds: int):
        self.vectorstore = vectorstore
        self.forget_after_seconds = forget_after_seconds
        self.last_cleanup_time = datetime.utcnow()

    def _get_current_timestamp(self) -> float:
        """获取当前UTC时间戳"""
        return datetime.utcnow().timestamp()

    def clean_old_memories(self, force: bool = False):
        """
        清理VectorStore中所有过期的记忆。
        这是一个物理删除操作。
        """
        if not force and (datetime.utcnow() - self.last_cleanup_time).total_seconds() < 3600:
            # 默认每小时清理一次，除非强制清理
            return

        print("n[MemoryManager]: 正在执行记忆清理任务...")
        expiration_timestamp = self._get_current_timestamp() - self.forget_after_seconds

        # FAISS没有直接的按元数据过滤删除接口，需要遍历所有文档并手动删除
        # 注意：对于大型VectorStore，这可能效率低下。
        # 生产环境中，推荐使用支持元数据过滤和批量删除的云端VectorStore。

        ids_to_delete = []
        # 遍历FAISS的docstore来检查每个文档的元数据
        # docstore._dict 存储了 id -> Document 对象
        for doc_id, doc in self.vectorstore.docstore._dict.items():
            if doc.metadata and "timestamp" in doc.metadata:
                if doc.metadata["timestamp"] < expiration_timestamp:
                    ids_to_delete.append(doc_id)

        if ids_to_delete:
            self.vectorstore.delete(ids_to_delete)
            print(f"[MemoryManager]: 清理了 {len(ids_to_delete)} 个过期记忆片段。")
        else:
            print("[MemoryManager]: 没有发现过期记忆片段。")

        self.last_cleanup_time = datetime.utcnow()

# 重新初始化VectorStore，并使用新的ForgettingVectorStoreRetrieverMemory
print("n--- 重新初始化并测试带遗忘机制的记忆系统 ---")

# 重新初始化VectorStore
vectorstore_forget = FAISS.from_texts(["初始化记忆，等待对话。"], embeddings_model)
vectorstore_forget.delete(vectorstore_forget.index_to_docstore_id.keys())

# 设置记忆在10分钟（600秒）后过期
FORGET_SECONDS = 600 
memory_forget = ForgettingVectorStoreRetrieverMemory(
    retriever=vectorstore_forget.as_retriever(search_kwargs={"k": 3}),
    memory_key="chat_history",
    return_messages=True,
    forget_after_seconds=FORGET_SECONDS,
    retriever_k=3
)

# 重新构建对话链
conversation_chain_forget = (
    RunnablePassthrough.assign(
        chat_history=lambda x: memory_forget.load_memory_variables(x)["chat_history"]
    )
    | prompt
    | llm
    | StrOutputParser()
)

# 封装函数来运行对话并保存记忆
def chat_with_forgetting_memory(user_input: str, current_memory: ForgettingVectorStoreRetrieverMemory):
    print(f"n用户: {user_input}")
    ai_response = conversation_chain_forget.invoke({"input": user_input})
    print(f"AI: {ai_response}")
    current_memory.save_context({"input": user_input}, {"output": ai_response})
    return ai_response

# 初始化记忆管理器
memory_manager = MemoryManager(vectorstore_forget, FORGET_SECONDS)

# 进行几轮对话
print("n进行第一组对话...")
chat_with_forgetting_memory("你好，我叫王五。", memory_forget)
chat_with_forgetting_memory("我喜欢园艺，尤其是种月季花。", memory_forget)
chat_with_forgetting_memory("你觉得月季花有哪些常见的病虫害？", memory_forget)

# 模拟时间流逝（例如等待10秒，假设FORGET_SECONDS设置得很小，方便演示）
# 在实际应用中，FORGET_SECONDS会是小时、天或更长
# 这里为了快速测试“遗忘”，我们暂时将FORGET_SECONDS设置小一些，例如10秒
# 为了演示效果，我们暂时不等待，而是先添加一些“旧”记忆
# 假设 FORGET_SECONDS 足够长，这里添加的记忆不会立即过期
# 但为了演示清理机制，我们会手动模拟时间
print(f"n当前VectorStore中记忆片段数量: {len(vectorstore_forget.docstore._dict)}")

# 添加一些将被标记为“旧”的记忆
print("n添加一些旧记忆，模拟之前的对话...")
# 这里我们直接向vectorstore添加，并手动设置一个较早的时间戳
old_timestamp = (datetime.utcnow() - timedelta(seconds=FORGET_SECONDS + 100)).timestamp() # 100秒前就过期了
vectorstore_forget.add_texts(
    ["用户: 我有一个关于编程的问题。 AI: 请问是什么问题呢？"], 
    metadatas=[{"timestamp": old_timestamp, "type": "old_memory"}]
)
vectorstore_forget.add_texts(
    ["用户: 我想学习新的编程语言。 AI: 推荐你试试Rust。"], 
    metadatas=[{"timestamp": old_timestamp + 10, "type": "old_memory"}]
)
print(f"添加旧记忆后，VectorStore中记忆片段数量: {len(vectorstore_forget.docstore._dict)}")

# 尝试询问旧记忆相关的问题，看是否能被检索到 (此时应该能，因为还没清理，但检索器会过滤)
print("n尝试询问旧记忆相关的问题（在清理前）...")
chat_with_forgetting_memory("你还记得我问过你关于编程的问题吗？", memory_forget) # 检索器会过滤掉过期的，所以即使没清理，也检索不到

# 强制执行记忆清理
memory_manager.clean_old_memories(force=True)
print(f"清理后，VectorStore中记忆片段数量: {len(vectorstore_forget.docstore._dict)}")

# 再次尝试询问旧记忆相关的问题 (此时应该彻底找不到了)
print("n再次尝试询问旧记忆相关的问题（在清理后）...")
chat_with_forgetting_memory("你还记得我问过你关于编程的问题吗？", memory_forget)

# 询问当前对话相关的问题，看新记忆是否还在
print("n询问当前对话相关的问题...")
chat_with_forgetting_memory("我最喜欢什么花？", memory_forget)

print("n带遗忘机制的对话测试结束。")

代码解释与遗忘曲线实现：

1. ForgettingVectorStoreRetrieverMemory 类：
   • 继承自 VectorStoreRetrieverMemory，重写了 _get_relevant_documents 和 save_context 方法。
   • save_context：在每次保存记忆时，向其 metadata 中添加一个 timestamp 字段，记录记忆创建的UTC时间戳。
   • _get_relevant_documents：在检索相关文档时，首先获取所有相关文档。然后，它会检查每个文档的 timestamp 元数据，如果该时间戳早于 当前时间 - forget_after_seconds，则该文档被认为是过期的，不会被返回给LLM。这实现了“检索时衰减”的效果。
2. MemoryManager 类：
   • 这个类负责物理删除 VectorStore 中真正过期的记忆。
   • clean_old_memories 方法会遍历 VectorStore 中的所有文档，找出 timestamp 早于设定过期时间的文档ID，并调用 vectorstore.delete() 方法将其从数据库中删除。
   • 在实际生产环境中，MemoryManager 应该作为一个独立的后台服务或定时任务运行，定期执行清理，而不是在每次对话后同步执行。

通过这种方式，我们不仅在逻辑上实现了记忆的“遗忘”（即不再被检索），还在物理上实现了记忆的“删除”，从而确保了记忆系统的容量不会无限增长，并保持了相关性。

六、 构建无限容量系统

虽然我们讨论了“无限容量”，但实际上这指的是理论上无上限，而非不计成本。真正的无限容量系统还需要考虑持久化、增量更新和多租户等问题。

6.1 持久化 VectorStore

FAISS 是内存驻留的，这意味着程序结束后数据会丢失。为了实现长久记忆，我们需要将 VectorStore 持久化到磁盘，或者使用云端 VectorStore。

• FAISS 的持久化：

  # 保存VectorStore到本地文件
  vectorstore_forget.save_local("my_conversation_memory_faiss_index")
  
  # 从本地文件加载VectorStore
  # 重新加载时需要提供Embedding模型
  loaded_vectorstore = FAISS.load_local("my_conversation_memory_faiss_index", embeddings_model, allow_dangerous_deserialization=True)
  
  # 再次创建memory实例，使用加载的vectorstore
  loaded_memory = ForgettingVectorStoreRetrieverMemory(
      retriever=loaded_vectorstore.as_retriever(search_kwargs={"k": 3}),
      memory_key="chat_history",
      return_messages=True,
      forget_after_seconds=FORGET_SECONDS,
      retriever_k=3
  )
  print("nVectorStore已从本地文件加载并重新初始化记忆。")

• 云端 VectorStore： Chroma、Pinecone、Weaviate 等云服务本身就提供持久化存储，无需额外操作。它们通常有各自的初始化和连接方式。

6.2 增量更新与查询

VectorStoreRetrieverMemory 天然支持增量更新。每次调用 save_context，新的记忆片段就会被转换为Embedding并添加到 VectorStore 中，而不会影响已有的数据。查询时，检索器会在整个 VectorStore 中进行搜索。

对于非常大的 VectorStore，增量添加和查询的性能会是一个挑战，这通常需要 VectorStore 自身的索引优化和分布式架构来解决。

6.3 多租户与隔离

在多用户场景下，每个用户的对话记忆必须是独立的。这可以通过 VectorStore 的元数据过滤功能来实现。

实现思路：

1. 存储用户ID： 在存储记忆时，除了时间戳，还为每个记忆片段添加一个 user_id 元数据。
2. 检索时过滤： 在检索时，告诉 Retriever 只搜索 user_id 匹配当前用户的记忆片段。

# 修改 ForgettingVectorStoreRetrieverMemory 以支持多租户
class MultiTenantForgettingVectorStoreRetrieverMemory(ForgettingVectorStoreRetrieverMemory):
    user_id: str # 当前用户的ID

    def __init__(self, **kwargs: Any):
        super().__init__(**kwargs)
        # 重新配置检索器，使其在查询时能根据user_id进行过滤
        # 注意：FAISS的as_retriever默认不支持直接的元数据过滤，
        # 需要自定义Wrapper或使用支持元数据过滤的VectorStore
        # 这里为了演示，我们将在_get_relevant_documents中手动过滤
        self.retriever = self.vectorstore.as_retriever(search_kwargs={"k": self.retriever_k})

    def _get_relevant_documents(self, input: str) -> List[Any]:
        # 调用父类（ForgettingVectorStoreRetrieverMemory）的检索方法，已处理时间衰减
        relevant_docs_from_parent = super()._get_relevant_documents(input)

        # 进一步根据user_id过滤
        filtered_by_user_docs = []
        for doc in relevant_docs_from_parent:
            if doc.metadata and "user_id" in doc.metadata and doc.metadata["user_id"] == self.user_id:
                filtered_by_user_docs.append(doc)
        return filtered_by_user_docs

    def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
        text = self.get_buffer_string(inputs, outputs)
        metadata = {"timestamp": self._get_current_timestamp(), "user_id": self.user_id}
        self.vectorstore.add_texts([text], metadatas=[metadata])

# 示例：为不同用户创建记忆实例
print("n--- 测试多租户记忆系统 ---")
user1_id = "user_001"
user2_id = "user_002"

# 共享同一个VectorStore
shared_vectorstore = FAISS.from_texts(["共享初始化。"], embeddings_model)
shared_vectorstore.delete(shared_vectorstore.index_to_docstore_id.keys())

memory_user1 = MultiTenantForgettingVectorStoreRetrieverMemory(
    vectorstore=shared_vectorstore,
    user_id=user1_id,
    memory_key="chat_history",
    return_messages=True,
    forget_after_seconds=FORGET_SECONDS,
    retriever_k=3
)

memory_user2 = MultiTenantForgettingVectorStoreRetrieverMemory(
    vectorstore=shared_vectorstore,
    user_id=user2_id,
    memory_key="chat_history",
    return_messages=True,
    forget_after_seconds=FORGET_SECONDS,
    retriever_k=3
)

def chat_with_multi_tenant_memory(user_input: str, current_memory: MultiTenantForgettingVectorStoreRetrieverMemory):
    print(f"n用户 {current_memory.user_id}: {user_input}")

    conversation_chain = (
        RunnablePassthrough.assign(
            chat_history=lambda x: current_memory.load_memory_variables(x)["chat_history"]
        )
        | prompt
        | llm
        | StrOutputParser()
    )

    ai_response = conversation_chain.invoke({"input": user_input})
    print(f"AI ({current_memory.user_id}): {ai_response}")
    current_memory.save_context({"input": user_input}, {"output": ai_response})
    return ai_response

# 用户1对话
chat_with_multi_tenant_memory("你好，我是用户1，我喜欢蓝色。", memory_user1)
chat_with_multi_tenant_memory("你觉得蓝色代表什么？", memory_user1)

# 用户2对话
chat_with_multi_tenant_memory("你好，我是用户2，我喜欢绿色。", memory_user2)
chat_with_multi_tenant_memory("你觉得绿色代表什么？", memory_user2)

# 用户1再次提问，看是否能记住自己的偏好
chat_with_multi_tenant_memory("我之前跟你说过我喜欢什么颜色吗？", memory_user1)

# 用户2再次提问，看是否能记住自己的偏好
chat_with_multi_tenant_memory("我最喜欢的颜色是什么？", memory_user2)

print(f"n共享VectorStore中所有记忆片段数量: {len(shared_vectorstore.docstore._dict)}")
print("n多租户记忆测试结束。")

注意： 对于FAISS这种本地VectorStore，其 as_retriever() 方法默认不支持基于元数据的过滤。在 MultiTenantForgettingVectorStoreRetrieverMemory 中，我们通过在 _get_relevant_documents 方法中手动遍历和过滤，实现了多租户逻辑。但在生产环境中，强烈建议使用原生支持元数据过滤的云端VectorStore（如Pinecone、Weaviate、Chroma的服务器模式），这样可以在VectorStore层面高效执行过滤，避免在Python代码中遍历大量文档。

七、 高级主题与优化

7.1 Retriever 优化

• MMR (Maximal Marginal Relevance) 检索： 传统的相似度搜索可能返回大量语义相似但内容重复的文档。MMR 旨在返回与查询相关且彼此之间差异最大的文档，从而提供更丰富和多样化的上下文。LangChain的 VectorStore.as_retriever() 通常支持 search_type="mmr"。
• 混合检索 (Hybrid Search)： 结合关键词搜索（如BM25）和语义搜索。关键词搜索擅长匹配精确术语，语义搜索擅长理解意图。两者的结合能弥补各自的不足。
• Re-ranking (重排序)： 在检索到一批初步相关的文档后，可以使用一个更小的、更强大的模型（或LLM本身）对这些文档进行二次排序，以提升最终送给LLM的上下文质量。

7.2 分块策略 (Chunking Strategies)

将对话历史分割成多大的“记忆单元”会直接影响检索效果。

• 过小的块： 可能丢失上下文，导致检索到的信息碎片化。
• 过大的块： 可能引入不相关的信息，增加LLM处理的噪音，且更容易超出Embedding模型的输入限制。

常用的策略包括：

• CharacterTextSplitter： 按字符数分割。
• RecursiveCharacterTextSplitter： 递归地尝试使用不同的分隔符（如nn, n, ）分割，直到块大小满足要求。这是目前最推荐的文本分割器之一。
• 基于对话轮次的分割： 将每一轮用户输入和AI响应作为一个块。
• 基于主题的分割： 使用LLM或聚类算法将相关对话分组为更大的记忆单元。

在我们的示例中，VectorStoreRetrieverMemory 默认是将 input 和 output 拼接后作为一个文档进行存储。这是一种简单的对话轮次分割。

7.3 元数据在记忆中的作用

除了 timestamp 和 user_id，元数据还可以存储更多有用的信息，以帮助检索和记忆管理：

• topic / tags： 对话的主题标签，可以用于在检索时过滤特定主题的记忆。
• sentiment： 对话的情感倾向，用于个性化响应或识别用户情绪变化。
• summary： 对长对话片段的摘要，检索时可以优先查看摘要，节省LLM的token。
• importance_score： 结合之前提到的基于重要性的遗忘策略。

这些元数据可以被 Retriever 用作过滤条件（例如 vectorstore.as_retriever(search_kwargs={"filter": {"topic": "gardening"}})），从而实现更精准的记忆检索。

7.4 记忆压缩与摘要

对于那些很久以前发生，但仍具有一定价值的记忆，我们不希望完全删除，但也不希望它们占用大量存储或每次都以原始形式检索。

策略：

1. 定期摘要： 使用LLM定期对一组相关且较旧的对话片段进行摘要，将其压缩成一个更短、信息密度更高的记忆单元，然后替换掉原始片段。
2. 分层记忆： 建立多层记忆结构。例如，短期记忆（ConversationBufferWindowMemory）存储最新几轮对话，中期记忆（VectorStoreRetrieverMemory）存储数小时或数天内的详细对话，长期记忆（VectorStoreRetrieverMemory 存储摘要）存储更久远的、高度压缩的总结。

7.5 Prompt Engineering 与记忆交互

即使我们提供了最相关的记忆片段，LLM是否能有效利用它们，很大程度上取决于Prompt的设计。

• 明确指令： 在Prompt中明确告诉LLM，它拥有历史对话，并鼓励它在回答中引用这些历史。

  SystemMessage("你是一个友好的AI助手，请参考提供的历史对话，确保回答连贯且个性化。")

• 格式化历史： 确保传递给LLM的历史对话格式清晰易读，例如使用“用户：”和“AI：”前缀。VectorStoreRetrieverMemory 的 return_messages=True 选项可以返回消息对象列表，这对于LLM来说是更友好的格式。
• 引导性问题： 在需要时，可以通过Prompt中的问题来引导LLM从记忆中提取特定信息。

八、 实际应用场景与挑战

8.1 实际应用场景

• 智能客服机器人： 记住用户的历史咨询、购买记录、偏好，提供个性化和连贯的客户支持。
• 个性化助手： 记住用户的日程、兴趣、习惯，提供更贴心的服务。
• 教育与辅导： 跟踪学生的学习进度、掌握的知识点、遇到的问题，提供定制化的学习路径和辅导。
• 创意写作助手： 记住用户设定的故事背景、人物关系、情节发展，帮助用户保持创作的一致性。

8.2 面临的挑战

• 计算资源： Embedding模型的推理成本、VectorStore 的存储和检索成本，尤其是在大规模用户和海量数据下，会显著增加。
• 实时性要求： 对于高并发、低延迟的实时对话系统，Embedding生成和VectorStore检索的速度至关重要。
• 记忆的准确性与幻觉： 即使检索到的信息是准确的，LLM在整合这些信息时仍可能产生“幻觉”或误解，导致不准确的回答。
• 隐私与安全： 存储用户敏感的对话历史需要严格遵守数据隐私法规（如GDPR、CCPA），并确保数据安全。多租户隔离是基本要求。
• 遗忘策略的平衡： “遗忘曲线”的参数（如 forget_after_seconds）需要根据具体应用场景仔细调优，平衡记忆的持久性和系统的效率。

九、 展望未来

我们今天探讨的 VectorStoreRetrieverMemory 结合“长久遗忘曲线”机制，为构建智能、高效的对话记忆系统奠定了坚实基础。然而，这仅仅是开始。

未来，我们可以期待更智能、更动态的遗忘机制，例如结合强化学习，让系统根据对话效果自动调整遗忘策略。多模态记忆将允许系统记住图片、视频、音频等非文本信息。与知识图谱的结合，则能让记忆系统不仅仅是存储事实，更能理解事实之间的关系，从而实现更深层次的推理和回答。

最终，我们追求的，是一个能够像人类一样，拥有学习、记忆、遗忘并能深刻理解和响应用户的AI系统。这不仅是技术的进步，更是人机交互体验的革新。

感谢各位的聆听！希望今天的讲座能为大家在LLM记忆系统构建的道路上，带来一些启发和帮助。