Zaawansowane RAG w Milvus: hybrydowe wyszukiwanie, HyDE, przepisywanie zapytań i ponowna sortowanie.

Stworzenie systemu RAG typu proof-of-concept zajmuje zaledwie kilka minut: dzieli się plik PDF na fragmenty, przekazuje te fragmenty modelowi embeddingowemu, a następnie przekazuje wyniki dopasowań modelowi LLM. To właśnie w fazie produkcji taka konfiguracja oparta wyłącznie na wyszukiwaniu wektorowym napotyka problemy. Prawdziwi użytkownicy nie formułują zapytań w taki sam sposób jak optymalizowane silniki przetwarzania danych. Używają niejasnego języka, pomijają kluczowe słowa i łączą kilka pytań w jedno, przez co pojedyncze wyszukiwanie semantyczne często pomija istotne informacje. Gdy wyszukiwanie nie udaje się, model LLM musi podsumować nieistotny kontekst, co prowadzi do subtelnych halucynacji lub odpowiedzi, które nie są pomocne.

Zamknięcie luki pomiędzy podstawowym wyszukiwaniem semantycznym a wieloetapowym procesem pozyskiwania informacji wymaga kilku warstw rozwiązań. Poniższe rozwiązanie wykorzystuje Milvus. Zaczyna od gęstego wyszukiwania wektorowego, dodaje wyszukiwanie hybrydowe, następnie łączy cztery techniki (filtrowanie metadanych, przepisywanie zapytań, HyDE oraz dekompozycja zapytań) i kończy na ponownym sortowaniu za pomocą cross-encodera. Każdy krok jest opatrzony gotowym kodem.

Dlaczego istnieje RAG

Modele takie jak GPT-4 uczą się na podstawie ogólnych danych z Internetu do określonego terminu aktualizacji wiedzy. Dokumenty prywatne, wewnętrzne wiki oraz zasady firmy pozostają poza zasięgiem. Jeśli zapytasz model o zasady urlopowe firmy, albo powie, że nie wie, albo wymyśli wiarygodną, lecz błędną odpowiedź.

RAG (Retrieval-Augmented Generation) dzieli ten proces na dwa kroki:

1. Pobieranie najbardziej istotnych fragmentów z dokumentów za pomocą wyszukiwania semantycznego.
2. Generowanie odpowiedzi poprzez przekazanie tych fragmentów modelowi LLM jako kontekstu.

Ogólny przebieg procesu wygląda następująco:

User Question
     │
     ▼
[Embed the question]  →  query vector
     │
     ▼
[Search Vector DB]  →  top-K relevant document chunks
     │
     ▼
[LLM prompt: "Given these passages, answer: {question}"]
     │
     ▼
  Accurate, Grounded Answer

Jakość ostatecznej odpowiedzi zależy niemal wyłącznie od jakości wyszukiwania, więc prawidłowe wyszukiwanie jest kluczowe.

Gdzie wchodzi baza wektorowa

Tradycyjne bazy danych porównują dokładne wartości, więc zapytanie WHERE text LIKE '%leave%' pomija dokumenty zawierające sformułowania takie jak „prawo do urlopu” lub „brak pracy z powodów medycznych”, mimo że mają one to samo znaczenie. Baza wektorowa natomiast przechowuje każdy dokument jako wielowymiarowy wektor liczbowy (embedding) i wyszukuje na podstawie podobieństwa semantycznego, czyli tego, jak bliskie jest znaczenie zapytania znaczeniu poszczególnych fragmentów tekstu. Milvus nadaje się do tej roli z następujących powodów:

• Jest otwartym oprogramowaniem i można go uruchomić lokalnie, co zapewnia prywatność danych.
• Milvus Lite działa z pojedynczego pliku .db, więc do rozwoju nie potrzeba serwera.

Poиск hybrydowy jest wbudowany, umożliwiając łączenie wektorów gęstych (semantycznych) i rzadkich (kluczowych słów) w jednej zapytaniu.

Działa skutecznie od laptopa aż po miliardy wektorów bez konieczności zmian w kodzie aplikacji.

Wektory gęste i rzadkie

Proces wykorzystuje dwa rodzaje wektorów, dlatego warto je znać przed napisaniem kodu. Wektor gęsty to wektor o stałej długości, na przykład 1024 liczb, wytwarzany przez sieć neuronową. Wszystkie wymiary zawierają wartości, a trening uczy modelu umieszczania zdań o podobnym znaczeniu blisko siebie:

"sick leave policy"           →  [0.12, -0.87, 0.34, 0.56, ...]  (1024 numbers)
"medical absence entitlement" →  [0.13, -0.85, 0.31, 0.54, ...]  ← very close
"quarterly revenue target"    →  [0.91,  0.23, -0.67, 0.02, ...] ← far away

Wektory gęste dobrze oddają znaczenie i synonimy. Modele takie jak text-embedding-3-large od OpenAI je wytwarzają. Wektor rzadki to bardzo duży wektor, w którym większość pozycji ma wartość zero. Każda niezerowa pozycja reprezentuje słowo z słownika, a jej wartość to stopień jego znaczenia, czyli waga w stylu TF-IDF:

"sick leave policy" → {word_index_for_"sick": 0.82, word_index_for_"leave": 0.91, ...}

BM25 to klasyczny algorytm rzadki, który doskonale radzi sobie z dokładnym dopasowywaniem słów kluczowych. Potrafi wykrywać nazwy własne, nazwy produktów oraz terminy techniczne, których może nie dostrzec wyszukiwanie gęste ze względu na brak bliskiego sąsiada semantycznego.

Wyszukiwanie hybrydowe wykorzystuje oba podejścia, czerpiąc zalety z każdego z nich.

Ustawianie i konfiguracja

Najpierw zainstaluj wymagane zależności:

pip install --upgrade pymilvus
pip install "pymilvus[model]"
pip install sentence-transformers
pip install langchain-text-splitters
pip install langchain-openai
pip install langchain-community
pip install scipy
pip install nltk

Następnie importy:

import uuid
from tqdm import tqdm
from pymilvus import (
    MilvusClient, DataType,
    AnnSearchRequest, RRFRanker
)
from pymilvus.model.sparse import BM25EmbeddingFunction
from pymilvus.model.sparse.bm25.tokenizers import build_default_analyzer
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
import scipy.sparse as sp
import re, json
import nltk
nltk.download('stopwords')

Zachowaj wszystkie parametry dostępne do ustawienia razem na górze pliku:

PDF_PATH        = "./data/sample_employee_handbook.pdf" # path of you document
COLLECTION_NAME = "rag_documents_hybrid"
MILVUS_DB_PATH  = "./db/milvus_demo.db"
API_KEY         = "sk-..."
EMBEDDING_MODEL = "text-embedding-3-large"
EMBEDDING_DIM   = 1024
CHUNK_SIZE      = 500
CHUNK_OVERLAP   = 100
TOP_K           = 5

Jeśli chodzi o rozmiar fragmentów: 500 znaków z 100-znakowym nakładaniem się nadaje się jako punkt wyjścia dla prostych dokumentów. Tekst narracyjny radzi sobie lepiej przy większych fragmentach o rozmiarze 800–1000 znaków, natomiast złożona dokumentacja techniczna – przy mniejszych fragmentach o rozmiarze 200–300 znaków. Nakładanie się zapewnia, że zdania przerwane granicą fragmentu nadal można odnaleźć.

Proces indeksowania

Indeksowanie jest wykonywane raz, lub ponownie w momencie zmiany dokumentów. Przekształca surowe dokumenty w wektory i ładowa je do Milvus. Ścieżka jest następująca:

PDF  →  Pages  →  Chunks  →  Dense Embeddings
                           →  Sparse Embeddings
                           →  Milvus Collection

Najpierw zainicjuj modele i połącz je:

# Dense embedding model here we'll be using OpenAI's embedding model
embedding_obj = OpenAIEmbeddings(
    model=EMBEDDING_MODEL,
    api_key=API_KEY,
    dimensions=EMBEDDING_DIM
)
# Milvus Lite - single file, no server needed
client = MilvusClient(MILVUS_DB_PATH)
print("Models and DB connection ready.")

Następnie załaduj PDF do osobnego obiektu dokumentu na każdą stronę:

# Load PDF — one Document object per page
loader = PyPDFLoader(PDF_PATH)
documents = loader.load()
print(f"Loaded {len(documents)} pages.")

# Split into overlapping chunks
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=CHUNK_SIZE,
    chunk_overlap=CHUNK_OVERLAP,
    separators=["\n\n", "\n", ".", " ", ""]
)
chunks = text_splitter.split_documents(documents)
print(f"Created {len(chunks)} chunks.")

RecursiveCharacterTextSplitter najpierw dzieli tekst na przerwy między akapitami, potem na nowe linie, a następnie na zdania, dzięki czemu powiązane zdania pozostają razem w jednym fragmencie tak długo, jak to możliwe. Następnie przychodzą embeddingi dla obu typów wektorów:

texts = [doc.page_content for doc in chunks]

# Dense embeddings - one API call for the entire corpus
print("Generating dense embeddings...")
dense_embeddings = embedding_obj.embed_documents(texts)
print(f"Dense dimension: {len(dense_embeddings[0])}")

# Sparse embeddings - BM25 must be fit on YOUR corpus first
print("Fitting BM25 on corpus...")
analyzer = build_default_analyzer(language="en") # for this you will require nltk-stopwords
bm25_ef = BM25EmbeddingFunction(analyzer)
bm25_ef.fit(texts)  # Builds vocabulary from your documents
sparse_embeddings = bm25_ef.encode_documents(texts)
print("Sparse embeddings generated.")

Wywołanie bm25_ef.fit(texts) jest kluczowe. BM25 uczy się słownictwa oraz częstotliwości występowania terminów na podstawie własnego korpusu, i właśnie dlatego działa skutecznie. Użycie ogólnego modelu BM25 znacząco obniża jakość dopasowywania słów kluczowych. Oto zbiór wraz ze schematem i indeksami:

# Drop and recreate for a clean state
if COLLECTION_NAME in client.list_collections():
    client.drop_collection(COLLECTION_NAME)

# Define schema
schema = client.create_schema()
schema.add_field("id",            DataType.VARCHAR,           is_primary=True, max_length=100)
schema.add_field("vector",        DataType.FLOAT_VECTOR,      dim=EMBEDDING_DIM)
schema.add_field("sparse_vector", DataType.SPARSE_FLOAT_VECTOR)
schema.add_field("text",          DataType.VARCHAR,           max_length=65535)
schema.add_field("page_number",   DataType.INT64)
schema.add_field("source",        DataType.VARCHAR,           max_length=500)
schema.add_field("chunk_id",      DataType.INT64)

# Create the collection
client.create_collection(collection_name=COLLECTION_NAME, schema=schema)

# Build indexes separately
index_params = client.prepare_index_params()

index_params.add_index(
    field_name="vector",
    index_type="FLAT",          # Exact search - swap to HNSW for production
    metric_type="COSINE"
)

index_params.add_index(
    field_name="sparse_vector",
    index_type="SPARSE_INVERTED_INDEX",
    metric_type="IP"            # Inner Product is the only valid metric for sparse
)

client.create_index(collection_name=COLLECTION_NAME, index_params=index_params)

print("Collection and indexes created.")

Tutaj istotne są twoje wybory indeksu:

• FLAT dla wektorów gęstych. Przeprowadza dokładne wyszukiwanie najbliższego sąsiada, jest całkowicie precyzyjne, ale wolniejsze przy dużych skaliach. Nadaje się do rozwoju oraz do zbiorków danych liczących mniej niż około 100K elementów, natomiast w środowisku produkcyjnym powinno zostać zastąpione przez HNSW.
• SPARSE_INVERTED_INDEX dla wektorów rzadkich. Jest to jedyny dopuszczalny typ indeksu dla danych rzadkich w Milvus i funkcjonuje jak klasyczny indeks odwrotny, w taki sam sposób, w jaki silniki wyszukiwania indeksują dokumenty według słów.

Na koniec przygotuj rekordy i wstaw je:

def sparse_to_dict(s_emb) -> dict:
    """Convert a scipy sparse row into Milvus-compatible {index: value} dict."""
    if sp.issparse(s_emb):
        coo = s_emb.tocoo()
        return {int(col): float(val) for col, val in zip(coo.col, coo.data)}
    elif isinstance(s_emb, dict):
        return s_emb
    else:
        return {int(i): float(v) for i, v in enumerate(s_emb) if v != 0.0}

# Build the records list
data = []
for idx, (chunk, d_emb) in enumerate(tqdm(zip(chunks, dense_embeddings), total=len(chunks))):
    sparse_dict = sparse_to_dict(sparse_embeddings[idx])
    if not sparse_dict:
        print(f"Warning: empty sparse vector at chunk {idx}, skipping.")
        continue
    data.append({
        "id":            str(uuid.uuid4()),
        "vector":        d_emb,
        "sparse_vector": sparse_dict,
        "text":          chunk.page_content,
        "page_number":   int(chunk.metadata.get("page", -1)),
        "source":        PDF_PATH,
        "chunk_id":      idx
    })

# Insert into Milvus
res = client.insert(collection_name=COLLECTION_NAME, data=data)
print(f"Inserted {res['insert_count']} records.")

# Load into memory - required before any search operation
client.load_collection(COLLECTION_NAME)
print(f"Load state: {client.get_load_state(COLLECTION_NAME)}")

Wsparcie sparse_to_dict jest konieczne, ponieważ BM25 tworzy rzadką macierz scipy, podczas gdy Milvus wymaga zwykłego słownika Pythona typu {int: float}. Funkcja ta radzi sobie z 2D rzadkimi wierszami, 1D tablicami oraz słownikami, które zostały już przekonwertowane. Jedna z częstych pułapek to konieczność zawsze wywoływania client.load_collection() przed wyszukiwaniem. Indeksy przechowywane są na dysku w Milvus i są ładowane do pamięci RAM na żądanie, więc bez tego wywołania wyszukiwanie kończy się błędem lub zwraca pusty wynik.

Wyszukiwanie gęste jako punkt odniesienia

Gdy dane są zindeksowane, najprostsze wyszukiwanie polega na włączeniu zapytania i znalezieniu wektorów najbliższych dokumentów na podstawie podobieństwa kosinowego:

# ════════════════════════════════════════════════════════════
# Dense Vector Search
# ════════════════════════════════════════════════════════════

query = "What is the leave policy?"

# Step 1: Embed the query using the same model used at index time
query_dense_embedding = embedding_obj.embed_query(query)

# Step 2: Search
results = client.search(
    collection_name=COLLECTION_NAME,
    data=[query_dense_embedding],
    anns_field="vector",
    search_param={"metric_type": "COSINE"},
    limit=TOP_K,
    output_fields=["text", "page_number", "source"]
)

# Step 3: Display results
for idx, hit in enumerate(results[0], start=1):
    entity = hit["entity"]
    print(f"Rank {idx} | Cosine Score: {hit['distance']:.4f} | Page: {entity['page_number']}")
    print(f"  {entity['text'][:300]}\n")

Zgodnie z metryką COSINE, wyniki waha się od -1 do 1. Wartość powyżej 0,7 zazwyczaj oznacza silne dopasowanie semantyczne, a najlepsze wyniki znajdują się zwykle w przedziale od 0,6 do 0,95. Szukanie gęste wystarcza w przypadku pytań w języku naturalnym, których odpowiedzi znajdują się w bliskiej przestrzeni semantycznej, np. gdy prosi się o wyjaśnienie świadczeń związanych z urlopem rodzicielskim lub co stanowi kodeks postępowania. Metoda ta ma trudności, gdy zapytanie zawiera konkretne kody, akronimy lub nazwy własne, których model embedding rzadko widział.

Szukanie hybrydowe z RRF

Szukanie hybrydowe wykonuje jednocześnie szukanie gęste i rzadkie BM25, a następnie łączy obie listy uporządkowane za pomocą Reciprocal Rank Fusion (RRF). Idea jest prosta – wynik RRF dokumentu to suma wartości 1 / (k + rank) z każdej listy, przy czym k=60 stanowi stałą wygładzającą, dzięki czemu dokumenty dobrze uporządkowane w obu listach znajdują się na szczycie.

# ════════════════════════════════════════════════════════════
# Hybrid Search (Dense + Sparse)
# ════════════════════════════════════════════════════════════

query = "leave policy?"

# Dense query vector
query_dense = embedding_obj.embed_query(query)

# Sparse query vector - uses the same BM25 model fitted on the corpus
sparse_raw  = bm25_ef.encode_queries([query])
sparse_dict = sparse_to_dict(sparse_raw[0])
print(f"Sparse query terms: {len(sparse_dict)}")  # Should be > 0

# Build two separate ANN search requests
dense_req = AnnSearchRequest(
    data=[query_dense],
    anns_field="vector",
    param={"metric_type": "COSINE"},
    limit=TOP_K
)
sparse_req = AnnSearchRequest(
    data=[sparse_dict],
    anns_field="sparse_vector",
    param={"metric_type": "IP"},
    limit=TOP_K
)

# Execute hybrid search with RRF fusion
results = client.hybrid_search(
    collection_name=COLLECTION_NAME,
    reqs=[dense_req, sparse_req],
    ranker=RRFRanker(k=60),
    limit=TOP_K,
    output_fields=["text", "page_number", "source"]
)
for idx, hit in enumerate(results[0], start=1):
    entity = hit["entity"]
    print(f"Rank {idx} | RRF Score: {hit['distance']:.4f} | Page: {entity['page_number']}")
    print(f"  {entity['text'][:300]}\n")

O wynikach: wartość hit['distance'] z funkcji hybrid_search to połączony wynik rankingu RRF, a nie surowa podobieństwo kosinowe. Im większa wartość, tym lepiej, chociaż zakres nie jest ograniczony do wartości od 0 do 1, co oznacza, że nie powinna być porównywana z wynikami kosinowymi uzyskanymi przy zwykłej funkcji search(). W środowiskach RAG produkcyjnych hybrydowe wyszukiwanie jest zalecanym standardem. BM25 wpływa znikomo na opóźnienia i wyraźnie poprawia stopień odzyskiwania informacji w zapytaniach zawierających wiele słów kluczowych.

Nawet hybrydowe metody wyszukiwania mogą zawieść w określonych sytuacjach, a cztery techniki adresują najczęstsze przyczyny takich awarii.

Filtrowanie metadanych

Problem. Gdy zbiór obejmuje kilka dokumentów, okresów lub działów, wyszukiwanie wektorowe może zwrócić fragmenty wydające się istotne, ale pochodzące z niewłaściwego źródła – na przykład ze starszej wersji polityki, podręcznika innego działu lub stron poza zakresem pytania.

Jak to działa. Milvus stosuje wyrażenie filtru skalarnego przed wyszukiwaniem wektorowym, dzięki czemu tylko te wiersze, które odpowiadają kryteriom, trafiają do etapu ANN. Nie jest to filtr post-processowy. Filtr najpierw redukuje zbiór kandydatów, co zapewnia dobrą dokładność w filtrowanym podzbiórze.

# ════════════════════════════════════════════════════════════
# METADATA FILTERING
# ════════════════════════════════════════════════════════════

def search_with_metadata_filter(
    client, collection_name, embedding_obj, bm25_ef,
    query: str,
    page_range: tuple = None,
    source_file: str = None,
    top_k: int = 5
):
    filter_parts = []

    if page_range:
        lo, hi = page_range
        filter_parts.append(f"page_number >= {lo} && page_number <= {hi}")

    if source_file:
        filter_parts.append(f'source == "{source_file}"')

    filter_expr = " && ".join(filter_parts) if filter_parts else None

    print(f"\n[Metadata Filter] Query : '{query}'")
    print(f"[Metadata Filter] Filter: {filter_expr or 'None (unfiltered)'}")

    results = hybrid_search(
        client, collection_name, embedding_obj, bm25_ef,
        query_text=query,
        top_k=top_k,
        filters=filter_expr
    )
    return results


# ── Run ──────────────────────────────────────────────────────
meta_results = search_with_metadata_filter(
    client, COLLECTION_NAME, embedding_obj, bm25_ef,
    query      = "What is the leave policy?",
    page_range = (1, 30),
    source_file= None,
    top_k      = 5
)

# ── Print Results ─────────────────────────────────────────────
print("\nMETADATA-FILTERED RESULTS")
print("=" * 55)
if not meta_results or not meta_results[0]:
    print("No results returned.")
else:
    for idx, hit in enumerate(meta_results[0], start=1):
        entity = hit["entity"]
        print(f"\nRank  : {idx}")
        print(f"Score : {hit['distance']:.4f}")
        print(f"Page  : {entity['page_number']}")
        print(f"Text  :\n{entity['text'][:400]}")

Wyrażenia filtrujące spełniają następującą składnię:

'page_number >= 1 && page_number <= 30'      # page range
'source == "hr_policy.pdf"'                  # exact source
'category in ["leave", "performance"]'       # in a list
'source like "hr%"'                          # prefix match

Każdy system RAG obsługujący wiele dokumentów powinien zawierać funkcję filtrowania metadanych. Jeśli użytkownicy pracują w określonym zakresie, na przykład tylko z podręcznikiem z 2024 roku lub plikami finansowymi, filtr powinien być automatycznie wygenerowany na podstawie tego zakresu.

Ponowne sformułowanie zapytania

Problem. Użytkownicy piszą swobodnie, na przykład „Ile dni wolnego dostanę?”, podczas gdy dokumenty używają formalnego języka, takiego jak „prawo do urlopu rocznego” lub „zasady naliczania dni urlopowych”. Intensywne wyszukiwanie pomaga w części pokrycia tej luki, a dobre ponowne sformułowanie znacznie poprawia efektywność wyszukiwania oraz ogólną dokładność wyników.

Jak to działa. Sztuczna inteligencja generuje kilka formalnych wersji zapytania, bogatych w słowa kluczowe; każda z tych wersji jest wyszukiwana oddzielnie, a system RRF łączy i usuwa duplikaty uzyskanych wyników.

# ════════════════════════════════════════════════════════════
# QUERY REWRITING
# ════════════════════════════════════════════════════════════

import re, json

REWRITE_PROMPT = """You are an expert at reformulating search queries to improve document retrieval.

Given a user query, produce {n} alternative search queries that:
- Use formal, document-style language
- Include relevant keywords and synonyms
- Cover different angles of the same question

User query: {query}

Respond ONLY with a JSON array of strings. Example:
["rewritten query 1", "rewritten query 2", "rewritten query 3"]"""


def rewrite_query(query: str, n: int = 3) -> list[str]:
    prompt   = REWRITE_PROMPT.format(query=query, n=n)
    response = llm.invoke(prompt)
    raw      = re.sub(r"^```json|^```|```quot;, "", response.content.strip(), flags=re.MULTILINE).strip()
    try:
        variants = json.loads(raw)
        return [query] + variants       # always keep the original
    except json.JSONDecodeError:
        print("Warning: Could not parse rewrites, using original query only.")
        return [query]


def search_with_query_rewriting(
    client, collection_name, embedding_obj, bm25_ef,
    query: str,
    n_rewrites: int = 3,
    top_k: int = 5
):
    variants = rewrite_query(query, n=n_rewrites)

    print(f"\n[Query Rewriting] Original  : '{query}'")
    for i, v in enumerate(variants[1:], 1):
        print(f"[Query Rewriting] Variant {i} : '{v}'")

    seen_ids    = {}
    rank_scores = {}

    for variant in variants:
        results = hybrid_search(
            client, collection_name, embedding_obj, bm25_ef,
            query_text=variant,
            top_k=top_k
        )
        if not results or not results[0]:
            continue
        for rank, hit in enumerate(results[0], start=1):
            hit_id = hit["id"]
            rank_scores[hit_id] = rank_scores.get(hit_id, 0) + 1.0 / (60 + rank)
            if hit_id not in seen_ids:
                seen_ids[hit_id] = hit

    merged = sorted(seen_ids.values(), key=lambda h: rank_scores[h["id"]], reverse=True)[:top_k]
    return [merged]


# ── Run ──────────────────────────────────────────────────────
rewrite_results = search_with_query_rewriting(
    client, COLLECTION_NAME, embedding_obj, bm25_ef,
    query      = "What is the leave policy?",
    n_rewrites = 3,
    top_k      = 5
)

# ── Print Results ─────────────────────────────────────────────
print("\nQUERY-REWRITTEN RESULTS")
print("=" * 55)
if not rewrite_results or not rewrite_results[0]:
    print("No results returned.")
else:
    for idx, hit in enumerate(rewrite_results[0], start=1):
        entity = hit["entity"]
        print(f"\nRank  : {idx}")
        print(f"Score : {hit['distance']:.4f}")
        print(f"Page  : {entity['page_number']}")
        print(f"Text  :\n{entity['text'][:400]}")

Przykład transformacji:

Input:  "how many days off do I get?"

Output variants:
  1. "annual leave entitlement number of days employee handbook"
  2. "vacation days accrual policy full-time employee"
  3. "paid time off PTO allowance per calendar year"

Każda wariantowa wersja rozszerza obszar wyszukiwania. Fragment, który nie znajduje się w pierwszej piątce wyników dla oryginalnego zapytania, może zająć drugie miejsce w przypadku trzeciej wariantowej wersji, a RRF ujawnia dokumenty, które regularnie pojawiają się we wszystkich zapytaniach.

HyDE (Hypothetical Document Embeddings)

Problem. Krótka pytanie i jego odpowiedź znajdują się w różnych miejscach we przestrzeni embeddingów. „Ile dni chorobowych?” to pytanie, które jest embedowane w pobliżu innych pytań. Zapis polityki mówiący, że pracownicy otrzymują 7 dni płatnego urlopu chorobowego rocznie, jest embedowany w pobliżu innych zapisów dotyczących polityki. Ta różnica obniża precyzję wyników.

Jak to działa. Zamiast wstawiać pytanie, model językowy tworzy hipotetyczny dokument z odpowiedzią, który następnie jest wstawiany. Ta wymyślona odpowiedź trafia w pobliże prawdziwych dokumentów z odpowiedziami, nawet jeśli zawarte w niej fakty są błędne. Wykorzystywany jest tylko jej embedding, a tekst jest odrzucany, więc wymyślone szczegóły nigdy nie docierają do użytkownika.

# ════════════════════════════════════════════════════════════
# HyDE (Hypothetical Document Embeddings)
# ════════════════════════════════════════════════════════════

HYDE_PROMPT = """You are a corporate policy document writer.

Write a 2-3 paragraph excerpt from an official HR policy or company document
that would DIRECTLY ANSWER the following question.
Write in formal document style. Do not mention the question itself.

Question: {query}

Document excerpt:"""


def generate_hypothetical_document(query: str) -> str:
    response = llm.invoke(HYDE_PROMPT.format(query=query))
    return response.content.strip()


def search_with_hyde(
    client, collection_name, embedding_obj, bm25_ef,
    query: str,
    top_k: int = 5
):
    hypothetical_doc = generate_hypothetical_document(query)

    print(f"\n[HyDE] Query            : '{query}'")
    print(f"[HyDE] Hypothetical doc :\n  {hypothetical_doc[:300]}...\n")

    # Search using the hypothetical document's embedding
    hyde_results = hybrid_search(
        client, collection_name, embedding_obj, bm25_ef,
        query_text=hypothetical_doc,    # embed the answer, not the question
        top_k=top_k
    )

    # Also search with the original query and merge both via RRF
    original_results = hybrid_search(
        client, collection_name, embedding_obj, bm25_ef,
        query_text=query,
        top_k=top_k
    )

    seen_ids    = {}
    rank_scores = {}

    for result_set in [hyde_results, original_results]:
        if not result_set or not result_set[0]:
            continue
        for rank, hit in enumerate(result_set[0], start=1):
            hit_id = hit["id"]
            rank_scores[hit_id] = rank_scores.get(hit_id, 0) + 1.0 / (60 + rank)
            if hit_id not in seen_ids:
                seen_ids[hit_id] = hit

    merged = sorted(seen_ids.values(), key=lambda h: rank_scores[h["id"]], reverse=True)[:top_k]
    return [merged]


# ── Run ──────────────────────────────────────────────────────
hyde_results = search_with_hyde(
    client, COLLECTION_NAME, embedding_obj, bm25_ef,
    query = "What is the leave policy?",
    top_k = 5
)

# ── Print Results ─────────────────────────────────────────────
print("\nHyDE RESULTS")
print("=" * 55)
if not hyde_results or not hyde_results[0]:
    print("No results returned.")
else:
    for idx, hit in enumerate(hyde_results[0], start=1):
        entity = hit["entity"]
        print(f"\nRank  : {idx}")
        print(f"Score : {hit['distance']:.4f}")
        print(f"Page  : {entity['page_number']}")
        print(f"Text  :\n{entity['text'][:400]}")

HyDE najlepiej sprawdza się przy krótkich, niejasnych zapytaniach, takich jak procedura oceny wynagrodzeń, brakach w słownictwie pomiędzy pytaniami a dokumentami oraz bardzo nieformalnym sformułowaniu. Jest przesadą przy precyzyjnych zapytaniach bogatych w słowa kluczowe, gdzie hybrydowy wyszukiwarka już dobrze funkcjonuje, a także gdy liczy się opóźnienie LLM, ponieważ HyDE dodaje jedną dodatkową próbę wywołania LLM na każde zapytanie.

Rozbicie zapytania

Problem. Złożone, wieloczęściowe pytanie nie może zostać odpowiedziane na podstawie jednego tekstu. Pytanie o politykę urlopów i wpływ oceny wyników na pensję dotyczy dwóch tematów w dwóch różnych sekcjach. Zwykłe wyszukiwanie wektorowe albo skupia się na jednym temacie, pomijając drugi, albo dostarcza przeciętne wyniki dla obu.

Jak to działa. LLM dzieli pytanie na małe, atomowe podpytania, każde z nich jest wyszukiwane osobno, RRF łączy uzyskane wyniki, a ostateczne wywołanie LLM przekazuje wszystkie istotne fragmenty informacji.

# ════════════════════════════════════════════════════════════
# QUERY DECOMPOSITION
# ════════════════════════════════════════════════════════════

DECOMPOSE_PROMPT = """You are an expert at breaking down complex questions for document retrieval.

Decompose the following question into 2-4 simple, self-contained sub-questions.
Each sub-question should target a single distinct piece of information.

Complex question: {query}

Respond ONLY with a JSON array of strings. Example:
["sub-question 1", "sub-question 2", "sub-question 3"]"""


def decompose_query(query: str) -> list[str]:
    response = llm.invoke(DECOMPOSE_PROMPT.format(query=query))
    raw      = re.sub(r"^```json|^```|```quot;, "", response.content.strip(), flags=re.MULTILINE).strip()
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        print("Warning: Could not parse decomposition, using original query.")
        return [query]


def search_with_decomposition(
    client, collection_name, embedding_obj, bm25_ef,
    query: str,
    top_k: int = 5
):
    sub_questions = decompose_query(query)

    print(f"\n[Decomposition] Original query : '{query}'")
    for i, sq in enumerate(sub_questions, 1):
        print(f"[Decomposition] Sub-question {i} : '{sq}'")

    per_subquery_results = {}
    seen_ids             = {}
    rank_scores          = {}

    for sq in sub_questions:
        results = hybrid_search(
            client, collection_name, embedding_obj, bm25_ef,
            query_text=sq,
            top_k=top_k
        )
        per_subquery_results[sq] = results

        if not results or not results[0]:
            continue
        for rank, hit in enumerate(results[0], start=1):
            hit_id = hit["id"]
            rank_scores[hit_id] = rank_scores.get(hit_id, 0) + 1.0 / (60 + rank)
            if hit_id not in seen_ids:
                seen_ids[hit_id] = hit

    merged = sorted(seen_ids.values(), key=lambda h: rank_scores[h["id"]], reverse=True)[:top_k]

    # Per sub-question breakdown
    print("\n── Per Sub-question Results ──")
    for sq, res in per_subquery_results.items():
        print(f"\n  SUB-QUERY: '{sq[:60]}'")
        if res and res[0]:
            for i, hit in enumerate(res[0], start=1):
                print(f"    {i}. Page {hit['entity']['page_number']} | Score {hit['distance']:.4f} | {hit['entity']['text'][:150]}")

    return {"per_subquery": per_subquery_results, "merged": [merged]}


# ── Run ──────────────────────────────────────────────────────
decomp_results = search_with_decomposition(
    client, COLLECTION_NAME, embedding_obj, bm25_ef,
    query = "What is the leave policy and how does it affect salary deductions?",
    top_k = 5
)

# ── Print Merged Results ──────────────────────────────────────
print("\nDECOMPOSED — MERGED FINAL RESULTS")
print("=" * 55)
merged_hits = decomp_results["merged"]
if not merged_hits or not merged_hits[0]:
    print("No results returned.")
else:
    for idx, hit in enumerate(merged_hits[0], start=1):
        entity = hit["entity"]
        print(f"\nRank  : {idx}")
        print(f"Score : {hit['distance']:.4f}")
        print(f"Page  : {entity['page_number']}")
        print(f"Text  :\n{entity['text'][:400]}")

Przykład rozkładu:

Input:  "What is the leave policy and how does performance review affect salary?"
Sub-questions:
  1. "What is the annual leave policy?"
  2. "How many sick days are employees entitled to?"
  3. "How does performance review affect salary?"
  4. "What is the performance review schedule?"

Każde podpytanie samodzielnie pobiera potrzebne fragmenty tekstu, a połączona zbiór obejmuje wszystkie elementy oryginalnego pytania, czego nie byłaby w stanie osiągnąć pojedyncza wyszukiwarka.

Ponowne rankowanie za pomocą cross-encoderów

Do tej pory wszystko było przetwarzane za pomocą bi-encoderów: zapytania i dokumenty były embedowane niezależnie, a podobieństwo obliczano na podstawie iloczynu skalarnego lub odległości kosinusowej. Są szybkie, ale nie stanowią najskuteczniejszego sposobu oceny istotności. Cross-encodery analizują zapytanie i jeden dokument razem jako parę, oceniając stopień ich dopasowania poprzez analizę całej interakcji między nimi, co sprawia, że są znacznie dokładniejsze. Kosztem jest jednak wolniejsza prędkość działania – są zbyt powolne przy tysiącach dokumentów, ale szybkie przy małej liczbie kandydatów, np. 20–50. Stąd powstaje taki schemat: szerokie przeszukiwanie za pomocą bi-encodera, a węższe ponowne rankowanie z użyciem cross-encodera.

# ============================================================
# RERANKING WITH CROSS-ENCODER
# ============================================================

from sentence_transformers import CrossEncoder

# Huggingface: cross-encoder/ms-marco-MiniLM-L12-v2
cross_encoder = CrossEncoder("cross-encoder/ms-marco-MiniLM-L12-v2")
query       = "What is the leave policy?"
RETRIEVAL_K = 20   # fetch more than you need
FINAL_K     = 5    # rerank down to this

# Step 1: Broad retrieval - fetch 20 candidates
query_dense = embedding_obj.embed_query(query)

results = client.search(
    collection_name=COLLECTION_NAME,
    data=[query_dense],
    anns_field="vector",
    search_param={"metric_type": "COSINE"},
    limit=RETRIEVAL_K,
    output_fields=["text", "page_number", "source"]
)

hits = results[0]
print(f"Retrieved {len(hits)} candidates for reranking.")

# Step 2: Score each (query, chunk) pair with the cross-encoder
pairs        = [[query, hit["entity"]["text"]] for hit in hits]
rerank_scores = cross_encoder.predict(pairs)

# Step 3: Sort by cross-encoder score
for hit, score in zip(hits, rerank_scores):
    hit["rerank_score"] = float(score)
reranked = sorted(hits, key=lambda x: x["rerank_score"], reverse=True)[:FINAL_K]

# Step 4: Display
for idx, hit in enumerate(reranked, start=1):
    entity = hit["entity"]
    print(f"Rank {idx} | Rerank: {hit['rerank_score']:.4f} | Vector: {hit['distance']:.4f}")
    print(f"  Page {entity['page_number']}: {entity['text'][:300]}\n")

Wyniki cross-encodera to nienormalizowane wartości logits, a nie prawdopodobieństwa. Wyższa wartość wskazuje na silniejszą istotność, więc 8,0 znacznie przewyższa 2,0, jednak absolutna skala różni się między modelami. Użyty model, ms-marco-MiniLM-L12-v2, to silny i lekki wybór, wyszkolony na zbiorze danych MS MARCO Passage Ranking, który dobrze radzi sobie z zapytaniami w formie pytań dotyczących fragmentów dokumentów.

Jak porównują się te techniki

Rozsądna domyślna konfiguracja do produkcji to:

1. Hybrydowe wyszukiwanie jako metoda pozyskiwania danych dostępna zawsze.
2. Filtrowanie metadanych za każdym razem, gdy można określić zakres zainteresowań użytkownika.
3. Ponowna sortowanie za pomocą cross-encodera najlepszych 20 kandydatów przed przekazaniem ich do LLM.

Przepisywanie zapytania lub HyDE zostały dodane dla domen z dużą różnicą w słownictwie pomiędzy językiem użytkownika a językiem dokumentu.

Postęp realizacji wygląda następująco:

• Szukanie gęste jest punktem wyjścia: szybkie, niezależne od słów kluczowych i dobrze radzące sobie z synonimami.
• Szukanie hybrydowe stanowi ulepszenie, ponieważ dodanie rzadkich wektorów BM25 skutecznie poprawia dokładność wyników przy niewielkiej dodatkowej złożoności.
• Filtrowanie, przepisywanie, HyDE oraz dekompozycja adresują po jednym konkretnym problemie. Nie stosuj ich wszystkich bezrefleksyjnie. Wybieraj w zależności od dokumentów oraz wzorców zapytań używanych przez użytkowników.
• Ponowna sortowanie za pomocą kodera krzyżowego prawie zawsze warto dodać. Działa lokalnie, powoduje niewielki opóźnienie przy małej liczbie kandydatów i skutecznie poprawia ostateczny kontekst.

Ostatecznie jakość wydobywania informacji decyduje o jakości odpowiedzi. Model językowy może być tylko tak dobry, jak kontekst, który otrzymuje, a każda z opisanych powyżej technik ma na celu uczynienie tego kontekstu bardziej istotnym, kompletnym i wiarygodnym. Wszystko działa lokalnie w Milvus Lite, przy czym jedyną zewnętrzną zależnością służącą do generowania i przechowywania wektorów jest API OpenAI.