跳至主要內容

Re-ranking 让 RAG 准确率提升了 15%——我们怎么做的

老张2026/4/30大约 6 分钟

Re-ranking 让 RAG 准确率提升了 15%——我们怎么做的

适读人群：RAG 系统已上线但回答质量还不够好的开发者 | 阅读时长：约 13 分钟 | 核心价值：完整的 Re-ranking 实现方案和真实数据对比

加 Re-ranking 这个决定，是被用户逼出来的。

系统上线三个月，有个用户一直跟我反馈说："查 XXX 产品的安装步骤，出来的第一个结果总是不对。"

我拉了日志看，检索回来的 Top-5 文档里，真正有用的那篇排在第 4 位，每次都是第 4 位。LLM 处理 Top-4 的时候，注意力分配在前面的文档上，对第 4 条的参考权重低，生成的答案就偏了。

问题很清楚：向量检索的排序不够精准，相关的文档确实检索到了，但没排在前面。

Re-ranking 就是在向量检索之后，加一个更精细的重排序步骤，把真正相关的文档排到更靠前的位置。

Re-ranking 的原理

向量检索用的是 Bi-encoder 结构：query 和 document 分别 Embedding，用向量相似度比较。速度很快，但损失了 query 和 document 之间的交叉信息。

Re-ranking 用的是 Cross-encoder 结构：把 query 和 document 拼接在一起，一次性输入模型，让模型直接判断相关性分数。

Cross-encoder 的精度高得多，但速度慢——每对 (query, doc) 都要单独计算，不能像向量检索一样提前存好向量。所以不能用它做第一阶段的召回（候选集太大），只能用来对已经召回的少量文档重排序。

这就是 RAG 两阶段检索的基本思路：

  • 第一阶段（Recall）：向量检索 + BM25，宽松召回 Top-20 或 Top-50
  • 第二阶段（Rank）：Re-ranker 精排，取 Top-5

选哪个 Re-ranking 模型

市面上常用的：

bge-reranker 系列（BAAI）

中文效果最好的开源 Re-ranker，bge-reranker-v2-m3 支持多语言，是我目前的默认选择。

Cohere Rerank API

商业 API，效果好，不用自己部署，按调用量收费。适合早期验证阶段，量大了成本高。

jina-reranker

Jina 出的 Re-ranker，开源版本可以本地部署，效果不错。

我的建议：先用 Cohere API 验证效果，效果好了再换成本地部署的 bge-reranker。

完整实现代码

方案一：用 Cohere API（验证阶段）

import cohere
from typing import List, Dict, Any

class CohereReranker:
    def __init__(self, api_key: str, model: str = "rerank-multilingual-v3.0"):
        self.client = cohere.Client(api_key)
        self.model = model
    
    def rerank(
        self, 
        query: str, 
        documents: List[Dict[str, Any]], 
        top_n: int = 5,
        text_field: str = "content"
    ) -> List[Dict[str, Any]]:
        """
        documents: 第一阶段检索的结果，每个元素是 {"content": "...", ...}
        返回: 重排后的 top_n 个文档
        """
        if not documents:
            return []
        
        # Cohere 接受纯文本列表
        texts = [doc[text_field] for doc in documents]
        
        response = self.client.rerank(
            query=query,
            documents=texts,
            model=self.model,
            top_n=min(top_n, len(documents))
        )
        
        reranked = []
        for result in response.results:
            original_doc = documents[result.index]
            reranked.append({
                **original_doc,
                "rerank_score": result.relevance_score,
                "original_rank": result.index
            })
        
        return reranked

方案二：本地 bge-reranker（生产阶段）

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch
import numpy as np
from typing import List, Dict, Any

class BGEReranker:
    def __init__(
        self, 
        model_name: str = "BAAI/bge-reranker-v2-m3",
        device: str = None,
        batch_size: int = 32
    ):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
        self.batch_size = batch_size
        
        if device is None:
            self.device = "cuda" if torch.cuda.is_available() else "cpu"
        else:
            self.device = device
        
        self.model.to(self.device)
        self.model.eval()
        print(f"BGE Reranker 加载完成，使用设备: {self.device}")
    
    def _compute_scores(self, pairs: List[tuple]) -> List[float]:
        """
        pairs: [(query, doc_text), ...]
        """
        all_scores = []
        
        for i in range(0, len(pairs), self.batch_size):
            batch = pairs[i:i + self.batch_size]
            
            inputs = self.tokenizer(
                batch,
                padding=True,
                truncation=True,
                max_length=512,
                return_tensors="pt"
            ).to(self.device)
            
            with torch.no_grad():
                outputs = self.model(**inputs)
                # 取 logits，Cross-encoder 输出的是相关性 logit
                scores = outputs.logits.squeeze(-1).float()
                # 用 sigmoid 转为 0-1 的相关性概率
                scores = torch.sigmoid(scores).cpu().numpy()
            
            if isinstance(scores, np.floating):
                all_scores.append(float(scores))
            else:
                all_scores.extend(scores.tolist())
        
        return all_scores
    
    def rerank(
        self,
        query: str,
        documents: List[Dict[str, Any]],
        top_n: int = 5,
        text_field: str = "content"
    ) -> List[Dict[str, Any]]:
        if not documents:
            return []
        
        pairs = [(query, doc[text_field]) for doc in documents]
        scores = self._compute_scores(pairs)
        
        # 按分数排序
        scored_docs = list(zip(documents, scores))
        scored_docs.sort(key=lambda x: x[1], reverse=True)
        
        result = []
        for rank, (doc, score) in enumerate(scored_docs[:top_n]):
            result.append({
                **doc,
                "rerank_score": score,
                "rerank_rank": rank + 1
            })
        
        return result

把两个阶段串起来

class TwoStageRAGRetriever:
    def __init__(
        self,
        vector_retriever,      # 向量检索器
        bm25_retriever,        # BM25检索器  
        reranker,              # Re-ranker
        recall_top_k: int = 20,  # 第一阶段召回数量
        final_top_k: int = 5     # 最终返回数量
    ):
        self.vector_retriever = vector_retriever
        self.bm25_retriever = bm25_retriever
        self.reranker = reranker
        self.recall_top_k = recall_top_k
        self.final_top_k = final_top_k
    
    def retrieve(self, query: str) -> List[Dict[str, Any]]:
        # 第一阶段：混合检索，宽松召回
        vector_results = self.vector_retriever.search(query, top_k=self.recall_top_k)
        bm25_results = self.bm25_retriever.search(query, top_k=self.recall_top_k)
        
        # 用 RRF 合并两路结果，去重
        merged = self._rrf_merge(vector_results, bm25_results)
        
        # 取前 recall_top_k 个候选
        candidates = merged[:self.recall_top_k]
        
        if not candidates:
            return []
        
        # 第二阶段：Re-ranking 精排
        reranked = self.reranker.rerank(
            query=query,
            documents=[c["doc"] for c in candidates],
            top_n=self.final_top_k
        )
        
        return reranked
    
    def _rrf_merge(self, list1, list2, k=60):
        """简单 RRF 合并，去重"""
        scores = {}
        docs = {}
        
        for rank, item in enumerate(list1):
            doc_id = item["doc"]["id"]
            scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank + 1)
            docs[doc_id] = item["doc"]
        
        for rank, item in enumerate(list2):
            doc_id = item["doc"]["id"]
            scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank + 1)
            docs[doc_id] = item["doc"]
        
        sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)
        return [{"doc": docs[doc_id], "rrf_score": scores[doc_id]} for doc_id in sorted_ids]

Before / After 数据对比

在一个客服知识库项目（约 8000 个文档，100 个标注测试问题）上的对比：

检索质量对比（人工标注）：

方案Precision@3Recall@5MRR@5
向量检索（Top-5）64.3%71.2%0.61
混合检索（Top-5）69.1%77.8%0.66
混合检索 + Re-ranking81.4%82.3%0.79

端到端回答质量（GPT-4o 评分，1-5分）：

方案平均评分完全正确率
向量检索（Top-5）3.451%
混合检索（Top-5）3.758%
混合检索 + Re-ranking4.273%

加了 Re-ranking 之后，完全正确率从 58% 提升到 73%，提升了 15 个百分点。这就是文章标题里那个 15% 的来源。

什么时候值得加 Re-ranking

不是所有场景都值得加这一层，以下几个信号说明你需要它：

值得加的场景：

  • 检索回来的文档明明相关，但排序不对，导致 LLM 生成答案时参考了错误的部分
  • 用户问题多样，有精确查询也有语义查询，单一检索策略顾此失彼
  • 文档长度差异大，短文档总是比长文档排名靠前（向量归一化的问题）
  • 准确率要求高（比如医疗、法律、合规类场景）

可以暂时不加的场景：

  • 文档量很少（几百个 chunk），向量检索精度已经很高
  • 对响应速度要求极高（本地 Re-ranker 会增加 100-500ms 延迟）
  • 预算有限，Cohere API 每千次约 $1，本地部署需要 GPU

性能影响

本地 bge-reranker-v2-m3，CPU 推理，对 20 个候选文档重排序，耗时大概 200-600ms（取决于文档长度）。有 GPU 的话 50ms 以内。

如果用 Cohere API，网络往返大概 200-400ms。

这个延迟加在用户体验上是感知不到的，LLM 本身就要 1-3 秒，多个 200ms 不影响使用感受。

编辑此页
最近更新:
贡献者: xiaozhangge