RAG 시스템 실전 구축 (v34)

1. RAG 시스템의 핵심 구성 요소

RAG(Retrieve-Augment-Generate) 시스템은 대규모 언어 모델(LLM)의 지식 범위를 확장하는 핵심 아키텍처입니다. 세 가지 주요 단계로 구성됩니다:

Retrieval 단계

사용자의 질문을 기반으로 관련 문서를 검색합니다.

Augmentation 단계

검색된 문서와 질문을 결합하여 LLM 입력을 확장합니다.

Generation 단계

확장된 입력을 기반으로 답변을 생성합니다.

# RAG 루프의 기본 구조
class SimpleRAG:
    def __init__(self, embedding_model, vector_db, llm):
        self.embedding_model = embedding_model
        self.vector_db = vector_db
        self.llm = llm

    def process_query(self, query):
        # 1. 질문 임베딩 생성
        query_embedding = self.embedding_model.encode(query)

        # 2. 관련 문서 검색
        relevant_docs = self.vector_db.search(query_embedding, k=5)

        # 3. 증강된 프롬프트 생성
        augmented_prompt = self.augment_prompt(query, relevant_docs)

        # 4. 답변 생성
        response = self.llm.generate(augmented_prompt)
        return response

    def augment_prompt(self, query, docs):
        context = "\n\n".join([doc.content for doc in docs])
        return f"Context: {context}\n\nQuestion: {query}"

2. Chunking 전략

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

1. Semantic Chunking

의미적 단위로 청킹하여 문맥을 유지합니다.

from langchain_text_splitters import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# Semantic Chunking 예제
def semantic_chunking(text, embedding_model):
    semantic_splitter = SemanticChunker(embedding_model)
    chunks = semantic_splitter.split_text(text)
    return chunks

# 예시 사용
chunks = semantic_chunking(
    "Large language models can process natural language effectively. They require significant computational resources.",
    OpenAIEmbeddings()
)

2. Recursive Chunking

정규식 기반으로 재귀적으로 청킹합니다.

from langchain_text_splitters import RecursiveCharacterTextSplitter

def recursive_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

3. Agentic Chunking

LLM가 청킹 결정을 내리는 방식

# Agentic Chunking을 위한 프롬프트
def agentic_chunking_prompt(text):
    prompt = f"""
    분할하세요. 각 청크는 300-500 단어 사이여야 하며, 문맥이 유지되어야 합니다.
    원본 텍스트: {text}
    청크 목록:
    """
    return prompt

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

다양한 임베딩 모델의 성능을 비교하세요.

# 임베딩 모델 비교 테스트
import numpy as np
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity

class EmbeddingBenchmark:
    def __init__(self):
        self.models = {
            'all-MiniLM-L6-v2': SentenceTransformer('all-MiniLM-L6-v2'),
            'all-mpnet-base-v2': SentenceTransformer('all-mpnet-base-v2'),
            'sentence-t5-base': SentenceTransformer('sentence-t5-base')
        }

    def compare_models(self, sentences):
        results = {}
        for name, model in self.models.items():
            embeddings = model.encode(sentences)
            # 간단한 유사도 계산 예시
            if len(embeddings) >= 2:
                similarity = cosine_similarity([embeddings[0]], [embeddings[1]])[0][0]
                results[name] = similarity
        return results

# 성능 평가
benchmark = EmbeddingBenchmark()
sentences = ["Hello world", "Goodbye world"]
print(benchmark.compare_models(sentences))

4. Vector Database 비교

DB	특징	장점	단점
Chroma	로컬, 간단	빠른 개발	대용량 처리 부족
Qdrant	고성능, 클라우드	검색 속도 빠름	설정 복잡
pgvector	PostgreSQL 연동	관계형 데이터와 통합	복잡한 관리
Milvus	분산 처리	대용량 데이터	높은 설정 비용

# Chroma DB 예제
import chromadb
from chromadb import Client

class ChromaVectorStore:
    def __init__(self, collection_name="rag_collection"):
        self.client = Client()
        self.collection = self.client.get_or_create_collection(collection_name)

    def add_documents(self, documents, embeddings, metadata=None):
        ids = [str(i) for i in range(len(documents))]
        self.collection.add(
            ids=ids,
            embeddings=embeddings,
            documents=documents,
            metadatas=metadata or [{}] * len(documents)
        )

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

# Qdrant DB 예제
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition

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

    def search(self, query_embedding, top_k=5):
        results = self.client.search(
            collection_name=self.collection_name,
            query_vector=query_embedding,
            limit=top_k
        )
        return [hit.payload for hit in results]

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

import os
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain.chains import RetrievalQA
from langchain.vectorstores import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter

class CompleteRAGPipeline:
    def __init__(self, api_key, model_name="gpt-3.5-turbo"):
        # 환경 설정
        os.environ["OPENAI_API_KEY"] = api_key

        # 구성 요소 초기화
        self.embedding_model = OpenAIEmbeddings()
        self.llm = ChatOpenAI(model_name=model_name, temperature=0)
        self.vector_store = Chroma(
            collection_name="rag_collection",
            embedding_function=self.embedding_model
        )

        # 청킹 전략
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200
        )

    def ingest_documents(self, documents):
        """문서 인덱싱"""
        # 문서 청킹
        texts = self.splitter.split_text(documents)

        # 임베딩 생성 및 저장
        self.vector_store.add_texts(texts)
        print(f"인덱싱 완료: {len(texts)}개 청크")

    def query(self, question):
        """질문 처리"""
        # Retrieval
        retriever = self.vector_store.as_retriever()

        # 프롬프트 템플릿
        template = """
        다음 문맥을 바탕으로 질문에 답변하세요:

        {context}

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

        prompt = PromptTemplate.from_template(template)

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

        # 답변 생성
        result = qa_chain.run(question)
        return result

# 사용 예시
# rag = CompleteRAGPipeline("your-api-key-here")
# rag.ingest_documents("문서 내용...")
# answer = rag.query("질문 내용?")

6. 고급 기능들

Query Transformation

질문을 더 효과적으로 검색할 수 있도록 변환합니다.

python
def transform_query(query, llm):
    """질문 변환"""
    transform_prompt = f"""
    질문을 검색에 최적화된 형태로 변환하세요.
    원본 질문: {query}
    변환된 질문:
    """
    return llm.invoke(transform_prompt).content

# 사용 예시
# transformed = transform_query("AI

---

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