RAG용 DRAG with KNEE: 동적 가지치기

무릎 기반 동적 청크화: 고성능 RAG의 핵심 기술

DRAG with KNEE는 무릎점 절단 기법을 활용해 컨텍스트 선택을 최적화하는 동적 검색 증강 생성 알고리즘입니다. 이 방법은 Qdrant에서 계층형 벡터 트리를 구축하고, 밀도와 희소 임베딩을 결합한 하이브리드 검색과 상호 순위 융합(RRF)을 통해 유연하게 관련 컨텍스트를 추출함으로써, 정적 top_k 접근 방식의 한계—예를 들어 컨텍스트 과잉 또는 정보 손실—을 해결합니다.

계층적 인덱스 구축

문서는 페이지 단위로 분할됩니다. 각 페이지에 대해 LLM이 구조화된 출력을 통해 간결한 요약문과 핵심 용어를 생성하며, 이는 트리의 리프 노드가 됩니다.

리프들은 부모 노드(3~5개씩 그룹화)로 묶입니다. 각 부모는 자식의 키워드를 상속받고, 자식 콘텐츠를 집계하여 자체 요약문을 생성합니다. 이 과정은 루트 노드(parent_id = -1)에 도달할 때까지 반복됩니다.

Qdrant에서는 데이터가 메타데이터 필드를 갖는 평면 리스트 형태로 저장되며, 이는 페이로드 필드를 통해 계층 구조를 유연하게 조작할 수 있게 합니다.

하이브리드 검색 및 RRF

점 비교는 밀도 임베딩과 희소 임베딩을 결합한 Reciprocal Rank Fusion(RRF) 방식을 사용합니다:

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

여기서 $D$는 문서 집합, $R$은 순위 기준(밀도, 희소), $k=60$입니다.

검색 알고리즘

브랜치 검색 (Branch Search)

최상위 루트를 찾은 후, 부모와 자식을 재귀적으로 비교합니다. 자식이 더 높은 점수를 받으면 깊이 탐색을 진행하고, 그렇지 않으면 부모 노드를 고정합니다.

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)

사용 예시:

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

장점: 높은 재현율. 단점: 무관한 분기 포함 가능성 있음.

비임계 빔 검색 (Fixed Beam Search)

• 현재 부모와 자식 세대를 취합니다.
• 부모 중 우승자는 자식을 제거하고, 자식 중 우승자는 부모를 제거합니다.
• 생존자들이 고정된 빔 너비를 위해 경쟁합니다.

노이즈 감소 효과 있지만, 관련 콘텐츠를 놓칠 수 있거나 무관한 내용을 포함할 위험이 있습니다.

적응형 빔 검색 (무릎점 기반)

무릎점 방법을 활용한 동적 빔 너비. RRF 곡선의 전환점(하향 볼록성)을 식별합니다.

동적으로 노이즈를 제거: 각 반복마다 먼저 무릎 절단을 적용한 후 다음 빔을 생성합니다. 약 62%의 무관 문서를 배제합니다.

핵심 요약

• Qdrant 내 계층적 트리 구조, 키워드 상속 가능.
• 하이브리드 검색 + RRF로 효율적인 노드 비교.
• 무릎점 절단을 통한 동적 관련성 기반 청크화.
• 완전성 확보를 위한 브랜치 검색, 효율성 강화를 위한 적응형 빔 검색.
• 구조화된 LLM 출력으로 환각 현상 최소화.

— Editorial Team