Векторный поиск для базы знаний: от прототипа до продакшена без боли

Почему ваш прототип RAG никогда не выйдет в продакшен

Вы взяли FAISS, запустили Sentence Transformers, накидали документов — и всё работает. Пока не работает. Настоящая боль начинается, когда вы пытаетесь запихнуть это в продакшен. Векторный поиск — это не про «взять и заработало». Это про детали, которые кусают вас в самый неподходящий момент.

Эмбеддинги: ваш первый и самый коварный выбор

Все начинают с all-MiniLM-L6-v2. Это нормально для прототипа. Проблема в том, что эта модель не понимает вашу предметную область. Юридические документы, медицинские термины, код программы — для неё это просто слова.

Как НЕ надо делать:

# ПЛОХО: статическая модель для всего
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('all-MiniLM-L6-v2')  # Уже 2022 год

Как надо:

# ХОРОШО: выбор модели под задачу
import torch
from transformers import AutoTokenizer, AutoModel

# Для смешанного контента (текст + код)
model_name = "BAAI/bge-m3"  # Актуально на 2026
# Или для английского: "Snowflake/snowflake-arctic-embed-l"

tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModel.from_pretrained(model_name, torch_dtype=torch.float16)

FAISS: быстрее, чем вы думали. И сложнее тоже

FAISS от Facebook — это чудо инженерии. Но его документация — это катастрофа. Вы думаете, что IndexFlatIP решит все проблемы? Подождите, пока ваша база вырастет до миллиона векторов.

1 Выбор индекса: от чего реально зависит скорость

Код, который работает в продакшене:

import faiss
import numpy as np

# Размерность ваших эмбеддингов (например, 1024 для bge-m3)
d = 1024
nlist = 100  # Количество кластеров

quantizer = faiss.IndexFlatIP(d)  # Косинусная близость
index = faiss.IndexIVFFlat(quantizer, d, nlist, faiss.METRIC_INNER_PRODUCT)

# ОБЯЗАТЕЛЬНО: тренировка на репрезентативной выборке
training_vectors = np.random.rand(10000, d).astype('float32')
index.train(training_vectors)

# Только после тренировки добавляем данные
index.add(your_vectors)

Порог релевантности: магическое число, которое не работает

Вы видели в каждом туториале: if similarity > 0.7: return result. Это ложь. Абсолютная. Порог зависит от:

• Размера базы знаний
• Качества эмбеддингов
• Семантической плотности домена
• Даже от времени суток (серьёзно, если у вас динамический контент)

Вот что происходит на самом деле:

# НАИВНЫЙ ПОДХОД (так делают все и потом плачут)
def naive_search(query_embedding, index, threshold=0.7):
    distances, indices = index.search(query_embedding, k=10)
    results = []
    for dist, idx in zip(distances[0], indices[0]):
        if dist > threshold:  # Магическое число!
            results.append((idx, dist))
    return results  # Часто возвращает пустой список

А вот адаптивный подход:

def adaptive_search(query_embedding, index, base_threshold=0.5, min_results=3):
    """Ищет минимум min_results, но не ниже base_threshold"""
    distances, indices = index.search(query_embedding, k=20)
    
    # Динамический порог: медиана + отклонение
    if len(distances[0]) > 5:
        median_dist = np.median(distances[0][:5])
        dynamic_threshold = max(base_threshold, median_dist * 0.8)
    else:
        dynamic_threshold = base_threshold
    
    results = []
    for dist, idx in zip(distances[0], indices[0]):
        if dist >= dynamic_threshold or len(results) < min_results:
            results.append((idx, dist))
        if len(results) >= 10:  # Лимит результатов
            break
            
    return results

Чанкинг: искусство резать документы без потери смысла

Разбивать по 512 токенов — это как резать пиццу линейкой. Иногда предложение разрывается посередине, и смысл теряется. Особенно больно с таблицами, кодом и многоуровневыми списками.

Проблема, о которой молчат:

# ПЛОХО: тупое разбиение по символам
def bad_chunking(text, chunk_size=500):
    return [text[i:i+chunk_size] for i in range(0, len(text), chunk_size)]
# Результат: "...контракт должен быть подписан\nдо 31 декабря 202..."
# Смысл потерян: дата оторвана от контекста

Решение с семантическим чанкингом:

from langchain.text_splitter import RecursiveCharacterTextSplitter

def smart_chunking(text, chunk_size=1000, chunk_overlap=200):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=['\n\n', '\n', '. ', ', ', ' ', '']  # Иерархия разделителей
    )
    return splitter.split_text(text)

Продакшен-архитектура: где живет ваш векторный поиск

Локальный скрипт — это не архитектура. В продакшене нужно думать о:

1. Обновлении индекса без даунтайма
2. Мониторинге качества поиска
3. Бэкапах векторов и метаданных
4. Масштабировании при росте данных

Типичная ошибка — хранить всё в памяти. А потом сервер падает, и вы теряете все индексы.

# ПРОДАКШЕН-ГОТОВОЕ РЕШЕНИЕ
import pickle
import json
from datetime import datetime

class VectorSearchProduction:
    def __init__(self, index_path, metadata_path):
        self.index_path = index_path
        self.metadata_path = metadata_path
        self.index = None
        self.metadata = {}
        self.load()
    
    def load(self):
        """Загрузка с проверкой целостности"""
        try:
            self.index = faiss.read_index(self.index_path)
            with open(self.metadata_path, 'rb') as f:
                self.metadata = pickle.load(f)
            print(f"Загружен индекс с {len(self.metadata)} документами")
        except Exception as e:
            print(f"Ошибка загрузки: {e}")
            self.index = faiss.IndexFlatL2(1024)  # Фолбэк
            
    def save(self):
        """Сохранение с атомарной записью"""
        temp_index = f"{self.index_path}.tmp"
        temp_meta = f"{self.metadata_path}.tmp"
        
        faiss.write_index(self.index, temp_index)
        with open(temp_meta, 'wb') as f:
            pickle.dump(self.metadata, f)
        
        # Атомарная замена
        import os
        os.replace(temp_index, self.index_path)
        os.replace(temp_meta, self.metadata_path)
        
    def add_document(self, doc_id, embedding, metadata):
        """Добавление с версионированием"""
        self.index.add(embedding.reshape(1, -1))
        self.metadata[doc_id] = {
            **metadata,
            'added_at': datetime.utcnow().isoformat(),
            'version': 1
        }
        self.save()

Гибридный поиск: когда векторов недостаточно

Векторный поиск находит семантически похожие документы. Но он не понимает, что «2026 год» — это дата, а «Python 3.12» — это версия языка. Для этого нужен ключевой поиск.

Объединяем лучшее из двух миров:

def hybrid_search(query, vector_index, keyword_index, alpha=0.7):
    """alpha = вес векторного поиска (0.7), (1-alpha) = вес ключевого"""
    
    # Векторный поиск
    query_embedding = model.encode([query])[0]
    vector_results = vector_index.search(query_embedding, k=20)
    
    # Ключевой поиск (упрощённо)
    keyword_results = keyword_index.search(query, k=20)
    
    # Слияние результатов ( Reciprocal Rank Fusion )
    fused_scores = {}
    
    for rank, (doc_id, score) in enumerate(vector_results):
        fused_scores[doc_id] = fused_scores.get(doc_id, 0) + alpha * (1 / (rank + 1))
        
    for rank, (doc_id, score) in enumerate(keyword_results):
        fused_scores[doc_id] = fused_scores.get(doc_id, 0) + (1-alpha) * (1 / (rank + 1))
    
    # Сортировка по комбинированному скору
    sorted_results = sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
    return sorted_results[:10]

Это именно та архитектура, которая описана в статье «Когда SQL и векторный поиск дерутся за ваши данные». Там разобраны все подводные камни.

Мониторинг: как понять, что ваш поиск сломался

Без мониторинга вы узнаете о проблеме от пользователей. А это уже поздно. Отслеживайте:

• Среднюю косинусную близость возвращаемых результатов
• Процент запросов, возвращающих пустой результат
• Время ответа (p95, p99)
• Хитрейт кэша эмбеддингов

# Простейший мониторинг
import time
from collections import deque

class SearchMonitor:
    def __init__(self, window_size=1000):
        self.latencies = deque(maxlen=window_size)
        self.empty_results = deque(maxlen=window_size)
        self.scores = deque(maxlen=window_size)
    
    def record_search(self, latency_ms, results_count, avg_score):
        self.latencies.append(latency_ms)
        self.empty_results.append(1 if results_count == 0 else 0)
        self.scores.append(avg_score)
        
    def get_metrics(self):
        return {
            'p95_latency': np.percentile(list(self.latencies), 95) if self.latencies else 0,
            'empty_rate': sum(self.empty_results) / len(self.empty_results) if self.empty_results else 0,
            'avg_score': np.mean(list(self.scores)) if self.scores else 0
        }

Типичные ошибки, которые взорвут ваш продакшен

Что дальше? Когда векторного поиска недостаточно

Ваша база знаний выросла до миллиона документов. Качество поиска падает. Знакомо? Это называется «проклятием размерности».

Следующий шаг — графы знаний. Они добавляют связи между документами, которые не уловить эмбеддингами. Как это работает на практике, смотрите в гайде по Knowledge Graph.

А если вы хотите глубоко разобраться, почему эмбеддинги иногда не находят очевидные документы, почитайте про математический потолок RAG.

И последнее: если вы устали отлаживать FAISS вслепую, посмотрите VectorDBZ — инструмент для визуальной отладки векторных баз. Иногда проще увидеть проблему, чем пытаться её вычислить.