RAG на простом Python: разбиение на части, TF-IDF, косинусное сходство и формулировка запросов

В типичном учебнике по RAG вам предлагают установить пять библиотек и вызвать метод .from_documents(). Через десять минут демо-версия работает, но вы так и не понимаете, как это происходит. Когда система возвращает бесполезные данные или модель выдает ошибочную информацию, вы оказываетесь в тупике, потому что всю работу выполнила библиотека.

Здесь конвейер построен наоборот — простым языком Python, с использованием только модуля math и одного вызова LLM в формате простого текста в самом конце. Никаких моделей эмбеддингов, ни FAISS, ни Pinecone. Понимание внутренних механизмов позволяет при использовании LangChain или настоящей векторной базы данных понимать, что происходит на самом деле. В итоге получается рабочий конвейер, а также представление о каждом числе, которое он генерирует.

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

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

1. Переобучить модель. Это медленно и дорого, к тому же изменения, внесённые завтра, потребуют ещё одного цикла обучения.
2. Вставить всё в промпт. Такой подход неэффективен в масштабе. Даже окно объёмом в миллион токенов не может вместить 50 000 документов, а исследования моделей с длинным контекстом неоднократно указывают на явление «потери информации в середине», при котором данные, находящиеся глубоко внутри длинного промпта, используются хуже, несмотря на то что они туда входят. Кроме того, дополнительные токены приводят к увеличению задержек и стоимости каждого вызова.

На самом деле секрет RAG проще, чем кажется. Документы остаются на внешнем хранилище, и при обработке запроса извлекаются и передаются модели только те небольшие фрагменты, которые к нему относятся. Языковая модель никогда не хранит ваши данные — каждый раз она получает свежий, релевантный фрагмент.

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

Шаг 1: разбиение на фрагменты

Преобразование целого 50-страничного PDF в один вектор уничтожает почти все полезные сигналы, поскольку детали, которые могли бы ответить на вопрос, средневзвешиваются до нуля. Первой задачей является разделение документов на небольшие, целенаправленные фрагменты.

Самый простой метод — это слайдинг-окно фиксированного размера, подсчитываемое по словам, с перекрытием между соседними блоками:

def chunk_text(text: str, chunk_size: int = 200, overlap: int = 50) -> list[str]:
    if overlap >= chunk_size:
        raise ValueError("overlap must be smaller than chunk_size")

    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunks.append(" ".join(words[start:end]))
        if end >= len(words):
            break
        start = end - overlap  # step back so chunks share context
    return chunks

Эта оговорка-защита — не просто декоративный элемент. Если значение overlap достигает или превышает chunk_size, позиция начала перестаёт увеличиваться, и цикл никогда не заканчивается. Попытка увеличить значение overlap «для получения большего контекста» часто приводит к ошибкам, поэтому функция отказывается выполняться вместо того, чтобы молча застревать.

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

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

Шаг 2: преобразование текста в числа с помощью TF-IDF

Компьютер не может напрямую сравнивать степень схожести двух строк. Для всех последующих шагов требуются векторы — списки чисел, с которыми можно выполнять арифметические операции. Классическим способом создания таких векторов без учета зависимостей является TF-IDF, то есть частота вхождения терма умноженная на обратную частоту в документе.

• TF показывает, как часто слово встречается в данном фрагменте. Высокое значение для слова «kidney» указывает на то, что фрагмент касается почек.
• IDF оценивает редкость слова во всем наборе. Слова вроде «the» и «is» встречаются повсюду и не дают информации о том, какой фрагмент вы читаете, поэтому их значение близко к нулю. Редкое слово в фрагменте является гораздо более сильным сигналом.

Код формирует словарь из всех фрагментов, вычисляет значение IDF для каждого термина, а затем преобразует каждый фрагмент в вектор TF-IDF:

import math
from collections import Counter

def build_vocabulary(chunks: list[str]) -> list[str]:
    vocab = set()
    for chunk in chunks:
        vocab.update(chunk.lower().split())
    return sorted(vocab)

def compute_idf(chunks: list[str], vocab: list[str]) -> dict[str, float]:
    num_chunks = len(chunks)
    idf = {}
    for word in vocab:
        containing = sum(1 for c in chunks if word in c.lower().split())
        idf[word] = math.log(num_chunks / (1 + containing))
    return idf

def compute_tf(chunk: str, vocab: list[str]) -> dict[str, float]:
    words = chunk.lower().split()
    counts = Counter(words)
    total = len(words)
    return {word: counts.get(word, 0) / total for word in vocab}

def vectorize(chunk: str, vocab: list[str], idf: dict[str, float]) -> list[float]:
    tf = compute_tf(chunk, vocab)
    return [tf[word] * idf[word] for word in vocab]

Попробуйте применить его к трём фрагментам текста — тогда всё станет ясно. Слово, присутствующее практически в каждом фрагменте, такое как «the», получает значение IDF, близкое к нулю; поэтому его повторение ничего не добавляет к вектору. Редкое слово, связанное с конкретной темой, имеет явно положительное значение IDF. TF-IDF буквально представляет собой формулу для определения того, сообщает ли слово какую-либо уникальную информацию о фрагменте текста.

Недостатком является масштабируемость. При наличии 100 000 документов словарь может содержать более 500 000 уникальных слов, поэтому вектор каждого фрагмента, даже если он состоит из трёх слов, имеет длину 500 000 чисел, причём большинство из них равны нулю. Хранение и сравнение таких разреженных векторов требует значительных ресурсов. Именно по этой причине в производственных системах используются плотные нейронные эмбеддинги с размерностью от 384 до 1536, в которых практически нет нулей.

Шаг 3: оценка релевантности с помощью косинусного сходства

Если рассматривать фрагменты текста как векторы, следующий вопрос заключается в том, насколько данный фрагмент релевантен запросу. Чистое расстояние не является подходящим инструментом, поскольку для текста более важен угол между векторами, чем их длина. Фрагмент, в котором слово «кошка» упоминается один раз, и фрагмент, в котором оно повторяется двадцать раз, по-прежнему касаются примерно одной и той же темы, и коэффициент косинусного сходства относится к ним одинаково, не учитывая, какой из вариантов более выражен.

Для реализации необходимы скалярное произведение, модуль вектора и соотношение между ними:

def dot_product(a, b):
    return sum(x * y for x, y in zip(a, b))

def magnitude(v):
    return math.sqrt(sum(x * x for x in v))

def cosine_similarity(a, b):
    denom = magnitude(a) * magnitude(b)
    return 0.0 if denom == 0 else dot_product(a, b) / denom

По принципу косинусное сходство варьируется от -1 до 1, и при использовании данной настройки TF-IDF оно почти всегда остается в диапазоне от 0 до 1:

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

Оценка 0 — это не аномалия. Это означает, что система правильно указывает на отсутствие связи между фрагментом текста и запросом.

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

Слабым местом метода TF-IDF с использованием косинусного сходства являются синонимы. Запрос с словом «automobile» в отношении документа, содержащего слово «car», получает оценку близкую к нулю, поскольку учитывается только буквальное совпадение лексики, без учета смысла. Модели плотных вкладок, которые располагают синонимы рядом друг с другом, созданы именно для преодоления этой проблемы.

Шаг 4: механизм поиска

После того как векторизация и расчет сходства готовы, процесс поиска заключается в однократной векторизации каждого фрагмента, векторизации входящего запроса, оценке сходства каждого фрагмента с запросом, сортировке резултатов и сохранении k лучших:

class VectorStore:
    def __init__(self):
        self.chunks, self.vocab, self.idf, self.vectors = [], [], {}, []

    def index(self, documents, chunk_size=50, overlap=10):
        all_chunks = []
        for doc in documents:
            all_chunks.extend(chunk_text(doc, chunk_size, overlap))
        self.chunks = all_chunks
        self.vocab = build_vocabulary(all_chunks)
        self.idf = compute_idf(all_chunks, self.vocab)
        self.vectors = [vectorize(c, self.vocab, self.idf) for c in all_chunks]

    def search(self, query, top_k=3):
        query_vector = vectorize(query, self.vocab, self.idf)
        scored = [(c, cosine_similarity(query_vector, v))
                  for c, v in zip(self.chunks, self.vectors)]
        scored.sort(key=lambda pair: pair[1], reverse=True)
        return scored[:top_k]

Одна мелочь может тихо разрушить всё. Запрос необходимо преобразовать с использованием словаря и значений IDF, рассчитанных на основе документов; использование словаря, полученного непосредственно из самого запроса, является ошибкой. Коэффициент косинусного сходства предполагает векторы одинаковой длины, при которых позиция i соответствует одному и тому же слову в каждом из них. Если дать запросу собственный словарь, то на позиции 5 в одном векторе может находиться слово «heart», а в другом — «and», что делает значение коэффициента сходства структурно бессмысленным, а не просто незначительно неточным.

Шаг 5: дополнение промпта

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

def build_augmented_prompt(query: str, retrieved_chunks: list[str]) -> str:
    context = "\n\n".join(
        f"[Source {i+1}]: {chunk}" for i, chunk in enumerate(retrieved_chunks)
    )
    return f"""Answer the question using ONLY the context below.
If the answer is not contained in the context, say "I don't know based on the provided context."
Cite which source number(s) you used.

Context:
{context}

Question: {query}

Answer:"""

Два аспекта здесь не являются косметическими.

«Использовать ТОЛЬКО приведённый ниже контекст». Без этого чёткого правила модель смешивает полученные фрагменты с тем, что запомнила во время обучения, и тогда уже невозможно определить, какая часть ответа взята из надёжных документов, а какая — это предположение. Если такое смешивание делается уверенно, оно наносит больший ущерб, чем полное игнорирование технологии RAG, поскольку создаётся впечатление наличия источника там, где его нет.

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

Шаг 6: объединение компонентов

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

class RagPipeline:
    def __init__(self, client, model="openai/gpt-oss-20b"):
        self.store = VectorStore()
        self.client = client
        self.model = model

    def build_index(self, documents: list[str]):
        """Run once, whenever documents are loaded or updated."""
        self.store.index(documents, chunk_size=50, overlap=10)

    def answer(self, query: str, top_k: int = 3) -> str:
        """Run on every user query — never re-indexes."""
        results = self.store.search(query, top_k=top_k)
        chunks = [c for c, _ in results]
        prompt = build_augmented_prompt(query, chunks)

        response = self.client.chat.completions.create(
            model=self.model,
            max_tokens=500,
            messages=[{"role": "user", "content": prompt}],
        )
        return response.choices[0].message.content

Если бы индексация выполнялась внутри функции на уровне каждого запроса, каждый запрос заставлял бы снова обрабатывать текст, пересоздавать словарь и пересчитывать все векторы TF-IDF для всей коллекции, хотя ничто в ней не меняется между запросами. При 10 000 документах и 1 000 запросах это приводит к 999 одинаковым процедурам пересоздания. Правильный подход — сначала создать индекс один раз, а затем выполнять запросы многократно; именно к такому подходу склоняют FAISS, Chroma и Pinecone. Это правильная архитектура с самого начала, а не преждевременная оптимизация.

В чём проблема этой системы

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

• Отсутствие семантического понимания. Термины «автомобиль» и «машина» никогда не совпадут, поскольку алгоритм TF-IDF учитывает только буквальное совпадение в словаре.

Редкие, очень широкие векторы. Размер словаря соответствует объему корпуса, поэтому каждый вектор увеличивается вместе с ним. При количестве документов, превышающем несколько тысяч, это становится неудобным для использования.

В коде отсутствует проверка на то, что процесс поиска нашел что-то полезное. Даже когда оценки всех элементов из группы top-k близки к нулю, система все равно передает их в модель. Запрос в промпте о признании незнания полностью зависит от готовности большой языковой модели сотрудничать, поэтому слабый результат поиска может все равно попасть в процесс генерации без проверки.