Jak dać maszynie „zrozumienie”?
W Lab 3 zbudowaliśmy Agenta, który potrafi używać narzędzi. Jednym z nich było kb.lookup – prosta wyszukiwarka oparta na słowach kluczowych. Działała, ale miała wadę: jeśli użytkownik zapytał o „auto”, a w bazie mieliśmy tylko „samochód”, system nie widział powiązań.
Dzisiaj to zmienimy. Wkraczamy w świat osadzeń (embeddings) i wektorowych baz danych. To technologia, która stoi za rewolucją RAG.
🎯 Cel na dziś
Zbudujemy własną wyszukiwarkę, która „rozumie” znaczenie tekstu, a nie tylko porównuje literki.
- Syntetyczne dane: Stworzymy kontrolowany zbiór dokumentów, aby móc zmierzyć jakość wyszukiwania.
- Chunking: Nauczymy się dzielić tekst na strawne dla modelu kawałki.
- Dense Retrieval (FAISS): Użyjemy biblioteki od Meta AI do szybkiego wyszukiwania wektorowego.
- Baseline (BM25): Porównamy nowoczesne AI z klasycznym algorytmem słów kluczowych.
Czym są te całe Embeddingi?
Komputery nie rozumieją słów, rozumieją liczby. Embedding to proces zmiany tekstu (zdania, akapitu) na długi ciąg liczb (wektor), np. [0.12, -0.98, 0.55, ...].
Magia polega na tym, że w tej matematycznej przestrzeni, teksty o podobnym znaczeniu znajdują się blisko siebie.
- „Król” – „Mężczyzna” + „Kobieta” ~ „Królowa”
- „Pies” będzie blisko „Szczeniak”, nawet jeśli nie dzielą zbyt wiele liter.
Dzięki temu możemy zrobić tzw. Semantic Search; szukamy po znaczeniu, a nie po słowach.
Krok 1: Dane i chunking (krojenie słonia 🐘)
Systemy RAG rzadko indeksują całe książki jako jeden rekord. Dlaczego? Bo LLM ma limit „pamięci” tzw. okno kontekstowe (Context Window). Musimy podzielić wiedzę na mniejsze fragmenty – chunki.
W tym laboratorium generujemy syntetyczne dokumenty z 5 dziedzin (AI, Sport, Gotowanie, Geo, Zdrowie), aby mieć 100% pewności, który dokument jest poprawną odpowiedzią (Ground Truth). Przygotujmy dane:
pip install pandas
import random
import pandas as pd
SEED = 42
random.seed(SEED)
TOPICS = {
"ai": [
"Large language models predict the next token using transformer architectures.",
"Embeddings map text into dense vectors enabling semantic search.",
"RAG combines retrieval with generation to ground responses."
],
"sport": [
"Marathon training plans balance long runs and recovery days.",
"Strength training improves running economy and power.",
"Interval sessions develop speed and lactate threshold."
],
"cooking": [
"Sourdough starter needs regular feeding to stay active.",
"Sous vide cooking keeps precise temperatures for tenderness.",
"Spices bloom in hot oil enhancing aroma and flavor."
],
"geo": [
"Rivers shape valleys through erosion and sediment transport.",
"Plate tectonics explains earthquakes and mountain building.",
"Deserts form where evaporation exceeds precipitation."
],
"health": [
"Sleep supports memory consolidation and hormonal balance.",
"Aerobic exercise benefits cardiovascular health and VO2 max.",
"Protein intake supports muscle repair and satiety."
]
}
def synth_docs(n_per_topic=40):
"""
Metoda synth_docs generuje syntetyczny zbiór dokumentów do testowania algorytmów wyszukiwania semantycznego. Dla każdego tematu z listy TOPICS tworzy określoną liczbę dokumentów (n_per_topic). Każdy dokument zawiera tekst złożony z dwóch losowo wybranych zdań z danego tematu oraz informację o temacie i unikalny identyfikator. Tak przygotowany korpus pozwala porównywać skuteczność różnych metod wyszukiwania na kontrolowanych danych, gdzie znana jest przynależność dokumentów do tematów.
:param n_per_topic:
:return:
"""
docs = []
for topic, seeds in TOPICS.items():
for i in range(n_per_topic):
base = random.choice(seeds)
noise = random.choice(seeds)
txt = f"{base} {noise} ({topic} #{i})"
docs.append({"doc_id": f"{topic}-{i}", "topic": topic, "text": txt})
return docs
DOCS = synth_docs(40)
Kolejnym krokiem będzie przygotowanie funkcji do dzielenia tekstu.
⚠️ Ważne: Parametr overlap (zakładka) jest kluczowy. Zapobiega sytuacji, w której ważne słowo kluczowe, paragraf, akapit zostaje przecięty na granicy dwóch chunków.
CHUNK_SIZE=280
OVERLAP=40
def simple_chunk(text, chunk_chars=280, overlap=40):
out = []
i = 0
while i < len(text):
j = min(len(text), i+chunk_chars)
out.append((i, j, text[i:j]))
if j == len(text): break
i = max(0, j-overlap)
return out
def build_chunks(docs, chunk_chars=280, overlap=40):
rows = []
for d in docs:
for k,(a,b,txt) in enumerate(simple_chunk(d["text"], chunk_chars, overlap)):
rows.append({"doc_id": d["doc_id"], "topic": d["topic"], "chunk_id": k, "start": a, "end": b, "chunk": txt})
return pd.DataFrame(rows)
df = build_chunks(DOCS, CHUNK_SIZE, OVERLAP)
Krok 2: Dense Retrieval z FAISS
Teraz zamienimy nasze chunki na wektory i wrzucimy do indeksu FAISS (Facebook AI Similarity Search). Użyjemy modelu all-MiniLM-L6-v2 – jest lekki, szybki i zaskakująco dobry.
pip install sentence-transformers faiss-cpu
import time, faiss
from sentence_transformers import SentenceTransformer
# 1. Ładujemy model "tłumaczący" tekst na liczby
MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2"
embedder = SentenceTransformer(MODEL_NAME)
# 2. Zamieniamy wszystkie chunki na macierz (listę wektorów)
def embed_texts(texts, batch_size=64):
return embedder.encode(texts, batch_size=batch_size, convert_to_numpy=True, normalize_embeddings=True).astype("float32")
chunks = df["chunk"].tolist()
embeddings = embed_texts(chunks, 64)
# 3. Budujemy indeks (baza wektorowa w RAM)
index = faiss.IndexFlatIP(embeddings.shape[1])
index.add(embeddings)
# 4. Wyszukiwanie
query_vector = embedder.encode(["What improves running?"], normalize_embeddings=True)
scores, ids = index.search(query_vector, k=3)
W tym momencie stworzyłeś Retriever. Możesz zadać pytanie, a on zwróci 3 najbardziej pasujące semantycznie fragmenty tekstu.
Krok 3: Stary mistrz BM25
Nie zawsze „AI” jest lepsze. W wyszukiwaniu nazw własnych, kodów produktów (np. „EAN-13”), czy specyficznych fraz, klasyczne wyszukiwanie słów kluczowych (BM25) często wygrywa.
pip install rank_bm25
W profesjonalnych systemach RAG często stosuję się Hybrid Search: Wynik = 0.7 * VectorSearch + 0.3 * KeywordSearch.
Dlatego w naszym laboratorium implementujemy też BM25 jako punkt odniesienia (baseline).
import re
from rank_bm25 import BM25Okapi
def tokenize(text: str):
return re.findall(r"[a-z0-9]+", text.lower())
bm25_corpus = [tokenize(c) for c in chunks]
bm25 = BM25Okapi(bm25_corpus)
Krok 4: Ewaluacja – liczby nie kłamią 📊
Przygotujmy funkcje dla naszych Retriver-ów, żeby łatwiej było nam porównać ich skuteczność.
import numpy as np
def retrieve_dense(query: str, k: int=5):
q = embed_texts([query], batch_size=1)
scores, idxs = index.search(q, k)
return [(float(scores[0][i]), df.iloc[idxs[0][i]].to_dict()) for i in range(k)]
def retrieve_bm25(query: str, k: int=5):
toks = tokenize(query)
scores = bm25.get_scores(toks)
idxs = np.argsort(scores)[::-1][:k]
return [(float(scores[i]), df.iloc[i].to_dict()) for i in idxs]
retrieve_dense("What improves running economy?", 3)[0]
retrieve_bm25("What improves running economy?", 3)[0]
Skąd wiesz, czy Twój system jest dobry? „Na oko”? Inżynierowie AI używają metryk. Zdefiniowaliśmy „Golden Set” – listę pytań i przypisanych do nich poprawnych kategorii.
Sprawdzamy 4 kluczowe metryki:
Recall@k:Czy poprawna odpowiedź znalazła się w ogóle w top-5 wyników? (często najważniejsze dla RAG)Precision@k:Ile śmieci (niepoprawnych wyników) zwrócił system?MRR(Mean Reciprocal Rank): Jak wysoko na liście była poprawna odpowiedź? (lepiej być 1. niż 5.)- nDCG: Bardziej zaawansowana metryka jakości rankingu.
import math
GOLDEN = [
("How do transformers predict tokens?", "ai"),
("What is an embedding used for?", "ai"),
("How does RAG work?", "ai"),
("How to train for a marathon?", "sport"),
("What improves running economy?", "sport"),
("What is a threshold workout?", "sport"),
("How to feed sourdough starter?", "cooking"),
("Why sous vide is precise?", "cooking"),
("How to bloom spices?", "cooking"),
("How do rivers shape valleys?", "geo"),
("What causes earthquakes?", "geo"),
("Why do deserts form?", "geo"),
("Why is sleep important?", "health"),
("Benefits of aerobic exercise?", "health"),
("Why eat protein?", "health"),
]
def dcg(rels):
return sum((rel / math.log2(i+2) for i, rel in enumerate(rels)))
def ndcg_at_k(rels, k):
rels_k = rels[:k]
ideal = sorted(rels_k, reverse=True)
denom = dcg(ideal) or 1e-9
return dcg(rels_k)/denom
def eval_query(q, target_topic, retriever, k=5):
hits = retriever(q, k=k)
rels = [1 if h[1]["topic"]==target_topic else 0 for h in hits]
rec = sum(rels) * 1.0 # dla jednego relewantnego tematu interpretujemy jako Recall@k (liczba trafień w top-k)
prec = sum(rels)/len(rels) if rels else 0.0
rr = 0.0
for i, r in enumerate(rels, start=1):
if r==1: rr = 1.0/i; break
ndcg = ndcg_at_k(rels, k)
return {"recall@k": rec, "precision@k": prec, "mrr": rr, "ndcg@k": ndcg}
def evaluate(golden, retriever, k=5):
rows = []
for q,t in golden:
rows.append({"query": q, "topic": t, **eval_query(q,t,retriever,k)})
return pd.DataFrame(rows)
K=5
dense_df = evaluate(GOLDEN, retrieve_dense, k=K)
bm25_df = evaluate(GOLDEN, retrieve_bm25, k=K)
summary = pd.DataFrame({
"metric": ["recall@k","precision@k","mrr","ndcg@k"],
"dense": [dense_df[m].mean() for m in ["recall@k","precision@k","mrr","ndcg@k"]],
"bm25": [bm25_df[m].mean() for m in ["recall@k","precision@k","mrr","ndcg@k"]],
})
print(summary)
Na koniec sprawdzimy, jak zmiana wielkości chunka i innych hiperparametrów wpływa na metryki. Takie badanie nazywamy Grid Search.
def run_setting(chunk_size, overlap, kk):
dff = build_chunks(DOCS, chunk_size, overlap)
embs = embedder.encode(dff["chunk"].tolist(), batch_size=64, convert_to_numpy=True, normalize_embeddings=True).astype("float32")
idx = faiss.IndexFlatIP(embs.shape[1]); idx.add(embs)
def retr(q, k):
qv = embedder.encode([q], convert_to_numpy=True, normalize_embeddings=True).astype("float32")
scores, ids = idx.search(qv, k)
return [(float(scores[0][i]), dff.iloc[ids[0][i]].to_dict()) for i in range(k)]
dfres = evaluate(GOLDEN, retr, k=kk)
return dfres[["recall@k","precision@k","mrr","ndcg@k"]].mean().to_dict()
grid = []
for cs in [50, 200, 400, 800]:
for ov in [0, 80]:
for kk in [1, 3, 5]:
met = run_setting(cs, ov, kk)
grid.append({"chunk_size": cs, "overlap": ov, "k": kk, **met})
grid_df = pd.DataFrame(grid).sort_values(["k","recall@k"], ascending=[True, False])
print(grid_df.head(10))
# Wyniki możemy logować w następujący sposób
import os
from datetime import datetime
def log_row(path: str, row: dict):
exists = os.path.exists(path)
import csv
with open(path, "a", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=row.keys())
if not exists:
w.writeheader()
w.writerow(row)
log_row("lab04_logs.csv", {
"timestamp": datetime.utcnow().isoformat(),
"model_name": "all-MiniLM-L6-v2",
"index_type": "faiss-flatip",
"k": K,
"chunk_size": CHUNK_SIZE,
"overlap": OVERLAP,
"dense_recall@k": float(summary.loc[summary.metric=='recall@k','dense'].values[0]),
"bm25_recall@k": float(summary.loc[summary.metric=='recall@k','bm25'].values[0]),
})
