Начинаем работать с FAISS: создание индекса, добавление векторов, поиск. Простые примеры для понимания базовых принципов.
Начинаем работать с FAISS: создание индекса, добавление векторов, поиск. Простые примеры для понимания базовых принципов.
За 45 минут вы:
FAISS (Facebook AI Similarity Search) — библиотека от Meta для эффективного поиска похожих векторов.
Преимущества:
Недостатки:
Когда использовать:
pip install faiss-cpu # CPU-версия
# или
pip install faiss-gpu # GPU-версия (требует CUDA)Проверка установки:
import faiss
print(faiss.__version__)Самый простой индекс — полный перебор. Точно, но медленно для больших данных.
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer
# Подготовка данных
model = SentenceTransformer('all-MiniLM-L6-v2')
documents = [
"Python — язык программирования общего назначения",
"Питон — это змея из семейства удавов",
"Django — веб-фреймворк на Python для быстрой разработки",
"Flask — микрофреймворк для создания веб-приложений",
"FastAPI — современный фреймворк для создания API"
]
# Генерация эмбеддингов
embeddings = model.encode(documents).astype('float32')
dimension = embeddings.shape[1] # 384
print(f"Размерность: {dimension}")
print(f"Количество векторов: {len(embeddings)}")
# Создание индекса (косинусное сходство)
index = faiss.IndexFlatIP(dimension) # Inner Product
# Нормализация для косинусного сходства
faiss.normalize_L2(embeddings)
# Добавление векторов
index.add(embeddings)
print(f"Векторов в индексе: {index.ntotal}")# Запрос
query = "веб-разработка на Python"
query_embedding = model.encode([query]).astype('float32')
# Нормализация запроса (обязательно для косинусного сходства!)
faiss.normalize_L2(query_embedding)
# Поиск: k=2 ближайших документа
k = 2
distances, indices = index.search(query_embedding, k)
print(f"\nЗапрос: {query}")
print(f"Индексы: {indices[0]}")
print(f"Сходство: {distances[0]}")
print("\nРезультаты:")
for idx, score in zip(indices[0], distances[0]):
print(f" [{score:.3f}] {documents[idx]}")Вывод:
Запрос: веб-разработка на Python
Индексы: [2 4]
Сходство: [0.823 0.756]
Результаты:
[0.823] Django — веб-фреймворк на Python для быстрой разработки
[0.756] FastAPI — современный фреймворк для создания API
distances, indices = index.search(query_embedding, k)
# distances: массив сходства (или расстояния)
# Форма: (n_queries, k)
print(distances.shape) # (1, 2)
# indices: индексы найденных векторов в базе
# Форма: (n_queries, k)
print(indices.shape) # (1, 2)def interpret_score(score):
"""Интерпретация косинусного сходства."""
if score >= 0.9:
return "Практически идентичные"
elif score >= 0.7:
return "Очень похожие"
elif score >= 0.5:
return "Умеренное сходство"
elif score >= 0.3:
return "Слабое сходство"
else:
return "Почти не связаны"
score = 0.823
print(f"{score:.3f} → {interpret_score(score)}")
# 0.823 → Очень похожиеFAISS предоставляет разные индексы для разных задач:
| Индекс | Тип | Точность | Скорость | Память | Когда использовать |
|---|---|---|---|---|---|
| IndexFlatIP | Точный | 100% | Медленно | Большая | Малые данные (<10K векторов), косинусное сходство |
| IndexFlatL2 | Точный | 100% | Медленно | Большая | Когда нужно L2 расстояние |
| IndexHNSW | ANN | 95–99% | Очень быстро | Средняя | Продакшен, баланс скорости/точности |
| IndexIVFFlat | ANN | 90–95% | Быстро | Малая | Большие данные (>100K) |
| IndexIVFPQ | ANN | 85–95% | Быстро | Очень малая | Ограниченная память, сжатие |
Используется для косинусного сходства (с нормализацией):
index = faiss.IndexFlatIP(dimension)
faiss.normalize_L2(embeddings)
index.add(embeddings)Для евклидова расстояния:
index = faiss.IndexFlatL2(dimension)
# Нормализация не нужна
index.add(embeddings)
distances, indices = index.search(query_embedding, k)
# distances теперь содержат L2-расстояние (меньше = ближе)# Сохранение индекса
faiss.write_index(index, "my_index.faiss")
print(f"Индекс сохранён: {faiss.read_index('my_index.faiss').ntotal} векторов")# Загрузка индекса
loaded_index = faiss.read_index("my_index.faiss")
print(f"Загружено векторов: {loaded_index.ntotal}")
# Поиск в загруженном индексе
distances, indices = loaded_index.search(query_embedding, k)FAISS хранит только векторы. Для метаданных используйте pickle:
import pickle
# Метаданные
metadata = {
"documents": documents,
"model_name": "all-MiniLM-L6-v2",
"created_at": "2026-03-17"
}
# Сохранение индекса и метаданных
faiss.write_index(index, "my_index.faiss")
with open("metadata.pkl", "wb") as f:
pickle.dump(metadata, f)
# Загрузка
loaded_index = faiss.read_index("my_index.faiss")
with open("metadata.pkl", "rb") as f:
loaded_metadata = pickle.load(f)
print(f"Документов: {len(loaded_metadata['documents'])}")Создадим обёртку для хранения метаданных:
import pickle
from pathlib import Path
class FAISSWithMetadata:
"""FAISS с хранением метаданных документов."""
def __init__(self, dimension: int):
self.index = faiss.IndexFlatIP(dimension)
self.metadata = [] # Список метаданных по индексам
def add(self, embeddings: np.ndarray, metadata: list):
"""
Добавить векторы с метаданными.
Args:
embeddings: Матрица эмбеддингов (n, dimension)
metadata: Список словарей с метаданными
"""
self.index.add(embeddings)
self.metadata.extend(metadata)
def search(self, query_embedding: np.ndarray, k: int = 5):
"""
Поиск с возвратом метаданных.
Args:
query_embedding: Вектор запроса
k: Количество результатов
Returns:
Список словарей с результатами
"""
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)
@classmethod
def load(cls, path: str, dimension: int):
"""Загрузить индекс и метаданные."""
instance = cls(dimension)
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 = FAISSWithMetadata(dimension=384)
db.add(
embeddings,
[
{"text": "Python — язык программирования", "category": "programming", "author": "Alice"},
{"text": "Питон — это змея", "category": "animals", "author": "Bob"},
{"text": "Django — веб-фреймворк", "category": "web", "author": "Alice"},
{"text": "Flask — микрофреймворк", "category": "web", "author": "Charlie"},
{"text": "FastAPI — современный фреймворк", "category": "web", "author": "Alice"}
]
)
# Поиск
query = "веб-разработка"
query_embedding = model.encode([query]).astype('float32')
faiss.normalize_L2(query_embedding)
results = db.search(query_embedding, k=3)
print(f"Запрос: {query}\n")
for r in results:
print(f"Score: {r['score']:.3f}")
print(f" Текст: {r['metadata']['text']}")
print(f" Категория: {r['metadata']['category']}")
print(f" Автор: {r['metadata']['author']}")
print()FAISS поддерживает пакетный поиск:
# Несколько запросов
queries = [
"язык программирования Python",
"веб-фреймворк",
"змеи и рептилии"
]
# Генерация эмбеддингов для запросов
query_embeddings = model.encode(queries).astype('float32')
faiss.normalize_L2(query_embeddings)
# Поиск для всех запросов сразу
k = 2
distances, indices = index.search(query_embeddings, k)
print(f"Форма результатов: {distances.shape}") # (3, 2) — 3 запроса, 2 результата
for i, query in enumerate(queries):
print(f"\nЗапрос {i+1}: {query}")
for idx, score in zip(indices[i], distances[i]):
print(f" [{score:.3f}] {documents[idx]}")Создадим полноценный прототип:
# search_prototype.py
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer
import pickle
from pathlib import Path
class SearchPrototype:
"""Прототип поисковика на FAISS."""
def __init__(self, model_name: str = 'all-MiniLM-L6-v2'):
self.model = SentenceTransformer(model_name)
self.index = None
self.documents = []
self.dimension = self.model.get_sentence_embedding_dimension()
def add_documents(self, documents: list[str], metadata: list[dict] = None):
"""Добавить документы."""
self.documents.extend(documents)
# Генерация эмбеддингов
embeddings = self.model.encode(documents).astype('float32')
faiss.normalize_L2(embeddings)
# Создание или обновление индекса
if self.index is None:
self.index = faiss.IndexFlatIP(self.dimension)
self.index.add(embeddings)
# Метаданные
if metadata:
if not hasattr(self, 'metadata'):
self.metadata = []
self.metadata.extend(metadata)
else:
if not hasattr(self, 'metadata'):
self.metadata = []
self.metadata.extend([{"text": doc} for doc in documents])
def search(self, query: str, k: int = 5) -> list[dict]:
"""Поиск документов."""
query_embedding = self.model.encode([query]).astype('float32')
faiss.normalize_L2(query_embedding)
distances, indices = self.index.search(query_embedding, k)
results = []
for idx, score in zip(indices[0], distances[0]):
if idx < len(self.documents):
results.append({
"score": float(score),
"text": self.documents[idx],
"metadata": self.metadata[idx] if hasattr(self, 'metadata') else {}
})
return results
def save(self, path: str):
"""Сохранить поисковик."""
faiss.write_index(self.index, f"{path}.faiss")
with open(f"{path}.data.pkl", "wb") as f:
pickle.dump({
"documents": self.documents,
"metadata": getattr(self, 'metadata', [])
}, f)
@classmethod
def load(cls, path: str, model_name: str = 'all-MiniLM-L6-v2'):
"""Загрузить поисковик."""
instance = cls(model_name)
instance.index = faiss.read_index(f"{path}.faiss")
with open(f"{path}.data.pkl", "rb") as f:
data = pickle.load(f)
instance.documents = data["documents"]
instance.metadata = data["metadata"]
return instance
# Использование
if __name__ == "__main__":
search = SearchPrototype()
# Добавление документов
search.add_documents([
"Python — язык программирования общего назначения",
"Питон — это змея из семейства удавов",
"Django — веб-фреймворк на Python",
"Flask — микрофреймворк для веб-приложений",
"FastAPI — современный фреймворк для API"
])
# Поиск
query = "веб-разработка на Python"
results = search.search(query, k=3)
print(f"Запрос: {query}\n")
for r in results:
print(f"[{r['score']:.3f}] {r['text']}")# ❌
query_embedding = model.encode([query])
distances, indices = index.search(query_embedding, k)
# ✅
query_embedding = model.encode([query])
faiss.normalize_L2(query_embedding) # Обязательно!
distances, indices = index.search(query_embedding, k)# ❌ Медленно
for embedding in embeddings:
index.add(embedding.reshape(1, -1))
# ✅ Быстро
index.add(embeddings) # Одна операция с матрицей# ❌ Нельзя сравнивать результаты индексов с разными метриками
index_ip = faiss.IndexFlatIP(dimension) # Косинусное
index_l2 = faiss.IndexFlatL2(dimension) # Евклидово
# Результаты search() будут incomparable# ❌ Ошибка: размерность не совпадает
embeddings_384 = model.encode(texts) # 384
index = faiss.IndexFlatIP(768) # 768!
index.add(embeddings_384) # Ошибка!
# ✅
dimension = embeddings.shape[1]
index = faiss.IndexFlatIP(dimension)Теперь вы умеете работать с базовыми индексами FAISS. Следующий шаг — FAISS для продакшена:
my_search.py с прототипом поисковикаГотовы? Открывайте faiss_production!
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.