RAG重排序Rerank策略

引言

在RAG系统中，向量检索通常返回Top-K候选文档，但这些文档的排序精度往往不够理想——原因在于向量检索使用的是双塔模型（Bi-Encoder），查询和文档独立编码，无法捕获细粒度的交互特征。重排序（Rerank）通过使用更精细的交叉编码器（Cross-Encoder）对候选文档重新打分，显著提升排序质量。本文深入Rerank策略的原理、模型选择和工程实践。

为什么需要Rerank

双塔模型的局限

向量检索使用双塔模型：查询和文档分别编码为向量，通过点积或余弦相似度计算相关性。这种方式的优势是速度快（可以预计算文档向量），但局限在于：

缺乏交互：查询和文档在编码时没有交互，无法捕获词级别的匹配关系
语义粗粒度：向量相似度高不等于真正相关，可能存在"语义假阳性"
排序精度有限：Top-1的准确率通常只有60-70%，存在改进空间

Cross-Encoder的优势

重排序模型使用Cross-Encoder：将查询和文档拼接在一起输入模型，模型可以捕获两者的细粒度交互特征。这相当于让模型"逐字对比"查询和文档，排序精度远高于双塔模型。

两阶段检索架构

查询 → 向量检索（召回阶段）→ Top-20候选 → Rerank（精排阶段）→ Top-5最终结果

两阶段架构平衡了效率和精度：向量检索负责高效召回，Rerank负责精确排序。

Rerank模型选择

通用Rerank模型

Cohere Rerank：商业API，效果优秀，使用简单：

import cohere

co = cohere.Client('your-api-key')

def cohere_rerank(query, documents, top_n=5):
    results = co.rerank(
        model='rerank-multilingual-v3.0',
        query=query,
        documents=documents,
        top_n=top_n
    )
    return [documents[r.index] for r in results.results]

BGE-Reranker：开源模型，支持本地部署：

from FlagEmbedding import FlagReranker

reranker = FlagReranker('BAAI/bge-reranker-large', use_fp16=True)

def bge_rerank(query, documents, top_n=5):
    pairs = [[query, doc] for doc in documents]
    scores = reranker.compute_score(pairs)
    
    # 按分数排序
    ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
    return [doc for doc, _ in ranked[:top_n]]

bge-reranker-v2-m3：多语言支持，轻量高效：

from sentence_transformers import CrossEncoder

model = CrossEncoder('BAAI/bge-reranker-v2-m3', max_length=512)

def rerank(query, documents, top_n=5):
    pairs = [[query, doc] for doc in documents]
    scores = model.predict(pairs)
    ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
    return [doc for doc, _ in ranked[:top_n]]

模型选择对比

模型	类型	语言支持	延迟	效果	成本
Cohere Rerank	API	多语言	中	优秀	按量付费
bge-reranker-large	本地	中英	中	优秀	免费（需GPU）
bge-reranker-v2-m3	本地	多语言	低	良好	免费（轻量）
ms-marco-MiniLM	本地	英文	低	良好	免费
GPT-4 Rerank	API	多语言	高	优秀	高（按token）

LLM作为Rerank器

使用LLM对候选文档进行重排序：

def llm_rerank(query, documents, llm, top_n=5):
    """使用LLM进行重排序"""
    doc_list = "\n".join([f"[{i}] {doc[:200]}..." for i, doc in enumerate(documents)])
    
    prompt = f"""
    查询：{query}
    
    候选文档：
    {doc_list}
    
    请根据与查询的相关性，对候选文档排序。
    只输出排序后的编号，用逗号分隔，如：3,1,4,0,2
    
    排序结果：
    """
    
    result = llm.generate(prompt)
    indices = [int(x.strip()) for x in result.split(',')]
    
    return [documents[i] for i in indices[:top_n]]

优势：可以利用LLM的深层理解能力、支持复杂相关性判断劣势：延迟高、成本高、不适合实时场景

高级Rerank策略

多阶段Rerank

向量检索Top-50 → 轻量Rerank Top-20 → 重量Rerank Top-5

def multi_stage_rerank(query, vector_db, light_reranker, heavy_reranker):
    # 第一阶段：向量检索
    candidates = vector_db.search(query, top_k=50)
    
    # 第二阶段：轻量Rerank
    light_scores = light_reranker.predict([[query, c] for c in candidates])
    light_ranked = sorted(zip(candidates, light_scores), 
                          key=lambda x: x[1], reverse=True)[:20]
    
    # 第三阶段：重量Rerank
    top_candidates = [c for c, _ in light_ranked]
    heavy_scores = heavy_reranker.predict([[query, c] for c in top_candidates])
    final_ranked = sorted(zip(top_candidates, heavy_scores),
                          key=lambda x: x[1], reverse=True)[:5]
    
    return [c for c, _ in final_ranked]

分数融合

将向量检索分数和Rerank分数加权融合：

def fused_rerank(query, vector_db, reranker, alpha=0.3, top_n=5):
    """分数融合重排序"""
    # 向量检索
    results = vector_db.search(query, top_k=20)
    
    documents = [r['text'] for r in results]
    vector_scores = [r['score'] for r in results]
    
    # Rerank打分
    rerank_scores = reranker.predict([[query, doc] for doc in documents])
    
    # 归一化
    vector_scores = normalize(vector_scores)
    rerank_scores = normalize(rerank_scores)
    
    # 融合分数
    fused_scores = [alpha * v + (1 - alpha) * r 
                    for v, r in zip(vector_scores, rerank_scores)]
    
    # 排序
    ranked = sorted(zip(documents, fused_scores), 
                    key=lambda x: x[1], reverse=True)
    
    return [doc for doc, _ in ranked[:top_n]]

def normalize(scores):
    """Min-Max归一化"""
    min_s, max_s = min(scores), max(scores)
    if max_s == min_s:
        return [1.0] * len(scores)
    return [(s - min_s) / (max_s - min_s) for s in scores]

上下文感知Rerank

考虑文档间的多样性，避免返回高度相似的多条结果：

def diversity_rerank(query, documents, reranker, top_n=5, lambda_div=0.3):
    """考虑多样性的重排序"""
    # 初始Rerank分数
    relevance_scores = reranker.predict([[query, doc] for doc in documents])
    
    selected = []
    remaining = list(range(len(documents)))
    
    for _ in range(top_n):
        best_idx = None
        best_score = -float('inf')
        
        for idx in remaining:
            # 相关性分数
            rel_score = relevance_scores[idx]
            
            # 多样性分数（与已选文档的最大相似度的补）
            if selected:
                max_sim = max(
                    cosine_similarity(
                        embed(documents[idx]),
                        embed(documents[s])
                    ) for s in selected
                )
            else:
                max_sim = 0
            
            div_score = 1 - max_sim
            combined = (1 - lambda_div) * rel_score + lambda_div * div_score
            
            if combined > best_score:
                best_score = combined
                best_idx = idx
        
        selected.append(best_idx)
        remaining.remove(best_idx)
    
    return [documents[i] for i in selected]

工程优化

批处理优化

def batch_rerank(query, documents, reranker, batch_size=32):
    """批处理Rerank"""
    all_scores = []
    
    for i in range(0, len(documents), batch_size):
        batch = documents[i:i+batch_size]
        pairs = [[query, doc] for doc in batch]
        scores = reranker.predict(pairs)
        all_scores.extend(scores)
    
    return all_scores

缓存策略

from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_rerank(query_hash, doc_hashes, reranker_name):
    """缓存Rerank结果"""
    # 对相同query+documents组合缓存结果
    pass

# 在RAG流程中使用
query_hash = hash(query)
doc_hashes = tuple(hash(doc) for doc in documents)
result = cached_rerank(query_hash, doc_hashes, "bge-reranker-large")

异步Rerank

import asyncio

async def async_rerank(query, documents, reranker, top_n=5):
    """异步Rerank"""
    pairs = [[query, doc] for doc in documents]
    
    # 分批异步处理
    tasks = []
    batch_size = 10
    for i in range(0, len(pairs), batch_size):
        batch = pairs[i:i+batch_size]
        tasks.append(asyncio.to_thread(reranker.predict, batch))
    
    results = await asyncio.gather(*tasks)
    all_scores = [score for batch in results for score in batch]
    
    ranked = sorted(zip(documents, all_scores), 
                    key=lambda x: x[1], reverse=True)
    return [doc for doc, _ in ranked[:top_n]]

效果评估

评估指标

def evaluate_rerank(test_set, rerank_fn):
    """评估Rerank效果"""
    metrics = {'ndcg@5': [], 'precision@5': [], 'recall@5': [], 'mrr': []}
    
    for query, relevant_docs, candidate_docs in test_set:
        # Rerank前
        before_ranking = candidate_docs[:5]
        
        # Rerank后
        after_ranking = rerank_fn(query, candidate_docs, top_n=5)
        
        # 计算指标
        metrics['ndcg@5'].append(ndcg(at=5)(after_ranking, relevant_docs))
        metrics['precision@5'].append(precision(at=5)(after_ranking, relevant_docs))
        metrics['mrr'].append(rr(after_ranking, relevant_docs))
    
    return {k: np.mean(v) for k, v in metrics.items()}

典型提升效果

指标	仅向量检索	向量+Rerank	提升
NDCG@5	0.62	0.81	+30.6%
Precision@5	0.55	0.74	+34.5%
MRR	0.58	0.79	+36.2%

结语

Rerank是RAG系统中投入产出比最高的优化环节之一——它不需要修改检索索引，只需在检索结果上增加一步精排，就能显著提升检索精度。选择合适的Rerank模型、设计合理的两阶段检索流程、结合分数融合和多样性策略，可以构建高质量的RAG检索系统。在RAG效果优化中，建议优先评估和优化Rerank环节，通常能获得最显著的改善。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。

引言#

为什么需要Rerank#

双塔模型的局限#

Cross-Encoder的优势#

两阶段检索架构#

Rerank模型选择#

通用Rerank模型#

模型选择对比#

LLM作为Rerank器#

高级Rerank策略#

多阶段Rerank#

分数融合#

上下文感知Rerank#

工程优化#

批处理优化#

缓存策略#

异步Rerank#

效果评估#

评估指标#

典型提升效果#

结语#

加入讨论#

📌 相关推荐

引言