LangChain：时间加权向量存储检索器

TimeWeightedVectorStoreRetriever（时间加权向量存储检索器）是 LangChain 中的一种高级检索工具，它在语义相似性的基础上，它结合时间衰减因子来优化检索结果。其核心思想是让近期被频繁访问的文档在检索时获得更高的权重，从而保持信息的新鲜度（recency）。

这一设计理念受到了 2023 年斯坦福大学与 Google 合作的论文《Generative Agents: Interactive Simulacra of Human Behavior》的启发。论文中的生成式智能体（Generative Agents）需要模拟人类记忆，其中一项关键机制就是：近期或频繁被调用的记忆会“保持鲜活”，在后续检索中获得更高权重。

TimeWeightedVectorStoreRetriever 为 LangChain 的检索系统引入了时间维度，使其能够更智能地模拟人类记忆机制。通过调节衰减率，开发者可以在语义相关性与信息时效性之间取得平衡，适用于生成式智能体、对话记忆管理、个性化推荐等需要权衡历史信息与最新动态的场景。

核心原理

评分算法

TimeWeightedVectorStoreRetriever 的评分算法是其核心，它将传统的语义相似度得分与一个基于时间的分数相加：

# 最终得分 = 语义相似度 + (1.0 - 衰减率) ^ 经过的小时数
score = semantic_similarity + (1.0 - decay_rate) ^ hours_passed

• semantic_similarity：查询内容与文档内容的语义匹配程度。
• decay_rate：控制时间影响衰减速度的参数，取值范围通常在 0 到 1 之间。
• hours_passed：自文档最后一次被访问以来经过的小时数，而非自创建以来的时间。这意味着频繁被检索和访问的文档会被系统自动“刷新”，保持其新鲜度分数。

衰减率参数

通过调整 decay_rate 参数，可以实现不同的记忆模式，以适应多样化的应用场景。

衰减率取值	模式	效果	行为特征	适用场景
极低（趋近于 0）	长期记忆	记忆持久，近乎永久保留	时间衰减极慢，历史信息不易被遗忘。 即使文档久未访问，仍能保持较高新鲜度分数。 当 decay_rate = 0 时， (1.0 - 0) ^ hours_passed = 1， 检索器等同于纯向量查找	知识库问答、需要保留长期历史记录的对话系统。
高（如 0.999）	短期记忆	新鲜度迅速衰减	时间衰减极快，系统倾向于快速“遗忘”旧信息。 文档一旦长时间未访问，新鲜度分数会迅速趋近于 0，检索主要依赖语义相似度	实时聊天记录处理、新闻或社交媒体动态流。
1.0		新鲜度完全失效	所有文档的新鲜度分数均为 0，等同于标准向量查找

API 参考与核心参数

TimeWeightedVectorStoreRetriever 的初始化参数如下：

参数	类型	默认值	说明
vectorstore	VectorStore	必填	用于存储文档向量和进行语义检索的向量存储实例
decay_rate	float	0.01	指数衰减因子，用于新鲜度分数的衰减计算
k	int	4	单次检索返回的最大文档数量
search_kwargs	dict	None	传递给底层向量存储检索的额外参数
memory_stream	List[Document]	None	初始的记忆文档列表
default_salience	Optional[float]	None	为未从向量存储中获取的文档分配的新鲜度分数；若为 None 则不对其赋分
tags	Optional[List[str]]	None	与检索器关联的标签，会传递给每次调用
metadata	Optional[Dict[str, Any]]	None	与检索器关联的元数据，会传递给每次调用

与普通向量检索的区别

• 普通向量检索：仅基于语义相似度排序，忽略时效性。
• 时间加权检索：在语义相似度之上叠加时间衰减因子，使得近期被访问过的文档在排序中占据优势，能够模拟人类记忆的“使用频率影响记忆强度”这一特征。

使用示例（Python）

衰减率示例

低衰减率：示例中使用了极低的 decay_rate（趋近于 0），因此即使 "hello world" 的文档是昨天添加的，其新鲜度分数仍然较高，在检索 "hello world" 时优先被返回。

高衰减率：若将 decay_rate 设置为较高值（如 0.999），则新鲜度衰减速度极快，旧文档的新鲜度分数会迅速降至接近于 0。这种情况下，检索结果主要由语义相似度决定，与普通向量检索的行为接近。

以下是一个完整的 Python 示例，展示如何初始化并实际使用 TimeWeightedVectorStoreRetriever实现长期和短期记忆。

from datetime import datetime, timedelta
import faiss
from langchain.retrievers import TimeWeightedVectorStoreRetriever
from langchain_community.docstore import InMemoryDocstore
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_openai import OpenAIEmbeddings

# 1. 初始化向量存储
# 使用 OpenAI Embeddings 模型，维度为 1536
embeddings_model = OpenAIEmbeddings()
embedding_size = 1536
index = faiss.IndexFlatL2(embedding_size)
vectorstore = FAISS(embeddings_model, index, InMemoryDocstore({}), {})

# 2. 模拟一天前的时间点
yesterday = datetime.now() - timedelta(days=1)

# --- 场景一：低衰减率（长期记忆） ---
print("=== 长期记忆模式 ===")
long_term_retriever = TimeWeightedVectorStoreRetriever(
    vectorstore=vectorstore, 
    decay_rate=0.0000000000000000000000001, # 极低衰减率
    k=1
)
# 添加两个文档，其中一个是一天前的
long_term_retriever.add_documents([
    Document(page_content="Hello World", metadata={"last_accessed_at": yesterday}),
    Document(page_content="Hello Foo")
])
# 检索 "hello world"，由于衰减率极低，一天前的内容依然优先返回
result_long_term = long_term_retriever.invoke("hello world")
print(f"检索结果: {result_long_term.page_content}") 
# 预期输出: Hello World


# --- 场景二：高衰减率（短期记忆） ---
print("\n=== 短期记忆模式 ===")
short_term_retriever = TimeWeightedVectorStoreRetriever(
    vectorstore=vectorstore, 
    decay_rate=0.999, # 高衰减率
    k=1
)
# 清空并重新添加文档
short_term_retriever.vectorstore = FAISS(embeddings_model, index, InMemoryDocstore({}), {})
short_term_retriever.add_documents([
    Document(page_content="Hello World", metadata={"last_accessed_at": yesterday}),
    Document(page_content="Hello Foo")
])
# 检索 "hello world"，由于衰减率高，"Hello World" 因过时而被“遗忘”，返回更新的内容
result_short_term = short_term_retriever.invoke("hello world")
print(f"检索结果: {result_short_term.page_content}") 
# 预期输出: Hello Foo

开发调试模拟

from langchain_core.utils import mock_now

# 模拟时间前进一天
tomorrow = datetime.now() + timedelta(days=1)
with mock_now(tomorrow):
    # 在此上下文中，所有时间相关的操作都会认为当前时间是 'tomorrow'
    print("=== 虚拟时间测试 ===")
    # 此时再检索，可以观察到时间分数发生的变化
    result_mock = short_term_retriever.invoke("hello world")
    print(f"虚拟时间下的检索结果: {result_mock.page_content}")

注意事项

1. 必须通过检索器的 add_documents 方法添加文档

   由于 TimeWeightedVectorStoreRetriever 需要在每个文档的元数据中记录访问历史信息（如 last_accessed_at），所有文档必须通过检索器自身的 add_documents 方法添加，而非直接调用底层向量存储的添加方法。

   直接通过向量存储添加文档会导致元数据缺失，从而影响检索器的正常工作。

2. 新鲜度基于“访问时间”，而非“创建时间”

   算法中的 hours_passed 计算的是自文档最后一次被检索器访问以来经过的小时数。这意味着：

   • 文档被检索的次数越多、间隔越短，其新鲜度分数越高。

   • 即使文档创建时间较早，只要频繁被访问，其排名仍可能高于新加入但从未被访问的文档。

3. 衰减率的选择取决于具体应用场景

   • 需要长期保留重要记忆（如知识库问答系统）：选择较低的 decay_rate，使旧知识保持较高新鲜度。

   • 强调最新信息（如新闻推荐、实时对话上下文）：选择较高的 decay_rate，让过时信息快速降权。

   • 可通过实际场景测试不同衰减率下的检索效果，选择最优参数。

资料引用

• LangChain 官方检索器集成
• FAISS 向量存储文档
• Generative Agents 论文