混合搜索查询

混合搜索（Hybrid Search）是一种结合多种检索技术的搜索方法，旨在同时利用语义搜索（向量检索）和关键词搜索（如BM25、TF-IDF）的优势，以提高搜索结果的准确性和相关性。以下是几种主要的混合搜索查询方法：

1. 基于倒排索引 + 向量检索的混合搜索

• 原理：使用传统的倒排索引（如Elasticsearch、Lucene）进行关键词匹配，同时结合向量数据库（如Milvus、Pinecone）进行语义相似度计算。

• 适用场景：适用于需要同时支持精确关键词匹配和语义相似性搜索的应用，如电商搜索、文档检索等。

• 示例：

  • 在Elasticsearch中结合BM25进行关键词搜索，同时使用向量数据库进行语义搜索，最后融合结果24。

  • Pinecone的混合搜索模式，支持稀疏向量（BM25）和密集向量（如CLIP、BERT）的联合查询1。

2. 基于多路召回 + 重排序（RRF, Weighted Fusion）

• 原理：先通过不同检索方式（如关键词搜索、向量搜索、稀疏向量搜索）分别召回候选集，再使用融合算法（如倒数排序融合RRF、加权融合）进行重排序。

• 适用场景：适用于需要高召回率的场景，如推荐系统、问答系统（RAG）。

• 示例：

  • RRF（Reciprocal Rank Fusion）：对每个检索路径的结果按排名分配分数（如1/rank），最后合并得分27。

  • 加权融合：对不同检索方法的结果赋予不同权重（如语义搜索权重0.7，关键词搜索权重0.3）7。

3. 基于结构化 + 非结构化数据的混合查询

• 原理：在向量数据库中，同时支持结构化字段（如价格、时间）过滤和非结构化数据（如文本、图像）的向量搜索。

• 适用场景：适用于需要结合业务规则（如筛选条件）和语义搜索的应用，如商品搜索、人脸检索等。

• 示例：

  • Milvus支持在向量检索时附加结构化过滤条件（如price > 100 AND category = 'electronics'）39。

  • PostgreSQL的pgvector扩展支持结合SQL条件和向量相似度搜索8。

4. 基于稀疏向量 + 密集向量的混合搜索

• 原理：

  • 稀疏向量（如BM25、SPLADE）：适用于关键词精确匹配，维度高但大部分为零。

  • 密集向量（如BERT、CLIP）：适用于语义搜索，维度固定且连续。

• 适用场景：适用于需要同时支持精确术语匹配和语义理解的场景，如学术论文检索、法律文档搜索等。

• 示例：

  • Milvus的Sparse-BM25功能，支持稀疏向量和密集向量的联合查询4。

  • BGE-M3模型同时生成稀疏和密集向量，用于混合检索7。

5. 基于图索引的混合查询（如HNSW + 过滤搜索）

• 原理：在近似最近邻搜索（ANNS）图结构（如HNSW、NSG）中，结合结构化约束进行动态剪枝或路由优化。

• 适用场景：适用于高维向量搜索+强业务规则过滤的场景，如人脸库检索、商品推荐等。

• 示例：

  • 过滤贪心搜索（FGS）：在HNSW搜索时仅遍历满足结构化条件的节点9。

  • 容忍因子算法（TF-FGS）：允许部分不满足条件的节点参与路由，提高召回率9。

这些方法可根据具体业务需求组合使用，例如：

• 电商搜索：结构化过滤（价格/品牌） + 关键词搜索（BM25） + 语义搜索（向量）38。

• RAG问答：密集向量（语义） + 稀疏向量（关键词） + 全文搜索（BM25）三重召回7。

6.Milvus+python实现多路召回 + 重排序的混合搜索查询

1. 环境准备

 pip install pymilvus numpy

2. 初始化 Milvus 连接

 from pymilvus import MilvusClient, DataType
 import numpy as np
 ​
 # 连接到 Milvus
 client = MilvusClient(
     uri="http://localhost:19530",
     db_name="default"
 )

3. 创建支持混合搜索的集合

假设集合包含：

• 主键 id（INT64）

• 密集向量 embedding（FLOAT_VECTOR, 128维）

• 文本字段 text（用于关键词召回）

 # 删除已存在的集合（可选）
 client.drop_collection("hybrid_search_demo")
 ​
 # 创建集合 Schema
 schema = MilvusClient.create_schema(
     auto_id=False,
     enable_dynamic_field=False
 )
 ​
 # 添加字段
 schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
 schema.add_field(field_name="embedding", datatype=DataType.FLOAT_VECTOR, dim=128)
 schema.add_field(field_name="text", datatype=DataType.VARCHAR, max_length=512)
 ​
 # 创建集合
 client.create_collection(
     collection_name="hybrid_search_demo",
     schema=schema,
     index_params={
         "index_type": "IVF_FLAT",
         "metric_type": "L2",
         "params": {"nlist": 128}
     }
 )

4. 插入测试数据

 # 生成示例数据
 data = [
     {"id": 1, "text": "苹果手机 iPhone 13", "embedding": np.random.rand(128).tolist()},
     {"id": 2, "text": "华为 Mate 60 Pro", "embedding": np.random.rand(128).tolist()},
     {"id": 3, "text": "小米 14 Ultra", "embedding": np.random.rand(128).tolist()},
     # 添加更多数据...
 ]
 ​
 # 插入数据
 client.insert("hybrid_search_demo", data)

5. 实现多路召回 + RRF 重排序

5.1 定义召回方法

 def vector_recall(query_embedding, top_k=5):
     """向量召回：语义相似度搜索"""
     results = client.search(
         collection_name="hybrid_search_demo",
         data=[query_embedding],
         limit=top_k,
         output_fields=["id", "text"],
         search_params={"metric_type": "L2", "params": {"nprobe": 10}}
     )
     return [(hit["entity"]["id"], hit["distance"]) for hit in results[0]]
 ​
 def keyword_recall(query_text, top_k=5):
     """关键词召回：使用标量字段过滤（简化版BM25模拟）"""
     results = client.query(
         collection_name="hybrid_search_demo",
         filter=f"text like '%{query_text}%'",
         limit=top_k,
         output_fields=["id"]
     )
     return [(item["id"], 1.0) for item in results]  # 假设相关性得分为1.0

5.2 实现 RRF 重排序

 def reciprocal_rank_fusion(results_list, k=60):
     """RRF 重排序算法"""
     scores = {}
     for results in results_list:
         for rank, (doc_id, _) in enumerate(results, 1):
             scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank)
     
     # 按分数降序排序
     return sorted(scores.items(), key=lambda x: x[1], reverse=True)
 ​
 def hybrid_search_rrf(query_text, query_embedding, top_k=5):
     # 多路召回
     vector_results = vector_recall(query_embedding, top_k*2)  # 扩大召回量
     keyword_results = keyword_recall(query_text, top_k*2)
     
     # RRF 融合
     fused_results = reciprocal_rank_fusion([vector_results, keyword_results])
     
     # 获取最终结果详情
     final_ids = [doc_id for doc_id, _ in fused_results[:top_k]]
     final_results = client.get(
         collection_name="hybrid_search_demo",
         ids=final_ids,
         output_fields=["id", "text"]
     )
     return final_results

6. 执行混合搜索

 # 模拟查询
 query_text = "苹果手机"
 query_embedding = np.random.rand(128).tolist()  # 实际应用中应使用模型生成
 ​
 # 执行搜索
 results = hybrid_search_rrf(query_text, query_embedding)
 for item in results:
     print(f"ID: {item['id']}, Text: {item['text']}")

7. 加权融合替代方案（可选）

如果更倾向于加权分数而非RRF：

 def weighted_fusion(vector_results, keyword_results, vector_weight=0.7):
     scores = {}
     # 向量结果加权
     for doc_id, distance in vector_results:
         scores[doc_id] = (1 - distance) * vector_weight  # 假设distance是L2距离
     
     # 关键词结果加权
     for doc_id, score in keyword_results:
         scores[doc_id] = scores.get(doc_id, 0) + score * (1 - vector_weight)
     
     return sorted(scores.items(), key=lambda x: x[1], reverse=True)

关键点解析

1. 多路召回：

   • 向量召回：捕捉语义相似性

   • 关键词召回：保证术语精确匹配

2. 重排序算法：

   • RRF：无需调参，对异构召回结果融合效果稳定

   • 加权融合：需调整权重（通常向量权重更高）

3. 性能优化：

   • 扩大每路召回的 top_k（如2倍最终需求）以提高召回率

   • 对结构化字段建立标量索引加速过滤

8.检测召回是否正确

检测混合搜索中多路召回结果的正确性需要从 语义相关性、关键词匹配度 和 业务逻辑 三个维度综合评估。以下是详细的检测方法和 Python 实现示例：

1. 基础验证方法

(1) 人工抽样检查

 def manual_check(query, results):
     print(f"查询: '{query}'")
     for i, item in enumerate(results, 1):
         print(f"{i}. ID:{item['id']} | Text: {item['text']} | Score: {item.get('score', 'N/A')}")
     print("-" * 50)
 ​
 # 使用示例
 results = hybrid_search_rrf("苹果手机", query_embedding)
 manual_check("苹果手机", results)

(2) 自动化测试框架

 test_cases = [
     {
         "query": "苹果手机",
         "expected_ids": [1, 5, 8],  # 预期应召回的文档ID
         "min_precision": 0.7         # 要求前3结果中至少70%是相关的
     }
 ]
 ​
 def run_test_cases():
     for case in test_cases:
         results = hybrid_search_rrf(case["query"], get_embedding(case["query"]))
         retrieved_ids = [item['id'] for item in results]
         relevant = len(set(retrieved_ids) & set(case["expected_ids"]))
         precision = relevant / len(retrieved_ids)
         assert precision >= case["min_precision"], \
             f"测试失败: 查询 '{case['query']}' 精确度 {precision:.2f} < {case['min_precision']}"

2. 定量评估指标

(1) 计算召回率 (Recall) & 精确率 (Precision)

 def evaluate_search(gold_standard, results, top_k=5):
     """
     gold_standard: 人工标注的相关文档ID列表
     results: 实际召回结果列表
     """
     retrieved = [item['id'] for item in results[:top_k]]
     relevant_retrieved = len(set(retrieved) & set(gold_standard))
     
     precision = relevant_retrieved / len(retrieved)
     recall = relevant_retrieved / len(gold_standard)
     f1 = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0
     
     return {"precision": precision, "recall": recall, "f1": f1}
 ​
 # 使用示例
 gold_ids = [1, 3, 7]  # 已知与查询"苹果手机"相关的文档
 results = hybrid_search_rrf("苹果手机", query_embedding)
 metrics = evaluate_search(gold_ids, results)
 print(metrics)  # 输出: {'precision': 0.8, 'recall': 0.67, 'f1': 0.73}

(2) NDCG@K (衡量排序质量)

 import numpy as np
 ​
 def ndcg_at_k(gold_scores, results, k=5):
     """
     gold_scores: {doc_id: 相关性分数} (如 0=不相关, 1=相关, 2=高度相关)
     results: 召回结果列表
     """
     dcg = 0
     for i, item in enumerate(results[:k], 1):
         rel = gold_scores.get(item['id'], 0)
         dcg += (2 ** rel - 1) / np.log2(i + 1)
     
     ideal_scores = sorted(gold_scores.values(), reverse=True)[:k]
     idcg = sum((2 ** rel - 1) / np.log2(i + 2) for i, rel in enumerate(ideal_scores))
     
     return dcg / idcg if idcg > 0 else 0
 ​
 # 使用示例
 gold_scores = {1: 2, 3: 1, 7: 1}  # 文档相关性标注
 print(f"NDCG@5: {ndcg_at_k(gold_scores, results):.3f}")

3. 多路召回专项检测

(1) 检查各路召回结果

 def debug_recall_paths(query_text, query_embedding):
     print("=== 向量召回结果 ===")
     vec_results = vector_recall(query_embedding)
     manual_check(query_text, [{"id": r[0], "text": get_text_by_id(r[0])} for r in vec_results])
     
     print("=== 关键词召回结果 ===")
     kw_results = keyword_recall(query_text)
     manual_check(query_text, [{"id": r[0], "text": get_text_by_id(r[0])} for r in kw_results])
 ​
 # 辅助函数：根据ID获取文档文本
 def get_text_by_id(doc_id):
     res = client.get("hybrid_search_demo", ids=[doc_id], output_fields=["text"])
     return res[0]["text"]

(2) 验证融合效果

 def compare_fusion_methods(query_text, query_embedding):
     # 原始各路结果
     vec = vector_recall(query_embedding)
     kw = keyword_recall(query_text)
     
     # RRF融合结果
     rrf_results = reciprocal_rank_fusion([vec, kw])
     
     # 加权融合结果
     weighted_results = weighted_fusion(vec, kw)
     
     print("RRF Top3:", rrf_results[:3])
     print("Weighted Top3:", weighted_results[:3])

4. 业务规则验证

(1) 结构化字段过滤检查

 def test_filter_condition():
     # 测试价格过滤是否生效
     results = client.search(
         collection_name="hybrid_search_demo",
         data=[query_embedding],
         filter="price < 1000",  # 验证是否只有低价商品被召回
         output_fields=["price"]
     )
     prices = [hit["entity"]["price"] for hit in results[0]]
     assert all(p < 1000 for p in prices), "过滤条件未生效！"

(2) 动态字段检测

 def test_dynamic_fields():
     # 插入带动态字段的数据
     client.insert("hybrid_search_demo", {"id": 100, "text": "测试", "extra_field": "动态值"})
     
     # 验证是否能召回
     results = client.search(
         collection_name="hybrid_search_demo",
         data=[query_embedding],
         expr="exists(extra_field)"  # 动态字段查询
     )
     assert len(results[0]) > 0, "动态字段查询失败"

关键检查清单