Table of Contents
RAG 챗봇 구축 가이드
개요
RAG(Retrieval-Augmented Generation) 챗봇은 외부 지식베이스를 활용해 더 정확하고 최신 정보를 제공하는 AI 챗봇입니다. 이 가이드는 RAG 챗봇을 구축하는 과정을 단계별로 설명합니다.
RAG란 무엇인가?
RAG는 검색(Retrieval)과 생성(Generation)을 결합한 접근 방식입니다. 사용자 질문에 대해 관련 문서를 먼저 검색하고, 그 정보를 바탕으로 답변을 생성합니다.
RAG의 장점
- 정확성: 실제 문서에 기반한 답변 제공
- 최신성: 지식베이스 업데이트로 최신 정보 반영
- 투명성: 답변의 출처를 명확히 제시
- 비용 효율성: 모델 재학습 없이 지식 확장 가능
시스템 아키텍처
사용자 질문 → 임베딩 → 벡터 검색 → 문서 검색 → 컨텍스트 + 질문 → LLM → 답변주요 구성 요소
- 문서 처리 파이프라인: 원본 문서를 청크 단위로 분할하고 벡터화
- 벡터 데이터베이스: 문서 임베딩을 저장하고 유사도 검색 수행
- 검색 엔진: 사용자 질문과 관련된 문서 청크를 찾아 반환
- 생성 모델: 검색된 컨텍스트를 활용해 최종 답변 생성
단계별 구축 가이드
1단계: 환경 설정
# 필요한 라이브러리 설치
pip install langchain chromadb openai sentence-transformers pypdf
pip install -U langchain-community
pip install -U langchain-openai2단계: 문서 전처리
from langchain_community.document_loaders import PyPDFLoader
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
def load_documents(file_path):
"""문서 로드"""
if file_path.endswith('.pdf'):
loader = PyPDFLoader(file_path)
else:
loader = TextLoader(file_path)
return loader.load()
def split_documents(documents):
"""문서를 청크로 분할"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len
)
return text_splitter.split_documents(documents)3단계: 벡터 데이터베이스 구축
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import OpenAIEmbeddings
# from langchain_openai import OpenAIEmbeddings
def create_vectorstore(documents):
"""벡터 저장소 생성"""
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
documents=documents,
embedding=embeddings,
persist_directory="./chroma_db"
)
return vectorstore4단계: 검색 시스템 구현
def create_retriever(vectorstore, search_type="similarity", k=5):
"""검색기 생성"""
retriever = vectorstore.as_retriever(
search_type=search_type,
search_kwargs={"k": k}
)
return retriever5단계: RAG 체인 구성
from langchain.chains import RetrievalQA
from langchain_community.llms import OpenAI
def create_rag_chain(retriever):
"""RAG 체인 생성"""
llm = OpenAI(temperature=0)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
return_source_documents=True
)
return qa_chain6단계: 챗봇 인터페이스 구현
import os
from langchain_community.document_loaders import PyPDFLoader
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain_openai import OpenAI
from langchain_openai import ChatOpenAI
os.environ["OPENAI_API_KEY"] = "your-openai-api-key-here"
def load_documents(file_path):
"""문서 로드"""
if file_path.endswith('.pdf'):
loader = PyPDFLoader(file_path)
else:
loader = TextLoader(file_path)
return loader.load()
def split_documents(documents):
"""문서를 청크로 분할"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len
)
return text_splitter.split_documents(documents)
def create_vectorstore(documents):
"""벡터 저장소 생성"""
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
documents=documents,
embedding=embeddings,
persist_directory="./chroma_db"
)
return vectorstore
def create_retriever(vectorstore, search_type="similarity", k=5):
"""검색기 생성"""
retriever = vectorstore.as_retriever(
search_type=search_type,
search_kwargs={"k": k}
)
return retriever
def create_rag_chain(retriever):
"""RAG 체인 생성"""
# llm = OpenAI(temperature=0)
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0,
max_tokens=200 # 응답 길이 제한
)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
return_source_documents=True
)
return qa_chain
class RAGChatbot:
def __init__(self, documents_path):
# 문서 로드 및 전처리
documents = load_documents(documents_path)
splits = split_documents(documents)
# 벡터 저장소 생성
self.vectorstore = create_vectorstore(splits)
# 검색기 및 체인 생성
retriever = create_retriever(self.vectorstore)
self.qa_chain = create_rag_chain(retriever)
def chat(self, question):
"""사용자 질문에 대한 답변 생성"""
result = self.qa_chain.invoke({"query": question})
return {
"answer": result["result"],
"sources": result["source_documents"]
}
# 사용 예시
chatbot = RAGChatbot("./law.pdf")
response = chatbot.chat("성년의 기준이 되는 나이가 어떻게 되지?")
print(response["answer"])성능 최적화
여기서부터 아래의 문서는 인공지능에 의해 작성되었으며, 테스트 되지 않은 코드입니다.
청크 크기 조정
# 다양한 청크 크기 실험
chunk_sizes = [500, 1000, 1500, 2000]
for size in chunk_sizes:
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=size,
chunk_overlap=size//5
)
# 성능 평가하이브리드 검색
from langchain.retrievers import EnsembleRetriever
from langchain.retrievers import BM25Retriever
def create_hybrid_retriever(documents, vectorstore):
"""하이브리드 검색기 생성"""
# 키워드 기반 검색
bm25_retriever = BM25Retriever.from_documents(documents)
# 의미 기반 검색
vector_retriever = vectorstore.as_retriever()
# 앙상블 검색기
ensemble_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, vector_retriever],
weights=[0.5, 0.5]
)
return ensemble_retriever재랭킹 시스템
from langchain.document_transformers import LongContextReorder
def rerank_documents(documents, query):
"""문서 재랭킹"""
reordering = LongContextReorder()
reordered_docs = reordering.transform_documents(documents)
return reordered_docs고급 기능
대화 메모리 추가
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationalRetrievalChain
def create_conversational_chain(retriever):
"""대화형 RAG 체인 생성"""
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
qa_chain = ConversationalRetrievalChain.from_llm(
llm=OpenAI(temperature=0),
retriever=retriever,
memory=memory
)
return qa_chain실시간 문서 업데이트
def update_vectorstore(vectorstore, new_documents):
"""벡터 저장소 업데이트"""
splits = split_documents(new_documents)
vectorstore.add_documents(splits)
vectorstore.persist()답변 품질 평가
from langchain.evaluation.qa import QAEvalChain
def evaluate_rag_system(qa_chain, test_questions):
"""RAG 시스템 평가"""
eval_chain = QAEvalChain.from_llm(OpenAI(temperature=0))
results = []
for question in test_questions:
result = qa_chain({"query": question})
evaluation = eval_chain.evaluate(
question,
result["result"],
result["source_documents"]
)
results.append(evaluation)
return results배포 및 모니터링
FastAPI를 이용한 API 서버
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
chatbot = RAGChatbot("./documents/")
class ChatRequest(BaseModel):
question: str
@app.post("/chat")
async def chat_endpoint(request: ChatRequest):
response = chatbot.chat(request.question)
return response로깅 및 모니터링
import logging
from datetime import datetime
def setup_logging():
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('rag_chatbot.log'),
logging.StreamHandler()
]
)
def log_interaction(question, answer, sources):
"""사용자 상호작용 로그"""
logging.info(f"Question: {question}")
logging.info(f"Answer: {answer}")
logging.info(f"Sources: {len(sources)} documents")주의사항 및 모범 사례
보안 고려사항
- API 키 및 민감한 정보는 환경 변수로 관리
- 사용자 입력 검증 및 sanitization 필수
- 접근 권한 및 인증 시스템 구현
성능 최적화
- 캐싱 시스템 도입으로 응답 시간 단축
- 비동기 처리를 통한 동시 요청 처리
- 벡터 데이터베이스 인덱싱 최적화
데이터 품질 관리
- 정기적인 문서 업데이트 및 품질 검토
- 중복 문서 제거 및 일관성 유지
- 사용자 피드백을 통한 지속적인 개선
결론
RAG 챗봇은 기존 LLM의 한계를 극복하고 도메인 특화된 고품질 답변을 제공하는 강력한 솔루션입니다. 이 가이드를 통해 기본적인 RAG 시스템부터 고급 기능까지 단계별로 구축할 수 있습니다.
성공적인 RAG 챗봇 구축을 위해서는 지속적인 모니터링과 개선이 필요하며, 사용자 피드백을 적극적으로 반영해 시스템의 품질을 향상시켜야 합니다.