초기 버전: 소방 복무관리 RAG 챗봇

주요 기능:
- RAG 기반 질의응답 시스템 구현
- 다양한 문서 형식 지원 (PDF, Word, TXT, Excel)
- 한국어 임베딩 모델 최적화 (jhgan/ko-sbert-nli)
- FAISS 벡터 데이터베이스 활용한 고성능 검색
- Gradio 기반 웹 인터페이스
- 허깅페이스 Spaces 배포 최적화
- 샘플 복무관리 문서 포함

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

README.md

@@ -0,0 +1,276 @@
+# 🚒 소방 복무관리 RAG 챗봇
+
+소방공무원 복무관리 규정 및 절차에 대한 질문에 답변해 드리는 RAG(Retrieval-Augmented Generation) 기반 챗봇입니다.
+
+## 📋 프로젝트 개요
+
+- **목적**: 소방업무 복무관리 관련 문서들을 활용한 지능형 질의응답 시스템
+- **기술**: RAG, Sentence-BERT, FAISS, Gradio, Hugging Face
+- **문서 유형**: PDF, Word, TXT, Excel, CSV
+- **특징**: 한국어 최적화, 실시간 검색, 웹 인터페이스
+
+## 🎯 주요 기능
+
+### ✅ RAG 검색 시스템
+- 문서 내용 자동 청킹
+- 의미 기반 검색
+- 유사도 측정
+- 다중 문서 검색
+
+### ✅ 인터페이스
+- Gradio 기반 웹 채팅
+- 실시간 응답
+- 출처 정보 표시
+- 신뢰도 점수 제공
+
+### ✅ 문서 처리
+- 다양한 파일 형식 지원 (PDF, DOCX, TXT, XLSX, CSV)
+- 한국어 텍스트 전처리
+- 소방 용어 최적화
+- 동적 문서 추가
+
+## 🏗️ 시스템 아키텍처
+
+```
+📂 문서 폴더
+   ↓
+📄 문서 처리기 (DocumentProcessor)
+   ↓
+🧠 임베딩 모델 (Sentence-BERT)
+   ↓
+🔍 벡터 DB (FAISS)
+   ↓
+💬 RAG 챗봇 (Chatbot)
+   ↓
+🌐 웹 인터페이스 (Gradio)
+```
+
+## 🚀 시작하기
+
+### 1. 환경 설정
+
+```bash
+# 1. 저장소 복제
+git clone <repository-url>
+cd 119_chatbot
+
+# 2. Python 가상환경 생성
+python -m venv venv
+
+# 3. 가상환경 활성화
+# Windows
+venv\Scripts\activate
+# Linux/Mac
+source venv/bin/activate
+
+# 4. 라이브러리 설치
+pip install -r requirements.txt
+```
+
+### 2. 문서 준비
+
+```bash
+# documents 폴더 생성
+mkdir documents
+
+# 복무관리 관련 문서 추가
+# 지원 형식: .pdf, .docx, .txt, .xlsx, .csv
+```
+
+### 3. 로컬 실행
+
+```bash
+# 방법 1: 간단한 테스트
+python document_processor.py
+
+# 방법 2: 챗봇 테스트
+python rag_chatbot.py
+
+# 방법 3: 웹 인터페이스 실행
+python gradio_interface.py
+
+# 방법 4: 허깅페이스 배포용 실행
+python app.py
+```
+
+## 📁 프로젝트 구조
+
+```
+119_chatbot/
+├── app.py                   # 허깅페이스 배포용 메인 파일
+├── config.py               # 시스템 설정
+├── requirements.txt        # 라이브러리 목록
+├── README.md              # 프로젝트 설명서
+├── document_processor.py  # 문서 처리 모듈
+├── vector_store.py        # 벡터 데이터베이스
+├── rag_chatbot.py         # RAG 챗봇 핵심 로직
+├── gradio_interface.py    # Gradio 웹 인터페이스
+├── documents/             # 문서 폴더
+│   ├── 복무관리규정.txt
+│   ├── 인사평가규정.txt
+│   └── ...
+└── faiss_index/           # 벡터 인덱스 캐시
+```
+
+## ⚙️ 설정 옵션
+
+### config.py 주요 설정
+
+```python
+# 모델 설정
+EMBEDDING_MODEL = "jhgan/ko-sbert-nli"      # 한국어 임베딩 모델
+LLM_MODEL = "beomi/Llama-3-Open-Ko-8B"      # 한국어 LLM
+
+# RAG 파라미터
+CHUNK_SIZE = 500                            # 문서 청크 크기
+CHUNK_OVERLAP = 50                          # 청크 중복 크기
+MAX_RETRIEVE_DOCS = 3                       # 검색할 문서 수
+
+# 경로 설정
+DOCS_FOLDER = "documents"                   # 문서 폴더
+VECTOR_DB_PATH = "faiss_index"             # 벡터 DB 경로
+```
+
+## 🧪 사용 예시
+
+### 기본 질문
+
+```
+Q: 연차휴가 사용 방법을 알려주세요
+A: 연차휴가는 1년간 정상 근무한 자에게 15일을 부여합니다.
+   사용 시 3일 전까지 신청서를 제출하고 부서장의 승인을 받아야 합니다.
+
+   📚 출처: 복무관리규정.txt (신뢰도: 92%)
+```
+
+### 고급 기능
+
+```python
+# 직접 코드로 사용
+from rag_chatbot import RAGChatbot
+
+# 챗봇 초기화
+chatbot = RAGChatbot()
+chatbot.initialize()
+
+# 질문하고 답변 받기
+response = chatbot.generate_answer("연차휴가 절차를 알려주세요")
+print(response.answer)
+```
+
+## 🌐 배포 방법
+
+### 1. 허깅페이스 Spaces
+
+```bash
+# 1. 허깅페이스 계정으로 로그인
+huggingface-cli login
+
+# 2. 새로운 Space 생성
+# - Space 이름: fire-service-rag-chatbot
+# - SDK: Gradio
+# - Hardware: CPU Basic (무료)
+
+# 3. 파일 업로드
+git add .
+git commit -m "초기 버전 배포"
+git push origin main
+```
+
+### 2. 로컬 서버
+
+```bash
+# 포트 7860으로 실행
+python app.py --port 7860
+
+# 공유 링크 생성
+python app.py --share
+```
+
+## 📊 성능 최적화
+
+### 🔍 검색 성능 향상
+- **Chunk 크기**: 500-1000 토큰으로 최적화
+- **검색 문서 수**: 3-5개로 설정하여 속도와 정확도 균형
+- **임베딩 모델**: 한국어 전용 모델 사용
+
+### 💾 메모리 관리
+- **캐싱**: 벡터 인덱스 로컬 저장
+- **동적 로드**: 필요 시에만 LLM 모델 로드
+- **청크 최적화**: 너무 긴 문서 분리
+
+### ⚡ 응답 속도
+- **템플릿 모드**: LLM 없이 빠른 응답
+- **캐싱**: 자�� 묻는 질문 응답 저장
+- **비동기 처리**: 대용량 문서 처리 시 배치 작업
+
+## 🔧 개발 가이드
+
+### 새로운 문서 추가
+
+```python
+from document_processor import DocumentProcessor
+
+# 문서 처리기 생성
+processor = DocumentProcessor()
+
+# 새 문서 로드
+new_docs = processor.load_documents_from_folder("new_documents")
+
+# 기존 문서에 추가
+from vector_store import VectorStore
+vector_store = VectorStore()
+vector_store.add_documents(new_docs)
+```
+
+### 커스텀 프롬프트
+
+```python
+# config.py에서 시스템 프롬프트 수정
+SYSTEM_PROMPT = """
+당신은 소방 복무관리 전문가입니다...
+[원하는 프롬프트 내용]
+"""
+```
+
+## 🚨 주의사항
+
+### 보안
+- 민감한 개인정보가 포함된 문서는 업로드하지 마세요
+- 법적 효력이 있는 최신 규정만 사용하세요
+- 정기적으로 문서를 업데이트해야 합니다
+
+### 성능
+- 대용량 문서 처리 시 시간이 오래 걸릴 수 있습니다
+- LLM 모델 로드 시 충분한 메모리가 필요합니다
+- 무료 플랜의 경우 리소스 제한이 있을 수 있습니다
+
+### 정확성
+- 생성된 답변은 반드시 실제 규정과 교차 확인하세요
+- 중요한 결정 시 반드시 담당자와 상담하세요
+- 챗봇 답변을 법적 효력으로 사용하지 마세요
+
+## 🤝 기여 방법
+
+1. 이슈 등록: 버그나 개선사항 등록
+2. 코드 제출: Pull Request를 통한 기여
+3. 문서 개선: README나 코드 주석 개선
+4. 테스트: 다양한 문서로 테스트 및 피드백
+
+## 📞 문의
+
+- 개발자: Claude (AI Assistant)
+- 이메일: [개발자 이메일]
+- 라이선스: MIT License
+
+## 🙏 감사
+
+- **Hugging Face**: 오픈소스 모델 및 플랫폼 제공
+- **LangChain**: RAG 프레임워크 제공
+- **Sentence-Transformers**: 한국어 임베딩 모델 제공
+- **FAISS**: 고성능 벡터 검색 라이브러리 제공
+- **Gradio**: 간편한 웹 인터페이스 제공
+
+---
+
+**⚠️ 본 챗봇은 보조 도구입니다. 중요한 업무 결정 시 반드시 관련 규정 원본과 담당자의 확인을 받으세요!**

app.py

@@ -0,0 +1,395 @@
+#!/usr/bin/env python3
+"""
+허깅페이스 Spaces 배포용 메인 파일
+소방 복무관리 RAG 챗봇
+"""
+
+import os
+import sys
+import argparse
+import gradio as gr
+import pandas as pd
+from pathlib import Path
+from typing import List, Dict, Tuple
+
+# 현재 디렉토리를 Python 경로에 추가
+current_dir = os.path.dirname(os.path.abspath(__file__))
+sys.path.append(current_dir)
+
+# 모듈 임포트
+from config import Config
+from rag_chatbot import RAGChatbot
+from document_processor import DocumentProcessor
+
+class HuggingFaceApp:
+    """허깅페이스 Spaces 배포용 앱 클래스"""
+
+    def __init__(self):
+        self.chatbot = RAGChatbot()
+        self.is_initialized = False
+
+        # 예시 질문 (허깅페이스 환경 최적화)
+        self.example_questions = [
+            "연차휴가 사용 방법을 알려주세요",
+            "정규근무시간은 어떻게 되나요?",
+            "당직근무 절차가 궁금합니다",
+            "인사평가는 언제 진행되나요?",
+            "파견근무 신청 방법",
+            "복무규정 위반 시 처리"
+        ]
+
+        # 앱 초기화
+        self._initialize_app()
+
+    def _initialize_app(self):
+        """앱 초기화"""
+        try:
+            print("🚀 소방 복무관리 RAG 챗봇 시작 중...")
+
+            # 문서 폴더 확인 및 샘플 데이터 생성
+            self._ensure_documents()
+
+            # 챗봇 초기화
+            success = self.chatbot.initialize()
+            if success:
+                self.is_initialized = True
+                print("✅ 챗봇 초기화 완료")
+            else:
+                print("⚠️ 챗봇 초기화 실패 - 템플릿 모드로 동작")
+
+        except Exception as e:
+            print(f"❌ 초기화 오류: {str(e)}")
+            # 오류가 있어도 앱은 계속 실행
+
+    def _ensure_documents(self):
+        """문서 폴더 및 샘플 데이터 확인"""
+        docs_folder = Path("documents")
+        docs_folder.mkdir(exist_ok=True)
+
+        # 샘플 문서가 없으면 생성
+        sample_files = list(docs_folder.glob("*.txt"))
+        if not sample_files:
+            print("📝 샘플 문서 생성 중...")
+            self._create_sample_documents()
+
+    def _create_sample_documents(self):
+        """샘플 복무관리 문서 생성"""
+        sample_docs = [
+            {
+                "filename": "복무관리규정.txt",
+                "content": """소방공무원 복무관리 규정
+
+제1장 총칙
+제1조 (목적)
+이 규정은 소방공무원의 복무에 관한 기본사항을 규정하여 직무수행의 효율성을 높이고 조직의 발전에 기여함을 목적으로 한다.
+
+제2조 (근무시간)
+1. 정규근무시간은 09:00부터 18:00까지로 한다.
+2. 점심시간은 12:00부터 13:00까지로 한다.
+3. 토요일, 일요일 및 법정공휴일은 휴무일로 한다.
+
+제3조 (연차휴가)
+1. 연차휴가는 1년간 정상 근무한 자에게 15일을 부여한다.
+2. 연차휴가 사용 시 3일 전까지 신청서를 제출해야 한다.
+3. 부서장의 승인을 받아 사용하며, 긴급한 경우에는 사후 승인도 가능하다.
+
+제4조 (당직근무)
+1. 당직근무는 정규근무시간 외에 수행하는 근무를 말한다.
+2. 당직자는 비상상황에 대비한 통신장비를 항시 점검해야 한다.
+3. 당직 중에는 음주를 엄금하며, 직무 수행에 지장이 없는 행위만 가능하다.
+"""
+            },
+            {
+                "filename": "인사평가규정.txt",
+                "content": """소방공무원 인사평가 규정
+
+제1장 평가의 기본원칙
+제1조 (평가목적)
+소방공무원의 직무수행 능력과 성과를 객관적으로 평가하여 능력위주의 인사관리를 정립하고 공정한 보상 및 승진의 기초 자료로 활용한다.
+
+제2조 (평가주기)
+1. 정기평가는 연 1회 실시하며, 평가기간은 매년 1월 1일부터 12월 31일까지로 한다.
+2. 수시평가는 특별한 사유가 있을 경우 실시할 수 있다.
+
+제3조 (평가항목)
+1. 직무수행 능력 (40점)
+2. 업무 성과 (30점)
+3. 근무 태도 (20점)
+4. 협업 능력 (10점)
+
+제4조 (평가등급)
+- 수 (90점 이상)
+- 우 (80점 이상 90점 미만)
+- 양 (70점 이상 80점 미만)
+- 가 (60점 이상 70점 미만)
+- 미 (60점 미만)
+"""
+            },
+            {
+                "filename": "교육훈련.txt",
+                "content": """소방공무원 교육훈련 안내
+
+제1조 (교육목적)
+소방공무원의 전문성 향상과 직무능력 개발을 위한 체계적인 교육훈련을 실시한다.
+
+제2조 (필수교육)
+1. 신임교육: 신규 임용자 대상 2주간 집체교육
+2. 직무연수: 매년 1회, 직무별 전문교육
+3. 안전교육: 분기별 1회, 안전사고 예방 교육
+
+제3조 (선택교육)
+1. 외국어 교육
+2. 정보통신 기술 교육
+3. 리더십 교육
+4. 전문 자격증 취득 지원 교육
+
+제4조 (교육신청)
+1. 교육 희망자는 소속 기관을 통해 신청한다.
+2. 신청 시기는 교육 시작일 1개월 전까지이다.
+3. 업무에 지장이 없는 경우 우선 선발한다.
+"""
+            }
+        ]
+
+        for doc in sample_docs:
+            file_path = Path("documents") / doc["filename"]
+            try:
+                with open(file_path, 'w', encoding='utf-8') as f:
+                    f.write(doc["content"])
+                print(f"✅ {doc['filename']} 생성 완료")
+            except Exception as e:
+                print(f"❌ {doc['filename']} 생성 실패: {str(e)}")
+
+    def format_response(self, response) -> str:
+        """응답을 Gradio 형식으로 변환"""
+        try:
+            # 메시지 형식으로 변환
+            answer = response.answer
+
+            # 신뢰도 표시
+            confidence = getattr(response, 'confidence', 0.0)
+            if confidence >= 0.8:
+                confidence_emoji = "🟢"
+            elif confidence >= 0.5:
+                confidence_emoji = "🟡"
+            else:
+                confidence_emoji = "🔴"
+
+            # 응답 시간
+            response_time = getattr(response, 'response_time', 0.0)
+
+            # 출처 정보
+            sources = getattr(response, 'sources', [])
+            source_text = ""
+            if sources:
+                source_text = "\n\n📚 **참고자료:**\n"
+                for i, source in enumerate(sources[:3], 1):  # 최대 3개만 표시
+                    source_name = source.get('source', '알 수 없음')
+                    source_text += f"{i}. {source_name}\n"
+
+            # 전체 응답
+            full_response = f"""{answer}
+
+---
+{confidence_emoji} 신뢰도: {confidence:.1%}
+⏱️ 응답시간: {response_time:.2f}초
+📄 참고문서: {len(sources)}개{source_text}"""
+
+            return full_response
+
+        except Exception as e:
+            return f"응답 형식 변환 중 오류 발생: {str(e)}"
+
+    def chat_function(self, message: str, history: List[List[str]]) -> List[List[str]]:
+        """채팅 함수"""
+        if not message.strip():
+            return history
+
+        try:
+            # 챗봇 응답 생성
+            if self.is_initialized:
+                response = self.chatbot.generate_answer(message, use_llm=False)
+                answer = self.format_response(response)
+            else:
+                answer = "죄송합니다. 챗봇이 초기화되지 않았습니다. 페이지를 새로고침해주세요."
+
+            # 히스토리에 추가
+            history.append([message, answer])
+
+        except Exception as e:
+            error_msg = f"답변 생성 중 오류 발생: {str(e)}\n\n관리자에게 문의해주세요."
+            history.append([message, error_msg])
+
+        return history
+
+    def create_demo(self):
+        """Gradio 데모 생성"""
+        # 커스텀 CSS
+        custom_css = """
+        .gradio-container {
+            max-width: 900px !important;
+            margin: auto !important;
+        }
+        .message.user {
+            background-color: #e3f2fd;
+            border-radius: 15px 15px 0 15px;
+        }
+        .message.assistant {
+            background-color: #f1f8e9;
+            border-radius: 15px 15px 15px 0;
+        }
+        """
+
+        with gr.Blocks(
+            title="소방 복무관리 RAG 챗봇",
+            theme=gr.themes.Soft(),
+            css=custom_css
+        ) as demo:
+
+            gr.Markdown("""
+            # 🚒 소방 복무관리 RAG 챗봇
+
+            소방공무원 복무관리 규정, 인사평가, 교육훈련 등 업무 관련 질문에 답변해 드립니다.
+
+            💡 **사용 방법**: 아래에 복무관리 관련 질문을 입력하고 Enter 키를 누르세요.
+            """)
+
+            # 상태 표시
+            with gr.Row():
+                status_text = gr.HTML(
+                    "✅ **챗봇 준비 완료** - 복무관리 관련 질문을 입력해주세요"
+                )
+
+            # 채팅 인터페이스
+            chatbot = gr.Chatbot(
+                height=500,
+                show_copy_button=True,
+                bubble_full_width=False,
+                placeholder="안녕하세요! 소방 복무관리에 대해 무엇이 궁금하신가요?",
+                avatar_images=["👤", "🤖"]
+            )
+
+            # 입력 영역
+            with gr.Row():
+                msg = gr.Textbox(
+                    placeholder="복무관리 관련 질문을 입력해주세요 (예: 연차휴가 사용 방법)",
+                    container=False,
+                    scale=7
+                )
+                submit_btn = gr.Button("전송", scale=1, variant="primary")
+                clear_btn = gr.Button("초기화", scale=1)
+
+            # 예시 질문
+            gr.Markdown("### 💡 예시 질문 (클릭하면 자동 입력됩니다)")
+
+            with gr.Row():
+                with gr.Column():
+                    gr.Button("연차휴가 사용 방법", size="sm").click(
+                        fn=lambda: "연차휴가 사용 방법을 알려주세요",
+                        outputs=msg
+                    )
+                    gr.Button("정규근무시간", size="sm").click(
+                        fn=lambda: "정규근무시간은 어떻게 되나요?",
+                        outputs=msg
+                    )
+                    gr.Button("당직근무 절차", size="sm").click(
+                        fn=lambda: "당직근무 절차가 궁금합니다",
+                        outputs=msg
+                    )
+                with gr.Column():
+                    gr.Button("인사평가 기준", size="sm").click(
+                        fn=lambda: "인사평가는 어떤 기준으로 이루어지나요?",
+                        outputs=msg
+                    )
+                    gr.Button("교육훈련 안내", size="sm").click(
+                        fn=lambda: "교육훈련 종류와 신청 방법을 알려주세요",
+                        outputs=msg
+                    )
+                    gr.Button("파견근무 신청", size="sm").click(
+                        fn=lambda: "파견근무 신청 방법을 알려주세요",
+                        outputs=msg
+                    )
+
+            # 정보 섹션
+            with gr.Accordion("📊 시스템 정보", open=False):
+                if self.is_initialized:
+                    stats = self.chatbot.get_stats()
+                    vector_stats = stats.get("vector_store", {})
+                    gr.Markdown(f"""
+                    - **문서 수**: {vector_stats.get('total_documents', 0)}개
+                    - **임베딩 모델**: {vector_stats.get('embedding_model', 'N/A')}
+                    - **응답 모드**: 템플릿 기반
+                    - **최대 검색 문서**: {Config.MAX_RETRIEVE_DOCS}개
+                    """)
+                else:
+                    gr.Markdown("⚠️ 챗봇이 초기화되지 않았습니다.")
+
+            # 이벤트 핸들러
+            def user_input(user_message, history):
+                """사용자 입력 처리"""
+                return "", history + [[user_message, None]]
+
+            def bot_response(history):
+                """봇 응답 처리"""
+                if history and history[-1][1] is None:
+                    user_message = history[-1][0]
+                    bot_message = self.chat_function(user_message, history[:-1])
+                    if bot_message:
+                        history[-1][1] = bot_message[-1][1]  # 마지막 응답만 가져오기
+                    else:
+                        history[-1][1] = "죄송합니다. 응답을 생성할 수 없습니다."
+                return history
+
+            # 메시지 전송 이벤트
+            msg.submit(
+                user_input,
+                [msg, chatbot],
+                [msg, chatbot],
+                queue=False
+            ).then(
+                bot_response,
+                chatbot,
+                chatbot
+            )
+
+            submit_btn.click(
+                user_input,
+                [msg, chatbot],
+                [msg, chatbot],
+                queue=False
+            ).then(
+                bot_response,
+                chatbot,
+                chatbot
+            )
+
+            clear_btn.click(
+                lambda: ([], ""),
+                outputs=[chatbot, msg]
+            )
+
+        return demo
+
+def main():
+    """메인 실행 함수"""
+    parser = argparse.ArgumentParser(description="소방 복무관리 RAG 챗봇")
+    parser.add_argument("--share", action="store_true", help="공유 링크 생성")
+    parser.add_argument("--port", type=int, default=7860, help="서버 포트")
+    args = parser.parse_args()
+
+    # 앱 생성
+    app = HuggingFaceApp()
+    demo = app.create_demo()
+
+    # 실행
+    print("🚀 허깅페이스 Spaces 앱 시작 중...")
+    demo.launch(
+        share=args.share,
+        server_port=args.port,
+        server_name="0.0.0.0",
+        show_error=True,
+        show_tips=False
+    )
+
+if __name__ == "__main__":
+    main()

config.py

@@ -0,0 +1,43 @@
+# 복무관리 RAG 챗봇 설정
+import os
+from dotenv import load_dotenv
+
+# 환경 변수 로드
+load_dotenv()
+
+class Config:
+    # 모델 설정
+    EMBEDDING_MODEL = "jhgan/ko-sbert-nli"  # 한국어 임베딩 모델
+    LLM_MODEL = "beomi/Llama-3-Open-Ko-8B"   # 한국어 LLM
+
+    # RAG 설정
+    CHUNK_SIZE = 500
+    CHUNK_OVERLAP = 50
+    MAX_RETRIEVE_DOCS = 3
+
+    # 벡터 DB 설정
+    VECTOR_DB_PATH = "faiss_index"
+
+    # 문서 폴더 설정
+    DOCS_FOLDER = "documents"
+
+    # 웹 인터페이스 설정
+    APP_NAME = "소방 복무관리 RAG 챗봇"
+    APP_DESCRIPTION = "소방업무 복무관리 규정 및 절차에 대한 질문에 답변해 드립니다."
+
+    # 토큰 설정 (필요시)
+    HUGGINGFACE_TOKEN = os.getenv("HUGGINGFACE_TOKEN", "")
+
+    # 시스템 프롬프트
+    SYSTEM_PROMPT = """
+    당신은 소방업무 복무관리 전문가입니다. 복무관리 규정, 인사운영, 근무절차 등에 대한 질문에 정확하고 친절하게 답변해 주세요.
+
+    답변 시 다음 사항을 준수해 주세요:
+    1. 관련 규정 조문이나 근거를 명확히 제시
+    2. 절차가 있는 경우 단계별로 설명
+    3. 필요한 서류나 양식을 안내
+    4. 주의사항이나 중요한 사항은 강조 표시
+    5. 모든 답변은 한국어로 제공
+
+    사용자의 질문에 최대한 상세하고 정확한 정보를 제공하세요.
+    """

document_processor.py

@@ -0,0 +1,291 @@
+import os
+import re
+from typing import List, Dict
+from pathlib import Path
+import PyMuPDF
+import docx
+import pandas as pd
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from langchain.schema import Document
+
+class DocumentProcessor:
+    """복무관리 문서 처리 클래스"""
+
+    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
+        self.chunk_size = chunk_size
+        self.chunk_overlap = chunk_overlap
+        self.text_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=chunk_size,
+            chunk_overlap=chunk_overlap,
+            separators=["\n\n", "\n", " ", ""]
+        )
+
+    def load_documents_from_folder(self, folder_path: str) -> List[Document]:
+        """폴더에서 모든 문서 로드"""
+        documents = []
+        folder = Path(folder_path)
+
+        if not folder.exists():
+            print(f"⚠️ 폴더가 존재하지 않습니다: {folder_path}")
+            return documents
+
+        # 지원하는 파일 형식
+        supported_extensions = ['.pdf', '.docx', '.txt', '.xlsx', '.csv']
+
+        for file_path in folder.rglob('*'):
+            if file_path.suffix.lower() in supported_extensions:
+                try:
+                    print(f"📄 문서 로드: {file_path.name}")
+                    docs = self.load_single_document(str(file_path))
+                    documents.extend(docs)
+                except Exception as e:
+                    print(f"❌ 문서 로드 실패 ({file_path.name}): {str(e)}")
+
+        return documents
+
+    def load_single_document(self, file_path: str) -> List[Document]:
+        """단일 문서 로드"""
+        file_ext = Path(file_path).suffix.lower()
+
+        if file_ext == '.pdf':
+            return self._load_pdf(file_path)
+        elif file_ext == '.docx':
+            return self._load_docx(file_path)
+        elif file_ext == '.txt':
+            return self._load_txt(file_path)
+        elif file_ext in ['.xlsx', '.csv']:
+            return self._load_table(file_path)
+        else:
+            raise ValueError(f"지원하지 않는 파일 형식: {file_ext}")
+
+    def _load_pdf(self, file_path: str) -> List[Document]:
+        """PDF 파일 로드"""
+        documents = []
+
+        try:
+            with PyMuPDF.open(file_path) as doc:
+                full_text = ""
+
+                for page_num in range(len(doc)):
+                    page = doc[page_num]
+                    page_text = page.get_text()
+
+                    # 페이지 정제
+                    page_text = self._clean_text(page_text)
+
+                    if page_text.strip():
+                        full_text += f"\n\n--- 페이지 {page_num + 1} ---\n\n{page_text}"
+
+                if full_text.strip():
+                    chunks = self.text_splitter.split_text(full_text)
+
+                    for i, chunk in enumerate(chunks):
+                        metadata = {
+                            "source": Path(file_path).name,
+                            "page": "multiple",
+                            "chunk_id": i,
+                            "file_type": "pdf"
+                        }
+
+                        documents.append(Document(page_content=chunk, metadata=metadata))
+
+        except Exception as e:
+            print(f"PDF 로드 중 오류: {str(e)}")
+            raise
+
+        return documents
+
+    def _load_docx(self, file_path: str) -> List[Document]:
+        """Word 문서 로드"""
+        documents = []
+
+        try:
+            doc = docx.Document(file_path)
+            paragraphs = []
+
+            for para in doc.paragraphs:
+                if para.text.strip():
+                    paragraphs.append(para.text)
+
+            full_text = "\n\n".join(paragraphs)
+
+            if full_text.strip():
+                chunks = self.text_splitter.split_text(full_text)
+
+                for i, chunk in enumerate(chunks):
+                    metadata = {
+                        "source": Path(file_path).name,
+                        "chunk_id": i,
+                        "file_type": "docx"
+                    }
+
+                    documents.append(Document(page_content=chunk, metadata=metadata))
+
+        except Exception as e:
+            print(f"DOCX 로드 중 오류: {str(e)}")
+            raise
+
+        return documents
+
+    def _load_txt(self, file_path: str) -> List[Document]:
+        """텍스트 파일 로드"""
+        documents = []
+
+        try:
+            with open(file_path, 'r', encoding='utf-8') as f:
+                text = f.read()
+
+            text = self._clean_text(text)
+
+            if text.strip():
+                chunks = self.text_splitter.split_text(text)
+
+                for i, chunk in enumerate(chunks):
+                    metadata = {
+                        "source": Path(file_path).name,
+                        "chunk_id": i,
+                        "file_type": "txt"
+                    }
+
+                    documents.append(Document(page_content=chunk, metadata=metadata))
+
+        except Exception as e:
+            print(f"TXT 로드 중 오류: {str(e)}")
+            raise
+
+        return documents
+
+    def _load_table(self, file_path: str) -> List[Document]:
+        """엑셀/CSV 파일 로드"""
+        documents = []
+
+        try:
+            if file_path.endswith('.xlsx'):
+                df = pd.read_excel(file_path)
+            else:
+                df = pd.read_csv(file_path, encoding='utf-8')
+
+            # 데이터프레임을 텍스트로 변환
+            text_parts = []
+            text_parts.append(f"파일: {Path(file_path).name}")
+            text_parts.append(f"컬럼: {', '.join(df.columns.tolist())}")
+
+            for index, row in df.iterrows():
+                row_text = " | ".join([f"{col}: {val}" for col, val in row.items() if pd.notna(val)])
+                text_parts.append(f"행 {index + 1}: {row_text}")
+
+            full_text = "\n\n".join(text_parts)
+
+            if full_text.strip():
+                chunks = self.text_splitter.split_text(full_text)
+
+                for i, chunk in enumerate(chunks):
+                    metadata = {
+                        "source": Path(file_path).name,
+                        "chunk_id": i,
+                        "file_type": "table",
+                        "total_rows": len(df)
+                    }
+
+                    documents.append(Document(page_content=chunk, metadata=metadata))
+
+        except Exception as e:
+            print(f"테이블 로드 중 오류: {str(e)}")
+            raise
+
+        return documents
+
+    def _clean_text(self, text: str) -> str:
+        """텍스트 정제"""
+        # 불필요한 공백 제거
+        text = re.sub(r'\s+', ' ', text)
+
+        # 특수문자 정리
+        text = re.sub(r'[^\w\s\.\,\?\!\:\;\-\(\)\/\&\@]', ' ', text)
+
+        # 연속된 공백 제거
+        text = re.sub(r'\s+', ' ', text).strip()
+
+        return text
+
+    def process_documents(self, documents: List[Document]) -> List[Document]:
+        """문서 후처리"""
+        processed_docs = []
+
+        for doc in documents:
+            content = doc.page_content.strip()
+
+            if content and len(content) > 20:  # 너무 짧은 청크는 제외
+                # 복무관리 특화 키워드 강화
+                content = self._enhance_fire_service_terms(content)
+
+                processed_doc = Document(
+                    page_content=content,
+                    metadata=doc.metadata
+                )
+                processed_docs.append(processed_doc)
+
+        return processed_docs
+
+    def _enhance_fire_service_terms(self, text: str) -> str:
+        """소방 용어 강화"""
+        # 복무관리 관련 키워드 매핑
+        term_mappings = {
+            "연차": "연차휴가",
+            "연장": "연장근무",
+            "당직": "당직근무",
+            "파견": "파견근무",
+            "인사": "인사평가",
+            "승진": "승진시험",
+            "교육": "교육훈련",
+            "휴가": "휴가사용",
+            "상벌": "상벌규정",
+            "징계": "징계절차"
+        }
+
+        enhanced_text = text
+        for standard_term, enhanced_term in term_mappings.items():
+            enhanced_text = enhanced_text.replace(standard_term, enhanced_term)
+
+        return enhanced_text
+
+# 테스트용 함수
+def test_document_processor():
+    """문서 처리기 테스트"""
+    processor = DocumentProcessor()
+
+    # 샘플 documents 폴더 생성
+    docs_folder = "documents"
+    os.makedirs(docs_folder, exist_ok=True)
+
+    # 샘플 문서 생성
+    sample_text = """
+    복무관리 규정
+
+    제1장 총칙
+    제1조 (목적)
+    이 규정은 소방공무원의 복무에 관한 사항을 규정하여 직무수행의 효율성을 높이고
+    조직의 발전에 기여함을 목적으로 한다.
+
+    제2조 (근무시간)
+    1. 정규근무시간은 09:00부터 18:00까지로 한다.
+    2. 점심시간은 12:00부터 13:00까지로 한다.
+    3. 당직근무는 정규근무시간 외에 수행하는 근무를 말한다.
+
+    제3조 (연차휴가)
+    1. 연차휴가는 1년간 정상 근무한 자에게 15일을 부여한다.
+    2. 연차휴가 사용 시 3일 전까지 신청서를 제출해야 한다.
+    3. 부서장의 승인을 받아 사용한다.
+    """
+
+    with open(os.path.join(docs_folder, "sample_policy.txt"), "w", encoding="utf-8") as f:
+        f.write(sample_text)
+
+    # 문서 로드 테스트
+    documents = processor.load_documents_from_folder(docs_folder)
+    print(f"✅ {len(documents)}개 문서 청크 생성 완료")
+
+    return documents
+
+if __name__ == "__main__":
+    test_document_processor()

gradio_interface.py

@@ -0,0 +1,329 @@
+import gradio as gr
+import time
+from typing import List, Dict, Tuple
+import os
+from rag_chatbot import RAGChatbot, ChatResponse
+
+class GradioInterface:
+    """Gradio 웹 인터페이스 클래스"""
+
+    def __init__(self):
+        self.chatbot = RAGChatbot()
+        self.is_initialized = False
+
+        # 예시 질문
+        self.example_questions = [
+            "연차휴가 사용 방법을 알려주세요",
+            "정규근무시간은 어떻게 되나요?",
+            "당직근무 절차가 궁금합니다",
+            "인사평가는 언제 어떻게 진행되나요?",
+            "파견근무 신청 방법을 알려주세요",
+            "복무규정 위반 시 어떻게 되나요?"
+        ]
+
+    def initialize_chatbot(self, docs_folder: str = None, force_rebuild: bool = False) -> str:
+        """챗봇 초기화"""
+        try:
+            success = self.chatbot.initialize(docs_folder, force_rebuild)
+            if success:
+                self.is_initialized = True
+                return "✅ RAG 챗봇이 성공적으로 초기화되었습니다!"
+            else:
+                return "❌ 챗봇 초기화에 실패했습니다. documents 폴더에 파일이 있는지 확인해주세요."
+        except Exception as e:
+            return f"❌ 초기화 중 오류 발생: {str(e)}"
+
+    def format_chat_response(self, response: ChatResponse) -> Tuple[str, str]:
+        """챗봇 응답을 채팅 형식으로 변환"""
+        # 메인 답변
+        answer_html = response.answer.replace('\n', '<br>')
+
+        # 신뢰도 색상
+        if response.confidence >= 0.8:
+            confidence_color = "green"
+            confidence_text = "높음"
+        elif response.confidence >= 0.5:
+            confidence_color = "orange"
+            confidence_text = "보통"
+        else:
+            confidence_color = "red"
+            confidence_text = "낮음"
+
+        # 정보 메시지
+        info_html = f"""
+        <div style="background-color: #f0f0f0; padding: 10px; border-radius: 5px; margin-top: 10px;">
+            <strong>📊 답변 정보</strong><br>
+            • 신뢰도: <span style="color: {confidence_color}; font-weight: bold;">{confidence_text} ({response.confidence:.2%})</span><br>
+            • 응답시간: {response.response_time:.2f}초<br>
+            • 참고문서: {len(response.sources)}개
+        </div>
+        """
+
+        # 출처 정보
+        if response.sources:
+            sources_html = "<br><strong>📚 참고자료:</strong><ul>"
+            for i, source in enumerate(response.sources, 1):
+                sources_html += f"<li><strong>{source['source']}</strong> (유사도: {source['similarity']})<br><small>{source['content'][:150]}...</small></li>"
+            sources_html += "</ul>"
+            info_html += sources_html
+
+        return answer_html, info_html
+
+    def chat_interface(self, message: str, history: List[List[str]]) -> List[List[str]]:
+        """채팅 인터페이스 핸들러"""
+        if not self.is_initialized:
+            history.append([message, "⚠️ 챗봇이 초기화되지 않았습니다. 먼저 초기화를 눌러주세요."])
+            return history
+
+        if not message.strip():
+            return history
+
+        try:
+            # 답변 생성
+            response = self.chatbot.generate_answer(message, use_llm=False)  # 템플릿 모드로 가볍게
+            answer, info = self.format_chat_response(response)
+
+            # 전체 응답
+            full_response = f"{answer}<br><br>{info}"
+
+            history.append([message, full_response])
+
+        except Exception as e:
+            error_response = f"❌ 답변 생성 중 오류 발생: {str(e)}"
+            history.append([message, error_response])
+
+        return history
+
+    def test_question(self, question: str) -> str:
+        """예시 질문 테스트"""
+        if not self.is_initialized:
+            return "⚠️ 먼저 챗봇 초기화를 진행해주세요."
+
+        try:
+            response = self.chatbot.generate_answer(question, use_llm=False)
+            answer, info = self.format_chat_response(response)
+            return f"❓ 질문: {question}\n\n🤖 답변:\n{answer}\n\n{info}"
+        except Exception as e:
+            return f"❌ 테스트 중 오류 발생: {str(e)}"
+
+    def get_chatbot_stats(self) -> str:
+        """챗봇 통계 정보"""
+        if not self.is_initialized:
+            return "챗봇이 초기화되지 않았습니다."
+
+        try:
+            stats = self.chatbot.get_stats()
+            stats_html = "<h3>🤖 챗봇 상태 정보</h3>"
+
+            if stats.get("status") == "initialized":
+                vector_stats = stats.get("vector_store", {})
+                stats_html += f"""
+                <ul>
+                    <li>상태: <span style="color: green;">✅ 정상 작동 중</span></li>
+                    <li>총 문서 수: {vector_stats.get('total_documents', 0)}개</li>
+                    <li>임베딩 모델: {vector_stats.get('embedding_model', 'N/A')}</li>
+                    <li>벡터 차원: {vector_stats.get('index_dimension', 'N/A')}</li>
+                    <li>LLM 사용: {'✅' if stats.get('llm_available') else '❌ (템플릿 모드)'}</li>
+                </ul>
+                """
+            else:
+                stats_html += "<p>⚠️ 챗봇이 초기화되지 않았습니다.</p>"
+
+            return stats_html
+
+        except Exception as e:
+            return f"❌ 통계 정보 조회 실패: {str(e)}"
+
+    def create_interface(self):
+        """Gradio 인터페이스 생성"""
+        # 커스텀 CSS
+        custom_css = """
+        .chat-message {
+            padding: 15px;
+            border-radius: 10px;
+            margin: 10px 0;
+        }
+        .user-message {
+            background-color: #e3f2fd;
+            border-left: 4px solid #2196f3;
+        }
+        .assistant-message {
+            background-color: #f1f8e9;
+            border-left: 4px solid #4caf50;
+        }
+        .info-box {
+            background-color: #fff3e0;
+            border: 1px solid #ffb74d;
+            border-radius: 8px;
+            padding: 12px;
+            margin: 10px 0;
+        }
+        .stats-box {
+            background-color: #f5f5f5;
+            border-radius: 8px;
+            padding: 15px;
+            margin: 10px 0;
+        }
+        """
+
+        with gr.Blocks(
+            title="소방 복무관리 RAG 챗봇",
+            theme=gr.themes.Soft(),
+            css=custom_css
+        ) as interface:
+
+            gr.Markdown("# 🚒 소방 복무관리 RAG 챗봇")
+            gr.Markdown("소방업무 복무관리 규정 및 절차에 대한 질문에 답변해 드립니다.")
+
+            with gr.Tab("💬 채팅"):
+                with gr.Row():
+                    with gr.Column(scale=4):
+                        chatbot = gr.Chatbot(
+                            height=500,
+                            show_copy_button=True,
+                            bubble_full_width=False,
+                            avatar_images=["👤", "🤖"]
+                        )
+                        msg = gr.Textbox(
+                            placeholder="복무관리 관련 질문을 입력해주세요 (예: 연차휴가 사용 방법)",
+                            label="질문 입력",
+                            submit_btn="전송"
+                        )
+                        with gr.Row():
+                            submit_btn = gr.Button("💬 전송", variant="primary")
+                            clear_btn = gr.Button("🗑️ 대화 초기화")
+
+                    with gr.Column(scale=1):
+                        gr.Markdown("### 🚀 빠른 시작")
+                        init_btn = gr.Button("🔧 챗봇 초기화", variant="secondary")
+                        init_status = gr.HTML("⏳ 초기화를 눌러주세요.")
+
+                        gr.Markdown("### 💡 예시 질문")
+                        for question in self.example_questions:
+                            gr.Button(question, size="sm").click(
+                                fn=lambda q=question: q,
+                                outputs=msg
+                            )
+
+                        gr.Markdown("### 📊 상태 정보")
+                        stats_info = gr.HTML("상태 정보를 확인해주세요.")
+
+            with gr.Tab("🔍 테스트"):
+                gr.Markdown("### 예시 질문 테스트")
+                with gr.Row():
+                    with gr.Column():
+                        test_question = gr.Textbox(
+                            placeholder="테스트할 질문을 입력하세요",
+                            label="질문"
+                        )
+                        test_btn = gr.Button("🧪 테스트 실행", variant="primary")
+                    with gr.Column():
+                        test_result = gr.HTML("테스트 결과가 여기에 표시됩니다.")
+
+                gr.Markdown("### 통계 정보")
+                stats_display = gr.HTML(self.get_chatbot_stats())
+
+            with gr.Tab("⚙️ 설정"):
+                gr.Markdown("### 시스템 설정")
+                with gr.Row():
+                    with gr.Column():
+                        docs_folder = gr.Textbox(
+                            value="documents",
+                            label="문서 폴더 경로"
+                        )
+                        force_rebuild = gr.Checkbox(
+                            label="강제 재구축",
+                            info="체크 시 기존 인덱스를 새로构建"
+                        )
+                    with gr.Column():
+                        rebuild_btn = gr.Button("🔄 인덱스 재구축", variant="primary")
+
+                gr.Markdown("### 시스템 정보")
+                system_info = gr.HTML("""
+                <div class="stats-box">
+                    <h4>🤖 시스템 사양</h4>
+                    <ul>
+                        <li>임베딩 모델: jhgan/ko-sbert-nli</li>
+                        <li>LLM 모델: beomi/Llama-3-Open-Ko-8B</li>
+                        <li>청크 크기: 500</li>
+                        <li>검색 문서 수: 3개</li>
+                    </ul>
+                </div>
+                """)
+
+            # 이벤트 핸들러
+            def init_chatbot_wrapper():
+                status = self.initialize_chatbot()
+                return status, self.get_chatbot_stats()
+
+            init_btn.click(
+                fn=init_chatbot_wrapper,
+                outputs=[init_status, stats_info]
+            )
+
+            def rebuild_wrapper(docs_path, force):
+                return self.initialize_chatbot(docs_path, force)
+
+            rebuild_btn.click(
+                fn=rebuild_wrapper,
+                inputs=[docs_folder, force_rebuild],
+                outputs=stats_display
+            )
+
+            # 채팅 이벤트
+            def chat_wrapper(message, history):
+                return self.chat_interface(message, history), ""
+
+            msg.submit(
+                fn=chat_wrapper,
+                inputs=[msg, chatbot],
+                outputs=[chatbot, msg]
+            )
+
+            submit_btn.click(
+                fn=chat_wrapper,
+                inputs=[msg, chatbot],
+                outputs=[chatbot, msg]
+            )
+
+            clear_btn.click(
+                fn=lambda: ([], ""),
+                outputs=[chatbot, msg]
+            )
+
+            # 테스트 이벤트
+            test_btn.click(
+                fn=self.test_question,
+                inputs=test_question,
+                outputs=test_result
+            )
+
+            # 자동 새로고침 (통계 정보)
+            interface.load(
+                fn=self.get_chatbot_stats,
+                outputs=stats_info,
+                every=30  # 30초마다 새로고침
+            )
+
+        return interface
+
+    def launch(self, share: bool = False, server_port: int = 7860):
+        """인터페이스 실행"""
+        interface = self.create_interface()
+
+        print("🚀 Gradio 웹 인터페이스 시작 중...")
+        print(f"📍 접속 주소: http://localhost:{server_port}")
+
+        interface.launch(
+            share=share,
+            server_port=server_port,
+            show_error=True,
+            show_tips=True,
+            inbrowser=True
+        )
+
+# 메인 실행
+if __name__ == "__main__":
+    # 웹 인터페이스 실행
+    gradio_app = GradioInterface()
+    gradio_app.launch(share=False, server_port=7860)

rag_chatbot.py

@@ -0,0 +1,376 @@
+import os
+import re
+from typing import List, Dict, Tuple
+from dataclasses import dataclass
+import torch
+from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
+from langchain.schema import Document
+from document_processor import DocumentProcessor
+from vector_store import VectorStore
+from config import Config
+
+@dataclass
+class ChatResponse:
+    """챗봇 응답 결과 클래스"""
+    answer: str
+    sources: List[Dict]
+    confidence: float
+    response_time: float
+
+class RAGChatbot:
+    """소방 복무관리 RAG 챗봇"""
+
+    def __init__(self):
+        self.document_processor = DocumentProcessor(
+            chunk_size=Config.CHUNK_SIZE,
+            chunk_overlap=Config.CHUNK_OVERLAP
+        )
+        self.vector_store = VectorStore()
+        self.llm = None
+        self.llm_tokenizer = None
+        self.is_initialized = False
+
+    def initialize(self, docs_folder: str = None, force_rebuild: bool = False):
+        """챗봇 초기화"""
+        print("🤖 소방 복무관리 RAG 챗봇 초기화 중...")
+
+        # 1. 문서 로드 및 처리
+        docs_folder = docs_folder or Config.DOCS_FOLDER
+        documents = self._load_documents(docs_folder)
+
+        if not documents:
+            print("❌ 처리할 문서가 없습니다. documents 폴더에 파일을 넣어주세요.")
+            return False
+
+        # 2. 벡터 데이터베이스 구축
+        success = self.vector_store.rebuild_if_needed(documents, force_rebuild)
+        if not success:
+            print("❌ 벡터 데이터베이스 구축 실패")
+            return False
+
+        # 3. LLM 모델 로드 (선택적 - 메모리 부족 시 스킵)
+        try:
+            self._load_llm()
+        except Exception as e:
+            print(f"⚠️ LLM 모델 로드 실패: {str(e)}")
+            print("📝 템플릿 기반 응답 모드로 동작합니다.")
+
+        self.is_initialized = True
+        print("✅ RAG 챗봇 초기화 완료")
+        return True
+
+    def _load_documents(self, docs_folder: str) -> List[Document]:
+        """문서 로드 및 처리"""
+        if not os.path.exists(docs_folder):
+            print(f"⚠️ 문서 폴더가 존재하지 않습니다: {docs_folder}")
+            return []
+
+        print(f"📂 문서 폴더: {docs_folder}")
+        raw_documents = self.document_processor.load_documents_from_folder(docs_folder)
+        processed_documents = self.document_processor.process_documents(raw_documents)
+
+        print(f"✅ 총 {len(processed_documents)}개 문서 청크 생성 완료")
+        return processed_documents
+
+    def _load_llm(self):
+        """LLM 모델 로드"""
+        print(f"🧠 LLM 모델 로드: {Config.LLM_MODEL}")
+
+        try:
+            self.llm_tokenizer = AutoTokenizer.from_pretrained(
+                Config.LLM_MODEL,
+                trust_remote_code=True
+            )
+
+            # 패딩 토큰 설정
+            if self.llm_tokenizer.pad_token is None:
+                self.llm_tokenizer.pad_token = self.llm_tokenizer.eos_token
+
+            self.llm = AutoModelForCausalLM.from_pretrained(
+                Config.LLM_MODEL,
+                torch_dtype=torch.float16,
+                device_map="auto",
+                trust_remote_code=True
+            )
+
+            print("✅ LLM 모델 로드 완료")
+
+        except Exception as e:
+            raise Exception(f"LLM 모델 로드 실패: {str(e)}")
+
+    def search_relevant_docs(self, query: str, k: int = 3) -> List[Tuple[Document, float]]:
+        """관련 문서 검색"""
+        if not self.is_initialized:
+            print("⚠️ 챗봇이 초기화되지 않았습니다.")
+            return []
+
+        # 쿼리 전처리
+        processed_query = self._preprocess_query(query)
+
+        # 벡터 검색
+        results = self.vector_store.search_similar(processed_query, k)
+
+        # 유사도 필터링
+        filtered_results = [
+            (doc, similarity) for doc, similarity in results
+            if similarity > 0.3  # 최소 유사도 임계값
+        ]
+
+        return filtered_results
+
+    def _preprocess_query(self, query: str) -> str:
+        """쿼리 전처리"""
+        # 불필요한 공백 제거
+        query = re.sub(r'\s+', ' ', query.strip())
+
+        # 복무관리 관련 키워드 강화
+        keyword_mappings = {
+            "연차": "연차휴가",
+            "휴가": "휴가사용",
+            "근무": "근무시간",
+            "당직": "당직근무",
+            "인사": "인사평가",
+            "승진": "승진시험"
+        }
+
+        for keyword, enhanced in keyword_mappings.items():
+            if keyword in query and enhanced not in query:
+                query = query.replace(keyword, enhanced)
+
+        return query
+
+    def generate_answer(self, query: str, use_llm: bool = True) -> ChatResponse:
+        """질문에 대한 답변 생성"""
+        import time
+        start_time = time.time()
+
+        if not self.is_initialized:
+            return ChatResponse(
+                answer="죄송합니다. 챗봇이 초기화되지 않았습니다. 관리자에게 문의해주세요.",
+                sources=[],
+                confidence=0.0,
+                response_time=time.time() - start_time
+            )
+
+        # 1. 관련 문서 검색
+        relevant_docs = self.search_relevant_docs(query, k=Config.MAX_RETRIEVE_DOCS)
+
+        if not relevant_docs:
+            return ChatResponse(
+                answer="죄송합니다. 질문과 관련된 정보를 찾을 수 없습니다. 다른 방식으로 질문해주시거나 관련 부서에 문의해주시기 바랍니다.",
+                sources=[],
+                confidence=0.0,
+                response_time=time.time() - start_time
+            )
+
+        # 2. 답변 생성
+        if use_llm and self.llm is not None:
+            answer = self._generate_llm_answer(query, relevant_docs)
+        else:
+            answer = self._generate_template_answer(query, relevant_docs)
+
+        # 3. 출처 정보 준비
+        sources = [
+            {
+                "source": doc.metadata.get("source", "알 수 없음"),
+                "content": doc.page_content[:200] + "..." if len(doc.page_content) > 200 else doc.page_content,
+                "similarity": f"{similarity:.4f}"
+            }
+            for doc, similarity in relevant_docs
+        ]
+
+        # 4. 신뢰도 계산
+        confidence = min(sum(similarity for _, similarity in relevant_docs) / len(relevant_docs), 1.0)
+
+        return ChatResponse(
+            answer=answer,
+            sources=sources,
+            confidence=confidence,
+            response_time=time.time() - start_time
+        )
+
+    def _generate_llm_answer(self, query: str, relevant_docs: List[Tuple[Document, float]]) -> str:
+        """LLM으로 답변 생성"""
+        try:
+            # 문맥 구성
+            context = "\n\n".join([
+                f"[출처 {i+1}] {doc.page_content}"
+                for i, (doc, _) in enumerate(relevant_docs)
+            ])
+
+            # 프롬프트 구성
+            prompt = f"""{Config.SYSTEM_PROMPT}
+
+[참고자료]
+{context}
+
+[질문]
+{query}
+
+위 참고자료를 바탕으로 질문에 답변해주세요. 정확하고 친절하게 설명해주세요."""
+
+            # 토크나이징
+            inputs = self.llm_tokenizer(
+                prompt,
+                return_tensors="pt",
+                max_length=2048,
+                truncation=True
+            )
+
+            # 생성
+            with torch.no_grad():
+                outputs = self.llm.generate(
+                    inputs.input_ids,
+                    max_new_tokens=512,
+                    temperature=0.7,
+                    do_sample=True,
+                    pad_token_id=self.llm_tokenizer.eos_token_id
+                )
+
+            # 결과 디코딩
+            answer = self.llm_tokenizer.decode(
+                outputs[0][inputs.input_ids.shape[1]:],
+                skip_special_tokens=True
+            ).strip()
+
+            return answer
+
+        except Exception as e:
+            print(f"⚠️ LLM 답변 생성 실패: {str(e)}")
+            return self._generate_template_answer(query, relevant_docs)
+
+    def _generate_template_answer(self, query: str, relevant_docs: List[Tuple[Document, float]]) -> str:
+        """템플릿 기반 답변 생성"""
+        # 쿼리 분석
+        query_lower = query.lower()
+
+        # 가장 관련성 높은 문서
+        top_doc, top_similarity = relevant_docs[0]
+
+        # 기본 답변 형식
+        if "연차" in query_lower or "휴가" in query_lower:
+            return self._format_leave_answer(top_doc, query)
+        elif "근무시간" in query_lower or "시간" in query_lower:
+            return self._format_work_hours_answer(top_doc, query)
+        elif "당직" in query_lower:
+            return self._format_duty_answer(top_doc, query)
+        elif "인사" in query_lower or "평가" in query_lower:
+            return self._format_evaluation_answer(top_doc, query)
+        else:
+            return self._format_general_answer(top_doc, query)
+
+    def _format_leave_answer(self, doc: Document, query: str) -> str:
+        """휴가 관련 답변 형식"""
+        content = doc.page_content
+
+        answer = f"📅 연차휴가 안내\n\n"
+
+        # 숫자와 관련된 내용 추출
+        import re
+        days = re.findall(r'(\d+)일', content)
+        periods = re.findall(r'(\d+)일 전', content)
+
+        if days:
+            answer += f"• 연차휴가 일수: {days[0]}일\n"
+        if periods:
+            answer += f"• 신청 기한: {periods[0]}일 전\n"
+
+        answer += f"\n{content[:300]}..."
+
+        if len(content) > 300:
+            answer += "\n\n📋 자세한 내용은 관련 규정을 확인하시거나 인사담당자에게 문의해주세요."
+
+        return answer
+
+    def _format_work_hours_answer(self, doc: Document, query: str) -> str:
+        """근무시간 관련 답변 형식"""
+        content = doc.page_content
+
+        answer = f"⏰ 근무시간 안내\n\n"
+        answer += f"{content[:400]}..."
+
+        # 시간 정보 추출
+        import re
+        times = re.findall(r'\d{2}:\d{2}', content)
+        if times:
+            answer += f"\n\n🕐 주요 시간: {', '.join(times)}"
+
+        return answer
+
+    def _format_duty_answer(self, doc: Document, query: str) -> str:
+        """당직 관련 답변 형식"""
+        answer = f"🌙 당직근무 안내\n\n"
+        answer += f"{doc.page_content[:400]}..."
+        answer += "\n\n📞 당직 관련 추가 문의는 관리부서로 연락주세요."
+        return answer
+
+    def _format_evaluation_answer(self, doc: Document, query: str) -> str:
+        """인사평가 관련 답변 형식"""
+        answer = f"📊 인사평가 안내\n\n"
+        answer += f"{doc.page_content[:400]}..."
+        answer += "\n\n💡 평가 관련 구체적인 문의는 인사담당자에게 문의해주세요."
+        return answer
+
+    def _format_general_answer(self, doc: Document, query: str) -> str:
+        """일반 답변 형식"""
+        answer = f"📋 복무관리 안내\n\n"
+        answer += f"질문: {query}\n\n"
+        answer += f"관련 정보:\n{doc.page_content[:400]}..."
+
+        if len(doc.page_content) > 400:
+            answer += "\n\n📖 더 자세한 정보는 관련 규정 파일을 확인해주세요."
+
+        return answer
+
+    def get_stats(self) -> Dict:
+        """챗봇 통계 정보"""
+        if not self.is_initialized:
+            return {"status": "not_initialized"}
+
+        vector_stats = self.vector_store.get_stats()
+
+        return {
+            "status": "initialized",
+            "vector_store": vector_stats,
+            "llm_available": self.llm is not None,
+            "system_prompt": Config.SYSTEM_PROMPT[:100] + "..."
+        }
+
+# 테스트용 함수
+def test_rag_chatbot():
+    """RAG 챗봇 테스트"""
+    # 샘플 문서 폴더 확인
+    if not os.path.exists("documents"):
+        print("⚠️ documents 폴더가 없습니다. document_processor.py를 먼저 실행해주세요.")
+        return
+
+    # 챗봇 초기화
+    chatbot = RAGChatbot()
+    success = chatbot.initialize()
+
+    if not success:
+        return
+
+    # 테스트 질문
+    test_questions = [
+        "연차휴가는 어떻게 사용하나요?",
+        "정규근무시간은 어떻게 되나요?",
+        "당직근무가 무엇인가요?",
+        "인사평가 절차가 궁금합니다."
+    ]
+
+    # 질문 테스트
+    for question in test_questions:
+        print(f"\n❓ 질문: {question}")
+        response = chatbot.generate_answer(question, use_llm=False)  # 템플릿 모드로 테스트
+
+        print(f"🤖 답변: {response.answer[:300]}...")
+        print(f"📊 신뢰도: {response.confidence:.4f}")
+        print(f"⏱️ 응답시간: {response.response_time:.4f}초")
+        print(f"📚 출처: {len(response.sources)}개")
+
+    # 통계 정보
+    print(f"\n📈 챗봇 통계: {chatbot.get_stats()}")
+
+if __name__ == "__main__":
+    test_rag_chatbot()

requirements.txt

@@ -0,0 +1,30 @@
+# RAG 챗봇을 위한 필수 라이브러리
+langchain>=0.1.0
+langchain-community>=0.0.20
+langchain-huggingface>=0.0.3
+langchain-text-splitters>=0.0.1
+
+# 허깅페이스 관련
+transformers>=4.36.0
+torch>=2.1.0
+sentence-transformers>=2.2.2
+huggingface-hub>=0.20.0
+
+# 벡터 데이터베이스
+faiss-cpu>=1.7.4
+chromadb>=0.4.18
+
+# 웹 인터페이스
+gradio>=4.7.1
+streamlit>=1.28.0
+
+# 문서 처리
+pypdf>=3.17.0
+python-docx>=1.1.0
+openpyxl>=3.1.2
+PyMuPDF>=1.23.8
+
+# 유틸리티
+python-dotenv>=1.0.0
+tqdm>=4.66.0
+pandas>=2.1.0

vector_store.py

@@ -0,0 +1,266 @@
+import os
+import pickle
+import numpy as np
+from typing import List, Dict, Tuple
+from pathlib import Path
+from sentence_transformers import SentenceTransformer
+import faiss
+from langchain.schema import Document
+from config import Config
+
+class VectorStore:
+    """FAISS 기반 벡터 데이터베이스 클래스"""
+
+    def __init__(self, embedding_model: str = None, cache_dir: str = None):
+        self.embedding_model_name = embedding_model or Config.EMBEDDING_MODEL
+        self.cache_dir = cache_dir or Config.VECTOR_DB_PATH
+        self.model = None
+        self.index = None
+        self.documents = []
+        self.doc_ids = []
+
+        # 캐시 디렉토리 생성
+        Path(self.cache_dir).mkdir(parents=True, exist_ok=True)
+
+    def load_embedding_model(self):
+        """임베딩 모델 로드"""
+        if self.model is None:
+            print(f"📥 임베딩 모델 로드: {self.embedding_model_name}")
+            try:
+                self.model = SentenceTransformer(self.embedding_model_name)
+                print("✅ 임베딩 모델 로드 완료")
+            except Exception as e:
+                print(f"❌ 임베딩 모델 로드 실패: {str(e)}")
+                print("🔄 다국어 모델로 대체 시도...")
+                self.model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
+
+    def create_embeddings(self, texts: List[str]) -> np.ndarray:
+        """텍스트 목록에 대한 임베딩 생성"""
+        if self.model is None:
+            self.load_embedding_model()
+
+        print(f"🔄 {len(texts)}개 텍스트 임베딩 생성 중...")
+        embeddings = self.model.encode(
+            texts,
+            batch_size=32,
+            show_progress_bar=True,
+            convert_to_numpy=True,
+            normalize_embeddings=True
+        )
+        return embeddings
+
+    def build_vector_index(self, documents: List[Document]) -> bool:
+        """문서 목록으로부터 벡터 인덱스 구축"""
+        if not documents:
+            print("⚠️ 처리할 문서가 없습니다.")
+            return False
+
+        print(f"🏗️ {len(documents)}개 문서로 벡터 인덱스 구축 시작...")
+
+        # 문서 저장
+        self.documents = documents
+
+        # 텍스트 추출
+        texts = [doc.page_content for doc in documents]
+
+        # 임베딩 생성
+        embeddings = self.create_embeddings(texts)
+
+        # FAISS 인덱스 생성
+        dimension = embeddings.shape[1]
+        self.index = faiss.IndexFlatIP(dimension)  # 내적 기반 유사도 검색
+
+        # 임베딩 추가
+        self.index.add(embeddings.astype('float32'))
+
+        # 문서 ID 생성
+        self.doc_ids = list(range(len(documents)))
+
+        print(f"✅ 벡터 인덱스 구축 완료 (차원: {dimension}, 문서: {len(documents)})")
+
+        # 인덱스 저장
+        self.save_index()
+
+        return True
+
+    def search_similar(self, query: str, k: int = 5) -> List[Tuple[Document, float]]:
+        """유사 문서 검색"""
+        if self.index is None:
+            print("⚠️ 벡터 인덱스가 생성되지 않았습니다.")
+            return []
+
+        if self.model is None:
+            self.load_embedding_model()
+
+        # 쿼리 임베딩 생성
+        query_embedding = self.model.encode([query], normalize_embeddings=True)
+        query_embedding = query_embedding.astype('float32')
+
+        # 검색
+        k = min(k, len(self.documents))
+        similarities, indices = self.index.search(query_embedding, k)
+
+        # 결과 변환
+        results = []
+        for i in range(k):
+            idx = indices[0][i]
+            similarity = similarities[0][i]
+
+            if 0 <= idx < len(self.documents):
+                doc = self.documents[idx]
+                results.append((doc, float(similarity)))
+
+        return results
+
+    def save_index(self):
+        """벡터 인덱스 및 문서 저장"""
+        if self.index is None:
+            return
+
+        try:
+            # FAISS 인덱스 저장
+            index_path = os.path.join(self.cache_dir, "faiss_index.bin")
+            faiss.write_index(self.index, index_path)
+
+            # 문서 및 메타데이터 저장
+            metadata_path = os.path.join(self.cache_dir, "metadata.pkl")
+            metadata = {
+                'documents': self.documents,
+                'doc_ids': self.doc_ids,
+                'embedding_model': self.embedding_model_name,
+                'total_documents': len(self.documents)
+            }
+
+            with open(metadata_path, 'wb') as f:
+                pickle.dump(metadata, f)
+
+            print(f"💾 벡터 인덱스 저장 완료: {self.cache_dir}")
+
+        except Exception as e:
+            print(f"❌ 인덱스 저장 실패: {str(e)}")
+
+    def load_index(self) -> bool:
+        """저장된 벡터 인덱스 로드"""
+        try:
+            index_path = os.path.join(self.cache_dir, "faiss_index.bin")
+            metadata_path = os.path.join(self.cache_dir, "metadata.pkl")
+
+            if not os.path.exists(index_path) or not os.path.exists(metadata_path):
+                return False
+
+            # FAISS 인덱스 로드
+            self.index = faiss.read_index(index_path)
+
+            # 메타데이터 로드
+            with open(metadata_path, 'rb') as f:
+                metadata = pickle.load(f)
+
+            self.documents = metadata['documents']
+            self.doc_ids = metadata['doc_ids']
+            self.embedding_model_name = metadata.get('embedding_model', Config.EMBEDDING_MODEL)
+
+            # 임베딩 모델 로드
+            self.load_embedding_model()
+
+            print(f"📖 벡터 인덱스 로드 완료 (문서: {len(self.documents)}개)")
+            return True
+
+        except Exception as e:
+            print(f"❌ 인덱스 로드 실패: {str(e)}")
+            return False
+
+    def get_stats(self) -> Dict:
+        """벡터 데이터베이스 통계 정보"""
+        if self.index is None:
+            return {"status": "no_index"}
+
+        return {
+            "total_documents": len(self.documents),
+            "embedding_model": self.embedding_model_name,
+            "index_dimension": self.index.d,
+            "cache_directory": self.cache_dir,
+            "is_trained": self.index.is_trained if hasattr(self.index, 'is_trained') else True
+        }
+
+    def rebuild_if_needed(self, documents: List[Document], force_rebuild: bool = False) -> bool:
+        """필요시 인덱스 재구축"""
+        # 기존 인덱스가 있고 강제 재구축이 없는 경우
+        if not force_rebuild and self.load_index():
+            # 문서 개수가 크게 변경되지 않았으면 재사용
+            if abs(len(self.documents) - len(documents)) / max(len(self.documents), 1) < 0.1:
+                print("📦 기존 인덱스 재사용")
+                return True
+
+        print("🔄 벡터 인덱스 재구축")
+        return self.build_vector_index(documents)
+
+    def add_documents(self, new_documents: List[Document]) -> bool:
+        """새 문서 추가 (동적 업데이트)"""
+        if not new_documents:
+            return False
+
+        # 임베딩 생성
+        new_texts = [doc.page_content for doc in new_documents]
+        new_embeddings = self.create_embeddings(new_texts)
+
+        if self.index is None:
+            # 인덱스가 없으면 새로 생성
+            return self.build_vector_index(new_documents)
+
+        # 기존 인덱스에 추가
+        self.index.add(new_embeddings.astype('float32'))
+
+        # 문서 목록 업데이트
+        start_id = len(self.documents)
+        self.documents.extend(new_documents)
+        self.doc_ids.extend(range(start_id, start_id + len(new_documents)))
+
+        print(f"➕ {len(new_documents)}개 문서 추가 완료")
+
+        # 저장
+        self.save_index()
+        return True
+
+    def delete_document(self, doc_id: int) -> bool:
+        """문서 삭제 (실제로는 인덱스 재구축 필요)"""
+        if doc_id < 0 or doc_id >= len(self.documents):
+            return False
+
+        # 해당 문서 제외하고 재구축
+        remaining_docs = [doc for i, doc in enumerate(self.documents) if i != doc_id]
+        return self.build_vector_index(remaining_docs)
+
+# 테스트용 함수
+def test_vector_store():
+    """벡터 데이터베이스 테스트"""
+    from document_processor import DocumentProcessor
+
+    # 문서 처리
+    processor = DocumentProcessor()
+    documents = processor.load_documents_from_folder("documents")
+
+    if not documents:
+        print("⚠️ 테스트할 문서가 없습니다.")
+        return
+
+    # 벡터 데이터베이스 생성
+    vector_store = VectorStore()
+    vector_store.build_vector_index(documents)
+
+    # 검색 테스트
+    test_queries = [
+        "연차휴가 사용 방법",
+        "근무시간은 어떻게 되나요?",
+        "당직근무 절차"
+    ]
+
+    for query in test_queries:
+        print(f"\n🔍 검색: {query}")
+        results = vector_store.search_similar(query, k=3)
+
+        for i, (doc, similarity) in enumerate(results):
+            print(f"  {i+1}. 유사도: {similarity:.4f}")
+            print(f"     내용: {doc.page_content[:100]}...")
+
+if __name__ == "__main__":
+    test_vector_store()