Source code for langchain.retrievers.time_weighted_retriever
"""Retriever that combines embedding similarity with recency in retrieving values."""
import datetime
from copy import deepcopy
from typing import Any, Dict, List, Optional, Tuple
from pydantic import BaseModel, Field
from langchain.schema import BaseRetriever, Document
from langchain.vectorstores.base import VectorStore
def _get_hours_passed(time: datetime.datetime, ref_time: datetime.datetime) -> float:
    """Get the hours passed between two datetime objects."""
    return (time - ref_time).total_seconds() / 3600
[docs]class TimeWeightedVectorStoreRetriever(BaseRetriever, BaseModel):
    """Retriever combining embedding similarity with recency."""
    vectorstore: VectorStore
    """The vectorstore to store documents and determine salience."""
    search_kwargs: dict = Field(default_factory=lambda: dict(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: Optional[float] = None
    """The salience to assign memories not retrieved from the vector store.
    None assigns no salience to documents not fetched from the vector store.
    """
    class Config:
rtdocs_stable/api.python.langchain.com/en/stable/_modules/langchain/retrievers/time_weighted_retriever.html