Практические советы: Создание системы RAG с нуля — практическое руководство без затрат на API

В следующих заметках описывается практический подход к созданию системы RAG с нуля — без использования API и с акцентом на практическую реализацию. Основное внимание уделяется контрактам, проверкам и шаблонам кода, а не мотивирующим формулировкам. На этапе обзора сначала запишите контракт: необходимые входные данные, сигнал успешного выполнения и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Документируйте как успешный, так и восстановительный сценарии работы. Повторные попытки, проверки человеком и обработка ошибок являются неотъемлемой частью продукта, а не элементами последующей доработки.

Что такое RAG на самом деле (60 секунд)

Подход, заключающийся в рассмотрении RAG как измеримой поверхности, наиболее эффективен. Сначала соберите один идеальный пример транскрипции, один пример сбоя и записку о возврате к предыдущему состоянию, прежде чем расширять объём работы. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. Когда какой-то шаг терпит неудачу, причина сбоя должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Разделяйте политику разбиения данных на части и политику их поиска. Изменение одной из них не должно вынуждать переписывать другую при изменении показателей качества.

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

Шаг 0 — Настройка

Этап настройки шага 0 работает наилучшим образом, если рассматривать его как измеримую поверхность. Соберите один идеальный пример вывода, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объема работ. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия создаваемым элементам, определите критерии успеха и не соглашайтесь на молчаливое частичное выполнение задачи. Разделяйте политику разбиения данных на части и политику их извлечения. Изменение одной из них не должно приводить к переписыванию другой при изменении показателей качества.

pip install sentence-transformers transformers torch numpy

Шаг 1 — База знаний, которую модель никогда раньше не видела

Этап знаний Step 1 A работает наилучшим образом, когда его рассматривают как измеримую поверхность. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объема работ. Записывайте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные счета при переходе от демо-версии к общедоступным средам. Установите лимиты на количество токенов за один ход и за сессию. Инструменты агентов активно расширяют контекст; жесткие ограничения не позволяют демо-версиям превращаться в неожиданные счета. Этап знаний Step 1 A работает наилучшим образом, когда его рассматривают как измеримую поверхность. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объема работ. Документируйте одновременно успешный путь выполнения и путь восстановления. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не элементами последующей доработки.

# 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.""",
]

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], "...")

Шаг 3 — Эмбеддинги: преобразование текста в векторы

На этапе встраивания данных на шаге 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)

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

Шаг 4 — Получение данных: поиск фрагментов, отвечающих на вопрос

На этапе поиска информации в рамках Шага 4 необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг, исходя из известной точки контроля, без необходимости угадывать скрытое состояние. Регистрируйте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости на раннем этапе предотвращает неожиданные счета при переходе с демо-среды в общедоступные среды. Указывайте конкретные фрагменты текста, на которых основан ответ. Без цитат операторы не смогут отличить галлюцинации от пробелов в индексации. На этапе поиска информации в рамках Шага 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]}...")

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

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

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]}...")

Шаг 6 — Соберите всё вместе

При работе над этапом 6 «Внедрение» сначала запишите контракт: необходимые входные данные, сигнал о успехе и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия элементам, определите критерии успеха и не допускайте молчаливого частичного выполнения задачи. Оцените уровень воспроизводимости на фиксированном наборе вопросов перед настройкой подсказок. Частая смена подсказок редко помогает улучшить качество поиска.

# 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

Этап 7 — Подтверждение эффективности RAG (тест A/B)

При работе над этапом Step 7 Prove RAG сначала запишите условия работы системы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Запишите также время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. Измерьте уровень воспроизведения информации на фиксированном наборе вопросов перед настройкой подсказок. Частая смена подсказок редко помогает улучшить качество поиска. При работе над этапом Step 7 Prove RAG сначала запишите условия работы системы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Одновременно задокументируйте успешный и восстановительный сценарии работы. Повторные попытки, проверка человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки.

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))

Шаг 8 — Улучшить (выберите то, что вас интересует)

Шаг 8 «Улучшить» работает наилучшим образом, когда его рассматривают как измеримую основу. Соберите один идеальный пример выполнения, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объема работ. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. Когда какой-либо шаг терпит неудачу, причина сбоя должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Разделяйте политику разбиения на части и политику извлечения данных. Изменение одной из них не должно принуждать к переписыванию другой при изменении показателей качества.

Ментальная модель, которую следует сохранять

Ментальная модель данного этапа работает наилучшим образом, когда её рассматривают как измеримую структуру. Соберите один идеальный пример результата, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объёма работ. Рассматривайте этот этап как контракт между входными данными и проверенными результатами. Дайте названия создаваемым элементам, определите критерии успеха и не соглашайтесь на молчаливое частичное выполнение задач. Установите лимиты на количество операций за раз и за сессию. Инструменты агентов активно расширяют объём контекста; строгие ограничения предотвращают появление неожиданных счетов.

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

Этап устранения неполадок работает наилучшим образом, когда его рассматривают как измеримую поверхность. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию до расширения объема работ. Записывайте временные показатели, а также стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. Разделяйте политику разбиения данных на части и политику их извлечения. Изменение одной из них не должно приводить к необходимости переписывания другой при изменении показателей качества. Этап устранения неполадок работает наилучшим образом, когда его рассматривают как измеримую поверхность. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию до расширения объема работ. Документируйте одновременно успешный путь выполнения и путь восстановления. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не элементами последующей доработки.

Чек-лист операций

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

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

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

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

Документируйте как успешный, так и восстановительный пути работы. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не последующими улучшениями.

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

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

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