RAG 시스템 실전 구축 (v13)

1. RAG 기초 개념

Retrieval-Augmented Generation (RAG)은 정보 검색과 생성 모델을 통합한 아키텍처입니다. RAG 시스템은 다음 세 단계로 작동합니다:

1. 검색 (Retrieval): 사용자 쿼리와 유사한 문서 검색
2. 증강 (Augmentation): 검색된 문서와 쿼리를 결합하여 프롬프트 구성
3. 생성 (Generation): 증강된 프롬프트를 기반으로 답변 생성

# 간단한 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\nDocuments:\n" + "\n".join(retrieved_docs)
        # LLM 호출은 여기서 수행
        return self.llm.generate(prompt)

2. Chunking 전략

문서를 적절한 크기로 나누는 것이 중요합니다. 세 가지 주요 전략은 다음과 같습니다:

2.1 Semantic Chunking

의미 기반으로 문단을 나누어 의미 단위 유지

from langchain.text_splitter import RecursiveCharacterTextSplitter
from sentence_transformers import SentenceTransformer

class SemanticChunker:
    def __init__(self, model_name="all-MiniLM-L6-v2"):
        self.model = SentenceTransformer(model_name)
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50,
            separators=["\n\n", "\n", " ", ""]
        )

    def chunk_document(self, text):
        # 의미적 유사도를 기반으로 chunk 분할
        chunks = self.splitter.split_text(text)
        return chunks

2.2 Recursive Chunking

문서의 구조를 고려한 계층적 분할

def recursive_chunking(text, max_chunk_size=1000):
    """계층적 chunking 전략"""
    if len(text) <= max_chunk_size:
        return [text]

    chunks = []
    # 제목 기준으로 분할
    sections = text.split('\n\n')
    current_chunk = ""

    for section in sections:
        if len(current_chunk) + len(section) < max_chunk_size:
            current_chunk += section + '\n\n'
        else:
            if current_chunk:
                chunks.append(current_chunk.strip())
            current_chunk = section + '\n\n'

    if current_chunk:
        chunks.append(current_chunk.strip())

    return chunks

3. 임베딩 모델 선택

다양한 임베딩 모델의 성능을 비교하여 최적의 선택을 합니다.

import numpy as np
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity

class EmbeddingEvaluator:
    def __init__(self):
        self.models = {
            'all-MiniLM-L6-v2': SentenceTransformer('all-MiniLM-L6-v2'),
            'all-mpnet-base-v2': SentenceTransformer('all-mpnet-base-v2'),
            'BAAI/bge-small-en': SentenceTransformer('BAAI/bge-small-en'),
            'sentence-transformers/gtr-t5-base': SentenceTransformer('sentence-transformers/gtr-t5-base')
        }

    def benchmark_models(self, test_queries, test_docs):
        results = {}
        for name, model in self.models.items():
            # 벡터 생성
            query_vectors = model.encode(test_queries)
            doc_vectors = model.encode(test_docs)

            # 유사도 계산
            similarities = cosine_similarity(query_vectors, doc_vectors)
            avg_similarity = np.mean(similarities)

            results[name] = {
                'avg_similarity': avg_similarity,
                'model': model
            }

        return results

# 성능 비교
evaluator = EmbeddingEvaluator()
benchmark_results = evaluator.benchmark_models(
    test_queries=["machine learning", "deep learning"],
    test_docs=["ML은 데이터 기반 학습을 포함하는 컴퓨터 과학 분야입니다", 
               "DL은 신경망 기반의 머신 러닝입니다"]
)

4. Vector Database 비교

DB	특징	성능	저장 공간	비용
Chroma	로컬, 간단한 사용	빠름	낮음	무료
Qdrant	고성능, 클라우드	높음	중간	유료
pgvector	PostgreSQL 연동	중간	높음	무료
Milvus	대규모, 분산	높음	높음	유료

# Chroma 예제
import chromadb
from chromadb.config import Settings

class ChromaVectorDB:
    def __init__(self, collection_name="rag_docs"):
        self.client = chromadb.Client()
        self.collection = self.client.get_or_create_collection(collection_name)

    def add_documents(self, documents, embeddings, ids):
        self.collection.add(
            embeddings=embeddings,
            documents=documents,
            ids=ids
        )

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

# pgvector 예제
import psycopg2
import numpy as np

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 embeddings (
                    id UUID PRIMARY KEY,
                    content TEXT,
                    embedding VECTOR(768)
                )
            """)
        self.conn.commit()

    def search(self, query_embedding, n_results=5):
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT content, 
                       1 - (embedding <-> %s) as similarity
                FROM embeddings 
                ORDER BY similarity DESC
                LIMIT %s
            """, (np.array(query_embedding), n_results))
            return cur.fetchall()

5. RAG 파이프라인 전체 구현

import os
from sentence_transformers import SentenceTransformer
from chromadb import Client
from chromadb.config import Settings
from langchain.text_splitter import RecursiveCharacterTextSplitter
import numpy as np

class CompleteRAGPipeline:
    def __init__(self, model_name="all-MiniLM-L6-v2", db_path="./chroma_db"):
        # 임베딩 모델 초기화
        self.embedding_model = SentenceTransformer(model_name)

        # Vector DB 설정
        self.chroma_client = Client(Settings(chroma_db_impl="duckdb", persist_directory=db_path))
        self.collection = self.chroma_client.get_or_create_collection("rag_documents")

        # Chunking 설정
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50
        )

    def add_document(self, document_id, content):
        """문서 추가"""
        # 문서 분할
        chunks = self.text_splitter.split_text(content)

        # 임베딩 생성
        embeddings = self.embedding_model.encode(chunks)

        # DB에 저장
        self.collection.add(
            embeddings=embeddings.tolist(),
            documents=chunks,
            ids=[f"{document_id}_{i}" for i in range(len(chunks))]
        )

    def retrieve(self, query, k=5):
        """쿼리 검색"""
        query_embedding = self.embedding_model.encode([query])
        results = self.collection.query(
            query_embeddings=query_embedding.tolist(),
            n_results=k
        )
        return results['documents'][0]

    def generate_answer(self, query, retrieved_docs):
        """답변 생성 (예시)"""
        context = "\n\n".join(retrieved_docs)
        prompt = f"""
        질문: {query}
        문서 컨텍스트:
        {context}

        답변:
        """

        # 실제 LLM 호출은 여기서 수행
        # 예: openai.ChatCompletion.create() 또는 local LLM 호출
        return f"질문: {query}\n\n참고 문서:\n{context}"

    def process_query(self, query):
        """전체 쿼리 처리"""
        retrieved_docs = self.retrieve(query)
        answer = self.generate_answer(query, retrieved_docs)
        return answer

# 사용 예시
rag_pipeline = CompleteRAGPipeline()
rag_pipeline.add_document("doc1", "Python은 인터프리터 기반의 고급 프로그래밍 언어입니다. 다양한 분야에서 사용됩니다.")
answer = rag_pipeline.process_query("Python은 어떤 언어인가요?")
print(answer)

6. 고급 기능

6.1 Query Transformation

python
class QueryTransformer

---

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