RAG 시스템 실전 구축 (v3)
개요
이 가이드는 실전에서 RAG(Retrieval-Augmented Generation) 시스템을 구축하는 데 필요한 모든 것을 다룹니다. ML 엔지니어와 백엔드 개발자들이 데이터를 기반으로 의미 있는 질문에 답변할 수 있는 시스템을 구축할 때 실용적인 코드와 접근 방식을 제공합니다.
1. RAG 기본 개념
RAG는 검색 기반 생성 시스템으로, 두 가지 주요 단계를 거칩니다:
검색 (Retrieval)
문서에서 관련 정보를 찾는 과정
보강 (Augmentation)
검색된 정보를 프롬프트에 추가하여 LLM이 답변 생성에 활용
생성 (Generation)
LLM이 검색된 정보와 원본 질문을 기반으로 답변 생성
질문 → 검색 → 보강 → 생성 → 답변
2. 청킹 전략
2.1 의미적 청킹 (Semantic Chunking)
from sentence_transformers import SentenceTransformer
import numpy as np
from sklearn.cluster import KMeans
def semantic_chunking(text, model, threshold=0.7):
# 텍스트를 문장 단위로 분할
sentences = text.split('. ')
sentence_embeddings = model.encode(sentences)
# 클러스터링을 통한 의미적 청킹
kmeans = KMeans(n_clusters=min(len(sentences), 10))
kmeans.fit(sentence_embeddings)
# 같은 클러스터 내 문장들을 하나의 청크로 결합
chunks = []
for i in range(len(sentences)):
cluster = kmeans.labels_[i]
if cluster not in [c[0] for c in chunks]:
chunks.append((cluster, [sentences[i]]))
else:
for j, (c_cluster, c_sentences) in enumerate(chunks):
if c_cluster == cluster:
chunks[j] = (c_cluster, c_sentences + [sentences[i]])
break
return [chunk[1] for chunk in chunks]
# 사용 예시
model = SentenceTransformer('all-MiniLM-L6-v2')
text = "AI 기술은 빠르게 발전하고 있습니다. 인공지능은 많은 분야에서 활용됩니다. 특히 자연어 처리 분야에서는 큰 진전이 있습니다."
chunks = semantic_chunking(text, model)
2.2 재귀적 청킹 (Recursive Chunking)
import re
def recursive_chunking(text, chunk_size=512):
# 문단 단위 청킹
paragraphs = text.split('\n\n')
chunks = []
current_chunk = ""
for para in paragraphs:
if len(current_chunk) + len(para) < chunk_size:
current_chunk += para + "\n\n"
else:
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = para + "\n\n"
if current_chunk:
chunks.append(current_chunk.strip())
return chunks
2.3 에이전트 기반 청킹 (Agentic Chunking)
class AgenticChunker:
def __init__(self, model_name="gpt-3.5-turbo"):
self.model = model_name
def chunk_with_context(self, text, context_length=1000):
# 문맥을 고려한 청킹
sentences = re.split(r'[.!?]+', text)
chunks = []
current_chunk = ""
for sentence in sentences:
if len(current_chunk) + len(sentence) < context_length:
current_chunk += sentence + ". "
else:
if current_chunk.strip():
chunks.append(current_chunk.strip())
current_chunk = sentence + ". "
if current_chunk.strip():
chunks.append(current_chunk.strip())
return chunks
3. 임베딩 모델 선택
3.1 비교 테이블
| 모델 | 성능 | 속도 | 메모리 | 용도 |
|---|---|---|---|---|
| all-MiniLM-L6-v2 | 86.4 | 빠름 | 100MB | 일반 |
| all-mpnet-base-v2 | 88.2 | 중간 | 400MB | 정확도 |
| BGE-M3 | 90.1 | 중간 | 500MB | 다국어 |
| Sentence-BERT | 89.8 | 느림 | 700MB | 높은 정확도 |
3.2 성능 비교 코드
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import time
def evaluate_embedding_models():
models = {
'all-MiniLM-L6-v2': SentenceTransformer('all-MiniLM-L6-v2'),
'all-mpnet-base-v2': SentenceTransformer('all-mpnet-base-v2'),
'BGE-M3': SentenceTransformer('BAAI/bge-m3')
}
test_sentences = [
"Machine learning is a subset of artificial intelligence.",
"Deep learning uses neural networks with multiple layers.",
"Natural language processing deals with human language understanding."
]
results = {}
for name, model in models.items():
start_time = time.time()
embeddings = model.encode(test_sentences)
end_time = time.time()
# 유사도 계산
similarity = cosine_similarity([embeddings[0]], [embeddings[1]])[0][0]
results[name] = {
'time': end_time - start_time,
'similarity': similarity,
'memory': model._modules['0'].state_dict()['weight'].numel() * 4 / (1024**2) # MB
}
return results
# 성능 평가
perf_results = evaluate_embedding_models()
for model, metrics in perf_results.items():
print(f"{model}: Time={metrics['time']:.3f}s, Similarity={metrics['similarity']:.3f}")
4. 벡터 데이터베이스 비교
4.1 Chroma (가볍고 빠름)
import chromadb
from chromadb import Client
class ChromaVectorStore:
def __init__(self, collection_name="documents"):
self.client = Client()
self.collection = self.client.get_or_create_collection(collection_name)
def add_documents(self, documents, embeddings, metadatas=None):
self.collection.add(
documents=documents,
embeddings=embeddings,
metadatas=metadatas
)
def search(self, query_embedding, n_results=5):
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=n_results
)
return results
# 사용 예시
chroma_store = ChromaVectorStore()
# embeddings = model.encode(documents)
# chroma_store.add_documents(documents, embeddings)
4.2 Qdrant (고성능)
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue
class QdrantVectorStore:
def __init__(self, host="localhost", port=6333, collection_name="documents"):
self.client = QdrantClient(host=host, port=port)
self.collection_name = collection_name
# 컬렉션 생성
self.client.recreate_collection(
collection_name=self.collection_name,
vectors_config={"size": 384, "distance": "Cosine"}
)
def add_documents(self, documents, embeddings, metadatas=None):
points = [
{
"id": i,
"vector": embedding,
"payload": {"text": doc, "metadata": metadata if metadata else {}}
}
for i, (doc, embedding, metadata) in enumerate(zip(documents, embeddings, metadatas or [None]*len(documents)))
]
self.client.upsert(
collection_name=self.collection_name,
points=points
)
def search(self, query_embedding, n_results=5):
results = self.client.search(
collection_name=self.collection_name,
query_vector=query_embedding,
limit=n_results
)
return results
# 사용 예시
qdrant_store = QdrantVectorStore()
# qdrant_store.add_documents(documents, embeddings)
4.3 pgvector (SQL 기반)
python
import psycopg2
from psycopg2.extras import Json
import numpy as np
class PGVectorStore:
def __init__(self, connection_string):
self.conn = psycopg2.connect(connection_string)
self.create_table()
def create_table(self):
with self.conn.cursor() as cur:
cur.execute("""
CREATE TABLE IF NOT EXISTS document_embeddings (
id SERIAL PRIMARY KEY,
document TEXT,
embedding VECTOR(384),
metadata JSONB,
created_at TIMESTAMP DEFAULT NOW()
)
""")
cur.execute("CREATE INDEX IF NOT EXISTS idx_embedding ON document_embeddings USING ivfflat (embedding vector_cosine_ops)")
self.conn.commit()
def add_documents(self,
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($7)