RAG 시스템 실전 구축 (v47)

개요

RAG(Retrieval-Augmented Generation) 시스템은 LLM의 지식 범위를 확장하고, 특정 도메인 전문 지식을 통합하는 데 핵심적인 기술입니다. 이 가이드는 실제 개발 환경에서 RAG 시스템을 구축하는 실전 가이드를 제공하며, 특히 로컬 환경에서의 성능과 비용 효율성을 중심으로 다룹니다.

1. RAG 기본 구조

RAG는 세 가지 핵심 단계로 구성됩니다:

1. 검색(Retrieval): 주어진 쿼리와 관련된 문서 조각을 찾습니다.
2. 보강(Augmentation): 검색된 문서를 프롬프트에 추가합니다.
3. 생성(Generation): LLM이 보강된 프롬프트를 기반으로 응답을 생성합니다.

# RAG 기본 구조 구현
class BasicRAG:
    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])
        return self.vector_db.search(query_embedding, k)

    def generate(self, query, retrieved_docs):
        prompt = f"Query: {query}\n\nRelevant Docs:\n" + "\n\n".join(retrieved_docs)
        return self.llm.generate(prompt)

    def process(self, query):
        docs = self.retrieve(query)
        response = self.generate(query, docs)
        return response

2. Chunking 전략

문서를 적절한 크기로 나누는 전략은 RAG 성능에 결정적 영향을 미칩니다.

2.1 의미적 Chunking

from sentence_transformers import SentenceTransformer
import numpy as np

def semantic_chunking(documents, model, threshold=0.7):
    """의미적 기반으로 문서를 chunking"""
    embeddings = model.encode(documents)

    chunks = []
    current_chunk = []
    current_embedding = np.zeros(embeddings[0].shape)

    for i, (doc, embedding) in enumerate(zip(documents, embeddings)):
        # 이전 chunk와의 유사도 계산
        similarity = np.dot(current_embedding, embedding) / (
            np.linalg.norm(current_embedding) * np.linalg.norm(embedding)
        )

        if similarity < threshold:
            if current_chunk:
                chunks.append(' '.join(current_chunk))
            current_chunk = [doc]
            current_embedding = embedding
        else:
            current_chunk.append(doc)
            # 평균 임베딩 업데이트
            current_embedding = (current_embedding + embedding) / 2

    if current_chunk:
        chunks.append(' '.join(current_chunk))
    return chunks

2.2 Recursive Chunking

def recursive_chunking(text, max_chunk_size=500):
    """재귀적 chunking으로 문장 단위로 분할"""
    chunks = []
    sentences = text.split('. ')

    current_chunk = ""
    for sentence in sentences:
        if len(current_chunk) + len(sentence) < max_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

3. 임베딩 모델 선택 및 비교

# 임베딩 모델 비교
from sentence_transformers import SentenceTransformer
from transformers import AutoTokenizer, AutoModel
import torch

class EmbeddingComparison:
    def __init__(self):
        self.models = {
            'all-MiniLM-L6-v2': SentenceTransformer('all-MiniLM-L6-v2'),
            'paraphrase-multilingual-MiniLM-v2': SentenceTransformer('paraphrase-multilingual-MiniLM-v2'),
            'bge-small-en': SentenceTransformer('BAAI/bge-small-en'),
            'fast-sentence-bert': SentenceTransformer('sentence-transformers/paraphrase-multilingual-mpnet-base-v2')
        }

    def benchmark_models(self, test_sentences):
        results = {}
        for name, model in self.models.items():
            start_time = time.time()
            embeddings = model.encode(test_sentences)
            end_time = time.time()

            results[name] = {
                'time': end_time - start_time,
                'dimension': embeddings.shape[1],
                'memory': embeddings.nbytes
            }
        return results

# 사용 예시
comparison = EmbeddingComparison()
test_data = ["Hello world", "How are you?", "I am fine"]
results = comparison.benchmark_models(test_data)

4. Vector Database 비교

# Vector DB별 비교
class VectorDBComparison:
    def __init__(self):
        self.dbs = {
            'chroma': ChromaDB(),
            'qdrant': QdrantDB(),
            'pgvector': PostgreSQLVector(),
            'milvus': MilvusDB()
        }

    def test_insert_performance(self, documents, embeddings):
        results = {}
        for name, db in self.dbs.items():
            start_time = time.time()
            db.insert(documents, embeddings)
            end_time = time.time()
            results[name] = end_time - start_time
        return results

# ChromaDB 구현 예시
class ChromaDB:
    def __init__(self, path="./chroma_db"):
        import chromadb
        self.client = chromadb.Client(path)
        self.collection = self.client.get_or_create_collection("docs")

    def insert(self, documents, embeddings):
        self.collection.add(
            documents=documents,
            embeddings=embeddings,
            ids=[f"doc_{i}" for i in range(len(documents))]
        )

    def search(self, query_embedding, k=5):
        results = self.collection.query(
            query_embeddings=query_embedding,
            n_results=k
        )
        return results['documents'][0]

5. 전체 RAG 파이프라인 코드

import numpy as np
from sentence_transformers import SentenceTransformer
import faiss
import time
import pickle

class LocalRAGPipeline:
    def __init__(self, model_name='all-MiniLM-L6-v2'):
        self.embedding_model = SentenceTransformer(model_name)
        self.vector_index = None
        self.documents = []
        self.doc_ids = []

    def create_vector_index(self, documents):
        """문서 벡터 인덱스 생성"""
        # 문서 임베딩 생성
        embeddings = self.embedding_model.encode(documents)
        self.documents = documents

        # FAISS 인덱스 생성
        dimension = embeddings.shape[1]
        self.vector_index = faiss.IndexFlatIP(dimension)
        self.vector_index.add(np.array(embeddings, dtype=np.float32))

        # 문서 ID 매핑
        self.doc_ids = [f"doc_{i}" for i in range(len(documents))]

    def search_relevant_docs(self, query, k=5):
        """관련 문서 검색"""
        query_embedding = self.embedding_model.encode([query])
        distances, indices = self.vector_index.search(
            np.array(query_embedding, dtype=np.float32), k
        )

        relevant_docs = []
        for idx, dist in zip(indices[0], distances[0]):
            if idx < len(self.documents):
                relevant_docs.append({
                    'content': self.documents[idx],
                    'similarity': float(dist)
                })
        return relevant_docs

    def generate_response(self, query, relevant_docs):
        """LLM 응답 생성"""
        # 간단한 프롬프트 생성
        context = "\n\n".join([doc['content'] for doc in relevant_docs])
        prompt = f"""
        주어진 문서 내용을 바탕으로 질문에 답변해주세요.

        문서 내용:
        {context}

        질문: {query}
        """

        # 실제 LLM 호출은 여기서 수행 (예: local LLM)
        # 예시로 간단한 응답 생성
        return f"질문: {query}\n답변: 문서 내용을 바탕으로 답변을 생성했습니다."

    def process_query(self, query):
        """전체 쿼리 처리 파이프라인"""
        relevant_docs = self.search_relevant_docs(query)
        response = self.generate_response(query, relevant_docs)
        return response

# 사용 예시
pipeline = LocalRAGPipeline()

# 샘플 문서
documents = [
    "Python은 고급 프로그래밍 언어로, 간결하고 가독성이 뛰어납니다.",
    "장고(Django)는 웹 프레임워크로, 빠르고 안전한 웹 개발을 지원합니다.",
    "FastAPI는 현대적인 웹 프레임워크로, 높은 성능과 자동 문서화 기능을 제공합니다."
]

# 벡터 인덱스 생성
pipeline.create_vector_index(documents)

# 쿼리 처리
response = pipeline.process_query("Python 프로그래밍 언어의 특징은?")
print(response)

6. 고급 기

---

📥 Get the full guide on Gumroad: https://gumroad.com/l/auto ($7)