RAG 시스템 실전 구축 (v12)

실제로 사용할 수 있는 RAG 구축 가이드

1. RAG 기초 개념

RAG (Retrieval-Augmented Generation)는 검색 기반 생성 모델로, LLM이 외부 문서에서 정보를 검색하고 이를 기반으로 답변을 생성하는 아키텍처입니다.

RAG 루프 구성:

1. 검색 (Retrieval): 사용자 질문에 관련된 문서 조각 검색
2. 증강 (Augmentation): 검색된 문서와 질문을 프롬프트로 결합
3. 생성 (Generation): LLM이 증강된 프롬프트를 기반으로 답변 생성

2. 청킹 전략

2.1 의미 기반 청킹 (Semantic Chunking)

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings

# 문맥 기반 청킹
def semantic_chunking(text, chunk_size=500, chunk_overlap=50):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", " ", ""]
    )
    chunks = splitter.split_text(text)
    return chunks

# 예제 사용
text = "AI 기술은 빠르게 발전하고 있습니다. 특히 자연어 처리와 컴퓨터 비전 분야에서 많은 진전이 있었습니다."
chunks = semantic_chunking(text)

2.2 재귀적 청킹

# 재귀적 청킹: 단락 단위로 청킹
def recursive_chunking(text, chunk_size=1000):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=200,
        separators=["\n\n", "\n", "\t", " "]
    )
    return splitter.split_text(text)

2.3 에이전트 기반 청킹

# 청킹 전략 비교
def evaluate_chunking_strategies(texts):
    strategies = {
        'semantic': semantic_chunking,
        'recursive': recursive_chunking,
        'agentic': lambda x: agentic_chunking(x)  # 사용자 정의
    }

    results = {}
    for name, strategy in strategies.items():
        chunks = [strategy(text) for text in texts]
        results[name] = {
            'avg_chunks_per_doc': np.mean([len(chunk) for chunk in chunks]),
            'avg_chunk_size': np.mean([len(chunk) for chunk in chunks])
        }
    return results

3. 임베딩 모델 선택

3.1 다양한 임베딩 모델 비교

from sentence_transformers import SentenceTransformer
from chromadb.utils import embedding_functions
import numpy as np

# 임베딩 모델 비교
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')
        }

    def evaluate_models(self, texts, queries):
        results = {}
        for name, model in self.models.items():
            # 임베딩 생성
            text_embeddings = model.encode(texts)
            query_embeddings = model.encode(queries)

            # 유사도 계산
            similarities = np.dot(query_embeddings, text_embeddings.T)
            results[name] = {
                'mean_similarity': np.mean(similarities),
                'top_k_accuracy': self.calculate_top_k_accuracy(similarities, k=3)
            }
        return results

# 사용 예
evaluator = EmbeddingEvaluator()

3.2 성능 기준

# 임베딩 모델 성능 비교
def compare_embedding_performance():
    # 빠른 모델 (적은 리소스)
    fast_models = ['all-MiniLM-L6-v2', 'BAAI/bge-small-en']

    # 정확도 높은 모델
    accurate_models = ['all-mpnet-base-v2', 'sentence-transformers/all-mpnet-base-v2']

    return {
        'fast': fast_models,
        'accurate': accurate_models,
        'balanced': ['all-MiniLM-L6-v2', 'BAAI/bge-small-en']
    }

4. 벡터 데이터베이스 비교

4.1 Chroma vs Qdrant vs pgvector vs Milvus

# Chroma 구현
import chromadb
from chromadb.config import Settings

class ChromaVectorDB:
    def __init__(self, collection_name="rag_collection"):
        self.client = chromadb.Client(Settings(chroma_db_impl="duckdb"))
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            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, top_k=5):
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=top_k
        )
        return results

# Qdrant 구현
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Filter, FieldCondition

class QdrantVectorDB:
    def __init__(self, host="localhost", port=6333):
        self.client = QdrantClient(host=host, port=port)

    def create_collection(self, collection_name, vector_size):
        self.client.create_collection(
            collection_name=collection_name,
            vectors_config=VectorParams(size=vector_size, distance="Cosine")
        )

    def search(self, collection_name, query_vector, top_k=5):
        results = self.client.search(
            collection_name=collection_name,
            query_vector=query_vector,
            limit=top_k
        )
        return results

4.2 벡터 데이터베이스 성능 비교

def benchmark_vector_dbs():
    # 벡터 데이터베이스별 성능 지표
    performance_metrics = {
        'Chroma': {
            'setup_time': '10s',
            'search_time': '<100ms',
            'memory_usage': 'Low',
            'ease_of_use': 'High'
        },
        'Qdrant': {
            'setup_time': '5s',
            'search_time': '<50ms',
            'memory_usage': 'Medium',
            'ease_of_use': 'Medium'
        },
        'pgvector': {
            'setup_time': '15s',
            'search_time': '<150ms',
            'memory_usage': 'High',
            'ease_of_use': 'Low'
        },
        'Milvus': {
            'setup_time': '20s',
            'search_time': '<20ms',
            'memory_usage': 'Low',
            'ease_of_use': 'Low'
        }
    }

    return performance_metrics

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

from langchain_openai import OpenAI, OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
import chromadb

class RAGPipeline:
    def __init__(self, model_name="gpt-3.5-turbo"):
        # 임베딩 모델
        self.embeddings = OpenAIEmbeddings()

        # LLM 모델
        self.llm = OpenAI(model_name=model_name)

        # 벡터 저장소
        self.vectorstore = Chroma(
            client=chromadb.Client(),
            collection_name="rag_collection",
            embedding_function=self.embeddings
        )

        # 프롬프트 템플릿
        self.prompt_template = PromptTemplate(
            input_variables=["context", "question"],
            template="""
            다음 컨텍스트를 기반으로 질문에 답변하세요:

            {context}

            질문: {question}
            답변:
            """
        )

    def add_documents(self, documents):
        """문서 추가"""
        self.vectorstore.add_documents(documents)

    def query(self, question):
        """질문 처리"""
        # 검색
        retriever = self.vectorstore.as_retriever()

        # QA 체인 생성
        qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=retriever,
            prompt=self.prompt_template
        )

        return qa_chain.run(question)

# 사용 예
pipeline = RAGPipeline()
pipeline.add_documents([
    "Python은 고급 프로그래밍 언어입니다.",
    "Docker는 컨테이너화 기술입니다."
])
response = pipeline.query("Python은 어떤 언어인가요?")

6. 고급 기능

6.1

---

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