Spaces:
Sleeping
Sleeping
| #!/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, Optional | |
| import hashlib | |
| import time | |
| import shutil | |
| # 현재 디렉토리를 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 | |
| # OpenAI/Supabase 관련 모듈은 에러 처리 후 임포트 | |
| try: | |
| from openai_chatbot import OpenAIRAGChatbot | |
| OPENAI_AVAILABLE = True | |
| except ImportError as e: | |
| print(f"OpenAI 모듈 임포트 오류: {e}") | |
| print("FAISS 모드로 실행됩니다.") | |
| OPENAI_AVAILABLE = False | |
| OpenAIRAGChatbot = None | |
| class AdminManager: | |
| """관리자 기능을 담당하는 클래스""" | |
| def __init__(self): | |
| self.admin_password = os.getenv("ADMIN_PASSWORD") | |
| self.enabled = bool(self.admin_password) | |
| self.authenticated = False | |
| self.auth_time = None | |
| self.session_timeout = 3600 # 1시간 | |
| self.uploaded_files_dir = Path("uploaded_documents") | |
| self.uploaded_files_dir.mkdir(exist_ok=True) | |
| def _hash_password(self, password: str) -> str: | |
| """비밀번호 해싱""" | |
| return hashlib.sha256(password.encode()).hexdigest() | |
| def authenticate(self, password: str) -> Tuple[bool, str]: | |
| """관리자 인증""" | |
| if not self.enabled: | |
| return False, "❌ 관리자 기능이 비활성화되었습니다. ADMIN_PASSWORD를 설정해주세요." | |
| if self._hash_password(password) == self._hash_password(self.admin_password): | |
| self.authenticated = True | |
| self.auth_time = time.time() | |
| return True, "✅ 인증 성공! 관리자 기능에 접근할 수 있습니다." | |
| else: | |
| self.authenticated = False | |
| return False, "❌ 비밀번호가 올바르지 않습니다." | |
| def is_authenticated(self) -> bool: | |
| """인증 상태 확인""" | |
| if not self.enabled: | |
| return False | |
| if not self.authenticated: | |
| return False | |
| # 세션 타임아웃 확인 | |
| if self.auth_time and (time.time() - self.auth_time) > self.session_timeout: | |
| self.authenticated = False | |
| return False | |
| return True | |
| def logout(self) -> Tuple[bool, str]: | |
| """로그아웃""" | |
| self.authenticated = False | |
| self.auth_time = None | |
| return False, "👋 로그아웃되었습니다." | |
| def get_document_list(self) -> pd.DataFrame: | |
| """업로드된 문서 목록 가져오기""" | |
| documents_folder = Path(Config.DOCS_FOLDER) | |
| docs = [] | |
| if documents_folder.exists(): | |
| for file_path in documents_folder.glob("*"): | |
| if file_path.is_file(): | |
| stat = file_path.stat() | |
| docs.append({ | |
| "파일명": file_path.name, | |
| "크기": f"{stat.st_size / 1024:.1f} KB", | |
| "수정일": time.strftime("%Y-%m-%d %H:%M", time.localtime(stat.st_mtime)), | |
| "타입": file_path.suffix.upper() | |
| }) | |
| return pd.DataFrame(docs) if docs else pd.DataFrame(columns=["파일명", "크기", "수정일", "타입"]) | |
| def process_uploaded_files(self, files) -> str: | |
| """업로드된 파일 처리""" | |
| if not self.enabled: | |
| return "❌ 관리자 기능이 비활성화되어 있습니다. ADMIN_PASSWORD를 설정해주세요." | |
| if not self.is_authenticated(): | |
| return "❌ 관리자 인증이 필요합니다." | |
| if not files: | |
| return "⚠️ 업로드된 파일이 없습니다." | |
| try: | |
| # documents 폴더로 파일 복사 | |
| docs_folder = Path(Config.DOCS_FOLDER) | |
| docs_folder.mkdir(exist_ok=True) | |
| processed_files = [] | |
| errors = [] | |
| for file_obj in files: | |
| try: | |
| # Gradio File 객체에서 파일 경로 가져오기 | |
| if hasattr(file_obj, 'name'): | |
| file_path = file_obj.name | |
| original_name = Path(file_path).name | |
| elif isinstance(file_obj, str): | |
| file_path = file_obj | |
| original_name = Path(file_path).name | |
| else: | |
| # file_obj이 파일 경로 문자열인 경우 | |
| file_path = str(file_obj) | |
| original_name = Path(file_path).name | |
| # 파일명 안전 처리 | |
| safe_name = "".join(c for c in original_name if c.isalnum() or c in "._-") | |
| if not safe_name: | |
| safe_name = f"document_{len(processed_files)}.txt" | |
| dest_path = docs_folder / safe_name | |
| # 파일 복사 | |
| shutil.copy2(file_path, dest_path) | |
| processed_files.append(safe_name) | |
| except Exception as e: | |
| file_name = getattr(file_obj, 'name', str(file_obj)) | |
| errors.append(f"{Path(file_name).name}: {str(e)}") | |
| # 결과 메시지 생성 | |
| result_msg = f"📁 **파일 처리 완료**\n\n" | |
| if processed_files: | |
| result_msg += f"✅ **성공 ({len(processed_files)}개)**:\n" | |
| for file_name in processed_files: | |
| result_msg += f"- {file_name}\n" | |
| if errors: | |
| result_msg += f"\n❌ **오류 ({len(errors)}개)**:\n" | |
| for error in errors: | |
| result_msg += f"- {error}\n" | |
| result_msg += f"\n💡 **다음 단계**: 하단 '벡터 DB 재구축' 버튼을 눌러 임베딩을 진행해주세요." | |
| return result_msg | |
| except Exception as e: | |
| return f"❌ 파일 처리 중 오류 발생: {str(e)}" | |
| def rebuild_vector_db(self, force_rebuild: bool = True) -> str: | |
| """벡터 DB 재구축""" | |
| if not self.enabled: | |
| return "❌ 관리자 기능이 비활성화되어 있습니다. ADMIN_PASSWORD를 설정해주세요." | |
| if not self.is_authenticated(): | |
| return "❌ 관리자 인증이 필요합니다." | |
| try: | |
| # 챗봇 재초기화 | |
| docs_folder = Path(Config.DOCS_FOLDER) | |
| if not docs_folder.exists() or not list(docs_folder.glob("*")): | |
| return "⚠️ documents 폴더에 파일이 없습니다. 먼저 파일을 업로드해주세요." | |
| # 기존 챗봇 재초기화 | |
| success = self.chatbot.initialize(str(docs_folder), force_rebuild) | |
| if success: | |
| # 통계 정보 가져오기 | |
| stats = self.chatbot.get_stats() | |
| vector_stats = stats.get("vector_store", {}) | |
| result_msg = f"🔄 **벡터 DB 재구축 완료**\n\n" | |
| result_msg += f"- 📄 처리 문서: {vector_stats.get('total_documents', 0)}개\n" | |
| result_msg += f"- 🧠 임베딩 모델: {vector_stats.get('embedding_model', 'N/A')}\n" | |
| result_msg += f"- 📏 벡터 차원: {vector_stats.get('index_dimension', 'N/A')}\n" | |
| result_msg += f"- ✅ 챗봇 업데이트 완료" | |
| return result_msg | |
| else: | |
| return "❌ 벡터 DB 재구축 실패" | |
| except Exception as e: | |
| return f"❌ 벡터 DB 재구축 중 오류 발생: {str(e)}" | |
| def delete_document(self, filename: str) -> str: | |
| """문서 삭제""" | |
| if not self.enabled: | |
| return "❌ 관리자 기능이 비활성화되어 있습니다. ADMIN_PASSWORD를 설정해주세요." | |
| if not self.is_authenticated(): | |
| return "❌ 관리자 인증이 필요합니다." | |
| try: | |
| file_path = Path(Config.DOCS_FOLDER) / filename | |
| if file_path.exists(): | |
| file_path.unlink() | |
| return f"✅ '{filename}' 파일이 삭제되었습니다." | |
| else: | |
| return f"⚠️ '{filename}' 파일을 찾을 수 없습니다." | |
| except Exception as e: | |
| return f"❌ 파일 삭제 중 오류 발생: {str(e)}" | |
| class HuggingFaceApp: | |
| """허깅페이스 Spaces 배포용 앱 클래스""" | |
| def __init__(self): | |
| # 벡터 DB 타입에 따라 챗봇 선택 | |
| if self._should_use_supabase(): | |
| try: | |
| self.chatbot = OpenAIRAGChatbot() | |
| except Exception as e: | |
| print(f"Supabase 챗봇 초기화 오류: {e}") | |
| print("FAISS 모드로 전환됩니다.") | |
| self.chatbot = RAGChatbot() | |
| else: | |
| self.chatbot = RAGChatbot() | |
| self.is_initialized = False | |
| # 관리자 매니저 초기화 | |
| self.admin_manager = AdminManager() | |
| # 관리자 매니저에 챗봇 참조 전달 | |
| self.admin_manager.chatbot = self.chatbot | |
| # 예시 질문 (허깅페이스 환경 최적화) | |
| self.example_questions = [ | |
| "연차휴가 사용 방법을 알려주세요", | |
| "정규근무시간은 어떻게 되나요?", | |
| "당직근무 절차가 궁금합니다", | |
| "인사평가는 언제 진행되나요?", | |
| "파견근무 신청 방법", | |
| "복무규정 위반 시 처리" | |
| ] | |
| # 앱 초기화 | |
| self._initialize_app() | |
| def _should_use_supabase(self) -> bool: | |
| """Supabase 경로 사용 여부 판단""" | |
| has_supabase_env = bool(Config.SUPABASE_URL and Config.SUPABASE_KEY) | |
| has_openai = bool(Config.OPENAI_API_KEY) | |
| if Config.VECTOR_DB_TYPE != "supabase": | |
| return False | |
| if not has_supabase_env: | |
| print("Supabase 환경 변수가 없어 FAISS 모드로 실행됩니다.") | |
| return False | |
| if not has_openai: | |
| print("OpenAI API Key가 없어 FAISS 모드로 실행됩니다.") | |
| return False | |
| if not (OPENAI_AVAILABLE and OpenAIRAGChatbot): | |
| print("OpenAI 모듈을 사용할 수 없어 FAISS 모드로 실행됩니다.") | |
| return False | |
| return True | |
| def _initialize_app(self): | |
| """앱 초기화""" | |
| try: | |
| print("Starting Fire Service Management RAG Chatbot...") | |
| # 문서 폴더 확인 및 샘플 데이터 생성 | |
| self._ensure_documents() | |
| # 챗봇 초기화 | |
| success = self.chatbot.initialize() | |
| if success: | |
| self.is_initialized = True | |
| print("Chatbot initialization completed") | |
| else: | |
| print("Chatbot initialization failed - running in template mode") | |
| except Exception as e: | |
| print(f"Initialization error: {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("Creating sample documents...") | |
| 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 데모 생성""" | |
| custom_css = """ | |
| .gradio-container { | |
| max-width: 1200px !important; | |
| margin: auto !important; | |
| font-family: 'Segoe UI', 'Pretendard', system-ui, -apple-system, sans-serif; | |
| background: #0f172a; | |
| color: #e2e8f0; | |
| } | |
| .main-hero { | |
| background: radial-gradient(circle at 20% 20%, rgba(255,94,94,0.25), rgba(15,23,42,0.4)), | |
| radial-gradient(circle at 80% 0%, rgba(255,214,102,0.25), rgba(15,23,42,0.4)), | |
| linear-gradient(135deg, #111827 0%, #0b1222 100%); | |
| padding: 28px; | |
| border-radius: 18px; | |
| border: 1px solid #1f2937; | |
| box-shadow: 0 20px 60px rgba(0,0,0,0.35); | |
| margin-bottom: 18px; | |
| } | |
| .hero-title { font-size: 2.4rem; font-weight: 800; margin: 0 0 8px 0; color: #fff; letter-spacing: -0.5px; } | |
| .hero-sub { font-size: 1.05rem; color: #cbd5e1; margin: 0; } | |
| .pill { | |
| display: inline-flex; | |
| align-items: center; | |
| gap: 8px; | |
| padding: 8px 12px; | |
| border-radius: 999px; | |
| background: rgba(255,255,255,0.08); | |
| color: #e2e8f0; | |
| font-size: 0.95rem; | |
| border: 1px solid rgba(255,255,255,0.12); | |
| } | |
| .stat-grid { | |
| display: grid; | |
| grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); | |
| gap: 12px; | |
| margin-top: 14px; | |
| } | |
| .stat-card { | |
| background: rgba(255,255,255,0.03); | |
| border: 1px solid rgba(255,255,255,0.08); | |
| border-radius: 14px; | |
| padding: 14px; | |
| } | |
| .stat-value { font-size: 1.4rem; font-weight: 700; color: #fff; } | |
| .stat-label { font-size: 0.9rem; color: #cbd5e1; } | |
| .card { | |
| background: #0b1222; | |
| border: 1px solid #1f2937; | |
| border-radius: 14px; | |
| padding: 14px; | |
| box-shadow: 0 12px 30px rgba(0,0,0,0.3); | |
| } | |
| .card h3 { margin-top: 0; color: #fff; } | |
| .badge { | |
| display: inline-block; | |
| padding: 6px 10px; | |
| border-radius: 10px; | |
| font-size: 0.85rem; | |
| background: #f97316; | |
| color: #0b1222; | |
| font-weight: 700; | |
| } | |
| .info-note { | |
| background: rgba(255,255,255,0.05); | |
| border: 1px dashed rgba(255,255,255,0.15); | |
| padding: 12px; | |
| border-radius: 12px; | |
| color: #cbd5e1; | |
| } | |
| """ | |
| stats = self.chatbot.get_stats() if self.is_initialized else {} | |
| vector_stats = stats.get("vector_store", {}) | |
| doc_count = vector_stats.get("total_documents", 0) | |
| index_dim = vector_stats.get("index_dimension", "N/A") | |
| embedding_model = vector_stats.get("embedding_model", "N/A") | |
| status_message = ( | |
| "✅ **챗봇 준비 완료** - 복무관리 관련 질문을 입력해주세요" | |
| if self.is_initialized else | |
| "⚠️ **챗봇 초기화 필요** - 관리자 탭에서 문서를 업로드하고 재구축하세요" | |
| ) | |
| admin_enabled = self.admin_manager.enabled | |
| admin_status_init = ( | |
| "🔒 **관리자 기능 비활성화** - ADMIN_PASSWORD를 설정하세요" | |
| if not admin_enabled else | |
| "🔒 **인증 필요** - 관리자 비밀번호를 입력해주세요" | |
| ) | |
| admin_notice = ( | |
| "<div style='color:#fca5a5; font-weight:bold;'>ADMIN_PASSWORD가 설정되지 않아 관리자 기능이 비활성화되었습니다.</div>" | |
| if not admin_enabled else "" | |
| ) | |
| with gr.Blocks( | |
| title="소방 복무관리 RAG 챗봇", | |
| theme=gr.themes.Soft(), | |
| css=custom_css | |
| ) as demo: | |
| gr.HTML(f""" | |
| <div class="main-hero"> | |
| <div style="display:flex; gap:18px; align-items:flex-start; flex-wrap:wrap;"> | |
| <div style="flex:3; min-width:260px;"> | |
| <div class="pill">🚒 소방 복무관리 · RAG</div> | |
| <h1 class="hero-title">규정·절차를 바로 찾는<br>복무관리 AI 컨시어지</h1> | |
| <p class="hero-sub">연차/당직/인사평가/교육훈련까지 신뢰할 수 있는 출처 기반 답변을 제공합니다.</p> | |
| <div style="margin-top:12px; display:flex; gap:10px; flex-wrap:wrap;"> | |
| <span class="pill">🔍 실시간 검색</span> | |
| <span class="pill">📚 출처 명시</span> | |
| <span class="pill">🛡️ 보안 인증</span> | |
| </div> | |
| </div> | |
| <div class="card" style="flex:2; min-width:240px; background:rgba(255,255,255,0.03);"> | |
| <div style="display:flex; justify-content:space-between; align-items:center;"> | |
| <h3 style="margin:0;">실시간 상태</h3> | |
| <span class="badge">{'LIVE' if self.is_initialized else 'SETUP'}</span> | |
| </div> | |
| <div style="margin-top:10px; color:#cbd5e1;">{status_message}</div> | |
| <div class="stat-grid" style="margin-top:10px;"> | |
| <div class="stat-card"> | |
| <div class="stat-value">{doc_count}</div> | |
| <div class="stat-label">인덱싱 문서</div> | |
| </div> | |
| <div class="stat-card"> | |
| <div class="stat-value">{index_dim}</div> | |
| <div class="stat-label">벡터 차원</div> | |
| </div> | |
| <div class="stat-card"> | |
| <div class="stat-value">{embedding_model}</div> | |
| <div class="stat-label">임베딩 모델</div> | |
| </div> | |
| </div> | |
| </div> | |
| </div> | |
| </div> | |
| """) | |
| with gr.Row(): | |
| with gr.Column(scale=7): | |
| gr.Markdown("### ⚡ 바로 묻기") | |
| quick_note = "채팅 탭에서 추천 질문을 클릭해 입력창에 불러온 뒤 바로 전송하세요." | |
| gr.HTML(f"<div class='info-note'>{quick_note}</div>") | |
| with gr.Column(scale=5): | |
| gr.Markdown("### 🧭 사용 흐름") | |
| gr.HTML(""" | |
| <div class="card"> | |
| <ol style="margin:0; padding-left:16px; color:#cbd5e1; line-height:1.6;"> | |
| <li>채팅 탭에서 질문 입력 → 답변/출처 확인</li> | |
| <li>문서 추가 필요 시 관리자 탭에서 업로드</li> | |
| <li>“벡터 DB 재구축” 후 새 문서를 반영</li> | |
| </ol> | |
| <div style="margin-top:10px; color:#94a3b8;">챗봇이 초기화되지 않았다면 문서를 업로드하고 재구축을 실행하세요.</div> | |
| </div> | |
| """) | |
| with gr.Tabs(): | |
| with gr.Tab("💬 채팅"): | |
| gr.HTML(""" | |
| <div style="padding:12px; border-radius:12px; background:#0b1222; border:1px solid #1f2937;"> | |
| <h2 style="margin:0 0 6px 0; color:#fff;">복무관리 전문가에게 물어보세요</h2> | |
| <p style="margin:0; color:#cbd5e1;">연차/당직/인사평가/교육훈련/징계 등 규정 기반 답변을 제공합니다.</p> | |
| </div> | |
| """) | |
| with gr.Row(): | |
| status_text = gr.HTML(status_message) | |
| chatbot = gr.Chatbot( | |
| height=600, | |
| placeholder="안녕하세요! 복무관리 관련 무엇이든 물어보세요.", | |
| avatar_images=["👤", "🤖"], | |
| type="tuples" | |
| ) | |
| 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(): | |
| for question in self.example_questions[:4]: | |
| gr.Button(question, size="sm").click(fn=lambda q=question: q, 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("⚠️ 챗봇이 초기화되지 않았습니다.") | |
| with gr.Tab("🔐 관리자"): | |
| gr.Markdown("### 🛡️ 관리자 전용 기능") | |
| if admin_notice: | |
| gr.HTML(admin_notice) | |
| with gr.Row(): | |
| auth_status = gr.HTML(admin_status_init) | |
| with gr.Group(): | |
| gr.Markdown("#### 🔑 관리자 인증") | |
| with gr.Row(): | |
| admin_password = gr.Textbox( | |
| placeholder="관리자 비밀번호를 입력하세요", | |
| type="password", | |
| label="비밀번호", | |
| scale=3 | |
| ) | |
| auth_btn = gr.Button("🔐 인증", variant="primary", scale=1) | |
| logout_btn = gr.Button("🚪 로그아웃", scale=1) | |
| with gr.Group(visible=admin_enabled) as admin_panel: | |
| gr.Markdown("#### 📁 문서 업로드") | |
| upload_status = gr.HTML("⏳ 파일을 선택하고 업로드해주세요.") | |
| uploaded_files = gr.File( | |
| file_count="multiple", | |
| file_types=[".txt", ".pdf", ".docx", ".xlsx", ".csv"], | |
| label="복무관리 문서 파일" | |
| ) | |
| with gr.Row(): | |
| upload_btn = gr.Button("📤 파일 업로드 및 처리", variant="primary") | |
| refresh_btn = gr.Button("🔄 목록 새로고침") | |
| gr.Markdown("#### 📋 현재 문서 목록") | |
| document_list = gr.DataFrame( | |
| headers=["파일명", "크기", "수정일", "타입"], | |
| datatype=["str", "str", "str", "str"] | |
| ) | |
| with gr.Row(): | |
| delete_selected = gr.Button("🗑️ 선택 삭제", variant="stop") | |
| rebuild_btn = gr.Button("🔄 벡터 DB 재구축", variant="secondary") | |
| gr.Markdown("#### 📊 작업 결과") | |
| operation_result = gr.HTML("작업 결과가 여기에 표시됩니다.") | |
| gr.Markdown("#### 📈 시스템 통계") | |
| system_stats = gr.HTML("시스템 상태를 확인해주세요.") | |
| with gr.Tab("📊 정보"): | |
| gr.Markdown("### 시스템 정보 및 설정") | |
| with gr.Row(): | |
| with gr.Column(): | |
| gr.Markdown("#### 🤖 챗봇 정보") | |
| chatbot_info = gr.HTML("챗봇 정보 로딩 중...") | |
| with gr.Column(): | |
| gr.Markdown("#### ⚙️ 시스템 설정") | |
| system_settings = gr.HTML(f""" | |
| - **벡터 DB 타입**: {Config.VECTOR_DB_TYPE} | |
| - **청크 크기**: {Config.CHUNK_SIZE} | |
| - **최대 검색 문서**: {Config.MAX_RETRIEVE_DOCS} | |
| - **문서 폴더**: {Config.DOCS_FOLDER} | |
| """) | |
| gr.Markdown("#### 📋 사용 방법") | |
| usage_guide = gr.HTML(""" | |
| <div class="info-note"> | |
| <strong>관리자 비밀번호</strong>: 환경변수 <code>ADMIN_PASSWORD</code>를 설정해야 관리자 기능이 활성화됩니다.<br> | |
| 초기화 필요 시 관리자 탭에서 문서를 업로드한 뒤 “벡터 DB 재구축”을 클릭하세요. | |
| </div> | |
| """) | |
| # 이벤트 핸들러 | |
| 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 | |
| def authenticate_admin(password): | |
| """관리자 인증 처리""" | |
| is_auth, message = self.admin_manager.authenticate(password) | |
| if is_auth: | |
| status_html = "🔓 **인증 성공!** 관리자 기능에 접근할 수 있습니다." | |
| doc_list = self.admin_manager.get_document_list() | |
| stats_html = self.get_system_stats_html() | |
| return status_html, doc_list, stats_html, "", "" | |
| else: | |
| status_html = f"🔒 **인증 실패** - {message}" | |
| return status_html, pd.DataFrame(columns=["파일명", "크기", "수정일", "타입"]), "인증이 필요합니다.", password, "" | |
| def logout_admin(): | |
| """관리자 로그아웃 처리""" | |
| self.admin_manager.logout() | |
| return "🔒 **로그아웃되었습니다** - 다시 인증해주세요", pd.DataFrame(columns=["파일명", "크기", "수정일", "타입"]), "인증이 필요합니다." | |
| def upload_files(files): | |
| """파일 업로드 처리""" | |
| result = self.admin_manager.process_uploaded_files(files) | |
| doc_list = self.admin_manager.get_document_list() | |
| return result, doc_list, None | |
| def rebuild_vector_db(): | |
| """벡터 DB 재구축""" | |
| result = self.admin_manager.rebuild_vector_db(force_rebuild=True) | |
| stats_html = self.get_system_stats_html() | |
| return result, stats_html | |
| def delete_selected_files(selected_rows): | |
| """선택된 파일 삭제""" | |
| if not selected_rows or not self.admin_manager.is_authenticated(): | |
| return "❌ 관리자 인증이 필요하거나 삭제할 파일을 선택해주세요.", self.admin_manager.get_document_list() | |
| results = [] | |
| for row in selected_rows: | |
| filename = row[0] | |
| result = self.admin_manager.delete_document(filename) | |
| results.append(result) | |
| doc_list = self.admin_manager.get_document_list() | |
| result_msg = "🗑️ **파일 삭제 결과**\n\n" + "\n".join(results) | |
| return result_msg, doc_list | |
| def refresh_document_list(): | |
| """문서 목록 새로고침""" | |
| if self.admin_manager.is_authenticated(): | |
| return self.admin_manager.get_document_list(), self.get_system_stats_html() | |
| else: | |
| return pd.DataFrame(columns=["파일명", "크기", "수정일", "타입"]), "인증이 필요합니다." | |
| def get_system_stats_html(): | |
| """시스템 통계 HTML 생성""" | |
| if not self.admin_manager.is_authenticated(): | |
| return "🔒 인증이 필요합니다." | |
| try: | |
| stats = self.chatbot.get_stats() if self.is_initialized else {} | |
| vector_stats = stats.get("vector_store", {}) | |
| return f""" | |
| <div style="background-color: #f8f9fa; padding: 15px; border-radius: 8px;"> | |
| <h4>📊 시스템 상태</h4> | |
| <ul> | |
| <li><strong>챗봇 상태</strong>: {'✅ 정상' if self.is_initialized else '⚠️ 초기화 필요'}</li> | |
| <li><strong>관리자 인증</strong>: {'✅ 인증됨' if self.admin_manager.is_authenticated() else '❌ 미인증'}</li> | |
| <li><strong>문서 수</strong>: {vector_stats.get('total_documents', 0)}개</li> | |
| <li><strong>임베딩 모델</strong>: {vector_stats.get('embedding_model', 'N/A')}</li> | |
| <li><strong>벡터 차원</strong>: {vector_stats.get('index_dimension', 'N/A')}</li> | |
| <li><strong>벡터 DB 타입</strong>: {Config.VECTOR_DB_TYPE}</li> | |
| </ul> | |
| </div> | |
| """ | |
| except Exception as e: | |
| return f"❌ 통계 정보 로딩 실패: {str(e)}" | |
| def get_chatbot_info_html(): | |
| """챗봇 정보 HTML 생성""" | |
| try: | |
| if self.is_initialized: | |
| stats = self.chatbot.get_stats() | |
| vector_stats = stats.get("vector_store", {}) | |
| return f""" | |
| <div style="background-color: #e8f5e8; padding: 15px; border-radius: 8px;"> | |
| <h4>🤖 챗봇 정보</h4> | |
| <ul> | |
| <li><strong>상태</strong>: ✅ 정상 작동 중</li> | |
| <li><strong>임베딩 모델</strong>: {vector_stats.get('embedding_model', 'N/A')}</li> | |
| <li><strong>문서 수</strong>: {vector_stats.get('total_documents', 0)}개</li> | |
| <li><strong>벡터 차원</strong>: {vector_stats.get('index_dimension', 'N/A')}</li> | |
| <li><strong>LLM 사용</strong>: {'사용 가능' if stats.get('llm_available') else '템플릿 모드'}</li> | |
| </ul> | |
| </div> | |
| """ | |
| else: | |
| return """ | |
| <div style="background-color: #fff3cd; padding: 15px; border-radius: 8px;"> | |
| <h4>🤖 챗봇 정보</h4> | |
| <p>⚠️ 챗봇이 초기화되지 않았습니다.</p> | |
| </div> | |
| """ | |
| except Exception as e: | |
| return f"❌ 정보 로딩 실패: {str(e)}" | |
| # 메시지 전송 이벤트 | |
| 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] | |
| ) | |
| # 관리자 이벤트 핸들러 | |
| auth_btn.click( | |
| authenticate_admin, | |
| inputs=[admin_password], | |
| outputs=[auth_status, document_list, system_stats, admin_password, uploaded_files] | |
| ) | |
| logout_btn.click( | |
| logout_admin, | |
| outputs=[auth_status, document_list, system_stats] | |
| ) | |
| upload_btn.click( | |
| upload_files, | |
| inputs=[uploaded_files], | |
| outputs=[operation_result, document_list, uploaded_files] | |
| ) | |
| rebuild_btn.click( | |
| rebuild_vector_db, | |
| outputs=[operation_result, system_stats] | |
| ) | |
| delete_selected.click( | |
| delete_selected_files, | |
| inputs=[document_list], | |
| outputs=[operation_result, document_list] | |
| ) | |
| refresh_btn.click( | |
| refresh_document_list, | |
| outputs=[document_list, system_stats] | |
| ) | |
| # 초기 정보 로드 | |
| demo.load( | |
| fn=lambda: ( | |
| self.admin_manager.get_document_list(), | |
| get_system_stats_html(), | |
| get_chatbot_info_html() | |
| ), | |
| outputs=[document_list, system_stats, chatbot_info] | |
| ) | |
| demo.queue() # FastAPI/Starlette 응답 길이 이슈 완화 및 안정적 이벤트 처리 | |
| 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 | |
| ) | |
| if __name__ == "__main__": | |
| main() | |