DRAG with KNEE dla RAG: dynamiczne przycinanie

DRAG z KNEE: dynamiczna obcinanie dla skutecznego RAG

DRAG z KNEE — algorytm dynamicznego uzupełniania generacji z obcinaniem w punkcie kolana. Buduje hierarchiczne drzewa wektorowe dokumentów w Qdrant i wykorzystuje hybrydowy wyszukiwarkę z RRF do adaptacyjnego wyboru kontekstu. Rozwiązuje problemy związane ze statycznym top_k: nadmiar kontekstu lub jego niedobór.

Budowanie hierarchicznego indeksu

Dokumenty dzielone są na strony. Dla każdej LLM generuje skrócone opisy i kluczowe słowa za pomocą strukturalnego wyjścia. Te dane tworzą liście drzewa.

Liście grupowane są po 3–5 sztuk w węzły nadrzędne. Węzeł nadrzędny dziedziczy kluczowe słowa dzieci i otrzymuje własne opisy na podstawie ich agregacji. Proces powtarza się aż do węzła głównego z parent_id=-1.

W Qdrant przechowywany jest płaski list punktów z metadane do filtrowania. Pozwala to manipulować drzewami poprzez payload.

Hybrydowe wyszukiwanie i RRF

Porównanie punktów wykorzystuje gęste i rzadkie embeddowania z połączeniem przez Reciprocal Rank Fusion:

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

gdzie $D$ — dokumenty, $R$ — rankingi (gęste, rzadkie), $k=60$.

Algorytmy wyszukiwania

Branch Search

Znajduje top-k korzeni, a następnie rekurencyjnie porównuje rodziców z dziećmi. Jeśli dzieci są lepsze — przechodzi głębiej, w przeciwnym razie zapisuje rodzica.

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 poprzedniego życia :)
        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)

Użycie:

from rag_lib.search import branch_search

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

Zalety: pełność. Wady: może przyciągać nieistotne gałęzie.

Beam Search (fixed)

• Bierze pokolenie rodziców i ich dzieci.
• Zwycięzcy rodziców wykluczają dzieci, zwycięzcy dzieci — rodziców.
• Przetrwali konkuruje o stałą szerokość promienia.

Zmniejsza szum, ale top_k może obciąć istotne lub dodać zbędne.

Adaptive Beam z punktem kolena

Dynamiczna szerokość promienia przez metodę punktu kolena. Znajduje punkt maksymalnego zgięcia krzywej RRF (wypukłość w dół).

Dynamicznie obcina szum: na każdej iteracji najpierw obcinanie kolena, potem generacja nowego promienia. Odrzuca ok. 62% nieistotnych dokumentów.

Co ważne

• Hierarchiczne drzewa w Qdrant z dziedziczeniem kluczowych słów.
• Hybrydowe wyszukiwanie + RRF do porównania węzłów.
• Obcinanie punktu kolena do dynamicznego obcinania według relevancji.
• Branch_search do pełności, adaptive beam do efektywności.
• Strukturalne wyjście LLM minimalizuje halucynacje.

— Editorial Team