RAG 시스템 실전 구축 (v29)

실전 가이드: 효율적이고 확장 가능한 RAG 시스템 구축

1. RAG 기초 개념

RAG (Retrieval-Augmented Generation) 시스템은 검색과 생성을 통합한 아키텍처입니다. 세 가지 핵심 단계로 구성됩니다:

• 검색 (Retrieval): 사용자 질문과 관련된 문서 조각 검색
• 증강 (Augmentation): 검색된 문서와 질문 결합
• 생성 (Generation): 증강된 정보를 기반으로 답변 생성

사용자 질문 → 검색 엔진 → 관련 문단 → 생성 모델 → 답변

2. 청킹 전략 (Chunking Strategies)

2.1 의미 기반 청킹 (Semantic Chunking)

from langchain_text_splitters import RecursiveCharacterTextSplitter
from sentence_transformers import SentenceTransformer

# 의미 기반 청킹
def semantic_chunking(text, model_name="all-MiniLM-L6-v2"):
    model = SentenceTransformer(model_name)
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
        separators=["\n\n", "\n", " ", ""]
    )
    chunks = splitter.split_text(text)

    # 임베딩 생성
    embeddings = model.encode(chunks)
    return chunks, embeddings

2.2 재귀적 청킹 (Recursive Chunking)

# 재귀적 청킹 구현
def recursive_chunking(text, max_chunk_size=1000, overlap=100):
    import re

    # 문장 단위로 분할
    sentences = re.split(r'[.!?]+', text)
    chunks = []
    current_chunk = ""

    for sentence in sentences:
        if len(current_chunk) + len(sentence) < max_chunk_size:
            current_chunk += sentence + ". "
        else:
            if current_chunk:
                chunks.append(current_chunk.strip())
            current_chunk = sentence + ". "

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

    return chunks

3. 임베딩 모델 선택

3.1 모델 비교 및 평가

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

# 모델 비교 함수
def compare_models(texts, models_config):
    results = {}

    for model_name, model_path in models_config.items():
        model = SentenceTransformer(model_path)
        embeddings = model.encode(texts)

        # 유사도 계산 (예: 첫 번째 텍스트와 나머지 비교)
        similarities = cosine_similarity([embeddings[0]], embeddings[1:])[0]

        results[model_name] = {
            "avg_similarity": np.mean(similarities),
            "embedding_size": embeddings.shape[1]
        }

    return results

# 사용 예시
models = {
    "all-MiniLM-L6-v2": "all-MiniLM-L6-v2",
    "all-mpnet-base-v2": "all-mpnet-base-v2",
    "multi-qa-MiniLM-L6-v2": "multi-qa-MiniLM-L6-v2"
}

# 테스트 데이터
test_texts = [
    "RAG 시스템은 검색과 생성을 통합하는 아키텍처입니다.",
    "검색 단계에서 관련 문서를 찾습니다.",
    "생성 단계에서는 검색된 정보를 기반으로 답변을 만듭니다."
]

comparison_results = compare_models(test_texts, models)
print(comparison_results)

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

4.1 Chroma vs Qdrant vs pgvector vs Milvus

# Chroma 예제
from chromadb import Client
import numpy as np

def setup_chroma():
    client = Client()
    collection = client.get_or_create_collection("rag_docs")

    # 문서 추가
    def add_documents(documents, embeddings):
        collection.add(
            embeddings=embeddings,
            documents=documents,
            ids=[f"doc_{i}" for i in range(len(documents))]
        )

    # 검색
    def search(query_embedding, top_k=5):
        results = collection.query(
            query_embeddings=[query_embedding],
            n_results=top_k
        )
        return results

    return add_documents, search

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

def setup_qdrant():
    client = QdrantClient(host="localhost", port=6333)

    def add_documents(documents, embeddings):
        client.upsert(
            collection_name="rag_docs",
            points=[
                {
                    "id": i,
                    "vector": embedding.tolist(),
                    "payload": {"text": doc}
                }
                for i, (doc, embedding) in enumerate(zip(documents, embeddings))
            ]
        )

    def search(query_embedding, top_k=5):
        results = client.search(
            collection_name="rag_docs",
            query_vector=query_embedding.tolist(),
            limit=top_k
        )
        return results

    return add_documents, search

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

import os
import numpy as np
from sentence_transformers import SentenceTransformer
from chromadb import Client
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import ChatOpenAI

class RAGPipeline:
    def __init__(self, model_name="all-MiniLM-L6-v2", chroma_path="./chroma_db"):
        self.embedding_model = SentenceTransformer(model_name)
        self.chroma_client = Client()
        self.collection = self.chroma_client.get_or_create_collection("rag_docs")
        self.llm = ChatOpenAI(model="gpt-3.5-turbo")

    def add_document(self, text, doc_id):
        """문서 추가 및 임베딩"""
        # 청킹
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50
        )
        chunks = splitter.split_text(text)

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

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

        return len(chunks)

    def retrieve(self, query, top_k=5):
        """쿼리 검색"""
        query_embedding = self.embedding_model.encode([query])[0]

        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=top_k
        )

        return results['documents'][0]

    def generate(self, query, context):
        """답변 생성"""
        prompt = f"""
        다음 컨텍스트를 기반으로 질문에 답변하세요:
        컨텍스트: {context}
        질문: {query}
        답변:
        """

        response = self.llm.invoke(prompt)
        return response.content

    def process_query(self, query):
        """전체 파이프라인 실행"""
        # 검색
        context = self.retrieve(query)
        # 생성
        answer = self.generate(query, " ".join(context))
        return answer

# 사용 예시
rag = RAGPipeline()
rag.add_document("RAG 시스템은 검색과 생성을 통합하여 정확한 답변을 제공합니다.", "doc1")
answer = rag.process_query("RAG 시스템의 주요 특징은 무엇인가요?")
print(answer)

6. 고급 기능 구현

6.1 쿼리 변환 (Query Transformation)

class QueryTransformer:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-3.5-turbo")

    def transform_query(self, query):
        """쿼리 변환"""
        prompt = f"""
        다음 질문을 더 구체적이고 검색하기 쉬운 형태로 변환하세요:
        원래 질문: {query}
        변환된 질문:
        """

        response = self.llm.invoke(prompt)
        return response.content

# 하이브리드 검색
def hybrid_search(query, vector_results, keyword_results, alpha=0.7):
    """벡터 검색과 키워드 검색의 하이브리드"""
    # 가중치 합산
    combined_scores = []
    for i, vector_score in enumerate(vector_results['distances'][0]):
        keyword_score = keyword_results[i] if i < len(keyword_results) else 0
        combined_score = alpha * vector_score + (1 - alpha) * keyword_score
        combined_scores.append(combined_score)

    # 정렬 후 반환
    return [idx for idx, _ in sorted(enumerate(combined_scores), key=lambda x: x[1])]

6.2 재정렬 (Re-ranking)

python
# 재정렬 모델 사용
from sentence_transformers import CrossEncoder

class ReRanker:
    def __init__(

---

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