使用 Python 构建的混合搜索流程结合了关键词搜索(特别是BM25)与向量相似度搜索。此流程整合了这些不同的检索方法,并合并它们的结果以提升相关性。我们假设您有一个可运行的Python环境,其中包含用于向量操作的库,以及可能存在的向量数据库客户端或Faiss等库。您还需要关键词搜索的库。前提条件与设置在我们开始之前,请确保已安装所需的库。在本例中,我们将使用 rank_bm25 进行关键词搜索,sentence-transformers 用于生成嵌入(尽管在搜索阶段我们假定向量是预先计算好的),以及 numpy。pip install rank_bm25 numpy sentence-transformers我们将使用一个文档数据集,其中每个文档都包含ID、文本内容和预先计算好的向量嵌入。# 示例数据结构(请替换为您的实际数据加载方式) documents = [ {"id": "doc1", "text": "Introduction to machine learning concepts.", "vector": [0.1, 0.2, ..., 0.5]}, {"id": "doc2", "text": "Advanced deep learning techniques for NLP.", "vector": [0.3, 0.1, ..., 0.7]}, {"id": "doc3", "text": "A guide to Python programming basics.", "vector": [0.8, 0.7, ..., 0.1]}, {"id": "doc4", "text": "Optimizing machine learning models.", "vector": [0.2, 0.3, ..., 0.6]}, # ... 更多文档 ] # 模拟将预计算的向量加载到结构中以供查找 vector_index_data = {doc["id"]: doc["vector"] for doc in documents} document_texts = {doc["id"]: doc["text"] for doc in documents} # 查询示例 query_text = "machine learning optimization" # 假设我们有一个获取查询嵌入的函数 # 在实际场景中,您会使用与文档向量相同的模型 from sentence_transformers import SentenceTransformer embedding_model = SentenceTransformer('all-MiniLM-L6-v2') # 示例模型 def get_query_embedding(query): return embedding_model.encode(query) query_vector = get_query_embedding(query_text)步骤1:实现关键词搜索 (BM25)首先,让我们设置BM25组件。我们将对文档的文本内容进行索引。rank_bm25 库提供了一个直接的实现。import numpy as np from rank_bm25 import BM25Okapi # 对文档进行分词(本例中使用简单的空格分隔) tokenized_corpus = [doc["text"].lower().split() for doc in documents] doc_ids = [doc["id"] for doc in documents] # 创建BM25索引 bm25 = BM25Okapi(tokenized_corpus) def perform_keyword_search(query_text, top_n=10): """执行BM25搜索并返回排名的文档ID和分数。""" tokenized_query = query_text.lower().split() # 获取所有文档的分数 doc_scores = bm25.get_scores(tokenized_query) # 获取前N个分数的索引 top_indices = np.argsort(doc_scores)[::-1][:top_n] # 将索引映射回文档ID并创建结果列表 results = [] for index in top_indices: if doc_scores[index] > 0: # 只包含分数非零的文档 results.append({"id": doc_ids[index], "score": doc_scores[index]}) return results # 执行关键词搜索 keyword_results = perform_keyword_search(query_text, top_n=5) print("关键词搜索结果 (BM25):") print(keyword_results) # 预期输出(示例): # 关键词搜索结果 (BM25): # [{'id': 'doc4', 'score': 0.937...}, {'id': 'doc1', 'score': 0.937...}, {'id': 'doc2', 'score': 0.0}] # 注意:分数很大程度上取决于语料库这会给我们一个文档ID列表,它们按其与查询的BM25相关性分数进行排序。步骤2:实现向量搜索接下来,我们模拟向量搜索组件。在实际系统中,这会涉及查询一个从 vector_index_data 构建的ANN索引(如Faiss、OpenSearch、Pinecone、Weaviate等中的HNSW)。在这里,我们将通过计算与所有向量的余弦相似度来模拟它(对于大型数据集效率不高,但足以演示混合方法的想法)。from numpy.linalg import norm def cosine_similarity(vec_a, vec_b): """计算两个向量之间的余弦相似度。""" return np.dot(vec_a, vec_b) / (norm(vec_a) * norm(vec_b)) def perform_vector_search(query_vector, index_data, top_n=10): """执行向量搜索并返回排名的文档ID和分数。""" scores = [] for doc_id, doc_vector in index_data.items(): similarity = cosine_similarity(query_vector, doc_vector) scores.append({"id": doc_id, "score": similarity}) # 按分数降序排序 scores.sort(key=lambda x: x["score"], reverse=True) return scores[:top_n] # 执行向量搜索 vector_results = perform_vector_search(query_vector, vector_index_data, top_n=5) print("\n向量搜索结果:") print(vector_results) # 预期输出(示例): # 向量搜索结果: # [{'id': 'doc4', 'score': 0.85...}, {'id': 'doc1', 'score': 0.78...}, {'id': 'doc2', 'score': 0.65...}, {'id': 'doc3', 'score': 0.12...}]现在我们有了第二个文档ID列表,按语义相似度进行排序。请注意,分数范围不同(余弦相似度通常在[-1, 1]或[0, 1]之间,BM25通常为正且无上限),并且排名可能与关键词结果有显著差异。步骤3:使用倒数排名融合 (RRF) 合并结果如前所述,由于分数尺度和分布不同,简单地相加或平均分数会带来问题。倒数排名融合 (RRF) 提供了一种有效的方法来组合排名,而无需分数归一化。文档 $d$ 的RRF分数计算公式如下: $$Score_{RRF}(d) = \sum_{i \in \text{结果列表}} \frac{1}{k + rank_i(d)}$$ 其中 $rank_i(d)$ 是文档 $d$ 在结果列表 $i$ 中的排名(从1开始),而 $k$ 是一个常数,控制着较低排名项的影响。$k$ 的一个常用值是60。让我们实现RRF函数:def apply_rrf(results_lists, k=60): """应用倒数排名融合来组合多个排名列表。""" fused_scores = {} doc_data = {} # 用于存储原始分数或仅存储ID(如果需要) for results in results_lists: for rank, result in enumerate(results): doc_id = result["id"] if doc_id not in fused_scores: fused_scores[doc_id] = 0 doc_data[doc_id] = {"id": doc_id} # 存储关联数据 # 计算此列表的RRF贡献 rrf_score = 1.0 / (k + rank + 1) # 排名是基于0的,所以加1 fused_scores[doc_id] += rrf_score # 组合ID和分数,然后排序 fused_results = [{"id": doc_id, "score": fused_scores[doc_id]} for doc_id in fused_scores] fused_results.sort(key=lambda x: x["score"], reverse=True) return fused_results # 组合关键词和向量搜索结果 hybrid_results = apply_rrf([keyword_results, vector_results], k=60) print("\n混合搜索结果 (RRF):") print(hybrid_results) # 预期输出(示例,取决于输入排名): # 混合搜索结果 (RRF): # [{'id': 'doc4', 'score': 0.032...}, {'id': 'doc1', 'score': 0.032...}, {'id': 'doc2', 'score': 0.016...}, {'id': 'doc3', 'score': 0.015...}] # 注意:RRF分数是相对的;排名是主要输出。 # 在本例中,doc4和doc1在两个列表中都排名靠前,获得了更高的RRF分数。组合起来:混合流程工作原理我们可以将整个过程封装到一个函数中,表示我们的混合搜索流程。digraph HybridSearchPipeline { rankdir=LR; node [shape=box, style=filled, fillcolor="#a5d8ff"]; edge [color="#495057"]; Query [label="用户查询", shape=ellipse, style=filled, fillcolor="#ffec99"]; Embedding [label="生成查询嵌入"]; VectorSearch [label="向量搜索\n(ANN 索引)"]; KeywordSearch [label="关键词搜索\n(BM25 索引)"]; Fusion [label="结果合并\n(RRF)", style=filled, fillcolor="#96f2d7"]; Results [label="排序结果", shape=ellipse, style=filled, fillcolor="#ffec99"]; Query -> Embedding [label=" 文本 "]; Query -> KeywordSearch [label=" 文本 "]; Embedding -> VectorSearch [label=" 向量 "]; subgraph cluster_retrieval { label = "检索阶段"; style=filled; color="#e9ecef"; VectorSearch; KeywordSearch; } VectorSearch -> Fusion [label=" 向量排名 "]; KeywordSearch -> Fusion [label=" 关键词排名 "]; Fusion -> Results; }这是一个图表,说明了混合搜索流程。用户查询既用于关键词搜索,也用于生成向量搜索的嵌入。然后将两者的结果合并(例如,使用RRF)以生成最终的排名列表。def hybrid_search_pipeline(query_text, top_n=10, rrf_k=60): """执行完整的混合搜索流程。""" print(f"正在为查询 '{query_text}' 执行混合搜索") # 1. 获取查询嵌入 query_vector = get_query_embedding(query_text) # 2. 执行关键词搜索 keyword_results = perform_keyword_search(query_text, top_n=top_n * 2) # 最初检索更多结果 print(f" BM25 返回了 {len(keyword_results)} 个结果。") # 3. 执行向量搜索 # 请将此占位符替换为对您实际向量搜索系统的调用 vector_results = perform_vector_search(query_vector, vector_index_data, top_n=top_n * 2) # 最初检索更多结果 print(f" 向量搜索返回了 {len(vector_results)} 个结果。") # 4. 应用RRF合并 fused_results = apply_rrf([keyword_results, vector_results], k=rrf_k) print(f" 合并结果生成了 {len(fused_results)} 个候选。") # 5. 返回最终的前N个结果及文本 final_results = [] for result in fused_results[:top_n]: doc_id = result["id"] final_results.append({ "id": doc_id, "text": document_texts.get(doc_id, "N/A"), # 获取文本以提供上下文 "rrf_score": result["score"] }) return final_results # 使用示例 final_ranked_list = hybrid_search_pipeline(query_text, top_n=5) print("\n最终混合排序列表:") for i, item in enumerate(final_ranked_list): print(f"{i+1}. ID: {item['id']}, 分数: {item['rrf_score']:.4f}, 文本: {item['text'][:80]}...")考量与后续步骤性能: 该流程执行两次独立的搜索查询(关键词和向量),然后进行合并。总延迟大约是两次搜索的最大延迟加上合并开销(通常可以忽略不计)。对于生产系统,优化各个搜索组件(第二章)并可能并行化查询是很重要的。参数 k 的调整: RRF常数 k 影响着对较低排名项给予的权重。较小的 k 会给予靠前排名更高的重要性。其最优值可能取决于特定的数据集和查询模式,并可以通过离线评估(第五章)进行调整。检索量: 请注意,在合并之前,我们从每个系统检索了 top_n * 2 个结果。在合并之前从每个来源检索更多候选(例如,前50或100个)通常会带来更好的最终结果,但代价是合并计算量略有增加。其他合并方法: 尽管RRF很常见,但可以考虑其他方法,例如加权分数组合(如果分数可以有意义地归一化),或更复杂的基于两种搜索结果特征训练的排序学习模型。评估: 使用相关性指标(如NDCG、召回率)在真实数据集上对混合系统进行严格评估,并与纯向量和纯关键词搜索进行对比,详细内容在第五章。这有助于量化效益并找出可以改进的地方。这个动手练习为构建混合搜索系统提供了一个基本蓝图。通过结合词法和语义检索的优势,您通常可以获得比单独使用任一方法显著更好的相关性,尤其是在RAG等LLM应用中常见的复杂信息需求场景下。