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)
Top comments (0)