langchain Architecture Docs
Home / File/ merger_retriever.py — langchain Source File

merger_retriever.py — langchain Source File

Architecture documentation for merger_retriever.py, a python file in the langchain codebase. 4 imports, 0 dependents.

File python LangChainCore Runnables 4 imports 1 classes

Entity Profile

Dependency Diagram

graph LR
  950f69ef_bc0c_23ff_44c8_712d43ac7834["merger_retriever.py"]
  e6310202_b39e_ec9f_8b57_921c9c39c97d["asyncio"]
  950f69ef_bc0c_23ff_44c8_712d43ac7834 --> e6310202_b39e_ec9f_8b57_921c9c39c97d
  17a62cb3_fefd_6320_b757_b53bb4a1c661["langchain_core.callbacks"]
  950f69ef_bc0c_23ff_44c8_712d43ac7834 --> 17a62cb3_fefd_6320_b757_b53bb4a1c661
  6a98b0a5_5607_0043_2e22_a46a464c2d62["langchain_core.documents"]
  950f69ef_bc0c_23ff_44c8_712d43ac7834 --> 6a98b0a5_5607_0043_2e22_a46a464c2d62
  2b1aa4a8_5352_1757_010a_46ac9ef4b0b0["langchain_core.retrievers"]
  950f69ef_bc0c_23ff_44c8_712d43ac7834 --> 2b1aa4a8_5352_1757_010a_46ac9ef4b0b0
  style 950f69ef_bc0c_23ff_44c8_712d43ac7834 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

import asyncio

from langchain_core.callbacks import (
    AsyncCallbackManagerForRetrieverRun,
    CallbackManagerForRetrieverRun,
)
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever


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.

        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 = await asyncio.gather(
            *(
                retriever.ainvoke(
                    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

Domain

Subdomains

Classes

Dependencies

  • asyncio
  • langchain_core.callbacks
  • langchain_core.documents
  • langchain_core.retrievers

Source

Frequently Asked Questions

What does merger_retriever.py do?
merger_retriever.py is a source file in the langchain codebase, written in python. It belongs to the LangChainCore domain, Runnables subdomain.
What does merger_retriever.py depend on?
merger_retriever.py imports 4 module(s): asyncio, langchain_core.callbacks, langchain_core.documents, langchain_core.retrievers.
Where is merger_retriever.py in the architecture?
merger_retriever.py is located at libs/langchain/langchain_classic/retrievers/merger_retriever.py (domain: LangChainCore, subdomain: Runnables, directory: libs/langchain/langchain_classic/retrievers).

Analyze Your Own Codebase

Get architecture documentation, dependency graphs, and domain analysis for your codebase in minutes.

Try Supermodel Free