RAG 시스템 실전 구축 (v37)
실제로 구축할 수 있는 RAG 시스템 구현 가이드
1. RAG 기초 개념: 검색 → 보완 → 생성 루프
RAG(Retrieval-Augmented Generation)는 대규모 언어 모델(LLM)이 외부 지식을 활용하여 정확한 답변을 생성할 수 있도록 하는 아키텍처입니다. 이 루프는 세 가지 주요 단계로 구성됩니다:
# RAG 루프 구조
class RAGPipeline:
def __init__(self):
self.retriever = None
self.generator = None
def process_query(self, query):
# 1. 검색: 쿼리와 유사한 문서 검색
retrieved_docs = self.retriever.retrieve(query)
# 2. 보완: 검색된 문서를 쿼리와 결합
augmented_context = self.augment_context(query, retrieved_docs)
# 3. 생성: LLM이 보완된 문맥을 기반으로 답변 생성
answer = self.generator.generate(query, augmented_context)
return answer
2. 청킹 전략: 의미적, 재귀적, 에이전트 기반
청킹은 문서를 검색 가능한 단위로 나누는 작업입니다:
from langchain.text_splitter import (
RecursiveCharacterTextSplitter,
CharacterTextSplitter
)
import tiktoken
class ChunkingStrategies:
@staticmethod
def semantic_chunking(documents, model="text-embedding-3-small"):
"""의미적 청킹: 의미 단위로 문서 분할"""
# 의미적 단위를 찾기 위한 임베딩 기반 분할
from sentence_transformers import SentenceTransformer
encoder = SentenceTransformer(model)
# 문서를 의미 단위로 분할
# 예시: 문장 단위 분할
sentences = doc.split('.')
embeddings = encoder.encode(sentences)
# 임베딩 간 유사도를 기준으로 청킹
return sentences
@staticmethod
def recursive_chunking(documents, chunk_size=512, chunk_overlap=50):
"""재귀적 청킹: 계층적 구조로 문서 분할"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", " ", ""]
)
return text_splitter.split_text(documents)
@staticmethod
def agentic_chunking(documents, chunk_size=512):
"""에이전트 기반 청킹: 문서 구조를 고려한 청킹"""
# 문서 헤더, 본문, 결론 구분하여 청킹
sections = documents.split('\n\n')
chunks = []
for section in sections:
if len(section) > chunk_size:
chunks.extend(
[section[i:i+chunk_size]
for i in range(0, len(section), chunk_size)]
)
else:
chunks.append(section)
return chunks
# 사용 예시
strategy = ChunkingStrategies()
documents = "..."
chunks = strategy.recursive_chunking(documents, chunk_size=512)
3. 임베딩 모델 선택과 비교
임베딩 모델은 문서와 쿼리를 벡터 공간으로 변환하는 핵심 요소입니다:
from sentence_transformers import SentenceTransformer
from transformers import AutoTokenizer, AutoModel
import numpy as np
import time
class EmbeddingComparison:
def __init__(self):
self.models = {
'all-MiniLM-L6-v2': SentenceTransformer('all-MiniLM-L6-v2'),
'all-mpnet-base-v2': SentenceTransformer('all-mpnet-base-v2'),
'text-embedding-3-small': SentenceTransformer('text-embedding-3-small'),
'gte-small': SentenceTransformer('thenlper/gte-small')
}
def compare_models(self, texts):
"""모델별 성능 비교"""
results = {}
for model_name, model in self.models.items():
start_time = time.time()
embeddings = model.encode(texts)
end_time = time.time()
results[model_name] = {
'embedding_size': embeddings.shape,
'processing_time': end_time - start_time,
'memory_usage': embeddings.nbytes
}
return results
def benchmark_embeddings(self, query, corpus, top_k=5):
"""임베딩 성능 벤치마킹"""
# 모든 모델로 벡터 생성
vectors = {}
for name, model in self.models.items():
vectors[name] = model.encode(corpus)
# 쿼리 벡터 생성
query_vector = self.models['all-MiniLM-L6-v2'].encode([query])
# 유사도 계산
similarities = {}
for name, embeddings in vectors.items():
similarity = np.dot(embeddings, query_vector.T).flatten()
top_indices = np.argsort(similarity)[-top_k:][::-1]
similarities[name] = top_indices.tolist()
return similarities
# 사용 예시
benchmark = EmbeddingComparison()
texts = ["문서 1", "문서 2", "문서 3"]
results = benchmark.compare_models(texts)
4. 벡터 데이터베이스 비교
다양한 벡터 데이터베이스의 특징과 성능 비교:
# Chroma 벡터 데이터베이스
import chromadb
from chromadb import Client
class ChromaVectorDB:
def __init__(self, persist_directory="./chroma_db"):
self.client = Client()
self.collection = self.client.get_or_create_collection(
name="rag_collection",
metadata={"hnsw:space": "cosine"}
)
def add_documents(self, documents, embeddings, ids):
self.collection.add(
documents=documents,
embeddings=embeddings,
ids=ids
)
def search(self, query_embedding, n_results=5):
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=n_results
)
return results
# 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)
def create_collection(self, collection_name, vector_size=384):
self.client.recreate_collection(
collection_name=collection_name,
vectors_config={"size": vector_size, "distance": "Cosine"}
)
def search(self, query_vector, collection_name, limit=5):
results = self.client.search(
collection_name=collection_name,
query_vector=query_vector,
limit=limit
)
return results
# pgvector 예시
import psycopg2
from psycopg2.extras import Json
class PGVectorDB:
def __init__(self, connection_string):
self.conn = psycopg2.connect(connection_string)
def create_table(self):
with self.conn.cursor() as cur:
cur.execute("""
CREATE TABLE IF NOT EXISTS documents (
id UUID PRIMARY KEY,
content TEXT,
embedding VECTOR(384),
metadata JSONB
)
""")
cur.execute("CREATE INDEX IF NOT EXISTS idx_embedding ON documents USING ivfflat (embedding vector_cosine_ops)")
def insert_document(self, doc_id, content, embedding, metadata):
with self.conn.cursor() as cur:
cur.execute("""
INSERT INTO documents (id, content, embedding, metadata)
VALUES (%s, %s, %s, %s)
ON CONFLICT (id) DO UPDATE SET
content = EXCLUDED.content,
embedding = EXCLUDED.embedding,
metadata = EXCLUDED.metadata
""", (doc_id, content, embedding, Json(metadata)))
def search(self, query_embedding, limit=5):
with self.conn.cursor() as cur:
cur.execute("""
SELECT content, metadata, embedding <-> %s as distance
FROM documents
ORDER BY distance
LIMIT %s
""", (query_embedding, limit))
return cur.fetchall()
5. 전체 RAG 파이프라인 코드
실제 구현 예시:
python
import os
from sentence_transformers import SentenceTransformer
import numpy as np
from typing import List, Dict, Any
class FullRAGPipeline:
def __init__(self,
embedding_model_name="all-MiniLM-L6-v2",
vector_db_type="chroma",
local_model_path=None):
# 임베딩 모델 로드
self.embedding_model = SentenceTransformer(embedding_model_name)
# 벡터 데이터베이스 초기화
if vector_db_type == "chroma":
self.vector_db = ChromaVectorDB()
elif vector_db_type == "qdrant":
self.vector_db = QdrantVectorDB()
else:
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($7)
Top comments (0)