LLM(대규모 언어 모델)은 놀라운 텍스트 생성 능력을 가지고 있지만, 학습 데이터에 없는 내용을 그럴듯하게 지어내는 환각(Hallucination) 문제가 있습니다. 사내 문서, 최신 데이터, 도메인 특화 지식을 기반으로 정확한 답변을 생성해야 하는 엔터프라이즈 환경에서 이는 치명적입니다. RAG(Retrieval-Augmented Generation)는 이 문제를 해결하는 가장 실용적인 접근법입니다.

RAG란 무엇인가?

RAG는 사용자의 질문에 대해 외부 지식 저장소에서 관련 문서를 먼저 검색(Retrieve)한 뒤, 그 문서를 컨텍스트로 제공하여 LLM이 답변을 생성(Generate)하는 패턴입니다. 파인 튜닝과 달리 모델을 재학습할 필요가 없고, 지식을 실시간으로 업데이트할 수 있다는 장점이 있습니다.

RAG vs 파인 튜닝: 파인 튜닝은 모델의 행동 패턴을 변경할 때 적합하고, RAG는 최신 또는 특정 도메인 지식을 주입할 때 적합합니다. 대부분의 실무 시나리오에서는 RAG가 비용 대비 효과적입니다.

RAG의 핵심 장점

RAG 기본 아키텍처: 3단계 파이프라인

RAG 시스템은 크게 Indexing(색인), Retrieval(검색), Generation(생성) 세 단계로 구성됩니다. 각 단계를 올바르게 설계하는 것이 전체 시스템의 품질을 좌우합니다.

1단계: Indexing (색인)

원본 문서를 벡터 DB에 저장 가능한 형태로 변환하는 단계입니다. 문서 로딩, 청킹(분할), 임베딩 생성, 벡터 저장의 과정을 거칩니다.

# LangChain을 활용한 기본 인덱싱 파이프라인
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 1. 문서 로딩
loader = PyPDFLoader("company_docs.pdf")
documents = loader.load()

# 2. 문서 청킹
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ".", " ", ""]
)
chunks = text_splitter.split_documents(documents)

# 3. 임베딩 생성 및 벡터 DB 저장
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

2단계: Retrieval (검색)

사용자 질문을 임베딩으로 변환한 뒤, 벡터 DB에서 의미적으로 유사한 문서 청크를 검색합니다. 검색 품질이 RAG 전체의 성능을 결정하는 핵심 단계입니다.

3단계: Generation (생성)

검색된 문서 청크를 프롬프트에 포함시켜 LLM에 전달하고, 컨텍스트 기반 답변을 생성합니다.

# 검색 + 생성 통합 파이프라인
from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

# 프롬프트 템플릿 정의
prompt_template = """아래 컨텍스트를 참고하여 질문에 답변하세요.
컨텍스트에 없는 내용은 "해당 정보를 찾을 수 없습니다"라고 답변하세요.

컨텍스트:
{context}

질문: {question}
답변:"""

PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

# RAG 체인 구성
llm = ChatOpenAI(model="gpt-4o", temperature=0)
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5}
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    chain_type_kwargs={"prompt": PROMPT},
    return_source_documents=True
)

# 질의 실행
result = qa_chain.invoke({"query": "회사의 연차 정책은 무엇인가요?"})
print(result["result"])
print("참조 문서:", [doc.metadata for doc in result["source_documents"]])

벡터 DB 비교: 어떤 것을 선택할까?

RAG 시스템의 핵심 인프라인 벡터 DB는 프로젝트의 규모와 요구사항에 따라 선택이 달라집니다.

주요 벡터 DB 비교

실무 추천: PoC 단계에서는 Chroma로 빠르게 검증하고, 프로덕션에서는 이미 PostgreSQL을 사용 중이면 pgvector, 관리형 서비스가 필요하면 Pinecone, 하이브리드 검색이 중요하면 Weaviate를 고려하세요.

문서 청킹 전략

문서를 어떻게 분할하느냐에 따라 검색 품질이 크게 달라집니다. 청크가 너무 작으면 문맥이 손실되고, 너무 크면 관련 없는 내용이 포함되어 노이즈가 증가합니다.

청킹 전략별 비교

# 시맨틱 청킹 구현 예시
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 의미적 유사도 기반 분할
semantic_splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95
)

semantic_chunks = semantic_splitter.split_documents(documents)
print(f"시맨틱 청킹 결과: {len(semantic_chunks)}개 청크 생성")

# Markdown 구조 기반 분할
from langchain.text_splitter import MarkdownHeaderTextSplitter

headers_to_split_on = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]

md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on
)
md_chunks = md_splitter.split_text(markdown_document)
주의: chunk_overlap(청크 간 겹침)을 적절히 설정하세요. 일반적으로 chunk_size의 10~20%가 권장됩니다. 겹침이 없으면 청크 경계에서 문맥이 끊겨 검색 품질이 떨어집니다.

임베딩 모델 선택

임베딩 모델은 텍스트를 벡터로 변환하는 핵심 구성요소입니다. 모델에 따라 의미 파악 능력, 다국어 지원, 차원 수, 비용이 달라집니다.

주요 임베딩 모델 비교

# 오픈소스 임베딩 모델 활용 (BGE-M3)
from langchain_huggingface import HuggingFaceEmbeddings

# BGE-M3: 다국어 + 다중 기능 임베딩 모델
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-m3",
    model_kwargs={"device": "cuda"},
    encode_kwargs={
        "normalize_embeddings": True,
        "batch_size": 32
    }
)

# 한국어 테스트
test_texts = [
    "쿠버네티스에서 파드 오토스케일링을 설정하는 방법",
    "Kubernetes HPA 구성 가이드",
    "파이썬으로 웹 크롤러 만들기"
]
vectors = embeddings.embed_documents(test_texts)
print(f"벡터 차원: {len(vectors[0])}")

하이브리드 검색: 키워드 + 시맨틱

순수 벡터 검색(시맨틱 검색)만으로는 고유 명사, 코드 스니펫, 약어 등을 정확히 매칭하기 어렵습니다. 키워드 검색(BM25)과 시맨틱 검색을 결합하는 하이브리드 검색이 실무에서 더 나은 결과를 제공합니다.

# LangChain EnsembleRetriever로 하이브리드 검색 구현
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

# BM25 키워드 검색 리트리버
bm25_retriever = BM25Retriever.from_documents(
    chunks,
    k=5
)

# 벡터 시맨틱 검색 리트리버
vector_retriever = vectorstore.as_retriever(
    search_kwargs={"k": 5}
)

# 하이브리드 앙상블 리트리버 (가중치 조절 가능)
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # 시맨틱 검색에 더 높은 가중치
)

# 검색 실행
results = ensemble_retriever.invoke("pgvector 인덱스 생성 DDL")
for doc in results:
    print(f"[{doc.metadata.get('source', 'unknown')}] {doc.page_content[:100]}...")
가중치 튜닝 팁: 기술 문서 검색에서는 키워드 가중치를 높이고(0.5:0.5), 일반 Q&A에서는 시맨틱 가중치를 높이는(0.3:0.7) 것이 효과적입니다. A/B 테스트를 통해 최적 비율을 찾으세요.

RAG 평가 메트릭

RAG 시스템의 품질을 객관적으로 측정하려면 체계적인 평가 프레임워크가 필요합니다. RAGAS(Retrieval Augmented Generation Assessment) 프레임워크에서 제안하는 핵심 메트릭들을 살펴보겠습니다.

핵심 평가 지표

# RAGAS를 활용한 RAG 파이프라인 평가
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall
)
from datasets import Dataset

# 평가 데이터셋 구성
eval_data = {
    "question": [
        "연차 휴가는 몇 일인가요?",
        "재택근무 신청 절차는?"
    ],
    "answer": [
        qa_chain.invoke({"query": "연차 휴가는 몇 일인가요?"})["result"],
        qa_chain.invoke({"query": "재택근무 신청 절차는?"})["result"]
    ],
    "contexts": [
        [doc.page_content for doc in result1["source_documents"]],
        [doc.page_content for doc in result2["source_documents"]]
    ],
    "ground_truth": [
        "입사 1년 미만 시 월 1일, 1년 이상 시 15일의 연차가 부여됩니다.",
        "사내 포털에서 최소 3일 전에 재택근무를 신청합니다."
    ]
}

eval_dataset = Dataset.from_dict(eval_data)

# 평가 실행
result = evaluate(
    eval_dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall
    ]
)
print(result)

Advanced RAG 기법

기본 RAG로 충분하지 않을 때 적용할 수 있는 고급 기법들을 소개합니다.

Query Transformation (질문 변환)

사용자의 원래 질문을 검색에 더 적합한 형태로 변환하여 검색 품질을 높이는 기법입니다.

# Multi-Query: 하나의 질문을 여러 관점으로 변환
from langchain.retrievers.multi_query import MultiQueryRetriever

multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
)

# "연차 정책" -> 다양한 변형 질의로 확장 검색
results = multi_query_retriever.invoke("연차 정책이 어떻게 되나요?")

# HyDE: 가상 답변을 생성하여 해당 답변과 유사한 문서 검색
from langchain.chains import HypotheticalDocumentEmbedder

hyde_embeddings = HypotheticalDocumentEmbedder.from_llm(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    base_embeddings=embeddings,
    prompt_key="web_search"
)

Re-Ranking (재순위화)

초기 검색 결과를 Cross-Encoder 모델로 재순위화하여 가장 관련성 높은 문서를 상위로 올립니다.

# Cohere Reranker를 활용한 재순위화
from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank

# 1차 검색: 넓은 범위에서 후보 문서 검색 (k=20)
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 20})

# 2차 재순위화: 상위 5개로 압축
compressor = CohereRerank(
    model="rerank-multilingual-v3.0",
    top_n=5
)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever
)

reranked_docs = compression_retriever.invoke("퇴직금 계산 방법")
성능 주의: Re-Ranking은 Cross-Encoder 모델을 사용하므로 지연 시간이 추가됩니다. 첫 번째 검색에서 가져오는 문서 수(k)를 적절히 제한하고, Re-Ranker 호출은 캐싱을 적용하여 반복 질의에 대한 응답 속도를 개선하세요.

프로덕션 운영 팁

RAG 시스템을 프로덕션 환경에서 안정적으로 운영하기 위한 실무 가이드입니다.

문서 업데이트 파이프라인

사내 문서가 변경될 때마다 벡터 DB를 동기화하는 파이프라인이 필요합니다. 전체 재인덱싱은 비용이 크므로, 변경된 문서만 증분 업데이트하는 전략을 사용합니다.

# 증분 업데이트를 위한 문서 해시 기반 변경 감지
import hashlib
from datetime import datetime

class DocumentSyncManager:
    def __init__(self, vectorstore, redis_client):
        self.vectorstore = vectorstore
        self.redis = redis_client

    def compute_hash(self, content: str) -> str:
        return hashlib.sha256(content.encode()).hexdigest()

    def sync_document(self, doc_id: str, content: str, metadata: dict):
        new_hash = self.compute_hash(content)
        old_hash = self.redis.get(f"doc_hash:{doc_id}")

        if old_hash and old_hash.decode() == new_hash:
            return False  # 변경 없음, 스킵

        # 기존 청크 삭제 후 재인덱싱
        self.vectorstore.delete(
            filter={"doc_id": doc_id}
        )

        chunks = self.text_splitter.split_text(content)
        self.vectorstore.add_texts(
            texts=chunks,
            metadatas=[{**metadata, "doc_id": doc_id} for _ in chunks]
        )

        # 해시 업데이트
        self.redis.set(f"doc_hash:{doc_id}", new_hash)
        self.redis.set(f"doc_updated:{doc_id}", datetime.now().isoformat())
        return True

모니터링 및 관찰 가능성

비용 최적화

마무리

RAG는 LLM의 환각 문제를 해결하고, 사내 지식을 활용한 AI 서비스를 구축하는 데 가장 실용적인 접근법입니다. 핵심은 문서 청킹 품질과 검색 정확도에 있습니다. 작은 규모의 PoC부터 시작하여 평가 메트릭을 기반으로 지속적으로 개선해 나가는 것을 권장합니다. 하이브리드 검색과 Re-Ranking 같은 고급 기법을 점진적으로 도입하면 프로덕션 수준의 RAG 시스템을 구축할 수 있습니다.

← 목록으로
Jaeseong
Jaeseong

10년차 풀스택 개발자. Spring Boot, Flutter, AI 등 실무 경험을 기록합니다.

GitHub →

💬 댓글