matias yoon

Posted on May 25

RAG 시스템 실전 구축 (v32)

#ai #llm #developers #tutorial

RAG 시스템 실전 구축 (v32)

개요

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

1. RAG 기초 원리

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

Retrieval: 사용자 질문에 관련된 문서 또는 정보를 검색
Augmentation: 검색된 정보를 입력으로 사용하여 프롬프트를 보강
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)

DEV Community

RAG 시스템 실전 구축 (v32)

RAG 시스템 실전 구축 (v32)

개요

1. RAG 기초 원리

2. Chunking 전략

Semantic Chunking

Recursive Chunking

Agentic Chunking

3. 임베딩 모델 선택

모델 비교

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

Chroma (가장 쉬운 선택)

Qdrant (성능 우수)

pgvector (관계형 DB 통합)

Milvus (대규모 스케일링)

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

Top comments (0)