RAG с нуля: как 17-летний написал алгоритм для хакатона по финансам — разбор кода и архитектуры

Школьник, PDF-ки и куча финансовых терминов

Представьте хакатон по финтеху. Вокруг — команды из банков с опытными дата-сайентистами. А в углу — 17-летний парень с ноутбуком, который за 48 часов с нуля собрал систему, отвечающую на вопросы по годовым отчетам компаний. Без LangChain, без готовых облачных API. Чистый Python, кастомные эмбеддинги и упрямство.

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

Проблема: ИИ, который галлюцинирует на финансовых цифрах

Финансовые документы — это ад. Цифры, таблицы, перекрестные ссылки, юридические формулировки. Попросите обычную LLM вроде GPT ответить, как изменилась чистая прибыль компании X в 2023 году — она сгенерирует красивый, уверенный и абсолютно выдуманный ответ.

Задача хакатона была конкретной: по набору PDF с годовыми отчетами (10-К форм) отвечать на точные вопросы инвесторов. Точность — священная корова. Ошибка в цифре может стоить денег.

Решение: Свой велосипед RAG

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

1 Загрузка и чанкинг: не просто split по символам

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

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

# Плохо: наивное разделение
def naive_chunk(text, size=500):
    return [text[i:i+size] for i in range(0, len(text), size)]

А вот стратегия, которая сработала на хакатоне:

import re
from typing import List

def smart_chunk(text: str, chunk_size: int = 1000, overlap: int = 200) -> List[str]:
    """
    Резьба по смыслу: старается разбивать по абзацам,
    предложениям, сохраняя перекрытие.
    """
    # Сначала разбиваем по двойным переводам строк (абзацы)
    paragraphs = re.split(r'\n\n+', text)
    chunks = []
    current_chunk = []
    current_size = 0
    
    for para in paragraphs:
        para = para.strip()
        if not para:
            continue
        para_size = len(para)
        
        # Если абзац огромный (например, таблица), режем его по строкам
        if para_size > chunk_size:
            lines = para.split('\n')
            for line in lines:
                line = line.strip()
                if current_size + len(line) > chunk_size and current_chunk:
                    chunks.append(' '.join(current_chunk))
                    # Перекрытие: оставляем последние N слов предыдущего чанка
                    overlap_words = ' '.join(current_chunk).split()[-overlap//5:]
                    current_chunk = overlap_words + [line]
                    current_size = len(' '.join(current_chunk))
                else:
                    current_chunk.append(line)
                    current_size += len(line) + 1
        else:
            # Обычный абзац
            if current_size + para_size > chunk_size and current_chunk:
                chunks.append(' '.join(current_chunk))
                overlap_words = ' '.join(current_chunk).split()[-overlap//5:]
                current_chunk = overlap_words + [para]
                current_size = len(' '.join(current_chunk))
            else:
                current_chunk.append(para)
                current_size += para_size + 1
    
    if current_chunk:
        chunks.append(' '.join(current_chunk))
    return chunks

Перекрытие (overlap) критически важно. Иначе контекст теряется на границах чанков. Но overlap в 200 символов — не догма. Для таблиц лучше перекрывать по строкам, для текста — по предложениям.

2 Эмбеддинги: выбираем модель, которая понимает финансы

Все бросились использовать text-embedding-ada-002. Но она платная и не всегда оптимальна для узкой предметной области. Парень выбрал открытую модель, которую можно запустить локально — all-MiniLM-L6-v2 от Sentence Transformers.

Почему? Маленький размер (80 МБ), быстро работает на CPU, и ее можно дообучить на финансовых текстах (хотя на хакатоне не было времени).

from sentence_transformers import SentenceTransformer
import numpy as np

class Embedder:
    def __init__(self, model_name='all-MiniLM-L6-v2'):
        self.model = SentenceTransformer(model_name)
    
    def encode(self, texts: List[str]) -> np.ndarray:
        """Возвращает numpy массив с эмбеддингами."""
        return self.model.encode(texts, show_progress_bar=False, normalize_embeddings=True)

3 Векторная база: не обязательно Pinecone или Weaviate

На хакатоне не было времени разворачивать сложные кластеры. Решение — использовать FAISS от Facebook. Библиотека, которая работает в памяти, бешено быстрая и имеет индекс для точного или приближенного поиска.

import faiss
import pickle
import os

class VectorStore:
    def __init__(self, dimension=384):  # Размерность all-MiniLM-L6-v2
        self.index = faiss.IndexFlatIP(dimension)  # IndexFlatIP для скалярного произведения (косинусного сходства)
        self.chunks = []
        self.metadata = []  # Можно хранить источник, номер страницы и т.д.
    
    def add(self, embeddings: np.ndarray, chunks: List[str], meta: List[dict] = None):
        """Добавляет векторы и связанные с ними чанки."""
        self.index.add(embeddings.astype('float32'))
        self.chunks.extend(chunks)
        if meta:
            self.metadata.extend(meta)
        else:
            self.metadata.extend([{}] * len(chunks))
    
    def search(self, query_embedding: np.ndarray, k=5) -> List[tuple]:
        """Ищет k ближайших соседей, возвращает чанки и сходство."""
        distances, indices = self.index.search(query_embedding.astype('float32'), k)
        results = []
        for i, idx in enumerate(indices[0]):
            if idx != -1:  # FAISS может возвращать -1, если индекс пустой
                results.append((self.chunks[idx], distances[0][i], self.metadata[idx]))
        return results
    
    def save(self, path):
        """Сохраняет индекс и данные на диск."""
        faiss.write_index(self.index, os.path.join(path, 'index.faiss'))
        with open(os.path.join(path, 'data.pkl'), 'wb') as f:
            pickle.dump({'chunks': self.chunks, 'metadata': self.metadata}, f)
    
    def load(self, path):
        """Загружает с диска."""
        self.index = faiss.read_index(os.path.join(path, 'index.faiss'))
        with open(os.path.join(path, 'data.pkl'), 'rb') as f:
            data = pickle.load(f)
            self.chunks = data['chunks']
            self.metadata = data['metadata']

IndexFlatIP (Inner Product) идеален для нормализованных векторов. Если вам нужна скорость на миллионах векторов, переходите на HNSW индекс. Но для нескольких тысяч отчетов Flat достаточно.

4 Сборка конвейера: от PDF до ответа

Теперь склеиваем все компоненты. Главное — сохранять метаданные для каждого чанка: имя файла, номер страницы. Иначе как вы поймете, откуда пришел ответ?

import PyPDF2
from pathlib import Path

def process_pdfs(pdf_folder: str, vector_store: VectorStore, embedder: Embedder):
    """Читает все PDF в папке, чанкирует, создает эмбеддинги, добавляет в хранилище."""
    for pdf_file in Path(pdf_folder).glob('*.pdf'):
        print(f"Обрабатываю {pdf_file.name}")
        text = ""
        metadata = []
        
        with open(pdf_file, 'rb') as f:
            reader = PyPDF2.PdfReader(f)
            for page_num, page in enumerate(reader.pages):
                page_text = page.extract_text()
                # Простая очистка
                page_text = ' '.join(page_text.split())
                text += page_text + '\n\n'
                # Сохраняем метаданные для каждой строки (упрощенно)
                metadata.append({'file': pdf_file.name, 'page': page_num + 1})
        
        chunks = smart_chunk(text)
        # Привязываем метаданные к чанкам (упрощенно: каждый чанк получает метаданные первой страницы в нем)
        chunk_meta = [metadata[0] for _ in chunks]  # Здесь нужна более сложная логика
        
        embeddings = embedder.encode(chunks)
        vector_store.add(embeddings, chunks, chunk_meta)

5 Генерация: заставляем LLM говорить только по делу

Вот где многие ошибаются. Они просто склеивают найденные чанки и говорят модели: «Ответь на вопрос». Но LLM может проигнорировать контекст и начать галлюцинировать.

Промпт должен быть строгим. Вот шаблон, который использовался:

def build_prompt(question: str, retrieved_chunks: List[tuple]) -> str:
    """Строит промпт для LLM с жесткими инструкциями."""
    context = "\n\n---\n\n".join([f"[Источник: {meta.get('file', 'N/A')}, стр. {meta.get('page', 'N/A')}]\n{chunk}" 
                                      for chunk, score, meta in retrieved_chunks])
    
    prompt = f"""Ты — финансовый аналитик. Ответь на вопрос, используя ТОЛЬКО предоставленный контекст.
Если в контексте нет информации для ответа, скажи "В предоставленных документах нет информации для ответа на этот вопрос."

Контекст:
{context}

Вопрос: {question}

Ответ (будь краток и точен, указывай цифры и даты если они есть в контексте):"""
    return prompt

Затем этот промпт отправляется в LLM. На хакатоне использовался GPT-3.5 Turbo через API, но для локального варианта можно взять Llama 3 или Mistral. Если интересно, как собрать полностью локальную систему, посмотрите статью «Полное руководство: как собрать Agentic RAG систему полностью локально».

Где спрятаны грабли: 5 ошибок, которые сломают ваш RAG

1. Чанкинг без перекрытия. Контекст обрывается, и модель теряет смысл. Всегда добавляйте overlap хотя бы 10% от размера чанка.
2. Поиск только по косинусному сходству. Для финансовых документов важен также ключевой поиск (по терминам). Комбинируйте семантический и лексический поиск (гибридный). О будущем гибридного поиска читайте в RAG 2026: roadmap.
3. Игнорирование метаданных. Без номеров страниц и названий файлов вы не сможете проверить ответ. Храните их с каждым чанком.
4. Слабый промпт. Если не дать модели строгих инструкций, она начнет выдумывать. Явно указывайте: «Используй только контекст».
5. Эмбеддинги для таблиц. Текстовая модель плохо кодирует таблицы. Рассмотрите специальные методы, например, распознавание структуры таблицы и преобразование ее в текст вида «Строка: Выручка, 2022 год: 100 млн, 2023 год: 120 млн».

Что дальше? Эволюция простого RAG

Базовая система работает. Но в production этого мало. Нужны:

• Рерактор (Reranker). После первичного поиска 20 чанков, пропустите их через модель для переранжирования (например, cross-encoder). Это повысит точность.
• Агентская логика. Если вопрос сложный («Сравни прибыль компаний A и B»), разбейте его на подзапросы, выполните несколько поисков, синтезируйте ответ. Это уже Agentic RAG.
• Файн-тюнинг эмбеддера. Дообучите модель на финансовых текстах, чтобы она лучше понимала термины. Практические кейсы есть в статье про файн-тюнинг LLM для RAG.

История 17-летнего — не про гениальность. Она про то, что сложные системы становятся простыми, если разобрать их на части и понять каждую. Вы можете повторить этот путь. Не ищите волшебную библиотеку. Напишите свой конвейер. Потом уже оптимизируйте.

Соберите свою систему, протестируйте на своих документах. И когда что-то сломается (а это случится), вы будете знать, в каком модуле копать. Это и есть инженерия.