포스팅 개요
앞선 3편에서는 RAG의 개념, LangChain, 벡터 데이터베이스를 다뤘습니다. 이번 글에서는 이를 합쳐 실제로 동작하는 RAG 시스템을 만듭니다.
로컬 LLM 실행 도구인 Ollama를 사용해 외부 API 없이 내 컴퓨터에서만 돌아가는 개인용 AI 어시스턴트를 구성합니다. API 비용이 들지 않고, 인터넷 연결 없이 쓸 수 있으며, 데이터가 밖으로 나가지 않습니다.
1. Ollama 소개
Ollama는 오픈소스 LLM(Llama, Mistral, Gemma 등)을 로컬 환경에서 간단히 실행하게 해주는 도구입니다.
Ollama의 장점
- 비용: 모델을 내려받은 뒤에는 호출 횟수에 따른 API 비용이 없음
- 프라이버시: 데이터가 외부로 나가지 않음
- 오프라인: 모델을 내려받은 뒤에는 인터넷 없이 동작
- 커스터마이징: 모델 파라미터를 직접 조정
(Ollama vs 클라우드 API 비교)
2. Ollama 설치 및 설정
2-1. Ollama 설치
Linux:
curl -fsSL https://ollama.com/install.sh | shmacOS/Windows:
- ollama.com (opens in a new tab)에서 설치 파일을 내려받습니다. macOS는 Homebrew(
brew install ollama)로도 설치할 수 있습니다.
설치 확인:
ollama --version
# ollama version is 0.x.x2-2. 모델 다운로드
# 추천 모델들 (용량 순)
ollama pull gemma2:2b # 1.6GB - 가장 빠름, 기본 성능
ollama pull llama3.2:3b # 2.0GB - 균형잡힌 성능
ollama pull llama3.1:8b # 4.7GB - 고성능, 추천!
ollama pull qwen2.5:7b # 4.7GB - 다국어 지원
# 다운로드 확인
ollama list모델별 특징:
- gemma2:2b: 가장 가볍고 응답이 빠릅니다. 기본적인 대화에 적합합니다.
- llama3.2:3b: 크기와 성능이 중간 수준입니다.
- llama3.1:8b: 성능과 속도의 균형이 좋아 이 글에서 기본으로 사용합니다.
- qwen2.5:7b: Alibaba에서 만든 다국어 모델입니다. 한국어도 지원하며, 같은 크기의 다른 모델과 비교해 보고 고르면 좋습니다.
2-3. Ollama 기본 사용법
# 모델 실행 (대화 모드)
ollama run llama3.1:8b
# API 서버 직접 실행 (설치 앱이나 서비스가 이미 실행 중이면 필요 없음)
ollama serve
# 특정 포트에서 실행
OLLAMA_HOST=127.0.0.1:11435 ollama serve3. 프로젝트 환경 설정
3-1. 가상환경 및 라이브러리 설치
# 프로젝트 폴더 생성
mkdir my-local-rag
cd my-local-rag
# 가상환경 생성
python -m venv venv
# 가상환경 활성화
# Windows
venv\Scripts\activate
# macOS/Linux
source venv/bin/activate
# 필수 라이브러리 설치
pip install langchain
pip install chromadb
pip install sentence-transformers
pip install streamlit
pip install ollama
pip install pypdf
pip install docx2txt임베딩 모델(jhgan/ko-sroberta-multitask)은 처음 실행할 때 Hugging Face에서 내려받습니다. 이후에는 캐시를 사용하므로 오프라인에서도 동작합니다.
3-2. 프로젝트 구조 설정
my-local-rag/
├── documents/ # 검색 대상 문서들
│ ├── manuals/
│ ├── guides/
│ └── faqs/
├── database/ # 벡터 DB 저장소
├── models/ # 모델 관련 설정
├── utils/ # 유틸리티 함수들
├── app.py # 메인 어플리케이션
├── config.py # 설정 파일
└── requirements.txt # 라이브러리 목록# 폴더 구조 생성
mkdir documents documents/manuals documents/guides documents/faqs
mkdir database models utils4. RAG 시스템 핵심 구현
4-1. 설정 파일 (config.py)
import os
class Config:
"""RAG 시스템 설정"""
# Ollama 설정
OLLAMA_BASE_URL = "http://localhost:11434"
OLLAMA_MODEL = "llama3.1:8b" # 사용할 모델
# 임베딩 모델 설정
EMBEDDING_MODEL = "jhgan/ko-sroberta-multitask" # 한국어 특화
# 벡터 DB 설정
VECTOR_DB_PATH = "./database/chroma_db"
COLLECTION_NAME = "local_rag_collection"
# 텍스트 분할 설정
CHUNK_SIZE = 1000
CHUNK_OVERLAP = 200
# 검색 설정
SEARCH_K = 5 # 검색할 문서 수
# 문서 경로
DOCUMENTS_PATH = "./documents"
# 지원하는 파일 확장자
SUPPORTED_EXTENSIONS = [".txt", ".md", ".pdf", ".docx"]
@classmethod
def get_prompt_template(cls):
"""RAG용 프롬프트 템플릿"""
return """당신은 도움이 되는 AI 어시스턴트입니다. 제공된 문서들을 바탕으로 질문에 정확하게 답변해주세요.
참고 문서:
{context}
질문: {question}
답변 가이드라인:
1. 제공된 문서의 내용만을 바탕으로 답변하세요
2. 문서에 없는 내용은 추측하지 마세요
3. 불확실한 경우 "제공된 문서에서 해당 정보를 찾을 수 없습니다"라고 말씀해주세요
4. 가능한 한 구체적이고 자세한 답변을 제공해주세요
5. 한국어로 자연스럽게 답변해주세요
답변:"""4-2. 문서 처리 유틸리티 (utils/document_processor.py)
import os
import logging
from typing import List, Dict, Any
from langchain.document_loaders import (
DirectoryLoader,
TextLoader,
PyPDFLoader,
Docx2txtLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class DocumentProcessor:
"""문서 처리 전담 클래스"""
def __init__(self, chunk_size: int = 1000, chunk_overlap: int = 200):
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", ".", "!", "?", ",", " ", ""]
)
def load_documents_from_directory(self, directory_path: str) -> List[Document]:
"""디렉토리에서 다양한 형식의 문서 로딩"""
all_documents = []
if not os.path.exists(directory_path):
logger.warning(f"디렉토리가 존재하지 않습니다: {directory_path}")
return all_documents
# 파일 형식별 로더 매핑
loaders_config = [
("**/*.txt", TextLoader, {'encoding': 'utf-8'}),
("**/*.md", TextLoader, {'encoding': 'utf-8'}),
("**/*.pdf", PyPDFLoader, {}),
("**/*.docx", Docx2txtLoader, {})
]
for glob_pattern, loader_class, loader_kwargs in loaders_config:
try:
logger.info(f"📁 {glob_pattern} 파일들 로딩 중...")
loader = DirectoryLoader(
directory_path,
glob=glob_pattern,
loader_cls=loader_class,
loader_kwargs=loader_kwargs,
show_progress=True
)
documents = loader.load()
# 메타데이터 보강
for doc in documents:
file_path = doc.metadata.get('source', '')
doc.metadata.update({
'file_type': os.path.splitext(file_path)[1],
'file_name': os.path.basename(file_path),
'directory': os.path.dirname(file_path),
'char_count': len(doc.page_content),
'processed_at': str(os.path.getctime(file_path)) if os.path.exists(file_path) else 'unknown'
})
all_documents.extend(documents)
logger.info(f"✅ {len(documents)}개 문서 로딩 완료")
except Exception as e:
logger.error(f"❌ {glob_pattern} 로딩 중 오류: {e}")
logger.info(f"🎉 총 {len(all_documents)}개 문서 로딩 완료!")
return all_documents
def split_documents(self, documents: List[Document]) -> List[Document]:
"""문서를 청크로 분할"""
if not documents:
logger.warning("분할할 문서가 없습니다.")
return []
logger.info(f"✂️ {len(documents)}개 문서 분할 중...")
try:
doc_chunks = self.text_splitter.split_documents(documents)
# 청크별 메타데이터 추가
for i, chunk in enumerate(doc_chunks):
chunk.metadata.update({
'chunk_id': i,
'chunk_size': len(chunk.page_content),
'chunk_index': i
})
logger.info(f"✅ {len(doc_chunks)}개 청크 생성 완료")
# 청크 크기 통계
chunk_sizes = [len(chunk.page_content) for chunk in doc_chunks]
logger.info(f"📊 청크 크기 - 평균: {sum(chunk_sizes)/len(chunk_sizes):.0f}자, "
f"최소: {min(chunk_sizes)}자, 최대: {max(chunk_sizes)}자")
return doc_chunks
except Exception as e:
logger.error(f"❌ 문서 분할 중 오류: {e}")
return []
def get_document_stats(self, documents: List[Document]) -> Dict[str, Any]:
"""문서 통계 생성"""
if not documents:
return {"error": "문서가 없습니다"}
stats = {
"total_documents": len(documents),
"file_types": {},
"total_characters": 0,
"avg_doc_size": 0,
"directories": set()
}
for doc in documents:
# 파일 타입별 카운트
file_type = doc.metadata.get('file_type', 'unknown')
stats["file_types"][file_type] = stats["file_types"].get(file_type, 0) + 1
# 총 문자 수
char_count = len(doc.page_content)
stats["total_characters"] += char_count
# 디렉토리 정보
directory = doc.metadata.get('directory', '')
if directory:
stats["directories"].add(directory)
stats["avg_doc_size"] = stats["total_characters"] / len(documents)
stats["directories"] = list(stats["directories"])
return stats
# 사용 예제
if __name__ == "__main__":
processor = DocumentProcessor()
docs = processor.load_documents_from_directory("./documents")
chunks = processor.split_documents(docs)
stats = processor.get_document_stats(docs)
print("📊 문서 처리 결과:")
print(f"원본 문서: {stats['total_documents']}개")
print(f"생성된 청크: {len(chunks)}개")
print(f"파일 형식: {stats['file_types']}")
(문서 로딩 및 분할 처리 과정)
4-3. 벡터 DB 관리자 (utils/vector_store_manager.py)
import os
import logging
from typing import List, Optional, Dict, Any
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from langchain.schema import Document
logger = logging.getLogger(__name__)
class VectorStoreManager:
"""벡터 스토어 관리 클래스"""
def __init__(self,
embedding_model: str = "jhgan/ko-sroberta-multitask",
persist_directory: str = "./database/chroma_db"):
logger.info("🔧 벡터 스토어 매니저 초기화 중...")
self.persist_directory = persist_directory
# 임베딩 모델 초기화
logger.info(f"📥 임베딩 모델 로딩 중: {embedding_model}")
self.embeddings = HuggingFaceEmbeddings(
model_name=embedding_model,
model_kwargs={
'device': 'cpu', # GPU 사용시 'cuda'로 변경
'trust_remote_code': True
},
encode_kwargs={
'normalize_embeddings': True,
'batch_size': 32 # 배치 사이즈 조정으로 성능 최적화
}
)
self.vectorstore = None
logger.info("✅ 벡터 스토어 매니저 초기화 완료")
def create_or_load_vectorstore(self,
documents: Optional[List[Document]] = None) -> Chroma:
"""벡터 스토어 생성 또는 기존 것 로드"""
# 기존 DB가 있는지 확인
if os.path.exists(self.persist_directory) and os.listdir(self.persist_directory):
logger.info("📂 기존 벡터 DB 발견, 로딩 중...")
self.vectorstore = Chroma(
persist_directory=self.persist_directory,
embedding_function=self.embeddings,
collection_metadata={"hnsw:space": "cosine"}
)
# 새 문서가 있다면 추가
if documents:
logger.info(f"📄 {len(documents)}개 새 문서 추가 중...")
self.vectorstore.add_documents(documents)
self.vectorstore.persist()
logger.info("✅ 새 문서 추가 완료")
else:
# 새로운 벡터 DB 생성
if not documents:
raise ValueError("새 벡터 DB 생성을 위해서는 문서가 필요합니다.")
logger.info(f"🆕 새 벡터 DB 생성 중... ({len(documents)}개 문서)")
# 디렉토리 생성
os.makedirs(self.persist_directory, exist_ok=True)
self.vectorstore = Chroma.from_documents(
documents=documents,
embedding=self.embeddings,
persist_directory=self.persist_directory,
collection_metadata={"hnsw:space": "cosine"}
)
logger.info("✅ 새 벡터 DB 생성 완료")
return self.vectorstore
def search_similar_documents(self,
query: str,
k: int = 5,
filter_dict: Optional[Dict] = None) -> List[Document]:
"""유사 문서 검색"""
if not self.vectorstore:
raise ValueError("벡터 스토어가 초기화되지 않았습니다.")
try:
logger.info(f"🔍 검색 중: '{query}' (상위 {k}개)")
if filter_dict:
results = self.vectorstore.similarity_search(
query,
k=k,
filter=filter_dict
)
else:
results = self.vectorstore.similarity_search(query, k=k)
logger.info(f"✅ {len(results)}개 관련 문서 발견")
return results
except Exception as e:
logger.error(f"❌ 검색 중 오류: {e}")
return []
def search_with_scores(self,
query: str,
k: int = 5) -> List[tuple]:
"""점수와 함께 검색"""
if not self.vectorstore:
raise ValueError("벡터 스토어가 초기화되지 않았습니다.")
try:
results = self.vectorstore.similarity_search_with_score(query, k=k)
# 점수를 관련도로 변환 (거리 -> 유사도)
scored_results = []
for doc, distance in results:
similarity = max(0, 1 - distance) # 거리를 유사도로 변환
scored_results.append((doc, similarity))
return scored_results
except Exception as e:
logger.error(f"❌ 점수 기반 검색 중 오류: {e}")
return []
def get_vectorstore_info(self) -> Dict[str, Any]:
"""벡터 스토어 정보 조회"""
if not self.vectorstore:
return {"error": "벡터 스토어가 초기화되지 않았습니다."}
try:
collection = self.vectorstore._collection
doc_count = collection.count()
# 샘플 문서 조회
sample_docs = self.vectorstore.get(limit=5)
info = {
"document_count": doc_count,
"persist_directory": self.persist_directory,
"embedding_model": self.embeddings.model_name,
"collection_name": collection.name,
"sample_sources": [
metadata.get('source', 'unknown')
for metadata in sample_docs.get('metadatas', [])
]
}
return info
except Exception as e:
logger.error(f"❌ 벡터 스토어 정보 조회 중 오류: {e}")
return {"error": str(e)}
def delete_vectorstore(self) -> bool:
"""벡터 스토어 삭제"""
try:
import shutil
if os.path.exists(self.persist_directory):
shutil.rmtree(self.persist_directory)
logger.info("🗑️ 벡터 스토어 삭제 완료")
self.vectorstore = None
return True
except Exception as e:
logger.error(f"❌ 벡터 스토어 삭제 중 오류: {e}")
return False
# 사용 예제
if __name__ == "__main__":
manager = VectorStoreManager()
# 문서 로딩 (예시)
from document_processor import DocumentProcessor
processor = DocumentProcessor()
docs = processor.load_documents_from_directory("./documents")
chunks = processor.split_documents(docs)
# 벡터 스토어 생성
vectorstore = manager.create_or_load_vectorstore(chunks)
# 검색 테스트
results = manager.search_similar_documents("파이썬 설치 방법", k=3)
for doc in results:
print(f"📄 {doc.metadata.get('source', 'unknown')}")
print(f" {doc.page_content[:100]}...")
# 벡터 스토어 정보
info = manager.get_vectorstore_info()
print(f"\n📊 벡터 DB 정보: {info}")4-4. RAG 체인 구현 (utils/rag_chain.py)
import logging
from typing import List, Dict, Any, Optional
from langchain.llms import Ollama
from langchain.prompts import PromptTemplate
from langchain.chains import RetrievalQA
from langchain.schema import Document
logger = logging.getLogger(__name__)
class LocalRAGChain:
"""로컬 RAG 체인 구현"""
def __init__(self,
model_name: str = "llama3.1:8b",
base_url: str = "http://localhost:11434",
temperature: float = 0.7):
logger.info("🤖 RAG 체인 초기화 중...")
# Ollama LLM 초기화
logger.info(f"📡 Ollama 모델 연결 중: {model_name}")
self.llm = Ollama(
model=model_name,
base_url=base_url,
temperature=temperature,
# 추가 파라미터들
num_ctx=4096, # 컨텍스트 길이
num_predict=2048, # 최대 생성 토큰 수
repeat_penalty=1.1, # 반복 방지
top_k=40,
top_p=0.9
)
# 프롬프트 템플릿
self.prompt_template = PromptTemplate(
input_variables=["context", "question"],
template="""당신은 도움이 되는 AI 어시스턴트입니다. 제공된 문서들을 바탕으로 질문에 정확하게 답변해주세요.
참고 문서:
{context}
질문: {question}
답변 가이드라인:
1. 제공된 문서의 내용만을 바탕으로 답변하세요
2. 문서에 없는 내용은 추측하지 마세요
3. 불확실한 경우 "제공된 문서에서 해당 정보를 찾을 수 없습니다"라고 말씀해주세요
4. 가능한 한 구체적이고 자세한 답변을 제공해주세요
5. 한국어로 자연스럽게 답변해주세요
답변:"""
)
self.vectorstore = None
self.qa_chain = None
logger.info("✅ RAG 체인 초기화 완료")
def setup_retrieval_qa(self, vectorstore):
"""검색 기반 QA 체인 설정"""
logger.info("🔗 검색 기반 QA 체인 구성 중...")
self.vectorstore = vectorstore
# 검색기 설정
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={
"k": 5, # 상위 5개 문서 검색
}
)
# QA 체인 생성
self.qa_chain = RetrievalQA.from_chain_type(
llm=self.llm,
chain_type="stuff", # 모든 문서를 하나의 프롬프트에 포함
retriever=retriever,
chain_type_kwargs={"prompt": self.prompt_template},
return_source_documents=True, # 참조 문서 반환
verbose=False
)
logger.info("✅ 검색 기반 QA 체인 구성 완료")
def query(self, question: str) -> Dict[str, Any]:
"""질문 처리 및 답변 생성"""
if not self.qa_chain:
return {
"answer": "RAG 체인이 설정되지 않았습니다. setup_retrieval_qa()를 먼저 실행해주세요.",
"source_documents": [],
"error": "Chain not initialized"
}
try:
logger.info(f"❓ 질문 처리 중: '{question[:50]}...'")
# 질문 처리
result = self.qa_chain({"query": question})
# 결과 정리
response = {
"answer": result["result"].strip(),
"source_documents": [
{
"content": doc.page_content[:300] + "..." if len(doc.page_content) > 300 else doc.page_content,
"source": doc.metadata.get("source", "알 수 없음"),
"file_name": doc.metadata.get("file_name", "알 수 없음"),
"chunk_id": doc.metadata.get("chunk_id", 0)
}
for doc in result.get("source_documents", [])
],
"question": question,
"timestamp": str(pd.Timestamp.now()) if 'pd' in globals() else "unknown"
}
logger.info("✅ 답변 생성 완료")
return response
except Exception as e:
logger.error(f"❌ 질문 처리 중 오류: {e}")
return {
"answer": f"죄송합니다. 처리 중 오류가 발생했습니다: {str(e)}",
"source_documents": [],
"error": str(e)
}
def direct_llm_query(self, prompt: str) -> str:
"""벡터 검색 없이 LLM에 직접 질문"""
try:
logger.info("🤖 LLM 직접 호출 중...")
response = self.llm(prompt)
logger.info("✅ LLM 응답 완료")
return response.strip()
except Exception as e:
logger.error(f"❌ LLM 직접 호출 중 오류: {e}")
return f"오류가 발생했습니다: {str(e)}"
def batch_query(self, questions: List[str]) -> List[Dict[str, Any]]:
"""여러 질문 일괄 처리"""
results = []
for i, question in enumerate(questions, 1):
logger.info(f"📝 질문 {i}/{len(questions)} 처리 중...")
result = self.query(question)
results.append(result)
logger.info(f"🎉 {len(questions)}개 질문 일괄 처리 완료")
return results
def get_chain_info(self) -> Dict[str, Any]:
"""RAG 체인 정보 조회"""
info = {
"llm_model": getattr(self.llm, 'model', 'unknown'),
"llm_base_url": getattr(self.llm, 'base_url', 'unknown'),
"temperature": getattr(self.llm, 'temperature', 'unknown'),
"chain_initialized": self.qa_chain is not None,
"vectorstore_connected": self.vectorstore is not None
}
if self.vectorstore:
try:
collection = self.vectorstore._collection
info["vector_db_doc_count"] = collection.count()
except:
info["vector_db_doc_count"] = "unknown"
return info
# 사용 예제
if __name__ == "__main__":
from vector_store_manager import VectorStoreManager
from document_processor import DocumentProcessor
# 문서 처리
processor = DocumentProcessor()
documents = processor.load_documents_from_directory("./documents")
chunks = processor.split_documents(documents)
# 벡터 스토어 생성
vector_manager = VectorStoreManager()
vectorstore = vector_manager.create_or_load_vectorstore(chunks)
# RAG 체인 설정
rag_chain = LocalRAGChain()
rag_chain.setup_retrieval_qa(vectorstore)
# 테스트 질문들
test_questions = [
"파이썬 설치 방법을 알려주세요",
"라이브러리 사용법이 궁금합니다",
"에러가 발생했을 때 해결 방법은?"
]
# 질문별 답변
for question in test_questions:
print(f"\n{'='*60}")
print(f"❓ 질문: {question}")
result = rag_chain.query(question)
print(f"💬 답변: {result['answer']}")
print(f"\n📚 참고 문서 ({len(result['source_documents'])}개):")
for i, doc in enumerate(result['source_documents'], 1):
print(f"{i}. {doc['file_name']} (청크 #{doc['chunk_id']})")
print(f" {doc['content'][:80]}...")5. 웹 인터페이스 구현 (app.py)
import streamlit as st
import os
import time
import logging
from datetime import datetime
from typing import Dict, Any
# 유틸리티 함수들 import
from utils.document_processor import DocumentProcessor
from utils.vector_store_manager import VectorStoreManager
from utils.rag_chain import LocalRAGChain
from config import Config
# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 페이지 설정
st.set_page_config(
page_title="로컬 RAG 시스템",
page_icon="🤖",
layout="wide",
initial_sidebar_state="expanded"
)
class LocalRAGApp:
"""Streamlit 기반 로컬 RAG 애플리케이션"""
def __init__(self):
self.config = Config()
self.init_session_state()
def init_session_state(self):
"""세션 상태 초기화"""
if 'rag_chain' not in st.session_state:
st.session_state.rag_chain = None
if 'vector_manager' not in st.session_state:
st.session_state.vector_manager = None
if 'documents_loaded' not in st.session_state:
st.session_state.documents_loaded = False
if 'chat_history' not in st.session_state:
st.session_state.chat_history = []
def render_sidebar(self):
"""사이드바 렌더링"""
with st.sidebar:
st.header("🛠️ 시스템 설정")
# Ollama 연결 상태 체크
self.check_ollama_connection()
# 모델 선택
st.subheader("🤖 LLM 모델 설정")
available_models = ["llama3.1:8b", "gemma2:2b", "llama3.2:3b", "qwen2.5:7b"]
selected_model = st.selectbox(
"모델 선택",
available_models,
index=0 if self.config.OLLAMA_MODEL in available_models else 0
)
temperature = st.slider(
"Temperature (창의성)",
min_value=0.0,
max_value=1.0,
value=0.7,
step=0.1,
help="낮을수록 일관적, 높을수록 창의적"
)
# 문서 관리
st.subheader("📚 문서 관리")
# 파일 업로드
uploaded_files = st.file_uploader(
"문서 업로드",
accept_multiple_files=True,
type=['txt', 'md', 'pdf', 'docx'],
help="지원 형식: TXT, MD, PDF, DOCX"
)
# 업로드된 파일 처리
if uploaded_files:
if st.button("📄 업로드 파일 처리"):
self.process_uploaded_files(uploaded_files)
# 로컬 문서 로딩
if st.button("📁 로컬 문서 로딩"):
self.load_local_documents()
# RAG 시스템 초기화
if st.button("🚀 RAG 시스템 초기화"):
self.initialize_rag_system(selected_model, temperature)
# 시스템 상태 표시
self.show_system_status()
def check_ollama_connection(self):
"""Ollama 연결 상태 확인"""
try:
import requests
response = requests.get(f"{self.config.OLLAMA_BASE_URL}/api/tags", timeout=5)
if response.status_code == 200:
st.success("✅ Ollama 연결 성공")
models = response.json().get('models', [])
if models:
st.write(f"사용 가능한 모델: {len(models)}개")
else:
st.warning("⚠️ 다운로드된 모델이 없습니다")
else:
st.error("❌ Ollama 연결 실패")
except Exception as e:
st.error(f"❌ Ollama 연결 확인 중 오류: {str(e)}")
st.info("💡 'ollama serve' 명령으로 Ollama를 시작해주세요")
def process_uploaded_files(self, uploaded_files):
"""업로드된 파일 처리"""
with st.spinner("📤 파일 업로드 처리 중..."):
try:
# 임시 디렉토리에 파일 저장
temp_dir = "./temp_uploads"
os.makedirs(temp_dir, exist_ok=True)
saved_files = []
for uploaded_file in uploaded_files:
file_path = os.path.join(temp_dir, uploaded_file.name)
with open(file_path, "wb") as f:
f.write(uploaded_file.getbuffer())
saved_files.append(file_path)
st.success(f"✅ {len(saved_files)}개 파일 업로드 완료")
st.session_state.uploaded_files = saved_files
except Exception as e:
st.error(f"❌ 파일 업로드 중 오류: {str(e)}")
def load_local_documents(self):
"""로컬 문서 로딩"""
with st.spinner("📁 로컬 문서 로딩 중..."):
try:
processor = DocumentProcessor(
chunk_size=self.config.CHUNK_SIZE,
chunk_overlap=self.config.CHUNK_OVERLAP
)
# 문서 로딩
documents = processor.load_documents_from_directory(
self.config.DOCUMENTS_PATH
)
if not documents:
st.warning(f"⚠️ {self.config.DOCUMENTS_PATH}에서 문서를 찾을 수 없습니다")
return
# 문서 분할
chunks = processor.split_documents(documents)
# 세션에 저장
st.session_state.processed_documents = chunks
st.session_state.documents_loaded = True
# 통계 표시
stats = processor.get_document_stats(documents)
st.success(f"✅ 문서 로딩 완료: {stats['total_documents']}개 → {len(chunks)}개 청크")
except Exception as e:
st.error(f"❌ 문서 로딩 중 오류: {str(e)}")
def initialize_rag_system(self, model_name: str, temperature: float):
"""RAG 시스템 초기화"""
if not st.session_state.documents_loaded:
st.error("❌ 먼저 문서를 로딩해주세요")
return
with st.spinner("🚀 RAG 시스템 초기화 중..."):
try:
# 벡터 스토어 매니저 초기화
st.session_state.vector_manager = VectorStoreManager(
embedding_model=self.config.EMBEDDING_MODEL,
persist_directory=self.config.VECTOR_DB_PATH
)
# 벡터 스토어 생성/로드
vectorstore = st.session_state.vector_manager.create_or_load_vectorstore(
st.session_state.processed_documents
)
# RAG 체인 초기화
st.session_state.rag_chain = LocalRAGChain(
model_name=model_name,
base_url=self.config.OLLAMA_BASE_URL,
temperature=temperature
)
# 검색 기반 QA 설정
st.session_state.rag_chain.setup_retrieval_qa(vectorstore)
st.success("🎉 RAG 시스템 초기화 완료!")
except Exception as e:
st.error(f"❌ RAG 시스템 초기화 중 오류: {str(e)}")
def show_system_status(self):
"""시스템 상태 표시"""
st.subheader("📊 시스템 상태")
# 문서 로딩 상태
if st.session_state.documents_loaded:
st.success("✅ 문서 로딩 완료")
else:
st.warning("⏳ 문서 미로딩")
# RAG 체인 상태
if st.session_state.rag_chain:
st.success("✅ RAG 시스템 활성")
# 상세 정보
if st.session_state.vector_manager:
info = st.session_state.vector_manager.get_vectorstore_info()
st.write(f"📄 저장된 문서: {info.get('document_count', 0)}개")
else:
st.warning("⏳ RAG 시스템 대기")
def render_main_chat(self):
"""메인 채팅 인터페이스"""
st.title("🤖 로컬 RAG 시스템")
st.markdown("**Ollama 기반 개인용 AI 어시스턴트**")
if not st.session_state.rag_chain:
st.info("👈 사이드바에서 RAG 시스템을 초기화해주세요")
return
# 채팅 히스토리 표시
for message in st.session_state.chat_history:
with st.chat_message(message["role"]):
st.write(message["content"])
# 참고 문서 표시 (AI 응답인 경우)
if message["role"] == "assistant" and "sources" in message:
with st.expander("📚 참고 문서"):
for i, source in enumerate(message["sources"], 1):
st.write(f"**{i}. {source['file_name']}**")
st.write(f"```{source['content'][:200]}...```")
# 사용자 입력
if prompt := st.chat_input("질문을 입력하세요..."):
# 사용자 메시지 추가
st.session_state.chat_history.append({
"role": "user",
"content": prompt,
"timestamp": datetime.now().isoformat()
})
with st.chat_message("user"):
st.write(prompt)
# AI 응답 생성
with st.chat_message("assistant"):
with st.spinner("🤔 답변 생성 중..."):
result = st.session_state.rag_chain.query(prompt)
# 응답 표시
st.write(result["answer"])
# 참고 문서 표시
if result["source_documents"]:
with st.expander("📚 참고 문서"):
for i, doc in enumerate(result["source_documents"], 1):
st.write(f"**{i}. {doc['file_name']}**")
st.write(f"```{doc['content']}```")
# 히스토리에 추가
st.session_state.chat_history.append({
"role": "assistant",
"content": result["answer"],
"sources": result["source_documents"],
"timestamp": datetime.now().isoformat()
})
# 채팅 기록 관리
col1, col2 = st.columns(2)
with col1:
if st.button("🗑️ 채팅 기록 삭제"):
st.session_state.chat_history = []
st.rerun()
with col2:
if st.button("💾 채팅 기록 저장"):
self.save_chat_history()
def save_chat_history(self):
"""채팅 기록 저장"""
try:
import json
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"chat_history_{timestamp}.json"
with open(filename, 'w', encoding='utf-8') as f:
json.dump(st.session_state.chat_history, f, ensure_ascii=False, indent=2)
st.success(f"✅ 채팅 기록 저장 완료: {filename}")
except Exception as e:
st.error(f"❌ 채팅 기록 저장 중 오류: {str(e)}")
def render_analytics_tab(self):
"""분석 탭 렌더링"""
st.header("📈 시스템 분석")
if not st.session_state.vector_manager:
st.info("벡터 스토어가 초기화되지 않았습니다.")
return
# 벡터 DB 정보
col1, col2 = st.columns(2)
with col1:
st.subheader("📊 벡터 데이터베이스")
info = st.session_state.vector_manager.get_vectorstore_info()
st.metric("저장된 문서 수", info.get("document_count", 0))
st.write(f"**임베딩 모델:** {info.get('embedding_model', 'unknown')}")
st.write(f"**저장 위치:** {info.get('persist_directory', 'unknown')}")
with col2:
st.subheader("🤖 RAG 체인")
if st.session_state.rag_chain:
chain_info = st.session_state.rag_chain.get_chain_info()
st.write(f"**LLM 모델:** {chain_info.get('llm_model', 'unknown')}")
st.write(f"**Temperature:** {chain_info.get('temperature', 'unknown')}")
st.write(f"**연결 상태:** {'🟢 연결됨' if chain_info.get('chain_initialized') else '🔴 미연결'}")
# 채팅 통계
if st.session_state.chat_history:
st.subheader("💬 채팅 통계")
total_messages = len(st.session_state.chat_history)
user_messages = len([msg for msg in st.session_state.chat_history if msg["role"] == "user"])
col1, col2, col3 = st.columns(3)
with col1:
st.metric("총 메시지", total_messages)
with col2:
st.metric("사용자 질문", user_messages)
with col3:
st.metric("AI 응답", total_messages - user_messages)
# 검색 테스트
st.subheader("🔍 검색 테스트")
test_query = st.text_input("테스트 검색어를 입력하세요:")
if test_query and st.button("검색 실행"):
with st.spinner("검색 중..."):
results = st.session_state.vector_manager.search_with_scores(test_query, k=5)
st.write(f"**검색 결과 ({len(results)}개):**")
for i, (doc, score) in enumerate(results, 1):
st.write(f"**{i}. 유사도: {score:.3f}**")
st.write(f"출처: {doc.metadata.get('source', '알 수 없음')}")
st.write(f"내용: {doc.page_content[:200]}...")
st.write("---")
def run(self):
"""애플리케이션 실행"""
# 사이드바
self.render_sidebar()
# 메인 탭 구성
tab1, tab2 = st.tabs(["💬 채팅", "📈 분석"])
with tab1:
self.render_main_chat()
with tab2:
self.render_analytics_tab()
# 메인 실행
if __name__ == "__main__":
app = LocalRAGApp()
app.run()6. 실행 및 사용법
6-1. 시스템 시작하기
# 1. Ollama 서버 시작 (터미널 1, 이미 실행 중이면 생략)
ollama serve
# 2. 모델 다운로드 (터미널 2)
ollama pull llama3.1:8b
# 3. 문서 준비
# documents/ 폴더에 검색 대상 문서들 복사
# 4. 웹 애플리케이션 실행
streamlit run app.py6-2. 사용 단계
Step 1: 문서 준비
documents/
├── manuals/
│ ├── python_guide.txt
│ └── api_documentation.md
├── guides/
│ ├── installation_guide.pdf
│ └── troubleshooting.docx
└── faqs/
└── common_questions.txtStep 2: 시스템 초기화
- 웹 인터페이스 접속 (http://localhost:8501)
- 사이드바에서 "로컬 문서 로딩" 클릭
- "RAG 시스템 초기화" 클릭
Step 3: 질문하기
- 채팅창에 질문을 입력합니다.
- 답변 아래에서 참고 문서를 확인할 수 있습니다.
- 채팅 기록을 JSON 파일로 저장할 수 있습니다.
7. 고급 기능 및 최적화
7-1. 모델별 성능 비교
def benchmark_models():
"""다양한 모델 성능 비교"""
models_to_test = [
"gemma2:2b",
"llama3.2:3b",
"llama3.1:8b",
"qwen2.5:7b"
]
test_questions = [
"파이썬 설치 방법을 설명해주세요",
"API 사용 시 주의사항은 무엇인가요?",
"에러가 발생했을 때 해결 방법은?"
]
results = {}
for model in models_to_test:
print(f"\n🧪 {model} 테스트 중...")
# RAG 체인 초기화
rag_chain = LocalRAGChain(model_name=model)
rag_chain.setup_retrieval_qa(vectorstore)
model_results = []
for question in test_questions:
start_time = time.time()
result = rag_chain.query(question)
end_time = time.time()
response_time = end_time - start_time
model_results.append({
"question": question,
"answer_length": len(result["answer"]),
"response_time": response_time,
"source_count": len(result["source_documents"])
})
results[model] = model_results
# 평균 성능 계산
avg_time = sum(r["response_time"] for r in model_results) / len(model_results)
avg_length = sum(r["answer_length"] for r in model_results) / len(model_results)
print(f" 평균 응답시간: {avg_time:.2f}초")
print(f" 평균 답변길이: {avg_length:.0f}자")
return results
# 실행
# benchmark_results = benchmark_models()7-2. 메모리 최적화
def optimize_for_low_memory():
"""저사양 환경을 위한 최적화 설정"""
# 1. 작은 임베딩 모델 사용
embeddings = HuggingFaceEmbeddings(
model_name="sentence-transformers/all-MiniLM-L6-v2", # 더 작은 모델 (영어 중심이라 한국어 문서에는 품질이 떨어짐)
model_kwargs={
'device': 'cpu',
'trust_remote_code': True
},
encode_kwargs={
'batch_size': 16, # 배치 크기 줄임
'normalize_embeddings': True
}
)
# 2. 청크 크기 줄이기
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 더 작은 청크
chunk_overlap=50 # 오버랩 줄임
)
# 3. 검색 결과 수 제한
retriever = vectorstore.as_retriever(
search_kwargs={"k": 3} # 3개로 제한
)
# 4. Ollama 파라미터 조정
llm = Ollama(
model="gemma2:2b", # 가장 작은 모델
num_ctx=2048, # 컨텍스트 길이 줄임
num_predict=512, # 생성 토큰 줄임
temperature=0.7
)
return embeddings, text_splitter, retriever, llm7-3. 배치 처리 시스템
from datetime import datetime
class BatchRAGProcessor:
"""대량 질문 배치 처리"""
def __init__(self, rag_chain):
self.rag_chain = rag_chain
def process_question_file(self, file_path: str, output_path: str = None):
"""파일에서 질문을 읽어 배치 처리"""
# 질문 파일 읽기
with open(file_path, 'r', encoding='utf-8') as f:
questions = [line.strip() for line in f if line.strip()]
print(f"📝 {len(questions)}개 질문 배치 처리 시작...")
results = []
for i, question in enumerate(questions, 1):
print(f"처리 중: {i}/{len(questions)} - {question[:50]}...")
result = self.rag_chain.query(question)
results.append({
"question": question,
"answer": result["answer"],
"sources": [doc["file_name"] for doc in result["source_documents"]],
"timestamp": datetime.now().isoformat()
})
# 중간 저장 (10개마다)
if i % 10 == 0:
self.save_results(results, f"temp_results_{i}.json")
# 최종 결과 저장
if output_path:
self.save_results(results, output_path)
print(f"🎉 배치 처리 완료: {len(results)}개 결과 생성")
return results
def save_results(self, results: list, output_path: str):
"""결과를 JSON 파일로 저장"""
import json
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
print(f"💾 결과 저장 완료: {output_path}")
def generate_report(self, results: list) -> str:
"""처리 결과 리포트 생성"""
total_questions = len(results)
avg_answer_length = sum(len(r["answer"]) for r in results) / total_questions
# 가장 많이 참조된 소스 파일
source_count = {}
for result in results:
for source in result["sources"]:
source_count[source] = source_count.get(source, 0) + 1
top_sources = sorted(source_count.items(), key=lambda x: x[1], reverse=True)[:5]
report = f"""
📊 배치 처리 리포트
==================
📈 전체 통계:
- 처리된 질문 수: {total_questions}개
- 평균 답변 길이: {avg_answer_length:.0f}자
- 처리 완료 시간: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
📚 최다 참조 문서:
"""
for source, count in top_sources:
report += f"- {source}: {count}회 참조\n"
return report
# 사용 예시
# batch_processor = BatchRAGProcessor(rag_chain)
# results = batch_processor.process_question_file("questions.txt", "answers.json")
# report = batch_processor.generate_report(results)
# print(report)8. 문제 해결
8-1. 자주 발생하는 문제
"Ollama connection failed" 에러
# 해결 방법:
# 1. Ollama 서비스 시작
ollama serve
# 2. 방화벽 확인
# 3. 포트 11434가 사용 중인지 확인
netstat -an | grep 11434"CUDA out of memory" 에러
# CPU 모드로 강제 설정
embeddings = HuggingFaceEmbeddings(
model_name="jhgan/ko-sroberta-multitask",
model_kwargs={'device': 'cpu'} # GPU 대신 CPU 사용
)
# 또는 더 작은 모델 사용
embeddings = HuggingFaceEmbeddings(
model_name="sentence-transformers/all-MiniLM-L6-v2"
)"ChromaDB lock error"
# 기존 DB 삭제 후 재생성
import shutil
shutil.rmtree("./database/chroma_db")
# 또는 다른 경로 사용
vectorstore = Chroma(
persist_directory=f"./database/chroma_db_{int(time.time())}"
)8-2. 성능 최적화 가이드
응답 속도를 우선하는 설정
# 1. 작은 모델 사용
OLLAMA_MODEL = "gemma2:2b"
# 2. 검색 문서 수 줄이기
SEARCH_K = 3
# 3. 청크 크기 최적화
CHUNK_SIZE = 800
CHUNK_OVERLAP = 100
# 4. 임베딩 배치 크기 조정
encode_kwargs={'batch_size': 64}정확도를 우선하는 설정
# 1. 큰 모델 사용
OLLAMA_MODEL = "llama3.1:8b"
# 2. 더 많은 문서 검색
SEARCH_K = 7
# 3. 상세한 청킹
CHUNK_SIZE = 1200
CHUNK_OVERLAP = 300
# 4. MMR 검색 사용
retriever = vectorstore.as_retriever(
search_type="mmr",
search_kwargs={"k": 7, "lambda_mult": 0.25}
)9. 활용 예시와 확장
9-1. 회사 문서 관리 시스템
class CompanyRAGSystem(LocalRAGApp):
"""회사 전용 RAG 시스템"""
def __init__(self, company_name: str):
super().__init__()
self.company_name = company_name
def setup_department_collections(self):
"""부서별 컬렉션 설정"""
departments = {
"개발팀": "./documents/dev/",
"기획팀": "./documents/planning/",
"인사팀": "./documents/hr/",
"마케팅팀": "./documents/marketing/"
}
for dept, path in departments.items():
if os.path.exists(path):
# 부서별 벡터 스토어 생성
dept_vectorstore = Chroma(
collection_name=f"{self.company_name}_{dept}",
persist_directory=f"./database/{dept.lower()}_db"
)
st.session_state[f"{dept}_vectorstore"] = dept_vectorstore
def department_specific_query(self, query: str, department: str):
"""부서별 특화 질문"""
dept_vectorstore = st.session_state.get(f"{department}_vectorstore")
if not dept_vectorstore:
return "해당 부서 데이터가 준비되지 않았습니다."
# 부서 특화 프롬프트
dept_prompt = f"""
당신은 {self.company_name} {department}의 전문 AI 어시스턴트입니다.
{department}의 업무와 정책에 대해 정확한 정보를 제공해주세요.
참고 문서:
{{context}}
질문: {{question}}
답변: {department}의 관점에서 구체적이고 실용적인 답변을 제공해주세요.
"""
# 부서별 RAG 체인 실행
# ... 구현 생략9-2. 개인 지식 관리 시스템
class PersonalKnowledgeManager:
"""개인용 지식 관리 RAG"""
def __init__(self):
self.categories = {
"학습노트": "./documents/study/",
"프로젝트": "./documents/projects/",
"아이디어": "./documents/ideas/",
"참고자료": "./documents/references/"
}
def auto_categorize_documents(self, documents):
"""문서 자동 분류"""
for doc in documents:
content = doc.page_content.lower()
# 키워드 기반 분류
if any(word in content for word in ["학습", "공부", "tutorial", "guide"]):
doc.metadata["category"] = "학습노트"
elif any(word in content for word in ["프로젝트", "project", "개발"]):
doc.metadata["category"] = "프로젝트"
elif any(word in content for word in ["아이디어", "idea", "구상"]):
doc.metadata["category"] = "아이디어"
else:
doc.metadata["category"] = "참고자료"
return documents
def smart_search_suggestions(self, query: str) -> List[str]:
"""검색어 자동 추천"""
# 관련 키워드 생성
suggestions = []
if "파이썬" in query:
suggestions.extend(["파이썬 문법", "파이썬 라이브러리", "파이썬 예제"])
if "프로젝트" in query:
suggestions.extend(["프로젝트 구조", "프로젝트 관리", "프로젝트 배포"])
return suggestions[:5] # 상위 5개만 반환마무리
이번 글에서는 앞선 3편의 내용을 합쳐 외부 API 없이 내 컴퓨터에서 동작하는 문서 기반 질의응답 시스템을 구현했습니다.
시리즈 요약
- 1편: RAG의 개념과 필요성, 전체 아키텍처
- 2편: LangChain으로 구성 요소별 구현
- 3편: 벡터 데이터베이스의 동작 원리와 최적화
- 4편: 전체를 합친 로컬 RAG 시스템 구현
모든 처리가 로컬에서 이루어지므로 회사 문서나 개인 학습 자료처럼 외부로 보내기 어려운 문서도 다룰 수 있고, API 비용도 들지 않습니다.