RAG 구현 완벽 가이드 | 검색 증강 생성으로 LLM 성능 향상
이 글의 핵심
RAG(Retrieval-Augmented Generation)로 LLM 성능을 향상시키는 완벽 가이드. 벡터 임베딩, 유사도 검색, 청킹 전략, 하이브리드 검색, 리랭킹까지 실전 예제로 완벽 이해.
들어가며
RAG(Retrieval-Augmented Generation)는 LLM의 한계를 극복하는 핵심 기술입니다. 외부 문서를 검색하여 LLM에 제공함으로써 최신 정보, 특정 도메인 지식, 사실 기반 답변을 가능하게 합니다.
실무 경험: 고객 지원 챗봇에 RAG를 적용하면서, 5만 건의 FAQ와 정책 문서를 벡터화하여 답변 정확도를 70%에서 92%로 향상시킨 경험을 바탕으로 작성했습니다.
이 글에서 다룰 내용:
- RAG 아키텍처와 작동 원리
- 문서 전처리와 청킹 전략
- 벡터 임베딩과 유사도 검색
- 벡터 데이터베이스 선택
- 하이브리드 검색과 리랭킹
- 프로덕션 RAG 시스템 구축
목차
1. RAG란 무엇인가
RAG 작동 원리
flowchart LR
User[사용자 질문] --> Embed1[질문 임베딩]
Embed1 --> Search[벡터 검색]
Docs[문서 DB] --> Chunk[청킹]
Chunk --> Embed2[문서 임베딩]
Embed2 --> VectorDB[(벡터 DB)]
VectorDB --> Search
Search --> Relevant[관련 문서]
Relevant --> Prompt[프롬프트 구성]
User --> Prompt
Prompt --> LLM[LLM]
LLM --> Answer[답변]
RAG vs Fine-tuning
| 특성 | RAG | Fine-tuning |
|---|---|---|
| 비용 | 저렴 (검색만) | 비쌈 (학습 필요) |
| 업데이트 | 즉시 (문서 추가) | 재학습 필요 |
| 투명성 | 높음 (출처 확인) | 낮음 (블랙박스) |
| 정확도 | 사실 기반 | 학습 데이터 의존 |
| 용도 | 지식 검색, Q&A | 스타일, 특정 작업 |
2. RAG 아키텍처
기본 RAG 파이프라인
from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
class BasicRAG:
def __init__(self, documents_path):
# 1. 문서 로딩
loader = TextLoader(documents_path)
documents = loader.load()
# 2. 청킹
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
chunks = text_splitter.split_documents(documents)
# 3. 임베딩 및 벡터 스토어
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
self.vectorstore = Chroma.from_documents(
chunks,
embeddings,
persist_directory="./chroma_db"
)
# 4. LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 5. QA Chain
self.qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=self.vectorstore.as_retriever(search_kwargs={"k": 3})
)
def query(self, question):
return self.qa_chain.run(question)
# 사용
rag = BasicRAG("knowledge_base.txt")
answer = rag.query("What is the main topic?")
3. 문서 전처리
청킹 전략
from langchain.text_splitter import (
RecursiveCharacterTextSplitter,
CharacterTextSplitter,
TokenTextSplitter
)
# 1. Character-based (기본)
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
separators=["\n\n", "\n", ". ", " ", ""]
)
# 2. Token-based (정확한 토큰 수)
splitter = TokenTextSplitter(
chunk_size=500,
chunk_overlap=50
)
# 3. Semantic-based (의미 단위)
from langchain.text_splitter import SpacyTextSplitter
splitter = SpacyTextSplitter(chunk_size=1000)
청킹 크기 최적화
def find_optimal_chunk_size(documents, test_queries):
"""최적 청킹 크기 찾기"""
chunk_sizes = [250, 500, 1000, 2000]
results = {}
for size in chunk_sizes:
splitter = RecursiveCharacterTextSplitter(
chunk_size=size,
chunk_overlap=size // 5
)
chunks = splitter.split_documents(documents)
# 벡터 스토어 생성
vectorstore = Chroma.from_documents(chunks, embeddings)
# 테스트 쿼리 실행
scores = []
for query in test_queries:
docs = vectorstore.similarity_search(query, k=3)
# 관련성 점수 계산 (실제로는 더 복잡한 평가 필요)
scores.append(len(docs))
results[size] = sum(scores) / len(scores)
return results
메타데이터 추가
from langchain.schema import Document
documents = [
Document(
page_content="LangChain is a framework...",
metadata={
"source": "docs.langchain.com",
"date": "2026-04-01",
"category": "introduction",
"author": "LangChain Team"
}
)
]
# 메타데이터 필터링
retriever = vectorstore.as_retriever(
search_kwargs={
"k": 5,
"filter": {"category": "introduction"}
}
)
4. 벡터 임베딩
OpenAI Embeddings
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(
model="text-embedding-3-small" # 또는 text-embedding-3-large
)
# 텍스트 임베딩
text = "LangChain is awesome"
vector = embeddings.embed_query(text)
print(f"Vector dimension: {len(vector)}") # 1536
# 여러 텍스트 임베딩
texts = ["text1", "text2", "text3"]
vectors = embeddings.embed_documents(texts)
임베딩 모델 비교
| 모델 | 차원 | 가격 ($/1M 토큰) | 성능 |
|---|---|---|---|
| text-embedding-3-small | 1536 | $0.02 | 빠름, 저렴 |
| text-embedding-3-large | 3072 | $0.13 | 높은 정확도 |
| text-embedding-ada-002 | 1536 | $0.10 | 이전 모델 |
로컬 임베딩 (무료)
from langchain.embeddings import HuggingFaceEmbeddings
# 무료 오픈소스 모델
embeddings = HuggingFaceEmbeddings(
model_name="sentence-transformers/all-MiniLM-L6-v2"
)
vector = embeddings.embed_query("test")
5. 벡터 데이터베이스
Chroma (로컬)
from langchain.vectorstores import Chroma
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db"
)
# 검색
results = vectorstore.similarity_search("query", k=5)
# 점수와 함께 검색
results = vectorstore.similarity_search_with_score("query", k=5)
for doc, score in results:
print(f"Score: {score:.4f} | {doc.page_content[:100]}")
Pinecone (클라우드)
from langchain.vectorstores import Pinecone
import pinecone
# 초기화
pinecone.init(
api_key="...",
environment="us-west1-gcp"
)
# 인덱스 생성
index_name = "my-rag-index"
if index_name not in pinecone.list_indexes():
pinecone.create_index(
name=index_name,
dimension=1536,
metric="cosine"
)
# 벡터 스토어
vectorstore = Pinecone.from_documents(
documents=chunks,
embedding=embeddings,
index_name=index_name
)
Weaviate
from langchain.vectorstores import Weaviate
import weaviate
client = weaviate.Client(
url="http://localhost:8080"
)
vectorstore = Weaviate.from_documents(
documents=chunks,
embedding=embeddings,
client=client,
by_text=False
)
벡터 DB 비교
| DB | 특징 | 가격 | 추천 용도 |
|---|---|---|---|
| Chroma | 로컬, 간단 | 무료 | 개발, 소규모 |
| FAISS | 로컬, 빠름 | 무료 | 대규모 로컬 |
| Pinecone | 클라우드, 관리형 | 유료 | 프로덕션 |
| Weaviate | 오픈소스, 확장성 | 무료/유료 | 대규모 |
| Qdrant | 오픈소스, 빠름 | 무료/유료 | 고성능 |
6. 검색 전략
기본 유사도 검색
# Cosine similarity (기본)
results = vectorstore.similarity_search("query", k=5)
MMR (Maximum Marginal Relevance)
# 다양성을 고려한 검색
results = vectorstore.max_marginal_relevance_search(
"query",
k=5,
fetch_k=20, # 초기 후보
lambda_mult=0.5 # 0=다양성, 1=관련성
)
하이브리드 검색
from langchain.retrievers import BM25Retriever, EnsembleRetriever
# 1. 벡터 검색
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
# 2. 키워드 검색 (BM25)
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5
# 3. 하이브리드 (앙상블)
ensemble_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.5, 0.5] # 가중치
)
docs = ensemble_retriever.get_relevant_documents("query")
메타데이터 필터링
# 날짜 필터
retriever = vectorstore.as_retriever(
search_kwargs={
"k": 5,
"filter": {
"date": {"$gte": "2026-01-01"}
}
}
)
# 복합 필터
retriever = vectorstore.as_retriever(
search_kwargs={
"k": 5,
"filter": {
"$and": [
{"category": "technical"},
{"author": "John Doe"}
]
}
}
)
7. 프롬프트 구성
기본 RAG 프롬프트
from langchain.prompts import PromptTemplate
template = """Use the following context to answer the question.
If you don't know the answer, say so. Don't make up information.
Context:
{context}
Question: {question}
Answer:"""
prompt = PromptTemplate(
template=template,
input_variables=["context", "question"]
)
출처 포함 프롬프트
template = """Answer the question based on the context below.
Include the source document in your answer.
Context:
{context}
Question: {question}
Answer (include source):"""
다국어 프롬프트
template = """다음 문서를 참고하여 질문에 답하세요.
답을 모르면 모른다고 말하세요. 추측하지 마세요.
문서:
{context}
질문: {question}
답변:"""
8. 고급 기법
Query Transformation
from langchain.chains import LLMChain
# 질문을 여러 하위 질문으로 분해
decomposition_prompt = PromptTemplate(
template="""Break down this question into 3 simpler sub-questions:
Question: {question}
Sub-questions:""",
input_variables=["question"]
)
decomposition_chain = LLMChain(llm=llm, prompt=decomposition_prompt)
# HyDE (Hypothetical Document Embeddings)
hyde_prompt = PromptTemplate(
template="""Write a hypothetical document that would answer this question:
Question: {question}
Document:""",
input_variables=["question"]
)
hyde_chain = LLMChain(llm=llm, prompt=hyde_prompt)
Reranking
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain.retrievers import ContextualCompressionRetriever
# 기본 retriever (많은 문서 가져오기)
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 20})
# Compressor (관련성 높은 것만 선택)
compressor = LLMChainExtractor.from_llm(llm)
# Compression retriever
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=base_retriever
)
# 사용
docs = compression_retriever.get_relevant_documents("What is RAG?")
Multi-Query Retrieval
from langchain.retrievers.multi_query import MultiQueryRetriever
# 질문을 여러 형태로 변환하여 검색
retriever = MultiQueryRetriever.from_llm(
retriever=vectorstore.as_retriever(),
llm=llm
)
# "What is RAG?" →
# - "Explain RAG"
# - "How does RAG work?"
# - "RAG definition"
docs = retriever.get_relevant_documents("What is RAG?")
Parent Document Retriever
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
# 작은 청크로 검색, 큰 청크 반환
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
store = InMemoryStore()
retriever = ParentDocumentRetriever(
vectorstore=vectorstore,
docstore=store,
child_splitter=child_splitter,
parent_splitter=parent_splitter
)
9. 프로덕션 구현
전체 RAG 시스템
from typing import List, Dict
import logging
class ProductionRAG:
def __init__(self, config: Dict):
self.config = config
self.logger = logging.getLogger(__name__)
# LLM
self.llm = ChatOpenAI(
model=config.get("model", "gpt-4o-mini"),
temperature=config.get("temperature", 0),
max_retries=3
)
# Embeddings
self.embeddings = OpenAIEmbeddings(
model="text-embedding-3-small"
)
# Vector Store
self.vectorstore = self._init_vectorstore()
# QA Chain
self.qa_chain = self._create_qa_chain()
def _init_vectorstore(self):
"""벡터 스토어 초기화"""
if os.path.exists(self.config["db_path"]):
return Chroma(
persist_directory=self.config["db_path"],
embedding_function=self.embeddings
)
return None
def _create_qa_chain(self):
"""QA Chain 생성"""
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True,
output_key="answer"
)
return ConversationalRetrievalChain.from_llm(
llm=self.llm,
retriever=self.vectorstore.as_retriever(
search_kwargs={"k": self.config.get("top_k", 3)}
),
memory=memory,
return_source_documents=True
)
def ingest_documents(self, file_paths: List[str]):
"""문서 인제스트"""
all_chunks = []
for file_path in file_paths:
self.logger.info(f"Processing {file_path}")
# 로딩
if file_path.endswith('.pdf'):
loader = PyPDFLoader(file_path)
else:
loader = TextLoader(file_path)
documents = loader.load()
# 청킹
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=self.config.get("chunk_size", 1000),
chunk_overlap=self.config.get("chunk_overlap", 200)
)
chunks = text_splitter.split_documents(documents)
# 메타데이터 추가
for chunk in chunks:
chunk.metadata["source_file"] = file_path
all_chunks.extend(chunks)
# 벡터 스토어 생성/업데이트
if self.vectorstore is None:
self.vectorstore = Chroma.from_documents(
all_chunks,
self.embeddings,
persist_directory=self.config["db_path"]
)
else:
self.vectorstore.add_documents(all_chunks)
self.logger.info(f"Ingested {len(all_chunks)} chunks")
return len(all_chunks)
def query(self, question: str, return_sources: bool = True):
"""질문에 답변"""
try:
result = self.qa_chain({
"question": question
})
answer = {
"answer": result["answer"],
"sources": []
}
if return_sources and "source_documents" in result:
for doc in result["source_documents"]:
answer["sources"].append({
"content": doc.page_content[:200],
"metadata": doc.metadata
})
return answer
except Exception as e:
self.logger.error(f"Query error: {e}")
raise
# 사용
config = {
"model": "gpt-4o-mini",
"temperature": 0,
"db_path": "./chroma_db",
"chunk_size": 1000,
"chunk_overlap": 200,
"top_k": 3
}
rag = ProductionRAG(config)
# 문서 추가
rag.ingest_documents(["doc1.pdf", "doc2.txt"])
# 질문
result = rag.query("What is the main topic?")
print(result["answer"])
for source in result["sources"]:
print(f"Source: {source['metadata']['source_file']}")
성능 최적화
임베딩 캐싱
from langchain.embeddings import CacheBackedEmbeddings
from langchain.storage import LocalFileStore
# 로컬 캐시
store = LocalFileStore("./embedding_cache")
cached_embeddings = CacheBackedEmbeddings.from_bytes_store(
underlying_embeddings=OpenAIEmbeddings(),
document_embedding_cache=store,
namespace="openai-embeddings"
)
# 동일한 텍스트는 캐시에서 반환 (API 호출 없음)
배치 처리
# 대량 문서 처리
def batch_ingest(file_paths, batch_size=100):
for i in range(0, len(file_paths), batch_size):
batch = file_paths[i:i+batch_size]
rag.ingest_documents(batch)
print(f"Processed {i+len(batch)}/{len(file_paths)}")
비동기 처리
import asyncio
from langchain.callbacks import AsyncCallbackHandler
async def async_query(rag, questions):
tasks = [rag.aquery(q) for q in questions]
results = await asyncio.gather(*tasks)
return results
평가 및 모니터링
RAG 성능 평가
from langchain.evaluation import load_evaluator
# 답변 관련성 평가
evaluator = load_evaluator("qa")
test_cases = [
{
"query": "What is RAG?",
"expected": "RAG is Retrieval-Augmented Generation..."
}
]
for case in test_cases:
result = rag.query(case["query"])
score = evaluator.evaluate_strings(
prediction=result["answer"],
reference=case["expected"]
)
print(f"Score: {score}")
검색 품질 측정
def evaluate_retrieval(vectorstore, test_queries):
"""검색 품질 평가"""
metrics = {
"precision": [],
"recall": []
}
for query, relevant_docs in test_queries:
retrieved = vectorstore.similarity_search(query, k=5)
retrieved_ids = {doc.metadata["id"] for doc in retrieved}
relevant_ids = set(relevant_docs)
# Precision: 검색된 것 중 관련 있는 비율
precision = len(retrieved_ids & relevant_ids) / len(retrieved_ids)
# Recall: 관련 있는 것 중 검색된 비율
recall = len(retrieved_ids & relevant_ids) / len(relevant_ids)
metrics["precision"].append(precision)
metrics["recall"].append(recall)
return {
"avg_precision": sum(metrics["precision"]) / len(metrics["precision"]),
"avg_recall": sum(metrics["recall"]) / len(metrics["recall"])
}
실전 프로젝트: 문서 검색 API
from fastapi import FastAPI, UploadFile, File, HTTPException
from pydantic import BaseModel
import uuid
app = FastAPI()
class QueryRequest(BaseModel):
question: str
session_id: str = None
class QueryResponse(BaseModel):
answer: str
sources: List[Dict]
session_id: str
# RAG 인스턴스
rag_systems = {}
@app.post("/upload")
async def upload_documents(files: List[UploadFile] = File(...)):
"""문서 업로드 및 인덱싱"""
session_id = str(uuid.uuid4())
# 임시 파일 저장
file_paths = []
for file in files:
tmp_path = f"/tmp/{session_id}_{file.filename}"
with open(tmp_path, "wb") as f:
f.write(await file.read())
file_paths.append(tmp_path)
# RAG 시스템 생성
config = {
"model": "gpt-4o-mini",
"db_path": f"./db/{session_id}",
"chunk_size": 1000,
"top_k": 3
}
rag = ProductionRAG(config)
chunks_count = rag.ingest_documents(file_paths)
rag_systems[session_id] = rag
return {
"session_id": session_id,
"files": len(files),
"chunks": chunks_count
}
@app.post("/query", response_model=QueryResponse)
async def query_documents(request: QueryRequest):
"""문서 질의"""
if request.session_id not in rag_systems:
raise HTTPException(status_code=404, detail="Session not found")
rag = rag_systems[request.session_id]
try:
result = rag.query(request.question)
return QueryResponse(
answer=result["answer"],
sources=result["sources"],
session_id=request.session_id
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.delete("/session/{session_id}")
async def delete_session(session_id: str):
"""세션 삭제"""
if session_id in rag_systems:
del rag_systems[session_id]
# 벡터 DB 정리
import shutil
shutil.rmtree(f"./db/{session_id}", ignore_errors=True)
return {"status": "deleted"}
청킹 전략 비교
고정 크기 청킹
# 장점: 간단, 빠름
# 단점: 문맥 손실 가능
splitter = CharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
의미 기반 청킹
# 장점: 문맥 보존
# 단점: 느림, 복잡
from langchain.text_splitter import SpacyTextSplitter
splitter = SpacyTextSplitter(
chunk_size=1000,
pipeline="en_core_web_sm"
)
문서 구조 기반 청킹
# Markdown 헤딩 기준
from langchain.text_splitter import MarkdownHeaderTextSplitter
headers_to_split_on = [
("#", "Header 1"),
("##", "Header 2"),
("###", "Header 3"),
]
splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=headers_to_split_on
)
비용 분석
RAG 비용 구성
def estimate_rag_cost(
num_documents,
avg_doc_size,
queries_per_day,
top_k=3
):
"""RAG 시스템 비용 추정"""
# 1. 임베딩 비용 (1회)
total_tokens = num_documents * avg_doc_size
embedding_cost = total_tokens * 0.02 / 1_000_000 # $0.02/1M tokens
# 2. 검색 비용 (무료, 벡터 DB 비용 별도)
# 3. LLM 비용 (매일)
# 질문 + 검색된 문서 (top_k * chunk_size)
input_tokens_per_query = 100 + (top_k * 500) # 대략적
output_tokens_per_query = 200
daily_input_tokens = queries_per_day * input_tokens_per_query
daily_output_tokens = queries_per_day * output_tokens_per_query
# GPT-4o-mini 가격
daily_llm_cost = (
daily_input_tokens * 0.15 / 1_000_000 +
daily_output_tokens * 0.60 / 1_000_000
)
monthly_llm_cost = daily_llm_cost * 30
return {
"embedding_cost_once": f"${embedding_cost:.2f}",
"llm_cost_daily": f"${daily_llm_cost:.4f}",
"llm_cost_monthly": f"${monthly_llm_cost:.2f}",
"total_first_month": f"${embedding_cost + monthly_llm_cost:.2f}"
}
# 예시: 1000개 문서, 하루 100 쿼리
costs = estimate_rag_cost(
num_documents=1000,
avg_doc_size=2000,
queries_per_day=100
)
print(costs)
베스트 프랙티스
1. 청킹 전략
# ✅ 문서 타입별 최적화
def get_splitter(doc_type):
if doc_type == "code":
return RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
separators=["\nclass ", "\ndef ", "\n\n", "\n", " "]
)
elif doc_type == "markdown":
return MarkdownHeaderTextSplitter(...)
else:
return RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
2. 메타데이터 활용
# ✅ 풍부한 메타데이터
Document(
page_content="...",
metadata={
"source": "file.pdf",
"page": 5,
"date": "2026-04-01",
"author": "John Doe",
"category": "technical",
"language": "ko"
}
)
3. 모니터링
# ✅ 로깅 및 추적
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def query_with_logging(question):
logger.info(f"Query: {question}")
start = time.time()
result = rag.query(question)
elapsed = time.time() - start
logger.info(f"Response time: {elapsed:.2f}s")
logger.info(f"Sources: {len(result['sources'])}")
return result
참고 자료
한 줄 요약: RAG는 외부 문서 검색으로 LLM의 지식을 확장하는 핵심 기술이며, 적절한 청킹, 임베딩, 검색 전략을 통해 정확하고 사실 기반의 AI 애플리케이션을 구축할 수 있습니다.