InMemoryDocumentIndex Class — langchain Architecture

Architecture documentation for the InMemoryDocumentIndex class in in_memory.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  b727e55b_00f6_db29_cf86_d0e7c0c2c9c9["InMemoryDocumentIndex"]
  b6dfcb0b_d4ed_4574_14c7_fa7e485f1b07["DocumentIndex"]
  b727e55b_00f6_db29_cf86_d0e7c0c2c9c9 -->|extends| b6dfcb0b_d4ed_4574_14c7_fa7e485f1b07
  16d9b59c_7ced_1b5f_7142_710269a670f7["in_memory.py"]
  b727e55b_00f6_db29_cf86_d0e7c0c2c9c9 -->|defined in| 16d9b59c_7ced_1b5f_7142_710269a670f7
  cc7b6bcf_9d88_24e6_85ab_ca9258a66c7a["upsert()"]
  b727e55b_00f6_db29_cf86_d0e7c0c2c9c9 -->|method| cc7b6bcf_9d88_24e6_85ab_ca9258a66c7a
  922b4703_176c_265e_0cc2_6107fcf5d16c["delete()"]
  b727e55b_00f6_db29_cf86_d0e7c0c2c9c9 -->|method| 922b4703_176c_265e_0cc2_6107fcf5d16c
  63b027f9_3719_25d0_d1c0_5719a8ce5082["get()"]
  b727e55b_00f6_db29_cf86_d0e7c0c2c9c9 -->|method| 63b027f9_3719_25d0_d1c0_5719a8ce5082
  04a6353f_773a_e75a_cd9b_5936859e545c["_get_relevant_documents()"]
  b727e55b_00f6_db29_cf86_d0e7c0c2c9c9 -->|method| 04a6353f_773a_e75a_cd9b_5936859e545c

Relationship Graph

Source Code

libs/core/langchain_core/indexing/in_memory.py lines 19–104

class InMemoryDocumentIndex(DocumentIndex):
    """In memory document index.

    This is an in-memory document index that stores documents in a dictionary.

    It provides a simple search API that returns documents by the number of
    counts the given query appears in the document.
    """

    store: dict[str, Document] = Field(default_factory=dict)
    top_k: int = 4

    @override
    def upsert(self, items: Sequence[Document], /, **kwargs: Any) -> UpsertResponse:
        """Upsert documents into the index.

        Args:
            items: Sequence of documents to add to the index.
            **kwargs: Additional keyword arguments.

        Returns:
            A response object that contains the list of IDs that were
            successfully added or updated in the index and the list of IDs that
            failed to be added or updated.
        """
        ok_ids = []

        for item in items:
            if item.id is None:
                id_ = str(uuid.uuid4())
                item_ = item.model_copy()
                item_.id = id_
            else:
                item_ = item
                id_ = item.id

            self.store[id_] = item_
            ok_ids.append(cast("str", item_.id))

        return UpsertResponse(succeeded=ok_ids, failed=[])

    @override
    def delete(self, ids: list[str] | None = None, **kwargs: Any) -> DeleteResponse:
        """Delete by IDs.

        Args:
            ids: List of IDs to delete.

        Raises:
            ValueError: If IDs is None.

        Returns:
            A response object that contains the list of IDs that were successfully
            deleted and the list of IDs that failed to be deleted.
        """
        if ids is None:
            msg = "IDs must be provided for deletion"
            raise ValueError(msg)

        ok_ids = []

        for id_ in ids:
            if id_ in self.store:
                del self.store[id_]
                ok_ids.append(id_)

        return DeleteResponse(
            succeeded=ok_ids, num_deleted=len(ok_ids), num_failed=0, failed=[]
        )

    @override
    def get(self, ids: Sequence[str], /, **kwargs: Any) -> list[Document]:
        return [self.store[id_] for id_ in ids if id_ in self.store]

    @override
    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
    ) -> list[Document]:
        counts_by_doc = []

        for document in self.store.values():