DeepRAG

DeepRAG는 검색 쿼리를 원자 명제(atomic proposition) 단위로 분해하고, 트리 기반 구조에서 각 노드마다 “검색할 것인가, 더 분해할 것인가”를 결정하는 RAG 아키텍처입니다. 복잡한 질문을 가장 작은 사실 단위까지 분해한 뒤, 각 단위에 대해 정밀하게 검색하여 답변의 정확도를 높입니다.

핵심 아이디어

기존 쿼리 분해 방식은 질문을 하위 질문으로 나누지만, 분해 단위가 일정하지 않고 검색 시점도 고정적입니다. DeepRAG는 두 가지 핵심 메커니즘을 도입합니다.

원자 명제 분해: 질문을 더 이상 나눌 수 없는 최소 사실 단위(atomic proposition)까지 분해합니다
동적 검색 결정: 트리의 각 노드에서 “이 명제는 검색이 필요한가, 아니면 더 분해해야 하는가”를 모델이 판단합니다

원자 명제(atomic proposition)란 하나의 독립적인 사실만을 담는 최소 단위입니다. 예를 들어 “A는 B 대학에서 C를 전공한 D 분야의 연구자이다”는 “A는 B 대학 소속이다”, “A는 C를 전공했다”, “A는 D 분야 연구자이다”로 분해됩니다.

동작 방식

기존 쿼리 분해와의 차이

항목	일반 Decomposition	DeepRAG
분해 단위	하위 질문 (크기 불균일)	원자 명제 (최소 사실 단위)
분해 깊이	1단계 (고정)	다단계 (동적, 트리 구조)
검색 시점	분해 후 일괄 검색	각 노드에서 개별 결정
검색 결정	모든 하위 질문을 검색	원자 수준에서만 검색, 복합이면 추가 분해
구조	평면적 리스트	트리 (재귀적 분해)
검색 정밀도	중간 (넓은 쿼리)	높음 (좁고 정확한 쿼리)

LangGraph 구현

상태 정의

from typing import TypedDict, List

class DeepRAGState(TypedDict):
    question: str
    atomic_queries: List[str]
    retrieval_results: List[str]
    pending_queries: List[str]
    answer: str

노드 함수

질문 분해
검색/분해 결정
원자 검색
답변 종합

from langchain.chat_models import init_chat_model
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

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

embeddings = OpenAIEmbeddings()
vectorstore = Chroma(collection_name="documents", embedding_function=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

MAX_DECOMPOSITION_DEPTH = 3

def decompose(state: DeepRAGState) -> DeepRAGState:
    """질문을 원자 명제 수준의 하위 질문으로 분해합니다."""
    question = state["question"]
    pending = state.get("pending_queries", [])

    # 처음 호출 시 원본 질문을 분해
    target = pending[0] if pending else question

    prompt = ChatPromptTemplate.from_messages([
        ("system", (
            "복잡한 질문을 답변에 필요한 원자적(atomic) 하위 질문으로 분해하세요.\n"
            "원자적 질문이란 하나의 사실만을 묻는 가장 작은 단위의 질문입니다.\n"
            "각 하위 질문을 한 줄에 하나씩, 번호 없이 출력하세요.\n"
            "이미 충분히 원자적이면 원래 질문을 그대로 출력하세요."
        )),
        ("human", "{question}"),
    ])

    chain = prompt | llm | StrOutputParser()
    result = chain.invoke({"question": target})
    sub_queries = [q.strip() for q in result.strip().split("\n") if q.strip()]

    # pending에서 현재 처리한 쿼리 제거
    remaining_pending = pending[1:] if pending else []

    return {"pending_queries": sub_queries + remaining_pending}

def decide_action(state: DeepRAGState) -> DeepRAGState:
    """각 pending 쿼리가 원자적인지 판단하여 검색 또는 추가 분해를 결정합니다."""
    pending = state.get("pending_queries", [])
    atomic = state.get("atomic_queries", [])

    if not pending:
        return state

    current_query = pending[0]

    prompt = ChatPromptTemplate.from_messages([
        ("system", (
            "다음 질문이 원자적(atomic)인지 판단하세요.\n"
            "원자적 질문: 하나의 사실만을 묻는 단순한 질문 (예: 'X의 저자는?')\n"
            "복합 질문: 여러 사실을 동시에 묻는 질문 (예: 'X와 Y의 차이점은?')\n"
            "'atomic' 또는 'complex'만 출력하세요."
        )),
        ("human", "{question}"),
    ])

    chain = prompt | llm | StrOutputParser()
    result = chain.invoke({"question": current_query})

    if "atomic" in result.strip().lower():
        # 원자적이면 검색 대상에 추가
        updated_atomic = list(atomic) + [current_query]
        return {
            "atomic_queries": updated_atomic,
            "pending_queries": pending[1:],
        }
    else:
        # 복합이면 pending에 유지 (다음 decompose에서 처리)
        return state

def retrieve_atomic(state: DeepRAGState) -> DeepRAGState:
    """원자 쿼리에 대해 검색을 수행합니다."""
    atomic_queries = state.get("atomic_queries", [])
    existing_results = state.get("retrieval_results", [])

    new_results = []
    for query in atomic_queries:
        documents = retriever.invoke(query)
        context = "\n".join(doc.page_content for doc in documents)
        new_results.append(f"[{query}]\n{context}")

    return {
        "retrieval_results": existing_results + new_results,
        "atomic_queries": [],  # 처리 완료
    }

def synthesize(state: DeepRAGState) -> DeepRAGState:
    """모든 원자 검색 결과를 종합하여 최종 답변을 생성합니다."""
    question = state["question"]
    results = state.get("retrieval_results", [])

    context = "\n\n".join(results)

    prompt = ChatPromptTemplate.from_messages([
        ("system", (
            "다음은 질문을 원자 단위로 분해하여 각각 검색한 결과입니다.\n"
            "모든 결과를 종합하여 원래 질문에 답변하세요.\n\n"
            "{context}"
        )),
        ("human", "{question}"),
    ])

    chain = prompt | llm | StrOutputParser()
    answer = chain.invoke({"context": context, "question": question})
    return {"answer": answer}

조건부 엣지

from typing import Literal

def route_action(state: DeepRAGState) -> Literal["decompose", "retrieve", "synthesize"]:
    """pending 쿼리 상태에 따라 다음 행동을 결정합니다."""
    pending = state.get("pending_queries", [])
    atomic = state.get("atomic_queries", [])

    # 검색 대상이 있으면 먼저 검색
    if atomic:
        return "retrieve"
    # 아직 처리할 쿼리가 있으면 결정 계속
    if pending:
        return "decompose"
    # 모든 처리 완료
    return "synthesize"

def route_after_decide(state: DeepRAGState) -> Literal["decide", "decompose", "retrieve", "synthesize"]:
    """decide_action 후 다음 단계를 결정합니다."""
    pending = state.get("pending_queries", [])
    atomic = state.get("atomic_queries", [])

    if atomic:
        return "retrieve"
    if pending:
        return "decide"
    return "synthesize"

그래프 구성

from langgraph.graph import StateGraph, START, END

workflow = StateGraph(DeepRAGState)

# 노드 추가
workflow.add_node("decompose", decompose)
workflow.add_node("decide_action", decide_action)
workflow.add_node("retrieve_atomic", retrieve_atomic)
workflow.add_node("synthesize", synthesize)

# 엣지 연결
workflow.add_edge(START, "decompose")
workflow.add_edge("decompose", "decide_action")

workflow.add_conditional_edges("decide_action", route_after_decide, {
    "decide": "decide_action",
    "decompose": "decompose",
    "retrieve": "retrieve_atomic",
    "synthesize": "synthesize",
})

workflow.add_conditional_edges("retrieve_atomic", route_action, {
    "decompose": "decompose",
    "retrieve": "retrieve_atomic",
    "synthesize": "synthesize",
})

workflow.add_edge("synthesize", END)

# 컴파일
app = workflow.compile()

실행

result = app.invoke({
    "question": "BERT와 GPT의 사전학습 방식 차이를 설명하고, 각각 어떤 태스크에 적합한지 비교하라",
    "pending_queries": [],
    "atomic_queries": [],
    "retrieval_results": [],
})
print(result["answer"])

장단점

항목	장점	단점
검색 정밀도	원자 단위 검색으로 높은 정밀도	-
복잡한 질문	트리 기반 재귀 분해로 효과적 대응	-
불필요한 검색	검색 필요 여부를 동적 판단하여 절약	-
LLM 호출 비용	-	분해 + 판단 + 검색마다 호출 발생
지연 시간	-	재귀 분해로 인한 순차적 처리 시간 증가
분해 품질	-	LLM의 분해 능력에 크게 의존

재귀적 분해는 무한 루프의 위험이 있습니다. 위 구현에서는 decide_action이 원자적 판단을 수행하지만, 실제 프로덕션에서는 분해 깊이 제한(MAX_DECOMPOSITION_DEPTH)을 decompose 함수 내에서 엄격히 적용하여 과도한 분해를 방지해야 합니다.

적용 시나리오

복잡한 비교 질문

“A, B, C 세 기술의 성능, 비용, 확장성을 비교하라”처럼 여러 축과 여러 대상이 동시에 관여하는 질문을 원자 단위로 분해하여 누락 없이 검색합니다.

팩트 체크

복합적인 주장을 개별 사실 단위로 분해한 뒤, 각 사실에 대해 독립적으로 검증하여 정확한 팩트 체크를 수행합니다.

법률/규정 질의

“A 조건과 B 조건을 모두 만족할 때 C 규정이 적용되는가?”와 같은 조건부 질문을 각 조건별로 분해하여 정밀하게 검색합니다.

참고 논문

논문	학회	링크
DeepRAG: Thinking to Retrieval Step by Step for Large Language Models (Ling et al., 2025)	-	arXiv 2502.01142

RAG 개요

RAG 개발 단계

RAG 성능 향상 전략

RAG 아키텍처

DeepRAG

DeepRAG

핵심 아이디어

동작 방식

기존 쿼리 분해와의 차이

LangGraph 구현

상태 정의

노드 함수

조건부 엣지

그래프 구성

실행

장단점

적용 시나리오

참고 논문

RAG 개요

RAG 개발 단계

RAG 성능 향상 전략

RAG 아키텍처

​DeepRAG

​핵심 아이디어

​동작 방식

​기존 쿼리 분해와의 차이

​LangGraph 구현

​상태 정의

​노드 함수

​조건부 엣지

​그래프 구성

​실행

​장단점

​적용 시나리오

​참고 논문

DeepRAG

핵심 아이디어

동작 방식

기존 쿼리 분해와의 차이

LangGraph 구현

상태 정의

노드 함수

조건부 엣지

그래프 구성

실행

장단점

적용 시나리오

참고 논문