DRAG with KNEE for RAG: dynamic pruning

DRAG with KNEE: Dynamic Chunking for High-Performance RAG

DRAG with KNEE is a dynamic retrieval-augmented generation algorithm that uses knee-point pruning to optimize context selection. It builds hierarchical vector trees in Qdrant and leverages hybrid search with Reciprocal Rank Fusion (RRF) to adaptively retrieve relevant context—solving the limitations of static top_k approaches, such as context overflow or information loss.

Building Hierarchical Indexes

Documents are split into pages. For each page, an LLM generates a concise summary and key terms via structured output—these become leaf nodes in the tree.

Leaves are grouped into parent nodes (3–5 per group). Each parent inherits child keywords and generates its own summary by aggregating child content. This process repeats until reaching the root node (parent_id = -1).

In Qdrant, data is stored as a flat list of points with metadata for filtering. This enables flexible manipulation of the hierarchical structure through payload fields.

Hybrid Search & RRF

Point comparison combines dense and sparse embeddings using Reciprocal Rank Fusion:

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

where $D$ is the set of documents, $R$ represents rankings (dense, sparse), and $k=60$.

Search Algorithms

Branch Search

Finds top-level roots, then recursively compares parents against children. If children score better, it dives deeper; otherwise, it locks onto the parent.

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)

Usage:

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}):\n{chunk.text[:200]}...")

Pros: High recall. Cons: May include irrelevant branches.

Beam Search (Fixed)

• Take current generation of parents and their children.
• Winners among parents eliminate their children; winners among children eliminate their parents.
• Survivors compete for a fixed beam width.

Reduces noise but risks cutting off relevant content or including irrelevant ones.

Adaptive Beam with Knee-Point

Dynamic beam width using the knee-point method. Identifies the inflection point on the RRF curve (downward convexity).

Dynamically prunes noise: at each iteration, apply knee-pruning first, then generate the next beam. Discards ~62% of irrelevant documents.

Key Takeaways

• Hierarchical trees in Qdrant with inherited keywords.
• Hybrid search + RRF for effective node comparison.
• Knee-point pruning for dynamic relevance-based chunking.
• Branch search for completeness, adaptive beam for efficiency.
• Structured LLM output minimizes hallucinations.

— Editorial Team