How to implement time-weighted vector store retriever with InMemoryVectorStore?

[Percival-Chan]

Hi, I am trying out the InMemoryVectorStore demo. How can I implement time-weighted feature when saving to local like the time-weighted vector store retriever from langchain-community? or does InMemoryVectorStore already considered time as vectors?

[dosubot[bot]]

The InMemoryVectorStore in LangChain does not inherently support time-based weighting or decay when retrieving vectors. However, you can implement a time-weighted feature similar to the TimeWeightedVectorStoreRetriever from langchain-community, which combines embedding similarity with recency to retrieve documents ^[1].

To achieve this with InMemoryVectorStore, you can:

1. Add Timestamps as Metadata: When adding documents to the InMemoryVectorStore, include a timestamp in the metadata. This can be done using the add_documents method, where each Document object can have a metadata attribute.

   from langchain_core.documents import Document
   
   document = Document(
       page_content="Sample content",
       metadata={"timestamp": "2023-10-01T12:00:00Z"}
   )
   
   vector_store.add_documents(documents=[document])

2. Custom Retrieval Logic: Implement a custom retrieval function that considers both the similarity score and the time decay factor. You can use the similarity_search_with_score method to get the similarity scores and then apply a decay function based on the time elapsed since the document's timestamp.

3. Time Decay Function: Use a decay function similar to the one in TimeWeightedVectorStoreRetriever, which adjusts the relevance of documents based on the time since they were last accessed or created. The formula used is semantic_similarity + (1.0 - decay_rate) ** hours_passed.

By combining these steps, you can create a custom retriever that incorporates time-based features with the InMemoryVectorStore.

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[Percival-Chan]

@dosu

Here are my tools on save and search memories, how to implement time decay on searching memories?

from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
from langchain_core.messages import get_buffer_string
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool
from langchain_core.vectorstores import InMemoryVectorStore
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode
from langchain_community.utilities.serpapi import SerpAPIWrapper

########################################################################

import uuid

def get_user_id(config: RunnableConfig) -> str:
    user_id = config["configurable"].get("user_id")
    if user_id is None:
        raise ValueError("User ID needs to be provided to save a memory.")

    return user_id

vector_store = InMemoryVectorStore(azure_embeddings)

def verify_past_vector_store(user_id):
    global vector_store
    isExist = os.path.exists(f"{user_id}.json")
    if isExist:
        vector_store = vector_store.load(f"{user_id}.json", azure_embeddings)

@tool
def save_recall_memory(memory: str, config: RunnableConfig) -> str:
    """Save memory to vectorstore for later semantic retrieval."""
    user_id = get_user_id(config)
    document = Document(
        page_content=memory, 
        id=str(uuid.uuid4()), 
        metadata={"user_id": user_id, "timestamp": str(datetime.datetime.now())}
    )
    verify_past_vector_store(user_id)
    vector_store.add_documents([document])
    return memory

@tool
def search_recall_memories(query: str, config: RunnableConfig) -> List[str]:
    """Search for relevant memories."""
    user_id = get_user_id(config)

    def _filter_function(doc: Document) -> bool:
        return doc.metadata.get("user_id") == user_id

    verify_past_vector_store(user_id)
    documents = vector_store.similarity_search(
        query, k=5, filter=_filter_function
    )
    return [document.page_content for document in documents]

[dosubot[bot]]

To implement time decay on searching memories with your current setup using InMemoryVectorStore, you can modify your search_recall_memories function to incorporate a time decay factor. Here's how you can do it:

1. Add a Time Decay Function: Calculate the time decay factor based on the time elapsed since the document's timestamp.

2. Modify the Retrieval Logic: Adjust the similarity scores with the time decay factor to prioritize more recent documents.

Here's an example of how you can implement this:

from datetime import datetime
from typing import List

def time_decay_factor(timestamp: str, decay_rate: float = 0.999) -> float:
    """Calculate the time decay factor based on the timestamp."""
    time_elapsed = datetime.now() - datetime.fromisoformat(timestamp)
    hours_passed = time_elapsed.total_seconds() / 3600
    return (1.0 - decay_rate) ** hours_passed

@tool
def search_recall_memories(query: str, config: RunnableConfig) -> List[str]:
    """Search for relevant memories with time decay."""
    user_id = get_user_id(config)

    def _filter_function(doc: Document) -> bool:
        return doc.metadata.get("user_id") == user_id

    verify_past_vector_store(user_id)
    documents_with_scores = vector_store.similarity_search_with_score(
        query, k=5, filter=_filter_function
    )

    # Apply time decay to the scores
    decayed_documents = [
        (doc, score * time_decay_factor(doc.metadata["timestamp"]))
        for doc, score in documents_with_scores
    ]

    # Sort documents by the adjusted score
    decayed_documents.sort(key=lambda x: x[1], reverse=True)

    return [document.page_content for document, _ in decayed_documents]

This code calculates a decay factor for each document based on its timestamp and adjusts the similarity score accordingly. The documents are then sorted by the adjusted score to prioritize more recent memories. Adjust the decay_rate to control how quickly the relevance of older documents decreases.

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}