LLM이 “찾아보고(검색) → 판단하고(계획) → 답한다(생성)”까지 하는 2025 RAG Agent 구현 튜토리얼

들어가며

2025년의 RAG는 더 이상 “vector DB에서 top-k 뽑아 프롬프트에 붙이는” 수준에서 끝나지 않습니다. 제품/운영 환경에서는 (1) 질문이 애매하면 재질문하거나, (2) 내부 문서/외부 웹/DB/API 중 어디를 먼저 볼지 선택하고, (3) 찾은 근거가 부족하면 재검색·재랭크하고, (4) 최종 답변에 근거(citation)와 신뢰도까지 붙여야 합니다. 이 요구사항을 만족하는 패턴이 바로 Agentic RAG입니다. AWS도 “Agentic RAG는 multi-step, tool 사용, 적응형 흐름”으로 정의하며 단순 QA를 넘어선다고 명시합니다. (aws.amazon.com)

또한 OpenAI는 2025년에 Responses API + built-in tools(Web search / File search / MCP 등)로 “툴 실행을 인프라에서 자동 처리”하는 방향을 강화했고, 특히 File search는 query optimization·metadata filtering·reranking을 포함해 RAG 파이프라인의 번거로운 부분을 크게 줄였습니다. (developers.openai.com)
이 글에서는 “직접 구현(컨트롤 극대화)” 관점에서 RAG Agent의 내부 원리를 해부하고, 곧바로 실행 가능한 코드로 끝까지 연결합니다.

---

🔧 핵심 개념

1) RAG의 3계층: Indexing / Retrieval / Generation

• Indexing: 문서를 chunk로 쪼개고 embedding을 만들어 vector store에 저장
• Retrieval: 질문을 embedding → top-k 검색(+필요시 hybrid, rerank)
• Generation: 검색 결과를 “근거 컨텍스트”로 넣어 답변 생성(가능하면 citation 포함)

2025년 실전에서 중요한 건 Retrieval이 “단발”이 아니라는 점입니다. Agent는 검색 결과의 품질을 평가하고 부족하면 다음 액션을 합니다(재질문, k 조정, 다른 소스 조회 등). AWS가 말하는 “break down complex tasks / use external tools / adapt”가 여기서 나옵니다. (aws.amazon.com)

2) Agentic RAG의 제어 루프(의사결정)

Agentic RAG는 보통 아래 루프를 가집니다.

1. Router(의사결정): 이 질문은 “내부 문서 검색”인가? “웹 검색”인가? “DB 조회”인가?
2. Retriever(행동): 선택한 도구/리트리버 실행
3. Critic/Verifier(평가): 근거가 충분한가? 출처가 신뢰 가능한가? 상충하는가?
4. Synthesis(생성): 답변 + 근거 + 제한사항(모르면 모른다고)

OpenAI의 “built-in tools는 모델이 호출하면 자동 실행되고 결과가 대화 컨텍스트에 추가”된다는 설명은, 이 루프의 (2)를 매우 단순화합니다. (developers.openai.com)

3) 2025년 RAG에서 달라진 포인트: “검색 품질이 제품 품질”

• chunking/metadata/rerank가 성능을 좌우
• “top-k=3” 같은 고정값은 금방 한계
• 운영에서는 근거 노출(auditability)이 필수(특히 B2B/내부문서) OpenAI는 File search에 query optimization, metadata filtering, reranking을 포함한다고 밝히며 “강한 RAG를 추가 튜닝 없이” 구성할 수 있다고 설명합니다. (openai.com)

---

💻 실전 코드

아래 코드는 (A) 로컬 벡터 인덱스(FAISS) 기반 RAG + (B) 간단한 Agent 루프(검색→평가→재검색/답변) 를 한 파일에 구현한 예제입니다.

• 문서: 로컬 텍스트 파일들을 읽어 chunk → embedding → FAISS 저장
• 질의: 1차 검색 후 “근거 부족”이면 k를 늘려 재검색
• 답변: “근거 외 추측 금지” 프롬프트 + citation 형태로 source id를 같이 출력

# rag_agent.py
# 실행: pip install openai faiss-cpu tiktoken numpy
# 환경변수: OPENAI_API_KEY 설정
from __future__ import annotations

import os
import glob
import numpy as np
import faiss
import tiktoken
from dataclasses import dataclass
from typing import List, Tuple, Dict

from openai import OpenAI

client = OpenAI()

EMBED_MODEL = "text-embedding-3-small"
GEN_MODEL = "gpt-4o-mini"

@dataclass
class Chunk:
    chunk_id: str
    text: str
    source: str

def chunk_text(text: str, source: str, chunk_tokens: int = 350, overlap: int = 50) -> List[Chunk]:
    """
    토큰 기반 chunking.
    - chunk_tokens/overlap은 도메인에 맞게 튜닝 포인트
    """
    enc = tiktoken.get_encoding("cl100k_base")
    toks = enc.encode(text)
    out = []
    step = max(1, chunk_tokens - overlap)

    for i in range(0, len(toks), step):
        piece = toks[i:i + chunk_tokens]
        if not piece:
            continue
        chunk_text = enc.decode(piece).strip()
        if len(chunk_text) < 20:
            continue
        out.append(Chunk(
            chunk_id=f"{source}::tok{i}",
            text=chunk_text,
            source=source
        ))
    return out

def embed(texts: List[str]) -> np.ndarray:
    """
    OpenAI embeddings -> float32 numpy matrix
    """
    resp = client.embeddings.create(model=EMBED_MODEL, input=texts)
    vecs = [d.embedding for d in resp.data]
    return np.array(vecs, dtype="float32")

class FaissStore:
    def __init__(self, dim: int):
        self.index = faiss.IndexFlatIP(dim)  # cosine 유사도를 위해 내적 사용(벡터 정규화 전제)
        self.chunks: List[Chunk] = []

    def add(self, vecs: np.ndarray, chunks: List[Chunk]):
        faiss.normalize_L2(vecs)
        self.index.add(vecs)
        self.chunks.extend(chunks)

    def search(self, query_vec: np.ndarray, k: int = 4) -> List[Tuple[Chunk, float]]:
        q = query_vec.astype("float32")
        faiss.normalize_L2(q)
        D, I = self.index.search(q, k)
        results = []
        for score, idx in zip(D[0].tolist(), I[0].tolist()):
            if idx == -1:
                continue
            results.append((self.chunks[idx], float(score)))
        return results

def build_store(doc_glob: str = "./docs/*.txt") -> FaissStore:
    files = sorted(glob.glob(doc_glob))
    if not files:
        raise RuntimeError("docs/*.txt에 문서를 넣어주세요.")

    all_chunks: List[Chunk] = []
    for fp in files:
        with open(fp, "r", encoding="utf-8") as f:
            text = f.read()
        all_chunks.extend(chunk_text(text, source=os.path.basename(fp)))

    vecs = embed([c.text for c in all_chunks])
    store = FaissStore(dim=vecs.shape[1])
    store.add(vecs, all_chunks)
    return store

def judge_sufficiency(question: str, retrieved: List[Tuple[Chunk, float]]) -> bool:
    """
    휴리스틱 평가:
    - 상위 score가 너무 낮으면 근거 부족으로 판단
    - 운영에서는 별도 reranker/LLM-critic을 두는 게 일반적
    """
    if not retrieved:
        return False
    top_score = retrieved[0][1]
    return top_score >= 0.25  # 데이터에 따라 조정

def generate_answer(question: str, retrieved: List[Tuple[Chunk, float]]) -> str:
    context_lines = []
    for ch, score in retrieved:
        # citation을 chunk_id로 부여
        context_lines.append(f"[{ch.chunk_id} | score={score:.3f}]\n{ch.text}")

    system = (
        "You are a senior engineer. Answer ONLY using the provided CONTEXT.\n"
        "If the context is insufficient, say what is missing and ask a clarifying question.\n"
        "Include citations like [source_id] at the end of relevant sentences."
    )

    user = f"""QUESTION:
{question}

CONTEXT:
{'\n\n'.join(context_lines)}
"""

    resp = client.responses.create(
        model=GEN_MODEL,
        input=[
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ],
        temperature=0,
    )
    return resp.output_text

def rag_agent(question: str, store: FaissStore) -> str:
    """
    매우 단순화된 Agent 루프:
    - 1차 검색(k=4) -> 부족하면 k=10으로 재검색 -> 답변
    """
    qvec = embed([question])
    retrieved = store.search(qvec, k=4)

    if not judge_sufficiency(question, retrieved):
        # 재검색(“더 찾아보기” 액션)
        retrieved = store.search(qvec, k=10)

    return generate_answer(question, retrieved)

if __name__ == "__main__":
    store = build_store("./docs/*.txt")
    q = "우리 서비스의 데이터 보관 기간과 삭제 정책을 요약해줘."
    print(rag_agent(q, store))

핵심은 “RAG + Agent 루프”를 분리해서 생각하는 겁니다. OpenAI가 말하는 것처럼 built-in file search를 쓰면 retrieval(쿼리 최적화/리랭크 포함)을 플랫폼에 위임할 수 있고 (openai.com), 위 코드처럼 직접 구현하면 “내가 원하는 chunking/메타데이터/평가기준”을 끝까지 컨트롤할 수 있습니다.

---

⚡ 실전 팁

• Chunking은 ‘균일 길이’보다 ‘의미 단위’가 이깁니다. 토큰 기준 chunking은 시작점일 뿐이고, 제목/섹션/표/코드블록 경계를 살리는 “semantic chunking”이 검색 품질을 크게 올립니다(특히 기술 문서).
• Hybrid search + rerank를 고려하세요. vector 유사도만 쓰면 “정확한 키워드(버전/에러코드/설정키)”를 놓칠 수 있습니다. BM25+vector 혼합 후 reranker로 정렬하는 조합이 안정적입니다.
• Agent의 ‘평가 단계’를 코드로 분리하세요. 위 예제는 score 휴리스틱이지만, 운영에서는 (a) LLM-critic, (b) 규칙 기반(필수 섹션/키워드 존재), (c) 충돌 감지(서로 다른 근거) 등을 조합합니다.
• Citations는 기능이 아니라 안전장치입니다. “답변 문장 ↔ 근거 chunk” 링크가 있어야 디버깅이 됩니다. OpenAI도 File search를 RAG 파이프라인의 핵심 도구로 강조하면서 빠르고 정확한 검색을 목표로 합니다. (openai.com)
• 툴을 늘릴수록 관측 가능성(Observability)이 먼저입니다. Agentic RAG는 multi-step이 기본이므로, 각 스텝(쿼리 변환/검색/리랭크/생성)의 로그와 비용, latency를 분리해서 수집하세요.

---

🚀 마무리

2025년의 “RAG Agent 구현”은 결국 (1) Retrieval 품질을 올리는 엔지니어링과 (2) Agent 루프로 의사결정을 자동화하는 문제로 정리됩니다. OpenAI의 built-in tools(특히 File search)는 RAG의 복잡한 단계를 크게 줄여주고 (developers.openai.com), 반대로 직접 구현은 도메인 최적화(메타데이터/평가/보안/비용)에서 강합니다.

다음 학습으로는:

• OpenAI Responses API + File search 기반 “managed RAG” 패턴 (빠른 제품화) (openai.com)
• LlamaIndex의 agentic strategies/workflows로 “routing·query transformation·tool orchestration” 확장 (docs.llamaindex.ai)

원하시면, 위 코드에 (a) metadata 필터링, (b) reranker 추가, (c) 대화 메모리(세션/장기) 계층, (d) 웹 검색 툴까지 붙여서 “진짜 Agentic RAG” 형태로 확장한 버전도 이어서 작성해드릴게요.