Строим RAG вручную: как это работает под капотом

Имея некоторый опыт в построении классических ML и CV-проектов, я решил разобраться в NLP (Natural Language Processing) и собрать свою RAG-систему без использования сторонних RAG-фреймворков (LangChain, LlamaIndex и т.п.). Моя цель – понять, как на самом деле работает RAG под капотом: от разбиения текста до генерации ответа.

Проблема, которую решает RAG

Готовые LLM модели имеют несколько недостатков:

• модель не имеет доступа к документам, отсутствовавшим в обучающей выборке (внутренние базы знаний, личные файлы);

• модель не знает о событиях после даты отсечения обучающих данных;

• при отсутствии релевантных знаний модель всё равно максимизирует правдоподобие правдоподобного продолжения по обученному распределению, что проявляется как генерация синтаксически связного, но фактически неверного текста (галлюцинации).

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

1. Fine-tuning – дообучение весов на новом датасете градиентным спуском.

2. RAG (Retrieval-Augmented Generation) – веса не изменяются вообще. Используется механизм in-context learning: релевантные фрагменты текста подставляются в промпт, а self-attention на этапе инференса связывает вопрос с переданным контекстом.

Архитектура RAG целиком

1) Chunking

Эмбеддинг-модель преобразует текст в вектор так: сначала через self-attention получает представление каждого токена, затем сворачивает их в один вектор. Чем длиннее текст, тем больше разнородной информации “смешивается” в векторе фиксированной размерности косинусное расстояние до вектора одной конкретной темы растёт, точность поиска падает. Кроме того, self-attention имеет сложность O(n²), поэтому у модели есть жёсткий лимит на длину входа. Поэтому текст режут на чанки.

Стратегии

1. Fixed chunking — режем текст по N символов с overlap (перекрытием).

2. Recursive chunking — сначала разделяем текст по абзацам, потом по предложениям и т. д.

3. Semantic chunking — делим текст не по размеру, а по смыслу.

4. Structure-aware chunking — разделение по структуре документа (заголовки, секции и т. д.).

Здесь реализованы первые две – остальные требуют лишних проходов через embedding-модель или парсер.

Fixed chunking

def split_into_chunks(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    """
    Режет text на чанки, с overlap символами перекрытия между соседними чанками.
    """
    chunks = []
    if chunk_size <= 0:
        raise ValueError("Неверная длина")
        
    if overlap >= chunk_size:
        overlap = chunk_size - 1

    step = chunk_size - overlap
    for i in range (len(text) // step + 1):
        start = i * step
        end = start + chunk_size
        chunk = text[start:end]
        if len(chunk) != 0:
            chunks.append(chunk)
        else:
            pass
    return chunks

Recursive chunking

def recursive_split(text, chunk_size=500, separators=None):
    """
    Рекурсивно режет text на чанки
    """
    if separators is None:
        separators = ["nn", "n", ". ", " ", ""]

    if len(separators) == 0:
        return split_into_chunks(text, chunk_size=chunk_size, overlap=0)

    sep = separators[0]
    rest_separators = separators[1:]

    pieces = text.split(sep)

    chunks = []
    for piece in pieces:
        if len(piece) <= chunk_size:
            chunks.append(piece)
        else:
            sub_chunks = recursive_split(piece, chunk_size=chunk_size, separators=rest_separators)
            chunks.extend(sub_chunks)

    return chunks

Базовый случай рекурсии – разделители кончились, остаток режется fixed-chunking без overlap.

Когда рекурсия доходит до отдельных слов, у них почти нулевая информативность для семантического поиска. Постпроцессинг склеивает обрывки обратно до целевого размера:

def merge_small_chunks(pieces: list[str], chunk_size: int = 500, separator: str = " ") -> list[str]:
    """
    Склеивает соседние маленькие куски из pieces обратно вместе,
    пока суммарная длина не приблизится к chunk_size.
    """
    buffer = []
    result = []

    for piece in pieces:
        buffer.append(piece)
        total_length = len(separator.join(buffer))

        if total_length > chunk_size:
            buffer.pop()

            if buffer:
                result.append(separator.join(buffer))

            buffer = [piece]

    if buffer:
        result.append(separator.join(buffer))

    return result

2) Embeddings и Retrieval

После сегментации текста на чанки их необходимо преобразовать в эмбеддинги – векторные представления, которые позволяют искать семантически близкие фрагменты в пространстве признаков с помощью косинусного сходства. Чем ближе векторы друг к другу, тем более похожи соответствующие тексты по смыслу.

Bi-encoder vs Cross-encoder

Bi-encoder кодирует запрос и документ независимо через один энкодер, каждый получает свой вектор. Bi-encoder используется для поиска релевантных ответов в пространстве признаков с помощью косинусного сходства. Формула косинусного сходства:

Чем ближе векторы в пространстве признаков (чем меньше угол между ними), тем более похожи соответствующие тексты по смыслу с точки зрения математики. Деление на нормы убирает влияние длины векторов, оставляя только их направление: 1 максимальная семантическая близость, 0 ортогональность, -1 противоположность.

Cross-encoder подаёт запрос и документ как единую последовательность (query doc). Self-attention обрабатывает все токены обеих частей одновременно, на выходе формируется скаляр через классификационную голову. Такой подход точнее, но не допускает кеширования: для каждой пары запрос-документ требуется отдельный полный проход сети. Поэтому bi-encoder используется для быстрого поиска по всей базе, а cross-encoder — для дореранжирования уже отобранных кандидатов.

В проекте задействованы: nomic-embed-text (bi-encoder), cross-encoder/ms-marco-MiniLM-L-6-v2 (реранкер, обучен на MS MARCO) и llama3.1:8b (генерация) — все запускаются через Ollama.

def get_embedding(text: str, model: str = "nomic-embed-text") -> list[float]:
    """Получить эмбеддинг текста через локальный сервер Ollama."""
    response = requests.post(
        "http://localhost:11434/api/embeddings",
        json={"model": model, "prompt": text}
    )
    return response.json()["embedding"]

Дальше нам нужно где-то хранить и обрабатывать наши векторы. Для этого реализуем класс VectorStore

class VectorStore:
    def __init__(self, embed_model: str = "nomic-embed-text"):
        self.chunks = []
        self.embeddings = []
        self.metadata = []
        self.embed_model = embed_model

    def add(self, text: str, embedding: list[float], metadata: dict = None):
        """Добавить один чанк в хранилище"""
        self.chunks.append(text)
        self.embeddings.append(embedding)
        self.metadata.append(metadata if metadata else {})

    def add_text(self, text: str, metadata: dict = None):
        """Создать эмбединги"""
        embedding = get_embedding(f"search_document: {text}", model=self.embed_model)
        self.add(text, embedding, metadata)

    def save(self, path: str):
        data = {
            "chunks": self.chunks,
            "embeddings": self.embeddings,
            "metadata": self.metadata
        }
        with open(path, "wb") as f:
            pickle.dump(data, f)

    def load(self, path: str):
        with open(path, "rb") as f:
            data = pickle.load(f)
            self.chunks = data["chunks"]
            self.embeddings = data["embeddings"]
            self.metadata = data["metadata"]

    def search(self, query_text: str, top_k: int = 3, score_threshold: float = 0.0,
               filter_metadata: dict = None) -> list[dict]:
        """Найти top_k чанков, наиболее похожих на query_text"""
        if not self.embeddings:
            return []

        query_embedding = get_embedding(f"search_query: {query_text}", model=self.embed_model)

        if filter_metadata is None:
            valid_indices = list(range(len(self.chunks)))
        else:
            valid_indices = [i for i in range(len(self.chunks)) if matches_filter(self.metadata[i], filter_metadata)]

        if not valid_indices:
            return []

        matrix = np.array([self.embeddings[i] for i in valid_indices])
        query = np.array(query_embedding)

        # Формула косинусного сходства
        dot_products = matrix @ query
        norms = np.linalg.norm(matrix, axis=1) * np.linalg.norm(query)
        similarities = dot_products / norms

        top_indices = np.argsort(similarities)[::-1][:top_k]

        results = []
        for idx in top_indices:
            real_idx = valid_indices[idx]
            score = float(similarities[idx])
            if score < score_threshold:
                continue
            results.append({
                "text": self.chunks[real_idx],
                "score": score,
                "metadata": self.metadata[real_idx]
            })
        return results

Поиск реализован через векторизованные операции NumPy (matrix @ query и np.linalg.norm), что даёт высокую производительность даже при тысячах чанков.

Также функция, которая позволяет сократить количество документов для поиска, следовательно, и нагрузку на систему.

def matches_filter(metadata: dict, filter_metadata: dict) -> bool:
    for key, value in filter_metadata.items():
        if metadata.get(key) != value:
            return False
    return True

3) Reranking

После того как bi-encoder отобрал top-k кандидатов по косинусному сходству, мы применяем reranking – повторную сортировку этих кандидатов с помощью cross-encoder.

Bi-encoder даёт быстрый, но грубый отбор: он оценивает семантическую близость через косинусное сходство предвычисленных векторов. Однако этот подход не учитывает прямое взаимодействие токенов запроса и документа – информация теряется при независимом кодировании.

Cross-encoder решает эту проблему: он подаёт запрос и документ как единую последовательность (query doc), self-attention обрабатывает все токены обеих частей одновременно, а на выходе выдаётся скалярная оценка релевантности. Это даёт более точное ранжирование, поскольку модель может напрямую сопоставлять слова запроса со словами документа.

Но почему бы не использовать cross-encoder на всей базе? Всё упирается в вычислительные мощности. Cross-encoder не умеет вычислять эмбеддинг документа заранее и независимо от запроса — ему на вход нужны сразу оба текста вместе (вопрос и документ), и релевантность считается только для этой конкретной пары. Это значит, что для базы из 10 000 чанков пришлось бы на каждый новый вопрос пользователя прогонять модель 10 000 раз подряд

def rerank(query: str, candidates: list[dict], top_k: int = 2) -> list[dict]:
    """
    Пересортировывает candidates по релевантности к query через cross-encoder.
    """

    pairs = [(query, c["text"]) for c in candidates]
    scores = reranker.predict(pairs)
    for i, candidate in enumerate(candidates):
        candidate["rerank_score"] = float(scores[i])

    candidates_sorted = sorted(candidates, key=lambda x: x["rerank_score"], reverse=True)

    return candidates_sorted[:top_k]

4) Generation

Финальный этап создание функции для генерации ответа

def generate_answer(question: str, chunks: list[str], model: str = "llama3.1:8b") -> str:
    """
    Генерирует ответ на question, используя chunks как контекст.
    """
    context_text = "n---n".join(chunks)

    prompt = f"""Дай развёрнутый, подробный ответ. Объясни ключевые понятия своими словами,
приведи детали и примеры, если они есть в контексте. Не ограничивайся
одним предложением — раскрой тему полноценно, опираясь на все релевантные
части контекста.

Контекст:
---
{context_text}
---

Вопрос: {question}

Ответ:"""

    response = requests.post(
        "http://localhost:11434/api/generate",
        json={
            "model": model,
            "prompt": prompt,
            "stream": False
        }
    )
    return response.json()["response"]
  

Модель генерирует ответ авторегрессивно, токен за токеном. Self-attention на каждом шаге учитывает весь переданный контекст, включая вставленные чанки. Явная инструкция опираться только на контекст снижает вероятность того, что модель проигнорирует переданные факты в пользу параметрических знаний, выученных при обучении.

На этом момента все функции написаны и осталось собрать все в единный пайплайн

def rag_pipeline(store: VectorStore, question: str, top_k_retrieve: int = 50, top_k_final: int = 5) -> str:
    """
    Полный RAG pipeline
    """
    candidates = store.search(question, top_k=top_k_retrieve)

    if not candidates:
        return "В базе знаний не нашлось релевантной информации."

    reranked = rerank(question, candidates, top_k=top_k_final)

    chunks_text = [r["text"] for r in reranked]
    answer = generate_answer(question, chunks_text)

    return answer

Параметры top_k_retrieve и top_k_final регулируют разные вещи:

• top_k_retrieve — размер кандидатского пула для reranker (30-50), эти чанки не попадают напрямую в промпт

• top_k_final — сколько чанков реально войдёт в промпт LLM (3-5)

После этого создаём объект класса VectorStore и прописываем код для загрузки и выгрузки данных из хранилища. Пересчёт эмбеддингов для объёмного документа занимает заметное время, поэтому проиндексированное состояние сериализуется через pickle и загружается при последующих запусках вместо повторного обращения к embedding-модели:

store = VectorStore()

if os.path.exists(INDEX_PATH):
    print("Найден сохранённый индекс")
    store.load(INDEX_PATH)
    print(f"Загружено чанков: {len(store.chunks)}")
else:
    print("Сохранённого индекса нет")
    sample_text = read_pdf("test.pdf")
    raw_pieces = recursive_split(sample_text, chunk_size=1500)
    final_chunks = merge_small_chunks(raw_pieces, chunk_size=1000, separator=" ")

    for i, chunk in enumerate(final_chunks):
        store.add_text(chunk, metadata={"source": "test.pdf", "chunk_index": i})

    store.save(INDEX_PATH)
    print(f"Проиндексировано и сохранено чанков: {len(store.chunks)}")

Все готово! Теперь мы можем задать вопрос нашей RAG системе. Для примера я взял файл со 2 курса предмета “Теория информационных процессов и систем”.

question = "Метод Ветвей и границ, достоинства и недостатки"
answer = rag_pipeline(store, question)
print(f"Вопрос: {question}")
print(f"Ответ: {answer}")

Вопрос: Метод Ветвей и границ, достоинства и недостатки

Ответ: Метод Ветвей и границ – это алгоритм поиска, который используется для решения проблемы нахождения оптимального пути в графе. Он ориентирован на проверку операторов, преобразующих состояния, и определяет граничную глубину поиска, т. е. количество проводимых итераций поиска.

Один из ключевых понятий в методе Ветвей и границ – это процедура проверки, которая позволяет сохранять апостериорную информацию. Используя простейшие эвристические методы поиска на основе этой процедуры, можно выполнить первую и вторую задачу поиска, т. е. найти не только необходимое состояние, но и расстояние до него относительно начального.

Достоинства метода Ветвей и границ:

• Он позволяет найти оптимальный путь в графе.

• Он может быть использован для решения различных проблем, включая нахождение кратчайшего пути, а также других задач поиска.

• Он имеет универсальную природу, что означает, что он может быть применен для решения разных типов задач поиска.

Недостатки метода Ветвей и границ:

• Для определения оптимального пути необходимо проводить много итераций поиска, что может привести к значительному увеличению времени выполнения.

• Метод требует большого количества памяти для хранения информации об операторах и состояниях.

• Избегание циклов в методе Ветвей и границ может быть проблематичным.

Возможность использования различных способов генерации в методе Ветвей и границ может привести к различной эффективности решения проблемы. Например, использование генерации по лучу или генерации параллельным методом может дать различные результаты.

Пример: Рассмотрим задачу нахождения оптимального пути в графе с 10 вершинами и 20 ребрами. Используя метод Ветвей и границ с процедурой проверки, мы можем найти кратчайший путь от вершины A до вершины B. Однако если мы использовать генерацию по лучу вместо генерации параллельным методом, мы можем получить разные результаты.

В заключении, метод Ветвей и границ является мощным инструментом для решения проблем нахождения оптимального пути в графе. Однако его недостатки, такие как большой объем памяти и сложность избегания циклов, должны быть учтены при его применении.

Заключение

Мы построили RAG-систему с нуля, без использования готовых фреймворков вроде LangChain или LlamaIndex. Разобрали каждый компонент: от сегментации текста на чанки до генерации финального ответа.

Результат работы системы наглядно демонстрирует, что подход работает: ответ на вопрос о методе Ветвей и границ полностью совпал с информацией из исходного файла, без примеси общих знаний LLM о графовых алгоритмах.

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