RAG 시스템 실전 구축 (v9)
개요
이 가이드는 실제 ML 엔지니어와 백엔드 개발자가 RAG (Retrieval-Augmented Generation) 시스템을 구축할 때 필요한 모든 요소를 다룹니다. RAG는 LLM의 정보 확장과 생성 정확도를 높이는 핵심 기술로, 현대 대형 언어 모델의 실용성 향상에 필수적입니다.
1. RAG 기초 개념: 검색 → 보강 → 생성 루프
RAG는 세 가지 핵심 단계로 구성됩니다:
- 검색 (Retrieval): 사용자의 질문과 관련된 문서 또는 정보를 검색
- 보강 (Augmentation): 검색된 정보를 프롬프트에 추가하여 컨텍스트 제공
- 생성 (Generation): LLM이 보강된 컨텍스트를 기반으로 답변 생성
from langchain_core.prompts import PromptTemplate
from langchain_openai import OpenAI
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
class RAGPipeline:
def __init__(self):
self.embeddings = OpenAIEmbeddings()
self.llm = OpenAI()
self.vectorstore = Chroma(
persist_directory="./chroma_db",
embedding_function=self.embeddings
)
def process_query(self, query):
# 1. 검색
docs = self.vectorstore.similarity_search(query, k=3)
# 2. 보강
context = "\n".join([doc.page_content for doc in docs])
# 3. 생성
prompt = PromptTemplate.from_template("""
질문: {query}
컨텍스트: {context}
위의 정보를 바탕으로 질문에 답해주세요.
""")
response = self.llm.invoke(prompt.format(query=query, context=context))
return response
2. 청킹 전략 비교
청킹은 문서를 의미 단위로 분할하여 임베딩 생성의 효율성을 높입니다. 세 가지 주요 전략:
2.1 의미적 청킹 (Semantic Chunking)
의미 단위를 기준으로 문서를 분할하여 문맥의 연속성을 유지합니다:
import tiktoken
from langchain_text_splitters import RecursiveCharacterTextSplitter
def semantic_chunking(text, chunk_size=500, chunk_overlap=50):
# Recursive splitting을 통한 청킹
splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", " ", ""]
)
return splitter.split_text(text)
2.2 재귀적 청킹 (Recursive Chunking)
문서 구조를 고려하여 계층적으로 청킹:
from langchain_text_splitters import MarkdownHeaderTextSplitter
def recursive_chunking(markdown_text):
headers_to_split_on = [
("#", "Header 1"),
("##", "Header 2"),
("###", "Header 3"),
]
splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=headers_to_split_on
)
return splitter.split_text(markdown_text)
2.3 에이전트 기반 청킹 (Agentic Chunking)
LLM을 활용한 자동 청킹 전략:
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
def agentic_chunking(text, llm):
chunk_prompt = PromptTemplate.from_template("""
다음 문서를 의미 단위로 분할해주세요:
{document}
각 청크는 100-200단어 범위 내에 있도록 하며,
문맥의 연속성을 유지해야 합니다.
""")
response = llm.invoke(chunk_prompt.format(document=text))
return response.content.split('\n')
3. 임베딩 모델 선택 및 비교
3.1 모델 성능 비교
import numpy as np
from sentence_transformers import SentenceTransformer
from langchain_openai import OpenAIEmbeddings
def compare_embeddings(texts, models):
results = {}
for name, model in models.items():
if name == "openai":
embeddings = model.embed_documents(texts)
else:
embeddings = model.encode(texts)
# 벡터 정규화
normalized = np.array(embeddings) / np.linalg.norm(embeddings, axis=1, keepdims=True)
results[name] = normalized
return results
# 사용 예시
models = {
"sentence-transformers": SentenceTransformer('all-MiniLM-L6-v2'),
"openai": OpenAIEmbeddings()
}
texts = ["대한민국의 수도는 서울입니다.", "일본의 수도는 도쿄입니다."]
embeddings = compare_embeddings(texts, models)
3.2 성능 기준
| 모델 | 속도 | 정확도 | 비용 | 추천 사용 사례 |
|---|---|---|---|---|
| OpenAI Ada | 빠름 | 높음 | 높음 | 실시간 응답 필요 |
| Sentence-BERT | 중간 | 높음 | 저렴 | 로컬 환경 |
| BGE-M3 | 중간 | 중간 | 저렴 | 대용량 데이터 |
4. 벡터 데이터베이스 비교
4.1 Chroma
가장 간단한 로컬 벡터 저장소:
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
# Chroma 초기화
vectorstore = Chroma(
collection_name="rag_collection",
persist_directory="./chroma_db",
embedding_function=OpenAIEmbeddings()
)
# 문서 추가
texts = ["문서1 내용", "문서2 내용"]
vectorstore.add_texts(texts)
# 검색
docs = vectorstore.similarity_search("검색어", k=3)
4.2 Qdrant
고성능 분산 벡터 저장소:
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue
client = QdrantClient(path="./qdrant_db")
# 벡터 검색
results = client.search(
collection_name="rag_collection",
query_vector=[0.1, 0.2, 0.3], # 임베딩 벡터
limit=3
)
4.3 pgvector
PostgreSQL 확장으로 데이터베이스 통합:
from langchain_postgres import PGVector
from sqlalchemy import create_engine
engine = create_engine("postgresql://user:pass@localhost/db")
vectorstore = PGVector(
connection=engine,
embedding_function=OpenAIEmbeddings(),
collection_name="rag_collection"
)
5. 전체 RAG 파이프라인 구현
python
import asyncio
from typing import List, Dict
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
class CompleteRAGPipeline:
def __init__(self, db_path: str = "./chroma_db"):
self.embeddings = OpenAIEmbeddings()
self.llm = ChatOpenAI(model="gpt-4-turbo")
self.vectorstore = Chroma(
persist_directory=db_path,
embedding_function=self.embeddings
)
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
def add_documents(self, documents: List[str]):
"""문서 추가"""
texts = self.text_splitter.split_text("\n".join(documents))
self.vectorstore.add_texts(texts)
return len(texts)
def search_and_generate(self, query: str, k: int = 3) -> Dict:
"""검색 및 생성"""
# 검색
docs = self.vectorstore.similarity_search(query, k=k)
# 컨텍스트 생성
context = "\n".join([doc.page_content for doc in docs])
# 프롬프트 생성
prompt = PromptTemplate.from_template("""
당신은 전문 정보 검색 전문가입니다.
질문: {query}
제공된 정보:
{context}
위 정보를 바탕으로 답변해주세요.
""")
# 생성
response = self.llm.invoke(prompt.format(query=query, context=context))
return {
"query": query,
"context": context,
"answer": response.content,
"retrieved_docs": [doc.metadata for doc in docs]
}
# 사용 예시
pipeline = CompleteRAGPipeline()
pipeline.add_documents(["문서 내용 1", "문서 내용 2"])
result = pipeline.search_and_generate
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($7)
Top comments (0)