Fortgeschrittene RAG-Technologien in Milvus: Hybridsuche, HyDE, Abfragen umschreiben und Reranking.

Ein Proof-of-Concept-RAG-System wird in wenigen Minuten erstellt: Ein PDF wird in Blöcke aufgeteilt, diese Blöcke durch ein Embedding-Modell geleitet und die Ergebnisse an ein LLM übergeben. In der Produktivumgebung stößt diese ausschließlich auf Vektorabfragen basierende Konfiguration an ihre Grenzen. Reale Nutzer formulieren ihre Anfragen nicht wie strukturierte Datenmaschinen. Sie verwenden vage Formulierungen, lassen Schlüsselwörter aus und fassen mehrere Fragen zusammen – dabei wird durch eine einzige semantische Abfrage viel übersehen. Wenn die Suche fehlschlägt, muss das LLM irrelevante Kontexte zusammenfassen, was zu subtilen Halluzinationen oder nutzlosen Antworten führt.

Um die Lücke zwischen einfacher semantischer Suche und einem mehrstufigen Abrufprozess zu schließen, sind mehrere Schichten erforderlich. Die unten gezeigte Implementierung verwendet Milvus. Sie beginnt mit dichten Vektorabfragen, fügt eine hybride Suche hinzu, setzt anschließend vier Techniken ein (Metadatenfilterung, Anfrageumformulierung, HyDE sowie Anfragedekomposition) und endet mit einer erneuten Rangfolgebestimmung mittels Cross-Encoder. Jeder Schritt wird durch funktionierenden Code veranschaulicht.

Warum RAG existiert

Modelle wie GPT-4 lernen aus allgemeinen Internetdaten bis zu einem bestimmten Wissensstichtag. Private Dokumente, interne Wikis und Unternehmensrichtlinien sind dabei nicht erreichbar. Wenn man ein solches Modell nach der Urlaubsrichtlinie des Unternehmens fragt, antwortet es entweder, dass es es nicht weiß, oder es erfindet eine plausibel klingende, falsche Antwort.

RAG (Retrieval-Augmented Generation) teilt die Aufgabe in zwei Schritte auf:

1. Auswählen der relevantesten Abschnitte aus den Dokumenten mithilfe der semantischen Suche.
2. Erstellen einer Antwort, indem diese Abschnitte dem LLM als Kontext übergeben werden.

Der gesamte Ablauf sieht wie folgt aus:

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

Die Qualität der Endlösung hängt fast ausschließlich von der Qualität der Suche ab, daher ist es entscheidend, die Suche korrekt durchzuführen.

Wo eine Vektordatenbank ins Spiel kommt

Herkömmliche Datenbanken vergleichen exakte Werte, wodurch eine Abfrage wie WHERE text LIKE '%leave%' Dokumente überspringt, die „Urlaubsanspruch“ oder „medizinische Abwesenheit“ besagen, obwohl sie dieselbe Bedeutung haben. Eine Vektordatenbank hingegen speichert jedes Dokument als einen hochdimensionalen numerischen Vektor (eine Embedding) und führt die Suche nach semantischer Ähnlichkeit durch, also danach, wie nah die Bedeutung der Abfrage der Bedeutung jedes Dokumentsteils ist. Aus diesen Gründen eignet sich Milvus hervorragend für diese Aufgabe:

• Es handelt sich um eine Open-Source-Lösung, die selbst gehostet werden kann, wodurch die Daten privat bleiben.
• Milvus Lite läuft aus einer einzigen .db-Datei heraus, sodass für die Entwicklung kein Server erforderlich ist.

Die hybride Suche ist integriert und ermöglicht es, dichte (semantische) sowie spärliche (Schlüsselwort-)Vektoren in einer einzigen Abfrage zu kombinieren.

Sie skaliert von einem Laptop bis zu Milliarden von Vektoren ohne Änderungen am Anwendungscode.

Dichte und spärliche Embeddings

Der Pipeline werden zwei Arten von Vektoren verwendet, weshalb es sinnvoll ist, sie vor dem Schreiben von Code zu kennen. Ein dichtes Embedding ist ein Vektor fester Länge, beispielsweise mit 1024 Zahlen, der von einem neuronalen Netzwerk erzeugt wird. Alle Dimensionen enthalten Werte, und das Training lehrt das Modell, Sätze mit ähnlicher Bedeutung in der Nähe zueinander zu platzieren:

"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

Dichte Vektoren erfassen Bedeutung und Synonyme gut. Modelle wie OpenAI’s text-embedding-3-large erzeugen sie. Ein spärlicher Embedding ist ein großer Vektor, bei dem die meisten Werte null sind. Jeder nicht-nullen Wert steht für ein Wort aus dem Vokabular, und sein Wert gibt die Bedeutung dieses Wortes an – ein Gewicht im TF-IDF-Stil:

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

BM25 ist das klassische spärliche Algorithmus und zeichnet sich durch eine präzise Schlüsselwortabgleichfunktion aus. Er erkennt Eigennamen, Produktbezeichnungen und Fachbegriffe, die bei einem dichten Suchverfahren übersehen werden können, da es dafür keine naheliegenden semantischen Entsprechungen gibt.

Die hybride Suche nutzt beide Ansätze und kombiniert deren Stärken.

Einrichtung und Konfiguration

Installieren Sie zunächst die Abhängigkeiten:

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

Dann die Importe:

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

Halten Sie alle einstellbaren Parameter zusammen am Anfang der Datei:

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

Was die Größe der Blöcke angeht: 500 Zeichen mit einer Überlappung von 100 Zeichen eignen sich als Ausgangspunkt für einfache Dokumente. Erzähltexte profitieren von größeren Blöcken von 800 bis 1000 Zeichen, während dichte technische Dokumentatione besser mit kleineren Blöcken von 200 bis 300 Zeichen funktionieren. Die Überlappung sorgt dafür, dass Sätze, die an einer Blockgrenze abgeschnitten werden, dennoch auffindbar bleiben.

Der Indexierungsprozess

Die Indexierung wird einmal durchgeführt oder erneut, wenn sich die Dokumente ändern. Dabei werden Rohdokumente in Vektoren umgewandelt und in Milvus geladen. Der Ablauf ist wie folgt:

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

Zuerst initialisieren Sie die Modelle und verbinden Sie sie:

# 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.")

Laden Sie anschließend den PDF für jede Seite in ein eigenes Dokumentobjekt:

# 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 teilt zuerst an Absatzzeichen, dann an Zeilenumbrüchen und anschließend an Satzzeichen auf, wodurch verwandte Sätze so weit wie möglich in einem Block zusammenbleiben. Danach folgen die Embeddings für beide Vektortypen:

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.")

Der Aufruf bm25_ef.fit(texts) ist entscheidend. BM25 lernt das Vokabular sowie die Termfrequenzen aus Ihrem eigenen Korpus – deshalb funktioniert es. Ein generisches BM25-Modell führt zu einem deutlichen Rückgang der Qualität bei der Schlüsselwortabgleichsfunktion. Nun die Sammlung mit ihrem Schema und den Indizes:

# 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.")

Zwei Indexoptionen sind hier wichtig:

• FLAT für dichte Vektoren. Es führt eine präzise Suche nach dem nächsten Nachbarn durch, ist vollständig genau, arbeitet aber bei größeren Datenmengen langsamer. Er eignet sich für die Entwicklung sowie für Datensätze mit etwa weniger als 100.000 Elementen, während HNSW in der Produktion an seine Stelle treten sollte.
• SPARSE_INVERTED_INDEX für spärliche Vektoren. Es ist die einzige gültige Indexart für spärliche Daten in Milvus und funktioniert wie ein klassischer umgekehrter Index, ähnlich wie Suchmaschinen Dokumente nach Wörtern indizieren.

Zum Schluss werden die Datensätze vorbereitet und eingefügt:

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)}")

Der Hilfsfunktion sparse_to_dict ist erforderlich, weil BM25 eine scipy-sparse-Matrix erzeugt, während Milvus ein gewöhnliches Python-{int: float}-Wörterbuch benötigt. Sie kann mit 2D-sparse-Zeilen, 1D-Arrays sowie bereits konvertierten Wörterbüchern umgehen. Ein häufiges Problem: Rufen Sie immer zuerst client.load_collection() auf, bevor Sie suchen. Indizes werden in Milvus auf der Festplatte gespeichert und nach Bedarf in den RAM geladen; ohne diesen Aufruf führt die Suche zu Fehlern oder liefert leere Ergebnisse.

Dichte Suche als Basis

Sobald die Daten indiziert sind, beinhaltet die einfachste Suche die Eingabeabfrage und findet die nächsten Dokumentvektoren mithilfe der Kosinusähnlichkeit:

# ════════════════════════════════════════════════════════════
# 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")

Unter der COSINE-Metrik liegen die Scores zwischen -1 und 1. Werte über 0,7 deuten in der Regel auf eine starke semantische Übereinstimmung hin, während die besten Ergebnisse typischerweise zwischen 0,6 und 0,95 liegen. Für Fragen in natürlicher Sprache, deren Antworten sich im nahegelegenen semantischen Raum befinden – wie zum Beispiel die Erklärung von Elternurlaubsleistungen oder was im Verhaltenskodex steht – reicht eine dichte Suche aus. Sie hat Schwierigkeiten, wenn eine Abfrage spezifische Codes, Akronyme oder Eigennamen enthält, die das Embedding-Modell selten gesehen hat.

Hybride Suche mit RRF

Bei der hybriden Suche werden gleichzeitig eine dichte Suche und eine spärliche BM25-Suche durchgeführt, anschließend werden beide gerankten Listen mithilfe von Reciprocal Rank Fusion (RRF) zusammengeführt. Die Idee ist einfach: Der RRF-Score eines Dokuments setzt sich aus 1 / (k + Rank) aus jeder Liste zusammen, wobei k=60 als Glättungskonstante dient. Dadurch landen Dokumente, die in beiden Listen gut gerankt sind, an der Spitze.

# ════════════════════════════════════════════════════════════
# 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")

Zu den Scores: Der Wert hit['distance'] aus hybrid_search ist der kombinierte RRF-Ranking-Score und nicht die reine kosinusähnliche Ähnlichkeit. Je höher der Wert, desto besser – wobei der Bereich nicht auf 0 bis 1 beschränkt ist, was bedeutet, dass er nicht mit den kosinusähnlichen Scores einer normalen search()-Funktion verglichen werden sollte. Für produktive RAG-Anwendungen ist die hybride Suche die empfohlene Standardmethode. BM25 verursacht kaum Verzögerungen und verbessert deutlich die Recall-Rate bei Abfragen mit vielen Schlüsselwörtern.

Auch die hybride Suchmethode kann in bestimmten Situationen versagen, und vier Techniken beheben die häufigsten Fehlermuster.

Metadatenfilterung

Das Problem. Wenn eine Sammlung mehrere Dokumente, Zeiträume oder Abteilungen umfasst, kann die Vektorsuche relevante Passagen aus der falschen Quelle zurückgeben – beispielsweise eine veraltete Version einer Richtlinie, das Handbuch einer anderen Abteilung oder Seiten, die außerhalb des Themenbereichs der Frage liegen.

Wie es funktioniert. Milvus wendet eine skalare Filterausdruck vor der Vektorsuche an, sodass nur die passenden Zeilen in die ANN-Ebene gelangen. Es handelt sich dabei nicht um einen Nachfilter. Der Filter reduziert zunächst den Pool der Kandidaten, wodurch die Trefferquote innerhalb des gefilterten Untergrupsps erhalten bleibt.

# ════════════════════════════════════════════════════════════
# 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]}")

Filterausdrücke folgen dieser Syntax:

'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

Jedes mehrdokumentbasierte RAG-System sollte eine Metadatenfilterung beinhalten. Wenn Benutzer innerhalb eines bestimmten Umfangs arbeiten – beispielsweise nur mit dem Handbuch von 2024 oder nur mit Finanzdokumenten – sollte der Filter automatisch aus diesem Umfang abgeleitet werden.

Abfragen umformulieren

Das Problem. Benutzer formulieren ihre Anfragen ungezwungen, wie zum Beispiel „Wie viele freie Tage habe ich?“, während Dokumente formelle Begriffe wie „Jahresurlaubsanspruch“ oder „Richtlinie zur Anrechnung von Urlaubstagen“ verwenden. Dichte Suchalgorithmen verringern diesen Unterschied teilweise, doch eine gute Umformulierung verbessert erheblich die Ergebnisqualität bei spärlichen Suchergebnissen sowie die Gesamtrelevanz.

Wie es funktioniert. Ein LLM erstellt mehrere formelle, mit Schlüsselwörtern angereicherte Umformulierungen der ursprünglichen Abfrage; jede Variante wird separat gesucht, und RRF fügt die Ergebnisse zusammen und entfernt Duplikate.

# ════════════════════════════════════════════════════════════
# 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]}")

Ein Beispiel für eine Transformation:

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"

Jede Variante erweitert die Suchfläche. Ein Dokument, das bei der ursprünglichen Abfrage nicht unter den Top 5 liegt, könnte bei der dritten Variante auf Platz zwei geraten, und RRF bringt Dokumente zum Vorschein, die in verschiedenen Abfragen konstant auftauchen.

HyDE (Hypothetical Document Embeddings)

Das Problem. Eine kurze Frage und ihre Antwort befinden sich in unterschiedlichen Bereichen des Embeddingsraums. „Wie viele Krankheitstage?“ ist eine Frage und wird in der Nähe anderer Fragen eingebettet. Eine Richtlinie, die besagt, dass Mitarbeiter jährlich 7 bezahlte Krankheitstage erhalten, wird in der Nähe anderer Richtlinienangaben eingebettet. Diese Lücke verringert die Präzision.

Wie es funktioniert. Anstatt die Frage direkt einzubetten, erstellt das LLM ein hypothetisches Antwortdokument, das anschließend eingebettet wird. Die erfundene Antwort erscheint in der Nähe echter Antwortdokumente – selbst wenn ihre spezifischen Fakten falsch sind. Nur die Embedding-Daten werden verwendet, der Text wird weggeworfen, sodass die erfundenen Details den Benutzer niemals erreichen.

# ════════════════════════════════════════════════════════════
# 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 ist am nützlichsten bei kurzen, vagen Anfragen wie dem Prozess einer Gehaltsüberprüfung, bei Wortschatzlücken zwischen Fragen und Dokumenten sowie bei sehr informellen Formulierungen. Er ist überflüssig für präzise, mit Schlüsselwörtern reiche Anfragen, bei denen die hybride Suche bereits gut funktioniert, und wenn die Latenzzeit von LLMs eine Rolle spielt, da HyDE pro Anfrage einen zusätzlichen LLM-Aufruf erfordert.

Aufteilung der Anfrage

Das Problem. Eine komplexe, mehrteilige Frage kann nicht aus einem einzigen Abschnitt beantwortet werden. Die Frage nach der Urlaubsrichtlinie und deren Einfluss auf die Gehaltsüberprüfung betrifft zwei Themen in zwei Abschnitten. Eine einfache Vektorsuche konzentriert sich entweder auf ein Thema und übersieht das andere oder liefert für beide nur mittelmäßige Ergebnisse.

Wie es funktioniert. Ein LLM teilt die Frage in atomare Unterfragen auf, jede wird einzeln gesucht, RRF fügt die Ergebnisse zusammen, und der finale LLM-Aufruf erhält alle relevanten Teile.

# ════════════════════════════════════════════════════════════
# 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]}")

Ein Beispiel für eine Aufteilung:

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?"

Jede Unterfrage holt sich die benötigten Abschnitte eigenständig, und die zusammengeführte Menge umfasst jeden Teil der ursprünglichen Frage, was eine einzelne Suche nicht leisten könnte.

Cross-Encoder-Umrankung

Bis jetzt werden alle Daten mit Bi-Encodern abgerufen: Anfrage und Dokumente werden unabhängig eingebettet, wobei die Ähnlichkeit aus dem Skalarprodukt oder dem Kosinusabstand resultiert. Sie sind schnell, aber nicht die präziseste Methode zur Bewertung der Relevanz. Cross-Encodern lesen eine Anfrage und ein Dokument gemeinsam als Paar und bewerten, wie gut sie zueinander passen, indem sie ihre vollständige Interaktion berücksichtigen – dadurch sind sie weitaus genauer. Allerdings ist ihre Geschwindigkeit eingeschränkt; sie sind zu langsam für Tausende von Dokumenten, aber schnell bei einer kleinen Auswahl von 20 bis 50 Dokumenten. Daher das Muster: Mit einem Bi-Encoder eine breite Auswahl vornehmen, mit einem Cross-Encoder dann eine gezielte Umrankung durchführen.

# ============================================================
# 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")

Die Cross-Encoder-Werte sind unnormalisierte Logits und keine Wahrscheinlichkeiten. Ein höherer Wert signalisiert eine stärkere Relevanz – daher übertrifft 8,0 mit großem Abstand 2,0 – doch die absolute Skala unterscheidet sich zwischen den Modellen. Das verwendete Modell, ms-marco-MiniLM-L12-v2, ist eine leistungsstarke, leichtgewichtige Option, die auf dem MS MARCO Passage Ranking Dataset trainiert wurde und sich gut mit fragenbasierten Abfragen an Dokumentpassagen behauptet.

Vergleich der Techniken

Eine sinnvolle Standardkonfiguration für die Produktion ist:

1. Hybrides Suchen als immer verfügbare Grundmethode zur Informationsbeschaffung.
2. Metadatenfiltern, sobald der Suchbereich des Benutzers ermittelt werden kann.
3. Cross-Encoder-Reranking der 20 besten Kandidaten, bevor sie zum LLM weitergeleitet werden.

Abfrageschreibweise oder HyDE werden für Domänen hinzugefügt, bei denen es einen großen Wortschatzunterschied zwischen der Sprache des Benutzers und der Sprache der Dokumente gibt.

Der Fortschritt verläuft wie folgt:

• Dichte Suche ist der Ausgangspunkt: schnell, unabhängig von Schlüsselwörtern und gut bei Synonymen.
• Hybride Suche ist die Verbesserung, da das Hinzufügen von BM25-Sparse-Vektoren die Trefferquote zuverlässig bei geringer zusätzlicher Komplexität erhöht.
• Filtern, Schreibweiseanpassung, HyDE und Dekomposition beheben jeweils einen bestimmten Fehlermodus. Wenden Sie sie nicht alle blind an. Wählen Sie entsprechend den Dokumenten und den von den Benutzern tatsächlich generierten Abframustern.
• Cross-Encoder-Reranking lohnt sich fast immer. Es läuft lokal, verursacht bei einer kleinen Kandidatenliste nur geringe Verzögerungen und verbessert zuverlässig den endgültigen Kontext.

Letztendlich bestimmt die Qualität der Abfrage die Qualität der Antwort. Ein LLM kann nur so gut sein wie der ihm zur Verfügung gestellte Kontext, und jede der oben genannten Techniken dient dazu, diesen Kontext relevanter, vollständiger und zuverlässiger zu machen. Alles läuft lokal in Milvus Lite, wobei die OpenAI API die einzige externe Abhängigkeit für Embeddings und Generierung darstellt.