RAG 시스템 실전 구축 (v41)

1. RAG 기본 개념: 검색 → 보강 → 생성 루프

Retrieval-Augmented Generation (RAG)은 대규모 언어 모델(LLM)을 통해 질문에 대한 답변을 생성할 때, 관련된 외부 문서를 검색하여 정보를 보강하는 아키텍처입니다. 이는 LLM의 지식 범위를 확장하고, 최신 정보를 포함할 수 있게 해줍니다.

RAG 루프의 세 가지 핵심 단계:

1. 검색 (Retrieval): 사용자 질문과 유사한 문서/청크를 벡터 데이터베이스에서 찾습니다.
2. 보강 (Augmentation): 검색된 문서를 프롬프트에 포함시켜 LLM이 더 정확한 답변을 생성할 수 있도록 합니다.
3. 생성 (Generation): LLM은 보강된 프롬프트를 기반으로 답변을 생성합니다.

이 루프를 통해 LLM은 문서 기반의 정확성과 추론 능력을 동시에 갖출 수 있습니다.

2. 청킹 전략: 의미 기반 vs 재귀적 vs 에이전트 기반

청킹은 문서를 모델이 처리할 수 있는 단위로 나누는 과정입니다. 각 전략은 서로 다른 목적과 성능 특성을 가지고 있습니다.

2.1 의미 기반 청킹 (Semantic Chunking)

문서의 의미적 유사성에 따라 청킹합니다. 예를 들어, 문장의 의미가 변경될 때 청킹을 나눕니다.

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

def semantic_chunk(text, model, threshold=0.85):
    sentences = text.split('. ')
    embeddings = model.encode(sentences)
    chunks = []
    current_chunk = [sentences[0]]
    current_embedding = embeddings[0]

    for i in range(1, len(sentences)):
        similarity = cosine_similarity([current_embedding], [embeddings[i]])[0][0]
        if similarity < threshold:
            chunks.append(' '.join(current_chunk))
            current_chunk = [sentences[i]]
            current_embedding = embeddings[i]
        else:
            current_chunk.append(sentences[i])
            current_embedding = np.average([current_embedding, embeddings[i]], axis=0)

    chunks.append(' '.join(current_chunk))
    return chunks

2.2 재귀적 청킹 (Recursive Chunking)

문서를 고정된 길이로 분할하고, 문맥을 유지하기 위해 중복을 포함합니다.

def recursive_chunk(text, chunk_size=512, overlap=64):
    chunks = []
    start = 0
    while start < len(text):
        end = min(start + chunk_size, len(text))
        chunk = text[start:end]
        chunks.append(chunk)
        start = end - overlap
    return chunks

2.3 에이전트 기반 청킹 (Agentic Chunking)

문서의 구조와 의미에 따라 청킹합니다. 예: 제목, 섹션, 헤드라인 등을 기준으로 청킹.

def agentic_chunk(text):
    sections = text.split('\n\n')
    chunks = []
    for section in sections:
        if len(section) > 100:
            chunks.append(section)
    return chunks

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

임베딩 모델은 문서를 벡터 공간으로 변환하여 유사도를 계산하는 데 사용됩니다.

3.1 모델 비교

모델	파라미터	성능	용량	추천 사용 사례
Sentence-BERT (all-MiniLM-L6-v2)	28M	78%	100MB	일반 목적
BGE-M3	130M	84%	500MB	높은 정확도 필요
FastEmbed (BAAI/bge-small-en)	20M	75%	50MB	리소스 제한 환경

from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer('all-MiniLM-L6-v2')
embeddings = model.encode(['Hello world', 'How are you?'])
print(f"Embedding shape: {embeddings.shape}")

4. 벡터 데이터베이스 비교: Chroma vs Qdrant vs pgvector vs Milvus

데이터베이스	장점	단점	사용 사례
Chroma	간단한 API, 로컬 호스팅	대규모 데이터 처리 부족	개발용, 테스트
Qdrant	고성능, 복잡한 검색	설치 복잡	대규모 애플리케이션
pgvector	PostgreSQL 기반, 안정성	성능 저하	기존 DB 통합
Milvus	분산, 고성능	설정 복잡	대규모, 분산 환경

# Chroma 예시
import chromadb

client = chromadb.Client()
collection = client.get_or_create_collection("docs")
collection.add(
    documents=["Document content"],
    embeddings=[model.encode("Document content")],
    ids=["doc1"]
)
results = collection.query(
    query_embeddings=[model.encode("Query")],
    n_results=2
)

5. 완전한 RAG 파이프라인 코드

from sentence_transformers import SentenceTransformer
import chromadb
import numpy as np
from typing import List, Tuple

class SimpleRAG:
    def __init__(self, model_name="all-MiniLM-L6-v2"):
        self.model = SentenceTransformer(model_name)
        self.client = chromadb.Client()
        self.collection = self.client.get_or_create_collection("docs")

    def add_documents(self, docs: List[str]):
        embeddings = self.model.encode(docs)
        self.collection.add(
            documents=docs,
            embeddings=embeddings,
            ids=[f"doc_{i}" for i in range(len(docs))]
        )

    def search(self, query: str, top_k: int = 3) -> List[str]:
        query_embedding = self.model.encode([query])
        results = self.collection.query(
            query_embeddings=query_embedding,
            n_results=top_k
        )
        return results['documents'][0]

    def generate_response(self, query: str, context: List[str]) -> str:
        # 간단한 프롬프트 생성 예시
        prompt = f"질문: {query}\n\n참고 문서:\n" + "\n".join(context)
        # LLM 호출은 실제 환경에서 직접 구현
        return prompt  # 실제 구현에서는 LLM 호출

# 사용 예시
rag = SimpleRAG()
rag.add_documents(["Python은 인터프리터 기반의 고급 프로그래밍 언어입니다.", "Django는 웹 프레임워크입니다."])
context = rag.search("Python 언어 특징")
response = rag.generate_response("Python은 어떤 언어인가요?", context)
print(response)

6. 고급 기능: 쿼리 변환, 하이브리드 검색, 재순위

6.1 쿼리 변환 (Query Transformation)

사용자 쿼리를 더 정확한 검색어로 변환합니다.

def query_transform(query: str) -> List[str]:
    # 예: "Python 웹 프레임워크" → ["Django", "Flask", "Python web framework"]
    return [query, f"{query} framework", f"{query} development"]

6.2 하이브리드 검색 (Hybrid Search)

다양한 검색 방법을 결합합니다.

def hybrid_search(query: str, top_k: int = 3):
    # 의미 기반 검색 + 키워드 검색 조합
    semantic_results = semantic_search(query, top_k)
    keyword_results = keyword_search(query, top_k)
    return merge_results(semantic_results, keyword_results, top_k)

6.3 재순위 (Re-ranking)

검색된 문서의 순위를 다시 정렬합니다.

python
from transformers import AutoTokenizer, AutoModel
from sklearn.metrics.pairwise import cosine_similarity

def rerank(query: str, candidates: List[str]):
    tokenizer = AutoTokenizer.from_pretrained("cross-encoder/ms-marco-MiniLM-L-6-v2")
    model = AutoModel.from_pretrained("cross-encoder/ms-marco-MiniLM-L-6-v2")

    pairs = [[query, candidate] for candidate in candidates]
    inputs = tokenizer(pairs, return_tensors="pt", padding=True, truncation=True)
    scores = model(**inputs).logits.squeeze()
    ranked = sorted(zip(candidates, scores), key=lambda

---

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