Skip to main content

​
Adaptive RAG

Adaptive RAG는 쿼리의 복잡도를 먼저 분석하고, 그에 맞는 최적의 RAG 전략을 선택하는 아키텍처입니다. 단순한 질문에는 검색 없이 직접 답변하고, 복잡한 질문에는 반복 검색과 품질 평가를 거치는 다단계 파이프라인을 적용합니다.

​
핵심 아이디어

모든 질문에 동일한 RAG 파이프라인을 적용하는 것은 비효율적입니다. “오늘 날씨는?”처럼 단순한 질문에 다단계 검색과 자기 평가를 수행할 필요가 없고, “양자 컴퓨팅이 암호학에 미치는 영향을 분석하라”와 같은 복잡한 질문에 단순 검색만으로는 부족합니다. Adaptive RAG는 쿼리를 세 가지 복잡도로 분류하여 각각 다른 전략을 적용합니다.
복잡도전략설명
SimpleLLM 직접 답변검색 없이 LLM의 내재 지식으로 답변
Moderate단일 검색 + 생성한 번의 검색으로 충분한 경우
Complex반복 검색 + 품질 평가다단계 검색과 생성 결과 검증을 반복

​
동작 방식

위 다이어그램에서 Moderate와 Complex 경로는 동일한 검색 노드로 진입하지만, 실제 동작이 다릅니다. Moderate는 단일 검색 → 생성으로 충분한 경우가 대부분이며, Complex는 품질 평가에서 미달 시 재검색/재생성 루프를 반복합니다. 즉, 라우팅의 핵심 차이는 진입 후 루프 실행 여부에 있습니다.
이 그래프는 Self-RAG의 자기 평가 루프와 CRAG의 웹 검색 폴백을 하나로 통합한 구조입니다.

​
LangGraph 구현

​
상태 정의

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

class GraphState(TypedDict):
    question: str
    documents: List[Document]
    generation: str
    query_complexity: str  # "simple", "moderate", "complex"
    retry_count: int

​
노드 함수

from langchain.chat_models import init_chat_model
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_community.tools import TavilySearchResults
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": 4})

web_search = TavilySearchResults(max_results=3)

def route_question(state: GraphState) -> GraphState:
    """쿼리 복잡도를 분류합니다."""
    question = state["question"]

    prompt = ChatPromptTemplate.from_messages([
        ("system", (
            "질문의 복잡도를 분류하세요.\n"
            "- simple: 상식, 정의, 간단한 사실 (검색 불필요)\n"
            "- moderate: 특정 주제에 대한 설명 (단일 검색으로 충분)\n"
            "- complex: 분석, 비교, 최신 정보 필요 (반복 검색 필요)\n"
            "'simple', 'moderate', 'complex' 중 하나만 출력하세요."
        )),
        ("human", "{question}"),
    ])

    chain = prompt | llm | StrOutputParser()
    complexity = chain.invoke({"question": question})
    return {"query_complexity": complexity.strip().lower(), "retry_count": 0}

def direct_answer(state: GraphState) -> GraphState:
    """LLM이 검색 없이 직접 답변합니다."""
    question = state["question"]

    prompt = ChatPromptTemplate.from_messages([
        ("system", "질문에 간결하게 답변하세요."),
        ("human", "{question}"),
    ])

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

def retrieve(state: GraphState) -> GraphState:
    """벡터 데이터베이스에서 문서를 검색합니다."""
    question = state["question"]
    documents = retriever.invoke(question)
    return {"documents": documents}

def grade_documents(state: GraphState) -> GraphState:
    """검색된 문서의 관련성을 평가합니다."""
    question = state["question"]
    documents = state["documents"]

    prompt = ChatPromptTemplate.from_messages([
        ("system", (
            "문서가 질문에 관련이 있는지 판단하세요.\n"
            "'relevant' 또는 'irrelevant'만 출력하세요."
        )),
        ("human", "질문: {question}\n\n문서: {document}"),
    ])

    chain = prompt | llm | StrOutputParser()

    relevant_docs = []
    for doc in documents:
        result = chain.invoke({"question": question, "document": doc.page_content})
        if "relevant" in result.lower() and "irrelevant" not in result.lower():
            relevant_docs.append(doc)

    return {"documents": relevant_docs}

def web_search_node(state: GraphState) -> GraphState:
    """웹 검색으로 추가 정보를 확보합니다."""
    question = state["question"]
    results = web_search.invoke(question)
    web_docs = [
        Document(page_content=r["content"], metadata={"source": r["url"]})
        for r in results
    ]
    return {"documents": web_docs}

def generate(state: GraphState) -> GraphState:
    """검색된 문서를 기반으로 답변을 생성합니다."""
    question = state["question"]
    documents = state["documents"]

    prompt = ChatPromptTemplate.from_messages([
        ("system", "다음 컨텍스트를 참고하여 질문에 답변하세요.\n\n{context}"),
        ("human", "{question}"),
    ])

    chain = prompt | llm | StrOutputParser()
    context = "\n\n".join(doc.page_content for doc in documents)
    generation = chain.invoke({"context": context, "question": question})
    return {"generation": generation}

def grade_generation(state: GraphState) -> GraphState:
    """생성된 답변이 문서에 근거하고 질문에 유용한지 평가합니다."""
    question = state["question"]
    documents = state["documents"]
    generation = state["generation"]

    # 근거 확인
    grounded_prompt = ChatPromptTemplate.from_messages([
        ("system", (
            "답변이 문서에 근거하는지 판단하세요.\n"
            "'grounded' 또는 'not_grounded'만 출력하세요."
        )),
        ("human", "문서: {documents}\n\n답변: {generation}"),
    ])

    # 유용성 확인
    useful_prompt = ChatPromptTemplate.from_messages([
        ("system", (
            "답변이 질문을 충분히 해결하는지 판단하세요.\n"
            "'useful' 또는 'not_useful'만 출력하세요."
        )),
        ("human", "질문: {question}\n\n답변: {generation}"),
    ])

    docs_text = "\n".join(doc.page_content for doc in documents)

    grounded_chain = grounded_prompt | llm | StrOutputParser()
    useful_chain = useful_prompt | llm | StrOutputParser()

    grounded = grounded_chain.invoke({"documents": docs_text, "generation": generation})
    useful = useful_chain.invoke({"question": question, "generation": generation})

    state["is_grounded"] = "grounded" in grounded.lower() and "not_grounded" not in grounded.lower()
    state["is_useful"] = "useful" in useful.lower() and "not_useful" not in useful.lower()
    state["retry_count"] = state.get("retry_count", 0) + 1
    return state

​
조건부 엣지

def decide_route(state: GraphState) -> Literal["direct_answer", "retrieve"]:
    """쿼리 복잡도에 따라 경로를 결정합니다."""
    complexity = state.get("query_complexity", "moderate")
    if complexity == "simple":
        return "direct_answer"
    return "retrieve"

def decide_search_or_generate(state: GraphState) -> Literal["generate", "web_search"]:
    """관련 문서가 있으면 생성, 없으면 웹 검색으로 전환합니다."""
    if state["documents"]:
        return "generate"
    return "web_search"

def check_generation_quality(state: GraphState) -> Literal["end", "generate", "retrieve"]:
    """생성 결과의 품질을 확인하고 다음 단계를 결정합니다."""
    if state.get("retry_count", 0) >= 3:
        return "end"
    if state.get("is_grounded") and state.get("is_useful"):
        return "end"
    if not state.get("is_grounded"):
        return "generate"  # 근거 부족 → 재생성
    return "retrieve"  # 질문 불일치 → 재검색

​
그래프 구성

from langgraph.graph import StateGraph, START, END

workflow = StateGraph(GraphState)

# 노드 추가
workflow.add_node("route_question", route_question)
workflow.add_node("direct_answer", direct_answer)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("web_search", web_search_node)
workflow.add_node("generate", generate)
workflow.add_node("grade_generation", grade_generation)

# 엣지 연결
workflow.add_edge(START, "route_question")
workflow.add_conditional_edges("route_question", decide_route, {
    "direct_answer": "direct_answer",
    "retrieve": "retrieve",
})
workflow.add_edge("direct_answer", END)
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges("grade_documents", decide_search_or_generate, {
    "generate": "generate",
    "web_search": "web_search",
})
workflow.add_edge("web_search", "generate")
workflow.add_edge("generate", "grade_generation")
workflow.add_conditional_edges("grade_generation", check_generation_quality, {
    "end": END,
    "generate": "generate",
    "retrieve": "retrieve",
})

# 컴파일
app = workflow.compile()

​
실행

# Simple → LLM 직접 답변
result = app.invoke({"question": "Python이란?", "retry_count": 0})

# Complex → 반복 검색 + 품질 평가
result = app.invoke({"question": "2024년 RAG 아키텍처 트렌드를 분석하라", "retry_count": 0})

print(result["generation"])

​
관련 아키텍처와의 비교

Adaptive RAG의 핵심 기여는 쿼리 복잡도 분류기(query complexity classifier)입니다. 논문에서는 질문의 복잡도에 따라 기존의 서로 다른 RAG 전략(no retrieval, single-step, multi-step)으로 라우팅하는 방식을 제안합니다.
기능Self-RAGCRAGAdaptive RAG
쿼리 라우팅--복잡도 기반 분류기
문서 관련성 평가IsRel3단계 판정구현에 따라 다름
웹 검색 폴백-Incorrect 시 전환구현에 따라 다름
생성 결과 평가IsSup + IsUse-구현에 따라 다름
핵심 기여Reflection Token검색 품질 보정쿼리 복잡도 라우팅
위 LangGraph 구현에서는 설명의 편의를 위해 문서 관련성 평가, 웹 검색 폴백, 생성 결과 평가를 모두 포함했습니다. 이는 Adaptive RAG 논문 자체의 제안이 아니라, 쿼리 라우팅 개념과 다른 RAG 기법들을 조합한 실용적인 구현 예시입니다.
Adaptive RAG는 가장 복잡한 그래프 구조를 가지지만, 쿼리 복잡도에 따라 실제 실행 경로는 달라집니다. 단순한 질문은 빠르게 처리되고, 복잡한 질문만 전체 파이프라인을 거칩니다.

​
참고 논문

논문학회링크
Adaptive-RAG: Learning to Adapt Retrieval-Augmented Large Language Models through Question Complexity (Jeong et al., 2024)NAACL 2024arXiv 2403.14403
Self-RAG: Learning to Retrieve, Generate, and Critique (Asai et al., 2023)ICLR 2024arXiv 2310.11511
Corrective Retrieval Augmented Generation (Yan et al., 2024)-arXiv 2401.15884