Skip to main content

RAPTOR (Recursive Abstractive Processing for Tree-Organized Retrieval)

RAPTOR는 문서를 재귀적으로 클러스터링하고 요약하여 트리 구조를 구축한 뒤, 다양한 추상화 수준에서 검색을 수행하는 RAG 아키텍처입니다. 기존 RAG가 원본 텍스트 청크만 검색하는 반면, RAPTOR는 요약된 상위 노드도 검색 대상에 포함하여 세부 사실과 고수준 주제 모두를 커버합니다.

핵심 아이디어

기존 RAG의 검색은 원본 텍스트를 고정 크기 청크로 분할한 뒤, 질문과 가장 유사한 청크를 반환합니다. 이 방식은 세부 정보 검색에는 효과적이지만, 문서 전체를 아우르는 고수준 질문에는 취약합니다.
  • 고정 크기 청크는 문맥을 잘라내어 의미 단위가 분리됨
  • “이 문서의 핵심 주장은?” 같은 요약형 질문에 적절한 청크가 없음
  • 여러 청크에 걸친 정보를 종합해야 하는 질문에 답변 어려움
  • 검색 대상이 모두 동일한 추상화 수준(원본 텍스트)

트리 구축 과정

RAPTOR의 핵심은 인덱싱 단계에서 트리를 구축하는 것입니다.
1

텍스트 청킹

문서를 의미 단위의 청크로 분할합니다. 이 청크들이 트리의 리프 노드가 됩니다.
2

임베딩 및 클러스터링

청크를 임베딩한 뒤, UMAP으로 차원 축소하고 GMM(Gaussian Mixture Model)으로 클러스터링합니다. 소프트 클러스터링을 사용하여 하나의 청크가 여러 클러스터에 속할 수 있습니다 (Sarthi et al., 2024, Section 3).
3

클러스터 요약

각 클러스터의 텍스트를 LLM으로 요약합니다. 이 요약이 트리의 상위 노드가 됩니다.
4

재귀 반복

요약 노드들을 다시 임베딩 → 클러스터링 → 요약하여 더 상위 노드를 생성합니다. 클러스터가 하나로 수렴하거나 최대 깊이에 도달할 때까지 반복합니다.

검색 전략

RAPTOR는 두 가지 검색 전략을 제공합니다.
루트에서 시작하여 각 레이어에서 가장 관련성 높은 노드를 선택하며 하향 탐색합니다.
  • 탐색 범위를 점진적으로 좁혀 효율적
  • 상위 노드에서 잘못된 경로를 선택하면 관련 리프를 놓칠 수 있음

LangGraph 구현

상태 정의

from typing import TypedDict, List
from langchain_core.documents import Document

class RaptorState(TypedDict):
    question: str
    documents: List[Document]
    generation: str

트리 구축 (인덱싱)

import numpy as np
from sklearn.mixture import GaussianMixture
from langchain.chat_models import init_chat_model
from langchain_openai import OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = init_chat_model("gpt-4o-mini", temperature=0)
embeddings = OpenAIEmbeddings()

SUMMARY_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "다음 텍스트들의 핵심 내용을 종합하여 간결하게 요약하세요."),
    ("human", "{texts}"),
])

summary_chain = SUMMARY_PROMPT | llm | StrOutputParser()

def cluster_and_summarize(texts: list[str], n_clusters: int = None) -> list[str]:
    """텍스트를 클러스터링하고 각 클러스터를 요약합니다."""
    if len(texts) <= 1:
        return texts

    # 임베딩
    vecs = np.array(embeddings.embed_documents(texts))

    # 클러스터 수 결정
    if n_clusters is None:
        n_clusters = max(2, len(texts) // 3)
    n_clusters = min(n_clusters, len(texts))

    # GMM 클러스터링
    gmm = GaussianMixture(n_components=n_clusters, random_state=42)
    gmm.fit(vecs)
    probs = gmm.predict_proba(vecs)

    # 소프트 클러스터링: 확률 임계값 이상인 클러스터에 할당
    threshold = 0.1
    clusters = {i: [] for i in range(n_clusters)}
    for idx, prob in enumerate(probs):
        for cluster_id, p in enumerate(prob):
            if p >= threshold:
                clusters[cluster_id].append(texts[idx])

    # 각 클러스터 요약
    summaries = []
    for cluster_texts in clusters.values():
        if cluster_texts:
            combined = "\n\n".join(cluster_texts)
            summary = summary_chain.invoke({"texts": combined})
            summaries.append(summary)

    return summaries

def build_raptor_tree(leaf_texts: list[str], max_depth: int = 3) -> list[str]:
    """RAPTOR 트리를 구축하고 모든 레이어의 노드를 반환합니다."""
    all_nodes = list(leaf_texts)  # 리프 노드 포함
    current_layer = leaf_texts

    for depth in range(max_depth):
        if len(current_layer) <= 1:
            break
        summaries = cluster_and_summarize(current_layer)
        all_nodes.extend(summaries)
        current_layer = summaries

    return all_nodes

노드 함수

from langchain_chroma import Chroma

# 트리의 모든 노드를 벡터 DB에 저장 (인덱싱 시 1회 실행)
# all_nodes = build_raptor_tree(leaf_texts)
# vectorstore = Chroma.from_texts(all_nodes, embeddings, collection_name="raptor")

vectorstore = Chroma(collection_name="raptor", embedding_function=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

def retrieve(state: RaptorState) -> RaptorState:
    """Collapsed Tree 방식으로 모든 추상화 수준에서 검색합니다."""
    question = state["question"]
    documents = retriever.invoke(question)
    return {"documents": documents}

그래프 구성

from langgraph.graph import StateGraph, START, END

workflow = StateGraph(RaptorState)

workflow.add_node("retrieve", retrieve)
workflow.add_node("generate", generate)

workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", END)

app = workflow.compile()

실행

# 세부 사실 질문 → 리프 노드에서 검색
result = app.invoke({"question": "RAPTOR에서 GMM 클러스터링의 임계값은?"})

# 고수준 주제 질문 → 상위 요약 노드에서 검색
result = app.invoke({"question": "이 문서의 전체 논지를 요약해줘"})

print(result["generation"])
Collapsed Tree 방식은 리프 노드와 요약 노드를 모두 포함하므로 벡터 DB의 크기가 증가합니다 (구현에 따라 다르나, 일반적으로 원본 대비 1.5~2배 수준으로 추정). 대규모 문서에서는 트리 깊이(max_depth)를 조절하여 인덱스 크기를 관리하세요.

기존 RAG와의 비교

항목Naive RAGRAPTOR
검색 대상원본 청크만원본 청크 + 다층 요약
요약형 질문취약상위 요약 노드에서 검색
세부 사실 질문양호리프 노드에서 검색
인덱싱 비용낮음높음 (클러스터링 + LLM 요약)
벡터 DB 크기원본 크기증가 (구현에 따라 상이)
검색 전략단일Tree Traversal / Collapsed Tree
RAPTOR의 트리 구축에는 LLM 호출이 다수 발생합니다. 리프 노드 수가 많을수록 클러스터링과 요약 비용이 기하급수적으로 증가하므로, 대규모 문서에서는 청크 크기와 트리 깊이를 적절히 조절해야 합니다.

적용 시나리오

소설, 기술 보고서, 법률 문서 등 긴 문서에서 세부 사실과 전체 주제 모두에 답변해야 하는 경우에 적합합니다. 요약 트리를 통해 “3장의 핵심 논점은?”과 같은 질문에도 대응합니다.
교과서나 강의 자료에서 개념 정의(세부)와 단원 간 관계(고수준) 모두를 검색해야 하는 교육 Q&A 시스템에 활용됩니다.
여러 섹션으로 구성된 리서치 보고서에서, 특정 데이터 포인트 검색과 전체 인사이트 요약을 하나의 시스템으로 처리합니다.

참고 논문

논문학회링크
RAPTOR: Recursive Abstractive Processing for Tree-Organized Retrieval (Sarthi et al., 2024)ICLR 2024arXiv 2401.18059