RAG 시스템 실전 구축 (v39)
실제 생산 환경에서 RAG 시스템을 구축하고 운영하는 실전 가이드
1. RAG 기본 개념: 검색 → 보완 → 생성 루프
RAG(Retrieval-Augmented Generation)는 검색 기반의 정보를 활용하여 LLM이 더 정확한 응답을 생성하는 아키텍처입니다.
사용자 질문 → 검색 엔진 → 관련 문서 → LLM 보완 → 생성된 응답
기본적인 RAG 루프 구조는 다음과 같습니다:
class BasicRAGPipeline:
def __init__(self, embedding_model, vector_db):
self.embedding_model = embedding_model
self.vector_db = vector_db
def retrieve(self, query, k=5):
query_embedding = self.embedding_model.encode([query])[0]
return self.vector_db.search(query_embedding, k=k)
def generate(self, query, context):
prompt = f"질문: {query}\n문맥: {context}\n답변:"
return self.llm.generate(prompt)
def process(self, query):
context = self.retrieve(query)
return self.generate(query, context)
2. 청킹 전략: 의미, 재귀, 에이전트 기반
청킹 전략은 문서를 적절한 크기로 분할하여 임베딩 및 검색에 최적화합니다:
import tiktoken
from typing import List
class ChunkingStrategies:
@staticmethod
def semantic_chunking(documents: List[str], chunk_size: int = 500) -> List[str]:
"""의미 기반 청킹"""
# 문서 내에서 의미 단위를 기준으로 청킹
chunks = []
for doc in documents:
sentences = doc.split('. ')
current_chunk = ""
for sentence in sentences:
if len(current_chunk) + len(sentence) < chunk_size:
current_chunk += sentence + ". "
else:
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = sentence + ". "
if current_chunk:
chunks.append(current_chunk.strip())
return chunks
@staticmethod
def recursive_chunking(documents: List[str], chunk_size: int = 500) -> List[str]:
"""재귀적 청킹 - 토큰 기준"""
tokenizer = tiktoken.get_encoding("cl100k_base")
chunks = []
for doc in documents:
tokens = tokenizer.encode(doc)
for i in range(0, len(tokens), chunk_size):
chunk_tokens = tokens[i:i+chunk_size]
chunk_text = tokenizer.decode(chunk_tokens)
chunks.append(chunk_text)
return chunks
# 사용 예시
strategies = ChunkingStrategies()
docs = ["문서 내용..."]
semantic_chunks = strategies.semantic_chunking(docs)
recursive_chunks = strategies.recursive_chunking(docs)
3. 임베딩 모델 선택과 비교
임베딩 품질은 RAG 성능에 직접적인 영향을 미칩니다:
from sentence_transformers import SentenceTransformer
import numpy as np
class EmbeddingModelComparison:
def __init__(self):
self.models = {
'all-MiniLM-L6-v2': SentenceTransformer('all-MiniLM-L6-v2'),
'paraphrase-multilingual-MiniLM-v2': SentenceTransformer('paraphrase-multilingual-MiniLM-v2'),
'gte-small': SentenceTransformer('sentence-transformers/gte-small')
}
def compare_embeddings(self, texts: List[str], model_name: str) -> np.ndarray:
model = self.models[model_name]
return model.encode(texts)
def benchmark_models(self, test_queries: List[str], test_docs: List[str]) -> dict:
results = {}
for name, model in self.models.items():
# 성능 측정
embeddings = model.encode(test_queries + test_docs)
results[name] = {
'embedding_time': len(embeddings), # 실제 시간 측정 필요
'dimension': embeddings.shape[1]
}
return results
# 비교 실행
comparator = EmbeddingModelComparison()
benchmark_results = comparator.benchmark_models(["쿼리1", "쿼리2"], ["문서1", "문서2"])
4. 벡터 데이터베이스 비교: Chroma vs Qdrant vs pgvector vs Milvus
| 특성 | Chroma | Qdrant | pgvector | Milvus |
|---|---|---|---|---|
| 설치 간편성 | ★★★★★ | ★★★★☆ | ★★★★☆ | ★★★☆☆ |
| 성능 | ★★★★☆ | ★★★★★ | ★★★★☆ | ★★★★★ |
| 비용 | ★★★★☆ | ★★★★☆ | ★★★★★ | ★★★★☆ |
| 확장성 | ★★★☆☆ | ★★★★☆ | ★★★★☆ | ★★★★★ |
# Chroma 벡터 DB 구현
from chromadb import Client
from chromadb.config import Settings
class ChromaVectorDB:
def __init__(self, collection_name="rag_collection"):
self.client = Client(Settings(chroma_db_impl="duckdb", persist_directory="./chroma_db"))
self.collection = self.client.get_or_create_collection(collection_name)
def add_documents(self, documents: List[str], embeddings: List[List[float]]):
ids = [str(i) for i in range(len(documents))]
self.collection.add(
ids=ids,
embeddings=embeddings,
documents=documents
)
def search(self, query_embedding: List[float], k: int = 5):
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=k
)
return results['documents'][0]
# Qdrant 구현
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue
class QdrantVectorDB:
def __init__(self, host="localhost", port=6333):
self.client = QdrantClient(host=host, port=port)
self.collection_name = "rag_collection"
def create_collection(self, vector_size: int):
self.client.recreate_collection(
collection_name=self.collection_name,
vectors_config={"size": vector_size, "distance": "Cosine"}
)
def search(self, query_embedding: List[float], k: int = 5):
results = self.client.search(
collection_name=self.collection_name,
query_vector=query_embedding,
limit=k
)
return [hit.payload['text'] for hit in results]
5. 전체 RAG 파이프라인 코드
python
import numpy as np
from sentence_transformers import SentenceTransformer
from chromadb import Client
from chromadb.config import Settings
class RAGPipeline:
def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
# 1. 임베딩 모델 초기화
self.embedding_model = SentenceTransformer(model_name)
# 2. 벡터 DB 초기화
self.client = Client(Settings(chroma_db_impl="duckdb", persist_directory="./rag_db"))
self.collection = self.client.get_or_create_collection("documents")
# 3. LLM 초기화 (예시: HuggingFace)
from transformers import pipeline
self.llm = pipeline("text-generation",
model="gpt2",
max_length=200)
def add_documents(self, documents: List[str]):
"""문서 추가"""
embeddings = self.embedding_model.encode(documents)
ids = [str(i) for i in range(len(documents))]
self.collection.add(
ids=ids,
embeddings=embeddings.tolist(),
documents=documents
)
return f"Added {len(documents)} documents"
def retrieve(self, query: str, k: int = 5) -> List[str]:
"""문서 검색"""
query_embedding = self.embedding_model.encode([query])[0]
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=k
)
return results['documents'][0]
def generate_response(self, query: str, context: List[str]) -> str:
"""응답 생성"""
context_text = "\n".join(context)
prompt = f"""
질문: {query}
문맥: {context_text}
답변:"""
response = self.llm(prompt, return_full_text=False)[0]['generated_text']
return response.strip()
def process_query(self, query: str) -> str:
"""전체 파이프라인 실행"""
context = self.retrieve(query)
return self.generate_response(query, context)
# 사용 예시
pipeline = RAGPipeline()
documents = [
"Python은 고급 프로그래밍 언어입니다. 객체 지향 프로그래밍을 지원합니다.",
"JavaScript는 웹 브라우저에서 실행되는 스크립트 언어입니다.",
"Machine Learning은 데이터를 기반으로 학습하는 알고리즘입니다."
]
pipeline.add_documents(doc
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($7)