2.4 检索效果优化（上）

本节导读：本节将深入探讨RAG系统检索效果优化的核心技术，包括查询理解、相似度计算和多阶段排序策略，帮助你显著提升检索的准确性和召回率。

学习目标

• 理解影响检索效果的关键因素和优化方向
• 掌握查询理解与扩展的核心技术
• 学会多种相似度计算方法和混合策略
• 了解多阶段排序的优化方法
• 构建完整的检索优化框架

核心概念

检索效果优化是RAG系统性能提升的核心，它涉及查询处理、文档匹配、结果排序等多个环节的综合优化。高质量的检索是RAG系统准确回答用户问题的基础保障。

优化的多维视角

环境准备 / 前置知识

• Python 3.8+ 环境
• 向量数据库基础（FAISS、Milvus等）
• 相似度度量知识（余弦相似度、欧氏距离等）
• 基本的机器学习模型理解
• 熟悉RAG系统基本架构

分步实战

步骤 1：查询理解与预处理

查询规范化处理

import re
import jieba
from nltk.corpus import stopwords
from nltk.stem import WordNetLemmatizer

class QueryProcessor:
    """查询处理器"""
    
    def __init__(self):
        self.stop_words = set(stopwords.words('english'))
        self.lemmatizer = WordNetLemmatizer()
    
    def normalize_query(self, query):
        """查询规范化处理"""
        # 转换为小写
        query = query.lower()
        
        # 移除特殊字符，保留字母数字和中文
        query = re.sub(r'[^a-zA-Z0-9\u4e00-\u9fff\s]', '', query)
        
        # 移除多余空格
        query = ' '.join(query.split())
        
        return query
    
    def extract_keywords(self, query):
        """提取关键词"""
        normalized = self.normalize_query(query)
        
        # 英文关键词提取
        words = normalized.split()
        keywords = [word for word in words if word not in self.stop_words]
        
        # 中文分词
        chinese_words = jieba.lcut(normalized)
        keywords.extend([word for word in chinese_words if len(word) > 1])
        
        # 词形还原
        keywords = [self.lemmatizer.lemmatize(word) for word in keywords]
        
        return list(set(keywords))
    
    def expand_query(self, query):
        """查询扩展"""
        keywords = self.extract_keywords(query)
        
        # 基于同义词扩展
        expanded_terms = self._get_synonyms(keywords)
        
        # 基于上下文扩展
        contextual_terms = self._get_contextual_terms(keywords)
        
        # 合并扩展后的查询
        expanded_query = ' '.join(keywords + expanded_terms + contextual_terms)
        
        return expanded_query
    
    def _get_synonyms(self, keywords):
        """获取同义词"""
        # 这里可以实现同义词库的查询
        # 示例实现
        synonyms = {
            "优化": ["改进", "提升", "增强"],
            "检索": ["搜索", "查找", "查询"],
            "系统": ["平台", "框架", "架构"]
        }
        
        expanded = []
        for keyword in keywords:
            if keyword in synonyms:
                expanded.extend(synonyms[keyword])
        
        return list(set(expanded))
    
    def _get_contextual_terms(self, keywords):
        """获取上下文相关术语"""
        # 基于预训练模型获取上下文相关术语
        # 这里可以集成BERT等模型的上下文扩展能力
        contextual_terms = []
        
        for keyword in keywords:
            # 示例：基于word2vec的上下文词
            # 实际实现中应该使用更复杂的模型
            if keyword == "RAG":
                contextual_terms.extend(["检索", "增强", "生成", "向量"])
            elif keyword == "优化":
                contextual_terms.extend(["性能", "效率", "质量"])
        
        return list(set(contextual_terms))
​

查询意图识别

from transformers import pipeline

class IntentClassifier:
    """查询意图分类器"""
    
    def __init__(self):
        self.classifier = pipeline("zero-shot-classification", 
                                  model="facebook/bart-large-mnli")
    
    def classify_intent(self, query):
        """查询意图分类"""
        candidate_labels = [
            "事实查询", "解释说明", "操作指导", "比较分析",
            "技术问题", "学习指导", "创意写作", "代码帮助"
        ]
        
        result = self.classifier(query, candidate_labels)
        
        return {
            'primary_intent': result['labels'][0],
            'confidence': result['scores'][0],
            'all_scores': dict(zip(result['labels'], result['scores']))
        }
    
    def adjust_search_strategy(self, intent):
        """根据意图调整搜索策略"""
        strategies = {
            "事实查询": {"exact_match": True, "similarity_threshold": 0.8, "max_results": 5},
            "解释说明": {"exact_match": False, "similarity_threshold": 0.7, "max_results": 10},
            "操作指导": {"exact_match": True, "similarity_threshold": 0.85, "max_results": 8},
            "比较分析": {"exact_match": False, "similarity_threshold": 0.75, "max_results": 15},
            "技术问题": {"exact_match": True, "similarity_threshold": 0.9, "max_results": 6},
            "学习指导": {"exact_match": False, "similarity_threshold": 0.7, "max_results": 12},
            "创意写作": {"exact_match": False, "similarity_threshold": 0.6, "max_results": 20},
            "代码帮助": {"exact_match": True, "similarity_threshold": 0.95, "max_results": 5}
        }
        
        return strategies.get(intent, strategies["事实查询"])
​

步骤 2：相似度计算的优化

多种相似度度量

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from scipy.spatial.distance import euclidean

class SimilarityCalculator:
    """相似度计算器"""
    
    def __init__(self):
        self.similarity_methods = {
            'cosine': self._cosine_similarity,
            'euclidean': self._euclidean_distance,
            'manhattan': self._manhattan_distance,
            'dot_product': self._dot_product
        }
    
    def calculate_similarity(self, query_vector, doc_vectors, method='cosine'):
        """计算相似度"""
        if method not in self.similarity_methods:
            raise ValueError(f"不支持的相似度方法: {method}")
        
        if method == 'cosine':
            # 余弦相似度
            similarities = []
            for doc_vector in doc_vectors:
                sim = cosine_similarity([query_vector], [doc_vector])[0][0]
                similarities.append(sim)
        elif method == 'euclidean':
            # 欧氏距离（转换为相似度）
            distances = [euclidean(query_vector, doc_vector) for doc_vector in doc_vectors]
            max_dist = max(distances) if distances else 1.0
            similarities = [1 - (dist / max_dist) for dist in distances]
        elif method == 'manhattan':
            # 曼哈顿距离（转换为相似度）
            distances = [sum(abs(q - d) for q, d in zip(query_vector, doc_vector)) 
                        for doc_vector in doc_vectors]
            max_dist = max(distances) if distances else 1.0
            similarities = [1 - (dist / max_dist) for dist in distances]
        elif method == 'dot_product':
            # 点积相似度
            similarities = [np.dot(query_vector, doc_vector) for doc_vector in doc_vectors]
        
        return np.array(similarities)
    
    def _cosine_similarity(self, vec1, vec2):
        """余弦相似度计算"""
        return cosine_similarity([vec1], [vec2])[0][0]
    
    def _euclidean_distance(self, vec1, vec2):
        """欧氏距离计算"""
        return euclidean(vec1, vec2)
    
    def _manhattan_distance(self, vec1, vec2):
        """曼哈顿距离计算"""
        return sum(abs(q - d) for q, d in zip(vec1, vec2))
    
    def _dot_product(self, vec1, vec2):
        """点积计算"""
        return np.dot(vec1, vec2)
​

混合相似度策略

class HybridSimilarityCalculator(SimilarityCalculator):
    """混合相似度计算器"""
    
    def __init__(self):
        super().__init__()
        self.weights = {
            'semantic': 0.6,  # 语义相似度权重
            'lexical': 0.3,  # 词法相似度权重
            'structural': 0.1  # 结构相似度权重
        }
    
    def calculate_hybrid_similarity(self, query, doc):
        """计算混合相似度"""
        # 语义相似度
        semantic_sim = self._calculate_semantic_similarity(query, doc)
        
        # 词法相似度
        lexical_sim = self._calculate_lexical_similarity(query, doc)
        
        # 结构相似度
        structural_sim = self._calculate_structural_similarity(query, doc)
        
        # 加权混合
        hybrid_sim = (
            self.weights['semantic'] * semantic_sim +
            self.weights['lexical'] * lexical_sim +
            self.weights['structural'] * structural_sim
        )
        
        return hybrid_sim
    
    def _calculate_semantic_similarity(self, query, doc):
        """计算语义相似度"""
        # 使用预训练模型计算语义相似度
        # 这里简化实现，实际应该使用BERT等模型
        query_words = set(query.lower().split())
        doc_words = set(doc.lower().split())
        
        intersection = query_words & doc_words
        union = query_words | doc_words
        
        return len(intersection) / len(union) if union else 0.0
    
    def _calculate_lexical_similarity(self, query, doc):
        """计算词法相似度"""
        # 使用TF-IDF或词袋模型计算词法相似度
        from sklearn.feature_extraction.text import TfidfVectorizer
        from sklearn.metrics.pairwise import cosine_similarity
        
        vectorizer = TfidfVectorizer()
        tfidf_matrix = vectorizer.fit_transform([query, doc])
        
        return cosine_similarity(tfidf_matrix[0:1], tfidf_matrix[1:2])[0][0]
    
    def _calculate_structural_similarity(self, query, doc):
        """计算结构相似度"""
        # 考虑文档结构（标题、段落、列表等）
        query_structure = self._analyze_structure(query)
        doc_structure = self._analyze_structure(doc)
        
        return self._compare_structures(query_structure, doc_structure)
    
    def _analyze_structure(self, text):
        """分析文档结构"""
        # 统计不同结构元素的数量
        structure = {
            'headings': len(re.findall(r'^#{1,6}\s', text, re.MULTILINE)),
            'paragraphs': len(text.split('\n\n')),
            'lists': len(re.findall(r'^\*\s|^\d+\.\s', text, re.MULTILINE)),
            'code_blocks': len(re.findall(r'```', text))
        }
        return structure
    
    def _compare_structures(self, struct1, struct2):
        """比较文档结构相似度"""
        total_elements = sum(struct1.values()) + sum(struct2.values())
        if total_elements == 0:
            return 0.0
        
        similarity = 0.0
        for key in struct1:
            if key in struct2:
                similarity += min(struct1[key], struct2[key])
        
        return similarity / total_elements
​

步骤 3：相关性排序优化

多阶段排序策略

class MultiStageRanker:
    """多阶段排序器"""
    
    def __init__(self):
        self.ranking_stages = [
            'coarse_ranking',    # 粗排阶段
            'fine_ranking',      # 精排阶段
            'reranking'         # 重排序阶段
        ]
    
    def rank_documents(self, query, candidates, rankings_config):
        """多阶段文档排序"""
        ranked_docs = candidates
        
        for stage in self.ranking_stages:
            if stage in rankings_config:
                ranked_docs = self._apply_ranking_stage(
                    query, ranked_docs, rankings_config[stage]
                )
        
        return ranked_docs
    
    def _apply_ranking_stage(self, query, docs, config):
        """应用单阶段排序"""
        if config['method'] == 'vector_similarity':
            scores = self._vector_similarity_ranking(query, docs, config)
        elif config['method'] == 'bm25':
            scores = self._bm25_ranking(query, docs, config)
        elif config['method'] == 'learning_to_rank':
            scores = self._learning_to_rank_ranking(query, docs, config)
        elif config['method'] == 'cross_encoder':
            scores = self._cross_encoder_ranking(query, docs, config)
        else:
            scores = self._default_ranking(query, docs, config)
        
        # 根据分数重新排序
        doc_scores = list(zip(docs, scores))
        doc_scores.sort(key=lambda x: x[1], reverse=True)
        
        return [doc for doc, score in doc_scores[:config['max_results']]]
    
    def _vector_similarity_ranking(self, query, docs, config):
        """向量相似度排序"""
        query_vector = self._embed_query(query)
        doc_vectors = [self._embed_doc(doc) for doc in docs]
        
        calculator = SimilarityCalculator()
        similarities = calculator.calculate_similarity(
            query_vector, doc_vectors, config['similarity_method']
        )
        
        return similarities.tolist()
    
    def _bm25_ranking(self, query, docs, config):
        """BM25排序"""
        from rank_bm25 import BM25Okapi
        
        # 分词
        tokenized_corpus = [doc.lower().split() for doc in docs]
        bm25 = BM25Okapi(tokenized_corpus)
        
        # BM25得分
        tokenized_query = query.lower().split()
        scores = bm25.get_scores(tokenized_query)
        
        return scores
    
    def _learning_to_rank_ranking(self, query, docs, config):
        """学习排序"""
        # 使用预训练的学习到排序模型
        # 这里简化实现
        scores = [np.random.random() for _ in docs]  # 实际应该使用模型
        return scores
    
    def _cross_encoder_ranking(self, query, docs, config):
        """交叉编码器排序"""
        from transformers import AutoTokenizer, AutoModelForSequenceClassification
        
        model_name = config.get('model', 'cross-encoder/msmarco-MiniLM-L6-en-de-v1')
        tokenizer = AutoTokenizer.from_pretrained(model_name)
        model = AutoModelForSequenceClassification.from_pretrained(model_name)
        
        scores = []
        for doc in docs:
            # 准备输入
            inputs = tokenizer(query, doc, return_tensors="pt", truncation=True, max_length=512)
            
            # 计算分数
            with torch.no_grad():
                logits = model(**inputs).logits
            score = logits[0][1].item()  # 正类别的概率
            scores.append(score)
        
        return scores
    
    def _default_ranking(self, query, docs, config):
        """默认排序策略"""
        # 简单的词频排序
        query_words = set(query.lower().split())
        scores = []
        
        for doc in docs:
            doc_words = set(doc.lower().split())
            overlap = len(query_words & doc_words)
            scores.append(overlap)
        
        return scores
    
    def _embed_query(self, query):
        """查询向量化"""
        # 使用预训练模型进行向量化
        # 这里简化实现
        return np.random.rand(768)  # 768维向量
    
    def _embed_doc(self, doc):
        """文档向量化"""
        # 使用预训练模型进行向量化
        # 这里简化实现
        return np.random.rand(768)  # 768维向量
​

完整示例：检索优化管道（基础部分）

class BasicRetrievalOptimization:
    """基础检索优化管道"""
    
    def __init__(self):
        self.query_processor = QueryProcessor()
        self.similarity_calculator = HybridSimilarityCalculator()
        self.ranker = MultiStageRanker()
    
    def optimize_retrieval(self, query, candidate_docs, config):
        """基础检索优化流程"""
        print(f"原始查询: {query}")
        
        # 步骤1：查询理解
        processed_query = self.query_processor.normalize_query(query)
        expanded_query = self.query_processor.expand_query(query)
        intent = self.query_processor.classify_intent(query)
        
        print(f"处理后查询: {processed_query}")
        print(f"扩展查询: {expanded_query}")
        print(f"查询意图: {intent['primary_intent']}")
        
        # 步骤2：查询和文档向量化
        query_vector = self.similarity_calculator._embed_query(expanded_query)
        doc_vectors = [self.similarity_calculator._embed_doc(doc) for doc in candidate_docs]
        
        # 步骤3：相似度计算
        similarities = self.similarity_calculator.calculate_similarity(
            query_vector, doc_vectors, 'cosine'
        )
        
        # 步骤4：多阶段排序
        ranking_config = {
            'coarse_ranking': {
                'method': 'vector_similarity',
                'similarity_method': 'cosine',
                'max_results': min(len(candidate_docs), 20)
            },
            'fine_ranking': {
                'method': 'bm25',
                'max_results': 10
            }
        }
        
        scored_docs = []
        for i, (doc, sim) in enumerate(zip(candidate_docs, similarities)):
            scored_docs.append({
                'content': doc,
                'similarity': sim,
                'original_index': i
            })
        
        ranked_docs = self.ranker.rank_documents(expanded_query, scored_docs, ranking_config)
        
        print(f"原始候选文档: {len(candidate_docs)}")
        print(f"优化后结果: {len(ranked_docs)}")
        
        return ranked_docs
    
    def get_optimization_summary(self, original_results, optimized_results):
        """生成优化摘要"""
        improvement = len(optimized_results) / len(original_results) if original_results else 1.0
        
        return {
            'original_count': len(original_results),
            'optimized_count': len(optimized_results),
            'improvement_ratio': improvement,
            'optimization_stages': ['query_processing', 'similarity_calculation', 'multi_stage_ranking']
        }
​

常见问题 FAQ

Q1：如何确定最优的相似度计算方法？

A：选择相似度计算方法需要综合考虑以下因素：

• 数据类型：文本数据适合余弦相似度，向量数据可选欧氏距离
• 语义需求：需要语义理解时，使用预训练模型的语义相似度
• 性能考虑：BM25在传统文本检索中表现优秀且计算效率高
• 混合策略：在生产环境中通常采用多方法混合的相似度计算
• 场景适配：事实查询适合精确匹配，创意内容适合语义相似度

Q2：如何避免过度优化导致的检索性能下降？

A：避免过度优化的方法包括：

1. 基准测试：在优化前后进行对比测试，确保性能确实提升
2. 渐进优化：逐步引入优化技术，每次只改变一个变量
3. 监控指标：同时监控检索质量、响应时间和资源消耗
4. 回滚机制：建立优化回滚机制，发现问题能快速恢复
5. 用户反馈：收集实际用户的使用反馈，评估优化效果

Q3：混合相似度策略的优势是什么？

A：混合相似度策略的主要优势：

• 全面性：综合考虑语义、词法和结构多个维度
• 灵活性：可以根据不同场景调整权重配置
• 鲁棒性：单一维度失效时其他维度仍能发挥作用
• 准确性：多维度评估更接近人类的判断标准
• 适应性：可以针对不同类型的数据和查询进行定制

最佳实践与避坑

检索优化的最佳实践

• 数据质量优先：检索优化的基础是高质量的数据和文档分块
• 分阶段实施：按照粗排→精排的顺序逐步优化
• 持续监控：建立检索质量监控体系
• 性能权衡：在检索质量和响应时间之间找到平衡点
• 版本控制：对检索配置进行版本管理

常见陷阱及避免方法

• 过度依赖单一指标：不要只关注准确率或召回率
• 忽视查询理解：直接进行相似度计算而忽略预处理
• 缺乏多样性：优化后的结果可能过于集中
• 性能瓶颈：复杂的重排序模型可能导致响应时间过长

本节小结

本节系统介绍了RAG系统检索效果优化的核心技术，包括查询理解、相似度计算和多阶段排序。通过这些技术的组合应用，可以显著提升检索的准确性和用户体验。

关键要点回顾：

• 查询理解是检索优化的基础，直接影响后续所有步骤
• 混合相似度计算比单一方法更能适应复杂的检索场景
• 多阶段排序策略能在保证质量的同时兼顾性能
• 持续监控和评估是优化效果的重要保障

下一节将继续介绍检索优化的高级技术，包括结果过滤、去重策略和完整的优化管道构建。

延伸阅读

• 官方文档：LangChain检索优化器文档（文字描述，不带链接）
• 相关章节：本教程2.5节结果过滤与优化（文字描述，不带链接）

关键词：RAG高级优化, 检索效果, 查询理解, 相似度计算, 排序优化
难度：进阶
预计阅读：20分钟