acompress_documents() — langchain Function Reference

Source Code

libs/langchain/langchain_classic/retrievers/document_compressors/embeddings_filter.py lines 100–141

    async def acompress_documents(
        self,
        documents: Sequence[Document],
        query: str,
        callbacks: Callbacks | None = None,
    ) -> Sequence[Document]:
        """Filter documents based on similarity of their embeddings to the query."""
        try:
            from langchain_community.document_transformers.embeddings_redundant_filter import (  # noqa: E501
                _aget_embeddings_from_stateful_docs,
                get_stateful_documents,
            )
        except ImportError as e:
            msg = (
                "To use please install langchain-community "
                "with `pip install langchain-community`."
            )
            raise ImportError(msg) from e

        try:
            import numpy as np
        except ImportError as e:
            msg = "Could not import numpy, please install with `pip install numpy`."
            raise ImportError(msg) from e
        stateful_documents = get_stateful_documents(documents)
        embedded_documents = await _aget_embeddings_from_stateful_docs(
            self.embeddings,
            stateful_documents,
        )
        embedded_query = await self.embeddings.aembed_query(query)
        similarity = self.similarity_fn([embedded_query], embedded_documents)[0]
        included_idxs: np.ndarray = np.arange(len(embedded_documents))
        if self.k is not None:
            included_idxs = np.argsort(similarity)[::-1][: self.k]
        if self.similarity_threshold is not None:
            similar_enough = np.where(
                similarity[included_idxs] > self.similarity_threshold,
            )
            included_idxs = included_idxs[similar_enough]
        for i in included_idxs:
            stateful_documents[i].state["query_similarity_score"] = similarity[i]
        return [stateful_documents[i] for i in included_idxs]