引言

在RAG系统中，向量检索负责从海量文档中快速召回候选结果，但召回结果的排序不一定准确。重排序（Reranking）作为第二步，对候选结果进行更精确的排序，将最相关的文档排在最前面。

2026年，重排序已经成为RAG系统的标配组件。一个好的重排序模型可以将RAG的准确率提升20-40%。本文将深入探讨重排序技术。

一、为什么需要重排序

1.1 向量检索的局限

向量检索使用双塔模型（Bi-encoder）：查询和文档分别编码，通过向量相似度排序。这种方式速度快，但精度有限——因为查询和文档的交互发生在最后的相似度计算阶段。

1.2 重排序的优势

重排序使用交叉编码器（Cross-encoder）：将查询和文档拼接在一起输入模型，模型可以充分理解查询和文档的关系。精度更高，但速度更慢。

向量检索: encode(query) · encode(doc) → similarity score  (快，精度中等)
重排序:   encode([query, doc]) → relevance score  (慢，精度高)

1.3 两阶段架构

查询 → 向量检索（召回Top-100）→ 重排序（精排Top-10）→ 生成

向量检索负责"快而全"，重排序负责"准而精"。

二、重排序模型

2.1 交叉编码器

class CrossEncoderReranker:
    def __init__(self, model_name="cross-encoder/ms-marco-MiniLM-L-12-v2"):
        self.model = load_model(model_name)
    
    async def rerank(self, query, documents, top_k=10):
        # 1. 构建query-document对
        pairs = [(query, doc.text) for doc in documents]
        
        # 2. 模型打分
        scores = self.model.predict(pairs)
        
        # 3. 排序
        ranked = sorted(zip(documents, scores), key=lambda x: -x[1])
        
        return [doc for doc, score in ranked[:top_k]]

2.2 LLM重排序

class LLMReranker:
    def __init__(self, llm):
        self.llm = llm
    
    async def rerank(self, query, documents, top_k=10):
        # 方法1: 逐个打分
        scores = []
        for doc in documents:
            score = await self.score_document(query, doc)
            scores.append(score)
        
        # 方法2: 两两比较（更准确但更慢）
        # ranked = await self.pairwise_rank(query, documents)
        
        # 排序
        ranked = sorted(zip(documents, scores), key=lambda x: -x[1])
        return [doc for doc, score in ranked[:top_k]]
    
    async def score_document(self, query, document):
        """使用LLM对query-document对打分"""
        prompt = f"""
        查询: {query}
        文档: {document.text[:500]}
        
        请评估文档与查询的相关性:
        0 - 完全不相关
        1 - 弱相关
        2 - 部分相关
        3 - 相关
        4 - 高度相关
        
        只输出数字。
        """
        result = await self.llm.call(prompt)
        return int(result.strip())
    
    async def pairwise_rank(self, query, documents):
        """两两比较排序"""
        # 使用类似冒泡排序的方法
        for i in range(len(documents)):
            for j in range(i+1, len(documents)):
                comparison = await self.compare(query, documents[i], documents[j])
                if comparison == "B is better":
                    documents[i], documents[j] = documents[j], documents[i]
        return documents

2.3 ColBERT

ColBERT是一种延迟交互模型，介于双塔和交叉编码器之间：

class ColBERTReranker:
    def __init__(self):
        self.query_encoder = load_model("colbert-query-encoder")
        self.doc_encoder = load_model("colbert-doc-encoder")
    
    async def rerank(self, query, documents, top_k=10):
        # 1. 编码query（多向量表示）
        query_embeddings = self.query_encoder.encode(query)  # [num_query_tokens, dim]
        
        # 2. 编码documents（多向量表示，可以预计算）
        doc_embeddings = [self.doc_encoder.encode(doc.text) for doc in documents]
        
        # 3. 计算MaxSim
        scores = []
        for doc_emb in doc_embeddings:
            # 对query的每个token，找到doc中最相似的token
            sim_matrix = query_embeddings @ doc_emb.T  # [num_q_tokens, num_d_tokens]
            max_sim = sim_matrix.max(axis=1).sum()  # 每个query token取max，然后求和
            scores.append(max_sim)
        
        # 4. 排序
        ranked = sorted(zip(documents, scores), key=lambda x: -x[1])
        return [doc for doc, score in ranked[:top_k]]

三、重排序策略

3.1 多模型集成

class EnsembleReranker:
    def __init__(self):
        self.rerankers = [
            CrossEncoderReranker("cross-encoder/ms-marco-MiniLM-L-12-v2"),
            CrossEncoderReranker("cross-encoder/stsb-roberta-large"),
            LLMReranker(llm)
        ]
        self.weights = [0.4, 0.3, 0.3]
    
    async def rerank(self, query, documents, top_k=10):
        all_scores = []
        
        for reranker in self.rerankers:
            scores = await reranker.score(query, documents)
            # 归一化
            scores = self.normalize(scores)
            all_scores.append(scores)
        
        # 加权融合
        final_scores = np.average(all_scores, axis=0, weights=self.weights)
        
        # 排序
        ranked = sorted(zip(documents, final_scores), key=lambda x: -x[1])
        return [doc for doc, score in ranked[:top_k]]

3.2 分层重排序

class TieredReranker:
    async def rerank(self, query, documents, top_k=10):
        # 第一层：快速重排序（轻量模型），从100→30
        first_pass = await self.light_reranker.rerank(query, documents, top_k=30)
        
        # 第二层：精确重排序（重量模型），从30→10
        second_pass = await self.heavy_reranker.rerank(query, first_pass, top_k=top_k)
        
        return second_pass

3.3 上下文感知重排序

class ContextAwareReranker:
    async def rerank(self, query, documents, conversation_history=None):
        """考虑对话历史的重排序"""
        # 如果有对话历史，扩展query
        if conversation_history:
            expanded_query = await self.expand_query_with_history(
                query, conversation_history
            )
        else:
            expanded_query = query
        
        # 重排序
        return await self.cross_encoder.rerank(expanded_query, documents)

四、模型对比

模型类型延迟精度适用场景
MiniLM-L-12交叉编码器中中通用场景
MiniLM-L-6交叉编码器低中低低延迟需求
roberta-large交叉编码器高高高精度需求
ColBERT延迟交互中高平衡精度和速度
LLM (GPT-4o)LLM重排序很高最高最高精度需求
BGE-reranker交叉编码器低中高中文场景

五、优化

5.1 缓存

class CachedReranker:
    def __init__(self, reranker):
        self.reranker = reranker
        self.cache = {}
    
    async def rerank(self, query, documents):
        cache_key = hash(query + str([d.id for d in documents]))
        
        if cache_key in self.cache:
            return self.cache[cache_key]
        
        result = await self.reranker.rerank(query, documents)
        self.cache[cache_key] = result
        
        return result

5.2 批量化

class BatchReranker:
    async def rerank(self, query, documents, batch_size=32):
        scores = []
        
        for i in range(0, len(documents), batch_size):
            batch = documents[i:i+batch_size]
            batch_scores = await self.model.score_batch(
                [(query, doc.text) for doc in batch]
            )
            scores.extend(batch_scores)
        
        ranked = sorted(zip(documents, scores), key=lambda x: -x[1])
        return ranked

六、评估

class RerankerEvaluator:
    async def evaluate(self, reranker, test_cases):
        metrics = {
            "ndcg_10": [],
            "map": [],
            "recall_10": [],
            "latency_ms": []
        }
        
        for case in test_cases:
            start = time.time()
            reranked = await reranker.rerank(case.query, case.candidate_docs, top_k=10)
            elapsed = (time.time() - start) * 1000
            
            metrics["ndcg_10"].append(self.ndcg(reranked, case.relevant_docs, 10))
            metrics["map"].append(self.average_precision(reranked, case.relevant_docs))
            metrics["recall_10"].append(self.recall_at_k(reranked, case.relevant_docs, 10))
            metrics["latency_ms"].append(elapsed)
        
        return {k: np.mean(v) for k, v in metrics.items()}

结语

重排序是RAG系统中"投入产出比"最高的优化——用一个轻量级的重排序步骤，就能显著提升检索质量。

2026年的趋势是"LLM重排序"——随着LLM推理成本下降，越来越多的系统使用LLM进行重排序，因为LLM能理解更复杂的语义关系。

但记住：重排序不是万能的。如果向量检索阶段没有召回相关文档，重排序也无能为力。好的RAG系统需要在每个环节都做好——分块、编码、检索、重排序、生成，缺一不可。

加入讨论

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

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