TimeWeightedVectorStoreRetriever Class — langchain Architecture

Source Code

libs/langchain/langchain_classic/retrievers/time_weighted_retriever.py lines 21–198

class TimeWeightedVectorStoreRetriever(BaseRetriever):
    """Time Weighted Vector Store Retriever.

    Retriever that combines embedding similarity with recency in retrieving values.
    """

    vectorstore: VectorStore
    """The `VectorStore` to store documents and determine salience."""

    search_kwargs: dict = Field(default_factory=lambda: {"k": 100})
    """Keyword arguments to pass to the `VectorStore` similarity search."""

    # TODO: abstract as a queue
    memory_stream: list[Document] = Field(default_factory=list)
    """The memory_stream of documents to search through."""

    decay_rate: float = Field(default=0.01)
    """The exponential decay factor used as `(1.0-decay_rate)**(hrs_passed)`."""

    k: int = 4
    """The maximum number of documents to retrieve in a given call."""

    other_score_keys: list[str] = []
    """Other keys in the metadata to factor into the score, e.g. 'importance'."""

    default_salience: float | None = None
    """The salience to assign memories not retrieved from the vector store.

    None assigns no salience to documents not fetched from the vector store.
    """

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
    )

    def _document_get_date(self, field: str, document: Document) -> datetime.datetime:
        """Return the value of the date field of a document."""
        if field in document.metadata:
            if isinstance(document.metadata[field], float):
                return datetime.datetime.fromtimestamp(document.metadata[field])
            return document.metadata[field]
        return datetime.datetime.now()

    def _get_combined_score(
        self,
        document: Document,
        vector_relevance: float | None,
        current_time: datetime.datetime,
    ) -> float:
        """Return the combined score for a document."""
        hours_passed = _get_hours_passed(
            current_time,
            self._document_get_date("last_accessed_at", document),
        )
        score = (1.0 - self.decay_rate) ** hours_passed
        for key in self.other_score_keys:
            if key in document.metadata:
                score += document.metadata[key]
        if vector_relevance is not None:
            score += vector_relevance
        return score

    def get_salient_docs(self, query: str) -> dict[int, tuple[Document, float]]:
        """Return documents that are salient to the query."""
        docs_and_scores: list[tuple[Document, float]]
        docs_and_scores = self.vectorstore.similarity_search_with_relevance_scores(
            query,
            **self.search_kwargs,
        )
        results = {}
        for fetched_doc, relevance in docs_and_scores:
            if "buffer_idx" in fetched_doc.metadata:
                buffer_idx = fetched_doc.metadata["buffer_idx"]
                doc = self.memory_stream[buffer_idx]
                results[buffer_idx] = (doc, relevance)
        return results

    async def aget_salient_docs(self, query: str) -> dict[int, tuple[Document, float]]:
        """Return documents that are salient to the query."""
        docs_and_scores: list[tuple[Document, float]]
        docs_and_scores = (