Создайте локальную систему RAG на Python без затрат на API

Полный цикл генерации с использованием поиска может быть создан и запущен полностью на персональном компьютере, используя только процессор, бесплатные модели с открытым исходным кодом, без ключа OpenAI, без кредитной карты и без ограничений по скорости. Главное — понимать каждый компонент, а не просто копировать готовое решение.

Лучший способ освоения — практика: держите открытым терминал и файл на Python с названием rag.py, запускайте каждый блок и предсказывайте результат перед его выполнением. После большинства шагов представлены краткие упражнения, и именно в них происходит основная часть обучения.

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

Языковая модель знает только то, что содержалось в её данных обучения. Личные заметки, корпоративная вики, записи прошлой недели или любые неизвестные документы для неё недоступны. Если её спросить об этом, она либо признает свою некомпетентность, либо, что ещё хуже, уверенно выдумывает ответы.

RAG разделяет задачу на две части:

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

Модель прекращает догадки и начинает анализировать текст. Всё остальное — это методы для качественного выполнения этих двух шагов.

question ──► [embed] ──► [search your docs] ──► top chunks ──┐
                                                             ▼
                                          [LLM: "answer using this context"] ──► answer

Бесплатный набор инструментов

Всю систему обеспечивают три компонента. sentence-transformers преобразует текст в векторы с использованием небольшой модели, которая работает на CPU и загружается один раз. numpy обрабатывает вычисления сходства, а вся «база данных векторов» помещается примерно в пять строк, что позволяет видеть её работу. transformers с моделью Flan-T5 формирует ответ, поскольку она настроена под выполнение инструкций, работает на CPU и не требует API.

Установите всё один раз:

pip install sentence-transformers transformers torch numpy

При первом загрузке каждая модель получает свои веса с Hugging Face. Модуль эмбеддингов весит примерно 80 МБ, а flan-t5-base — около 1 ГБ. После этого они кэшируются и позволяют работать в автономном режиме. При ограниченном объеме диска или оперативной памяти flan-t5-small (примерно 300 МБ) может заменить базовую модель, при этом качество будет ниже, но скорость значительно выше.

Шаг 1: корпус данных, недоступный модели

Принцип RAG легче всего понять на примере документов, которые модель никогда не могла видеть; поэтому скрипт начинается с небольшой вымышленной корпоративной вики. Создайте файл rag.py и вставьте туда документы:

# rag.py
DOCUMENTS = [
    """Nimbus is a fictional note-taking app launched in 2023. The free plan,
    called Nimbus Lite, allows up to 50 notes and 1 GB of storage. There are no
    collaboration features on the free plan.""",

Вики продолжается дальше с новыми записями:

    """Nimbus Pro costs 8 dollars per month billed annually, or 10 dollars billed
    monthly. Pro removes the note limit, gives 50 GB of storage, and unlocks
    real-time collaboration with up to 5 people per note.""",    """Nimbus stores all notes encrypted at rest using AES-256. End-to-end
    encryption is only available on the Pro plan and must be enabled manually in
    Settings > Security. Once enabled it cannot be turned off for that note.""",    """The Nimbus mobile app supports offline editing. Changes made offline are
    queued and sync automatically the next time the device is online. If two
    devices edit the same note offline, Nimbus keeps both versions and flags a
    conflict for the user to resolve.""",    """Nimbus offers a 30-day refund policy on all paid plans, no questions asked.
    Refunds are processed to the original payment method within 5 business days.
    Annual plans cancelled after 30 days are not refundable but stay active until
    the end of the billing period.""",    """Nimbus support is available via email at help@nimbus.example and live chat.
    Live chat is only staffed for Pro customers, Monday to Friday, 9am to 6pm UTC.
    Free-plan users receive email support with a typical 48-hour response time.""",
]

Прежде чем что-либо создавать, докажите, что проблема существует. В оболочке Python задайте простой модели вопрос о Nimbus:

from transformers import pipeline
gen = pipeline("text2text-generation", model="google/flan-t5-base")
print(gen("How much does Nimbus Pro cost?", max_new_tokens=50)[0]["generated_text"])

Это приведёт к получению какого-то результата, но этот результат будет неверным или пустым, поскольку это лишь предположение. Запомните этот ответ, ведь далее в руководстве будет дано решение проблемы.

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

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

Добавьте функцию разбиения на фрагменты:

def chunk_text(text, chunk_size=60, overlap=15):
    """Split text into overlapping chunks of `chunk_size` words."""
    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 by `overlap` so context isn't cut
    return chunks

Затем сформируйте список кусков, фиксируя при этом, из какого документа каждый кусок взят:

# Build our chunk list, remembering which doc each chunk came from
chunks = []
for doc_id, doc in enumerate(DOCUMENTS):
    for c in chunk_text(doc):
        chunks.append({"doc_id": doc_id, "text": c})print(f"{len(DOCUMENTS)} documents -> {len(chunks)} chunks")
for c in chunks[:3]:
    print("-", c["text"][:70], "...")

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

Упражнение: выполните процесс снова, установив chunk_size=20, чтобы получить больше мелких частей, затем попробуйте chunk_size=500, при котором каждый из этих коротких документов станет отдельной частью. Перед запуском попробуйте угадать, какой параметр обеспечит более точный результат, но при этом повысит риск потери окружающего контекста. Мелкие части точны, но лишены контекста, и именно это противоречие лежит в основе процесса разбиения на части. Универсально правильного размера не существует.

Шаг 3: эмбеддинги

Эмбеддинг — это вектор чисел, кодирующий значение текста. Фрагменты текста с похожим значением превращаются в векторы, направленные в одну сторону, даже если у них нет общих слов; поэтому фраза «Сколько это стоит?» оказывается рядом с фразой «Какова цена?»

Импортируйте модель эмбеддингов:

from sentence_transformers import SentenceTransformer

Затем вставьте каждый фрагмент:

embedder = SentenceTransformer("all-MiniLM-L6-v2")# Embed every chunk. normalize_embeddings=True makes the vectors unit-length,
# which lets us measure similarity with a simple dot product later.
chunk_texts = [c["text"] for c in chunks]
chunk_vectors = embedder.encode(chunk_texts, normalize_embeddings=True)print("vector shape:", chunk_vectors.shape)   # (num_chunks, 384)

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

Упражнение: увидите, как работает смыслление, проведя небольшой эксперимент:

import numpy as np
pairs = embedder.encode(
    ["the price of the pro plan", "how much does it cost", "the weather in Paris"],
    normalize_embeddings=True,
)
print("price vs cost :", round(float(pairs[0] @ pairs[1]), 3))   # should be HIGH
print("price vs weather:", round(float(pairs[0] @ pairs[2]), 3)) # should be LOW

Выражение a @ b представляет собой скалярное произведение, которое для единичных векторов равно косинусу сходства. Значение 1.0 означает одинаковое направление, а 0 — отсутствие связи. Первая пара должна показать значительно более высокий результат, чем вторая, несмотря на отсутствие общих слов, и это и есть семантический поиск. Попробуйте угадать оба числа с точностью до 0.2 перед запуском.

Шаг 4: поиск

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

import numpy as np

Функция поиска:

def retrieve(question, k=3):
    q_vec = embedder.encode([question], normalize_embeddings=True)[0]
    scores = chunk_vectors @ q_vec              # cosine similarity to every chunk
    top_idx = np.argsort(scores)[::-1][:k]      # indices of the k highest scores
    return [(chunks[i]["text"], float(scores[i])) for i in top_idx]for text, score in retrieve("How much does Nimbus Pro cost?"):
    print(f"[{score:.3f}] {text[:80]}...")

Блок с ценой должен выходить на первое место с наивысшей оценкой, даже если в вопросе говорилось «стоимость», а в документе — «стоимость 8 долларов в месяц». Речь идет о сопоставлении смысла, а не ключевых слов.

Упражнение: попробуйте три вопроса и посмотрите, что будет отображаться:

• retrieve("Могу ли я вернуть свои деньги?") должен возвращать блок с информацией о возврате средств, хотя слово «возврат» никогда не упоминалось в вопросе.
• retrieve("Безопасны ли мои данные?") должен показывать блок с информацией об шифровании.
• retrieve("Работает ли он без подключения к Интернету?") должен возвращать блок с информацией об офлайн-синхронизации.

Многие недооценивают этот этап. Когда система возвращает неверные фрагменты информации, никакие умные подсказки не помогут получить правильный ответ. Что введено — то и получено, поэтому именно здесь в первую очередь следует заняться отладкой.

Шаг 5: генерация

Теперь две части информации соединяются. Необходимо составить подсказку, которая передаст полученные фрагменты в Flan-T5 и скажет ему отвечать исключительно на их основе.

from transformers import pipeline

Функция генерации:

generator = pipeline("text2text-generation", model="google/flan-t5-base")def rag_answer(question, k=3):
    retrieved = retrieve(question, k=k)
    context = "\n".join(text for text, _ in retrieved)    prompt = f"""Answer the question using only the context below.
If the answer is not in the context, say you don't know.Context:
{context}Question: {question}
Answer:"""    out = generator(prompt, max_new_tokens=80)[0]["generated_text"]
    return out.strip(), retrievedanswer, sources = rag_answer("How much does Nimbus Pro cost?")
print("ANSWER:", answer)
print("\nBased on:")
for text, score in sources:
    print(f"  [{score:.3f}] {text[:70]}...")

Сравните полученный результат с неверным или пустым ответом из шага 1. Модель и вопрос остаются одинаковыми, единственное изменение заключается в том, что для чтения предоставляется соответствующий текст. Именно это различие и представляет собой суть технологии RAG.

Упражнение: сформулируйте вопрос, на который документы не дают ответа, например о клиенте Linux для рабочего стола Nimbus, и вызовите функцию rag_answer с этим вопросом. Правильно настроенная система признает отсутствие информации. Если ваша система выдает вымышленный ответ, это является реальным сценарием сбоя, называемым галлюцинацией, несмотря на выполнение операции по поиску информации; этот вопрос рассматривается в следующем разделе.

Готовый скрипт

Ниже приведен полный код файла rag.py в виде единой запускаемой программы с интерактивным циклом. Он начинается с импортов:

# rag.py — a complete, local, no-API RAG system
import numpy as np
from sentence_transformers import SentenceTransformer
from transformers import pipeline

Далее следует описание всего тела:

DOCUMENTS = [
    """Nimbus is a fictional note-taking app launched in 2023. The free plan,
    called Nimbus Lite, allows up to 50 notes and 1 GB of storage. There are no
    collaboration features on the free plan.""",
    """Nimbus Pro costs 8 dollars per month billed annually, or 10 dollars billed
    monthly. Pro removes the note limit, gives 50 GB of storage, and unlocks
    real-time collaboration with up to 5 people per note.""",
    """Nimbus stores all notes encrypted at rest using AES-256. End-to-end
    encryption is only available on the Pro plan and must be enabled manually in
    Settings > Security. Once enabled it cannot be turned off for that note.""",
    """The Nimbus mobile app supports offline editing. Changes made offline are
    queued and sync automatically the next time the device is online. If two
    devices edit the same note offline, Nimbus keeps both versions and flags a
    conflict for the user to resolve.""",
    """Nimbus offers a 30-day refund policy on all paid plans, no questions asked.
    Refunds are processed to the original payment method within 5 business days.
    Annual plans cancelled after 30 days are not refundable but stay active until
    the end of the billing period.""",
    """Nimbus support is available via email at help@nimbus.example and live chat.
    Live chat is only staffed for Pro customers, Monday to Friday, 9am to 6pm UTC.
    Free-plan users receive email support with a typical 48-hour response time.""",
]def chunk_text(text, chunk_size=60, overlap=15):
    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
    return chunksprint("Loading models (first run downloads them)...")
embedder = SentenceTransformer("all-MiniLM-L6-v2")
generator = pipeline("text2text-generation", model="google/flan-t5-base")# Index the documents once at startup
chunks = []
for doc_id, doc in enumerate(DOCUMENTS):
    for c in chunk_text(doc):
        chunks.append({"doc_id": doc_id, "text": c})
chunk_vectors = embedder.encode(
    [c["text"] for c in chunks], normalize_embeddings=True
)def retrieve(question, k=3):
    q_vec = embedder.encode([question], normalize_embeddings=True)[0]
    scores = chunk_vectors @ q_vec
    top_idx = np.argsort(scores)[::-1][:k]
    return [(chunks[i]["text"], float(scores[i])) for i in top_idx]def rag_answer(question, k=3):
    retrieved = retrieve(question, k=k)
    context = "\n".join(text for text, _ in retrieved)
    prompt = (
        "Answer the question using only the context below. "
        "If the answer is not in the context, say you don't know.\n\n"
        f"Context:\n{context}\n\nQuestion: {question}\nAnswer:"
    )
    out = generator(prompt, max_new_tokens=80)[0]["generated_text"]
    return out.strip()if __name__ == "__main__":
    print("RAG ready. Ask about Nimbus (or type 'quit').\n")
    while True:
        q = input("You: ").strip()
        if q.lower() in {"quit", "exit", ""}:
            break
        print("Nimbus bot:", rag_answer(q), "\n")

Начните с:

python rag.py

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

Доказательство ценности с помощью теста A/B

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

def no_rag(question):
    out = generator(f"Question: {question}\nAnswer:", max_new_tokens=80)
    return out[0]["generated_text"].strip()

Затем сравнение:

q = "Can free-plan Nimbus users use live chat support?"
print("WITHOUT context:", no_rag(q))
print("WITH context   :", rag_answer(q))

Ответ без контекста — это лишь догадка, тогда как ответ с контекстом должен указывать, что чат в реальном времени доступен только пользователям премиум-тарифа, а бесплатные пользователи получают ответы по электронной почте. Попробуйте три разных вопроса. Этот пример A/B является самым убедительным демонстрацией технологии RAG, поэтому используйте его для объяснения концепции другим.

Обновления в зависимости от пользы

Когда у вас уже есть рабочая система, наиболее важными становятся следующие обновления, примерно в порядке их пользы:

1. Отображение источников. Функция уже возвращает значение retrieved, поэтому необходимо выводить фрагменты текста под каждым ответом. Успех или неудача настоящих продуктов на основе технологии RAG зависит именно от этого. Ответ с указанием источника кажется надежным, тогда как ответ без указания источника — это всего лишь догадка, представленная в красивом формате.

Сначала получите дополнительные результаты, затем отфильтруйте их. Увеличьте значение k до 5 и установите минимальный балл, отбрасывая все значения ниже примерно 0.25. Если ни один фрагмент не проходит проверку, ответьте «Я не знаю» и полностью пропустите вызов модели. Это устранит большинство галлюцинаций, подобных той, что была в упражнении на шаге 5.

Настоящий хранилище векторов. Команда chunk_vectors @ q_vec сканирует каждый фрагмент, что подходит для сотен элементов, но медленно работает с миллионами. Инструмент Chroma (pip install chromadb) индексирует векторы для быстрого поиска и сохраняет их на диске, так что при запуске не требуется повторная обработка данных. Концепции вкладки, хранения и косинусного поиска остаются прежними; Chroma лишь позволяет выполнять их в масштабах.

Более мощный бесплатный генератор. Flan-T5 имеет небольшие размеры и иногда краток в ответах. Для получения более качественных локальных ответов установите Ollama, который позволяет запускать такие модели, как Llama 3.2 или Phi, локально, и замените вызов generator(...) запросом к его локальному API. Бесплатный уровень API от Google Gemini или Groq также работает путем замены только строки генерации в rag_answer. Изменяется лишь одна функция, в то время как механизм поиска остается прежним.

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

Диагностика проблем

RAG — это поиск плюс помощник для написания текста, объединённые вместе, причём почти каждая проблема связана с одной из этих частей. Неверный ответ при получении правильного фрагмента данных — это проблема генерации или запроса. Неверный ответ, когда правильный фрагмент вообще не появился, — это проблема получения данных, связанная с разбиением на фрагменты, вкладками или параметром k. Сначала определите, какая часть неисправна, прежде чем что-либо менять. Функции retrieve() и rag_answer() оставлены отдельными именно для того, чтобы можно было проверять каждую часть по отдельности, и именно это разделение является самой важной практикой здесь.

Устранение неполадок

• Медленное или неудачное загрузчение: модели скачиваются с Hugging Face при первом запуске и хранятся в кэше по пути ~/.cache/huggingface. После кэширования они работают в автономном режиме.

Проблемы с памятью или медленная работа: используйте модель google/flan-t5-small с параметром chunk_size равным 40. Эта модель достаточно мала, чтобы не замедлять работу.

Проблемы с установкой torch: запросите версию, предназначенную только для CPU, из индекса PyTorch: pip install torch --index-url https://download.pytorch.org/whl/cpu.

Очень короткие ответы: увеличьте значение max_new_tokens, поскольку Flan-T5 склонен прекращать работу рано. Более крупный генератор, такой как тот, что предоставляется Ollama, поможет лучше, чем любые настройки.

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

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