Stwórz lokalny pipeline RAG w Pythonie bez kosztów API

Kompletny pipeline generacji wzbogaconej o wyszukiwanie można stworzyć i uruchomić wyłącznie na komputerze osobistym, korzystając tylko z procesora, za darmo, przy użyciu modeli otwartego oprogramowania – bez klucza OpenAI, karty kredytowej i ograniczeń szybkości. Chodzi o zrozumienie każdego elementu, a nie tylko kopiowanie „czarnej skrzynki”.

Najlepszym sposobem na opanowanie tematu jest praktyka: miej otwarty terminal oraz plik Python o nazwie rag.py, uruchamiaj każdy blok i przewiduj wynik przed jego wykonaniem. Po większości kroków pojawiają się krótkie ćwiczenia, w których odbywa się główna część nauki.

Problem, który rozwiązuje RAG

Model językowy zna tylko to, co zawierały jego dane treningowe. Prywatne notatki, wiki firmy, spotkanie z zeszłego tygodnia czy jakikolwiek nieznany dokument są dla niego niewidoczne. Gdy zostanie o to zapytany, albo przyznaje brak wiedzy, albo, co gorsza, z pewnością siebie coś wymyśla.

RAG dzieli to zadanie na dwie części:

1. Pobierz: znajdź fragmenty własnych dokumentów, które odnoszą się do pytania. To jest problem wyszukiwania.
2. Stwórz: przekaż te fragmenty modelowi z poleceniem udzielenia odpowiedzi na podstawie nich.

Model przestaje zgadywać i zaczyna czytać. Wszystko inne to elementy techniczne niezbędne do prawidłowego wykonania tych dwóch kroków.

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

Besplatny zestaw narzędzi

Cały system składa się z trzech komponentów. sentence-transformers przekształca tekst w wektory, wykorzystując mały model działający na CPU, który można pobrać tylko raz. numpy zajmuje się obliczeniami podobieństwa, a cała „baza danych wektorowych” mieści się w około pięciu liniach kodu, dzięki czemu jej działanie pozostaje widoczne. transformers z modelem Flan-T5 generuje odpowiedź, ponieważ jest dostosowany do instrukcji, działa na CPU i nie wymaga API.

Zainstaluj wszystko tylko raz:

pip install sentence-transformers transformers torch numpy

Podczas pierwszego załadunku każdy model pobiera swoje wagi z Hugging Face. Moduł embedder waży około 80 MB, a flan-t5-base około 1 GB. Następnie są one przechowywane w pamięci cache i umożliwiają pracę offline. Przy ograniczonej ilości pamięci dyskowej lub RAM flan-t5-small (około 300 MB) może zastąpić model bazowy, choć z niższą jakością, to z znacznie większą szybkością.

Krok 1: korpus, którego model nie może znać

RAG jest najłatwiej zrozumieć przy użyciu dokumentów, których żaden model nie mógłby zobaczyć, dlatego skrypt rozpoczyna się od małej, fikcyjnej witryny firmowej. Stwórz plik rag.py i wklej do niego dokumenty:

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

Wiki zawiera więcej wpisów:

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

Zanim cokolwiek zbudujesz, udowodnij, że problem istnieje. W shellu Pythona zapytaj prosty model o 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"])

Wygeneruje coś, ale to coś będzie błędne lub puste, ponieważ jest to tylko przypuszczenie. Pamiętaj o tej odpowiedzi, ponieważ reszta instrukcji ją poprawia.

Krok 2: dzielenie na fragmenty

Całe dokumenty nie trafiają do modelu. Są one dzielone na mniejsze fragmenty z dwóch powodów: wyszukiwanie staje się dokładniejsze, ponieważ model znajduje akurat ten akapit, który odpowiada na pytanie, a nie cały podręcznik, oraz okno kontekstowe modelu jest ograniczone.

Dodaj funkcję dzielenia na fragmenty:

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

Następnie utwórz listę fragmentów, odnotowując przy tym, z którego dokumentu pochodzi każdy fragment:

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

Przecięcie występuje, ponieważ dany fakt może obejmować kilka obszarów, np. zasady zwrotu pieniędzy określone jako „w ciągu 5 dni” i „w dni robocze”. Bez takiego przecięcia żaden z fragmentów nie zawiera całego faktu. Powtórzenie kilku słów w każdym miejscu przecięcia pozwala faktom przetrwać to rozdzielenie.

Ćwiczenie: uruchom ponownie z parametrem chunk_size=20, aby uzyskać większą liczbę mniejszych fragmentów, a następnie spróbuj chunk_size=500, przy którym każdy z tych krótkich dokumentów stanie się pojedynczym fragmentem. Przed uruchomieniem spróbuj odgadnąć, który ustawienie zapewnia dokładniejszy wynik, ale niesie ryzyko utraty otaczającego kontekstu. Małe fragmenty są dokładne, ale pozbawione kontekstu, a to napięcie leży u podstaw procesu dzielenia na fragmenty. Nie istnieje uniwersalnie poprawna wielkość.

Krok 3: embeddingi

Embedding to wektor liczb kodujący znaczenie tekstu. Fragmenty tekstu o podobnym znaczeniu stają się wektorami wskazującymi w tym samym kierunku, nawet jeśli nie mają wspólnych słów, dlatego „Ile to kosztuje?” znajduje się obok „Jaka jest cena?”

Importuj model embeddingu:

from sentence_transformers import SentenceTransformer

Następnie wstaw każdy fragment:

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)

Każdy fragment odpowiada teraz wektorowi o 384 wymiarach, przy czym 384 to po prostu liczba liczb, których używa ten konkretny model do wyrażania znaczenia.

Ćwiczenie: zobacz, jak znaczenie funkcjonuje, wykonując mały eksperyment:

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

Wyrażenie a @ b to iloczyn skalarny, który dla wektorów jednostkowych odpowiada podobieństwu kosinowemu. Wartość 1,0 oznacza tę samą kierunek, a 0 brak związku. Pierwsza para powinna uzyskać znacznie wyższy wynik niż druga, mimo że frazy nie zawierają żadnych wspólnych słów – to właśnie jest wyszukiwanie semantyczne. Spróbuj odgadnąć oba liczby z dokładnością do 0,2 przed uruchomieniem.

Krok 4: wydobycie wyników

W kroku wyszukiwania należy wstawić pytanie, porównać je z każdym wektorem fragmentu i zachować najlepsze wyniki. Ponieważ wektory są znormalizowane, „porównanie” sprowadza się do mnożenia macierzy.

import numpy as np

Funkcja wyszukiwania:

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

Część dotycząca ceny powinna znaleźć się na górze z najwyższą oceną, mimo że w pytaniu było napisane „koszt”, a w dokumencie „koszty 8 dolarów miesięcznie”. Chodzi tu o dopasowanie znaczenia, a nie słów kluczowych.

Ćwiczenie: spróbuj z trzema pytaniami i zobacz, co się pojawi:

• retrieve("Czy mogę otrzymać zwrot pieniędzy?") powinien zwrócić część dotyczącą zwrotu, chociaż słowo „zwrot” nigdy nie pojawiło się w pytaniu.
• retrieve("Czy moje dane są bezpieczne?") powinien pokazać część dotyczącą szyfrowania.
• retrieve("Czy nadal działa bez połączenia z Internetem?") ma przywrócić część dotyczącą synchronizacji offline.

Wielu ludzi lekceważy ten etap. Gdy system przywraca błędne fragmenty, żadne sprytne instrukcje nie mogą uratować odpowiedzi. Co włożysz, to i dostaniesz, dlatego tu najpierw należy poświęcić wysiłki na debugowanie.

Krok 5: generowanie

Teraz te dwie części się łączą. Stwórz instrukcję, która przekaże pobrane fragmenty do Flan-T5 i poinstruuje go, aby odpowiadał wyłącznie na ich podstawie.

from transformers import pipeline

Funkcja generowania:

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

Porównaj wynik z błędną lub pustą odpowiedzią z kroku 1. Model i pytanie są identyczne, a jedyną zmianą jest dostarczenie odpowiedniego tekstu do przeczytania. To właśnie ta różnica stanowi całą wartość RAG.

Ćwiczenie: postaw pytanie, którego dokumenty nie obejmują, na przykład o kliencie desktopowym Linux dla Nimbus, i wywołaj z nim funkcję rag_answer. Poprawnie skonfigurowane rozwiązanie przyzna, że nie wie odpowiedzi. Jeśli twoje rozwiązanie wymyśli odpowiedź, jest to prawdziwy problem zwany halucynacją pomimo odzyskania informacji, a późniejszy rozdział poświęcony jest temu zagadnieniu.

Skończony skrypt

Oto cały plik rag.py jako pojedynczy program, który można uruchomić i który zawiera pętlę interaktywną. Zaczyna się od importów:

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

Następuje całe ciało:

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

Zacznij od:

python rag.py

Następnie spróbuj zadań takich jak okno zwrotu pieniędzy, czy użytkownicy darmowi mają dostęp do czatu na żywo, lub jaka jest wielkość darmowej przestrzeni przechowywania.

Dowodzenie wartości za pomocą testu A/B

Jeden eksperyment jasno ilustruje ten problem, zadając to samo pytanie z użyciem kontekstu oraz bez niego. Najpierw funkcja bazowa:

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

Następnie porównanie:

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

Odpowiedź bez kontekstu to tylko domysł, natomiast ta z kontekstem powinna wskazać, że czat na żywo jest dostępny tylko dla użytkowników Pro, a użytkownicy darmowi otrzymują wiadomości e-mail. Spróbuj trzech różnych pytań. Ten przykład typu A/B jest najprzekonującą demonstracją technologii RAG, więc zachowaj go do wyjaśniania tej koncepcji innym.

Ulepszenia według korzyści

Gdy mamy już działający system, te ulepszenia są najważniejsze, mniej więcej w kolejności ich korzyści:

1. Pokazuj źródła. Funkcja już zwraca wartość retrieved, więc wyświetl fragmenty tekstu pod każdą odpowiedzią. Od tego zależy sukces lub porażka prawdziwych produktów opartych na technologii RAG. Odpowiedź z odniesieniami do źródeł jest wiarygodna, natomiast ta bez nich to tylko domysł przedstawiony w ładnym formacie.

Pobierz dodatkowe wyniki, a następnie przefiltruj je. Ustaw k na 5 i określ minimalną ocenę, odrzucając wszystko poniżej mniej więcej 0,25. Gdy żaden fragment nie przejdzie testu, odpowiedz „Nie wiem” i całkowicie pomiń wywołanie modelu. To eliminuje większość halucynacji, takich jak ta z ćwiczenia w kroku 5.

Prawdziwy magazyn wektorów. chunk_vectors @ q_vec przeszukuje każdy fragment, co jest w porządku przy setkach, ale wolne przy milionach. Chroma (pip install chromadb) indeksuje wektory w celu szybkiego wyszukiwania i zapisuje je na dysku, dzięki czemu nic nie musi być ponownie embedowane przy uruchamianiu. Koncepcje embedowania, przechowywania i wyszukiwania kosinusowego pozostają takie same, a Chroma realizuje je tylko w dużych skali.

Mocniejszy generator darmowy. Flan-T5 jest mały i czasami zbyt zwięzły. Aby uzyskać lepsze lokalne odpowiedzi, zainstaluj Ollamę, która uruchamia modele takie jak Llama 3.2 lub Phi lokalnie, i zastąp wywołanie generator(...) żądaniem do jego lokalnej API. Darmowy poziom API od Google’s Gemini lub Groq również działa poprzez zastąpienie jedynie linii generowania w rag_answer. Zmienia się tylko jedna funkcja, podczas gdy mechanizm wyszukiwania pozostaje bez zmian.

Ocena. Napisz dziesięć pytań z znanych odpowiedzi, przepuść je przez rag_answer i sprawdź poprawność. To nie jest efektowne, ale to właśnie odróżnia prototyp od produktu – bo tego, czego nie mierzy się, nie da się ulepszyć.

Diagnozowanie problemów

RAG to szukanie połączone z asystentem do pisania, które zostały ze sobą połączone, przy czym niemal każdy problem wynika z jednej z tych dwóch części. Błędna odpowiedź przy poprawnie pobraanym fragmencie to problem z generowaniem lub promptem. Błędna odpowiedź, gdy poprawny fragment w ogóle się nie pojawił, to problem z pobieraniem danych dotyczący podziału na fragmenty, embeddingów lub wartości k. Zanim cokolwiek zmienisz, określ, która z tych części jest uszkodzona. Zachowanie funkcji retrieve() i rag_answer() jako odrębnych funkcji ma na celu umożliwienie osobnego sprawdzenia każdej z tych części, a ta separacja jest tutaj najważniejszym nawykiem.

Rozwiązywanie problemów

• Słabe lub nieudane pobieranie: modele są pobierane po raz pierwszy z Hugging Face i przechowywane w pamięci cache pod adresem ~/.cache/huggingface. Po ich zapisaniu w cache działają offline.

Nacisk na pamięć lub wolne działanie: użyj modelu google/flan-t5-small z wartością chunk_size równą 40. Model embeddingowy jest na tyle mały, że nie spowolni pracy.

Problemy z instalacją torch: pobierz wersję dostępną wyłącznie na CPU z indeksu PyTorch, używając polecenia pip install torch --index-url https://download.pytorch.org/whl/cpu.

Bardzo krótkie odpowiedzi: zwiększ wartość max_new_tokens, ponieważ Flan-T5 ma tendencję do wcześniejszego zatrzymywania się. Większy generator, np. ten dostarczany przez Ollama, pomoże bardziej niż jakiekolwiek modyfikacje.

Zawsze najlepszy jest jeden chunk: albo dokumenty są niemal identyczne, albo wartość k jest zbyt niska. Sprawdź surowe wyniki, ponieważ pokazują one, w jaki sposób model sortuje elementy.

Rozwiązywanie ćwiczeń umożliwia głębsze zrozumienie RAG niż to, co większość ludzi uzyskuje dzięki prostym narzędziom bibliotecznych.