DRAG with KNEE pro RAG: dynamické prořezávání

DRAG s KNEE: dynamické zkrácení pro efektivní RAG

DRAG s KNEE — algoritmus dynamického retrieval augmented generation s odstřihnutím na úrovni kolena. Vytváří hierarchické vektorové stromy dokumentů v Qdrantu a používá hybridní vyhledávání s RRF pro adaptivní výběr relevantního kontextu. Řeší problémy statického top_k: přetížení kontextu nebo nedostatek informací.

Vytvoření hierarchického indexu

Dokumenty se rozdělí na stránky. Pro každou generuje LLM stručný popis a klíčová slova prostřednictvím strukturovaného výstupu. Tyto údaje tvoří listy stromu.

Listy se seskupují po 3–5 do rodičovských uzlů. Rodič zdědí klíčová slova dětí a získá svůj popis na základě jejich agregace. Proces se opakuje až do kořenového uzlu s parent_id=-1.

V Qdrantu je uložen plochý seznam bodů s metadaty pro filtrování. To umožňuje manipulaci se stromy prostřednictvím payloadu.

Hybridní vyhledávání a RRF

Porovnání bodů využívá husté a řídké vektory s fúzí prostřednictvím Reciprocal Rank Fusion:

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

kde $D$ — dokumenty, $R$ — řazení (husté, řídké), $k=60$.

Algoritmy vyhledávání

Branch Search

Najde top kořeny, poté rekurzivně porovnává rodiče s dětmi. Pokud jsou děti lepší, pokračuje hlouběji, jinak uchovává rodiče.

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: # artefakty z minulého života :)
        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)

Použití:

from rag_lib.search import branch_search

results = branch_search(query="váš_zápis", num_roots=3)
for chunk in results:
    print(f"{chunk.file_name} (strany {chunk.page_start}-{chunk.page_end}):
{chunk.text[:200]}...")

Výhoda: plnost. Nevýhoda: může přinést nepodstatné větve.

Beam Search (fixní)

• Vybereme generaci rodičů a jejich dětí.
• Vítězové rodičů vyloučí děti, vítězové dětí vyloučí rodiče.
• Přeživší soutěží o pevnou šířku paprsku.

Sníží šum, ale top_k může vyříznout relevantní nebo přidat nepotřebné.

Adaptive Beam s kolenním bodem

Dynamická šířka paprsku metodou kolena. Najde bod maximálního ohybu křivky RRF (konvexní dolů).

Dynamicky odstraňuje šum: na každé iteraci nejprve kolenní zkrácení, pak generování nového paprsku. Odpadá cca 62 % nepodstatných dokumentů.

Co je důležité

• Hierarchické stromy v Qdrantu s děděním klíčových slov.
• Hybridní vyhledávání + RRF pro porovnání uzlů.
• Kolenní zkrácení pro dynamické odstranění podle relevance.
• Branch_search pro plnost, adaptive beam pro efektivitu.
• Strukturovaný výstup LLM minimalizuje halucinace.

— Editorial Team