3.3 检索策略优化

3.3 检索策略优化 — 大模型应用开发从零到一 关键词短语

本节导读：深入掌握检索增强生成（RAG）的核心检索策略，优化检索精度和召回率，构建高效的问答系统和知识检索框架。

学习目标

• 理解检索策略的基本概念和重要性
• 掌握多种检索算法的优缺点和适用场景
• 学会优化检索精度的关键技术
• 了解检索召回率的提升策略
• 构建多级检索和混合检索系统
• 实现智能问答和知识检索应用

核心概念

什么是检索策略

检索策略是指在海量数据中快速找到最相关信息的算法和方法。在RAG系统中，检索策略直接影响检索结果的准确性和系统的整体性能。

检索策略的关键指标：

• 精确率（Precision）：检索结果中相关文档的比例
• 召回率（Recall）：所有相关文档中被检索到的比例
• F1分数：精确率和召回率的调和平均值
• 响应速度：检索操作的执行时间

检索策略的重要性

在RAG系统中，检索策略的重要性体现在：

1. 影响答案质量：高质量的检索结果能提供更准确和相关的上下文
2. 控制成本：优化检索策略可以减少不必要的API调用
3. 提升用户体验：快速的检索响应能提供更好的交互体验
4. 支持复杂查询：高级检索策略能处理复杂的查询需求

环境准备 / 前置知识

基础配置

# 检索策略优化工具安装
pip install rank-bm25 faiss-cpu sentence-transformers scikit-learn numpy pandas

# 环境检查
import rank_bm25
import faiss
import sentence_transformers
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from sklearn.feature_extraction.text import TfidfVectorizer

print("检索策略优化工具配置完成")
​

依赖管理

# requirements.txt
rank-bm25>=0.2.2
faiss-cpu>=1.7.0
sentence-transformers>=2.2.0
scikit-learn>=1.0.0
numpy>=1.21.0
pandas>=1.3.0
​

分步实战

步骤 1：关键词检索策略优化

关键词检索是基于词汇匹配的传统检索方法，虽然简单但仍然是很多系统的基础。

BM25算法实现

import rank_bm25
import numpy as np
from typing import List, Dict, Any
import re

class BM25Retriever:
    """BM25检索器实现"""
    
    def __init__(self, k1: float = 1.2, b: float = 0.75):
        """
        初始化BM25检索器
        
        Args:
            k1: 控制词频饱和度的参数
            b: 控制文档长度归一化的参数
        """
        self.k1 = k1
        self.b = b
        self.corpus = []
        self.tokenized_corpus = []
        self.idf = {}
        self.avgdl = 0
    
    def fit(self, documents: List[str]):
        """训练BM25模型"""
        self.corpus = documents
        self.tokenized_corpus = [self._tokenize(doc) for doc in documents]
        
        # 计算IDF
        df = {}
        for doc in self.tokenized_corpus:
            for term in set(doc):
                df[term] = df.get(term, 0) + 1
        
        # 计算平均文档长度
        self.avgdl = np.mean([len(doc) for doc in self.tokenized_corpus])
        
        # 计算IDF值
        self.idf = {}
        for term, freq in df.items():
            self.idf[term] = np.log((len(documents) - freq + 0.5) / (freq + 0.5))
    
    def _tokenize(self, text: str) -> List[str]:
        """文本分词"""
        text = text.lower()
        tokens = re.findall(r'\b\w+\b', text)
        return tokens
    
    def score(self, query: str, doc_id: int) -> float:
        """计算查询与文档的相似度分数"""
        query_tokens = self._tokenize(query)
        doc_tokens = self.tokenized_corpus[doc_id]
        doc_len = len(doc_tokens)
        
        score = 0
        for token in query_tokens:
            if token in self.idf:
                tf = doc_tokens.count(token)
                score += self.idf[token] * (tf * (self.k1 + 1)) / (tf + self.k1 * (1 - self.b + self.b * doc_len / self.avgdl))
        
        return score
    
    def search(self, query: str, top_k: int = 5) -> List[Dict]:
        """搜索最相关的文档"""
        scores = []
        for i, doc in enumerate(self.corpus):
            score = self.score(query, i)
            scores.append((i, score))
        
        scores.sort(key=lambda x: x[1], reverse=True)
        
        results = []
        for idx, score in scores[:top_k]:
            results.append({
                'id': idx,
                'score': score,
                'content': self.corpus[idx],
                'snippet': self._get_snippet(self.corpus[idx], query)
            })
        
        return results

# 使用示例
def bm25_example():
    """BM25检索示例"""
    documents = [
        "机器学习是人工智能的一个分支，它使计算机能够从数据中学习并做出预测。",
        "深度学习是机器学习的一个子集，使用神经网络来模拟人脑的学习过程。",
        "数据挖掘是从大量数据中提取有价值信息的技术，广泛应用于商业智能。",
        "自然语言处理是人工智能领域的重要分支，专注于处理和理解人类语言。",
        "计算机视觉是使计算机能够理解和解释视觉信息的技术，包括图像和视频处理。"
    ]
    
    retriever = BM25Retriever(k1=1.2, b=0.75)
    retriever.fit(documents)
    
    query = "机器学习"
    results = retriever.search(query, top_k=3)
    
    print(f"BM25检索结果（查询: '{query}'）:")
    for i, result in enumerate(results):
        print(f"{i+1}. 分数: {result['score']:.4f}")
        print(f"   内容: {result['snippet']}")

if __name__ == "__main__":
    bm25_example()
​

步骤 2：向量检索策略优化

向量检索是基于语义相似性的检索方法，能够理解查询的语义含义。

嵌入模型优化

import numpy as np
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
from typing import List, Dict, Any

class VectorRetriever:
    """向量检索器实现"""
    
    def __init__(self, model_name: str = 'all-MiniLM-L6-v2'):
        """初始化向量检索器"""
        self.model = SentenceTransformer(model_name)
        self.corpus = []
        self.doc_embeddings = None
    
    def fit(self, documents: List[str]):
        """训练向量模型"""
        self.corpus = documents
        print("正在计算文档嵌入...")
        self.doc_embeddings = self.model.encode(documents, convert_to_tensor=True)
        print(f"已生成 {len(documents)} 个文档嵌入")
    
    def search(self, query: str, top_k: int = 5, threshold: float = 0.3) -> List[Dict]:
        """搜索最相关的文档"""
        # 编码查询
        query_embedding = self.model.encode([query], convert_to_tensor=True)
        
        # 计算相似度
        cosine_scores = cosine_similarity(query_embedding, self.doc_embeddings)[0]
        
        # 获取top_k结果
        top_indices = np.argsort(cosine_scores)[::-1][:top_k]
        
        results = []
        for idx in top_indices:
            if cosine_scores[idx] >= threshold:
                results.append({
                    'id': idx,
                    'score': cosine_scores[idx],
                    'content': self.corpus[idx],
                    'snippet': self._get_snippet(self.corpus[idx], query)
                })
        
        return results

# 使用示例
def vector_example():
    """向量检索示例"""
    documents = [
        "机器学习是人工智能的一个分支，它使计算机能够从数据中学习并做出预测。",
        "深度学习是机器学习的一个子集，使用神经网络来模拟人脑的学习过程。",
        "数据挖掘是从大量数据中提取有价值信息的技术，广泛应用于商业智能。",
        "自然语言处理是人工智能领域的重要分支，专注于处理和理解人类语言。",
        "计算机视觉是使计算机能够理解和解释视觉信息的技术，包括图像和视频处理。"
    ]
    
    retriever = VectorRetriever(model_name='all-MiniLM-L6-v2')
    retriever.fit(documents)
    
    query = "机器学习"
    results = retriever.search(query, top_k=3)
    
    print(f"向量检索结果（查询: '{query}'）:")
    for i, result in enumerate(results):
        print(f"{i+1}. 分数: {result['score']:.4f}")
        print(f"   内容: {result['snippet']}")

if __name__ == "__main__":
    vector_example()
​

步骤 3：混合检索策略

混合检索策略结合关键词和向量检索的优势，获得更好的检索效果。

from typing import List, Dict, Any
import numpy as np

class HybridRetriever:
    """混合检索器实现"""
    
    def __init__(self, model_name: str = 'all-MiniLM-L6-v2'):
        """初始化混合检索器"""
        self.keyword_retriever = BM25Retriever()
        self.vector_retriever = VectorRetriever(model_name)
        self.corpus = []
    
    def fit(self, documents: List[str]):
        """训练混合检索器"""
        self.corpus = documents
        self.keyword_retriever.fit(documents)
        self.vector_retriever.fit(documents)
    
    def search(self, query: str, top_k: int = 5, keyword_weight: float = 0.3, 
              vector_weight: float = 0.7, threshold: float = 0.2) -> List[Dict]:
        """混合搜索"""
        # 关键词检索
        keyword_results = self.keyword_retriever.search(query, top_k * 2)
        
        # 向量检索
        vector_results = self.vector_retriever.search(query, top_k * 2, threshold)
        
        # 合并结果
        combined_scores = {}
        
        # 处理关键词检索结果
        for result in keyword_results:
            doc_id = result['id']
            score = result['score'] * keyword_weight
            if doc_id not in combined_scores:
                combined_scores[doc_id] = {
                    'id': doc_id,
                    'keyword_score': score,
                    'vector_score': 0,
                    'content': result['content'],
                    'snippet': result['snippet']
                }
            else:
                combined_scores[doc_id]['keyword_score'] = score
        
        # 处理向量检索结果
        for result in vector_results:
            doc_id = result['id']
            score = result['score'] * vector_weight
            if doc_id not in combined_scores:
                combined_scores[doc_id] = {
                    'id': doc_id,
                    'keyword_score': 0,
                    'vector_score': score,
                    'content': result['content'],
                    'snippet': result['snippet']
                }
            else:
                combined_scores[doc_id]['vector_score'] = score
        
        # 计算综合分数
        for doc_id, data in combined_scores.items():
            data['score'] = data['keyword_score'] + data['vector_score']
        
        # 排序并返回top_k结果
        sorted_results = sorted(combined_scores.values(), key=lambda x: x['score'], reverse=True)
        
        return sorted_results[:top_k]

# 使用示例
def hybrid_example():
    """混合检索示例"""
    documents = [
        "机器学习是人工智能的一个分支，它使计算机能够从数据中学习并做出预测。",
        "深度学习是机器学习的一个子集，使用神经网络来模拟人脑的学习过程。",
        "数据挖掘是从大量数据中提取有价值信息的技术，广泛应用于商业智能。",
        "自然语言处理是人工智能领域的重要分支，专注于处理和理解人类语言。",
        "计算机视觉是使计算机能够理解和解释视觉信息的技术，包括图像和视频处理。",
        "强化学习是机器学习的重要分支，通过奖惩机制让智能体学习最优策略。",
        "监督学习使用标记数据训练模型，是机器学习中最常用的方法之一。",
        "无监督学习从不标记数据中发现隐藏模式和结构，如聚类和降维。",
        "迁移学习将已训练的知识应用到新的相关任务中，减少训练时间和数据需求。"
    ]
    
    retriever = HybridRetriever()
    retriever.fit(documents)
    
    query = "学习算法"
    results = retriever.search(query, top_k=5, keyword_weight=0.3, vector_weight=0.7)
    
    print(f"混合检索结果（查询: '{query}'）:")
    for i, result in enumerate(results):
        print(f"{i+1}. 综合分数: {result['score']:.4f}")
        print(f"   关键词分数: {result['keyword_score']:.4f}")
        print(f"   向量分数: {result['vector_score']:.4f}")
        print(f"   内容: {result['snippet']}")

if __name__ == "__main__":
    hybrid_example()
​

完整项目示例：智能问答系统

from typing import List, Dict, Any
import numpy as np

class IntelligentQASystem:
    """智能问答系统"""
    
    def __init__(self, knowledge_base: List[str]):
        """初始化智能问答系统"""
        self.knowledge_base = knowledge_base
        self.retriever = HybridRetriever()
        self.retriever.fit(knowledge_base)
    
    def answer_question(self, question: str, top_k: int = 3) -> Dict:
        """回答问题"""
        # 检索相关文档
        retrieved_docs = self.retriever.search(question, top_k=top_k)
        
        # 生成回答
        answer = self._generate_answer(question, retrieved_docs)
        
        return {
            'question': question,
            'answer': answer,
            'retrieved_docs': retrieved_docs,
            'confidence': self._calculate_confidence(retrieved_docs)
        }
    
    def _generate_answer(self, question: str, docs: List[Dict]) -> str:
        """生成回答"""
        if not docs:
            return "抱歉，我没有找到相关的信息。"
        
        # 提取相关信息
        relevant_info = []
        for doc in docs:
            relevant_info.append(doc['snippet'])
        
        # 模拟回答生成
        return "根据您的查询，我找到了相关信息：" + " ".join(relevant_info)
    
    def _calculate_confidence(self, docs: List[Dict]) -> float:
        """计算回答置信度"""
        if not docs:
            return 0.0
        
        # 基于检索分数计算置信度
        scores = [doc['score'] for doc in docs]
        avg_score = np.mean(scores)
        confidence = min(max(avg_score / 2.0, 0.0), 1.0)
        return confidence

# 使用示例
def qa_system_example():
    """智能问答系统示例"""
    # 创建知识库
    knowledge_base = [
        "机器学习是人工智能的一个分支，它使计算机能够从数据中学习并做出预测。",
        "深度学习是机器学习的一个子集，使用神经网络来模拟人脑的学习过程。",
        "数据挖掘是从大量数据中提取有价值信息的技术，广泛应用于商业智能。",
        "自然语言处理是人工智能领域的重要分支，专注于处理和理解人类语言。",
        "计算机视觉是使计算机能够理解和解释视觉信息的技术，包括图像和视频处理。"
    ]
    
    # 初始化问答系统
    qa_system = IntelligentQASystem(knowledge_base)
    
    # 测试问题
    question = "什么是机器学习？"
    result = qa_system.answer_question(question)
    
    print(f"问题: {question}")
    print(f"回答: {result['answer']}")
    print(f"置信度: {result['confidence']:.2f}")
    print(f"检索到的文档:")
    for i, doc in enumerate(result['retrieved_docs']):
        print(f"  {i+1}. {doc['snippet']} (分数: {doc['score']:.4f})")

if __name__ == "__main__":
    qa_system_example()
​

常见问题 FAQ

Q1：如何选择合适的检索策略？

A：选择检索策略时应考虑以下因素：数据规模（大数据量适合向量检索）、查询类型（精确查询适合关键词检索）、性能要求（实时性要求高适合BM25）、精度要求（语义理解要求高适合向量检索）等。混合检索策略通常能获得更好的综合性能。

Q2：如何评估检索策略的效果？

A：评估检索策略的常用指标包括精确率（Precision）、召回率（Recall）、F1分数、平均精度（MAP）、归一化折扣累积增益（nDCG）等。在实际应用中，还需要考虑响应时间、资源消耗等指标。

Q3：如何优化检索的响应速度？

A：优化检索响应速度的方法包括：使用索引加速（如FAISS）、减少检索的文档数量、使用更快的模型、缓存频繁查询的结果、并行处理、使用近似最近邻算法等。

Q4：如何处理多语言检索？

A：多语言检索的处理方法包括：使用多语言嵌入模型、为每种语言建立独立的检索系统、使用翻译将查询转换为目标语言、使用跨语言嵌入模型等。

Q5：如何处理长文档检索？

A：处理长文档检索的方法包括：将长文档分割为较小的段落、使用滑动窗口技术、考虑文档结构（如章节、段落）、使用专门的文档结构化方法等。

最佳实践与避坑

• 选择合适的嵌入模型：根据应用场景选择合适大小的模型（如all-MiniLM-L6-v2适合实时应用，paraphrase-mpnet-base适合高精度要求）
• 优化检索参数：调整top_k值、相似度阈值、权重比例等参数以获得最佳性能
• 缓存频繁查询：对频繁查询的结果进行缓存，提高响应速度
• 定期更新知识库：保持知识库的时效性，定期更新文档和重新训练模型
• 监控检索性能：监控检索的精确率、召回率、响应时间等关键指标
• A/B测试：对不同检索策略进行A/B测试，选择最适合业务需求的方案

本节小结

本节详细介绍了检索策略的核心概念、优化技术和实现方法，包括关键词检索（BM25、TF-IDF）、向量检索（嵌入模型、相似度计算）、混合检索策略等。通过完整的智能问答系统示例，读者可以掌握如何构建高效的检索系统，提升RAG系统的性能和用户体验。

延伸阅读

• 官方文档：Sentence Transformers官方文档
• 相关章节：本教程3.2节向量数据库应用
• 推荐资源：信息检索经典教材、现代搜索引擎技术、深度学习在信息检索中的应用

关键词：大模型应用开发从零到一, 检索策略, BM25, 向量检索, 混合检索, RAG, 相似度计算, 问答系统
难度：进阶
预计阅读：40分钟