Zaawansowany RAG - chunking strategy, hybrid search i reranking w produkcji

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

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]

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

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