RAG 시스템 실전 구축 (v32)

개요

이 가이드는 실제 ML 엔지니어와 백엔드 개발자가 RAG(Retrieval-Augmented Generation) 시스템을 구축할 때 마주하는 실질적인 문제들을 해결하기 위한 실전 가이드입니다. 현대의 LLM 기반 애플리케이션 개발에서 RAG는 핵심 구성 요소이며, 이를 효율적으로 설계하고 구현하는 것은 성공적인 AI 제품의 기본입니다.

1. RAG 기초 원리

RAG는 세 가지 주요 단계로 구성됩니다:

1. Retrieval: 사용자 질문에 관련된 문서 또는 정보를 검색
2. Augmentation: 검색된 정보를 입력으로 사용하여 프롬프트를 보강
3. Generation: 보강된 프롬프트를 바탕으로 생성 모델이 답변 생성

# RAG 루프의 기본 구조
class RAGPipeline:
    def __init__(self, retriever, generator):
        self.retriever = retriever
        self.generator = generator

    def process_query(self, query):
        # 1. 검색
        retrieved_docs = self.retriever.retrieve(query)

        # 2. 보강
        augmented_prompt = self._augment_prompt(query, retrieved_docs)

        # 3. 생성
        answer = self.generator.generate(augmented_prompt)
        return answer

    def _augment_prompt(self, query, docs):
        # 문서 내용을 프롬프트에 포함
        context = "\n".join([doc.content for doc in docs])
        return f"질문: {query}\n참고 문서:\n{context}"

2. Chunking 전략

문서를 청크로 분할하는 것은 성능과 정확도에 직접적인 영향을 미칩니다.

Semantic Chunking

from sentence_transformers import SentenceTransformer
import numpy as np

def semantic_chunking(text, model, threshold=0.75):
    sentences = text.split('. ')
    embeddings = model.encode(sentences)

    chunks = []
    current_chunk = []
    current_embedding = None

    for i, (sentence, embedding) in enumerate(zip(sentences, embeddings)):
        if current_embedding is None:
            current_chunk.append(sentence)
            current_embedding = embedding
        else:
            # 유사도 계산
            similarity = np.dot(current_embedding, embedding) / (
                np.linalg.norm(current_embedding) * np.linalg.norm(embedding)
            )

            if similarity > threshold:
                current_chunk.append(sentence)
            else:
                chunks.append(' '.join(current_chunk))
                current_chunk = [sentence]
                current_embedding = embedding

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

    return chunks

# 사용 예시
model = SentenceTransformer('all-MiniLM-L6-v2')
text = "..." # 긴 문서
chunks = semantic_chunking(text, model)

Recursive Chunking

def recursive_chunking(text, max_chunk_size=500, overlap=50):
    chunks = []
    start = 0

    while start < len(text):
        end = min(start + max_chunk_size, len(text))

        # 단어 단위로 잘라서 청크
        chunk = text[start:end]
        chunks.append(chunk)

        start = end - overlap

    return chunks

Agentic Chunking

from langchain.text_splitter import RecursiveCharacterTextSplitter

def agentic_chunking(text, chunk_size=1000, chunk_overlap=200):
    # 다양한 구분자로 청크 생성
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", " ", ""]
    )

    chunks = text_splitter.split_text(text)
    return chunks

3. 임베딩 모델 선택

모델 비교

모델	크기	성능	추천 사용 사례
all-MiniLM-L6-v2	26MB	78%	일반적인 RAG
BGE-M3	128MB	82%	고정밀도 필요
e5-small	100MB	75%	영어 전용

from sentence_transformers import SentenceTransformer
import numpy as np

# 모델 비교
def benchmark_embeddings(texts, models):
    results = {}

    for model_name in models:
        model = SentenceTransformer(model_name)
        embeddings = model.encode(texts)
        results[model_name] = {
            'avg_similarity': np.mean([np.dot(embeddings[i], embeddings[i+1]) 
                                     for i in range(0, len(embeddings)-1, 2)])
        }

    return results

# 예시
texts = ["문서 1 내용", "문서 2 내용", "문서 3 내용"]
models = ['all-MiniLM-L6-v2', 'BGE-M3', 'e5-small']
results = benchmark_embeddings(texts, models)

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

Chroma (가장 쉬운 선택)

import chromadb
from chromadb.config import Settings

# Chroma 초기화
client = chromadb.Client(Settings(chroma_db_impl="duckdb", persist_directory="./chroma_db"))
collection = client.get_or_create_collection("rag_collection")

# 저장
def store_in_chroma(documents, ids):
    collection.add(
        documents=documents,
        ids=ids
    )

# 검색
def search_chroma(query, n_results=5):
    results = collection.query(
        query_texts=[query],
        n_results=n_results
    )
    return results['documents'][0]

Qdrant (성능 우수)

from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue

# Qdrant 초기화
client = QdrantClient(path="./qdrant_storage")

# 저장
def store_in_qdrant(documents, vectors, ids):
    client.upsert(
        collection_name="rag_collection",
        points=[
            {
                "id": id,
                "vector": vector,
                "payload": {"content": doc}
            }
            for id, doc, vector in zip(ids, documents, vectors)
        ]
    )

# 검색
def search_qdrant(query_vector, n_results=5):
    results = client.search(
        collection_name="rag_collection",
        query_vector=query_vector,
        limit=n_results
    )
    return [hit.payload['content'] for hit in results]

pgvector (관계형 DB 통합)

-- PostgreSQL + pgvector 설정
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE embeddings (
    id UUID PRIMARY KEY,
    content TEXT,
    embedding VECTOR(768)
);

-- 저장
INSERT INTO embeddings (id, content, embedding) 
VALUES ('uuid', '문서 내용', '[0.1, 0.2, ...]');

Milvus (대규모 스케일링)

from pymilvus import Milvus, Collection, FieldSchema, DataType, CollectionSchema

# Milvus 연결
milvus = Milvus(host='localhost', port='19530')

# 컬렉션 생성
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=False),
    FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768)
]

schema = CollectionSchema(fields)
collection = Collection("rag_collection", schema)

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

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

class SimpleRAG:
    def __init__(self, embedding_model='all-MiniLM-L6-v2'):
        # 초기화
        self.embedding_model = SentenceTransformer(embedding_model)
        self.client = Client(Settings(chroma_db_impl="duckdb", persist_directory="./rag_db"))
        self.collection = self.client.get_or_create_collection("docs")
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000, chunk_overlap=200
        )

    def add_documents(self, documents):
        """문서 추가"""
        # 청크 분할
        chunks = []
        ids = []
        for i, doc in enumerate(documents):
            doc_chunks = self.text_splitter.split_text(doc)
            chunks.extend(doc_chunks)
            ids.extend([f"{i}_{j}" for j in range(len(doc_chunks))])

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

        # 저장
        self.collection.add(
            documents=chunks,
            embeddings=embeddings.tolist(),
            ids=ids
        )

    def retrieve(self, query, k=5):
        """문서 검색"""
        query

---

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