Продвинутые индексы FAISS для больших данных: HNSW для скорости, IVF для экономии памяти. Сохранение, загрузка, метаданные.
Продвинутые индексы FAISS для больших данных: HNSW для скорости, IVF для экономии памяти. Сохранение, загрузка, метаданные.
За 60 минут вы:
IndexFlatIP (полный перебор) имеет сложность O(n):
| Векторов | Время поиска | Память |
|---|---|---|
| 1 000 | ~0.1 мс | 1.5 МБ |
| 10 000 | ~1 мс | 15 МБ |
| 100 000 | ~10 мс | 150 МБ |
| 1 000 000 | ~100 мс | 1.5 ГБ |
| 10 000 000 | ~1 сек | 15 ГБ |
Для продакшена нужны ANN-алгоритмы (Approximate Nearest Neighbors):
HNSW строит многоуровневый граф связей между векторами:
Уровень 3 (редкие связи): A ───── B
│ │
Уровень 2: A ── C ─ B
│ │ │
Уровень 1 (частые связи): A ─ D ─ B
│ ╱ │ ╲ │
Уровень 0 (все векторы): A─E─F─G─B
Принцип поиска:
Результат: Логарифмическая сложность O(log n) вместо O(n).
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer
# Параметры
dimension = 384
M = 32 # Количество связей на узел
# Создание HNSW индекса
# IndexHNSWFlat: HNSW с плоским хранением векторов
index = faiss.IndexHNSWFlat(dimension, M)
# Настройка параметров
index.hnsw.efConstruction = 200 # Точность при построении (выше = точнее, медленнее)
# Подготовка данных
model = SentenceTransformer('all-MiniLM-L6-v2')
documents = ["документ"] * 10000 # 10K документов
embeddings = model.encode(documents).astype('float32')
faiss.normalize_L2(embeddings)
# Добавление векторов
print("Добавление векторов...")
index.add(embeddings)
print(f"Векторов в индексе: {index.ntotal}")| Параметр | Описание | Рекомендации |
|---|---|---|
| M | Количество связей на узел | 16–64 (32 — баланс) |
| efConstruction | Точность при построении | 100–400 (200 — стандарт) |
| efSearch | Точность при поиске | 10–100 (40 — стандарт) |
import time
def benchmark_hnsw(M, efConstruction, efSearch, embeddings, query_embedding):
"""Тестирование параметров HNSW."""
# Создание индекса
index = faiss.IndexHNSWFlat(384, M)
index.hnsw.efConstruction = efConstruction
index.hnsw.efSearch = efSearch
# Добавление
start = time.time()
index.add(embeddings)
add_time = time.time() - start
# Поиск
start = time.time()
for _ in range(100):
index.search(query_embedding, k=10)
search_time = (time.time() - start) / 100
return {
"M": M,
"efConstruction": efConstruction,
"efSearch": efSearch,
"add_time": add_time,
"search_time_ms": search_time * 1000
}
# Тестирование
embeddings = np.random.rand(10000, 384).astype('float32')
faiss.normalize_L2(embeddings)
query = np.random.rand(1, 384).astype('float32')
faiss.normalize_L2(query)
configs = [
{"M": 16, "efConstruction": 100, "efSearch": 20},
{"M": 32, "efConstruction": 200, "efSearch": 40},
{"M": 64, "efConstruction": 400, "efSearch": 80},
]
print("Конфигурация HNSW:\n")
for cfg in configs:
result = benchmark_hnsw(cfg["M"], cfg["efConstruction"], cfg["efSearch"], embeddings, query)
print(f"M={result['M']}, efC={result['efConstruction']}, efS={result['efSearch']}")
print(f" Добавление: {result['add_time']:.2f} сек")
print(f" Поиск: {result['search_time_ms']:.2f} мс\n")Пример вывода:
Конфигурация HNSW:
M=16, efC=100, efS=20
Добавление: 2.34 сек
Поиск: 0.45 мс
M=32, efC=200, efS=40
Добавление: 4.12 сек
Поиск: 0.78 мс
M=64, efC=400, efS=80
Добавление: 8.56 сек
Поиск: 1.23 мс
# Настройка efSearch (точность поиска)
index.hnsw.efSearch = 40 # Стандарт
distances, indices = index.search(query_embedding, k=10)
# Для высокой точности
index.hnsw.efSearch = 100
distances, indices = index.search(query_embedding, k=10)
# Для максимальной скорости
index.hnsw.efSearch = 10
distances, indices = index.search(query_embedding, k=10)IVF разделяет векторы на кластеры (воронки):
1. Кластеризация (обучение):
┌────────┐ ┌────────┐ ┌────────┐
│Воронка 1│ │Воронка 2│ │Воронка 3│
│ ● ● ● │ │ ● ● │ │ ● ● │
└────────┘ └────────┘ └────────┘
2. Поиск:
- Найти ближайшую воронку к запросу
- Искать только внутри воронки (и соседних)
Преимущество: Ищем не среди всех n векторов, а только в одной из k воронок.
import faiss
import numpy as np
dimension = 384
n_clusters = 100 # Количество воронок (кластеров)
# Шаг 1: Создание квантователя (кластеризатор)
quantizer = faiss.IndexFlatL2(dimension)
# Шаг 2: Создание IVF индекса
# IndexIVFFlat: IVF с плоским хранением векторов в воронках
index = faiss.IndexIVFFlat(quantizer, dimension, n_clusters, faiss.METRIC_L2)
# Шаг 3: Обучение на представительной выборке
# Требуется перед добавлением векторов!
train_size = min(10000, len(embeddings))
train_data = embeddings[:train_size]
print("Обучение IVF...")
index.train(train_data)
print("Обучение завершено")
# Шаг 4: Добавление векторов
index.add(embeddings)
print(f"Векторов в индексе: {index.ntotal}")| Параметр | Описание | Рекомендации |
|---|---|---|
| n_clusters | Количество воронок | √n – 4√n (для 1M векторов: 1000–4000) |
| nprobe | Количество воронок для поиска | 1–100 (10–20 — баланс) |
# По умолчанию nprobe=1 (только ближайшая воронка)
index.nprobe = 1
# Для высокой точности: искать в нескольких воронках
index.nprobe = 10 # Стандарт
distances, indices = index.search(query_embedding, k=10)
# Для максимальной точности
index.nprobe = 50
distances, indices = index.search(query_embedding, k=10)Влияние nprobe:
| nprobe | Точность | Время поиска |
|---|---|---|
| 1 | 80–85% | 0.1 мс |
| 10 | 90–95% | 0.5 мс |
| 50 | 95–98% | 2 мс |
| 100 | 98–99% | 4 мс |
PQ сжимает векторы, разделяя их на подсегменты:
Исходный вектор (384 измерения, float32):
[0.12, -0.45, 0.78, ..., 0.33] # 384 × 4 байта = 1536 байт
Разбиение на 8 подсегментов по 48 измерений:
[0.12, ..., 0.78] | [-0.45, ..., 0.12] | ... | [..., 0.33]
Квантование каждого подсегмента (256 центров):
[42] | [17] | [203] | [89] | [156] | [34] | [201] | [78] # 8 байт
Сжатие: 1536 байт → 8 байт (в 192 раза!)
import faiss
dimension = 384
n_clusters = 1024 # Количество воронок
m = 8 # Количество подсегментов (для сжатия)
nbits = 8 # Бит на код (256 центров квантования)
# Создание квантователя
quantizer = faiss.IndexFlatL2(dimension)
# Создание IVF-PQ индекса
index = faiss.IndexIVFPQ(quantizer, dimension, n_clusters, m, nbits)
# Обучение
index.train(train_data)
# Добавление
index.add(embeddings)
print(f"Векторов: {index.ntotal}")
print(f"Размер сжатия: {dimension * 4 / m:.0f}x")Когда использовать IVF-PQ:
| Индекс | Точность | Скорость | Память | Когда использовать |
|---|---|---|---|---|
| IndexFlatIP | 100% | Медленно | Большая | <10K векторов, прототипы |
| IndexHNSW | 95–99% | Очень быстро | Средняя | Продакшен, баланс |
| IndexIVFFlat | 90–95% | Быстро | Малая | 100K–1M векторов |
| IndexIVFPQ | 85–95% | Быстро | Очень малая | >1M векторов, сжатие |
# Сохранение
faiss.write_index(index, "hnsw_index.faiss")
# Загрузка
index = faiss.read_index("hnsw_index.faiss")# Сохранение
faiss.write_index(index, "ivf_index.faiss")
# Загрузка
index = faiss.read_index("ivf_index.faiss")
# Проверка параметров после загрузки
print(f"nprobe: {index.nprobe}")# Сохранение
faiss.write_index(index, "ivfpq_index.faiss")
# Загрузка
index = faiss.read_index("ivfpq_index.faiss")FAISS не хранит метаданные. Создайте обёртку:
import pickle
from pathlib import Path
class FAISSProduction:
"""FAISS для продакшена с метаданными."""
def __init__(
self,
dimension: int,
index_type: str = "hnsw",
**kwargs
):
self.dimension = dimension
self.metadata = []
if index_type == "hnsw":
M = kwargs.get("M", 32)
efConstruction = kwargs.get("efConstruction", 200)
self.index = faiss.IndexHNSWFlat(dimension, M)
self.index.hnsw.efConstruction = efConstruction
elif index_type == "ivf":
n_clusters = kwargs.get("n_clusters", 100)
quantizer = faiss.IndexFlatL2(dimension)
self.index = faiss.IndexIVFFlat(
quantizer, dimension, n_clusters, faiss.METRIC_L2
)
elif index_type == "ivfpq":
n_clusters = kwargs.get("n_clusters", 1024)
m = kwargs.get("m", 8)
nbits = kwargs.get("nbits", 8)
quantizer = faiss.IndexFlatL2(dimension)
self.index = faiss.IndexIVFPQ(
quantizer, dimension, n_clusters, m, nbits
)
self.index_type = index_type
def train(self, train_data: np.ndarray):
"""Обучение для IVF индексов."""
if self.index_type in ["ivf", "ivfpq"]:
self.index.train(train_data)
print(f"Обучение {self.index_type} завершено")
def add(self, embeddings: np.ndarray, metadata: list = None):
"""Добавить векторы с метаданными."""
self.index.add(embeddings)
if metadata:
self.metadata.extend(metadata)
else:
self.metadata.extend([{} for _ in range(len(embeddings))])
def search(self, query_embedding: np.ndarray, k: int = 10):
"""Поиск с метаданными."""
# Настройка для HNSW
if self.index_type == "hnsw":
self.index.hnsw.efSearch = 40
# Настройка для IVF
if self.index_type == "ivf":
self.index.nprobe = 10
distances, indices = self.index.search(query_embedding, k)
results = []
for idx, dist in zip(indices[0], distances[0]):
if idx < len(self.metadata):
results.append({
"id": int(idx),
"score": float(dist),
"metadata": self.metadata[idx]
})
return results
def save(self, path: str):
"""Сохранить индекс и метаданные."""
faiss.write_index(self.index, f"{path}.faiss")
with open(f"{path}.metadata.pkl", "wb") as f:
pickle.dump(self.metadata, f)
with open(f"{path}.config.json", "w") as f:
import json
json.dump({
"dimension": self.dimension,
"index_type": self.index_type
}, f)
@classmethod
def load(cls, path: str):
"""Загрузить индекс и метаданные."""
import json
with open(f"{path}.config.json") as f:
config = json.load(f)
instance = cls(config["dimension"], config["index_type"])
instance.index = faiss.read_index(f"{path}.faiss")
with open(f"{path}.metadata.pkl", "rb") as f:
instance.metadata = pickle.load(f)
return instance
# Использование
db = FAISSProduction(
dimension=384,
index_type="hnsw",
M=32,
efConstruction=200
)
# Добавление
db.add(embeddings, [
{"text": "документ 1", "category": "web"},
{"text": "документ 2", "category": "api"},
])
# Поиск
results = db.search(query_embedding, k=5)
# Сохранение
db.save("production_index")
# Загрузка
db_loaded = FAISSProduction.load("production_index")Напишите функцию для выбора индекса на основе данных:
def recommend_index(n_vectors: int, memory_limit_mb: int = None) -> dict:
"""
Рекомендация индекса FAISS на основе данных.
Args:
n_vectors: Количество векторов
memory_limit_mb: Ограничение памяти (опционально)
Returns:
Словарь с рекомендациями
"""
dimension = 384 # Стандартная размерность
bytes_per_vector = 4 * dimension # float32
# Оценка памяти
total_memory_mb = (n_vectors * bytes_per_vector) / (1024 * 1024)
if n_vectors < 10_000:
return {
"index_type": "IndexFlatIP",
"reason": "Мало данных — точный поиск без потерь",
"memory_mb": total_memory_mb,
"params": {}
}
elif n_vectors < 1_000_000:
return {
"index_type": "IndexHNSW",
"reason": "Баланс скорости и точности для средних данных",
"memory_mb": total_memory_mb * 1.5, # HNSW требует больше памяти
"params": {
"M": 32,
"efConstruction": 200,
"efSearch": 40
}
}
elif n_vectors < 10_000_000:
if memory_limit_mb and total_memory_mb > memory_limit_mb:
return {
"index_type": "IndexIVFPQ",
"reason": "Сжатие для больших данных с ограничением памяти",
"memory_mb": total_memory_mb / 16, # PQ сжимает ~16x
"params": {
"n_clusters": 1024,
"m": 8,
"nbits": 8
}
}
else:
return {
"index_type": "IndexIVFFlat",
"reason": "Большие данные без жёсткого ограничения памяти",
"memory_mb": total_memory_mb,
"params": {
"n_clusters": int(np.sqrt(n_vectors) * 2),
"nprobe": 20
}
}
else:
return {
"index_type": "IndexIVFPQ",
"reason": "Очень большие данные — требуется сжатие",
"memory_mb": total_memory_mb / 16,
"params": {
"n_clusters": 4096,
"m": 16,
"nbits": 8
}
}
# Тестирование
configs = [
(5_000, None),
(100_000, None),
(1_000_000, 500),
(10_000_000, 1000),
]
for n_vec, mem_limit in configs:
rec = recommend_index(n_vec, mem_limit)
print(f"\n{n_vec:,} векторов (память: {mem_limit or 'нет'}):")
print(f" Индекс: {rec['index_type']}")
print(f" Причина: {rec['reason']}")
print(f" Память: {rec['memory_mb']:.1f} МБ")
print(f" Параметры: {rec['params']}")# ❌ Ошибка: добавление без обучения
index = faiss.IndexIVFFlat(quantizer, dimension, n_clusters)
index.add(embeddings) # Ошибка!
# ✅ Правильно: сначала обучение
index.train(train_data)
index.add(embeddings)# ❌ Слишком мало кластеров для больших данных
n_clusters = 10 # Для 1M векторов!
# ✅ Правило: √n – 4√n
n_clusters = int(np.sqrt(1_000_000)) # 1000# ❌ Поиск с параметрами по умолчанию
index.hnsw.efSearch = 8 # По умолчанию (низкая точность)
distances, indices = index.search(query_embedding, k)
# ✅ Настройка перед поиском
index.hnsw.efSearch = 40 # Стандарт
distances, indices = index.search(query_embedding, k)# ❌ HNSW с ненормализованными векторами
index = faiss.IndexHNSWFlat(dimension, M)
index.add(embeddings) # Без нормализации!
# ✅ Нормализация для косинусного сходства
faiss.normalize_L2(embeddings)
index.add(embeddings)Теперь вы умеете работать с продвинутыми индексами FAISS. Следующий шаг — Chroma:
hnsw_benchmark.py с тестированием HNSWГотовы? Открывайте chroma!
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.