RAG 시스템 실전 구축 (v17)
개요
RAG (Retrieval-Augmented Generation) 시스템은 현대 대형 언어 모델(Large Language Models)의 능력을 극대화하는 핵심 기술입니다. 이 가이드는 ML 엔지니어와 백엔드 개발자가 실전에서 RAG 시스템을 구축하기 위해 필요한 모든 요소를 다룹니다. 특히, 실제 비즈니스 문제를 해결할 수 있도록 설계된 구체적인 코드와 전략을 제공합니다.
1. RAG 기초 개념 (Retrieval → Augmentation → Generation Loop)
RAG는 세 가지 핵심 단계로 구성됩니다:
- Retrieval: 질문과 유사한 문서를 검색합니다.
- Augmentation: 검색된 문서를 프롬프트에 추가하여 컨텍스트를 확장합니다.
- Generation: 확장된 컨텍스트를 기반으로 답변을 생성합니다.
# 간단한 RAG 루프 예시
def rag_pipeline(query, vector_db, llm):
# 1. 검색
retrieved_docs = vector_db.search(query, k=5)
# 2. 증강
context = "\n".join([doc.content for doc in retrieved_docs])
augmented_prompt = f"Context: {context}\n\nQuestion: {query}"
# 3. 생성
answer = llm.generate(augmented_prompt)
return answer
2. Chunking 전략 (Semantic, Recursive, Agentic)
적절한 chunking은 검색 성능에 큰 영향을 줍니다.
2.1 Semantic Chunking
문서 의미 기반으로 chunk를 나누는 방식입니다.
# SentenceTransformers 기반 의미적 청킹
from sentence_transformers import SentenceTransformer
import numpy as np
def semantic_chunking(text, model, threshold=0.8):
sentences = text.split('. ')
embeddings = model.encode(sentences)
chunks = []
current_chunk = []
current_embedding = None
for i, (sentence, embedding) in enumerate(zip(sentences, embeddings)):
if i == 0:
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:
chunks.append('. '.join(current_chunk))
current_chunk = [sentence]
current_embedding = embedding
else:
current_chunk.append(sentence)
if current_chunk:
chunks.append('. '.join(current_chunk))
return chunks
2.2 Recursive Chunking
지정된 길이를 기준으로 재귀적으로 나누는 전략입니다.
def recursive_chunking(text, max_chunk_size=500):
if len(text) <= max_chunk_size:
return [text]
chunks = []
while text:
if len(text) <= max_chunk_size:
chunks.append(text)
break
# 중간에서 자르되, 문장 단위로 잘라서 오류 방지
split_point = text.rfind('. ', 0, max_chunk_size)
if split_point == -1:
split_point = max_chunk_size
chunks.append(text[:split_point])
text = text[split_point:].lstrip()
return chunks
2.3 Agentic Chunking
지식 그래프 또는 구조화된 데이터를 기반으로 chunk를 생성하는 방식입니다.
def agentic_chunking(data_structure, chunk_size=500):
"""구조화된 데이터 기반 chunk 생성"""
chunks = []
current_chunk = ""
for node in data_structure:
node_text = f"{node['title']}: {node['content']}\n"
if len(current_chunk) + len(node_text) > chunk_size:
chunks.append(current_chunk)
current_chunk = node_text
else:
current_chunk += node_text
if current_chunk:
chunks.append(current_chunk)
return chunks
3. 임베딩 모델 선택 및 비교
3.1 모델 비교 표
| 모델 | 크기 | 성능 (MTEB) | 속도 | 비용 |
|---|---|---|---|---|
| all-MiniLM-L6-v2 | 80MB | 76.3 | 빠름 | 무료 |
| BGE-M3 | 120MB | 79.4 | 중간 | 무료 |
| sentence-t5 | 170MB | 81.2 | 느림 | 무료 |
| Qwen3.6 Embedding | 200MB | 82.7 | 중간 | 유료 |
# 임베딩 모델 비교 코드
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
def compare_embeddings(models, sentences):
results = {}
for name, model in models.items():
embeddings = model.encode(sentences)
similarity_matrix = cosine_similarity(embeddings)
results[name] = similarity_matrix
return results
# 예시 사용
models = {
"all-MiniLM-L6-v2": SentenceTransformer("all-MiniLM-L6-v2"),
"BGE-M3": SentenceTransformer("BAAI/bge-m3"),
}
sentences = ["Hello world", "Goodbye world", "Hello everyone"]
similarities = compare_embeddings(models, sentences)
4. 벡터 데이터베이스 비교 (Chroma, Qdrant, pgvector, Milvus)
| 데이터베이스 | 장점 | 단점 |
|---|---|---|
| Chroma | 간단한 설치, Python 친화적 | 대규모 데이터 처리 불리 |
| Qdrant | 고성능, 다양한 검색 옵션 | 복잡한 설정 필요 |
| pgvector | PostgreSQL 통합, 안정적 | 복잡한 설치 및 관리 |
| Milvus | 고성능, 분산 시스템 | 복잡한 설정 및 운영 |
# Chroma 예시
import chromadb
from chromadb import Client
client = Client()
collection = client.create_collection("docs")
# Chroma에 문서 추가
def add_to_chroma(collection, documents):
ids = [str(i) for i in range(len(documents))]
collection.add(
documents=documents,
ids=ids
)
# 검색
def search_chroma(collection, query, k=5):
results = collection.query(
query_texts=[query],
n_results=k
)
return results['documents'][0]
# Qdrant 예시
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue
client = QdrantClient(host="localhost", port=6333)
collection_name = "docs"
def search_qdrant(collection, query, k=5):
results = client.search(
collection_name=collection_name,
query_vector=query,
limit=k
)
return [hit.payload['content'] for hit in results]
5. 전체 RAG 파이프라인 코드
import chromadb
from sentence_transformers import SentenceTransformer
from langchain_openai import ChatOpenAI
import numpy as np
class SimpleRAG:
def __init__(self, model_name="all-MiniLM-L6-v2"):
self.embedding_model = SentenceTransformer(model_name)
self.vector_db = chromadb.Client()
self.collection = self.vector_db.get_or_create_collection("docs")
self.llm = ChatOpenAI(model="gpt-3.5-turbo")
def add_documents(self, documents, ids):
embeddings = self.embedding_model.encode(documents)
self.collection.add(
documents=documents,
embeddings=embeddings.tolist(),
ids=ids
)
def search(self, query, k=5):
query_embedding = self.embedding_model.encode([query])
results = self.collection.query(
query_embeddings=query_embedding.tolist(),
n_results=k
)
return results['documents'][0]
def generate_answer(self, query):
retrieved_docs = self.search(query)
context = "\n".join(retrieved_docs)
prompt = f"Context: {context}\n\nQuestion: {query}"
response = self.llm.invoke(prompt)
return response.content
# 사용 예시
rag = SimpleRAG()
rag.add_documents([
"Python은 인터프리터 기반의 고급 프로그래밍 언어입니다.",
"Java는 객체지향 프로그래밍 언어로, 안정적인 플랫폼 독립성을 제공합니다."
], ["doc1", "doc2"])
answer = rag.generate_answer("Python의 특징은 무엇인가요?")
print(answer)
6. 고급 기능 (Query Transformation, Hybrid Search, Re-ranking)
6.1 Query Transformation
질문을 더 검색 가능한 형태로 변환
python
def transform_query(query):
# 예:
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($7)
Top comments (0)