Разработка RAG с FAISS для локального векторного поиска

Разработка RAG с FAISS для локального векторного поиска

FAISS (Facebook AI Similarity Search) — библиотека векторного поиска от Meta, не являющаяся базой данных в традиционном смысле. FAISS — это высокопроизводительный движок поиска по векторам, работающий в памяти или на диске, без сетевого взаимодействия. Идеален для встраивания в приложение напрямую, offline-сценариев и ситуаций, где внешний сервис неприемлем.

Типы индексов FAISS

Создание индекса и индексация

import faiss
import numpy as np
import pickle
from openai import OpenAI

openai_client = OpenAI()

def build_faiss_index(texts: list[str], dimension: int = 1536) -> tuple:
    """Создаёт FAISS индекс и соответствующий список текстов"""

    # Получаем embeddings батчами
    embeddings = []
    batch_size = 100
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        response = openai_client.embeddings.create(
            model="text-embedding-3-small",
            input=batch,
        )
        batch_embeddings = [e.embedding for e in response.data]
        embeddings.extend(batch_embeddings)

    # Конвертируем в numpy float32
    vectors = np.array(embeddings, dtype=np.float32)

    # Нормализуем для cosine similarity (через inner product)
    faiss.normalize_L2(vectors)

    # Создаём HNSW индекс
    index = faiss.IndexHNSWFlat(dimension, 16)  # M=16
    index.hnsw.efConstruction = 200
    index.add(vectors)

    return index, texts

# Сохранение на диск
def save_index(index, texts, path_prefix: str):
    faiss.write_index(index, f"{path_prefix}.index")
    with open(f"{path_prefix}_texts.pkl", "wb") as f:
        pickle.dump(texts, f)

# Загрузка
def load_index(path_prefix: str) -> tuple:
    index = faiss.read_index(f"{path_prefix}.index")
    with open(f"{path_prefix}_texts.pkl", "rb") as f:
        texts = pickle.load(f)
    return index, texts

Поиск и RAG-ответ

def faiss_rag_answer(
    question: str,
    index: faiss.Index,
    texts: list[str],
    top_k: int = 5
) -> str:
    # Embedding вопроса
    query_embedding = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=question,
    ).data[0].embedding

    query_vector = np.array([query_embedding], dtype=np.float32)
    faiss.normalize_L2(query_vector)

    # Поиск
    distances, indices = index.search(query_vector, top_k)

    # Извлечение текстов
    context_texts = [texts[i] for i in indices[0] if i >= 0]
    context = "\n\n---\n\n".join(context_texts)

    # Генерация ответа
    response = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Отвечай строго на основе предоставленного контекста."},
            {"role": "user", "content": f"Контекст:\n{context}\n\nВопрос: {question}"}
        ],
        temperature=0,
    )
    return response.choices[0].message.content

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

FAISS поддерживает перенос индекса на GPU для ускорения поиска в 10–100×:

# Перенос на GPU
res = faiss.StandardGpuResources()
gpu_index = faiss.index_cpu_to_gpu(res, 0, index)  # GPU 0

# Поиск на GPU
distances, indices = gpu_index.search(query_vectors, top_k)

Когда выбирать FAISS

FAISS предпочтителен:

• Offline-обработка (нет сетевого доступа)
• Embedding в существующее Python-приложение
• Очень высокий throughput при batch поиске (>10K QPS)
• Исследовательские задачи и экспериментирование

FAISS нежелателен:

• Нужны обновления в реальном времени (FAISS плохо поддерживает partial updates)
• Несколько сервисов обращаются к одному индексу (нет сетевого API)
• Нужна фильтрация по метаданным (нет встроенного payload filtering)

Сроки

• Разработка FAISS RAG pipeline: 3–7 дней
• Оптимизация индекса и тестирование: 2–4 дня
• Итого: 1–2 недели