Розробка 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-обробки (немає мережевого доступу)
• Вбудовування в існуючі Python-додатки
• Дуже високого throughput при batch пошуку (>10K QPS)
• Дослідницьких завдань та експериментування

FAISS небажаний для:

• Потрібні оновлення в реальному часі (FAISS погано підтримує частково оновлення)
• Кількох сервісів звертаються до одного індексу (немає мережевого API)
• Потрібна фільтрація за метаданими (немає вбудованого payload filtering)

Графіки розроблення

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