基于 Faiss 的相似度过滤与 Top-K 检索在推荐系统召回层的应用教程

在推荐系统的召回层，我们需要快速地从海量物料中找出与用户画像或查询物品最相似的K个结果。但仅仅依赖向量相似度往往不够，我们还需要结合业务需求进行过滤，例如只召回特定品类、特定库存状态的商品。由于 Faiss 自身不提供复杂的SQL式元数据过滤功能，我们必须采用“检索+后置过滤”的混合策略。

本文将详细介绍如何使用 Faiss 进行 Top-K 检索，并通过 ID 映射实现高效的相似度后置过滤，确保召回结果既相似又满足业务条件。

1. 环境准备

确保你安装了 faiss-cpu 和 numpy。

pip install faiss-cpu numpy

2. 数据准备与索引创建

我们创建一个包含 1000 个 64 维向量的模拟数据集，并为每个向量分配一个模拟的 category_id 作为元数据。

import faiss
import numpy as np
import time

# 1. 配置参数
N = 1000        # 数据库向量数量
D = 64          # 向量维度

# 2. 生成模拟数据
np.random.seed(42)
x_data = np.random.random((N, D)).astype('float32')

# 3. 生成元数据 (例如，品类ID)
# 假设我们有 3 个品类 (1, 2, 3)
metadata_categories = np.random.randint(1, 4, N)

# 4. 创建 Faiss 索引
index = faiss.IndexFlatL2(D) # 使用 IndexFlatL2 进行精确查找
index.add(x_data)

print(f"Faiss 索引包含 {index.ntotal} 个向量。\n")

3. 实现召回与后置过滤逻辑

我们的目标是：给定一个查询向量，找出与其最近的 50 个向量，但最终只保留那些 category_id 为 2 的结果。

在这个混合方法中，Faiss 负责速度极快的 Top-N 检索，而 Python 字典和数组负责将 Faiss 返回的全局索引（D_indices）映射回我们的业务元数据并进行过滤。

# 5. 定义查询向量
q_vector = np.random.random((1, D)).astype('float32')

# 6. 配置检索参数
K_retrieve = 50  # Faiss 召回的候选项数量 (N >> K_final)
K_final = 10     # 最终需要的数量
target_category_id = 2 # 我们的过滤条件

# ---- 检索阶段 ----

t0 = time.time()
# D: 距离矩阵, I: 索引矩阵
D, I = index.search(q_vector, K_retrieve)
t1 = time.time()
print(f"Faiss 检索 {K_retrieve} 个候选项耗时: {t1 - t0:.4f} 秒")

# ---- 后置过滤阶段 ----

retrieved_indices = I[0] # 获取检索到的全局索引
filtered_results = []

for rank, global_idx in enumerate(retrieved_indices):
    # 1. 查找元数据
    item_category = metadata_categories[global_idx]
    item_distance = D[0][rank]

    # 2. 应用过滤条件
    if item_category == target_category_id:
        filtered_results.append({
            'global_index': int(global_idx),
            'distance': float(item_distance),
            'category_id': int(item_category)
        })

    # 3. 达到最终数量后停止 (可选的优化)
    if len(filtered_results) >= K_final:
        break

print(f"\n过滤条件: 只保留 category_id = {target_category_id} 的结果")
print(f"最终召回数量: {len(filtered_results)}")

print("\n--- 最终召回结果 (按相似度排序) ---")
for result in filtered_results:
    print(f"Index: {result['global_index']}, Category: {result['category_id']}, Distance: {result['distance']:.4f}")

4. 结果分析与应用建议

为什么采用“检索 K_retrieve 后再过滤 K_final”？

如果直接在 Faiss 中只检索 Top-K_final，可能会因为过滤掉不符合条件的向量，导致最终返回的结果数量不足，甚至不是最相似的。因此，我们通常设置一个较大的召回池 (K_retrieve)，确保在进行元数据过滤后，我们仍能得到足量的、且在相似度上排名靠前的结果。

在实际推荐系统中的应用：

1. Faiss 负责高性能召回： 使用 Faiss 专注于向量相似度计算，快速产出原始的 ID 列表。
2. 外部系统管理元数据： 实际系统中，metadata_categories 数组会存储在像 Redis 或 MySQL 这样的外部数据库中，通过 Faiss 返回的 ID 批量查询元数据。
3. 应用层负责过滤： 推荐引擎的业务逻辑层根据查询到的元数据，对 Faiss 返回的结果进行精确的、业务驱动的过滤和重排序。