汤不热吧在现代搜索系统中，用户往往需要同时考虑关键词匹配（全文搜索）和语义相似性（向量搜索）。单独使用 Elasticsearch（ES）进行全文搜索，或单独使用 Faiss 进行向量搜索，都无法满足所有需求。本文将介绍一种高效、实操性强的混合检索方案：利用 Elasticsearch 进行元数据过滤和关键词检索，再利用 Faiss 对筛选后的数据集进行快速向量相似性排序。
1. 混合检索的价值
- Elasticsearch (ES): 擅长快速的结构化数据查询、全文搜索、过滤和聚合。它用于确定“哪些文档是相关的”。
- Faiss: 擅长高维向量的近邻搜索（kNN），尤其在亿级数据量下性能卓越。它用于确定“在相关文档中，哪些是最相似的”。
2. 混合检索的实施步骤
混合检索并非将向量直接存入 ES，而是将 ES 作为向量 ID 和元数据的索引库，Faiss 作为向量计算引擎。流程如下：
- 数据同步： 文档的元数据（标题、内容、ID）存储在 Elasticsearch 中。文档的向量表示存储在 Faiss 索引中。Faiss 内部的索引编号必须能映射回 Elasticsearch 的文档 ID。
- 预过滤/关键词检索 (ES 阶段)： 用户提交查询，首先在 Elasticsearch 中执行关键词查询或结构化过滤（例如：category: ‘Tech’ AND text: ‘AI’）。
- 获取候选 ID 列表： ES 返回符合条件的文档 ID 列表。
- 向量搜索 (Faiss 阶段)： 使用用户的查询向量，仅对步骤 3 中获得的候选 ID 对应的向量子集执行相似性搜索。
- 结果融合： 返回最终排序后的结果。
3. 代码实操：模拟 ES 预过滤与 Faiss 搜索
我们假设已经完成了数据生成和索引创建。这里的重点是，一旦我们从 ES 得到了一个合法的索引 ID 列表（即那些通过全文搜索或过滤的文档），我们如何高效地让 Faiss 只搜索这些特定的向量。
虽然 Faiss 没有内置的原生过滤机制（如 SQL WHERE IN），但我们可以通过构建特定的索引类型或在查询后进行后处理来实现。最实用的方法是使用 IndexIDMap 或 IndexIVF 配合 ID 映射，但对于小规模演示，我们通过 ID 映射和 IndexFlat 来说明流程。
前提： 假设我们使用一个简单的 Faiss 索引，其内部序号就是我们的文档 ID。
import faiss
import numpy as np
import time
# 1. 初始化数据
d = 64 # 向量维度
nb = 100000 # 总数据量
n_candidates = 500 # 假设这是ES预过滤得到的候选集大小
# 生成随机数据
np.random.seed(1234)
x_data = np.random.random((nb, d)).astype('float32')
x_data[:, 0] += np.arange(nb) / 1000.
# 生成查询向量
x_query = np.random.random((1, d)).astype('float32')
# 2. 创建并填充 Faiss 索引
index = faiss.IndexFlatL2(d)
index.add(x_data)
print(f"Faiss 索引大小: {index.ntotal}")
# 3. 模拟 ES 预过滤阶段
# 假设 ES 返回了一组符合关键词/元数据条件的内部索引 ID
# 这里我们随机选择 500 个作为候选集
np.random.seed(42)
es_filtered_indices = np.random.choice(nb, size=n_candidates, replace=False)
print(f"ES 过滤得到 {len(es_filtered_indices)} 个候选 ID")
# 4. Faiss 向量搜索（混合搜索的核心）
# 提取候选向量子集
candidate_vectors = x_data[es_filtered_indices]
# 创建一个临时的 Faiss 索引来搜索这个子集 (最简单但高效的方法)
# 注意：对于大规模实时系统，通常会使用 Faiss IndexIVF 或 IndexIDMap
# 但对于已经被过滤的小集合，FlatL2 性能优秀
temp_index = faiss.IndexFlatL2(d)
temp_index.add(candidate_vectors)
# 执行 kNN 搜索
k = 10 # 搜索最相似的 10 个结果
start_time = time.time()
D, I_temp = temp_index.search(x_query, k)
end_time = time.time()
# 5. 结果映射回原始 ID
# I_temp 是在 candidate_vectors 中的索引，我们需要将其映射回原始的 x_data ID
final_faiss_indices = es_filtered_indices[I_temp.flatten()]
print("\n--- 混合检索结果 ---")
print(f"搜索耗时: {end_time - start_time:.4f} 秒")
print("距离 (D):", D.flatten())
print("原始文档 ID (I):", final_faiss_indices)
# 6. 后续处理: 使用 final_faiss_indices 去 ES 中查询完整的元数据并展示结果
4. 总结与优化
该混合检索方案的关键在于削减搜索空间。通过 ES 的精准过滤，我们将需要进行高耗能向量搜索的数据量从 $N$ 降低到了 $M$（其中 $M ext{《} N$）。
对于超大规模数据：
- IndexIVF + IDMap： 如果需要对整个索引进行 Faiss 搜索，但又需要快速的 ID 映射，可以使用 IndexIVF 配合 IndexIDMap。
- 向量存储在 ES 中 (新版)： 新版本的 Elasticsearch 支持原生向量字段和近似近邻搜索（kNN），这在一定程度上简化了部署，但对于追求极致性能和索引灵活性的场景，Faiss 仍是首选。结合 Faiss 和 ES 仍是最高性能的混合检索架构。