跨编码器直接对每一对 (query, document) 进行评分，而不是比较独立嵌入，这虽然每个文档需要额外一次推理，但能产生更准确的排序。在向量搜索之上应用重排序器（通过嵌入检索前20条，再重排序至前5条）是 RAG 流水线中影响最大的质量改进之一，而且使用 Hugging Face 的小型跨编码器时可以在本地 CPU 上免费运行。

本指南展示如何将 HuggingFaceCrossEncoder 与 LangChain 的 CrossEncoderReranker 和 ContextualCompressionRetriever结合使用。该模式适用于任何 Hugging Face 上的跨编码器模型，包括 BAAI/bge-reranker-*, mixedbread-ai/mxbai-rerank-*, Alibaba-NLP/gte-multilingual-reranker-*, Qwen/Qwen3-Reranker-*，以及经典的 cross-encoder/ms-marco-* family.

设置

pip install -qU langchain-huggingface langchain-community langchain-classic faiss-cpu

构建基础检索器

从一个标准的向量存储检索器开始。检索相对较大的 k；重排序器会将其缩小。

from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

documents = TextLoader("../../how_to/state_of_the_union.txt").load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=100)
texts = text_splitter.split_documents(documents)

embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-m3",
    encode_kwargs={"normalize_embeddings": True},
)
retriever = FAISS.from_documents(texts, embeddings).as_retriever(
    search_kwargs={"k": 20}
)

使用跨编码器进行重排序

CrossEncoderReranker 包装任意跨编码器并插入 ContextualCompressionRetriever.

from langchain_classic.retrievers.contextual_compression import ContextualCompressionRetriever
from langchain_classic.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

cross_encoder = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-v2-m3")
reranker = CrossEncoderReranker(model=cross_encoder, top_n=3)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=retriever,
)

compressed_docs = compression_retriever.invoke("What is the plan for the economy?")
for i, doc in enumerate(compressed_docs, 1):
    print(f"Document {i}:\n{doc.page_content}\n")

选择跨编码器

HuggingFaceCrossEncoder 自动选择最佳可用设备（CUDA > MPS > CPU）。要固定到特定设备，请传递 model_kwargs={"device": "cpu"} 或类似参数。

部署到 SageMaker

您也可以将跨编码器托管在 SageMaker 端点上并使用 SagemakerEndpointCrossEncoder。以下是示例 inference.py ，可动态加载模型（无需 model.tar.gz 产物）。请参阅 本教程 获取分步指导。

from typing import List

from sagemaker_inference import encoder
from transformers import AutoModelForSequenceClassification, AutoTokenizer

PAIRS = "pairs"
SCORES = "scores"


class CrossEncoder:
    def __init__(self) -> None:
        self.device = (
            torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")
        )
        logging.info(f"Using device: {self.device}")
        model_name = "BAAI/bge-reranker-v2-m3"
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
        self.model = self.model.to(self.device)

    def __call__(self, pairs: List[List[str]]) -> List[float]:
        with torch.inference_mode():
            inputs = self.tokenizer(
                pairs,
                padding=True,
                truncation=True,
                return_tensors="pt",
                max_length=512,
            )
            inputs = inputs.to(self.device)
            scores = (
                self.model(**inputs, return_dict=True)
                .logits.view(
                    -1,
                )
                .float()
            )

        return scores.detach().cpu().tolist()


def model_fn(model_dir: str) -> CrossEncoder:
    try:
        return CrossEncoder()
    except Exception:
        logging.exception(f"Failed to load model from: {model_dir}")
        raise


def transform_fn(
    cross_encoder: CrossEncoder, input_data: bytes, content_type: str, accept: str
) -> bytes:
    payload = json.loads(input_data)
    model_output = cross_encoder(**payload)
    output = {SCORES: model_output}
    return encoder.encode(output, accept)