RAG w prostym Pythonie: dzielenie na fragmenty, TF-IDF, podobieństwo kosinowe i generowanie promptów
Stwórz generację wzbogaconą o wyszukiwanie, używając wyłącznie modułu matematycznego: dzielenie na fragmenty za pomocą okna przesuwającego się, wektory TF-IDF, ranking kosinusowy, sprecyzowany prompt oraz jego ograniczenia.
Źródła: Building RAG From First Principles with just plain python .
Klasyczny tutorial dotyczący RAG polega na zainstalowaniu pięciu bibliotek i wywołaniu metody .from_documents(). Po dziesięciu minutach demo działa, ale nikt nie rozumie, jak to funkcjonuje. Gdy proces wyszukiwania zwraca bezużyteczne dane lub model tworzy błędne informacje, jesteśmy w tarapatach, ponieważ to biblioteka wykonała za nas całą pracę intelektualną.
Tutaj pipeline jest budowany w odwrotny sposób, przy użyciu zwykłego Pythona, tylko modułu math oraz jednego wywołania modelu LLM obsługującego tekst prosty na końcu. Żadnych modeli embedding, żadnego FAISS, żadnego Pinecone. Znajomość tych mechanizmów oznacza, że gdy później skorzystamy z LangChain lub prawdziwej bazy danych wektorowej, będziemy rozumieć, co się pod spodem dzieje. Na końcu mamy działający pipeline, a co ważniejsze – pojęcie o każdej liczbie, którą generuje.
Problem, który rozwiązuje RAG
Wiedza modelu językowego zostaje zamrożona po zakończeniu treningu i znajduje się wyłącznie w jego wagach. Aby mógł odpowiadać na pytania dotyczące wewnętrznych dokumentów firmy, kodu źródłowego lub wiadomości z dziś rana, obie oczywiste opcje okazują się nieudane:
- Trening modelu od nowa. Jest to proces powolny i kosztowny, a zmiany wprowadzone jutro wymagają kolejnej rundy treningu.
- wklejenie wszystkiego do promptu. Ta metoda nie jest skalowalna. Nawet okno o pojemności miliona tokenów nie może pomieścić 50 000 dokumentów, a badania nad modelami długiego kontekstu wielokrotnie wskazują na efekt „utraty informacji pośrednich”, przy którym dane pogrzebane głęboko w długim promptie są wykorzystywane gorzej, mimo że mieścią się w jego zakresie. Dodatkowe tokeny oznaczają również większą opóźnioność i wyższy koszt za każdą próbę wywołania modelu.
Prawdziwy sekret RAG jest prostszy, niż się wydaje. Dokumenty pozostają w zewnętrznym magazynie, a podczas odpowiadania na pytanie pobierane są i przekazywane modelowi tylko te nieliczne fragmenty, które są do tego pytania istotne. Model LLM nigdy nie przechowuje twoich danych – za każdym razem otrzymuje nowy, odpowiedni fragment.
Zatem każdy system RAG składa się z dwóch połączonych ze sobą komponentów: odnajdywania, które znajduje odpowiedni tekst w znacznie większej kolekcji na podstawie danego zapytania, oraz generowania, które przekazuje ten tekst wraz z zapytaniem do modelu LLM i otrzymuje odpowiedź. W dalszej części omówimy, jak zbudować oba te elementy z podstawowych składników.
Krok 1: dzielenie na fragmenty
Zamiana całego 50-stronicowego pliku PDF na jeden wektor niszczy niemal wszystkie użyteczne informacje, ponieważ szczegóły, które mogłyby odpowiedzieć na pytanie, zostają zsumowane i tracą znaczenie. Pierwszym krokiem jest podzielenie dokumentów na małe, precyzyjne fragmenty.
Najprostszą metodą jest okno przesuwane o stałej wielkości mierzone w słowach, z nakładaniem się sąsiednich fragmentów:
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
Ta klauzula ochronna to nie dekoracja. Jeśli overlap osiągnie lub przekroczy wartość chunk_size, pozycja początkowa przestaje się zmieniać, a pętla nigdy się nie kończy. Dostosowywanie wartości overlap w górę „dla większego kontekstu” sprawia, że łatwo o błąd, dlatego funkcja odmawia działania zamiast cicho się zawiesić.
Współwystępowanie fragmentów ma większe znaczenie, niż się wydaje. Bez niego zdanie odpowiadające na pytanie może przekraczać granicę fragmentu i zostać podzielone, w wyniku czego żaden z fragmentów nie zawiera kompletnej informacji, którą można by odzyskać. Dzięki współwystępowaniu fragmenty te pozostają nienaruszone przynajmniej w jednym z nich.
Jednym z przypadków krawędziowych, o których należy wiedzieć, jest to, że okna słów o stałej długości ignorują strukturę zdań i akapitów, więc mogą przeciąć zdanie lub wiersz tabeli na pół. Dzielenie z uwzględnieniem struktury zdań lub akapitów unika tego problemu, ale powoduje nierównomierne rozmiary fragmentów, ponieważ zdanie składające się z 10 słów i zdanie z 150 słowami stają się po jednym fragmencie. To zaburza rozkład tokenów oraz wpływ poszczególnych fragmentów na wyniki późniejszych obliczeń. Żaden sposób dzielenia tekstu nie jest wolny od kompromisów, a wybór zależy naprawdę od tego, który problem wolisz naprawiać.
Krok 2: przekształcanie tekstu w liczby za pomocą TF-IDF
Komputer nie może bezpośrednio porównywać podobieństwa dwóch ciągów znaków. Każdy kolejny krok wymaga wektorów, czyli list liczb, na których można wykonywać operacje arytmetyczne. Klasycznym, niezależnym od zależności sposobem na ich tworzenie jest TF-IDF – częstotliwość występowania terminu pomnożona przez odwrotną częstotliwość dokumentu.
- TF pokazuje, jak często dana słowo występuje w danym fragmencie. Wysoka liczba wystąpień słowa „nerka” wskazuje, że fragment dotyczy nerek.
- IDF mierzy, jak rzadkie jest dane słowo we całej kolekcji. Słowa takie jak „the” i „is” pojawiają się wszędzie i nie dają żadnych informacji na temat tego, jaki fragment się czyta, dlatego ich wynik jest bliski zera. Rzadkie słowo w fragmencie stanowi znacznie silniejszy sygnał.
Kod tworzy słownictwo ze wszystkich fragmentów, oblicza IDF dla każdego terminu, a następnie przekształca każdy fragment w wektor 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]
Wypróbuj to na trzech fragmentach tekstu, a coś stanie się jasne. Słowo obecne w niemal każdym fragmencie, takie jak „the”, otrzymuje wartość IDF bliską zera, więc jego powtarzanie nic nie dodaje do wektora. Rzadkie, specyficzne dla danego tematu słowo ma wyraźnie dodatnią wartość IDF. TF-IDF to dosłownie formuła służąca do określenia, czy dane słowo dostarcza jakichś charakterystycznych informacji o fragmencie.
Negatywną stroną jest skalowalność. W przypadku 100 000 dokumentów słownictwo może zawierać ponad 500 000 unikalnych słów, więc każdy wektor fragmentu, nawet trzywyrazowego, składa się z 500 000 liczb, z których większość to zera. Takie rzadkie wektory są kosztowne w przechowywaniu i porównywaniu. To główny powód, dla którego systemy produkcyjne przechodzą na gęste embeddingi neuronalne o rozmiarze 384 do 1536 wymiarów, w których prawie nie ma zer.
Krok 3: mierzenie istotności za pomocą podobieństwa kosinowego
Gdy kawałki tekstu traktujemy jako wektory, następnym pytaniem jest to, na jakim poziomie istotny jest dany kawałek dla zapytania. Surowa odległość nie jest odpowiednim narzędziem, ponieważ w przypadku tekstu ważniejszy jest kąt między wektorami niż ich długość. Kawałek, który wspomina „kot” raz, oraz ten, który powtarza to słowo dwadzieścia razy, dotyczą wciąż mniej więcej tego samego tematu, a podobieństwo kosinowe traktuje je jednakowo, nie zwracając uwagi na to, który z nich jest bardziej wyraźny.
W implementacji potrzebna jest iloczyn skalarny, wartość modułu oraz stosunek tych dwóch wielkości:
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
Podobieństwo kosinowe zasadniczo waha się od -1 do 1, a przy tym ustawieniu TF-IDF prawie zawsze pozostaje w przedziale od 0 do 1:
- 1,0 oznacza, że wektory wskazują dokładnie w tym samym kierunku – to maksymalne podobieństwo.
- 0,0 oznacza, że są ortogonalne i nie mają żadnego wspólnego wymiaru o znaczeniu; mówiąc prościej, zapytanie i fragment nie mają żadnych wspólnych słów o istotnym znaczeniu.
Ocena 0 nie jest anomalią. To oznacza, że system poprawnie informuje, iż fragment nie ma żadnego związku z zapytaniem.
Jedna uwaga dotycząca precyzji: spłaszczona formuła IDF może dać nieco ujemną wartość słowu, które pojawia się we wszystkich fragmentach, więc podobieństwo kosinowe czasami może lekko spaść poniżej zera. W przypadku rzeczywistego tekstu ma to niewielkie znaczenie, ale twierdzenie, że nigdy nie jest ujemne, byłoby przesadą.
Niedostatkiem TF-IDF w połączeniu z podobieństwem kosinowym są synonimy. Zapytanie zawierające słowo „automobile” wobec dokumentu z wyrazem „car” uzyskuje wynik bliski zeru, ponieważ uwzględnia się jedynie dosłowne pokrycie słownictwa, a nie koncepcję znaczenia. Modele gęstego embeddingu, które umieszczają synonimy blisko siebie, zostały stworzone właśnie po to, by wypełnić tę lukę.
Krok 4: narzędzie wyszukiwania
Gdy wektoryzacja i obliczanie podobieństwa są już gotowe, proces wyszukiwania polega na wektorowaniu każdego fragmentu dokładnie raz, wektorowaniu przychodzącego zapytania, obliczaniu wyniku dla każdego fragmentu w odniesieniu do niego, sortowaniu i zachowaniu najlepszych k elementów:
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]
Jeden detal może w ciszy zniszczyć wszystko. Zapytanie musi zostać przekształcone przy użyciu słownictwa oraz wartości IDF obliczonych na podstawie dokumentów, a słownictwo wywiedzione wyłącznie z samego zapytania jest błędem. Podobieństwo kosinowe zakłada wektory o równej długości, w których pozycja i oznacza to samo słowo we wszystkich z nich. Jeśli zapytaniu przydzieli się własne słownictwo, pozycja 5 może oznaczać „serce” w jednym wektorze, a „i” w drugim, co sprawia, że wynik podobieństwa staje się strukturalnie bezsensowny, a nie tylko nieco błędny.
Krok 5: uzupełnianie promptu
W tym kroku nie ma potrzeby używania matematyki – wystarczy dyscyplina. Łączysz pozyskane fragmenty z pytaniem użytkownika w prompt, który precyzyjnie wskazuje modelowi, jak ma się zachować:
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:"""
Dwa szczegóły tutaj nie są czysto estetyczne.
„Używaj TYLKO kontekstu poniżej.” Bez tej wyraźnej zasady model łączy pozyskane fragmenty z tym, co pamięta z treningu, i wtedy nie można już odróżnić, która część odpowiedzi pochodzi z wiarygodnych dokumentów, a która to domysł. Taka mieszanka, podana z pewnością siebie, wyrządza większą szkodę niż całkowite pominięcie RAG, ponieważ wydaje się, że ma źródło, podczas gdy w rzeczywistości go nie ma.
Numerowane etykiety źródeł oraz instrukcja cytowania. Dzięki nim odpowiedzi są śledzone. Gdy odpowiedź jest błędna – a tak się zdarza – oznaczone źródła pokazują, który fragment wprowadził model w błąd, czy sam fragment był przestarzały lub nieprawidłowy, czy też model błędnie odczytał poprawny fragment. Bez takich etykiet ustalenie przyczyny złej odpowiedzi polega w dużej mierze na domysłach.
Krok 6: połączenie wszystkiego
Cały proces obejmuje najpierw pobieranie danych, następnie ich wzbogacanie, a potem jedną wywołanie modelu LLM. Kluczową decyzją projektową jest trzymanie indeksowania i wyszukiwania jako dwóch odrębnych etapów zamiast jednej funkcji wykonywającej obie czynności:
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
Gdyby indeksowanie odbywało się wewnątrz funkcji dla każdej zapytania, każde z nich ponownie rozdzielałoby tekst, odbudowywało słownictwo i przeliczało wszystkie wektory TF-IDF dla całej kolekcji, mimo że nic się pomiędzy zapytaniami nie zmienia. Przy 10 000 dokumentach i 1 000 zapytaniach oznaczałoby to 999 identycznych procesów odbudowy. Optymalne jest utworzenie indeksu raz, a następnie wykonywanie wielu zapytań – taki jest również podejście FAISS, Chromy i Pinecone. To właściwa architektura od samego początku, a nie przedwczesna optymalizacja.
Co jest nie tak z tym systemem
Ograniczenia te wymagają szczerości, ponieważ nie są błędami. Wynikają bezpośrednio z powyższych decyzji projektowych.
- „Samochód” i „automobil” nigdy się nie dopasowują, ponieważ TF-IDF dostrzega jedynie dosłowne pokrycie słownictwa.