VectorStoreRetrieverMemory Class — langchain Architecture

Architecture documentation for the VectorStoreRetrieverMemory class in vectorstore.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  ab43e995_7b7c_e56d_c8a5_135ecb6c094c["VectorStoreRetrieverMemory"]
  9071c852_df07_5016_b7fc_1000d3b44bc4["BaseMemory"]
  ab43e995_7b7c_e56d_c8a5_135ecb6c094c -->|extends| 9071c852_df07_5016_b7fc_1000d3b44bc4
  6a2e472d_918c_d305_4626_3bde9a897802["vectorstore.py"]
  ab43e995_7b7c_e56d_c8a5_135ecb6c094c -->|defined in| 6a2e472d_918c_d305_4626_3bde9a897802
  eca6cea8_cddd_6484_2ab1_4a3028e4449c["memory_variables()"]
  ab43e995_7b7c_e56d_c8a5_135ecb6c094c -->|method| eca6cea8_cddd_6484_2ab1_4a3028e4449c
  c09ac156_f316_fbd3_dc7b_13dd4a650e84["_get_prompt_input_key()"]
  ab43e995_7b7c_e56d_c8a5_135ecb6c094c -->|method| c09ac156_f316_fbd3_dc7b_13dd4a650e84
  eecb6b8e_4a7c_4ff1_156f_3eb87879245e["_documents_to_memory_variables()"]
  ab43e995_7b7c_e56d_c8a5_135ecb6c094c -->|method| eecb6b8e_4a7c_4ff1_156f_3eb87879245e
  6e7d3892_f5b4_805a_fb16_5718f3fb624e["load_memory_variables()"]
  ab43e995_7b7c_e56d_c8a5_135ecb6c094c -->|method| 6e7d3892_f5b4_805a_fb16_5718f3fb624e
  d0ddc2fb_cb5d_f9af_70c5_e9d621fd97bd["aload_memory_variables()"]
  ab43e995_7b7c_e56d_c8a5_135ecb6c094c -->|method| d0ddc2fb_cb5d_f9af_70c5_e9d621fd97bd
  c92792c2_c6e6_3e09_594b_7a9012845e1e["_form_documents()"]
  ab43e995_7b7c_e56d_c8a5_135ecb6c094c -->|method| c92792c2_c6e6_3e09_594b_7a9012845e1e
  45cededf_b43b_0819_5ab5_b642d483951b["save_context()"]
  ab43e995_7b7c_e56d_c8a5_135ecb6c094c -->|method| 45cededf_b43b_0819_5ab5_b642d483951b
  247baaa7_bc94_bf54_9208_f47b6b308156["asave_context()"]
  ab43e995_7b7c_e56d_c8a5_135ecb6c094c -->|method| 247baaa7_bc94_bf54_9208_f47b6b308156
  d6d872d7_43ab_cbef_1650_e1c0bd95ca08["clear()"]
  ab43e995_7b7c_e56d_c8a5_135ecb6c094c -->|method| d6d872d7_43ab_cbef_1650_e1c0bd95ca08
  e412e0e4_831f_0ae4_14d8_53991890b7b4["aclear()"]
  ab43e995_7b7c_e56d_c8a5_135ecb6c094c -->|method| e412e0e4_831f_0ae4_14d8_53991890b7b4

Relationship Graph

Source Code

libs/langchain/langchain_classic/memory/vectorstore.py lines 23–122

class VectorStoreRetrieverMemory(BaseMemory):
    """Vector Store Retriever Memory.

    Store the conversation history in a vector store and retrieves the relevant
    parts of past conversation based on the input.
    """

    retriever: VectorStoreRetriever = Field(exclude=True)
    """VectorStoreRetriever object to connect to."""

    memory_key: str = "history"
    """Key name to locate the memories in the result of load_memory_variables."""

    input_key: str | None = None
    """Key name to index the inputs to load_memory_variables."""

    return_docs: bool = False
    """Whether or not to return the result of querying the database directly."""

    exclude_input_keys: Sequence[str] = Field(default_factory=tuple)
    """Input keys to exclude in addition to memory key when constructing the document"""

    @property
    def memory_variables(self) -> list[str]:
        """The list of keys emitted from the load_memory_variables method."""
        return [self.memory_key]

    def _get_prompt_input_key(self, inputs: dict[str, Any]) -> str:
        """Get the input key for the prompt."""
        if self.input_key is None:
            return get_prompt_input_key(inputs, self.memory_variables)
        return self.input_key

    def _documents_to_memory_variables(
        self,
        docs: list[Document],
    ) -> dict[str, list[Document] | str]:
        result: list[Document] | str
        if not self.return_docs:
            result = "\n".join([doc.page_content for doc in docs])
        else:
            result = docs
        return {self.memory_key: result}

    def load_memory_variables(
        self,
        inputs: dict[str, Any],
    ) -> dict[str, list[Document] | str]:
        """Return history buffer."""
        input_key = self._get_prompt_input_key(inputs)
        query = inputs[input_key]
        docs = self.retriever.invoke(query)
        return self._documents_to_memory_variables(docs)

    async def aload_memory_variables(
        self,
        inputs: dict[str, Any],
    ) -> dict[str, list[Document] | str]:
        """Return history buffer."""
        input_key = self._get_prompt_input_key(inputs)
        query = inputs[input_key]
        docs = await self.retriever.ainvoke(query)
        return self._documents_to_memory_variables(docs)

    def _form_documents(
        self,
        inputs: dict[str, Any],
        outputs: dict[str, str],
    ) -> list[Document]:
        """Format context from this conversation to buffer."""
        # Each document should only include the current turn, not the chat history
        exclude = set(self.exclude_input_keys)
        exclude.add(self.memory_key)
        filtered_inputs = {k: v for k, v in inputs.items() if k not in exclude}
        texts = [
            f"{k}: {v}"
            for k, v in list(filtered_inputs.items()) + list(outputs.items())
        ]
        page_content = "\n".join(texts)
        return [Document(page_content=page_content)]