Reranker 接口 (Reranker)

多后端 Reranker 系统，支持 aliyun (DashScope)、cohere、jina 三种后端。LightRAGReranker 实现了 LangChain BaseDocumentCompressor 接口，可直接用于 ContextualCompressionRetriever 管道。Reranker 为 Protocol 定义，工厂函数 create_reranker() 根据配置自动分发到对应的后端适配器。

lightrag_langchain.reranker.LightRAGReranker

Bases: BaseDocumentCompressor

将 Reranker Protocol 实例包装为 LangChain 的 BaseDocumentCompressor，用于 ContextualCompressionRetriever。

Parameters

reranker: 任何满足 Reranker Protocol 的对象。 top_n: 重排序后返回的最大文档数量（转发到底层 reranker 调用）。

Example

from lightrag_langchain.config import settings
from lightrag_langchain.reranker import create_reranker, LightRAGReranker

reranker = create_reranker(settings.reranker)
compressor = LightRAGReranker(reranker, top_n=5)

源代码位于： src/lightrag_langchain/reranker.py

class LightRAGReranker(BaseDocumentCompressor):
    """将 ``Reranker`` Protocol 实例包装为 LangChain 的
    ``BaseDocumentCompressor``，用于 ``ContextualCompressionRetriever``。

    Parameters
    ----------
    reranker:
        任何满足 ``Reranker`` Protocol 的对象。
    top_n:
        重排序后返回的最大文档数量（转发到底层 reranker 调用）。

    Example:
        ```python
        from lightrag_langchain.config import settings
        from lightrag_langchain.reranker import create_reranker, LightRAGReranker

        reranker = create_reranker(settings.reranker)
        compressor = LightRAGReranker(reranker, top_n=5)
        ```
    """

    def __init__(self, reranker: Reranker, top_n: int | None = None) -> None:
        self._reranker = reranker
        self._top_n = top_n

    def compress_documents(
        self,
        documents: list[Document] | tuple[Document, ...],
        query: str,
        **kwargs: Any,
    ) -> list[Document]:
        """同步路径——使用 ``asyncio.run`` 桥接异步 reranker。

        提取每个 Document 的 ``page_content``，调用 reranker，
        将 ``relevance_score`` 附加到 metadata，并按分数降序返回排序后的文档。
        """
        texts = [doc.page_content for doc in documents]
        scores = asyncio.run(
            self._reranker.rerank(query, texts, self._top_n)
        )
        return _sort_and_attach_scores(documents, scores)

    async def acompress_documents(
        self,
        documents: list[Document] | tuple[Document, ...],
        query: str,
        **kwargs: Any,
    ) -> list[Document]:
        """异步路径——直接 await reranker，无需 ``asyncio.run``。"""
        texts = [doc.page_content for doc in documents]
        scores = await self._reranker.rerank(query, texts, self._top_n)
        return _sort_and_attach_scores(documents, scores)

compress_documents

compress_documents(
    documents: list[Document] | tuple[Document, ...],
    query: str,
    **kwargs: Any,
) -> list[Document]

同步路径——使用 asyncio.run 桥接异步 reranker。

提取每个 Document 的 page_content，调用 reranker， 将 relevance_score 附加到 metadata，并按分数降序返回排序后的文档。

源代码位于： src/lightrag_langchain/reranker.py

def compress_documents(
    self,
    documents: list[Document] | tuple[Document, ...],
    query: str,
    **kwargs: Any,
) -> list[Document]:
    """同步路径——使用 ``asyncio.run`` 桥接异步 reranker。

    提取每个 Document 的 ``page_content``，调用 reranker，
    将 ``relevance_score`` 附加到 metadata，并按分数降序返回排序后的文档。
    """
    texts = [doc.page_content for doc in documents]
    scores = asyncio.run(
        self._reranker.rerank(query, texts, self._top_n)
    )
    return _sort_and_attach_scores(documents, scores)

acompress_documents async

acompress_documents(
    documents: list[Document] | tuple[Document, ...],
    query: str,
    **kwargs: Any,
) -> list[Document]

异步路径——直接 await reranker，无需 asyncio.run。

源代码位于： src/lightrag_langchain/reranker.py

async def acompress_documents(
    self,
    documents: list[Document] | tuple[Document, ...],
    query: str,
    **kwargs: Any,
) -> list[Document]:
    """异步路径——直接 await reranker，无需 ``asyncio.run``。"""
    texts = [doc.page_content for doc in documents]
    scores = await self._reranker.rerank(query, texts, self._top_n)
    return _sort_and_attach_scores(documents, scores)

lightrag_langchain.reranker.Reranker

Bases: Protocol

用于 reranking 后端的可调用接口。

所有适配器都实现此协议，以便工厂和 compressor 能够透明地与任何 provider 一起工作。

源代码位于： src/lightrag_langchain/reranker.py

class Reranker(Protocol):
    """用于 reranking 后端的可调用接口。

    所有适配器都实现此协议，以便工厂和 compressor 能够透明地与任何 provider
    一起工作。
    """

    async def rerank(
        self, query: str, documents: list[str], top_n: int | None = None
    ) -> list[dict[str, Any]]: ...