RAG 시스템 실전 구축 (v15)
개요
이 가이드는 실전에서 사용 가능한 RAG(Retrieval-Augmented Generation) 시스템을 구축하는 데 필요한 모든 요소를 다룹니다. 특히 ML 엔지니어와 백엔드 개발자들이 실제 프로젝트에서 적용할 수 있도록 구체적인 코드와 전략을 제시합니다.
1. RAG 기본 개념
RAG 시스템은 정보 검색과 생성을 통합한 아키텍처입니다. 핵심 루프는 다음과 같습니다:
- 검색(Retrieval): 사용자 쿼리와 유사한 문서 찾기
- 보완(Augmentation): 검색된 문서를 프롬프트에 통합
- 생성(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('. ')
embeddings = model.encode(sentences)
# 유사도 기반 클러스터링
kmeans = KMeans(n_clusters=max(1, len(sentences)//3))
kmeans.fit(embeddings)
chunks = []
for i in range(len(sentences)):
chunks.append((sentences[i], kmeans.labels_[i]))
return chunks
# 사용 예시
model = SentenceTransformer('all-MiniLM-L6-v2')
text = "RAG 시스템은 검색과 생성을 통합합니다. 이 시스템은 대규모 언어 모델을 위한 전략입니다."
chunks = semantic_chunking(text, model)
2.2 재귀적 청킹 (Recursive Chunking)
def recursive_chunking(text, max_chunk_size=512, overlap=50):
chunks = []
start = 0
while start < len(text):
end = min(start + max_chunk_size, len(text))
chunk = text[start:end]
chunks.append(chunk)
start = end - overlap
return chunks
# 사용 예시
text = "긴 문서 내용..." * 100
chunks = recursive_chunking(text)
2.3 에이전트 기반 청킹 (Agentic Chunking)
def agentic_chunking(text, segmenter_model, min_chunk_size=100):
# 문서 구조 분석
segments = segmenter_model.analyze(text)
chunks = []
for segment in segments:
if len(segment) > min_chunk_size:
chunks.extend(recursive_chunking(segment, min_chunk_size))
else:
chunks.append(segment)
return chunks
3. 임베딩 모델 선택
3.1 모델 비교
from sentence_transformers import SentenceTransformer
import time
def benchmark_embedding_models(texts):
models = {
'all-MiniLM-L6-v2': 'all-MiniLM-L6-v2',
'all-mpnet-base-v2': 'all-mpnet-base-v2',
'sentence-t5-base': 'sentence-t5-base'
}
results = {}
for name, model_name in models.items():
model = SentenceTransformer(model_name)
start_time = time.time()
embeddings = model.encode(texts)
end_time = time.time()
results[name] = {
'time': end_time - start_time,
'size': len(embeddings),
'dimension': len(embeddings[0])
}
return results
# 성능 비교
texts = ["문서 1", "문서 2", "문서 3"] * 100
benchmarks = benchmark_embedding_models(texts)
print(benchmarks)
3.2 모델 성능 요약
- all-MiniLM-L6-v2: 384차원, 빠르고 효율적
- all-mpnet-base-v2: 768차원, 정확도 높음
- sentence-t5-base: 512차원, 다국어 지원
4. 벡터 데이터베이스 비교
4.1 Chroma
import chromadb
from chromadb.config import Settings
# Chroma 클라이언트 생성
client = chromadb.Client(Settings(
chroma_db_impl="duckdb",
persist_directory="./chroma_db"
))
# 컬렉션 생성
collection = client.get_or_create_collection("rag_collection")
# 문서 추가
collection.add(
documents=["문서 내용 1", "문서 내용 2"],
metadatas=[{"source": "file1"}, {"source": "file2"}],
ids=["doc1", "doc2"]
)
# 검색
results = collection.query(
query_texts=["검색어"],
n_results=3
)
4.2 Qdrant
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue
client = QdrantClient(path="./qdrant_db")
# 벡터 인덱스 생성
client.create_collection(
collection_name="rag_collection",
vectors_config={"size": 384, "distance": "Cosine"}
)
# 문서 삽입
client.upsert(
collection_name="rag_collection",
points=[
{
"id": 1,
"vector": [0.1, 0.2, 0.3],
"payload": {"text": "문서 내용", "source": "file1"}
}
]
)
# 검색
results = client.search(
collection_name="rag_collection",
query_vector=[0.1, 0.2, 0.3],
limit=3
)
4.3 pgvector
import psycopg2
from psycopg2.extras import Json
# PostgreSQL + pgvector 설정
conn = psycopg2.connect(
host="localhost",
database="rag_db",
user="user",
password="password"
)
# 테이블 생성
cursor.execute("""
CREATE TABLE IF NOT EXISTS documents (
id SERIAL PRIMARY KEY,
content TEXT,
embedding VECTOR(384),
metadata JSONB
)
""")
# 문서 삽입
cursor.execute(
"INSERT INTO documents (content, embedding, metadata) VALUES (%s, %s, %s)",
("문서 내용", [0.1, 0.2, 0.3], {"source": "file1"})
)
5. 전체 RAG 파이프라인 구현
python
import numpy as np
from sentence_transformers import SentenceTransformer
import chromadb
from typing import List, Dict, Tuple
class RAGPipeline:
def __init__(self, embedding_model_name: str = "all-MiniLM-L6-v2"):
self.embedding_model = SentenceTransformer(embedding_model_name)
self.vector_db = chromadb.Client()
self.collection = self.vector_db.get_or_create_collection("rag_collection")
def add_documents(self, documents: List[Dict]):
"""문서 추가 및 임베딩"""
embeddings = self.embedding_model.encode([doc['content'] for doc in documents])
self.collection.add(
documents=[doc['content'] for doc in documents],
metadatas=[doc['metadata'] for doc in documents],
ids=[doc['id'] for doc in documents],
embeddings=embeddings
)
def retrieve(self, query: str, k: int = 3) -> List[Dict]:
"""쿼리 기반 검색"""
query_embedding = self.embedding_model.encode([query])
results = self.collection.query(
query_embeddings=query_embedding,
n_results=k,
include=['documents', 'metadatas']
)
return [{
'content': doc,
'metadata': meta,
'score': score
} for doc, meta, score in zip(
results['documents'][0],
results['metadatas'][0],
results['distances'][0]
)]
def generate_response(self, query: str, retrieved_docs: List[Dict]) -> str:
"""LLM 응답 생성 (예시)"""
context = "\n".join([doc['content'] for doc in retrieved_docs])
prompt = f"질문: {query}\n참고 문서:\n{context}\n\n답변:"
# 실제 LLM 호출은 여기서 수행
# return llm.generate(prompt)
return f"질문에 대한 답변: {query}"
# 사용 예시
pipeline = RAGPipeline()
documents = [
{
'id': 'doc1',
'content': 'RAG 시스템은 검색과 생성을 통합합니다.',
'metadata': {'source': 'guide'}
},
{
'id': 'doc2',
'content': '임베딩 모델은 문서를 벡터로 변환합니다.',
'metadata': {'source': 'technical'}
}
]
pipeline.add
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($7)
Top comments (0)