RAG 检索质量优化：从混合检索到重排序的工程链路

一、RAG 系统的检索瓶颈：召回率与精度的矛盾

RAG（检索增强生成）系统在生产环境中最大的瓶颈不是生成质量，而是检索质量。当知识库规模从千级文档增长到十万级时，纯向量检索的召回率急剧下降。根本原因是向量相似度搜索在语义空间中的精度有限——相近的向量不一定代表相关的内容，相关的内容也不一定向量相近。具体表现为两类问题：语义漂移（检索到语义相近但主题不同的文档）和粒度失配（检索到的片段缺少关键上下文）。

更深层的问题是，大多数 RAG 系统只关注"检索到了什么"，而忽略了"检索结果的质量如何"。没有质量评估的检索链路，就像没有单元测试的代码——看起来能跑，但出问题时无从定位。

二、混合检索与重排序的架构原理

现代 RAG 系统的标准架构是"混合检索 + 重排序"两阶段设计。第一阶段用多种检索策略扩大召回，第二阶段用精排模型过滤噪声。

flowchart LR
    Q[用户查询] --> QR[查询改写与扩展]
    QR --> VS[向量检索]
    QR --> KS[关键词检索 BM25]
    VS --> RR[结果融合 RRF]
    KS --> RR
    RR --> RM[重排序模型 Cross-Encoder]
    RM --> TK[Top-K 截断]
    TK --> LLM[大模型生成]

    subgraph 第一阶段：召回
        VS
        KS
        RR
    end

    subgraph 第二阶段：精排
        RM
        TK
    end

混合检索的核心是互补性：向量检索擅长语义匹配（"如何优化数据库性能"匹配到"SQL 调优方法"），关键词检索擅长精确匹配（"Go 1.22"精确匹配到版本号）。两者融合使用 Reciprocal Rank Fusion（RRF）算法，按排名倒数加权求和，避免分数尺度不一致的问题。

重排序使用 Cross-Encoder 模型对（查询, 文档）对做精排。与 Bi-Encoder（向量检索）不同，Cross-Encoder 让查询和文档在 Transformer 中做深度交互，精度更高但速度更慢。因此只在 Top-N 候选集上做重排，而非全库扫描。

三、混合检索与重排序的工程实现

import hashlib
from dataclasses import dataclass, field
from typing import Any

@dataclass
class Document:
    doc_id: str
    content: str
    metadata: dict[str, Any] = field(default_factory=dict)
    score: float = 0.0

class HybridRetriever:
    """混合检索器：向量检索 + BM25 关键词检索"""
    def __init__(self, vector_store, bm25_index, rrf_k: int = 60):
        self.vector_store = vector_store
        self.bm25_index = bm25_index
        self.rrf_k = rrf_k  # RRF 平滑参数

    def vector_search(self, query: str, top_k: int = 20) -> list[Document]:
        """向量检索：语义匹配"""
        # 实际实现调用向量数据库（Milvus/Weaviate/Qdrant）
        results = self.vector_store.search(query_embedding=query, top_k=top_k)
        return [Document(doc_id=r["id"], content=r["text"], score=r["score"]) for r in results]

    def bm25_search(self, query: str, top_k: int = 20) -> list[Document]:
        """BM25 关键词检索：精确匹配"""
        results = self.bm25_index.search(query, top_k=top_k)
        return [Document(doc_id=r["id"], content=r["text"], score=r["score"]) for r in results]

    def reciprocal_rank_fusion(
        self,
        result_lists: list[list[Document]],
    ) -> list[Document]:
        """RRF 融合：按排名倒数加权，消除分数尺度差异"""
        rrf_scores: dict[str, float] = {}
        doc_map: dict[str, Document] = {}

        for results in result_lists:
            for rank, doc in enumerate(results, start=1):
                if doc.doc_id not in rrf_scores:
                    rrf_scores[doc.doc_id] = 0.0
                    doc_map[doc.doc_id] = doc
                # RRF 公式：1 / (k + rank)
                rrf_scores[doc.doc_id] += 1.0 / (self.rrf_k + rank)

        # 按 RRF 分数降序排列
        sorted_ids = sorted(rrf_scores, key=rrf_scores.get, reverse=True)
        result = []
        for doc_id in sorted_ids:
            doc = doc_map[doc_id]
            doc.score = rrf_scores[doc_id]
            result.append(doc)
        return result

    def retrieve(self, query: str, top_k: int = 20) -> list[Document]:
        """混合检索入口"""
        vector_results = self.vector_search(query, top_k=top_k)
        bm25_results = self.bm25_search(query, top_k=top_k)
        return self.reciprocal_rank_fusion([vector_results, bm25_results])


class Reranker:
    """重排序器：Cross-Encoder 精排"""
    def __init__(self, model_name: str = "BAAI/bge-reranker-v2-m3"):
        self.model_name = model_name
        # 实际实现加载 Cross-Encoder 模型
        self._model = None

    def rerank(self, query: str, documents: list[Document], top_k: int = 5) -> list[Document]:
        """对候选文档做精排，返回 Top-K"""
        # 构造 (query, doc) 对
        pairs = [(query, doc.content) for doc in documents]

        # Cross-Encoder 打分（此处为伪代码，实际调用模型推理）
        scores = self._compute_scores(pairs)

        # 按分数降序排列
        scored_docs = list(zip(documents, scores))
        scored_docs.sort(key=lambda x: x[1], reverse=True)

        result = []
        for doc, score in scored_docs[:top_k]:
            doc.score = score
            result.append(doc)
        return result

    def _compute_scores(self, pairs: list[tuple[str, str]]) -> list[float]:
        """Cross-Encoder 推理（生产环境使用 GPU 加速）"""
        # 简化实现，实际使用 sentence-transformers 或 vLLM
        return [0.5] * len(pairs)


class RAGPipeline:
    """RAG 完整管线：检索 → 重排 → 生成"""
    def __init__(self, retriever: HybridRetriever, reranker: Reranker):
        self.retriever = retriever
        self.reranker = reranker

    async def query(self, question: str, client, top_k: int = 5) -> dict:
        # 第一阶段：混合检索召回
        candidates = self.retriever.retrieve(question, top_k=20)

        # 第二阶段：重排序精排
        ranked_docs = self.reranker.rerank(question, candidates, top_k=top_k)

        # 构造 Prompt
        context = "\n\n".join([f"[文档{i+1}] {doc.content}" for i, doc in enumerate(ranked_docs)])
        prompt = f"""基于以下检索结果回答问题。如果检索结果不足以回答，请明确说明。

检索结果：
{context}

问题：{question}"""

        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.1,
        )
        return {
            "answer": response.choices[0].message.content,
            "sources": [{"doc_id": d.doc_id, "score": d.score} for d in ranked_docs],
        }

四、RAG 检索链路的 Trade-offs 分析

召回率与延迟的权衡：混合检索需要同时执行两路检索，延迟约为单路检索的 1.5-2 倍。对于实时性要求高的场景（如客服），可以考虑异步并行执行两路检索，但需要处理结果到达时间不一致的问题。

重排序的计算成本：Cross-Encoder 对每个（查询, 文档）对做完整 Transformer 前向传播，20 个候选文档的重排序耗时约为向量检索的 5-10 倍。GPU 部署可以缓解，但增加了基础设施成本。替代方案是使用轻量级 LLM 做打分，精度略低但成本可控。

分块粒度的两难：小分块（256 token）检索精度高但缺少上下文，大分块（1024 token）上下文完整但引入噪声。生产建议使用"小分块检索 + 大分块返回"策略：检索时匹配小分块，返回时扩展到包含该分块的父文档段落。

知识库更新的时效性：向量索引的增量更新比全量重建快，但频繁更新会导致索引碎片化，影响检索质量。建议设置更新窗口：高频变更的知识用关键词检索覆盖，低频稳定的知识用向量检索覆盖。

五、总结

RAG 检索质量优化的核心思路是"先广后精"：混合检索扩大召回面，重排序模型过滤噪声。RRF 融合算法解决了不同检索策略分数尺度不一致的问题，Cross-Encoder 重排序提供了精度保障。落地时需要关注召回率与延迟的权衡、重排序的计算成本、分块粒度的选择以及知识库更新的时效性。建议先用纯向量检索验证基线效果，再逐步引入 BM25 混合检索和重排序，每一步都做 A/B 评估，避免过度优化。