2.5 检索效果优化（下）

B -.->|速度快 10-100x| B
C -.->|精度高 10-30%| C
​

</div>

**两阶段检索的核心价值在于**：向量索引让检索从"线性扫描全库"变成"毫秒级 Top-K 查询"，代价是 Bi-Encoder 的双塔架构天然丢失了词级别的交互信号。比如查询"Python 中如何读取 CSV 文件"，向量检索可能返回一篇讲"如何用 Python 进行数据分析"的文档（语义接近但答非所问），因为双塔模型只能看到"整体语义相似度"，看不到"读取 CSV"这个关键短语的精确匹配。Cross-Encoder 能捕捉这种细粒度信号，因为它是把查询和文档拼在一起做联合编码的。

**重排序模型选型指南**：目前主流的重排序模型及其适用场景如下——

| 模型 | 参数量 | 语言 | 特点 | 适用场景 |
|------|--------|------|------|----------|
| BAAI/bge-reranker-v2-m3 | 560M | 中英 | 多语言、支持长文本、效果最好 | 中英文混合知识库、生产环境首选 |
| BAAI/bge-reranker-large | 560M | 中英 | 平衡效果与速度 | 通用场景，资源充足时选用 |
| BAAI/bge-reranker-base | 270M | 中英 | 速度快、效果不错 | 延迟敏感场景 |
| cross-encoder/ms-marco-MiniLM-L-6-v2 | 110M | 英文 | 轻量、推理极快 | 纯英文、高并发场景 |
| Cohere rerank-english-v3.0 | API | 英文 | 商业 API、免部署 | 不想自建模型的团队 |

选型建议：如果你的知识库以中文为主，优先选 BGE 系列（v2-m3 效果最好但需要更多显存）；如果延迟是第一优先级且知识库不大，选 bge-reranker-base；如果完全不想管模型部署，用 Cohere 或 Jina 的 Rerank API。关键原则——重排序模型和向量化模型不要求是同一个系列，但在实践中同系列（如都用 BGE）通常配合得更好，因为训练数据和分词器一致。

## 环境准备 / 前置知识

- 掌握检索算法基础（本教程 2.2 节）
- 掌握检索效果优化基础（本教程 2.4 节）
- Python 依赖：sentence-transformers, rank_bm25, numpy
- 建议环境：Python 3.10+, 8GB 以上显存（使用 GPU 推理）

## 分步实战

### 步骤 1：Cross-Encoder 重排序实现

先实现一个基础但完整的重排序器，理解其工作原理。

```python
from sentence_transformers import CrossEncoder
import numpy as np
from typing import List, Dict

class Reranker:
    """Cross-Encoder 重排序器
    
    使用方法：
    1. 初始化时加载模型
    2. 调用 rerank() 对粗排结果进行精排
    """
    
    def __init__(self, model_name: str = 'BAAI/bge-reranker-base'):
        self.model = CrossEncoder(model_name)
        self.model_name = model_name
    
    def rerank(self, query: str, documents: List[Dict],
               top_k: int = 5) -> List[Dict]:
        """对检索结果进行重排序
        
        Args:
            query: 用户查询
            documents: 候选文档列表，每个包含 content 和原始 score
            top_k: 返回 top-k 结果
        """
        if not documents:
            return []
        
        # 构建 (query, document) 对
        pairs = [(query, doc['content']) for doc in documents]
        
        # Cross-Encoder 评分——这里是批量推理，比逐个快很多
        scores = self.model.predict(pairs)
        
        # 按重排序分数排序
        scored_docs = []
        for doc, score in zip(documents, scores):
            scored_docs.append({
                **doc,
                'rerank_score': float(score),
                'original_score': doc.get('score', 0)
            })
        
        scored_docs.sort(key=lambda x: x['rerank_score'], reverse=True)
        return scored_docs[:top_k]

# 使用示例
reranker = Reranker('BAAI/bge-reranker-base')

# 模拟粗排结果——注意最后两条是相关性较低的噪声
coarse_results = [
    {"content": "RAG通过检索增强生成，有效减少大语言模型的幻觉问题。", "score": 0.85},
    {"content": "检索增强生成的核心挑战在于检索质量和上下文管理。", "score": 0.82},
    {"content": "向量数据库是RAG系统的重要基础设施。", "score": 0.78},
    {"content": "大语言模型的训练需要大量计算资源。", "score": 0.65},
    {"content": "自然语言处理是AI的重要分支领域。", "score": 0.60},
]

reranked = reranker.rerank("RAG如何减少幻觉？", coarse_results, top_k=3)
for doc in reranked:
    print(f"重排分数: {doc['rerank_score']:.3f} | "
          f"{doc['content'][:50]}...")
# 输出示例：
# 重排分数: 0.982 | RAG通过检索增强生成，有效减少大语言模型的幻觉问题...
# 重排分数: 0.945 | 检索增强生成的核心挑战在于检索质量和上下文管理...
# 重排分数: 0.312 | 向量数据库是RAG系统的重要基础设施...
​

为什么最后一条噪声被排除了？ 注意看分数的变化：粗排中"向量数据库"的分数是 0.78（排第三），但在重排后它的分数降到 0.312——因为 Cross-Encoder 发现这篇文档虽然"语义上跟 RAG 有关"，但并不回答"如何减少幻觉"这个具体问题。这就是两阶段检索的核心价值：粗排看语义相关性，精排看问题匹配度。

性能优化技巧：如果你需要在 CPU 上运行重排序（比如没有 GPU 的边缘部署场景），可以限制候选数量到 20 个以内，并使用 bge-reranker-base 而非 large 版本。实测在 CPU 上对 20 个候选做重排序，base 版本大约需要 200-500ms，可以接受。

步骤 2：查询意图识别与动态路由

不同类型的查询需要不同的检索策略。事实型查询（"什么是 RAG"）适合少量精准结果；操作型查询（"如何优化检索效果"）需要覆盖多个步骤的文档；对比型查询（"FAISS 和 Milvus 的区别"）需要从多个角度检索。

from enum import Enum
from typing import Dict

class QueryIntent(Enum):
    FACTUAL = "factual"         # 事实型：什么是X？
    PROCEDURAL = "procedural"   # 操作型：如何做X？
    COMPARATIVE = "comparative" # 对比型：X和Y的区别？
    EXPLORATORY = "exploratory" # 探索型：关于X有什么信息？

class QueryRouter:
    """查询意图识别与路由
    
    设计思路：
    1. 先用规则匹配明确的模式（速度快、准确率高）
    2. 规则无法判断时，用 LLM 做兜底分类
    """
    
    INTENT_PATTERNS = {
        QueryIntent.FACTUAL: ['什么是', '是什么', '定义', '概念', 
                               'who', 'what is', '解释一下'],
        QueryIntent.PROCEDURAL: ['如何', '怎么', '怎样', '步骤', 
                                 'how to', '方法', '实现'],
        QueryIntent.COMPARATIVE: ['区别', '对比', '比较', 'vs', 
                                  '不同', 'difference', '哪个好'],
        QueryIntent.EXPLORATORY: ['关于', '介绍', '分析', '了解', 
                                  'tell me about', '总结']
    }
    
    def classify_intent(self, query: str) -> QueryIntent:
        """基于规则的意图识别"""
        query_lower = query.lower()
        best_intent = QueryIntent.EXPLORATORY
        max_matches = 0
        
        for intent, keywords in self.INTENT_PATTERNS.items():
            matches = sum(1 for kw in keywords if kw in query_lower)
            if matches > max_matches:
                max_matches = matches
                best_intent = intent
        
        return best_intent
    
    def get_retrieval_config(self, intent: QueryIntent) -> Dict:
        """根据意图返回检索配置
        
        核心思想：不同意图需要不同的检索"形状"——
        事实型要"准"，操作型要"全"，对比型要"多角度"
        """
        configs = {
            QueryIntent.FACTUAL: {
                'top_k': 3, 'rerank_top_k': 3,
                'use_keyword': True, 'keyword_weight': 0.6,
                'description': '事实型查询：少量高质量结果，精确匹配优先'
            },
            QueryIntent.PROCEDURAL: {
                'top_k': 10, 'rerank_top_k': 5,
                'use_keyword': True, 'keyword_weight': 0.4,
                'description': '操作型查询：多步骤覆盖，确保流程完整'
            },
            QueryIntent.COMPARATIVE: {
                'top_k': 8, 'rerank_top_k': 5,
                'use_keyword': False, 'keyword_weight': 0.3,
                'description': '对比型查询：多角度覆盖，语义匹配为主'
            },
            QueryIntent.EXPLORATORY: {
                'top_k': 5, 'rerank_top_k': 5,
                'use_keyword': False, 'keyword_weight': 0.2,
                'description': '探索型查询：语义匹配为主，适度多样'
            }
        }
        return configs[intent]

# 演示：不同查询的意图识别和策略差异
router = QueryRouter()
demo_queries = [
    "什么是RAG技术？",
    "如何优化RAG的检索效果？",
    "FAISS和Milvus的区别是什么？",
    "关于向量数据库的性能优化"
]

for q in demo_queries:
    intent = router.classify_intent(q)
    config = router.get_retrieval_config(intent)
    print(f"查询: '{q}'")
    print(f"  意图: {intent.value} | {config['description']}")
    print(f"  Top-K: {config['top_k']} -> 重排后: {config['rerank_top_k']}")
    print()
​

意图路由为什么重要？ 一个实际案例：用户问"FAISS 和 Milvus 有什么区别"，如果系统用事实型查询的策略（top_k=3），很可能只能拿到一两篇相关文档，无法覆盖两个系统的多个维度对比。但如果系统识别出这是对比型查询，会把 top_k 调到 8，确保同时检索到关于 FAISS 和 Milvus 的多篇文档，再由重排序精选出最有对比价值的 5 篇。最终 LLM 拿到的上下文信息更全面，回答质量显著提升。

进阶：用 LLM 做意图分类。 规则方案对明确的模式效果很好，但对模糊查询（如"帮我看看 RAG"）就力不从心了。这时可以用 LLM 做兜底——把意图分类当作一个简单的分类任务，prompt 大约 100 token，延迟增加约 200ms，但准确率能从规则的 75% 提升到 92% 以上。建议只对规则置信度低的查询走 LLM 兜底，兼顾成本和准确率。

步骤 3：检索结果去重与多样性选择

检索结果中经常出现内容高度相似的多篇文档。比如同一份 API 文档被分成了多个 chunk，用户问一个问题可能召回 5 个内容几乎一样的片段。这不仅浪费 LLM 的上下文窗口，还可能导致回答信息单一。

from typing import List, Dict
import numpy as np

class ResultDeduplicator:
    """检索结果去重器
    
    原理：计算候选文档之间的语义相似度，
    如果新文档与已选文档的相似度超过阈值，则跳过。
    """
    
    def __init__(self, similarity_threshold: float = 0.92):
        self.threshold = similarity_threshold
    
    def deduplicate(self, results: List[Dict], 
                    encoder=None) -> List[Dict]:
        """去除语义重复的检索结果"""
        if not encoder or len(results) <= 1:
            return results
        
        unique_results = [results[0]]
        
        for result in results[1:]:
            existing_contents = [r['content'] for r in unique_results]
            all_texts = existing_contents + [result['content']]
            vectors = encoder.encode(all_texts)
            
            # 新结果与所有已选结果的最大相似度
            new_vec = vectors[-1]
            max_sim = max(vectors[i] @ new_vec 
                         for i in range(len(unique_results)))
            
            if max_sim < self.threshold:
                unique_results.append(result)
        
        return unique_results


class MMRSelector:
    """Maximal Marginal Relevance 多样性选择
    
    核心公式：MMR = λ · Sim(q, d) - (1-λ) · max(Sim(d, d_selected))
    
    - λ=1.0：只考虑相关性（退化为普通排序）
    - λ=0.0：只考虑多样性（结果可能完全不相关）
    - λ=0.7：推荐值，优先保证相关性，兼顾多样性
    """
    
    def __init__(self, encoder, lambda_param: float = 0.7):
        self.encoder = encoder
        self.lambda_param = lambda_param
    
    def select(self, query: str, candidates: List[Dict],
               top_k: int = 5) -> List[Dict]:
        """MMR 贪心选择"""
        if not candidates:
            return []
        
        query_vec = self.encoder.encode([query])[0]
        cand_vecs = self.encoder.encode([c['content'] for c in candidates])
        
        selected_indices = []
        remaining_indices = list(range(len(candidates)))
        
        for _ in range(min(top_k, len(candidates))):
            best_idx = None
            best_mmr = float('-inf')
            
            for idx in remaining_indices:
                # 相关性分数：查询与候选的相似度
                relevance = float(cand_vecs[idx] @ query_vec)
                
                # 多样性惩罚：与已选结果的最大相似度
                if selected_indices:
                    max_sim = max(
                        float(cand_vecs[idx] @ cand_vecs[si])
                        for si in selected_indices
                    )
                    diversity = -max_sim
                else:
                    diversity = 0
                
                mmr = (self.lambda_param * relevance + 
                       (1 - self.lambda_param) * diversity)
                
                if mmr > best_mmr:
                    best_mmr = mmr
                    best_idx = idx
            
            selected_indices.append(best_idx)
            remaining_indices.remove(best_idx)
        
        return [candidates[i] for i in selected_indices]
​

F --> Dedup[语义去重\n阈值 0.92]
P --> Dedup
C --> Dedup
E --> Dedup

Dedup --> MMR[MMR 多样性选择\nλ=0.7]
MMR --> Rerank[Cross-Encoder 重排序]
Rerank --> LLM[LLM 生成最终回答]
​

</div>

**MMR 的 lambda 参数怎么调？** 这是实践中最常被问到的问题。我的建议是：先从 0.7 开始（相关性优先），观察检索结果的多样性是否足够。如果发现返回的文档都在讲同一件事（比如都是介绍 RAG 的定义，没有具体实现方案），把 lambda 降到 0.5 试试。反过来说，如果结果太分散（有些文档跟问题明显不相关），把 lambda 升到 0.8。在大多数知识库问答场景中，0.6-0.7 是一个安全区间。

## 完整示例

把上面所有模块串起来，构成一个生产级的两阶段检索流水线：

```python
"""完整的两阶段检索流水线——可用于生产环境

流水线：查询 -> 意图路由 -> 向量粗排 -> 去重 -> MMR -> 重排序 -> 输出
"""

def full_retrieval_pipeline(query: str, documents: list, 
                            encoder, reranker):
    """端到端的检索流水线"""
    # 1. 意图识别与策略选择
    router = QueryRouter()
    intent = router.classify_intent(query)
    config = router.get_retrieval_config(intent)
    
    # 2. 向量粗排——从全库中快速召回
    query_vec = encoder.encode([query])[0]
    doc_vecs = encoder.encode([d['content'] for d in documents])
    similarities = doc_vecs @ query_vec
    
    top_k = config['top_k']
    top_indices = similarities.argsort()[-top_k:][::-1]
    coarse_results = [
        {**documents[i], 'score': float(similarities[i])}
        for i in top_indices
    ]
    
    # 3. 语义去重
    deduplicator = ResultDeduplicator(similarity_threshold=0.92)
    deduped = deduplicator.deduplicate(coarse_results, encoder)
    
    # 4. MMR 多样性选择
    mmr = MMRSelector(encoder, lambda_param=0.7)
    diverse = mmr.select(query, deduped, 
                        top_k=config['rerank_top_k'])
    
    # 5. Cross-Encoder 重排序——最终精选
    final = reranker.rerank(query, diverse, top_k=3)
    
    return {
        'query': query,
        'intent': intent.value,
        'config': config['description'],
        'pipeline': (
            f"全库 -> 粗排{len(coarse_results)}篇 -> "
            f"去重{len(deduped)}篇 -> "
            f"MMR{len(diverse)}篇 -> 重排{len(final)}篇"
        ),
        'results': final
    }

# 运行示例
result = full_retrieval_pipeline(
    query="如何减少RAG系统的幻觉问题？",
    documents=[...],  # 你的文档库
    encoder=encoder,
    reranker=reranker
)
print(result['pipeline'])
# 输出：全库 -> 粗排10篇 -> 去重8篇 -> MMR5篇 -> 重排3篇
​