RAG 시스템 실전 구축 (v8)
실제 서비스에서 사용할 수 있는 RAG 시스템 구축 가이드
1. RAG 기초: 검색 → 보강 → 생성 루프
Retrieval-Augmented Generation(RAG)은 검색과 생성을 결합한 시스템입니다. 다음의 루프를 따릅니다:
- 검색 (Retrieval): 사용자 질문에 관련된 문서 조각 검색
- 보강 (Augmentation): 검색된 문서를 프롬프트에 포함
- 생성 (Generation): LLM이 보강된 프롬프트로 응답 생성
이러한 구조는 LLM의 지식 범위를 확장하고, 특정 도메인에 대한 정확한 답변을 가능하게 합니다.
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
class SimpleRAGChain:
def __init__(self, retriever, llm):
self.retriever = retriever
self.llm = llm
self.prompt = PromptTemplate.from_template(
"다음의 문서를 바탕으로 질문에 답하세요:\n\n{context}\n\n질문: {question}"
)
def invoke(self, question):
docs = self.retriever.invoke(question)
context = "\n\n".join([doc.page_content for doc in docs])
return self.llm.invoke({
"question": question,
"context": context
})
2. 청킹 전략 (Chunking Strategies)
문서를 청킹하는 방법에 따라 검색 성능이 크게 달라집니다.
2.1 문맥 기반 청킹 (Semantic Chunking)
- 특징: 의미 단위로 청킹
- 장점: 의미 있는 단위로 분할되어 검색 정확도 향상
- 구현 예시:
from langchain_text_splitters import RecursiveCharacterTextSplitter
from sentence_transformers import SentenceTransformer
import numpy as np
class SemanticChunker:
def __init__(self, model_name="all-MiniLM-L6-v2"):
self.model = SentenceTransformer(model_name)
def chunk(self, text, chunk_size=512):
# 임베딩 계산을 통한 의미적 단위 분할
sentences = text.split('. ')
embeddings = self.model.encode(sentences)
# 유사도 기반으로 문장 그룹화
chunks = []
current_chunk = []
current_embedding = np.zeros(embeddings[0].shape)
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 > 0.85 and len(" ".join(current_chunk)) < chunk_size:
current_chunk.append(sentence)
else:
chunks.append(" ".join(current_chunk))
current_chunk = [sentence]
current_embedding = embedding
if current_chunk:
chunks.append(" ".join(current_chunk))
return chunks
2.2 재귀적 청킹 (Recursive Chunking)
- 특징: 텍스트의 구조적 요소를 기반으로 분할
- 예시 코드:
from langchain_text_splitters import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
separators=["\n\n", "\n", " ", ""]
)
2.3 에이전트 기반 청킹 (Agentic Chunking)
- 특징: 문서 구조와 내용을 분석하여 의미 있는 단위로 분할
- 활용: 복잡한 문서 구조 분석에 적합
3. 임베딩 모델 선택 및 비교
3.1 주요 모델 비교
| 모델 | 크기 | 속도 | 정확도 |
|---|---|---|---|
| all-MiniLM-L6-v2 | 10MB | 빠름 | 중간 |
| BGE-M3 | 13MB | 중간 | 높음 |
| e5-base-v2 | 14MB | 빠름 | 매우 높음 |
3.2 성능 비교 코드
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
def compare_embeddings():
model1 = SentenceTransformer("all-MiniLM-L6-v2")
model2 = SentenceTransformer("BGE-M3")
# 테스트 문장
sentences = [
"기계 학습은 데이터를 기반으로 예측을 만든다.",
"인공지능은 기계를 학습시키는 기술이다.",
"이 기술은 빠르게 발전하고 있다."
]
# 임베딩 생성
emb1 = model1.encode(sentences)
emb2 = model2.encode(sentences)
# 유사도 계산
sim1 = cosine_similarity(emb1[:2], emb1[1:3])
sim2 = cosine_similarity(emb2[:2], emb2[1:3])
print("all-MiniLM-L6-v2 유사도:", sim1)
print("BGE-M3 유사도:", sim2)
compare_embeddings()
4. 벡터 데이터베이스 비교
| 데이터베이스 | 특징 | 성능 |
|---|---|---|
| Chroma | 로컬 저장, 간편 | 중간 |
| Qdrant | 고성능, 클라우드 호환 | 높음 |
| pgvector | PostgreSQL 확장 | 높음 |
| Milvus | 분산, 대규모 | 매우 높음 |
4.1 Chroma 구현 예제
import chromadb
from chromadb.config import Settings
# Chroma 클라이언트 초기화
client = chromadb.Client(Settings(
chroma_api_impl="local",
chroma_db_impl="duckdb"
))
# 컬렉션 생성
collection = client.get_or_create_collection("my_collection")
# 문서 추가
collection.add(
documents=["문서1 내용...", "문서2 내용..."],
metadatas=[{"source": "doc1"}, {"source": "doc2"}],
ids=["doc1", "doc2"]
)
# 검색
results = collection.query(
query_texts=["검색어"],
n_results=3
)
4.2 Qdrant 구현 예제
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition
client = QdrantClient(host="localhost", port=6333)
collection_name = "rag_collection"
# 벡터 저장
client.upsert(
collection_name=collection_name,
points=[
{
"id": 1,
"vector": [0.1, 0.2, 0.3],
"payload": {"text": "문서 내용"}
}
]
)
# 검색
response = client.search(
collection_name=collection_name,
query_vector=[0.1, 0.2, 0.3],
limit=5
)
5. 전체 RAG 파이프라인 코드
python
from langchain_community.vectorstores import Chroma
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_core.retrievers import BaseRetriever
from langchain_core.callbacks import CallbackManagerForRetrieverRun
class RAGPipeline:
def __init__(self):
# 1. 임베딩 모델 초기화
self.embeddings = HuggingFaceEmbeddings(
model_name="BGE-M3",
model_kwargs={"device": "cuda"}
)
# 2. 벡터 저장소 초기화 (Chroma)
self.db = Chroma(
persist_directory="chroma_db",
embedding_function=self.embeddings
)
# 3. 검색기 초기화
self.retriever = self.db.as_retriever(
search_type="similarity",
search_kwargs={"k": 5}
)
# 4. LLM 초기화
from langchain_huggingface import HuggingFacePipeline
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
model_id = "Qwen/Qwen2-7B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_id)
model = AutoModelForCausalLM.from_pretrained(
model_id,
torch_dtype="auto",
device_map="auto"
)
pipe = pipeline(
"text-generation",
model=model,
tokenizer=tokenizer,
max_new_tokens=512
)
self.llm = HuggingFacePipeline(pipeline=
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($7)
Top comments (0)