Zaawansowany RAG — chunking strategy, hybrid search i reranking w produkcji
Prosty RAG z RecursiveCharacterTextSplitterem zawodzi przy złożonych dokumentach. Jak wdrożyć chunking semantyczny, hybrid search z BM25, RRF fusion i Cross-Encoder reranking — z pełnym kodem produkcyjnym i ewaluacją RAGAS.
Prosty RAG wygląda imponująco na demo. Wrzucasz PDF, embeddujesz chunki z RecursiveCharacterTextSplitterem, odpytujesz Qdrant i dostajesz sensowną odpowiedź. Problem pojawia się w produkcji — przy dokumentach prawnych liczących 400 stron, specyfikacjach technicznych pełnych tabel i wzorów, wielojęzycznych bazach wiedzy z terminologią branżową. Metryki RAGAS spadają poniżej akceptowalnego poziomu i zaczynasz dostawać zgłoszenia: "model halucynuje", "nie znalazł oczywistej informacji", "odpowiedź jest urwana w połowie zdania".
Trzy najczęstsze przyczyny produkcyjnych awarii RAG: chunking mechanicznie tnie kontekst — zdanie podzielone między dwa chunki sprawia, że model nie ma pełnej informacji; dense retrieval zawodzi na terminologii technicznej — embeddingi słabo rozróżniają sygle norm, numery modeli i nazwy własne; brak rerankingu — LLM dostaje Top-3 z retrieval bez oceny trafności, a efekt "lost in the middle" powoduje ignorowanie środkowych fragmentów kontekstu.
/// PROSTY RAG vs ZAAWANSOWANY RAG
Demo vs architektura produkcyjna
Zaawansowany RAG to pipeline, nie jednorazowa decyzja architektoniczna. Każdy etap — od strategii podziału dokumentu przez hybrid search po Cross-Encoder reranking — ma mierzalny wpływ na jakość odpowiedzi. W tym artykule omawiam każdy etap z kodem gotowym do wdrożenia na produkcji.
Chunking — decyzja o jednostce sensu
Chunk to nie "kawałek tekstu o rozmiarze 512 tokenów" — to jednostka sensu, którą model zobaczy jako kontekst do odpowiedzi. Zbyt mały chunk: brak wystarczającego kontekstu, model "halucynuje" brakujące informacje. Zbyt duży: kontekst jest "rozcieńczony" — odpowiedź gdzieś się w nim chowa, ale model nie może jej precyzyjnie wyodrębnić. Optymalna strategia zależy od typu dokumentu i charakteru pytań użytkowników.
/// STRATEGIE CHUNKINGU — PORÓWNANIE
4 podejścia do podziału dokumentów
Fixed-size (RecursiveCharacterTextSplitter z LangChain) — szybki, prosty, dobry do prototypów. Sentence boundary (spaCy) — zachowuje integralność zdań, lepszy dla dokumentów narracyjnych. Semantic chunking — grupuje zdania wg podobieństwa embeddingów, zmiana tematu to nowy chunk. Hierarchical parent-child (SmallToLarge z LlamaIndex) — małe węzły (128 tok.) do wyszukiwania, duże rodzice (512 tok.) do kontekstu LLM. Eliminuje problem urwanych odpowiedzi przy długich dokumentach technicznych.
from langchain.text_splitter import RecursiveCharacterTextSplitterimport spacyfrom sentence_transformers import SentenceTransformerimport numpy as np# ─── 1. Fixed-size chunking (baseline) ───────────────────────────────────────def fixed_size_chunking(text: str, chunk_size: int = 512, overlap: int = 64) -> list[str]: splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=overlap, separators=["\n\n", "\n", ". ", " "], ) return splitter.split_text(text)# ─── 2. Sentence boundary chunking (spaCy) ───────────────────────────────────def sentence_boundary_chunking(text: str, max_sentences: int = 5) -> list[str]: nlp = spacy.load("pl_core_news_sm") doc = nlp(text) sentences = [sent.text.strip() for sent in doc.sents] return [" ".join(sentences[i : i + max_sentences]) for i in range(0, len(sentences), max_sentences)]# ─── 3. Semantic chunking (cosine distance threshold) ────────────────────────def semantic_chunking(text: str, threshold: float = 0.3) -> list[str]: model = SentenceTransformer("intfloat/multilingual-e5-large") sentences = text.split(". ") embeddings = model.encode(sentences, normalize_embeddings=True) chunks: list[str] = [] current: list[str] = [sentences[0]] for idx in range(1, len(sentences)): sim = float(np.dot(embeddings[idx - 1], embeddings[idx])) if (1 - sim) > threshold: chunks.append(". ".join(current)) current = [sentences[idx]] else: current.append(sentences[idx]) if current: chunks.append(". ".join(current)) return chunks
Kiedy używać hierarchical parent-child: dokumenty techniczne z wieloma sekcjami (specyfikacje, instrukcje, umowy), bazy FAQ, wielosekcyjne raporty. Kiedy fixed-size wystarczy: jednorodne, krótkie artykuły bez złożonej struktury.
Hybrid Search — Dense + BM25 + RRF
Dense retrieval (embeddingi + cosine similarity) świetnie rozumie semantykę i synonimy, ale zawodzi na terminach technicznych: syglach norm (ISO 27001, RODO), numerach modeli i kodach produktów. BM25 precyzyjnie trafia w dokładne frazy, ale nie rozumie parafraz. Reciprocal Rank Fusion (RRF) łączy rankingi z obu metod bez kalibrowania wag — sumuje 1/(k+rank) per dokument i sortuje malejąco.
/// HYBRID SEARCH + CROSS-ENCODER RERANKER
Od zapytania do Top-5 reranked chunks
from qdrant_client import QdrantClientfrom sentence_transformers import SentenceTransformerfrom rank_bm25 import BM25Okapiimport numpy as npclient = QdrantClient(url="http://localhost:6333")embedder = SentenceTransformer("intfloat/multilingual-e5-large")COLLECTION = "knowledge_base"# ─── Dense retrieval via Qdrant ──────────────────────────────────────────────def dense_search(query: str, top_k: int = 20) -> list[tuple]: q_vec = embedder.encode(f"query: {query}", normalize_embeddings=True).tolist() hits = client.search(collection_name=COLLECTION, query_vector=q_vec, limit=top_k, with_payload=True) return [(r.id, r.score, r.payload["text"]) for r in hits]# ─── BM25 sparse retrieval (in-memory; dla skali → Elasticsearch) ────────────def bm25_search(query: str, corpus: list[str], top_k: int = 20) -> list[tuple]: bm25 = BM25Okapi([t.lower().split() for t in corpus]) scores = bm25.get_scores(query.lower().split()) idxs = np.argsort(scores)[::-1][:top_k] return [(int(i), float(scores[i])) for i in idxs]# ─── Reciprocal Rank Fusion ───────────────────────────────────────────────────def rrf_fusion(dense: list, sparse: list, k: int = 60) -> list[tuple]: rrf: dict[int, float] = {} for rank, (doc_id, *_) in enumerate(dense): rrf[doc_id] = rrf.get(doc_id, 0) + 1 / (k + rank + 1) for rank, (doc_id, *_) in enumerate(sparse): rrf[doc_id] = rrf.get(doc_id, 0) + 1 / (k + rank + 1) return sorted(rrf.items(), key=lambda x: x[1], reverse=True)def hybrid_search(query: str, corpus: list[str], top_k: int = 10) -> list[tuple]: return rrf_fusion(dense_search(query), bm25_search(query, corpus))[:top_k]
Alternatywa: Qdrant 1.7+ obsługuje natywny sparse vector search (SPLADE, BM42) bez zewnętrznego silnika. Dla małych baz (<100k dokumentów) BM25 in-memory (rank_bm25) jest w pełni wystarczający.
Reranking z Cross-Encoder
Bi-encoder (embeddingi) jest szybki, bo każdy dokument jest kodowany niezależnie od pytania. Cross-encoder analizuje parę (pytanie, chunk) razem — jest 50–100× wolniejszy, ale znacznie dokładniejszy. Strategia: bi-encoder zwraca Top-20–50 kandydatów, cross-encoder rerankes je i zostawia Top-3–5 dla LLM. Latency wzrasta o ~200–400 ms na GPU — akceptowalny koszt przy mierzalnym wzroście jakości.
from sentence_transformers import CrossEncoderfrom openai import OpenAI# mmarco-mMiniLMv2: wielojęzyczny reranker (PL/EN/DE/FR/…), szybki (MiniLM backbone)# alternatywa: BAAI/bge-reranker-v2-m3 — wyższa jakość, cięższy modelreranker = CrossEncoder("cross-encoder/mmarco-mMiniLMv2-L12-H384-v1")def rerank(query: str, candidates: list[dict], top_k: int = 5) -> list[dict]: pairs = [(query, c["text"]) for c in candidates] scores = reranker.predict(pairs) for i, score in enumerate(scores): candidates[i]["rerank_score"] = float(score) return sorted(candidates, key=lambda x: x["rerank_score"], reverse=True)[:top_k]def rag_answer(query: str, corpus: list[str], llm: OpenAI) -> str: from hybrid_search_qdrant import hybrid_search fused = hybrid_search(query, corpus, top_k=20) candidates = [{"id": doc_id, "text": corpus[doc_id]} for doc_id, _ in fused] top_docs = rerank(query, candidates, top_k=5) context = "\n---\n".join(d["text"] for d in top_docs) resp = llm.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": "Odpowiadaj wyłącznie na podstawie kontekstu. Jeśli odpowiedź nie wynika z kontekstu — powiedz: nie wiem."}, {"role": "user", "content": f"Kontekst:\n{context}\n\nPytanie: {query}"}, ], temperature=0, ) return resp.choices[0].message.content
Kluczowa zasada: nie zwiększaj Top-K dla LLM żeby "dać mu więcej". Efekt "lost in the middle" sprawia, że modele ignorują środkowe fragmenty długiego kontekstu. 3–5 zrankowanych chunków daje lepsze odpowiedzi niż 15 słabiej posortowanych. Docelowy schemat: Top-20 retrieval → Top-5 po rerankingu → LLM z ~1500 tokenów kontekstu.
Ewaluacja z RAGAS — gate jakości przed wdrożeniem
Ewaluacja RAG bez automatycznych metryk to błąd architektoniczny — nie wiesz czy zmiana chunkingu poprawiła wyniki, nie masz gate'a przed wdrożeniem nowej wersji embeddingów. RAGAS mierzy cztery wymiary: Context Precision (trafność pobranych chunków), Context Recall (kompletność informacji), Faithfulness (czy odpowiedź wynika z kontekstu, nie z wiedzy modelu) i Answer Relevancy (czy odpowiedź odpowiada na pytanie).
from datasets import Datasetfrom ragas import evaluatefrom ragas.metrics import ( context_precision, context_recall, faithfulness, answer_relevancy,)# min. 50–100 par (pytanie, ground_truth) dla rzetelnych wyników produkcyjnychtest_cases = [ { "question": "Jakie są wymagania wstępne do certyfikacji ISO 27001?", "answer": "ISO 27001 wymaga wdrożenia ISMS, analizy ryzyka i audytu wewnętrznego.", "contexts": [ "ISO 27001 to norma dla systemów zarządzania bezpieczeństwem informacji.", "Certyfikacja wymaga udokumentowania polityki bezpieczeństwa i oceny ryzyka.", ], "ground_truth": "Do certyfikacji ISO 27001 wymagane jest ISMS, ocena ryzyka i audyt wewnętrzny.", },]dataset = Dataset.from_list(test_cases)results = evaluate( dataset=dataset, metrics=[context_precision, context_recall, faithfulness, answer_relevancy],)df = results.to_pandas()print(df[["context_precision", "context_recall", "faithfulness", "answer_relevancy"]].mean())
Docelowe progi jakości: Context Precision ≥ 0.80, Context Recall ≥ 0.75, Faithfulness ≥ 0.85, Answer Relevancy ≥ 0.80. Traktuj te wartości jako gate w CI/CD — przed mergem nowej wersji embeddingów lub strategii chunkingu uruchamiaj ewaluację na stałym zestawie 100 pytań testowych.
| Metryka | Prosty RAG | Zaawansowany RAG | Akcja gdy wynik < 0.70 |
|---|---|---|---|
| Context Precision | 0.51 | 0.84 | Zmień strategię chunkingu lub model embeddingowy |
| Context Recall | 0.63 | 0.82 | Zwiększ top-K retrieval lub popraw coverage indeksu |
| Faithfulness | 0.74 | 0.91 | Wzmocnij system prompt lub zmień model bazowy LLM |
| Answer Relevancy | 0.68 | 0.87 | Popraw query reformulation i instrukcję systemową |
Uruchamiaj RAGAS co tydzień na losowej próbce rzeczywistych zapytań z logów produkcyjnych — nie tylko na syntetycznym zestawie testowym. Metryki syntetyczne bywają zawyżone. Zbieraj pary (pytanie, odpowiedź) z produkcji i stopniowo rozszerzaj zestaw ewaluacyjny.
---
Wdrażam zaawansowane pipeline RAG dla firm, które mają bazę wiedzy i chcą żeby pracownicy lub klienci dostawali precyzyjne, udokumentowane odpowiedzi — nie "halucynacje". Od audytu istniejącego systemu przez optymalizację chunkingu, hybrid search i rerankingu po monitoring jakości z RAGAS. Napisz do mnie — zaczynam od analizy Twojej bazy dokumentów i pierwszego benchmarku.
/// RELATED_RECORDS
Jak AI czyta faktury z maila i wprowadza je do ERP
AI odczytuje fakturę z załącznika e-mail — PDF, skan lub zdjęcie z telefonu — i wprowadza dane bezpośrednio do ERP bez ręcznego przepisywania. Pełna automatyzacja obiegu faktur kosztowych: od skrzynki mailowej do zaksięgowania dokumentu.
Od czego zacząć wdrażanie AI w firmie?
Wdrażanie AI w firmie zaczyna się nie od wyboru narzędzia, lecz od jednego powtarzalnego procesu, który dziś zabiera najwięcej czasu. Dowiedz się jak krok po kroku wybrać, opisać i zautomatyzować ten proces.
Jak zbudować wewnętrzną bazę wiedzy firmy z AI (RAG w praktyce)
Wewnętrzna baza wiedzy oparta na RAG pozwala stworzyć własnego chatbota firmowego, który odpowiada wyłącznie na podstawie dokumentów Twojej firmy — nie domysłów modelu. Bezpieczne, aktualne, precyzyjne AI z pełną kontrolą nad danymi.
Signal received?
Przerwij
Ciszę
Zainicjuj protokół. Nawiąż połączenie. Zbudujmy coś głośnego.