重排序技术：从粗排到精排的检索质量飞跃

引言

在 RAG 系统中，向量检索（Bi-encoder）速度快但精度有限。重排序（Reranking） 作为第二阶段的精排，对初选结果进行更精细的语义匹配，是提升检索质量最有效的手段之一。

第一阶段：初排（Bi-encoder）        第二阶段：精排（Cross-encoder）
    检索全部文档(10万+)              对 TOP-K 结果重排序
           │                                │
           ▼                                ▼
    ┌──────────────┐              ┌──────────────────┐
    │  向量检索     │              │  Cross-encoder   │
    │  快: 10ms    │              │  慢: 50-200ms    │
    │  粗: Recall  │   TOP-100    │  精: Precision   │
    │  @100: 95%   │ ──────────→  │  @10: 精准排序   │
    └──────────────┘              └──────────────────┘
           │                                │
           ▼                                ▼
       10万→100个                    100个→10个最佳结果

一、为什么需要重排序？

1.1 Bi-encoder 的局限

Bi-encoder（双编码器）将查询和文档分别编码为向量，使用余弦距离计算相似度。这种方式虽然快，但：

1.2 Cross-encoder 的优势

Cross-encoder（交叉编码器）将查询和文档拼接后一起输入模型，让它们可以 互相注意（cross-attention）：

Bi-encoder:                   Cross-encoder:
"RAG是什么" → [0.1, 0.2...]   "[CLS] RAG是什么 [SEP] RAG是一种...
"文档1"    → [0.3, 0.4...]    架构 [SEP]" → 相似度: 0.92
    ↓ 余弦相似度: 0.85        ↑
    
无交互                     深度交互
精度损失                   精度提升 10-20%

二、主流重排序模型对比

三、实践：使用 Cross-encoder 重排序

3.1 基础重排序实现

from sentence_transformers import CrossEncoder

# 加载 Cross-encoder 模型
model = CrossEncoder(
    'BAAI/bge-reranker-v2-m3',
    max_length=512,
    device='cuda'  # GPU 加速
)

def rerank(
    query: str,
    documents: list[dict],
    top_k: int = 5
) -> list[dict]:
    """
    对检索结果进行重排序
    documents: [{"id": "1", "text": "..."}, ...]
    """
    pairs = [
        [query, doc['text']]
        for doc in documents
    ]

    # 计算相关性分数
    scores = model.predict(pairs)

    # 按分数排序
    scored_docs = [
        {**doc, 'rerank_score': float(score)}
        for doc, score in zip(documents, scores)
    ]
    scored_docs.sort(key=lambda x: x['rerank_score'], reverse=True)

    return scored_docs[:top_k]

3.2 完整 RAG 管线中的重排序

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import (
    CrossEncoderReranker
)
from langchain_community.cross_encoders import (
    HuggingFaceCrossEncoder
)

# 配置重排序
reranker = CrossEncoderReranker(
    model=HuggingFaceCrossEncoder(
        model_name="BAAI/bge-reranker-v2-m3"
    ),
    top_n=5
)

# 包装检索器
retriever = VectorStoreRetriever(
    vectorstore=vector_store,
    search_kwargs={"k": 20}  # 先取 20 个
)

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

# 使用时自动执行：检索 20 条 → 重排序 → 返回 TOP 5
results = compression_retriever.invoke(
    "RAG系统中如何优化检索质量？"
)

四、实践：LlamaIndex 中的重排序

from llama_index.core import VectorStoreIndex
from llama_index.core.postprocessor import (
    SentenceTransformerRerank
)
from llama_index.core.query_engine import RetrieverQueryEngine

# 创建索引
index = VectorStoreIndex.from_documents(documents)

# 配置重排序
rerank_postprocessor = SentenceTransformerRerank(
    model="BAAI/bge-reranker-v2-m3",
    top_k=5,              # 保留前 5 个结果
    keep_retrieval_score=True  # 保留原始检索分数
)

# 构建查询引擎
query_engine = RetrieverQueryEngine.from_args(
    retriever=index.as_retriever(similarity_top_k=20),
    node_postprocessors=[rerank_postprocessor],
    response_mode="compact"
)

# 查询时会自动：检索 20 → 重排序 20→5 → 生成回答
response = query_engine.query(
    "Milvus和Pinecone在架构设计上有什么不同？"
)

五、高级策略

5.1 分层重排序

当文档数量很大时，可以用多层重排序策略平衡速度和精度：

class TwoStageReranker:
    """两阶段重排序"""
    def __init__(self):
        # 第一阶段：轻量模型（快速筛选）
        self.fast_model = CrossEncoder(
            'cross-encoder/ms-marco-MiniLM-L-6-v2'
        )
        # 第二阶段：重量模型（精确排序）
        self.precise_model = CrossEncoder(
            'BAAI/bge-reranker-v2-m3'
        )

    def rerank(self, query: str, documents: list, top_k: int = 5):
        # 阶段1：快速筛选 TOP-50 → TOP-20
        pairs = [[query, d['text']] for d in documents]
        fast_scores = self.fast_model.predict(pairs)
        top_20_indices = fast_scores.argsort()[-20:][::-1]

        # 阶段2：精确排序 TOP-20 → TOP-5
        top_20 = [documents[i] for i in top_20_indices]
        precise_pairs = [[query, d['text']] for d in top_20]
        precise_scores = self.precise_model.predict(precise_pairs)

        # 返回最终结果
        results = sorted(
            zip(top_20, precise_scores),
            key=lambda x: x[1],
            reverse=True
        )
        return [doc for doc, _ in results[:top_k]]

5.2 查询感知重排序

class QueryAwareReranker:
    """根据查询类型选择重排序策略"""
    def __init__(self):
        self.reranker = CrossEncoder('BAAI/bge-reranker-v2-m3')

    def is_precision_query(self, query: str) -> bool:
        """判断是否需要高精度重排序"""
        precision_indicators = [
            r'\d+\.\d+\.\d+',  # 版本号
            r'[A-Z]{2,}-\d+',  # 编号
            r'第.+章',          # 章节
        ]
        return any(
            re.search(pattern, query)
            for pattern in precision_indicators
        )

    def rerank(self, query, documents, top_k=5):
        if self.is_precision_query(query):
            # 精确查询：更宽的候选池
            return self._rerank(query, documents, top_k,
                              candidate_pool=50)
        else:
            # 语义查询：标准重排序
            return self._rerank(query, documents, top_k,
                              candidate_pool=20)

六、性能考量

6.1 速度与精度权衡

6.2 优化建议

# 1. 并行批处理
from concurrent.futures import ThreadPoolExecutor

def batch_rerank_parallel(query, docs_batches, model, top_k):
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = [
            executor.submit(rerank, query, batch, top_k // 4)
            for batch in docs_batches
        ]
        results = []
        for f in futures:
            results.extend(f.result())
        return results[:top_k]

# 2. 结果缓存
from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_rerank(query: str, doc_ids: tuple) -> list:
    """对有缓存的查询跳过重排序"""
    return rerank(query, get_docs_by_ids(doc_ids))

七、总结

相关资源：

• BGE Reranker
• Cohere Rerank
• Cross-encoder 论文