重排序技术

重排序（Reranking）是在初步检索后对结果进行精细排序的过程。向量检索虽然快速，但相关性判断不够精确；重排序模型可以更准确地评估文档与问题的相关性，显著提升检索质量。

为什么需要重排序？

向量检索的局限

向量检索使用单一向量表示整段文本，存在信息压缩带来的精度损失：

查询: "年假计算方法"

向量检索结果:
1. "公司制度汇编" (0.85) - 包含年假，但也有大量无关内容
2. "员工手册第三章" (0.82) - 相关性一般
3. "年假申请流程" (0.80) - 不是计算方法

问题：检索结果相关性不够精确

重排序的优势

重排序模型可以深入分析查询和文档的交互：

重排序后:
1. "年假天数根据工龄计算：1-5年享5天..." (0.95) - 直接回答
2. "年假申请流程及计算规则说明" (0.88) - 相关且聚焦
3. "员工手册第三章：休假制度详解" (0.72) - 有用但不够直接

重排序模型原理

Cross-Encoder vs Bi-Encoder

Bi-Encoder（双塔模型）：

• 分别编码查询和文档
• 通过向量相似度判断相关性
• 速度快，但精度有限

Cross-Encoder（交叉编码器）：

• 将查询和文档一起输入模型
• 模型内部进行深度交互
• 精度高，但速度慢

# Bi-Encoder（向量检索）
query_embedding = model.encode("查询")      # 单独编码
doc_embedding = model.encode("文档")        # 单独编码
score = cosine_similarity(query_embedding, doc_embedding)

# Cross-Encoder（重排序）
score = model.predict([("查询", "文档")])   # 一起编码，直接输出分数

为什么 Cross-Encoder 更准？

Cross-Encoder 允许查询和文档在每个层级进行交互：

查询: "Python 如何处理 JSON"

文档: "JavaScript 中 JSON.parse() 方法..."

Bi-Encoder:
- 分别理解"Python处理JSON"和"JavaScript JSON"
- 可能认为语义相似

Cross-Encoder:
- 注意到"Python"与"JavaScript"的对比
- 判断文档不相关

重排序模型选择

商业模型

模型	提供商	特点	定价
Cohere Rerank	Cohere	多语言支持好	$2/1000次
Jina Reranker	Jina	快速、API友好	免费/付费
Voyage Rerank	Voyage	高精度	$0.05/1000次

开源模型

模型	特点	适用场景
BGE-reranker-large	中文优秀、开源免费	中文场景
BGE-reranker-v2-m3	多语言、长文本	国际化项目
MonoT5	经典模型	英文场景
ColBERT	细粒度交互	高精度需求

重排序实现

使用 Cohere Rerank

import cohere

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

def rerank_with_cohere(query, documents, top_n=5):
    """使用 Cohere 重排序"""
    # 提取文档文本
    docs = [doc.page_content for doc in documents]
    
    # 调用 API
    results = co.rerank(
        model="rerank-multilingual-v3.0",
        query=query,
        documents=docs,
        top_n=top_n
    )
    
    # 整理结果
    reranked = []
    for result in results.results:
        reranked.append({
            "document": documents[result.index],
            "relevance_score": result.relevance_score
        })
    
    return reranked

# 使用
query = "年假如何计算？"
initial_results = retriever.invoke(query)
reranked_results = rerank_with_cohere(query, initial_results)

使用 BGE Reranker

from sentence_transformers import CrossEncoder

# 加载模型
model = CrossEncoder('BAAI/bge-reranker-large')

def rerank_with_bge(query, documents, top_k=5):
    """使用 BGE 重排序"""
    # 构建输入对
    pairs = [(query, doc.page_content) for doc in documents]
    
    # 计算分数
    scores = model.predict(pairs)
    
    # 排序
    scored_docs = list(zip(documents, scores))
    scored_docs.sort(key=lambda x: x[1], reverse=True)
    
    # 返回 top_k
    return scored_docs[:top_k]

# 使用
query = "如何提高团队效率？"
initial_results = vectorstore.similarity_search(query, k=20)
reranked = rerank_with_bge(query, initial_results, top_k=5)

LangChain 集成

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank

# 创建重排序压缩器
compressor = CohereRerank(
    model="rerank-multilingual-v3.0",
    top_n=5
)

# 创建重排序检索器
reranking_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20})
)

# 检索（自动重排序）
results = reranking_retriever.invoke("年假计算方法")

FlagEmbedding（BGE）集成

from FlagEmbedding import FlagReranker

# 初始化
reranker = FlagReranker(
    'BAAI/bge-reranker-large',
    use_fp16=True  # 加速
)

def rerank_documents(query, documents, top_k=5):
    """BGE 重排序"""
    pairs = [[query, doc.page_content] for doc in documents]
    scores = reranker.compute_score(pairs, normalize=True)
    
    # 排序
    results = sorted(
        zip(documents, scores),
        key=lambda x: x[1],
        reverse=True
    )
    
    return results[:top_k]

重排序流程

标准流程

def retrieve_with_reranking(query, vectorstore, reranker, k_initial=20, k_final=5):
    """带重排序的检索流程"""
    
    # 1. 初步检索（粗排）
    initial_results = vectorstore.similarity_search(query, k=k_initial)
    
    # 2. 重排序（精排）
    pairs = [(query, doc.page_content) for doc in initial_results]
    scores = reranker.predict(pairs)
    
    # 3. 合并分数并排序
    scored_results = list(zip(initial_results, scores))
    scored_results.sort(key=lambda x: x[1], reverse=True)
    
    # 4. 返回 top_k
    final_results = [doc for doc, score in scored_results[:k_final]]
    
    return final_results, scored_results[:k_final]

带分数过滤

def retrieve_with_threshold(query, vectorstore, reranker, min_score=0.5):
    """带分数阈值过滤"""
    
    # 初步检索
    initial_results = vectorstore.similarity_search(query, k=30)
    
    # 重排序
    pairs = [(query, doc.page_content) for doc in initial_results]
    scores = reranker.predict(pairs)
    
    # 过滤低分结果
    filtered = [
        (doc, score) 
        for doc, score in zip(initial_results, scores)
        if score >= min_score
    ]
    
    # 按分数排序
    filtered.sort(key=lambda x: x[1], reverse=True)
    
    return filtered

重排序效果评估

评估指标

def evaluate_reranking(test_cases, retriever, reranker):
    """评估重排序效果"""
    
    metrics = {
        "ndcg@5": [],
        "ndcg@10": [],
        "mrr": [],
        "hit_rate": []
    }
    
    for case in test_cases:
        query = case["query"]
        relevant_ids = set(case["relevant_ids"])
        
        # 初步检索
        initial = retriever.invoke(query)
        
        # 重排序
        pairs = [(query, doc.page_content) for doc in initial]
        scores = reranker.predict(pairs)
        
        ranked = sorted(zip(initial, scores), key=lambda x: x[1], reverse=True)
        ranked_ids = [doc.metadata["id"] for doc, _ in ranked]
        
        # 计算 NDCG@5
        metrics["ndcg@5"].append(ndcg_at_k(ranked_ids, relevant_ids, 5))
        
        # 计算 MRR
        for i, doc_id in enumerate(ranked_ids):
            if doc_id in relevant_ids:
                metrics["mrr"].append(1 / (i + 1))
                break
        else:
            metrics["mrr"].append(0)
    
    return {k: sum(v) / len(v) for k, v in metrics.items()}

A/B 测试

def ab_test_reranking(queries, retriever, reranker, llm, evaluator):
    """A/B 测试重排序效果"""
    
    results = {"with_rerank": [], "without_rerank": []}
    
    for query in queries:
        # 无重排序
        docs_no_rerank = retriever.invoke(query)
        answer_no_rerank = llm.invoke(query, docs_no_rerank)
        score_no_rerank = evaluator.evaluate(query, answer_no_rerank)
        results["without_rerank"].append(score_no_rerank)
        
        # 有重排序
        initial = retriever.invoke(query)
        reranked = rerank(query, initial, reranker)
        answer_with_rerank = llm.invoke(query, reranked)
        score_with_rerank = evaluator.evaluate(query, answer_with_rerank)
        results["with_rerank"].append(score_with_rerank)
    
    return {
        "avg_without": sum(results["without_rerank"]) / len(results["without_rerank"]),
        "avg_with": sum(results["with_rerank"]) / len(results["with_rerank"]),
        "improvement": ...
    }

重排序最佳实践

参数调优

# 初步检索数量 vs 重排序数量
# 经验值：初步检索 20-50，重排序后返回 5-10

# 分数阈值
# 建议：设置最低分数阈值（如 0.3-0.5）过滤低质量结果

# 模型选择
# 中文：BGE-reranker-large
# 多语言：Cohere rerank-multilingual 或 BGE-reranker-v2-m3
# 英文：Cohere rerank-english 或 MonoT5

性能优化

# 批量处理
def batch_rerank(queries, documents, reranker, batch_size=32):
    """批量重排序"""
    all_pairs = []
    for query in queries:
        all_pairs.extend([(query, doc.page_content) for doc in documents])
    
    # 分批计算
    all_scores = []
    for i in range(0, len(all_pairs), batch_size):
        batch = all_pairs[i:i+batch_size]
        scores = reranker.predict(batch)
        all_scores.extend(scores)
    
    return all_scores

# GPU 加速
reranker = CrossEncoder('BAAI/bge-reranker-large', device='cuda')

常见问题

问题 1：重排序后结果变差

可能原因：

• 重排序模型与嵌入模型不匹配
• 初步检索结果质量太差
• 领域特定术语处理不当

解决方案：

• 尝试不同的重排序模型
• 增加初步检索数量
• 使用领域微调的模型

问题 2：重排序延迟太高

解决方案：

• 减少初步检索数量
• 使用更快的模型
• 缓存重排序结果

小结

重排序是提升 RAG 检索质量的有效手段：

1. Cross-Encoder 比向量检索更精确
2. 商业模型如 Cohere 易用，开源模型如 BGE 免费
3. 初步检索 20-50，重排序返回 5-10 是常用配置
4. 持续评估效果，调优参数

下一步

• 高级技术：探索更多检索优化方法
• 实战实现：完整代码示例

参考资料

• Cohere Rerank API
• BGE Reranker
• ColBERT