RAG 시스템 실전 구축 (v21)
1. RAG 기초 개념
Retrieval-Augmented Generation (RAG)은 검색 기반 생성 시스템으로, LLM의 지식 범위를 확장하는 데 효과적입니다. 핵심 루프는 다음과 같습니다:
입력 질의 → 검색기 → 문서 조각 → 증강된 프롬프트 → LLM 생성 → 출력
RAG 시스템은 3가지 주요 구성 요소로 이루어집니다:
- Retriever: 관련 문서 찾기
- Generator: 증강된 프롬프트로 생성
- Embedding: 문서 및 질의 벡터화
2. 청킹 전략
2.1 의미적 청킹
from sentence_transformers import SentenceTransformer
from sklearn.cluster import KMeans
import numpy as np
class SemanticChunker:
def __init__(self, model_name='all-MiniLM-L6-v2'):
self.model = SentenceTransformer(model_name)
def chunk_by_semantic(self, text, threshold=0.7):
sentences = text.split('. ')
embeddings = self.model.encode(sentences)
# K-means 클러스터링으로 의미적 그룹 생성
kmeans = KMeans(n_clusters=max(1, len(sentences)//3))
kmeans.fit(embeddings)
clusters = {}
for i, cluster_id in enumerate(kmeans.labels_):
if cluster_id not in clusters:
clusters[cluster_id] = []
clusters[cluster_id].append(sentences[i])
return [' '.join(cluster) for cluster in clusters.values()]
2.2 재귀적 청킹
import re
class RecursiveChunker:
def __init__(self, chunk_size=512, overlap=50):
self.chunk_size = chunk_size
self.overlap = overlap
def chunk_recursive(self, text):
chunks = []
start = 0
while start < len(text):
end = min(start + self.chunk_size, len(text))
chunk = text[start:end]
# 중간에 문장 분리점 찾기
if end < len(text):
separator = chunk.rfind('. ')
if separator > self.chunk_size * 0.3:
end = start + separator + 1
chunk = text[start:end]
chunks.append(chunk)
start = max(0, end - self.overlap)
return chunks
3. 임베딩 모델 선택
3.1 모델 비교 테스트
from sentence_transformers import SentenceTransformer
import time
def compare_embedding_models():
models = {
'all-MiniLM-L6-v2': 'fast',
'all-mpnet-base-v2': 'medium',
'multi-qa-MiniLM-L6-v2': 'fast',
'paraphrase-multilingual-MiniLM-v2': 'fast'
}
test_sentences = [
"The quick brown fox jumps over the lazy dog.",
"Machine learning is a subset of artificial intelligence.",
"Natural language processing enables computers to understand text."
]
results = {}
for model_name, speed in models.items():
model = SentenceTransformer(model_name)
# 성능 측정
start_time = time.time()
embeddings = model.encode(test_sentences)
end_time = time.time()
results[model_name] = {
'speed': end_time - start_time,
'dimensions': len(embeddings[0]),
'speed_category': speed
}
return results
# 사용 예시
model_results = compare_embedding_models()
print("모델 성능 비교:")
for name, metrics in model_results.items():
print(f"{name}: {metrics['speed']:.2f}s, {metrics['dimensions']}차원")
4. 벡터 데이터베이스 비교
4.1 Chroma vs Qdrant vs pgvector
import chromadb
from qdrant_client import QdrantClient
import psycopg2
import numpy as np
class VectorDBComparison:
def __init__(self):
# Chroma
self.chroma_client = chromadb.Client()
self.chroma_collection = self.chroma_client.get_or_create_collection(
name="rag_test"
)
# Qdrant
self.qdrant_client = QdrantClient(host='localhost', port=6333)
# pgvector (PostgreSQL)
self.pg_conn = psycopg2.connect(
host="localhost",
database="rag_db",
user="user",
password="password"
)
def chroma_insert(self, documents, embeddings):
self.chroma_collection.add(
documents=documents,
embeddings=embeddings,
ids=[f"doc_{i}" for i in range(len(documents))]
)
def qdrant_insert(self, documents, embeddings):
self.qdrant_client.upsert(
collection_name="rag_test",
points=[
{
"id": i,
"vector": emb.tolist(),
"payload": {"text": doc}
}
for i, (doc, emb) in enumerate(zip(documents, embeddings))
]
)
def pgvector_insert(self, documents, embeddings):
with self.pg_conn.cursor() as cursor:
for i, (doc, emb) in enumerate(zip(documents, embeddings)):
cursor.execute(
"INSERT INTO documents (id, content, embedding) VALUES (%s, %s, %s)",
(i, doc, np.array(emb).tobytes())
)
self.pg_conn.commit()
5. 전체 RAG 파이프라인 코드
import chromadb
from sentence_transformers import SentenceTransformer
from typing import List, Dict
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
class SimpleRAGPipeline:
def __init__(self, chunk_size: int = 512):
self.chunker = RecursiveChunker(chunk_size=chunk_size)
self.embedder = SentenceTransformer('all-MiniLM-L6-v2')
self.chroma_client = chromadb.Client()
self.collection = self.chroma_client.get_or_create_collection(
name="documents"
)
def add_documents(self, documents: List[str]):
"""문서 추가 및 임베딩"""
all_chunks = []
all_texts = []
# 문서 청킹
for doc in documents:
chunks = self.chunker.chunk_recursive(doc)
all_chunks.extend(chunks)
all_texts.extend([doc] * len(chunks))
# 임베딩 생성
embeddings = self.embedder.encode(all_chunks)
# Chroma에 저장
self.collection.add(
documents=all_chunks,
embeddings=embeddings.tolist(),
ids=[f"chunk_{i}" for i in range(len(all_chunks))]
)
def retrieve(self, query: str, top_k: int = 3) -> List[Dict]:
"""검색"""
query_embedding = self.embedder.encode([query])
results = self.collection.query(
query_embeddings=query_embedding.tolist(),
n_results=top_k
)
return [
{
'text': text,
'score': score
}
for text, score in zip(results['documents'][0], results['distances'][0])
]
def generate_response(self, query: str, retrieved_docs: List[Dict]) -> str:
"""응답 생성"""
context = "\n".join([doc['text'] for doc in retrieved_docs])
prompt = f"""
주어진 컨텍스트를 기반으로 질문에 답변해주세요.
컨텍스트:
{context}
질문: {query}
답변:
"""
# 간단한 LLM 호출 예시 (실제 구현에서는 실제 LLM 사용)
return f"검색된 컨텍스트 기반 답변: {query}에 대한 정보는 {len(retrieved_docs)}개 문서에 포함되어 있습니다."
# 사용 예시
rag = SimpleRAGPipeline()
# 문서 추가
documents = [
"Python은 고급 프로그래밍 언어로, 간결하고 읽기 쉬운 문법을 갖추고 있습니다.",
"Django는 파이썬 기반 웹 프레임워크로, 빠르고 안전한 웹 개발을 지원합니다.",
"React는 JavaScript 라이브러리로, 사용자 인터페이스 개발에 최적화되어 있습니다."
]
rag.add_documents(documents)
# 검색 및 응답 생성
query = "Python 프로그래밍 언어의 특징은?"
retrieved = rag.retrieve(query)
response = rag.generate_response(query, retrieved)
print(f"질의: {query}")
print(f"응답: {response}")
6. 고급 기능
6.1 질의 변환
python
class QueryTransformer:
def __init__(self):
self.embedder = SentenceTransformer('all-MiniLM-L6-v2')
def transform_query(self, original_query: str) -> List[str]:
"""질의 변환
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($7)