LLMChainFilter Class — langchain Architecture

Source Code

libs/langchain/langchain_classic/retrievers/document_compressors/chain_filter.py lines 35–135

class LLMChainFilter(BaseDocumentCompressor):
    """Filter that drops documents that aren't relevant to the query."""

    llm_chain: Runnable
    """LLM wrapper to use for filtering documents.
    The chain prompt is expected to have a BooleanOutputParser."""

    get_input: Callable[[str, Document], dict] = default_get_input
    """Callable for constructing the chain input from the query and a Document."""

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
    )

    def compress_documents(
        self,
        documents: Sequence[Document],
        query: str,
        callbacks: Callbacks | None = None,
    ) -> Sequence[Document]:
        """Filter down documents based on their relevance to the query."""
        filtered_docs = []

        config = RunnableConfig(callbacks=callbacks)
        outputs = zip(
            self.llm_chain.batch(
                [self.get_input(query, doc) for doc in documents],
                config=config,
            ),
            documents,
            strict=False,
        )

        for output_, doc in outputs:
            include_doc = None
            if isinstance(self.llm_chain, LLMChain):
                output = output_[self.llm_chain.output_key]
                if self.llm_chain.prompt.output_parser is not None:
                    include_doc = self.llm_chain.prompt.output_parser.parse(output)
            elif isinstance(output_, bool):
                include_doc = output_
            if include_doc:
                filtered_docs.append(doc)

        return filtered_docs

    async def acompress_documents(
        self,
        documents: Sequence[Document],
        query: str,
        callbacks: Callbacks | None = None,
    ) -> Sequence[Document]:
        """Filter down documents based on their relevance to the query."""
        filtered_docs = []

        config = RunnableConfig(callbacks=callbacks)
        outputs = zip(
            await self.llm_chain.abatch(
                [self.get_input(query, doc) for doc in documents],
                config=config,
            ),
            documents,
            strict=False,
        )
        for output_, doc in outputs:
            include_doc = None
            if isinstance(self.llm_chain, LLMChain):
                output = output_[self.llm_chain.output_key]
                if self.llm_chain.prompt.output_parser is not None:
                    include_doc = self.llm_chain.prompt.output_parser.parse(output)
            elif isinstance(output_, bool):
                include_doc = output_
            if include_doc:
                filtered_docs.append(doc)

        return filtered_docs

    @classmethod
    def from_llm(
        cls,
        llm: BaseLanguageModel,