InMemoryVectorStore Class — langchain Architecture

Architecture documentation for the InMemoryVectorStore class in test_indexing.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  4269aa92_b131_014d_4134_d396f7de1f62["InMemoryVectorStore"]
  6c336ac6_f55c_1ad7_6db3_73dbd71fb625["VectorStore"]
  4269aa92_b131_014d_4134_d396f7de1f62 -->|extends| 6c336ac6_f55c_1ad7_6db3_73dbd71fb625
  9bc4e8b6_769a_ae11_3fc1_309cb678c248["test_indexing.py"]
  4269aa92_b131_014d_4134_d396f7de1f62 -->|defined in| 9bc4e8b6_769a_ae11_3fc1_309cb678c248
  1f4c9cb5_2db7_6261_9cc4_7389b49642b4["__init__()"]
  4269aa92_b131_014d_4134_d396f7de1f62 -->|method| 1f4c9cb5_2db7_6261_9cc4_7389b49642b4
  21806945_41ec_9f2f_618a_0bdc469d8590["delete()"]
  4269aa92_b131_014d_4134_d396f7de1f62 -->|method| 21806945_41ec_9f2f_618a_0bdc469d8590
  474b1a59_b1ab_4820_df1a_e0bae377a0a5["adelete()"]
  4269aa92_b131_014d_4134_d396f7de1f62 -->|method| 474b1a59_b1ab_4820_df1a_e0bae377a0a5
  1eba5b53_5901_6a6a_625e_2b6999ba648a["add_documents()"]
  4269aa92_b131_014d_4134_d396f7de1f62 -->|method| 1eba5b53_5901_6a6a_625e_2b6999ba648a
  8ef95604_bdb2_eec9_3235_0aaf60a866ef["aadd_documents()"]
  4269aa92_b131_014d_4134_d396f7de1f62 -->|method| 8ef95604_bdb2_eec9_3235_0aaf60a866ef
  5c0e8220_60b9_1333_496e_9ec38c04f4e7["add_texts()"]
  4269aa92_b131_014d_4134_d396f7de1f62 -->|method| 5c0e8220_60b9_1333_496e_9ec38c04f4e7
  cea39929_0e7c_461a_1a1d_0b0c7355c3d6["from_texts()"]
  4269aa92_b131_014d_4134_d396f7de1f62 -->|method| cea39929_0e7c_461a_1a1d_0b0c7355c3d6
  d01ca1cb_6ef0_b0a9_a7dc_7ee67dbf050c["similarity_search()"]
  4269aa92_b131_014d_4134_d396f7de1f62 -->|method| d01ca1cb_6ef0_b0a9_a7dc_7ee67dbf050c

Relationship Graph

Source Code

libs/langchain/tests/unit_tests/indexes/test_indexing.py lines 40–137

class InMemoryVectorStore(VectorStore):
    """In-memory implementation of VectorStore using a dictionary."""

    def __init__(self, *, permit_upserts: bool = False) -> None:
        """Vector store interface for testing things in memory."""
        self.store: dict[str, Document] = {}
        self.permit_upserts = permit_upserts

    @override
    def delete(self, ids: Sequence[str] | None = None, **kwargs: Any) -> None:
        """Delete the given documents from the store using their IDs."""
        if ids:
            for _id in ids:
                self.store.pop(_id, None)

    @override
    async def adelete(self, ids: Sequence[str] | None = None, **kwargs: Any) -> None:
        """Delete the given documents from the store using their IDs."""
        if ids:
            for _id in ids:
                self.store.pop(_id, None)

    @override
    def add_documents(
        self,
        documents: Sequence[Document],
        *,
        ids: Sequence[str] | None = None,
        **kwargs: Any,
    ) -> list[str]:
        """Add the given documents to the store (insert behavior)."""
        if ids and len(ids) != len(documents):
            msg = f"Expected {len(ids)} ids, got {len(documents)} documents."
            raise ValueError(msg)

        if not ids:
            msg = "This is not implemented yet."
            raise NotImplementedError(msg)

        for _id, document in zip(ids, documents, strict=False):
            if _id in self.store and not self.permit_upserts:
                msg = f"Document with uid {_id} already exists in the store."
                raise ValueError(msg)
            self.store[_id] = document

        return list(ids)

    @override
    async def aadd_documents(
        self,
        documents: Sequence[Document],
        *,
        ids: Sequence[str] | None = None,
        **kwargs: Any,
    ) -> list[str]:
        if ids and len(ids) != len(documents):
            msg = f"Expected {len(ids)} ids, got {len(documents)} documents."
            raise ValueError(msg)

        if not ids:
            msg = "This is not implemented yet."
            raise NotImplementedError(msg)

        for _id, document in zip(ids, documents, strict=False):
            if _id in self.store and not self.permit_upserts:
                msg = f"Document with uid {_id} already exists in the store."
                raise ValueError(msg)
            self.store[_id] = document
        return list(ids)

    def add_texts(
        self,
        texts: Iterable[str],
        metadatas: list[dict[Any, Any]] | None = None,
        **kwargs: Any,
    ) -> list[str]:
        """Add the given texts to the store (insert behavior)."""
        raise NotImplementedError

    @classmethod
    def from_texts(