Agentic RAG 첫걸음 — Query Routing과 Adaptive Retrieval

RAG가 "서울 날씨 알려줘"는 잘 답하는데, "작년 대비 올해 서울 날씨 변화를 분석해줘"는 못 답합니다. 왜? 단일 벡터 검색으로는 이런 복합 질문을 처리할 수 없기 때문입니다.

기존 RAG는 질문이 들어오면 무조건 벡터 DB에서 유사 문서를 검색합니다. 그런데 현실의 질문은 훨씬 복잡합니다. 실시간 뉴스가 필요할 수도 있고, SQL 쿼리로 구조화 데이터를 뽑아야 할 수도 있고, 아예 검색이 필요 없는 일반 상식 질문일 수도 있습니다.

Agentic RAG는 이 문제를 해결합니다. LLM이 질문을 분석하고, 최적의 검색 전략을 스스로 결정하고, 여러 소스를 조합해서 답변을 생성합니다. 이 글에서는 Agentic RAG의 첫 번째 핵심 기법인 Query Routing과 Adaptive Retrieval을 다룹니다.

시리즈: Part 1 (이 글) | Part 2: Self-RAG과 Corrective RAG | Part 3: 프로덕션 파이프라인

RAG 기본이 처음이라면 Temporal RAG와 Multi-hop RAG 시리즈를 먼저 읽어보세요. Agent 패턴이 처음이라면 AI Agent 첫걸음부터 시작하세요.

Naive RAG의 한계

대부분의 RAG 튜토리얼은 이런 구조입니다: 질문 → 벡터 검색 → LLM 생성. 간단하고 잘 동작합니다 — 질문이 단순할 때만요.

현실에서 마주치는 질문은 다양합니다. 실시간 뉴스, SQL 집계, 일반 상식, 여러 소스의 조합 — 그런데 Naive RAG는 이 모든 질문을 똑같이 처리합니다. 벡터 DB에 던지고, 나온 결과를 LLM에 넘깁니다.

다음은 전형적인 Naive RAG 코드입니다.

def naive_rag(query: str) -> str:
    """단순 RAG: 항상 벡터 검색만 수행합니다."""
    docs = vector_store.similarity_search(query, k=4)
    context = "\n".join(d.page_content for d in docs)
    return llm.invoke(f"Context:\n{context}\n\nQ: {query}")

# 실패 케이스: "최근 OpenAI 매출 추이가 어떻게 돼?"
# → 벡터 DB에 실시간 데이터가 없어서 엉뚱한 답변 생성

핵심 인사이트: Naive RAG의 근본적 한계는 "모든 질문이 벡터 검색으로 해결된다"는 가정입니다. Agentic RAG는 이 가정을 깨고, 질문마다 최적의 전략을 선택합니다.

Query Analysis — 의도 분류

Agentic RAG의 첫 번째 단계는 Query Analysis입니다. 들어온 질문이 어떤 유형인지, 얼마나 복잡한지, 어떤 소스가 필요한지를 LLM이 먼저 판단합니다.

이때 Structured Output을 활용하면 분류 결과를 프로그래밍적으로 다룰 수 있습니다. Pydantic 모델로 출력 스키마를 정의하고, OpenAI의 response_format으로 강제합니다.

from pydantic import BaseModel, Field
from typing import Literal
from openai import OpenAI

client = OpenAI()

class QueryAnalysis(BaseModel):
    """사용자 질문을 분석한 결과 스키마입니다."""
    intent: Literal["factual", "analytical", "comparison", "temporal", "opinion"]
    complexity: Literal["simple", "multi_hop", "aggregation"]
    requires_retrieval: bool
    suggested_sources: list[Literal["vector_db", "web_search", "sql_db"]]
    sub_queries: list[str] = Field(default_factory=list)

SYSTEM_PROMPT = """당신은 사용자 질문을 분석하는 전문가입니다.
질문의 의도(intent), 복잡도(complexity), 검색 필요 여부(requires_retrieval),
적절한 소스(suggested_sources), 하위 질문(sub_queries)을 판단하세요.

규칙:
- 실시간 정보가 필요하면 web_search를 추천하세요.
- 수치/통계 질문은 sql_db를 추천하세요.
- 일반 상식이나 개념 설명은 requires_retrieval=false로 설정하세요.
- 복합 질문은 sub_queries로 분해하세요."""

def analyze_query(query: str) -> QueryAnalysis:
    """사용자 질문의 의도와 복잡도를 분석합니다."""
    response = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": query}
        ],
        response_format=QueryAnalysis
    )
    return response.choices[0].message.parsed

실제로 몇 가지 질문을 분석해보면 이렇게 됩니다.

# 예시 1: 단순 사실 확인
result = analyze_query("트랜스포머 모델의 어텐션 메커니즘이 뭐야?")
# → intent="factual", complexity="simple",
#   requires_retrieval=True, suggested_sources=["vector_db"]

# 예시 2: 실시간 정보 필요
result = analyze_query("최근 OpenAI 매출 추이가 어떻게 돼?")
# → intent="temporal", complexity="aggregation",
#   requires_retrieval=True, suggested_sources=["web_search"]

# 예시 3: 검색 불필요
result = analyze_query("파이썬에서 리스트와 튜플의 차이는?")
# → intent="comparison", complexity="simple",
#   requires_retrieval=False, suggested_sources=[]

질문 유형별로 어떤 소스가 최적인지 정리하면 다음과 같습니다.

Query Routing — 최적 소스로 라우팅

질문을 분석했으면 이제 실제로 적절한 소스에서 정보를 가져와야 합니다. Query Router는 분석 결과를 기반으로 검색 백엔드를 선택하고 실행합니다.

먼저 세 가지 검색 백엔드를 준비합니다.

import chromadb
from tavily import TavilyClient
import sqlite3

# 1. 벡터 검색: ChromaDB
chroma_client = chromadb.PersistentClient(path="./chroma_db")
collection = chroma_client.get_collection("documents")

def vector_search(query: str, k: int = 4) -> list[str]:
    """벡터 DB에서 유사 문서를 검색합니다."""
    results = collection.query(query_texts=[query], n_results=k)
    return results["documents"][0]

# 2. 웹 검색: Tavily
tavily_client = TavilyClient(api_key="tvly-...")

def web_search(query: str) -> list[str]:
    """실시간 웹 검색을 수행합니다."""
    response = tavily_client.search(query, max_results=3)
    return [r["content"] for r in response["results"]]

# 3. Text-to-SQL: SQLite
conn = sqlite3.connect("./company.db")

def sql_query(query: str) -> str:
    """자연어를 SQL로 변환하여 실행합니다."""
    # LLM으로 자연어 → SQL 변환
    sql = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": (
                "다음 스키마를 참고하여 SQL을 생성하세요.\n"
                "테이블: sales(date, product, revenue, region)\n"
                "SQL만 출력하세요. 설명 없이."
            )},
            {"role": "user", "content": query}
        ]
    ).choices[0].message.content.strip()

    # SQL 실행
    cursor = conn.execute(sql)
    rows = cursor.fetchall()
    columns = [desc[0] for desc in cursor.description]
    return f"SQL: {sql}\n결과: {[dict(zip(columns, row)) for row in rows]}"

이제 라우터 함수를 만듭니다. QueryAnalysis의 suggested_sources를 순회하면서 해당 백엔드를 호출합니다.

def route_query(analysis: QueryAnalysis, query: str) -> list[str]:
    """분석 결과에 따라 적절한 소스에서 정보를 검색합니다."""
    results = []

    for source in analysis.suggested_sources:
        if source == "vector_db":
            results.extend(vector_search(query))
        elif source == "web_search":
            results.extend(web_search(query))
        elif source == "sql_db":
            results.append(sql_query(query))

    return results

단순해 보이지만, 이 라우팅만으로도 Naive RAG 대비 큰 성능 향상을 얻습니다. 핵심은 "올바른 소스에 질문을 보내는 것"입니다. 벡터 DB에 실시간 뉴스를 물어보는 것은 도서관에서 오늘 주가를 물어보는 것과 같습니다.

핵심 인사이트: Query Routing의 본질은 "도구 선택"입니다. 검색 소스를 도구로 보면, 이것은 Agent가 어떤 Tool을 쓸지 결정하는 것과 동일한 패턴입니다.

Adaptive Retrieval — 검색이 필요 없을 때

모든 질문에 검색을 수행하는 것은 비효율적입니다. "HTTP와 HTTPS 차이가 뭐야?" 같은 질문은 LLM이 충분히 잘 알고 있습니다. 검색을 하면 오히려 불필요한 문맥이 들어가 답변 품질이 떨어질 수 있습니다.

Adaptive Retrieval은 검색 여부 자체를 LLM이 판단하게 합니다. QueryAnalysis의 requires_retrieval 필드가 이 역할을 합니다.

def adaptive_rag(query: str) -> str:
    """검색 필요 여부를 판단하고, 필요한 경우에만 검색합니다."""
    analysis = analyze_query(query)

    # 검색이 필요 없으면 LLM 지식으로 직접 답변
    if not analysis.requires_retrieval:
        return client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "당신의 지식으로 정확하게 답변하세요."},
                {"role": "user", "content": query}
            ]
        ).choices[0].message.content

    # 복합 질문이면 하위 질문으로 분해해서 각각 검색
    queries = analysis.sub_queries if analysis.sub_queries else [query]
    all_context = []
    for q in queries:
        sub_analysis = analyze_query(q) if q != query else analysis
        all_context.extend(route_query(sub_analysis, q))

    # 수집한 문맥을 기반으로 최종 답변 생성
    context_text = "\n---\n".join(all_context)
    return client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "주어진 문맥을 기반으로 정확하게 답변하세요."},
            {"role": "user", "content": f"문맥:\n{context_text}\n\n질문: {query}"}
        ]
    ).choices[0].message.content

이건 Agent Part 1의 ReAct 루프와 같은 원리입니다. Agent가 "어떤 행동을 취할지" 결정하는 것을 검색에 적용한 겁니다. 검색할지 말지, 어디서 검색할지, 질문을 쪼갤지 말지 — 이 모든 결정을 LLM이 내립니다.

LangGraph로 구현하기

지금까지의 로직을 LangGraph로 구조화하면 더 깔끔하고 확장 가능한 파이프라인이 됩니다. 각 단계를 노드로, 분기 조건을 엣지로 표현합니다.

LangGraph가 처음이라면 Agent Part 2: LangGraph 실전을 참고하세요.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END

class AgenticRAGState(TypedDict):
    """Agentic RAG 파이프라인의 상태를 정의합니다."""
    query: str                              # 사용자 원본 질문
    analysis: QueryAnalysis | None          # 질문 분석 결과
    documents: list[str]                    # 검색된 문서들
    generation: str                         # 최종 생성된 답변


def analyze_node(state: AgenticRAGState) -> dict:
    """질문을 분석하여 의도와 최적 소스를 결정합니다."""
    analysis = analyze_query(state["query"])
    return {"analysis": analysis}


def should_retrieve(state: AgenticRAGState) -> str:
    """검색이 필요한지 판단하여 다음 노드를 결정합니다."""
    if state["analysis"].requires_retrieval:
        return "retrieve"
    return "generate_direct"


def retrieve_node(state: AgenticRAGState) -> dict:
    """분석 결과에 따라 적절한 소스에서 문서를 검색합니다."""
    analysis = state["analysis"]
    query = state["query"]

    all_docs = []
    # 하위 질문이 있으면 각각 검색
    queries = analysis.sub_queries if analysis.sub_queries else [query]
    for q in queries:
        sub_analysis = analyze_query(q) if q != query else analysis
        all_docs.extend(route_query(sub_analysis, q))

    return {"documents": all_docs}


def generate_node(state: AgenticRAGState) -> dict:
    """검색된 문서를 기반으로 답변을 생성합니다."""
    context = "\n---\n".join(state["documents"])
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": (
                "주어진 문맥을 기반으로 정확하게 답변하세요. "
                "문맥에 없는 내용은 추측하지 마세요."
            )},
            {"role": "user", "content": f"문맥:\n{context}\n\n질문: {state['query']}"}
        ]
    ).choices[0].message.content
    return {"generation": response}


def generate_direct_node(state: AgenticRAGState) -> dict:
    """검색 없이 LLM 지식만으로 답변을 생성합니다."""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "당신의 지식으로 정확하고 친절하게 답변하세요."},
            {"role": "user", "content": state["query"]}
        ]
    ).choices[0].message.content
    return {"generation": response}


# 그래프 구성
graph = StateGraph(AgenticRAGState)

# 노드 추가
graph.add_node("analyze", analyze_node)
graph.add_node("retrieve", retrieve_node)
graph.add_node("generate", generate_node)
graph.add_node("generate_direct", generate_direct_node)

# 엣지 연결
graph.set_entry_point("analyze")
graph.add_conditional_edges(
    "analyze",
    should_retrieve,
    {
        "retrieve": "retrieve",           # 검색 필요 → 검색 노드로
        "generate_direct": "generate_direct"  # 검색 불필요 → 직접 답변
    }
)
graph.add_edge("retrieve", "generate")    # 검색 후 → 답변 생성
graph.add_edge("generate", END)           # 답변 완료
graph.add_edge("generate_direct", END)    # 직접 답변 완료

# 컴파일 및 실행
app = graph.compile()

# 실행 예시
result = app.invoke({
    "query": "작년 대비 올해 서울 날씨 변화를 분석해줘",
    "analysis": None,
    "documents": [],
    "generation": ""
})
print(result["generation"])

그래프 흐름은 사용자 질문 → [analyze] → 검색 필요? → Yes: [retrieve] → [generate] → END / No: [generate_direct] → END 입니다.

LangGraph의 장점은 각 노드를 독립적으로 테스트하고 교체할 수 있다는 것입니다. retrieve_node에 새로운 소스(GraphRAG, API 호출 등)를 추가하거나, should_retrieve 조건을 바꾸는 것이 간단합니다.

라우팅이 실제로 효과가 있나?

이론은 그럴듯한데, 실제로 얼마나 차이가 날까요? 4가지 유형의 질문 100개씩, 총 400개 질문으로 Naive RAG와 Agentic RAG(Query Routing)를 비교했습니다. 평가 지표는 정확도(GPT-4o judge 기반)입니다.

몇 가지 관찰 사항입니다.

• in-corpus 질문에서는 차이가 미미합니다. 벡터 DB에 답이 있는 질문은 Naive RAG도 잘 합니다.
• 실시간 정보에서 극적인 차이가 납니다. Naive RAG는 벡터 DB에 해당 정보가 없으니 환각을 생성하지만, Agentic RAG는 웹 검색으로 정확한 정보를 가져옵니다.
• 구조화 데이터는 아예 게임이 다릅니다. "지난 분기 매출 상위 5개 제품"을 벡터 검색으로 답할 수는 없습니다.
• 검색 불필요 질문에서도 의미 있는 개선이 있습니다. 불필요한 문맥을 제거하면 LLM이 더 정확하게 답변합니다.

평가 방법론은 RAG Evaluation에서 자세히 다뤘습니다.

핵심 인사이트: Query Routing의 가장 큰 가치는 "기존 RAG가 전혀 답할 수 없던 영역"을 커버하는 것입니다. in-corpus 질문 성능은 비슷하지만, 실시간 데이터와 구조화 데이터에서 판을 바꿉니다.

다음 편 예고

라우팅으로 올바른 소스를 찾았지만, 가져온 문서의 품질이 낮으면 어떻게 될까요? 검색 결과가 질문과 관련 없거나, 오래된 정보가 섞여 있다면?

Part 2에서는 이 문제를 해결하는 두 가지 기법을 다룹니다.

• Self-RAG: LLM이 스스로 검색 결과의 관련성을 평가하고, 필요하면 다시 검색합니다
• Corrective RAG (CRAG): 검색 결과가 불충분할 때 자동으로 웹 검색으로 폴백합니다

질문 분석 → 라우팅(Part 1) → 품질 검증(Part 2) → 프로덕션 배포(Part 3)의 완전한 Agentic RAG 파이프라인을 완성하겠습니다.

참고 자료

• Gao, Y., et al. (2024). "Retrieval-Augmented Generation for Large Language Models: A Survey." *arXiv:2312.10997*.
• Asai, A., et al. (2023). "Self-RAG: Learning to Retrieve, Generate, and Critique through Self-Reflection." *arXiv:2310.11511*.
• Yan, S., et al. (2024). "Corrective Retrieval Augmented Generation." *arXiv:2401.15884*.
• LangGraph 공식 문서
• Tavily API 문서
• 관련 시리즈: Temporal RAG · Multi-hop RAG · RAG Evaluation · AI Agent 시리즈