DRAG mit KNEE für RAG: dynamisches Pruning

DRAG mit KNEE: Dynamische Chunking für leistungsstarke RAG-Systeme

DRAG mit KNEE ist ein dynamischer Retrieval-augmented Generation-Algorithmus, der die Kniepunkt-Pruning-Methode nutzt, um die Kontextauswahl zu optimieren. Er erstellt hierarchische Vektortrees in Qdrant und setzt hybride Suche mit Reciprocal Rank Fusion (RRF) ein, um kontextuell relevante Informationen adaptiv abzurufen – und löst so die Grenzen statischer top_k-Ansätze wie Kontextüberlastung oder Informationsverlust.

Hierarchische Indizes aufbauen

Dokumente werden in Seiten aufgeteilt. Für jede Seite generiert ein LLM eine präzise Zusammenfassung und Schlüsselbegriffe über strukturierte Ausgabe – diese werden zu Blattknoten im Baum.

Blätter werden zu Elternknoten gruppiert (3–5 pro Gruppe). Jeder Elternknoten übernimmt die Schlüsselwörter seiner Kinder und erzeugt seine eigene Zusammenfassung durch Aggregation des Kind-Inhalts. Dieser Prozess wiederholt sich, bis der Wurzelknoten erreicht ist (parent_id = -1).

In Qdrant wird die Daten als flache Liste von Punkten mit Metadaten für Filterung gespeichert. Dadurch ermöglicht es flexible Manipulation der hierarchischen Struktur über Payload-Felder.

Hybride Suche & RRF

Die Punktvergleichslogik kombiniert dichte und spärliche Embeddings mittels Reciprocal Rank Fusion:

$$RRFscore(d \in D) = \sum_{r \in R} \frac{1}{k + r(d)}$$

wobei $D$ die Menge der Dokumente, $R$ die Rankings (dicht, spärlich) repräsentiert und $k=60$.

Suchalgorithmen

Zweig-Suche

Findet oberste Wurzeln, dann vergleicht rekursiv Eltern mit Kindern. Wenn Kinder besser abschneiden, taucht sie tiefer ein; andernfalls bleibt man beim Elternknoten.

def parent_vs_children(query: str, parent: ScoredPoint) -> list[ScoredPoint]:
    file_name = parent.payload["file_name"]
    parent_id = parent.payload["id"]
    child_ids = parent.payload["child_ids"]
    all_ids = [parent_id] + child_ids

    if len(child_ids) == 0:
        console.print(
            f"Parent with id {parent_id} is leave!", style="bold underline violet"
        )
        return []

    console.print()
    console.print("#" * 20, style="red")
    console.print(
        f"Children (ids: {child_ids}) [underline]VS[/underline] Parent (id: {parent_id}) [ FILE: {file_name} ]\nFor query: {query[:20]}...",
        style="bold violet",
    )

    with RAGalicClient() as client: # artifacts from a previous life :)
        dense_model: str = client.client.embedding_model_name  # type: ignore
        sparse_model: str = client.client.sparse_embedding_model_name  # type: ignore
        children_and_parent_filter = Filter(
            must=FieldCondition(key="id", match=MatchAny(any=all_ids))
        )
        sorted_points = client.client.query_points(
            collection_name="ragalic",
            prefetch=[
                Prefetch(
                    query=Document(text=query, model=dense_model),
                    using="dense",
                    limit=len(all_ids),
                    filter=children_and_parent_filter,
                ),
                Prefetch(
                    query=Document(text=query, model=sparse_model),
                    using="sparse",
                    limit=len(all_ids),
                    filter=children_and_parent_filter,
                ),
            ],
            query=FusionQuery(fusion=Fusion.RRF),
            query_filter=children_and_parent_filter,
            limit=len(all_ids),
        ).points

    children_better_then_parent = []
    for point in sorted_points:
        if point.payload["id"] == parent_id:
            break
        children_better_then_parent.append(point)

    ids = [child.payload["id"] for child in children_better_then_parent]
    console.print(
        f"Child ids which is better than parent: {ids if ids else 'N/A'}",
        style="italic purple",
    )
    console.print("#" * 20, style="red")
    console.print()

    return children_better_then_parent


def branch_search(query: str, num_roots: int = 3) -> list[Chunk]:
    roots = find_roots(query=query, num_to_find=num_roots)

    final_points = []
    points_to_process = roots
    while len(points_to_process) > 0:
        new_point_to_process = []
        for point in points_to_process:
            new_points = parent_vs_children(query=query, parent=point)
            if len(new_points) == 0:
                console.print(
                    f"--- NEW FINAL POINT (id: {point.payload['id']} | file: {point.payload['file_name']} | pages: {point.payload['page_start']} - {point.payload['page_end']}) ---",
                    style="bold green",
                )
                final_points.append(point)
            else:
                new_point_to_process.extend(new_points)
        points_to_process = new_point_to_process

    console.print(
        f"--- WAS FOUND {len(final_points)} POINTS FOR QUERY: '{query[:20]}...' ---",
        style="bold green on white",
    )

    return prepare_chunks(final_points)

Verwendung:

from rag_lib.search import branch_search

results = branch_search(query="your_query", num_roots=3)
for chunk in results:
    print(f"{chunk.file_name} (pages {chunk.page_start}-{chunk.page_end}):
{chunk.text[:200]}...")

Vorteile: Hohe Rückrufquote. Nachteile: Kann irrelevante Zweige einschließen.

Beam-Suche (fest)

• Nimm aktuelle Generation von Eltern und deren Kindern.
• Gewinner unter den Eltern eliminieren ihre Kinder; Gewinner unter den Kindern eliminieren ihre Eltern.
• Überlebende konkurrieren um eine feste Beam-Breite.

Reduziert Rauschen, birgt aber das Risiko, relevante Inhalte auszuschließen oder irrelevante einzubeziehen.

Adaptive Beam mit Kniepunkt

Dynamische Beam-Breite mittels Kniepunkt-Methode. Identifiziert den Wendepunkt auf der RRF-Kurve (abnehmende Konvexität).

Dynamisch prüft Rauschen: In jeder Iteration wird zunächst Knie-Pruning angewendet, danach wird der nächste Beam generiert. Verwirft ca. 62 % irrelevanter Dokumente.

Wichtige Erkenntnisse

• Hierarchische Bäume in Qdrant mit vererbten Schlüsselbegriffen.
• Hybride Suche + RRF für effektiven Knotenvergleich.
• Kniepunkt-Pruning für dynamisches, relevanzbasiertes Chunking.
• Zweig-Suche für Vollständigkeit, adaptiver Beam für Effizienz.
• Strukturierte LLM-Ausgabe minimiert Halluzinationen.

— Editorial Team