MergerRetriever Class — langchain Architecture

Architecture documentation for the MergerRetriever class in merger_retriever.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  d3431bb4_b3e7_972f_7bf0_e9f01c630ed7["MergerRetriever"]
  3a20478a_3692_141f_433b_a32429b00020["BaseRetriever"]
  d3431bb4_b3e7_972f_7bf0_e9f01c630ed7 -->|extends| 3a20478a_3692_141f_433b_a32429b00020
  950f69ef_bc0c_23ff_44c8_712d43ac7834["merger_retriever.py"]
  d3431bb4_b3e7_972f_7bf0_e9f01c630ed7 -->|defined in| 950f69ef_bc0c_23ff_44c8_712d43ac7834
  b36313ed_19de_20c2_3cb5_aea6296aa51a["_get_relevant_documents()"]
  d3431bb4_b3e7_972f_7bf0_e9f01c630ed7 -->|method| b36313ed_19de_20c2_3cb5_aea6296aa51a
  5a02cb7f_900b_f2de_bb3b_10f53debbc54["_aget_relevant_documents()"]
  d3431bb4_b3e7_972f_7bf0_e9f01c630ed7 -->|method| 5a02cb7f_900b_f2de_bb3b_10f53debbc54
  98b9ac16_6db8_bf8a_3842_55f116fd8b89["merge_documents()"]
  d3431bb4_b3e7_972f_7bf0_e9f01c630ed7 -->|method| 98b9ac16_6db8_bf8a_3842_55f116fd8b89
  24216958_8912_8cb9_1a2c_d54e9c9bcaca["amerge_documents()"]
  d3431bb4_b3e7_972f_7bf0_e9f01c630ed7 -->|method| 24216958_8912_8cb9_1a2c_d54e9c9bcaca

Relationship Graph

Source Code

libs/langchain/langchain_classic/retrievers/merger_retriever.py lines 11–119

class MergerRetriever(BaseRetriever):
    """Retriever that merges the results of multiple retrievers."""

    retrievers: list[BaseRetriever]
    """A list of retrievers to merge."""

    def _get_relevant_documents(
        self,
        query: str,
        *,
        run_manager: CallbackManagerForRetrieverRun,
    ) -> list[Document]:
        """Get the relevant documents for a given query.

        Args:
            query: The query to search for.
            run_manager: The callback handler to use.

        Returns:
            A list of relevant documents.
        """
        # Merge the results of the retrievers.
        return self.merge_documents(query, run_manager)

    async def _aget_relevant_documents(
        self,
        query: str,
        *,
        run_manager: AsyncCallbackManagerForRetrieverRun,
    ) -> list[Document]:
        """Asynchronously get the relevant documents for a given query.

        Args:
            query: The query to search for.
            run_manager: The callback handler to use.

        Returns:
            A list of relevant documents.
        """
        # Merge the results of the retrievers.
        return await self.amerge_documents(query, run_manager)

    def merge_documents(
        self,
        query: str,
        run_manager: CallbackManagerForRetrieverRun,
    ) -> list[Document]:
        """Merge the results of the retrievers.

        Args:
            query: The query to search for.
            run_manager: The callback handler to use.

        Returns:
            A list of merged documents.
        """
        # Get the results of all retrievers.
        retriever_docs = [
            retriever.invoke(
                query,
                config={"callbacks": run_manager.get_child(f"retriever_{i + 1}")},
            )
            for i, retriever in enumerate(self.retrievers)
        ]

        # Merge the results of the retrievers.
        merged_documents = []
        max_docs = max(map(len, retriever_docs), default=0)
        for i in range(max_docs):
            for _retriever, doc in zip(self.retrievers, retriever_docs, strict=False):
                if i < len(doc):
                    merged_documents.append(doc[i])

        return merged_documents

    async def amerge_documents(
        self,
        query: str,
        run_manager: AsyncCallbackManagerForRetrieverRun,
    ) -> list[Document]:
        """Asynchronously merge the results of the retrievers.