RAG 시스템 실전 구축 (v43)
1. RAG 기초 개념: 검색 → 보강 → 생성 루프
RAG(Retrieval-Augmented Generation)는 대규모 언어 모델(LLM)과 외부 지식 베이스를 통합하여 정확하고 최신 정보를 기반으로 답변을 생성하는 아키텍처입니다.
# 기본 RAG 루프 구현
class BasicRAG:
def __init__(self, retriever, generator):
self.retriever = retriever
self.generator = generator
def generate(self, query):
# 1. 검색: 질의와 유사한 문서 찾기
retrieved_docs = self.retriever.retrieve(query)
# 2. 보강: 문서와 질의를 결합하여 프롬프트 생성
prompt = self._augment_prompt(query, retrieved_docs)
# 3. 생성: LLM이 답변 생성
response = self.generator.generate(prompt)
return response
def _augment_prompt(self, query, docs):
context = "\n\n".join([doc.content for doc in docs])
return f"질의: {query}\n\n문맥: {context}"
2. 청킹 전략: 의미적, 재귀적, 에이전트 기반
효과적인 청킹은 RAG 성능의 핵심입니다.
# 의미적 청킹 (Semantic Chunking)
from sentence_transformers import SentenceTransformer
import numpy as np
from sklearn.cluster import KMeans
class SemanticChunker:
def __init__(self, model_name="all-MiniLM-L6-v2"):
self.model = SentenceTransformer(model_name)
def chunk_semantic(self, text, max_chunk_size=512):
# 텍스트를 문장 단위로 분할
sentences = text.split('. ')
embeddings = self.model.encode(sentences)
# 문장 간 유사도를 기반으로 클러스터링
kmeans = KMeans(n_clusters=max(1, len(sentences)//3))
kmeans.fit(embeddings)
clusters = {}
for i, label in enumerate(kmeans.labels_):
if label not in clusters:
clusters[label] = []
clusters[label].append((sentences[i], embeddings[i]))
# 클러스터별로 청크 생성
chunks = []
for cluster in clusters.values():
chunk_text = '. '.join([sent for sent, _ in cluster])
if len(chunk_text) > max_chunk_size:
# 너무 긴 청크는 다시 분할
chunks.extend(self._split_long_chunk(chunk_text, max_chunk_size))
else:
chunks.append(chunk_text)
return chunks
# 재귀적 청킹 (Recursive Chunking)
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]
# 문장 기반 정제
sentences = chunk.split('. ')
if len(sentences) > 1:
chunk = '. '.join(sentences[:-1]) + '.'
chunks.append(chunk)
start = end - self.overlap
return chunks
# 에이전트 기반 청킹
class AgentBasedChunker:
def __init__(self, chunk_size=512):
self.chunk_size = chunk_size
def chunk_with_agent(self, text):
# 텍스트 구조 분석
sections = self._analyze_structure(text)
chunks = []
for section in sections:
if len(section) > self.chunk_size:
# 재귀적 청킹
chunks.extend(self._recursive_chunk(section))
else:
chunks.append(section)
return chunks
def _analyze_structure(self, text):
# 헤더와 본문 구분
import re
sections = re.split(r'(\n\n|\n)', text)
return [s for s in sections if s.strip()]
def _recursive_chunk(self, text):
if len(text) <= self.chunk_size:
return [text]
return [text[i:i+self.chunk_size]
for i in range(0, len(text), self.chunk_size)]
3. 임베딩 모델 선택 및 비교
# 임베딩 모델 비교 클래스
class EmbeddingBenchmark:
def __init__(self):
self.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'),
'gte-small': SentenceTransformer('thenlper/gte-small')
}
def benchmark_models(self, test_queries, test_documents):
results = {}
for name, model in self.models.items():
start_time = time.time()
# 쿼리 임베딩
query_embeddings = model.encode(test_queries)
# 문서 임베딩
doc_embeddings = model.encode(test_documents)
# 유사도 계산
similarities = cosine_similarity(query_embeddings, doc_embeddings)
end_time = time.time()
results[name] = {
'latency': end_time - start_time,
'similarity_matrix': similarities,
'model': model
}
return results
# 사용 예시
benchmark = EmbeddingBenchmark()
results = benchmark.benchmark_models(
test_queries=["인공지능의 미래"],
test_documents=["AI는 인간의 삶을 혁신한다", "머신러닝은 데이터 기반으로 학습한다"]
)
4. 벡터 데이터베이스 비교
# 각 벡터 DB의 연결 및 쿼리
class VectorDBComparison:
def __init__(self):
# Chroma
self.chroma_client = chromadb.Client()
self.chroma_collection = self.chroma_client.get_or_create_collection("chroma_collection")
# Qdrant
self.qdrant_client = QdrantClient(path="./qdrant_storage")
self.qdrant_collection = "qdrant_collection"
# pgvector
self.pg_connection = psycopg2.connect(
host="localhost",
database="rag_db",
user="postgres",
password="password"
)
def chroma_search(self, query_vector, top_k=5):
results = self.chroma_collection.query(
query_embeddings=[query_vector],
n_results=top_k
)
return results
def qdrant_search(self, query_vector, top_k=5):
results = self.qdrant_client.search(
collection_name=self.qdrant_collection,
query_vector=query_vector,
limit=top_k
)
return results
def pgvector_search(self, query_vector, top_k=5):
cursor = self.pg_connection.cursor()
cursor.execute("""
SELECT id, content,
1 - (embedding <=> %s) as similarity
FROM documents
ORDER BY similarity DESC
LIMIT %s
""", (query_vector, top_k))
return cursor.fetchall()
# 성능 비교
def compare_vector_dbs():
db_comparison = VectorDBComparison()
# 테스트 데이터 생성
test_vector = [0.1] * 100
# 시간 측정
import time
# Chroma
start = time.time()
chroma_results = db_comparison.chroma_search(test_vector)
chroma_time = time.time() - start
# Qdrant
start = time.time()
qdrant_results = db_comparison.qdrant_search(test_vector)
qdrant_time = time.time() - start
print(f"Chroma: {chroma_time:.4f}s")
print(f"Qdrant: {qdrant_time:.4f}s")
5. 전체 RAG 파이프라인 코드
python
# 완전한 RAG 시스템 구현
import chromadb
from sentence_transformers import SentenceTransformer
import numpy as np
from typing import List, Dict, Any
import time
class CompleteRAGPipeline:
def __init__(self,
embedding_model="all-MiniLM-L6-v2",
vector_db_path="./chroma_db"):
# 임베딩 모델 초기화
self.embedding_model = SentenceTransformer(embedding_model)
# 벡터 DB 초기화
self.chroma_client = chromadb.Client()
self.collection = self.chroma_client.get_or_create_collection("rag_documents")
# 문서 ID 추적
self.document_ids = []
def add_documents(self, documents: List[Dict[str, str]]):
"""문서 추가"""
texts = [doc['content'] for doc in documents]
ids = [f"doc_{i}" for i in range(len(documents
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($7)
Top comments (0)