Створіть локальний пайплайн 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))

Відповідь без контексту — це лише здогадка, тоді як відповідь із контекстом має зазначати, що живий чат доступний лише для користувачів Pro, а безкоштовні користувачі отримують електронний лист. Спробуйте три різні запитання. Цей варіант 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, ніж це можливо через використання лише бібліотечних інтерфейсів.