3.2 关键词检索
文档摘要
3.2 关键词检索 本节导读：深入理解传统关键词检索技术，掌握Haystack中BM25检索器的使用方法和企业级关键词检索优化策略 学习目标 掌握BM25算法的原理和数学基础 理解关键词检索与向量检索的差异和适用场景 学会使用Haystack的BM25Retriever进行高效关键词检索 掌握文本预处理和模糊搜索技术 构建企业级关键词检索系统 核心概念 BM25算法原理 BM25 (Best Match 25) 是信息检索领域最经典的关键词匹配算法，基于概率框架计算文档与查询的相关性分数。它考虑了词频、文档长度和逆文档频率等多个因素，是现代搜索引擎的核心算法之一。 关键词检索特点 关键词检索基于文本中的字面匹配，通过统计词频和位置信息来计算相关性。
3.2 关键词检索
本节导读：深入理解传统关键词检索技术，掌握Haystack中BM25检索器的使用方法和企业级关键词检索优化策略
学习目标
- 掌握BM25算法的原理和数学基础
- 理解关键词检索与向量检索的差异和适用场景
- 学会使用Haystack的BM25Retriever进行高效关键词检索
- 掌握文本预处理和模糊搜索技术
- 构建企业级关键词检索系统
核心概念
BM25算法原理
BM25 (Best Match 25) 是信息检索领域最经典的关键词匹配算法，基于概率框架计算文档与查询的相关性分数。它考虑了词频、文档长度和逆文档频率等多个因素，是现代搜索引擎的核心算法之一。
关键词检索特点
关键词检索基于文本中的字面匹配，通过统计词频和位置信息来计算相关性。它擅长处理精确查询、特定术语匹配和基于关键词的信息过滤。
文本预处理
文本预处理包括分词、去除停用词、词形还原、大小写转换等步骤，这些步骤对关键词检索的准确性至关重要。
环境准备 / 前置知识
系统要求
- Python 3.8+
- 内存：至少4GB
- 存储：根据文档数量确定
必需依赖
pip install haystack-ai nltk scikit-learn jieba
基础概念
- 倒排索引：从词到文档的映射，是关键词检索的基础
- 词频(TF)：词在文档中出现的频率
- 逆文档频率(IDF)：衡量词在整个文档集合中的重要性
- BM25分数：结合TF和IDF的相关性计算公式
分步实战
步骤 1：基础BM25检索
from haystack.components.retrievers import BM25Retriever from haystack.document_stores import InMemoryDocumentStore # 1. 创建内存文档存储 document_store = InMemoryDocumentStore() # 2. 准备文档数据 documents = [ { "id": "doc1", "content": "机器学习是人工智能的一个分支，专注于让计算机系统从数据中学习模式和规律。监督学习、无监督学习和强化学习是机器学习的三大主要范式。", "meta": {"category": "AI", "language": "zh"} }, { "id": "doc2", "content": "深度学习是机器学习的子集，使用多层神经网络来学习数据的复杂表示。卷积神经网络（CNN）在图像处理中表现出色，循环神经网络（RNN）和Transformer在序列数据处理中效果显著。", "meta": {"category": "AI", "language": "zh"} }, { "id": "doc3", "content": "自然语言处理（NLP）是AI的一个领域，专注于计算机与人类语言之间的交互。现代NLP主要基于Transformer架构，BERT、GPT等模型在理解、生成和翻译任务中取得了突破性进展。", "meta": {"category": "NLP", "language": "zh"} }, { "id": "doc4", "content": "计算机视觉使计算机能够从图像和视频中获取高级理解，类似于人类视觉系统。目标检测、图像分割、图像生成是计算机视觉的核心任务，YOLO、SSD、Mask R-CNN等算法被广泛应用。", "meta": {"category": "CV", "language": "zh"} }, { "id": "doc5", "content": "检索增强生成（RAG）结合了信息检索和文本生成，通过外部知识库增强语言模型的能力。这种方法解决了大语言模型的知识时效性和幻觉问题，提高了回答的准确性和可靠性。", "meta": {"category": "RAG", "language": "zh"} } ] # 3. 写入文档 document_store.write_documents(documents) # 4. 创建BM25检索器 bm25_retriever = BM25Retriever(document_store=document_store, top_k=3) # 5. 执行关键词检索 query = "机器学习" results = bm25_retriever.run(query=query) # 6. 显示结果 print(f"查询: {query}") print(f"检索到 {len(results['documents'])} 个文档:\n") for i, doc in enumerate(results['documents'], 1): print(f"{i}. [分数: {doc.score:.4f}] {doc.content}") print(f" ID: {doc.id}\n")
步骤 2：高级BM25配置
# 1. 配置BM25检索器参数 bm25_retriever = BM25Retriever( document_store=document_store, top_k=5, all_terms_must_match=False, # 是否要求查询词全部匹配 scale_score=True, # 是否缩放分数 bm25_algorithm="okapi", # BM25算法变体 k1=1.2, # 词频饱和参数 b=0.75 # 文档长度归一化参数 ) # 2. 执行检索 query = "深度学习神经网络" results = bm25_retriever.run(query=query) # 3. 分析检索结果 print(f"查询: {query}") print(f"检索结果分析:") print(f"- 结果数量: {len(results['documents'])}") print(f"- 最高分数: {max(doc.score for doc in results['documents']):.4f}") print(f"- 最低分数: {min(doc.score for doc in results['documents']):.4f}") for i, doc in enumerate(results['documents'], 1): print(f"\n{i}. [分数: {doc.score:.4f}] {doc.content[:80]}...")
步骤 3：文本预处理优化
from haystack.components.preprocessors import TextPreprocessor # 1. 配置文本预处理器 preprocessor = TextPreprocessor( split_by="word", # 按单词分割 split_length=None, # 不限制长度 split_overlap=0, # 无重叠 remove_punctuation=True, # 去除标点符号 remove_numbers=False, # 保留数字 remove_stopwords=True, # 去除停用词 stemming=True, # 词形还原 lowercase=True # 转小写 ) # 2. 预处理查询文本 original_query = "深度学习是什么？" processed_query = preprocessor.run(texts=[original_query]) # 3. 显示处理结果 print(f"原始查询: {original_query}") print(f"处理后查询: {processed_query['processed_texts'][0]}") # 4. 使用预处理后的查询进行检索 results = bm25_retriever.run(query=processed_query['processed_texts'][0]) print(f"\n预处理后检索结果:") for i, doc in enumerate(results['documents'], 1): print(f"{i}. [分数: {doc.score:.4f}] {doc.content[:60]}...")
步骤 4：模糊BM25检索
# 1. 创建模糊BM25检索器 fuzzy_bm25 = BM25Retriever( document_store=document_store, top_k=5, fuzzy=True, # 启用模糊匹配 fuzziness=2, # 模糊程度（编辑距离） all_terms_must_match=False ) # 2. 测试模糊搜索 fuzzy_queries = [ "机械学习", # 机器学习的错别字 "深度学", # 部分匹配 "神经网路", # 神经网络的错误 "计算器视觉" # 计算机视觉的错误 ] print("模糊BM25检索测试:") print("-" * 50) for query in fuzzy_queries: results = fuzzy_bm25.run(query=query) print(f"查询: '{query}' -> 找到 {len(results['documents'])} 个结果:") for i, doc in enumerate(results['documents'][:3], 1): # 只显示前3个结果 print(f" {i}. [分数: {doc.score:.4f}] {doc.content[:50]}...") print()
完整示例
企业级关键词检索系统
import time import logging import jieba from typing import List, Dict, Any, Optional from dataclasses import dataclass from haystack.components.retrievers import BM25Retriever from haystack.document_stores import InMemoryDocumentStore from haystack.components.preprocessors import TextPreprocessor from haystack.core import Pipeline # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class KeywordRetrievalConfig: """关键词检索配置""" top_k: int = 10 fuzzy: bool = False fuzziness: int = 2 all_terms_must_match: bool = False scale_score: bool = True language: str = "zh" # zh/en/multi class ChineseTextPreprocessor: """中文文本预处理器""" def __init__(self): self.jieba = jieba # 初始化停用词 self.stopwords = self._load_stopwords() def _load_stopwords(self) -> set: """加载中文停用词""" # 这里使用一个基本的停用词集合，实际应用中可以加载更大的停用词表 basic_stopwords = { '的', '了', '在', '是', '我', '有', '和', '就', '不', '人', '都', '一', '一个', '上', '也', '很', '到', '说', '要', '去', '你', '会', '着', '没有', '看', '好', '自己', '这' } return basic_stopwords def preprocess(self, text: str) -> str: """预处理中文文本""" # 分词 words = self.jieba.cut(text) # 过滤停用词和单字词 filtered_words = [ word.strip() for word in words if word.strip() and word.strip() not in self.stopwords and len(word.strip()) > 1 ] return " ".join(filtered_words) class EnterpriseKeywordRetriever: """企业级关键词检索系统""" def __init__(self, config: KeywordRetrievalConfig): self.config = config self.document_store = None self.bm25_retriever = None self.preprocessor = None self.is_initialized = False def initialize(self): """初始化检索系统""" try: logger.info("正在初始化企业级关键词检索系统") # 1. 初始化文档存储 self.document_store = InMemoryDocumentStore() # 2. 初始化预处理器 if self.config.language == "zh": self.preprocessor = ChineseTextPreprocessor() else: self.preprocessor = TextPreprocessor( remove_punctuation=True, remove_stopwords=True, lowercase=True, stemming=True ) # 3. 初始化BM25检索器 self.bm25_retriever = BM25Retriever( document_store=self.document_store, top_k=self.config.top_k, fuzzy=self.config.fuzzy, fuzziness=self.config.fuzziness, all_terms_must_match=self.config.all_terms_must_match, scale_score=self.config.scale_score ) self.is_initialized = True logger.info("关键词检索系统初始化完成") except Exception as e: logger.error(f"初始化失败: {e}") raise def preprocess_text(self, text: str) -> str: """预处理文本""" if self.preprocessor: if self.config.language == "zh": return self.preprocessor.preprocess(text) else: result = self.preprocessor.run(texts=[text]) return result['processed_texts'][0] return text def index_documents(self, documents: List[Dict[str, Any]]): """索引文档""" if not self.is_initialized: raise RuntimeError("检索系统未初始化") try: logger.info(f"开始索引 {len(documents)} 个文档") # 对文档内容进行预处理 processed_documents = [] for doc in documents: processed_content = self.preprocess_text(doc['content']) processed_doc = doc.copy() processed_doc['content'] = processed_content processed_documents.append(processed_doc) # 写入文档 self.document_store.write_documents(processed_documents) logger.info(f"成功索引 {len(documents)} 个文档") except Exception as e: logger.error(f"索引文档失败: {e}") raise def search(self, query: str, filters: Optional[Dict[str, Any]] = None, use_preprocessing: bool = True) -> Dict[str, Any]: """执行关键词检索""" if not self.is_initialized: raise RuntimeError("检索系统未初始化") start_time = time.time() try: # 预处理查询 processed_query = self.preprocess_text(query) if use_preprocessing else query # 执行检索 results = self.bm25_retriever.run(query=processed_query, filters=filters) # 计算性能指标 query_time = time.time() - start_time return { "documents": results['documents'], "original_query": query, "processed_query": processed_query, "query_time": query_time, "result_count": len(results['documents']) } except Exception as e: logger.error(f"检索失败: {e}") raise def batch_search(self, queries: List[str], use_preprocessing: bool = True) -> List[Dict[str, Any]]: """批量检索""" results = [] for query in queries: result = self.search(query, use_preprocessing=use_preprocessing) results.append(result) return results # 使用示例 if __name__ == "__main__": # 配置检索系统 config = KeywordRetrievalConfig( top_k=5, fuzzy=True, fuzziness=2, language="zh" ) # 创建检索系统 retriever = EnterpriseKeywordRetriever(config) retriever.initialize() # 准备文档数据 documents = [ { "id": "doc1", "content": "机器学习是人工智能的一个重要分支，主要研究如何让计算机系统从数据中学习模式和规律。主要包括监督学习、无监督学习和强化学习三种主要范式。监督学习使用标记数据进行训练，无监督学习从无标记数据中发现隐藏模式，强化学习通过试错来学习最优策略。", "meta": {"category": "AI", "language": "zh"} }, { "id": "doc2", "content": "深度学习是机器学习的一个子领域，使用多层神经网络来学习数据的复杂表示。卷积神经网络（CNN）在图像识别任务中表现出色，循环神经网络（RNN）和长短期记忆网络（LSTM）在序列数据处理中效果显著。Transformer架构的出现彻底改变了自然语言处理领域，基于自注意力机制成为主流。", "meta": {"category": "AI", "language": "zh"} }, { "id": "doc3", "content": "自然语言处理（NLP）是人工智能的一个重要分支，专注于计算机与人类语言之间的交互。传统NLP方法基于规则和统计模型，现代NLP主要基于深度学习和Transformer架构。BERT、GPT、T5等预训练模型在各种NLP任务中取得了突破性进展，包括文本分类、情感分析、机器翻译、问答系统等。", "meta": {"category": "NLP", "language": "zh"} }, { "id": "doc4", "content": "计算机视觉是使计算机能够从图像和视频中获取高级理解的技术领域。它模拟人类视觉系统，目标包括图像分类、目标检测、图像分割、图像生成等任务。经典的计算机视觉算法包括SIFT、SURF、HOG等特征提取方法，现代方法主要基于深度学习，如YOLO、SSD、Mask R-CNN等目标检测算法。", "meta": {"category": "CV", "language": "zh"} }, { "id": "doc5", "content": "检索增强生成（RAG）是一种结合信息检索和文本生成的新兴技术。它通过外部知识库来增强语言模型的能力，解决了大语言模型的知识时效性和幻觉问题。RAG系统通常包括文档加载、预处理、存储、检索和生成等组件，Haystack框架提供了完整的RAG解决方案支持。", "meta": {"category": "RAG", "language": "zh"} } ] # 索引文档 retriever.index_documents(documents) # 测试检索 test_queries = [ "机器学习是什么", "深度学习神经网络", "自然语言处理应用", "计算机视觉目标检测", "检索增强生成优势" ] print("企业级关键词检索系统测试:") print("=" * 60) for query in test_queries: results = retriever.search(query) print(f"\n查询: '{query}'") print(f"处理后查询: '{results['processed_query']}'") print(f"检索时间: {results['query_time']:.4f}秒") print(f"检索到 {results['result_count']} 个结果:") for i, doc in enumerate(results['documents'], 1): print(f" {i}. [分数: {doc.score:.4f}] {doc.content[:70]}...") # 测试模糊搜索 print("\n" + "=" * 60) print("模糊搜索测试:") print("-" * 40) fuzzy_queries = [ "机械学习", # 机器学习的错别字 "深度学", # 部分匹配 "神经网路", # 神经网络的错误 "计算器视觉" # 计算机视觉的错误 ] for query in fuzzy_queries: results = retriever.search(query, use_preprocessing=True) print(f"\n模糊查询: '{query}' -> 找到 {results['result_count']} 个结果:") for i, doc in enumerate(results['documents'][:3], 1): print(f" {i}. [分数: {doc.score:.4f}] {doc.content[:50]}...")
常见问题 FAQ
Q1：BM25与TF-IDF有什么区别？
A：BM25是TF-IDF的改进版本，主要区别包括：
- 词频饱和：BM25引入了k1参数，避免了词频的线性增长，更加符合实际
- 文档长度归一化：BM25考虑文档长度对检索的影响
- 更好的效果：BM25在实际应用中通常比TF-IDF表现更好
Q2：什么时候应该选择关键词检索而不是向量检索？
A：关键词检索更适合的场景：
- 精确匹配需求：需要找到包含特定关键词的文档
- 术语密集型任务：如法律文档、技术文档检索
- 实时性要求高：关键词检索通常比向量检索更快
- 可解释性强：可以清楚知道为什么某个文档被检索到
- 小规模数据集：数据量不大时，关键词检索效果通常足够好
Q3：如何优化中文关键词检索的效果？
A：中文关键词检索优化策略：
- 专业分词：使用jieba等中文分词工具
- 停用词过滤：去除常见的无意义词汇
- 同义词扩展：添加同义词词典
- 领域词典：针对特定领域构建专业词典
- 拼音输入支持：支持拼音输入到汉字的转换
最佳实践与避坑
实践 1：合理配置BM25参数
# 根据数据特点调整BM25参数 def tune_bm25_retriever(document_store, sample_queries): """调优BM25参数""" param_combinations = [ {"k1": 1.2, "b": 0.75}, # 默认参数 {"k1": 2.0, "b": 0.75}, # 更高的词频饱和度 {"k1": 1.2, "b": 1.0}, # 更强的文档长度归一化 {"k1": 0.8, "b": 0.5}, # 更保守的参数 ] best_params = None best_score = 0 for params in param_combinations: bm25 = BM25Retriever( document_store=document_store, top_k=5, k1=params["k1"], b=params["b"] ) # 这里应该有评估指标，简化示例 score = evaluate_bm25(bm25, sample_queries) if score > best_score: best_score = score best_params = params return best_params
实践 2：混合检索策略
class HybridKeywordRetriever: """混合关键词检索系统""" def __init__(self, document_store): self.document_store = document_store self.exact_retriever = BM25Retriever( document_store=document_store, top_k=3, all_terms_must_match=True # 精确匹配 ) self.fuzzy_retriever = BM25Retriever( document_store=document_store, top_k=5, fuzzy=True, fuzziness=2 # 模糊匹配 ) def search(self, query): """执行混合检索""" exact_results = self.exact_retriever.run(query=query) fuzzy_results = self.fuzzy_retriever.run(query=query) # 合并结果，去重，重排序 all_docs = exact_results['documents'] + fuzzy_results['documents'] unique_docs = {} for doc in all_docs: if doc.id not in unique_docs or doc.score > unique_docs[doc.id].score: unique_docs[doc.id] = doc # 按分数排序 sorted_docs = sorted(unique_docs.values(), key=lambda x: x.score, reverse=True) return { "documents": sorted_docs[:10], # 返回前10个结果 "exact_count": len(exact_results['documents']), "fuzzy_count": len(fuzzy_results['documents']) }
坑点 1：中文分词问题
常见错误：直接按空格分词导致中文检索效果差。
解决方案：使用jieba等中文分词工具进行预处理。
坑点 2：模糊搜索过度使用
常见错误：过度使用模糊搜索导致检索精度下降。
解决方案：根据查询类型动态选择是否启用模糊匹配，对精确查询关闭模糊搜索。
本节小结
本章节详细介绍了关键词检索技术，特别是BM25算法的原理和应用。通过Haystack的BM25Retriever，我们可以构建高效的关键词检索系统。关键要点包括：理解BM25算法的数学原理、掌握文本预处理技术、合理配置检索参数、实现模糊搜索功能，以及构建企业级的混合检索系统。下一节我们将学习如何将关键词检索和向量检索有机结合，形成强大的混合检索策略。
延伸阅读
关键词：关键词检索, BM25, 倒排索引, 文本预处理, Haystack, 企业级搜索
难度：进阶
预计阅读：75分钟
评论区
(0)
U