RAG 시스템 구축 가이드

RAG 시스템 구축 단계

RAG 시스템을 처음부터 구축하는 전체 과정을 단계별로 설명합니다.

1. 환경 설정

필수 라이브러리 설치

# Python 가상환경 생성
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 기본 라이브러리 설치
pip install langchain langchain-community langchain-openai
pip install chromadb  # 벡터 데이터베이스
pip install openai    # OpenAI API
pip install tiktoken  # 토큰 카운팅
pip install pypdf     # PDF 처리
pip install python-dotenv

환경 변수 설정

# .env 파일 생성
OPENAI_API_KEY=your-api-key-here

2. 문서 준비 및 로딩

2.1 문서 로더 선택

from langchain_community.document_loaders import (
    PyPDFLoader,
    TextLoader,
    DirectoryLoader,
    CSVLoader,
    UnstructuredMarkdownLoader
)

# PDF 파일 로드
loader = PyPDFLoader("document.pdf")
documents = loader.load()

# 디렉토리 내 모든 문서 로드
loader = DirectoryLoader(
    "data/",
    glob="**/*.txt",
    loader_cls=TextLoader
)
documents = loader.load()

# Markdown 파일 로드
loader = UnstructuredMarkdownLoader("README.md")
documents = loader.load()

2.2 다양한 소스 통합

from langchain_community.document_loaders import (
    WebBaseLoader,
    GitbookLoader,
    NotionDirectoryLoader
)

# 웹 페이지 크롤링
web_loader = WebBaseLoader("https://example.com/docs")
web_docs = web_loader.load()

# Notion 문서 로드
notion_loader = NotionDirectoryLoader("notion_export/")
notion_docs = notion_loader.load()

# 모든 문서 통합
all_documents = documents + web_docs + notion_docs

3. 문서 분할 (Chunking)

3.1 기본 텍스트 분할

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 재귀적 문자 분할기 (가장 많이 사용)
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,        # 청크 크기
    chunk_overlap=200,      # 오버랩 크기
    length_function=len,
    separators=["\n\n", "\n", " ", ""]  # 우선순위별 구분자
)

chunks = text_splitter.split_documents(documents)
print(f"총 청크 수: {len(chunks)}")

3.2 코드 전용 분할기

from langchain.text_splitter import (
    Language,
    RecursiveCharacterTextSplitter
)

# Python 코드 분할
python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=1000,
    chunk_overlap=100
)

python_chunks = python_splitter.split_documents(python_docs)

3.3 의미 기반 분할

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai.embeddings import OpenAIEmbeddings

# 의미론적 유사성 기반 분할
semantic_chunker = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile"
)

semantic_chunks = semantic_chunker.split_documents(documents)

4. 임베딩 생성

4.1 OpenAI 임베딩

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",  # 또는 text-embedding-3-large
    openai_api_key=os.getenv("OPENAI_API_KEY")
)

# 단일 텍스트 임베딩
vector = embeddings.embed_query("Hello, world!")
print(f"벡터 차원: {len(vector)}")

4.2 오픈소스 임베딩 모델

from langchain_community.embeddings import HuggingFaceEmbeddings

# 무료 오픈소스 모델 사용
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)

# 한국어 특화 모델
embeddings = HuggingFaceEmbeddings(
    model_name="jhgan/ko-sroberta-multitask"
)

5. 벡터 데이터베이스 구축

5.1 ChromaDB (로컬)

from langchain_community.vectorstores import Chroma

# 벡터 스토어 생성
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"  # 저장 경로
)

# 저장
vectorstore.persist()

# 기존 DB 로드
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

5.2 Pinecone (클라우드)

from langchain_community.vectorstores import Pinecone
import pinecone

# Pinecone 초기화
pinecone.init(
    api_key=os.getenv("PINECONE_API_KEY"),
    environment="us-west1-gcp"
)

# 인덱스 생성 및 데이터 추가
index_name = "my-rag-index"
vectorstore = Pinecone.from_documents(
    chunks,
    embeddings,
    index_name=index_name
)

5.3 FAISS (로컬, 고성능)

from langchain_community.vectorstores import FAISS

# FAISS 벡터 스토어 생성
vectorstore = FAISS.from_documents(chunks, embeddings)

# 로컬 저장
vectorstore.save_local("faiss_index")

# 로드
vectorstore = FAISS.load_local(
    "faiss_index",
    embeddings,
    allow_dangerous_deserialization=True
)

6. 검색기 구성

6.1 기본 유사도 검색

# 벡터 스토어를 검색기로 변환
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}  # 상위 4개 문서 반환
)

# 검색 테스트
docs = retriever.get_relevant_documents("RAG란 무엇인가?")
for doc in docs:
    print(doc.page_content[:100])
    print("---")

6.2 MMR (Maximum Marginal Relevance) 검색

# 다양성을 고려한 검색
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 4,
        "fetch_k": 20,      # 초기 후보 수
        "lambda_mult": 0.5  # 다양성 가중치 (0~1)
    }
)

6.3 하이브리드 검색 (BM25 + Vector)

from langchain.retrievers import BM25Retriever, EnsembleRetriever

# BM25 키워드 검색기
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 4

# 벡터 검색기
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# 하이브리드 검색기
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.5, 0.5]  # 각 검색기 가중치
)

7. LLM 설정

7.1 OpenAI 모델

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0,  # 일관성 있는 답변
    openai_api_key=os.getenv("OPENAI_API_KEY")
)

7.2 오픈소스 모델 (Ollama)

# Ollama 설치 및 모델 다운로드
ollama pull llama3

from langchain_community.llms import Ollama

llm = Ollama(model="llama3")

8. RAG 체인 구성

8.1 기본 RetrievalQA 체인

from langchain.chains import RetrievalQA

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True,
    verbose=True
)

# 질문하기
result = qa_chain({"query": "RAG의 장점은 무엇인가요?"})
print(result["result"])
print("\n출처:")
for doc in result["source_documents"]:
    print(f"- {doc.metadata}")

8.2 커스텀 프롬프트 사용

from langchain.prompts import PromptTemplate
from langchain.chains import RetrievalQA

prompt_template = """당신은 친절한 AI 어시스턴트입니다.
아래의 컨텍스트를 사용하여 질문에 답변하세요.
답변을 모른다면 모른다고 솔직히 말하세요.

컨텍스트:
{context}

질문: {question}

답변:"""

PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    chain_type_kwargs={"prompt": PROMPT}
)

8.3 대화형 RAG (ConversationalRetrievalChain)

from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory

# 대화 메모리
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    output_key="answer"
)

# 대화형 체인
conversational_chain = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=retriever,
    memory=memory,
    return_source_documents=True
)

# 대화하기
response1 = conversational_chain({"question": "RAG란 무엇인가요?"})
print(response1["answer"])

response2 = conversational_chain({"question": "그것의 장점은?"})
print(response2["answer"])

9. 완전한 RAG 시스템 예제

import os
from dotenv import load_dotenv
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA

# 환경 변수 로드
load_dotenv()

class RAGSystem:
    def __init__(self, data_dir, persist_dir="./chroma_db"):
        self.data_dir = data_dir
        self.persist_dir = persist_dir
        self.embeddings = OpenAIEmbeddings()
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.vectorstore = None
        self.qa_chain = None

    def load_documents(self):
        """문서 로드"""
        loader = DirectoryLoader(
            self.data_dir,
            glob="**/*.txt",
            loader_cls=TextLoader
        )
        documents = loader.load()
        print(f"로드된 문서 수: {len(documents)}")
        return documents

    def split_documents(self, documents):
        """문서 분할"""
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200
        )
        chunks = text_splitter.split_documents(documents)
        print(f"생성된 청크 수: {len(chunks)}")
        return chunks

    def create_vectorstore(self, chunks):
        """벡터 스토어 생성"""
        self.vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory=self.persist_dir
        )
        self.vectorstore.persist()
        print("벡터 스토어 생성 완료")

    def load_vectorstore(self):
        """기존 벡터 스토어 로드"""
        self.vectorstore = Chroma(
            persist_directory=self.persist_dir,
            embedding_function=self.embeddings
        )
        print("벡터 스토어 로드 완료")

    def setup_qa_chain(self):
        """QA 체인 설정"""
        retriever = self.vectorstore.as_retriever(
            search_kwargs={"k": 4}
        )
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=retriever,
            return_source_documents=True
        )
        print("QA 체인 설정 완료")

    def query(self, question):
        """질문하기"""
        result = self.qa_chain({"query": question})
        return result

    def build(self):
        """전체 시스템 구축"""
        documents = self.load_documents()
        chunks = self.split_documents(documents)
        self.create_vectorstore(chunks)
        self.setup_qa_chain()
        print("RAG 시스템 구축 완료!")

# 사용 예제
if __name__ == "__main__":
    # 시스템 구축
    rag = RAGSystem(data_dir="./data")
    rag.build()

    # 또는 기존 시스템 로드
    # rag = RAGSystem(data_dir="./data")
    # rag.load_vectorstore()
    # rag.setup_qa_chain()

    # 질문하기
    result = rag.query("RAG 시스템의 구성 요소는 무엇인가요?")
    print("\n답변:", result["result"])
    print("\n출처:")
    for i, doc in enumerate(result["source_documents"], 1):
        print(f"{i}. {doc.metadata.get('source', 'Unknown')}")