Agentic RAG 자율 에이전트, “검색을 스스로 설계·검증하는” 구현 패턴 (2026년 3월판)

AI Agent

#ai #agent #trend #2026-03

들어가며

2024년식 RAG는 “질문 → 검색 → 답변” 파이프라인이 고정돼 있어, 애매한 질의(용어 불명확/범위 과대), 문서 품질 편차, 상충 근거가 섞인 상황에서 쉽게 무너집니다. 반면 2026년 3월 기준 현업에서 말하는 Agentic RAG는 LLM이 검색 자체를 도구(tool)로 다루며, 언제·무엇을·몇 번 검색할지, 결과를 어떻게 재질의/검증할지를 스스로 결정하는 형태로 진화했습니다. LangGraph 문서의 “retriever tool을 필요할 때만 호출하는 retrieval agent” 튜토리얼이 이 흐름을 가장 직설적으로 보여줍니다. (docs.langchain.com)

또한 최근 실전 사례에서는 “self-correcting RAG”처럼 실패를 전제로 디버깅/재시도를 설계하고(OpenSearch 예시), RAG를 단순 QA가 아니라 탐색(Research) 워크플로우로 다루는 관점이 강해졌습니다. (bigdataboutique.com)

---

🔧 핵심 개념

주요 개념 정의

• Agentic RAG: RAG 컴포넌트(검색/리랭크/요약/검증)를 LLM의 행동(action)으로 올려, 질의 계획(Query Planning) → 검색 → 관찰(Observation) → 재질의/확장 → 최종 합성을 반복하는 방식. (docs.langchain.com)
• Planner / Controller: “이번 턴에 검색이 필요한가?”, “검색 쿼리를 어떻게 쪼갤까?”, “근거가 충분한가?”를 판단하는 상위 정책. 최근 연구들은 전략(고수준)과 실행(저수준) 간 간극을 줄이려는 프레임을 제안하기도 합니다. (arxiv.org)
• Tool-based Retrieval: retriever를 함수 호출처럼 노출해, 에이전트가 필요할 때만 호출. LangGraph가 제공하는 대표 구현 패턴입니다. (docs.langchain.com)
• Self-correction loop: (a) 결과 부족/모순 감지 → (b) 쿼리 재작성/확장 → (c) 다른 검색 전략(hybrid, 필터) 시도 → (d) 답변 재합성의 루프. 실전 튜토리얼에서도 “실패를 디버깅하며 고친다”가 핵심 메시지입니다. (bigdataboutique.com)

어떻게 작동하는지(권장 상태 머신)

Agentic RAG를 “그래프/상태 머신”으로 모델링하면 구현이 단단해집니다.

1) Intake: 사용자 질문을 구조화(의도/범위/제약)
2) Plan: 검색이 필요한지 결정 + 서브쿼리 생성
3) Retrieve: 서브쿼리별 검색(벡터/키워드/hybrid)
4) Critique: 근거 커버리지/신뢰도/충돌 여부 평가
5) Refine: 쿼리 재작성, 범위 축소/확대, 추가 검색
6) Synthesize: 인용 가능한 근거 기반으로 답변 구성
7) Stop: (i) 근거 충분, (ii) 예산 초과, (iii) 더 검색해도 개선 미미

LangGraph는 이 “결정 노드 + tool 호출 + 조건부 라우팅” 구조를 그대로 코드로 옮기기 좋습니다. (docs.langchain.com)

---

💻 실전 코드

아래는 LangGraph 기반 Agentic RAG 최소 구현입니다. 핵심은 grade() 노드가 추가 검색 여부를 판정하고, 필요하면 rewrite_query()로 쿼리를 개선해 다시 retrieve()로 루프를 태우는 점입니다. (벡터 DB는 Chroma를 사용)

# python
# pip install -U langgraph langchain langchain-community langchain-text-splitters chromadb
# export OPENAI_API_KEY=...

from __future__ import annotations
from dataclasses import dataclass, field
from typing import List, Literal, Optional, Dict, Any

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma

# ---------------------------
# 1) 인덱스 준비(예시: 로컬 문서)
# ---------------------------
def build_vectorstore(texts: List[str]) -> Chroma:
    splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=120)
    docs = []
    for i, t in enumerate(texts):
        for chunk in splitter.split_text(t):
            docs.append(Document(page_content=chunk, metadata={"source": f"doc{i}"}))

    vs = Chroma.from_documents(
        docs,
        embedding=OpenAIEmbeddings(),
        collection_name="agentic_rag_demo",
    )
    return vs

# ---------------------------
# 2) 그래프 상태 정의
# ---------------------------
@dataclass
class RAGState:
    question: str
    query: str = ""
    docs: List[Document] = field(default_factory=list)
    answer: str = ""
    # 루프 제어
    iterations: int = 0
    max_iterations: int = 3
    need_more: bool = False
    critique: str = ""

# ---------------------------
# 3) 노드 구현
# ---------------------------
llm = ChatOpenAI(model="gpt-4.1-mini", temperature=0.1)

PLAN_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "You are a senior search planner. Decide the best search query for the user question."),
    ("user", "Question: {question}\nReturn a concise search query only.")
])

GRADE_PROMPT = ChatPromptTemplate.from_messages([
    ("system",
     "You are a strict evaluator. Decide if retrieved evidence is sufficient and non-contradictory.\n"
     "Return JSON with keys: need_more (true/false), critique (string)."),
    ("user",
     "Question: {question}\n\nRetrieved snippets:\n{snippets}\n\nEvaluate evidence sufficiency.")
])

REWRITE_PROMPT = ChatPromptTemplate.from_messages([
    ("system",
     "You rewrite queries to improve retrieval recall/precision.\n"
     "Use critique to refine query. Return only the rewritten query."),
    ("user",
     "Original question: {question}\nCurrent query: {query}\nCritique: {critique}\nRewrite:")
])

ANSWER_PROMPT = ChatPromptTemplate.from_messages([
    ("system",
     "Answer ONLY from the provided evidence. If evidence is insufficient, say so and list what to search next."),
    ("user",
     "Question: {question}\n\nEvidence:\n{snippets}\n\nWrite a grounded answer.")
])

def plan(state: RAGState) -> Dict[str, Any]:
    q = llm.invoke(PLAN_PROMPT.format_messages(question=state.question)).content.strip()
    return {"query": q}

def retrieve_factory(vs: Chroma):
    retriever = vs.as_retriever(search_kwargs={"k": 6})
    def retrieve(state: RAGState) -> Dict[str, Any]:
        docs = retriever.invoke(state.query or state.question)
        return {"docs": docs}
    return retrieve

def grade(state: RAGState) -> Dict[str, Any]:
    snippets = "\n\n".join([f"[{d.metadata.get('source')}] {d.page_content}" for d in state.docs[:6]])
    raw = llm.invoke(GRADE_PROMPT.format_messages(
        question=state.question,
        snippets=snippets
    )).content

    # 간단 파서(실무에선 pydantic/structured output 권장)
    need_more = "true" in raw.lower() and "need_more" in raw.lower()
    return {"need_more": need_more, "critique": raw, "iterations": state.iterations + 1}

def rewrite_query(state: RAGState) -> Dict[str, Any]:
    new_q = llm.invoke(REWRITE_PROMPT.format_messages(
        question=state.question,
        query=state.query,
        critique=state.critique
    )).content.strip()
    return {"query": new_q}

def answer(state: RAGState) -> Dict[str, Any]:
    snippets = "\n\n".join([f"[{d.metadata.get('source')}] {d.page_content}" for d in state.docs[:10]])
    a = llm.invoke(ANSWER_PROMPT.format_messages(
        question=state.question,
        snippets=snippets
    )).content.strip()
    return {"answer": a}

def should_continue(state: RAGState) -> Literal["rewrite", "answer", "__end__"]:
    if state.iterations >= state.max_iterations:
        return "answer"
    return "rewrite" if state.need_more else "answer"

# ---------------------------
# 4) 그래프 조립
# ---------------------------
def build_agentic_rag_app(vs: Chroma):
    g = StateGraph(RAGState)
    g.add_node("plan", plan)
    g.add_node("retrieve", retrieve_factory(vs))
    g.add_node("grade", grade)
    g.add_node("rewrite", rewrite_query)
    g.add_node("answer", answer)

    g.set_entry_point("plan")
    g.add_edge("plan", "retrieve")
    g.add_edge("retrieve", "grade")
    g.add_conditional_edges("grade", should_continue, {
        "rewrite": "rewrite",
        "answer": "answer",
    })
    g.add_edge("rewrite", "retrieve")
    g.add_edge("answer", END)

    return g.compile()

if __name__ == "__main__":
    # 데모용 문서(실무에선 사내 위키/DB/로그 등으로 대체)
    texts = [
        "Agentic RAG is a pattern where an agent decides when to retrieve, critiques evidence, and iterates.",
        "LangGraph models agent workflows as graphs with conditional routing between nodes and tool calls.",
        "Self-correcting RAG adds evaluation and query rewriting loops to handle ambiguous queries and missing evidence."
    ]
    vs = build_vectorstore(texts)
    app = build_agentic_rag_app(vs)

    out = app.invoke(RAGState(question="Agentic RAG를 자율 에이전트로 구현할 때 핵심 루프는?"))
    print(out["answer"])

이 코드는 LangGraph가 제안하는 “retriever tool을 에이전트가 필요할 때만 호출”하는 방향과 잘 맞고, 특히 grade→rewrite→retrieve 루프가 “agentic”함의 최소 단위입니다. (docs.langchain.com)

---

⚡ 실전 팁

• 쿼리 재작성은 ‘정답 찾기’가 아니라 ‘검색 공간 제어’다: critique에서 결측(coverage)인지 노이즈(precision)인지 먼저 분류하고, 결측이면 확장(동의어/상위개념), 노이즈면 축소(필터/키워드 고정)를 적용하세요. (RAG Blueprint에서도 결국 “retrieval agent가 무엇을/왜 찾는지”가 성능을 좌우한다고 정리합니다.) (langwatch.ai)
• Stop 조건을 제품 요구사항으로 명시: max_iterations는 임의 숫자가 아니라 (a) latency SLO, (b) 토큰 예산, (c) 도메인 위험도(의료/법률)로 결정해야 합니다. 무한 루프보다 더 나쁜 건 “그럴듯한 환각을 루프로 강화”하는 것입니다.
• Self-correction의 핵심은 ‘평가 기준의 외부화’: grade 노드가 “충분함”을 판단할 때, 단순히 LLM 감상평이 아니라 체크리스트를 강제하세요(예: 근거 2개 이상, 상충 시 둘 다 제시, 최신성 필요 시 날짜 포함 등).
• 관측(Observation)을 로그로 남겨 디버깅 가능하게: LangGraph 계열 접근이 강한 이유 중 하나가 그래프 단위 트레이싱/재현이 쉽기 때문입니다(실무에선 노드별 입력/출력을 반드시 저장). (docs.langchain.com)
• “프레임워크 종속”을 최소화: 커뮤니티에서도 에이전트 프레임워크 과의존을 경계하고, 워크플로우에 맞춘 bespoke 구현이 낫다는 의견이 반복됩니다. 핵심은 그래프(상태)와 도구 계약(interface)을 명확히 두고, 나머지는 교체 가능하게 만드는 것. (reddit.com)

---

🚀 마무리

2026년 3월의 Agentic RAG 구현 포인트는 한 문장으로 “검색을 파이프라인이 아니라 에이전트의 행동으로 만들고, 실패를 루프로 흡수하라”입니다. LangGraph 스타일로 상태 머신을 만들고(Plan/Retrieve/Grade/Rewrite/Answer), self-correction을 설계하면 “한 번에 맞히는 RAG”보다 훨씬 안정적으로 복잡 질의에 대응할 수 있습니다. (docs.langchain.com)

다음 학습 추천:

• LangGraph의 agentic RAG 공식 튜토리얼을 그대로 따라 하며 conditional routing과 tool 호출 경계를 손에 익히기 (docs.langchain.com)
• LlamaIndex의 agentic strategies를 참고해 “기존 RAG를 tool로 승격”시키는 패턴 비교해보기 (docs.llamaindex.ai)
• 연구 관점에선 “planner–executor 간 격차”를 다루는 최신 agentic RAG 프레임을 훑으며, 평가/보상 설계를 어떻게 가져올지 고민해보기 (arxiv.org)