如何使用时间加权向量存储检索器
它们的评分算法是:
semantic_similarity + (1.0 - decay_rate) ^ hours_passed
特别需要注意的是,hours_passed
指的是检索器中的对象上次被访问以来经过的小时数,而不是它被创建以来经过的小时数。这意味着经常被访问的对象会保持“新鲜”。
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
低衰减率
较低的 decay rate
(在此示例中,为了极端,我们将设置它接近 0)意味着记忆将被“记住”更长时间。 decay rate
为 0 意味着记忆永远不会被遗忘,这使得此检索器等效于向量查找。
# Define your embedding model
embeddings_model = OpenAIEmbeddings()
# Initialize the vectorstore as empty
embedding_size = 1536
index = faiss.IndexFlatL2(embedding_size)
vectorstore = FAISS(embeddings_model, index, InMemoryDocstore({}), {})
retriever = TimeWeightedVectorStoreRetriever(
vectorstore=vectorstore, decay_rate=0.0000000000000000000000001, k=1
)
yesterday = datetime.now() - timedelta(days=1)
retriever.add_documents(
[Document(page_content="hello world", metadata={"last_accessed_at": yesterday})]
)
retriever.add_documents([Document(page_content="hello foo")])
['73679bc9-d425-49c2-9d74-de6356c73489']
# "Hello World" is returned first because it is most salient, and the decay rate is close to 0., meaning it's still recent enough
retriever.invoke("hello world")
[Document(metadata={'last_accessed_at': datetime.datetime(2024, 10, 22, 16, 37, 40, 818583), 'created_at': datetime.datetime(2024, 10, 22, 16, 37, 37, 975074), 'buffer_idx': 0}, page_content='hello world')]
高衰减率
使用较高的 decay rate
(例如,多个 9),recency score
会很快变为 0!如果您将其设置为 1,则所有对象的 recency
都为 0,再次使其等效于向量查找。
# Define your embedding model
embeddings_model = OpenAIEmbeddings()
# Initialize the vectorstore as empty
embedding_size = 1536
index = faiss.IndexFlatL2(embedding_size)
vectorstore = FAISS(embeddings_model, index, InMemoryDocstore({}), {})
retriever = TimeWeightedVectorStoreRetriever(
vectorstore=vectorstore, decay_rate=0.999, k=1
)
yesterday = datetime.now() - timedelta(days=1)
retriever.add_documents(
[Document(page_content="hello world", metadata={"last_accessed_at": yesterday})]
)
retriever.add_documents([Document(page_content="hello foo")])
['379631f0-42c2-4773-8cc2-d36201e1e610']
# "Hello Foo" is returned first because "hello world" is mostly forgotten
retriever.invoke("hello world")
[Document(metadata={'last_accessed_at': datetime.datetime(2024, 10, 22, 16, 37, 46, 553633), 'created_at': datetime.datetime(2024, 10, 22, 16, 37, 43, 927429), 'buffer_idx': 1}, page_content='hello foo')]
虚拟时间
使用 LangChain 中的一些实用工具,您可以模拟时间组件。
from langchain_core.utils import mock_now
API 参考:mock_now
# Notice the last access time is that date time
tomorrow = datetime.now() + timedelta(days=1)
with mock_now(tomorrow):
print(retriever.invoke("hello world"))
[Document(metadata={'last_accessed_at': MockDateTime(2024, 10, 23, 16, 38, 19, 66711), 'created_at': datetime.datetime(2024, 10, 22, 16, 37, 43, 599877), 'buffer_idx': 0}, page_content='hello world')]