Как работает FAISS?

FAISS (Facebook AI Similarity Search)

Введение

FAISS — это библиотека для быстрого поиска похожих векторов в высокомерных пространствах. Разработана Facebook AI Research (Meta). Используется для:

• Поиска семантически похожих текстов
• Рекомендательных систем
• Поиска изображений по изображению
• Деупликации данных
• Извлечения информации (retrieval)

Проблема наивного поиска

import numpy as np
from scipy.spatial.distance import cdist
import time

# Наивный поиск: O(n*d) где n - количество векторов, d - размерность
n_vectors = 1_000_000
dim = 768  # размер BERT эмбедингов

db = np.random.randn(n_vectors, dim).astype('float32')
query = np.random.randn(1, dim).astype('float32')

# Поиск всех расстояний (МЕДЛЕННО!)
start = time.time()
distances = cdist(query, db, metric='cosine')[0]
top_k = np.argsort(distances)[:10]
print(f"Наивный поиск: {time.time() - start:.2f} сек")
# > 10 секунд! Слишком медленно для production

Как FAISS решает проблему

FAISS использует несколько техник для ускорения:

1. Квантизация: сжимаем высокоразмерные векторы в низкоразмерные
2. Индексирование: строим специальные индексы для быстрого поиска
3. Приблизительный поиск (ANN): находим ПОХОЖИЕ результаты, не все
4. GPU ускорение: используем видеокарты для параллельных вычислений

Основные индексы FAISS

1. Brute Force Index (базовый)

import faiss
import numpy as np

# Создание данных
n_vectors = 100000
dim = 128
db = np.random.randn(n_vectors, dim).astype('float32')
db /= np.linalg.norm(db, axis=1, keepdims=True)  # нормализуем

# Brute Force: полный перебор (медленно, но точно)
index = faiss.IndexFlatL2(dim)  # L2 метрика (Евклидово расстояние)
index.add(db)  # добавляем векторы

# Поиск
query = np.random.randn(10, dim).astype('float32')
query /= np.linalg.norm(query, axis=1, keepdims=True)

distances, indices = index.search(query, k=5)  # топ-5 результатов
print(f"Найдено индексов: {indices.shape}")
print(f"Расстояния: {distances[0]}")

# Альтернатива: косинусовое сходство
index_cosine = faiss.IndexFlatIP(dim)  # Inner Product (для нормализованных векторов)
index_cosine.add(db)
similarities, indices = index_cosine.search(query, k=5)
print(f"Сходство: {similarities[0]}")  # от 0 до 1

2. IVF Index (Inverted File)

# IVF: разделяем пространство на кластеры
nlist = 100  # количество кластеров
index = faiss.IndexIVFFlat(faiss.IndexFlatL2(dim), dim, nlist)

# Обучение (построение кластеров)
index.train(db)

# Добавление векторов
index.add(db)

# Поиск (быстрее, но менее точно)
index.nprobe = 10  # сколько кластеров проверять (больше = точнее, но медленнее)
distances, indices = index.search(query, k=5)

print(f"Время поиска: намного быстрее!")

3. HNSW Index (Hierarchical Navigable Small World)

# HNSW: граф-базированный индекс (очень быстро, хорошее качество)
index = faiss.IndexHNSWFlat(dim, 32)  # 32 - размер соседей
index.add(db)

# Поиск (очень быстро!)
distances, indices = index.search(query, k=5)
print("HNSW: быстро и точно")

4. IVF-PQ Index (Product Quantization)

# IVF-PQ: комбинирует IVF и квантизацию (лучший баланс)
nlist = 100  # IVF кластеры
m = 8        # количество подвекторов
index = faiss.IndexIVFPQ(faiss.IndexFlatL2(dim), dim, nlist, m, 8)
# 8 - количество бит на подвектор (256 центроидов)

index.train(db)
index.add(db)

index.nprobe = 10
distances, indices = index.search(query, k=5)

print(f"Размер индекса намного меньше (квантизация)")
print(f"Скорость: очень быстро")
print(f"Качество: хорошее")

Сравнение индексов

import time

n_vectors = 1_000_000
dim = 768
db = np.random.randn(n_vectors, dim).astype('float32')
db /= np.linalg.norm(db, axis=1, keepdims=True)
query = np.random.randn(100, dim).astype('float32')
query /= np.linalg.norm(query, axis=1, keepdims=True)

indices_to_test = [
    ("Flat", faiss.IndexFlatL2(dim)),
    ("IVF", faiss.IndexIVFFlat(faiss.IndexFlatL2(dim), dim, 100)),
    ("HNSW", faiss.IndexHNSWFlat(dim, 32)),
    ("IVF-PQ", faiss.IndexIVFPQ(faiss.IndexFlatL2(dim), dim, 100, 8, 8))
]

for name, index in indices_to_test:
    if hasattr(index, 'train'):
        index.train(db)
    if hasattr(index, 'nprobe'):
        index.nprobe = 10
    
    index.add(db)
    
    start = time.time()
    distances, indices = index.search(query, k=10)
    search_time = time.time() - start
    
    print(f"{name:10s} | Time: {search_time:.4f}s | Memory: {index.ntotal} vectors")

Практический пример: Поиск похожих текстов

import faiss
from sentence_transformers import SentenceTransformer

# Загружаем модель для эмбедингов
model = SentenceTransformer('distilbert-base-multilingual-cased')

# Тексты
texts = [
    "The cat is sleeping on the mat",
    "A dog is running in the park",
    "The cat is playing with a toy",
    "The weather is sunny today",
    "I love programming in Python",
    # ... тысячи других текстов
]

# Создаём эмбединги
embeddings = model.encode(texts)
embeddings = embeddings.astype('float32')
embeddings /= (np.linalg.norm(embeddings, axis=1, keepdims=True) + 1e-8)

# Строим индекс FAISS
index = faiss.IndexIVFPQ(faiss.IndexFlatIP(embeddings.shape[1]), 
                         embeddings.shape[1], 50, 8, 8)
index.train(embeddings)
index.add(embeddings)
index.nprobe = 10

# Поиск похожих текстов
query = "The cat is sleeping"
query_embedding = model.encode([query])
query_embedding = query_embedding.astype('float32')
query_embedding /= (np.linalg.norm(query_embedding, axis=1, keepdims=True) + 1e-8)

similarities, indices = index.search(query_embedding, k=5)

print(f"\nПоиск: '{query}'")
print("Похожие тексты:")
for i, (idx, sim) in enumerate(zip(indices[0], similarities[0])):
    print(f"{i+1}. {texts[idx]} (сходство: {sim:.4f})")

GPU ускорение с FAISS

import faiss

# Проверяем доступные GPU
ngpus = faiss.get_num_gpus()
print(f"Доступно GPU: {ngpus}")

# Создаём индекс на CPU
cpu_index = faiss.IndexIVFPQ(faiss.IndexFlatL2(dim), dim, 100, 8, 8)
cpu_index.train(db)
cpu_index.add(db)

# Переносим на GPU (если доступна видеокарта)
if ngpus > 0:
    co = faiss.GpuMultipleClonerOptions()
    co.shard = True  # распределяем по GPU
    gpu_index = faiss.index_cpu_to_gpu_multiple(list(range(ngpus)), cpu_index, co)
    
    # Поиск на GPU (намного быстрее!)
    distances, indices = gpu_index.search(query, k=5)
    print("Поиск выполнен на GPU")
else:
    print("GPU не доступна, используем CPU")

Сохранение и загрузка индекса

# Сохранение
index_path = 'my_index.faiss'
faiss.write_index(index, index_path)
print(f"Индекс сохранен в {index_path}")

# Загрузка
loaded_index = faiss.read_index(index_path)
print(f"Индекс загружен, всего векторов: {loaded_index.ntotal}")

# Поиск с загруженным индексом
distances, indices = loaded_index.search(query, k=5)

Метрики FAISS

# Различные метрики расстояния

# L2 (Евклидово расстояние): лучше для нормализованных данных
index_l2 = faiss.IndexFlatL2(dim)

# Inner Product: для нормализованных векторов (косинусовое сходство)
index_ip = faiss.IndexFlatIP(dim)

# Hamming: для бинарных векторов (очень быстро)
index_hamming = faiss.IndexBinaryFlat(dim)

Практические рекомендации

1. Для малых датасетов (< 100k): используй IndexFlatL2 или IndexFlatIP
2. Для средних датасетов (100k - 10M): используй IndexIVFFlat или IndexHNSW
3. Для больших датасетов (> 10M): используй IndexIVFPQ или IndexIVFPQR
4. Для памяти critical: используй квантизацию (PQ, SQ)
5. Для точности critical: используй более высокие nprobe или HNSW
6. Для скорости critical: используй GPU ускорение
7. Всегда нормализуй векторы перед добавлением в индекс
8. Используй правильную метрику: L2 или IP в зависимости от задачи

Сложность

FAISS стала стандартом для приблизительного поиска в высокомерных пространствах и критична для масштабируемых систем рекомендаций и поиска.