에이전트가 “스스로 검색하고, 검증하고, 다시 검색하는” Agentic RAG 구현 가이드 (2026년 4월 기준)

게시 2026/04/05

By Daewook Kwon

12 분읽는 시간

들어가며

전통적 RAG는 retrieve → generate 1회성 파이프라인에 가깝습니다. 그래서 (1) 질문이 애매하거나, (2) 검색 결과가 부정확/불충분하거나, (3) 최신 정보가 섞여 있어 교차검증이 필요한 경우 성능이 급격히 흔들립니다.
2026년 들어 업계 흐름은 “RAG를 도구(tool)로 두고, LLM이 플래닝/루프/반성(reflection)을 돌며 retrieval 자체를 제어”하는 Agentic RAG 쪽으로 명확히 이동했습니다. 특히 LangGraph의 Corrective RAG(CRAG)처럼 retrieval 결과를 LLM이 채점하고 부족하면 web search로 보강하는 패턴이 빠르게 표준화되는 분위기입니다. (langchain-ai.github.io)

또 하나의 큰 변화는 “그래프(DAG)보다 유연한 event-driven workflow”가 실전에서 강하다는 점입니다. LlamaIndex Workflows는 step/event/loop/state를 언어 레벨에서 잡아주어, 자율 에이전트의 반복 구조를 ‘코드로 통제’하기 쉽습니다. (docs.llamaindex.ai)

🔧 핵심 개념

1) Agentic RAG 정의

Agentic RAG는 RAG를 “한 번의 호출”이 아니라, 에이전트의 의사결정 루프 안에 넣는 아키텍처입니다. 핵심은 다음 4가지입니다.

Planner/Router: 질문을 보고 “벡터 검색 vs 웹 검색 vs 재질문(query rewrite)”을 선택
Retriever(들): Vector DB, BM25, SQL, Web search 등 복수 가능
Critic/Grader: 가져온 근거가 충분/정확한지 채점(자기검증)
Controller(Workflow/Graph): 루프, 분기, 타임아웃, 상태(state) 관리

LlamaIndex도 “routing, query transformation 같은 모듈 자체가 agentic”이며, 더 나아가 “RAG query engine을 tool로 올려 에이전트가 플래닝/반복 실행”하는 구성을 명시적으로 권장합니다. (docs.llamaindex.ai)

2) CRAG(Corrective RAG)의 실전적 의미

LangGraph 문서의 CRAG는 Agentic RAG의 ‘최소 기능’에 가깝습니다. 동작 요지는 단순합니다. (langchain-ai.github.io)

Top-k 문서를 가져온다
LLM이 문서 relevance를 채점한다
전부 낮으면 web search로 보강한다
최종 컨텍스트로 답변 생성

여기서 중요한 건 “web search”가 아니라, retrieval 품질을 에이전트가 판정한다는 점입니다. 이 한 단계가 들어가면, 데이터 품질/커버리지 문제를 ‘운영 로직’으로 흡수할 수 있습니다.

3) Workflow 기반 구현이 유리한 이유

자율 에이전트는 결국 “while-loop + 상태 + 안전장치”입니다. LlamaIndex Workflows는 이를 Event → Step → Event로 모델링하고, 루프/브랜치/상태(Context) 를 공식 API로 제공합니다. (huggingface.co)
또한 여러 에이전트를 엮는 AgentWorkflow(handoff 포함)도 제공해, “검색 담당/평가 담당/작성 담당”을 분리하기 쉽습니다. (huggingface.co)

💻 실전 코드

아래 예제는 “Vector RAG → 문서 채점 → 부족하면 Web search 보강 → 최종 답변”을 LlamaIndex Workflows 스타일의 event-driven loop로 구현한 최소 실전형 뼈대입니다. (Vector DB는 Chroma, Web search는 Tavily 예시로 둡니다. 실제 운영에서는 사내 검색/Elastic/OpenSearch로 교체하면 됩니다.)

  
# Python 3.11+
# pip install llama-index llama-index-workflows chromadb tavily-python
# (embedding/llm provider는 환경에 맞게 설정)

from __future__ import annotations

from dataclasses import dataclass
from typing import List, Optional

from llama_index.core.workflow import (
    Workflow, step, StartEvent, StopEvent, Event, Context
)

# --- Events ---
class QueryEvent(Event):
    query: str

class RetrievedEvent(Event):
    query: str
    docs: List[str]          # 간단화를 위해 text 리스트로 표현
    source: str              # "vector" | "web"

class NeedMoreEvent(Event):
    query: str
    reason: str

class FinalContextEvent(Event):
    query: str
    context: str


# --- 외부 의존: LLM / retrieval / web search 는 프로젝트별로 교체 ---
def vector_retrieve(query: str, k: int = 5) -> List[str]:
    """
    TODO: Chroma/FAISS/Elastic/OpenSearch 등으로 교체.
    여기서는 예시로 더미를 반환.
    """
    return [
        f"[VEC] ({i}) {query} 관련 내부 문서 스니펫..."
        for i in range(1, k + 1)
    ]

def web_search(query: str, k: int = 5) -> List[str]:
    """
    TODO: Tavily/SerpAPI/사내 검색 API 등으로 교체.
    여기서는 예시로 더미를 반환.
    """
    return [
        f"[WEB] ({i}) {query} 관련 웹 검색 스니펫..."
        for i in range(1, k + 1)
    ]

def grade_relevance(query: str, docs: List[str]) -> float:
    """
    TODO: 실제로는 LLM grader를 붙여 점수화.
    LangGraph CRAG처럼 'relevance threshold'로 분기하는 아이디어. ([langchain-ai.github.io](https://langchain-ai.github.io/langgraph/tutorials/rag/langgraph_crag_local/?utm_source=openai))
    여기서는 간단히 길이 기반 더미 점수.
    """
    joined = " ".join(docs)
    return min(1.0, len(joined) / 2000)

def generate_answer(query: str, context: str) -> str:
    """
    TODO: 실제 LLM 호출로 교체. (OpenAI/로컬 LLM 등)
    """
    return f"Q: {query}\n\n[컨텍스트 요약]\n{context[:400]}...\n\n[최종 답변]\n(여기에 LLM 생성 결과)"


# --- Workflow ---
class AgenticRAGWorkflow(Workflow):
    """
    event-driven step 구성은 LlamaIndex Workflows의 핵심. ([docs.llamaindex.ai](https://docs.llamaindex.ai/en/stable/module_guides/workflow/?utm_source=openai))
    Context를 이용해 상태를 공유/누적할 수 있음. ([huggingface.co](https://huggingface.co/learn/agents-course/unit2/llama-index/workflows?utm_source=openai))
    """

    @step
    async def start(self, ctx: Context, ev: StartEvent) -> QueryEvent:
        query = ev.get("query")
        await ctx.store.set("query", query)
        await ctx.store.set("tries", 0)
        return QueryEvent(query=query)

    @step
    async def retrieve_vector(self, ctx: Context, ev: QueryEvent) -> RetrievedEvent:
        docs = vector_retrieve(ev.query, k=5)
        return RetrievedEvent(query=ev.query, docs=docs, source="vector")

    @step
    async def critique(self, ctx: Context, ev: RetrievedEvent) -> NeedMoreEvent | FinalContextEvent:
        # 루프 횟수 제한(안전장치)
        tries = await ctx.store.get("tries")
        tries += 1
        await ctx.store.set("tries", tries)

        score = grade_relevance(ev.query, ev.docs)
        await ctx.store.set("last_score", score)

        threshold = 0.55
        if score >= threshold or tries >= 2:
            # 충분하거나(또는) 더 돌리면 비용만 커질 때 종료 컨텍스트 확정
            context = "\n".join(ev.docs)
            return FinalContextEvent(query=ev.query, context=context)

        # 부족하면 web 보강으로 분기 (CRAG의 핵심 분기) ([langchain-ai.github.io](https://langchain-ai.github.io/langgraph/tutorials/rag/langgraph_crag_local/?utm_source=openai))
        return NeedMoreEvent(query=ev.query, reason=f"low_relevance(score={score:.2f})")

    @step
    async def retrieve_web(self, ctx: Context, ev: NeedMoreEvent) -> RetrievedEvent:
        docs = web_search(ev.query, k=5)
        return RetrievedEvent(query=ev.query, docs=docs, source="web")

    @step
    async def answer(self, ctx: Context, ev: FinalContextEvent) -> StopEvent:
        # vector + web 결과를 합치는 전략도 흔함(여기선 간단화)
        answer = generate_answer(ev.query, ev.context)
        meta = {
            "tries": await ctx.store.get("tries"),
            "last_score": await ctx.store.get("last_score"),
        }
        return StopEvent(result={"answer": answer, "meta": meta})


# --- 실행 예시 ---
# import asyncio
# async def main():
#     w = AgenticRAGWorkflow(timeout=30, verbose=False)
#     out = await w.run(query="2026년 Agentic RAG 구현에서 CRAG 패턴을 적용하는 방법은?")
#     print(out["answer"])
#     print(out["meta"])
# asyncio.run(main())

⚡ 실전 팁

1) “에이전트 자율성”은 루프/상태/가드레일이 전부다
Agentic RAG는 멋있게 보이지만, 운영에서 중요한 건 “언제 멈추는가”입니다. max_tries, timeout, tool budget(검색 횟수/토큰)을 워크플로우 레벨에서 강제하세요. Workflows가 step 기반이라 이런 가드레일이 코드로 명확해집니다. (docs.llamaindex.ai)

2) retrieval grader는 ‘정답 판별’이 아니라 ‘다음 행동 결정’용
CRAG의 grader는 “정답 맞추기”가 아니라 “현재 근거로 답해도 되는지”를 판단합니다. 이 관점으로 프롬프트/스코어링을 설계하면 과도한 self-critique를 줄일 수 있습니다. (langchain-ai.github.io)

3) 멀티 에이전트는 ‘역할 분리’가 아니라 ‘실패 격리’에 가치가 있다
researcher(검색), critic(검증), writer(서술)로 나누면 좋다는 얘기는 많지만, 진짜 이점은 장애/환각을 격리하고 trace를 구조화하는 데 있습니다. LlamaIndex의 AgentWorkflow는 handoff를 기본 모델로 둡니다. (huggingface.co)

4) 관측가능성(Observability)을 먼저 붙이고 고도화
Workflows는 step 단위로 자동 계측을 제공하고, Phoenix 같은 도구로 관찰 가능하다고 명시합니다. “왜 이 에이전트가 web search로 갔지?” 같은 질문에 답할 수 있어야 운영이 됩니다. (docs.llamaindex.ai)

5) prompt injection은 ‘검색 단계’에서 터진다
OpenAI Agent Builder 문서도 워크플로우 구축 시 prompt injection / data leakage 위험을 직접 경고합니다. 웹/외부 문서를 넣는 순간부터는 “문서 내용은 지시가 아니라 데이터”라는 정책을 시스템적으로 강제(예: 컨텍스트 래핑, 출처 태깅, tool-output 분리)해야 합니다. (platform.openai.com)

🚀 마무리

Agentic RAG의 본질은 “RAG를 잘하는 법”이 아니라 retrieval을 스스로 통제하는 에이전트 루프를 설계하는 법입니다. 2026년 4월 기준으로 실전 구현의 중심은 (1) CRAG류의 self-grading 분기, (langchain-ai.github.io) (2) event-driven workflow로 루프/상태/가드레일을 코드로 고정하는 방식(LlamaIndex Workflows 등) (docs.llamaindex.ai) 으로 정리됩니다.

다음 학습 추천:

LangGraph의 CRAG 튜토리얼을 그대로 따라 하며 “grader → web fallback” 분기를 체화 (langchain-ai.github.io)
LlamaIndex Workflows에서 loop/branch/state 패턴을 익힌 뒤, AgentWorkflow로 역할 분리까지 확장 (huggingface.co)
마지막으로 OpenAI Agents SDK/Agent Builder의 보안 경고를 기준으로, tool-output 경계와 정책을 설계 (platform.openai.com)

AI, Agent

ai agent trend 2026-04