Add application file

.gitignore

@@ -0,0 +1,34 @@
+# 환경 변수
+.env
+
+# 캐시 및 임시 파일
+__pycache__/
+*.py[cod]
+*.so
+.Python
+env/
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# 폴더
+documents/
+faiss_index/
+cached_data/
+preprocessed_index/
+**/__pycache__/
+
+# 프로젝트 특화 파일
+parts_extraction_cache.json
+.venv/

app.py

@@ -0,0 +1,670 @@
+"""
+음성인식(STT) 기능이 구현된 RAG 챗봇 앱
+"""
+import os
+import time
+import hashlib
+import pickle
+import json
+import tempfile
+from typing import List, Dict, Tuple, Any
+
+from langchain.schema import Document
+
+from config import (
+    PDF_DIRECTORY, CHUNK_SIZE, CHUNK_OVERLAP, LLM_MODEL,
+    STT_LANGUAGE, IS_HUGGINGFACE
+)
+from optimized_document_processor import OptimizedDocumentProcessor
+from vector_store import VectorStore
+from clova_stt import ClovaSTT
+
+# 안전한 임포트
+try:
+    from rag_chain import RAGChain
+    RAG_CHAIN_AVAILABLE = True
+except ImportError:
+    print("RAG 체인 모듈을 로드할 수 없습니다.")
+    RAG_CHAIN_AVAILABLE = False
+
+
+class AutoRAGChatApp:
+    """
+    documents 폴더의 PDF 파일을 자동으로 처리하고 음성인식 기능을 제공하는 RAG 챗봇
+    """
+
+    def __init__(self):
+        """
+        RAG 챗봇 애플리케이션 초기화
+        """
+        # 데이터 디렉토리 정의
+        self.pdf_directory = PDF_DIRECTORY
+        self.cache_directory = "cached_data"
+        self.index_file = os.path.join(self.cache_directory, "file_index.json")
+        self.chunks_dir = os.path.join(self.cache_directory, "chunks")
+        self.vector_index_dir = os.path.join(self.cache_directory, "vector_index")
+
+        # 디렉토리 생성
+        os.makedirs(self.pdf_directory, exist_ok=True)
+        os.makedirs(self.cache_directory, exist_ok=True)
+        os.makedirs(self.chunks_dir, exist_ok=True)
+        os.makedirs(self.vector_index_dir, exist_ok=True)
+
+        print(f"PDF 문서 디렉토리: '{self.pdf_directory}'")
+        print(f"캐시 디렉토리: '{self.cache_directory}'")
+
+        # 컴포넌트 초기화
+        self.document_processor = OptimizedDocumentProcessor(
+            chunk_size=CHUNK_SIZE,
+            chunk_overlap=CHUNK_OVERLAP
+        )
+
+        # 벡터 저장소 초기화
+        self.vector_store = VectorStore(use_milvus=False)
+
+        # 문서 인덱스 로드
+        self.file_index = self._load_file_index()
+
+        # 기본 변수 초기화
+        self.documents = []
+        self.processed_files = []
+        self.is_initialized = False
+
+        # 클로바 STT 클라이언트 초기화
+        self.stt_client = ClovaSTT()
+        print("음성인식(STT) 기능이 초기화되었습니다.")
+
+        # 시작 시 자동으로 문서 로드 및 처리
+        print("문서 자동 로드 및 처리 시작...")
+        self.auto_process_documents()
+
+    def _process_pdf_file(self, file_path: str) -> List[Document]:
+        """
+        PDF 파일 처리 - docling 실패 시 PyPDFLoader 사용
+
+        Args:
+            file_path: 처리할 PDF 파일 경로
+
+        Returns:
+            처리된 문서 청크 리스트
+        """
+        try:
+            print(f"docling으로 처리 시도: {file_path}")
+
+            # docling 사용 시도
+            try:
+                # 10초 타임아웃 설정 (옵션)
+                import signal
+
+                def timeout_handler(signum, frame):
+                    raise TimeoutError("docling 처리 시간 초과")
+
+                # 리눅스/맥에서만 작동 (윈도우에서는 무시됨)
+                try:
+                    signal.signal(signal.SIGALRM, timeout_handler)
+                    signal.alarm(60)  # 60초 타임아웃
+                except:
+                    pass
+
+                # docling으로 처리 시도
+                chunks = self.document_processor.process_pdf(file_path, use_docling=True)
+
+                # 타임아웃 취소
+                try:
+                    signal.alarm(0)
+                except:
+                    pass
+
+                return chunks
+
+            except Exception as e:
+                # docling 오류 확인
+                error_str = str(e)
+                if "Invalid code point" in error_str or "RuntimeError" in error_str:
+                    print(f"docling 처리 오류 (코드 포인트 문제): {error_str}")
+                    print("PyPDFLoader로 대체합니다.")
+                else:
+                    print(f"docling 처리 오류: {error_str}")
+                    print("PyPDFLoader로 대체합니다.")
+
+                # PyPDFLoader로 대체
+                try:
+                    return self.document_processor.process_pdf(file_path, use_docling=False)
+                except Exception as inner_e:
+                    print(f"PyPDFLoader 처리 오류: {inner_e}")
+                    raise  # 두 방법 모두 실패하면 예외 발생
+
+        except Exception as e:
+            print(f"PDF 처리 중 심각한 오류: {e}")
+            # 빈 청크라도 반환하여 전체 처리가 중단되지 않도록 함
+            return []
+
+    def _load_file_index(self) -> Dict[str, Dict[str, Any]]:
+        """
+        파일 인덱스 로드
+
+        Returns:
+            파일 경로 -> 메타데이터 매핑
+        """
+        if os.path.exists(self.index_file):
+            try:
+                with open(self.index_file, 'r', encoding='utf-8') as f:
+                    return json.load(f)
+            except Exception as e:
+                print(f"인덱스 파일 로드 실패: {e}")
+                return {}
+        return {}
+
+    def _save_file_index(self) -> None:
+        """
+        파일 인덱스 저장
+        """
+        with open(self.index_file, 'w', encoding='utf-8') as f:
+            json.dump(self.file_index, f, ensure_ascii=False, indent=2)
+
+    def _calculate_file_hash(self, file_path: str) -> str:
+        """
+        파일 해시 계산
+
+        Args:
+            file_path: 파일 경로
+
+        Returns:
+            MD5 해시값
+        """
+        hasher = hashlib.md5()
+        with open(file_path, 'rb') as f:
+            buf = f.read(65536)
+            while len(buf) > 0:
+                hasher.update(buf)
+                buf = f.read(65536)
+        return hasher.hexdigest()
+
+    def _is_file_processed(self, file_path: str) -> bool:
+        """
+        파일이 이미 처리되었고 변경되지 않았는지 확인
+
+        Args:
+            file_path: 파일 경로
+
+        Returns:
+            처리 여부
+        """
+        if file_path not in self.file_index:
+            return False
+
+        # 현재 해시값 계산
+        current_hash = self._calculate_file_hash(file_path)
+
+        # 저장된 해시값과 비교
+        if self.file_index[file_path]['hash'] != current_hash:
+            print(f"파일 변경 감지: {file_path}")
+            return False
+
+        # 청크 파일 존재 확인
+        chunks_path = self.file_index[file_path]['chunks_path']
+        if not os.path.exists(chunks_path):
+            return False
+
+        return True
+
+    def _get_chunks_path(self, file_hash: str) -> str:
+        """
+        청크 파일 경로 생성
+
+        Args:
+            file_hash: 파일 해시값
+
+        Returns:
+            청크 파일 경로
+        """
+        return os.path.join(self.chunks_dir, f"{file_hash}.pkl")
+
+    def _save_chunks(self, file_path: str, chunks: List[Document]) -> None:
+        """
+        청크 데이터 저장
+
+        Args:
+            file_path: 원본 파일 경로
+            chunks: 문서 청크 리스트
+        """
+        # 해시 계산
+        file_hash = self._calculate_file_hash(file_path)
+
+        # 청크 파일 경로
+        chunks_path = self._get_chunks_path(file_hash)
+
+        # 청크 데이터 저장
+        with open(chunks_path, 'wb') as f:
+            pickle.dump(chunks, f)
+
+        # 인덱스 업데이트
+        self.file_index[file_path] = {
+            'hash': file_hash,
+            'chunks_path': chunks_path,
+            'last_processed': time.time(),
+            'chunks_count': len(chunks)
+        }
+
+        # 인덱스 저장
+        self._save_file_index()
+
+        print(f"청크 저장 완료: {file_path} ({len(chunks)}개 청크)")
+
+    def _load_chunks(self, file_path: str) -> List[Document]:
+        """
+        저장된 청크 데이터 로드
+
+        Args:
+            file_path: 파일 경로
+
+        Returns:
+            문서 청크 리스트
+        """
+        chunks_path = self.file_index[file_path]['chunks_path']
+        with open(chunks_path, 'rb') as f:
+            chunks = pickle.load(f)
+
+        print(f"청크 로드 완료: {file_path} ({len(chunks)}개 청크)")
+        return chunks
+
+    def auto_process_documents(self) -> str:
+        """
+        documents 폴더의 PDF 파일 자동 처리
+
+        Returns:
+            처리 결과 메시지
+        """
+        try:
+            start_time = time.time()
+
+            # PDF 파일 목록 수집
+            pdf_files = []
+            for filename in os.listdir(self.pdf_directory):
+                if filename.lower().endswith('.pdf'):
+                    pdf_files.append(os.path.join(self.pdf_directory, filename))
+
+            if not pdf_files:
+                return f"'{self.pdf_directory}' 폴더에 PDF 파일이 없습니다."
+
+            print(f"발견된 PDF 파일: {len(pdf_files)}개")
+
+            # 폴더 내 PDF 파일 처리
+            new_files = []
+            updated_files = []
+            cached_files = []
+            failed_files = []
+            all_chunks = []
+
+            for file_path in pdf_files:
+                if self._is_file_processed(file_path):
+                    # 캐시에서 청크 로드
+                    chunks = self._load_chunks(file_path)
+                    all_chunks.extend(chunks)
+                    cached_files.append(file_path)
+                    self.processed_files.append(os.path.basename(file_path))
+                else:
+                    # 새 파일 또는 변경된 파일 처리
+                    print(f"처리 중: {file_path}")
+
+                    try:
+                        # 개선된 PDF 처리 메서드 사용
+                        chunks = self._process_pdf_file(file_path)
+
+                        if chunks:  # 청크가 있는 경우에만 저장
+                            # 청크 저장
+                            self._save_chunks(file_path, chunks)
+
+                            all_chunks.extend(chunks)
+                            if file_path in self.file_index:
+                                updated_files.append(file_path)
+                            else:
+                                new_files.append(file_path)
+
+                            self.processed_files.append(os.path.basename(file_path))
+                        else:
+                            print(f"'{file_path}' 처리 실패: 추출된 청크 없음")
+                            failed_files.append(file_path)
+                    except Exception as e:
+                        print(f"'{file_path}' 처리 중 오류: {e}")
+                        failed_files.append(file_path)
+
+            # 모든 청크 저장
+            self.documents = all_chunks
+
+            processing_time = time.time() - start_time
+            print(f"문서 처리 완료: {len(all_chunks)}개 청크, {processing_time:.2f}초")
+
+            # 벡터 인덱스 저장 경로 확인
+            if os.path.exists(self.vector_index_dir) and any(os.listdir(self.vector_index_dir)):
+                # 기존 벡터 인덱스 로드
+                try:
+                    print("저장된 벡터 인덱스 로드 중...")
+                    vector_store_loaded = self.vector_store.load_local(self.vector_index_dir)
+
+                    # 인덱스 로드 성공 확인
+                    if self.vector_store.vector_store is not None:
+                        # 새 문서나 변경된 문서가 있으면 인덱스 업데이트
+                        if new_files or updated_files:
+                            print("벡터 인덱스 업데이트 중...")
+                            self.vector_store.add_documents(self.documents)
+
+                        print("벡터 인덱스 로드 완료")
+                    else:
+                        print("벡터 인덱스를 로드했으나 유효하지 않음, 새로 생성합니다.")
+                        self.vector_store.create_or_load(self.documents)
+
+                except Exception as e:
+                    print(f"벡터 인덱스 로드 실패, 새로 생성합니다: {e}")
+                    # 오류 상세 정보 출력
+                    import traceback
+                    traceback.print_exc()
+
+                    # 새 벡터 인덱스 생성
+                    self.vector_store.create_or_load(self.documents)
+            else:
+                # 새 벡터 인덱스 생성
+                print("새 벡터 인덱스 생성 중...")
+                self.vector_store.create_or_load(self.documents)
+
+            # 벡터 인덱스 저장
+            if self.vector_store and self.vector_store.vector_store is not None:
+                try:
+                    print(f"벡터 인덱스 저장 중: {self.vector_index_dir}")
+                    save_result = self.vector_store.save_local(self.vector_index_dir)
+                    print(f"벡터 인덱스 저장 완료: {self.vector_index_dir}")
+                except Exception as e:
+                    print(f"벡터 인덱스 저장 실패: {e}")
+                    # 오류 상세 정보 출력
+                    import traceback
+                    traceback.print_exc()
+            else:
+                print("벡터 인덱스가 초기화되지 않아 저장하지 않습니다.")
+
+            # RAG 체인 초기화
+            if RAG_CHAIN_AVAILABLE:
+                self.rag_chain = RAGChain(self.vector_store)
+                self.is_initialized = True
+
+                total_time = time.time() - start_time
+
+                status_message = (
+                    f"문서 처리 완료!\n"
+                    f"- 처리된 파일: {len(self.processed_files)}개\n"
+                    f"- 캐시된 파일: {len(cached_files)}개\n"
+                    f"- 새 파일: {len(new_files)}개\n"
+                    f"- 업데이트된 파일: {len(updated_files)}개\n"
+                    f"- 실패한 파일: {len(failed_files)}개\n"
+                    f"- 총 청크 수: {len(self.documents)}개\n"
+                    f"- 처리 시간: {total_time:.2f}초\n"
+                    f"이제 질문할 준비가 되었습니다!"
+                )
+
+                print(status_message)
+                return status_message
+            else:
+                return "RAG 체인을 초기화할 수 없습니다. 필요한 라이브러리가 설치되어 있는지 확인하세요."
+
+        except Exception as e:
+            error_message = f"문서 처리 중 오류 발생: {str(e)}"
+            print(error_message)
+            import traceback
+            traceback.print_exc()
+            return error_message
+
+    def reset_cache(self) -> str:
+        """
+        캐시 초기화
+
+        Returns:
+            결과 메시지
+        """
+        try:
+            # 청크 파일 삭제
+            for filename in os.listdir(self.chunks_dir):
+                file_path = os.path.join(self.chunks_dir, filename)
+                if os.path.isfile(file_path):
+                    os.remove(file_path)
+
+            # 인덱스 초기화
+            self.file_index = {}
+            self._save_file_index()
+
+            # 벡터 인덱스 삭제
+            for filename in os.listdir(self.vector_index_dir):
+                file_path = os.path.join(self.vector_index_dir, filename)
+                if os.path.isfile(file_path):
+                    os.remove(file_path)
+
+            self.documents = []
+            self.processed_files = []
+            self.is_initialized = False
+
+            return "캐시가 초기화되었습니다. 다음 실행 시 모든 문서가 다시 처리됩니다."
+        except Exception as e:
+            return f"캐시 초기화 중 오류 발생: {str(e)}"
+
+    def process_query(self, query: str, chat_history: List[Tuple[str, str]]) -> Tuple[str, List[Tuple[str, str]]]:
+        """
+        사용자 쿼리 처리
+
+        Args:
+            query: 사용자 질문
+            chat_history: 대화 기록
+
+        Returns:
+            응답 및 업데이트된 대화 기록
+        """
+        if not query:  # 비어있는 쿼리 처리
+            return "", chat_history
+
+        if not self.is_initialized:
+            response = "문서 로드가 초기화되지 않았습니다. 자동 로드를 시도합니다."
+            chat_history.append((query, response))
+
+            # 자동 로드 시도
+            try:
+                self.auto_process_documents()
+                if not self.is_initialized:
+                    response = "문서를 로드할 수 없습니다. 'documents' 폴더에 PDF 파일이 있는지 확인하세요."
+                    chat_history.append((query, response))
+                    return "", chat_history
+            except Exception as e:
+                response = f"문서 로드 중 오류 발생: {str(e)}"
+                chat_history.append((query, response))
+                return "", chat_history
+
+        try:
+            # RAG 체인 실행 및 응답 생성
+            start_time = time.time()
+            response = self.rag_chain.run(query)
+            end_time = time.time()
+
+            query_time = end_time - start_time
+            print(f"쿼리 처리 시간: {query_time:.2f}초")
+
+            chat_history.append((query, response))
+            return "", chat_history
+        except Exception as e:
+            error_msg = f"오류 발생: {str(e)}"
+            chat_history.append((query, error_msg))
+            return "", chat_history
+
+    def process_voice_query(self, audio, chat_history: List[Tuple[str, str]]) -> Tuple[str, List[Tuple[str, str]]]:
+        """
+        음성 쿼리 처리
+
+        Args:
+            audio: 녹음된 오디오 데이터
+            chat_history: 대화 기록
+
+        Returns:
+            응답 및 업데이트된 대화 기록
+        """
+        if audio is None:
+            return "", chat_history
+
+        try:
+            # 임시 파일에 오디오 저장
+            with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file:
+                temp_path = temp_file.name
+                temp_file.write(audio)
+
+            print(f"[STT] 임시 오디오 파일 생성: {temp_path}")
+
+            # config.py에서 설정한 언어 코드로 STT 실행
+            result = self.stt_client.recognize_file(temp_path, language=STT_LANGUAGE)
+
+            # 임시 파일 삭제
+            try:
+                os.unlink(temp_path)
+                print("[STT] 임시 오디오 파일 삭제됨")
+            except Exception as e:
+                print(f"[STT] 임시 파일 삭제 실패: {e}")
+
+            # STT 결과 처리
+            if "error" in result:
+                error_msg = f"음성인식 오류: {result.get('error')}"
+                print(f"[STT] {error_msg}")
+                chat_history.append(("음성 메시지", error_msg))
+                return "", chat_history
+
+            # 인식된 텍스트 추출
+            recognized_text = result.get("text", "")
+            if not recognized_text:
+                error_msg = "음성을 인식할 수 없습니다. 다시 시도해주세요."
+                print("[STT] 인식된 텍스트 없음")
+                chat_history.append(("음성 메시지", error_msg))
+                return "", chat_history
+
+            print(f"[STT] 인식된 텍스트: {recognized_text}")
+
+            # 인식된 텍스트로 쿼리 처리 (음성 메시지 접두어 추가)
+            return self.process_query(f"🎤 {recognized_text}", chat_history)
+
+        except Exception as e:
+            error_msg = f"음성 처리 중 오류 발생: {str(e)}"
+            print(f"[STT] {error_msg}")
+            chat_history.append(("음성 메시지", error_msg))
+            return "", chat_history
+
+    def launch_app(self) -> None:
+        """
+        음성인식 기능이 추가된 Gradio 앱 실행
+        """
+        import gradio as gr
+
+        with gr.Blocks(title="음성인식 기능이 추가된 PDF 문서 기반 RAG 챗봇") as app:
+            gr.Markdown("# 음성인식 기능이 추가된 PDF 문서 기반 RAG 챗봇")
+            gr.Markdown(f"* 사용 중인 LLM 모델: **{LLM_MODEL}**")
+            gr.Markdown(f"* PDF 문서 폴더: **{self.pdf_directory}**")
+            gr.Markdown("* 네이버 클로바 음성인식 API 통합")
+
+            with gr.Row():
+                with gr.Column(scale=1):
+                    # 문서 상태 섹션
+                    status_box = gr.Textbox(
+                        label="문서 처리 상태",
+                        value=f"처리된 문서 ({len(self.processed_files)}개): {', '.join(self.processed_files)}",
+                        lines=5,
+                        interactive=False
+                    )
+
+                    # 캐시 관리 버튼
+                    refresh_button = gr.Button("문서 새로 읽기", variant="primary")
+                    reset_button = gr.Button("캐시 초기화", variant="stop")
+
+                    # 처리된 파일 정보
+                    with gr.Accordion("캐시 세부 정보", open=False):
+                        file_info = ""
+                        for file_path, info in self.file_index.items():
+                            file_info += f"- {os.path.basename(file_path)}: {info['chunks_count']}개 청크\n"
+
+                        cache_info = gr.Textbox(
+                            label="캐시된 파일 정보",
+                            value=file_info or "캐시된 파일이 없습니다.",
+                            lines=5,
+                            interactive=False
+                        )
+
+                with gr.Column(scale=2):
+                    # 채팅 인터페이스
+                    chatbot = gr.Chatbot(
+                        label="대화 내용",
+                        bubble_full_width=False,
+                        height=500,
+                        show_copy_button=True
+                    )
+
+                    with gr.Tabs() as input_tabs:
+                        # 텍스트 입력 탭
+                        with gr.Tab("텍스트 입력"):
+                            # 텍스트 입력과 전송 버튼을 수평으로 배치
+                            with gr.Row():
+                                query_box = gr.Textbox(
+                                    label="질문",
+                                    placeholder="처리된 문서 내용에 대해 질문하세요...",
+                                    lines=2,
+                                    scale=4
+                                )
+                                submit_btn = gr.Button("전송", variant="primary", scale=1)
+
+                        # 음성 입력 탭
+                        with gr.Tab("음성 입력"):
+                            audio_input = gr.Audio(
+                                label="마이크 입력",
+                                sources=["microphone"],
+                                type="bytes",
+                                format="wav"
+                            )
+                            voice_submit_btn = gr.Button("음성 질문 전송", variant="primary")
+
+                    clear_chat_button = gr.Button("대화 초기화")
+
+            # 이벤트 핸들러 설정
+            refresh_button.click(
+                fn=self.auto_process_documents,
+                inputs=[],
+                outputs=[status_box]
+            )
+
+            reset_button.click(
+                fn=lambda: (self.reset_cache(), self.auto_process_documents()),
+                inputs=[],
+                outputs=[status_box]
+            )
+
+            # 텍스트 전송 버튼 클릭 이벤트
+            submit_btn.click(
+                fn=self.process_query,
+                inputs=[query_box, chatbot],
+                outputs=[query_box, chatbot]
+            )
+
+            # 엔터키 입력 이벤트
+            query_box.submit(
+                fn=self.process_query,
+                inputs=[query_box, chatbot],
+                outputs=[query_box, chatbot]
+            )
+
+            # 음성 전송 버튼 클릭 이벤트
+            voice_submit_btn.click(
+                fn=self.process_voice_query,
+                inputs=[audio_input, chatbot],
+                outputs=[audio_input, chatbot]
+            )
+
+            # 대화 초기화 버튼
+            clear_chat_button.click(
+                fn=lambda: [],
+                outputs=[chatbot]
+            )
+
+        # 앱 실행
+        app.launch(share=False)
+
+
+if __name__ == "__main__":
+    app = AutoRAGChatApp()
+    app.launch_app()

config.py

@@ -0,0 +1,60 @@
+"""
+벡터 스토어, 임베딩 모델, LLM 등 구성 요소 설정
+"""
+import os
+from dotenv import load_dotenv
+
+# .env 파일이 있으면 로드 (로컬 환경용)
+load_dotenv(verbose=True)
+
+# 환경 감지
+IS_HUGGINGFACE = os.getenv('SPACE_ID') is not None
+
+# API 키 및 환경 설정
+OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "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")
+LANGFUSE_PUBLIC_KEY = os.getenv("LANGFUSE_PUBLIC_KEY", "pk-lf-cd6248e2-59ad-496d-a4cb-487bb3ecfcd5")
+LANGFUSE_SECRET_KEY = os.getenv("LANGFUSE_SECRET_KEY", "sk-lf-61460a1d-e637-4c22-b5e9-9250ac2579ba")
+LANGFUSE_HOST = os.getenv("LANGFUSE_HOST", "https://cloud.langfuse.com")
+
+# 네이버 클로바 STT API 설정
+NAVER_CLIENT_ID = os.getenv("NAVER_CLIENT_ID", "xae4kga9s5")
+NAVER_CLIENT_SECRET = os.getenv("NAVER_CLIENT_SECRET", "aoSmmr3xMrdVopxGduFX5YfGZRJpu2MDUiUvlvQx")
+
+# 네이버 클로바 API 키 확인
+if NAVER_CLIENT_ID and NAVER_CLIENT_SECRET:
+    print("네이버 클로바 STT API 키가 설정되었습니다.")
+else:
+    print("경고: 네이버 클로바 STT API 키가 설정되지 않았습니다.")
+    print("STT 기능을 사용하려면 NAVER_CLIENT_ID와 NAVER_CLIENT_SECRET 환경 변수를 설정하세요.")
+
+# Milvus 벡터 DB 설정
+MILVUS_HOST = os.getenv("MILVUS_HOST", "localhost")
+MILVUS_PORT = os.getenv("MILVUS_PORT", "19530")
+MILVUS_COLLECTION = "pdf_documents"
+
+# 임베딩 모델 설정
+EMBEDDING_MODEL = "Alibaba-NLP/gte-multilingual-base"  # 다국어 지원 모델
+RERANKER_MODEL = "Alibaba-NLP/gte-multilingual-reranker-base"  # 다국어 지원 리랭커
+
+# LLM 모델 설정 (환경에 따라 자동 선택)
+if IS_HUGGINGFACE:
+    # HuggingFace 환경에서는 OpenAI 사용
+    USE_OPENAI = True
+    LLM_MODEL = "gpt-3.5-turbo"  # 또는 다른 적절한 모델
+    print("HuggingFace Spaces 환경 감지: OpenAI 모델 사용")
+else:
+    # 로컬 환경에서는 Ollama 사용
+    USE_OPENAI = os.getenv("USE_OPENAI", "False").lower() == "true"
+    LLM_MODEL = os.getenv("LLM_MODEL", "gemma3:latest" if not USE_OPENAI else "gpt-3.5-turbo")
+    OLLAMA_HOST = os.getenv("OLLAMA_HOST", "http://localhost:11434")
+    print(f"로컬 환경: {'OpenAI' if USE_OPENAI else 'Ollama'} 모델 사용")
+
+# 앱 설정
+CHUNK_SIZE = 1000
+CHUNK_OVERLAP = 200
+TOP_K_RETRIEVAL = 5  # 벡터 검색 결과 수
+TOP_K_RERANK = 3  # 리랭킹 후 선택할 결과 수
+PDF_DIRECTORY = "documents"  # PDF 문서가 저장된 디렉토리
+
+# 음성인식 설정
+STT_LANGUAGE = "Kor"  # 기본 언어 설정 (Kor, Eng, Jpn, Chn 등)

dir

@@ -0,0 +1,154 @@
+"""
+동의어 처리 모듈
+"""
+import os
+import sys
+import re
+from typing import Dict, List, Optional, Set
+
+# 기본 동의어 사전 (MP_synonyms.py 파일이 없을 경우 사용)
+DEFAULT_SYNONYMS = {
+    "엑츄레이터": "액츄에이터",
+    "액츄에이터": "액츄에이터",
+    "모터": "액츄에이터",
+    "컨박": "컨트롤박스"
+}
+
+
+class SynonymsHandler:
+    """
+    부품명의 동의어를 처리하는 클래스
+    """
+
+    def __init__(self, synonyms_file: Optional[str] = None):
+        """
+        동의어 핸들러 초기화
+
+        Args:
+            synonyms_file: 동의어 파일 경로 (선택적)
+        """
+        self.synonyms = {}
+        self.loaded = False
+
+        # 1. 기본 제공된 파일 경로 확인
+        if synonyms_file and os.path.exists(synonyms_file):
+            self._load_from_file(synonyms_file)
+
+        # 2. 일반적인 위치 확인 (.venv/SYNONYMS/MP_synonyms.py)
+        elif os.path.exists(".venv/SYNONYMS/MP_synonyms.py"):
+            self._load_from_file(".venv/SYNONYMS/MP_synonyms.py")
+
+        # 3. 현재 디렉토리 확인
+        elif os.path.exists("MP_synonyms.py"):
+            self._load_from_file("MP_synonyms.py")
+
+        # 4. 기본 동의어 사용
+        else:
+            print("동의어 파일을 찾을 수 없어 기본 동의어 사전을 사용합니다.")
+            self.synonyms = DEFAULT_SYNONYMS
+            self.loaded = True
+
+    def _load_from_file(self, file_path: str) -> None:
+        """
+        파일에서 동의어 사전 로드
+
+        Args:
+            file_path: 동의어 파일 경로
+        """
+        try:
+            # 파일 내용 읽기
+            with open(file_path, 'r', encoding='utf-8') as f:
+                content = f.read()
+
+            # SYNONYMS 딕셔너리 추출
+            synonyms_match = re.search(r'SYNONYMS\s*=\s*\{(.*?)\}', content, re.DOTALL)
+            if synonyms_match:
+                # 실행하지 않고 변환하는 방법
+                synonyms_str = "{" + synonyms_match.group(1) + "}"
+
+                # 정규식을 사용하여 딕셔너리 형태로 파싱
+                pattern = r'"([^"]*)"\s*:\s*"([^"]*)"'
+                matches = re.findall(pattern, synonyms_str)
+
+                self.synonyms = {key: value for key, value in matches}
+                self.loaded = True
+                print(f"동의어 사전 로드 완료: {file_path}, {len(self.synonyms)}개 항목")
+            else:
+                print(f"파일에서 SYNONYMS 딕셔너리를 찾을 수 없습니다: {file_path}")
+                self.synonyms = DEFAULT_SYNONYMS
+                self.loaded = True
+
+        except Exception as e:
+            print(f"동의어 사전 로드 중 오류: {e}")
+            self.synonyms = DEFAULT_SYNONYMS
+            self.loaded = True
+
+    def find_in_text(self, text: str) -> List[str]:
+        """
+        텍스트에서 동의어 찾기
+
+        Args:
+            text: 검색할 텍스트
+
+        Returns:
+            찾은 표준화된 부품명 리스트
+        """
+        if not text or not self.loaded:
+            return []
+
+        # 공백 제거 및 소문자 변환
+        text = text.lower()
+
+        found_parts = set()
+
+        # 동의어 키워드가 텍스트에 포함되어 있는지 확인
+        for keyword, standard_name in self.synonyms.items():
+            if keyword.lower() in text:
+                found_parts.add(standard_name)
+
+        return list(found_parts)
+
+    def standardize(self, part_name: str) -> str:
+        """
+        부품명을 표준화
+
+        Args:
+            part_name: 표준화할 부품명
+
+        Returns:
+            표준화된 부품명
+        """
+        if not part_name or not self.loaded:
+            return part_name
+
+        # 소문자 변환하여 비교
+        part_lower = part_name.lower().strip()
+
+        # 동의어 사전에서 검색
+        for keyword, standard_name in self.synonyms.items():
+            if part_lower == keyword.lower():
+                return standard_name
+
+        # 매칭되지 않으면 원래 이름 반환
+        return part_name
+
+    def standardize_parts_list(self, parts: List[str]) -> List[str]:
+        """
+        부품명 리스트를 표준화
+
+        Args:
+            parts: 표준화할 부품명 리스트
+
+        Returns:
+            표준화된 부품명 리스트
+        """
+        if not parts or not self.loaded:
+            return parts
+
+        standardized = set()
+
+        for part in parts:
+            if part:
+                standardized.add(self.standardize(part))
+
+        return list(standardized)

monitoring.py

@@ -0,0 +1,136 @@
+"""
+Langfuse를 활용한 모니터링 구현 (선택적)
+"""
+from typing import Dict, Any, Optional
+import time
+import os
+from dotenv import load_dotenv
+
+# 환경 변수 로드
+load_dotenv()
+
+# 설정 값 가져오기
+LANGFUSE_SECRET_KEY = os.getenv("LANGFUSE_SECRET_KEY", "")
+LANGFUSE_PUBLIC_KEY = os.getenv("LANGFUSE_PUBLIC_KEY", "")
+LANGFUSE_HOST = os.getenv("LANGFUSE_HOST", "https://cloud.langfuse.com")
+
+class LangfuseMonitoring:
+    def __init__(self):
+        """
+        Langfuse 모니터링 초기화 (선택적 기능)
+        """
+        self.enabled = False
+        print("모니터링 기능을 초기화합니다...")
+
+        # Langfuse가 설치되어 있는지 확인
+        try:
+            from langfuse import Langfuse
+
+            if LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY:
+                try:
+                    self.langfuse = Langfuse(
+                        public_key=LANGFUSE_PUBLIC_KEY,
+                        secret_key=LANGFUSE_SECRET_KEY,
+                        host=LANGFUSE_HOST,
+                    )
+                    self.enabled = True
+                    print("Langfuse 모니터링이 활성화되었습니다.")
+                except Exception as e:
+                    print(f"Langfuse 초기화 실패: {e}")
+            else:
+                print("Langfuse API 키가 설정되지 않았습니다. 모니터링은 비활성화됩니다.")
+        except ImportError:
+            print("langfuse 패키지가 설치되지 않았습니다. 모니터링 기능이 비활성화됩니다.")
+            print("pip install langfuse 명령으로 설치할 수 있습니다.")
+
+    def start_trace(self, name: str, user_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> Any:
+        """
+        새 트레이스 시작
+
+        Args:
+            name: 트레이스 이름
+            user_id: 사용자 ID (선택적)
+            metadata: 추가 메타데이터 (선택적)
+
+        Returns:
+            트레이스 객체 또는 None
+        """
+        if not self.enabled:
+            return None
+
+        try:
+            return self.langfuse.trace(
+                name=name,
+                user_id=user_id,
+                metadata=metadata or {},
+            )
+        except Exception as e:
+            print(f"트레이스 생성 실패: {e}")
+            return None
+
+    def log_generation(self, trace: Any, name: str, prompt: str, response: str, metadata: Optional[Dict[str, Any]] = None) -> None:
+        """
+        LLM 생성 로깅
+
+        Args:
+            trace: 트레이스 객체
+            name: 생성 이름
+            prompt: 입력 프롬프트
+            response: 모델 응답
+            metadata: 추가 메타데이터 (선택적)
+        """
+        if not self.enabled or trace is None:
+            return
+
+        try:
+            trace.generation(
+                name=name,
+                model="user-defined-model",
+                prompt=prompt,
+                completion=response,
+                metadata=metadata or {},
+            )
+        except Exception as e:
+            print(f"생성 로깅 실패: {e}")
+
+    def log_span(self, trace: Any, name: str, input_data: Any, output_data: Any, start_time: float, end_time: float) -> None:
+        """
+        처리 구간 로깅
+
+        Args:
+            trace: 트레이스 객체
+            name: 구간 이름
+            input_data: 입력 데이터
+            output_data: 출력 데이터
+            start_time: 시작 시간
+            end_time: 종료 시간
+        """
+        if not self.enabled or trace is None:
+            return
+
+        try:
+            trace.span(
+                name=name,
+                start_time=start_time,
+                end_time=end_time,
+                input=input_data,
+                output=output_data,
+                metadata={"duration_ms": (end_time - start_time) * 1000},
+            )
+        except Exception as e:
+            print(f"구간 로깅 실패: {e}")
+
+    def end_trace(self, trace: Any) -> None:
+        """
+        트레이스 종료
+
+        Args:
+            trace: 종료할 트레이스 객체
+        """
+        if not self.enabled or trace is None:
+            return
+
+        try:
+            trace.update(status="success")
+        except Exception as e:
+            print(f"트레이스 종료 실패: {e}")

optimized_document_processor.py

@@ -0,0 +1,346 @@
+"""
+CPU에 최적화된 문서 처리 모듈 - 병렬 처리 적용
+"""
+import os
+import time
+from typing import List, Dict, Any, Optional
+from langchain.schema import Document
+from concurrent.futures import ThreadPoolExecutor
+
+# 멀티프로세싱 가져오기
+import multiprocessing
+
+try:
+    CPU_COUNT = multiprocessing.cpu_count()
+except:
+    CPU_COUNT = 4
+
+print(f"CPU 코어 수: {CPU_COUNT}")
+
+# docling 라이브러리 존재 여부 확인
+try:
+    from docling.datamodel.base_models import InputFormat
+    from docling.document_converter import DocumentConverter, PdfFormatOption
+    from docling.datamodel.pipeline_options import PdfPipelineOptions, TableFormerMode
+    from docling.chunking import HybridChunker
+
+    DOCLING_AVAILABLE = True
+    print("docling 라이브러리 사용 가능")
+except ImportError:
+    print("docling 라이브러리를 찾을 수 없습니다. PyPDFLoader만 사용합니다.")
+    DOCLING_AVAILABLE = False
+
+# LangChain 문서 로더
+from langchain_community.document_loaders import PyPDFLoader
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+
+
+class OptimizedDocumentProcessor:
+    """
+    CPU에 최적화된 병렬 처리 문서 처리 클래스
+    """
+
+    def __init__(self,
+                 chunk_size: int = 1000,
+                 chunk_overlap: int = 200,
+                 tokenizer: str = "Alibaba-NLP/gte-multilingual-base",  # 올바른 모델 경로로 수정
+                 max_workers: int = CPU_COUNT):
+        """
+        문서 처리기 초기화
+
+        Args:
+            chunk_size: 텍스트 청크 크기
+            chunk_overlap: 청크 간 겹침 크기
+            tokenizer: HybridChunker에서 사용할 토크나이저
+            max_workers: 병렬 처리시 최대 작업자 수
+        """
+        self.chunk_size = chunk_size
+        self.chunk_overlap = chunk_overlap
+        self.tokenizer = tokenizer
+        self.max_workers = max(1, min(max_workers, CPU_COUNT))  # CPU 코어 수 초과하지 않도록
+
+        print(f"병렬 처리 작업자 수: {self.max_workers}")
+
+        # LangChain 텍스트 스플리터
+        self.text_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=chunk_size,
+            chunk_overlap=chunk_overlap,
+            separators=["\n\n", "\n", ". ", " ", ""],
+        )
+
+        # docling 관련 컴포넌트 초기화
+        if DOCLING_AVAILABLE:
+            # 파이프라인 옵션 설정
+            self.pipeline_options = PdfPipelineOptions(do_table_structure=True)
+            self.pipeline_options.table_structure_options.mode = TableFormerMode.ACCURATE
+
+            # 문서 변환기 초기화
+            self.doc_converter = DocumentConverter(
+                format_options={
+                    InputFormat.PDF: PdfFormatOption(pipeline_options=self.pipeline_options)
+                }
+            )
+
+            # HybridChunker 초기화 (trust_remote_code=True 추가)
+            self.hybrid_chunker = HybridChunker(
+                tokenizer=tokenizer,
+                chunk_size=chunk_size,
+                overlap=chunk_overlap,
+                tokenizer_kwargs={"trust_remote_code": True}  # 원격 코드 실행 허용
+            )
+
+            print(f"docling 초기화 완료: HybridChunker(청크 크기={chunk_size}, 오버랩={chunk_overlap})")
+
+    def process_with_docling(self, pdf_path: str) -> Dict[str, Any]:
+        """
+        docling을 사용하여 PDF 문서 처리
+
+        Args:
+            pdf_path: PDF 파일 경로
+
+        Returns:
+            처리된 문서 데이터
+        """
+        if not DOCLING_AVAILABLE:
+            raise ImportError("docling 라이브러리가 설치되지 않았습니다.")
+
+        try:
+            start_time = time.time()
+
+            # 문서 변환
+            conv_res = self.doc_converter.convert(pdf_path)
+            doc = conv_res.document
+
+            # 성능 측정
+            conversion_time = time.time() - start_time
+            print(f"PDF 변환 시간: {conversion_time:.2f}초")
+
+            # 메타데이터 추출
+            metadata = {
+                "source": pdf_path,
+                "title": os.path.basename(pdf_path),
+                "processing_time": conversion_time
+            }
+
+            return {
+                "content": doc.export_to_markdown(),
+                "metadata": metadata,
+                "raw_document": doc,
+            }
+
+        except Exception as e:
+            print(f"docling으로 문서 처리 중 오류 발생: {e}")
+            raise
+
+    def chunk_with_hybrid_chunker(self, doc: Any) -> List[Dict[str, Any]]:
+        """
+        HybridChunker를 사용하여 문서를 청크로 분할
+
+        Args:
+            doc: docling 문서 객체
+
+        Returns:
+            청크 리스트
+        """
+        start_time = time.time()
+
+        # 청킹 수행
+        chunk_iter = self.hybrid_chunker.chunk(doc)
+        chunks = list(chunk_iter)
+
+        chunking_time = time.time() - start_time
+        print(f"청킹 시간: {chunking_time:.2f}초 (청크 수: {len(chunks)})")
+
+        return chunks
+
+    def create_langchain_documents_from_chunks(self,
+                                               chunks: List[Dict[str, Any]],
+                                               metadata: Dict[str, Any]) -> List[Document]:
+        """
+        docling 청크를 LangChain Document 객체로 변환
+
+        Args:
+            chunks: docling HybridChunker로 생성한 청크 리스트
+            metadata: 문서 메타데이터
+
+        Returns:
+            LangChain Document 객체 리스트
+        """
+        documents = []
+
+        for i, chunk in enumerate(chunks):
+            # 각 청크에 대한 메타데이터
+            chunk_metadata = metadata.copy()
+            chunk_metadata["chunk_id"] = i
+
+            # 청크 내용 추출
+            if hasattr(chunk, "text"):
+                content = chunk.text
+            elif hasattr(chunk, "content"):
+                content = chunk.content
+            else:
+                content = str(chunk)
+
+            document = Document(
+                page_content=content,
+                metadata=chunk_metadata
+            )
+            documents.append(document)
+
+        return documents
+
+    def process_with_langchain(self, pdf_path: str) -> List[Document]:
+        """
+        LangChain의 PyPDFLoader를 사용하여 PDF 문서 로드
+
+        Args:
+            pdf_path: PDF 파일 경로
+
+        Returns:
+            LangChain Document 객체 리스트
+        """
+        start_time = time.time()
+
+        try:
+            loader = PyPDFLoader(pdf_path)
+            documents = loader.load()
+
+            processing_time = time.time() - start_time
+            print(f"PyPDFLoader 처리 시간: {processing_time:.2f}초")
+
+            return documents
+        except Exception as e:
+            print(f"PyPDFLoader로 문서 처리 중 오류 발생: {e}")
+            raise
+
+    def process_pdf(self, pdf_path: str, use_docling: bool = True) -> List[Document]:
+        """
+        PDF 파일 처리
+
+        Args:
+            pdf_path: PDF 파일 경로
+            use_docling: docling 사용 여부
+
+        Returns:
+            처리된 문서의 청크 리스트
+        """
+        total_start_time = time.time()
+
+        # docling 사용 가능 여부 확인
+        can_use_docling = use_docling and DOCLING_AVAILABLE
+
+        if can_use_docling:
+            try:
+                # 1. docling으로 PDF 처리
+                docling_result = self.process_with_docling(pdf_path)
+                doc = docling_result["raw_document"]
+                metadata = docling_result["metadata"]
+
+                # 2. HybridChunker로 청크 생성
+                chunks = self.chunk_with_hybrid_chunker(doc)
+
+                # 3. 청크를 LangChain Document로 변환
+                documents = self.create_langchain_documents_from_chunks(chunks, metadata)
+
+                total_time = time.time() - total_start_time
+                print(f"docling 처리 완료: '{pdf_path}', {len(documents)} 청크, 총 {total_time:.2f}초")
+
+                return documents
+            except Exception as e:
+                print(f"docling 처리 실패, PyPDFLoader로 대체: {e}")
+                can_use_docling = False
+
+        if not can_use_docling:
+            # PyPDFLoader로 처리 (대체 방안)
+            documents = self.process_with_langchain(pdf_path)
+            chunks = self.text_splitter.split_documents(documents)
+
+            total_time = time.time() - total_start_time
+            print(f"PyPDFLoader 처리 완료: '{pdf_path}', {len(chunks)} 청크, 총 {total_time:.2f}초")
+
+            return chunks
+
+    def process_directory_parallel(self, directory: str, use_docling: bool = True) -> List[Document]:
+        """
+        디렉토리 내 모든 PDF 파일 병렬 처리 (멀티스레딩)
+
+        Args:
+            directory: PDF 파일 디렉토리 경로
+            use_docling: docling 사용 여부
+
+        Returns:
+            처리된 모든 문서의 청크 리스트
+        """
+        all_documents = []
+        pdf_files = []
+
+        # PDF 파일 목록 수집
+        for file in os.listdir(directory):
+            if file.endswith(".pdf"):
+                pdf_path = os.path.join(directory, file)
+                pdf_files.append(pdf_path)
+
+        if not pdf_files:
+            print(f"'{directory}' 디렉토리에 PDF 파일이 없습니다.")
+            return []
+
+        print(f"총 {len(pdf_files)}개 PDF 파일 병렬 처리 시작 (최대 {self.max_workers} 작업자)")
+        start_time = time.time()
+
+        # 병렬 처리 실행
+        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
+            # 각 PDF 파일에 대해 process_pdf 함수 병렬 실행
+            future_to_pdf = {executor.submit(self.process_pdf, pdf_path, use_docling): pdf_path
+                             for pdf_path in pdf_files}
+
+            # 결과 수집
+            for future in future_to_pdf:
+                pdf_path = future_to_pdf[future]
+                try:
+                    # 결과 가져오기
+                    chunks = future.result()
+                    all_documents.extend(chunks)
+                    print(f"'{os.path.basename(pdf_path)}' 처리 완료: {len(chunks)} 청크")
+                except Exception as e:
+                    print(f"'{pdf_path}' 처리 중 오류 발생: {e}")
+
+        total_time = time.time() - start_time
+        print(f"병렬 처리 완료: 총 {len(all_documents)} 청크, 처리 시간: {total_time:.2f}초")
+
+        return all_documents
+
+    def process_directory(self, directory: str, use_docling: bool = True, parallel: bool = True) -> List[Document]:
+        """
+        디렉토리 내 모든 PDF 파일 처리
+
+        Args:
+            directory: PDF 파일 디렉토리 경로
+            use_docling: docling 사용 여부
+            parallel: 병렬 처리 사용 여부
+
+        Returns:
+            처리된 모든 문서의 청크 리스트
+        """
+        # 병렬 처리 사용
+        if parallel:
+            return self.process_directory_parallel(directory, use_docling)
+
+        # 순차 처리
+        all_documents = []
+        start_time = time.time()
+
+        for file in os.listdir(directory):
+            if file.endswith(".pdf"):
+                pdf_path = os.path.join(directory, file)
+                print(f"처리 중: {pdf_path}")
+
+                try:
+                    chunks = self.process_pdf(pdf_path, use_docling=use_docling)
+                    all_documents.extend(chunks)
+                except Exception as e:
+                    print(f"'{pdf_path}' 처리 중 오류 발생: {e}")
+
+        total_time = time.time() - start_time
+        print(f"순차 처리 완료: 총 {len(all_documents)} 청크, 처리 시간: {total_time:.2f}초")
+
+        return all_documents

rag_chain.py

@@ -0,0 +1,151 @@
+"""
+LangChain을 활용한 RAG 체인 구현
+"""
+from typing import List, Dict, Any
+from langchain.schema import Document
+from langchain.prompts import PromptTemplate
+from langchain_core.output_parsers import StrOutputParser
+from langchain_core.runnables import RunnablePassthrough
+from langchain_community.chat_models import ChatOllama
+from langchain_openai import ChatOpenAI
+
+from config import (
+    OLLAMA_HOST, LLM_MODEL, USE_OPENAI,
+    OPENAI_API_KEY, TOP_K_RETRIEVAL, TOP_K_RERANK
+)
+from vector_store import VectorStore
+from reranker import Reranker
+
+
+class RAGChain:
+    def __init__(self, vector_store: VectorStore, use_reranker: bool = True):
+        """
+        RAG 체인 초기화 (환경에 따른 LLM 선택)
+
+        Args:
+            vector_store: 벡터 스토어 인스턴스
+            use_reranker: 리랭커 사용 여부
+        """
+        try:
+            print("RAGChain 초기화 시작...")
+            self.vector_store = vector_store
+            self.use_reranker = use_reranker
+            print(f"리랭커 사용 여부: {use_reranker}")
+
+            if use_reranker:
+                try:
+                    self.reranker = Reranker()
+                    print("리랭커 초기화 성공")
+                except Exception as e:
+                    print(f"리랭커 초기화 실패: {str(e)}")
+                    self.reranker = None
+                    self.use_reranker = False
+            else:
+                self.reranker = None
+
+            # 환경에 따른 LLM 모델 설정
+            if USE_OPENAI or IS_HUGGINGFACE:
+                print(f"OpenAI 모델 초기화: {LLM_MODEL}")
+                print(f"API 키 존재 여부: {'있음' if OPENAI_API_KEY else '없음'}")
+                try:
+                    self.llm = ChatOpenAI(
+                        model_name=LLM_MODEL,
+                        temperature=0.2,
+                        api_key=OPENAI_API_KEY,
+                    )
+                    print("OpenAI 모델 초기화 성공")
+                except Exception as e:
+                    print(f"OpenAI 모델 초기화 실패: {str(e)}")
+                    raise
+            else:
+                try:
+                    print(f"Ollama 모델 초기화: {LLM_MODEL}")
+                    self.llm = ChatOllama(
+                        model=LLM_MODEL,
+                        temperature=0.2,
+                        base_url=OLLAMA_HOST,
+                    )
+                    print("Ollama 모델 초기화 성공")
+                except Exception as e:
+                    print(f"Ollama 모델 초기화 실패: {str(e)}")
+                    raise
+
+            # RAG 체인 구성 및 프롬프트 설정
+            print("RAG 체인 설정 시작...")
+            self.setup_chain()
+            print("RAG 체인 설정 완료")
+        except Exception as e:
+            print(f"RAGChain 초기화 중 상세 오류: {str(e)}")
+            import traceback
+            traceback.print_exc()
+            raise
+
+    def setup_chain(self) -> None:
+        """
+        RAG 체인 및 프롬프트 설정
+        """
+        # 프롬프트 템플릿 정의
+        template = """
+        다음 정보를 기반으로 질문에 정확하게 답변해주세요.
+
+        질문: {question}
+
+        참고 정보:
+        {context}
+
+        참고 정보에 답이 없는 경우 "제공된 문서에서 해당 정보를 찾을 수 없습니다."라고 답변하세요.
+        답변은 정확하고 간결하게 제공하되, 참고 정보에서 근거를 찾아 설명해주세요.
+        참고 정보의 출처도 함께 알려주세요.
+        """
+
+        self.prompt = PromptTemplate.from_template(template)
+
+        # RAG 체인 정의
+        self.chain = (
+                {"context": self._retrieve, "question": RunnablePassthrough()}
+                | self.prompt
+                | self.llm
+                | StrOutputParser()
+        )
+
+    def _retrieve(self, query: str) -> str:
+        """
+        쿼리에 대한 관련 문서 검색 및 컨텍스트 구성
+
+        Args:
+            query: 사용자 질문
+
+        Returns:
+            검색 결과를 포함한 컨텍스트 문자열
+        """
+        # 벡터 검색 수행
+        docs = self.vector_store.similarity_search(query, k=TOP_K_RETRIEVAL)
+
+        # 리랭커 적용 (선택적)
+        if self.use_reranker and docs:
+            docs = self.reranker.rerank(query, docs, top_k=TOP_K_RERANK)
+
+        # 검색 결과 컨텍스트 구성
+        context_parts = []
+        for i, doc in enumerate(docs, 1):
+            source = doc.metadata.get("source", "알 수 없는 출처")
+            page = doc.metadata.get("page", "")
+            source_info = f"{source}"
+            if page:
+                source_info += f" (페이지: {page})"
+
+            context_parts.append(f"[참고자료 {i}] - 출처: {source_info}\n{doc.page_content}\n")
+
+        return "\n".join(context_parts)
+
+    def run(self, query: str) -> str:
+        """
+        사용자 쿼리에 대한 RAG 파이프라인 실행
+
+        Args:
+            query: 사용자 질문
+
+        Returns:
+            모델 응답 문자열
+        """
+        return self.chain.invoke(query)

requirements.txt

@@ -0,0 +1,14 @@
+langchain>=0.1.0
+langchain-community>=0.0.10
+langchain-huggingface>=0.0.1
+sentence-transformers>=2.2.2
+faiss-cpu>=1.7.4
+pypdf>=3.15.1
+gradio>=4.0.0
+python-dotenv>=1.0.0
+torch>=2.0.0
+transformers>=4.34.0
+langchain-openai>=0.0.2
+openai>=1.0.0
+docling>=0.1.3
+requests>=2.28.0

reranker.py

@@ -0,0 +1,58 @@
+"""
+원격 코드 실행 옵션이 추가된 리랭커 모듈
+"""
+from typing import List, Dict, Tuple
+import numpy as np
+from sentence_transformers import CrossEncoder
+from langchain.schema import Document
+from config import RERANKER_MODEL
+
+class Reranker:
+    def __init__(self, model_name: str = RERANKER_MODEL):
+        """
+        Cross-Encoder 리랭커 초기화
+
+        Args:
+            model_name: 사용할 Cross-Encoder 모델 이름
+        """
+        print(f"리랭커 모델 로드 중: {model_name}")
+
+        # 원격 코드 실행 허용 옵션 추가
+        self.model = CrossEncoder(
+            model_name,
+            trust_remote_code=True  # 원격 코드 실행 허용 (필수)
+        )
+
+        print(f"리랭커 모델 로드 완료: {model_name}")
+
+    def rerank(self, query: str, documents: List[Document], top_k: int = 3) -> List[Document]:
+        """
+        검색 결과 재정렬
+
+        Args:
+            query: 검색 쿼리
+            documents: 벡터 검색 결과 문서 리스트
+            top_k: 반환할 상위 결과 수
+
+        Returns:
+            재정렬된 상위 문서 리스트
+        """
+        if not documents:
+            return []
+
+        # Cross-Encoder 입력 쌍 생성
+        document_texts = [doc.page_content for doc in documents]
+        query_doc_pairs = [(query, doc) for doc in document_texts]
+
+        # 점수 계산
+        print(f"리랭킹 수행 중: {len(documents)}개 문서")
+        scores = self.model.predict(query_doc_pairs)
+
+        # 점수에 따라 문서 재정렬
+        doc_score_pairs = list(zip(documents, scores))
+        doc_score_pairs.sort(key=lambda x: x[1], reverse=True)
+
+        print(f"리랭킹 완료: 상위 {top_k}개 문서 선택")
+
+        # 상위 k개 결과 반환
+        return [doc for doc, score in doc_score_pairs[:top_k]]

simple_rag_chain.py

@@ -0,0 +1,66 @@
+"""
+간단한 RAG 체인 구현 (디버깅용)
+"""
+import os
+from langchain_openai import ChatOpenAI
+from langchain.prompts import PromptTemplate
+from langchain_core.output_parsers import StrOutputParser
+from langchain_core.runnables import RunnablePassthrough
+
+
+class SimpleRAGChain:
+    def __init__(self, vector_store):
+        """간단한 RAG 체인 초기화"""
+        print("간단한 RAG 체인 초기화 중...")
+        self.vector_store = vector_store
+
+        # OpenAI API 키 확인
+        openai_api_key = os.environ.get("OPENAI_API_KEY", "")
+        print(f"API 키 설정됨: {bool(openai_api_key)}")
+
+        # OpenAI 모델 초기화
+        self.llm = ChatOpenAI(
+            model_name="gpt-3.5-turbo",
+            temperature=0.2,
+            api_key=openai_api_key,
+        )
+
+        # 프롬프트 템플릿
+        template = """
+        다음 정보를 기반으로 질문에 정확하게 답변해주세요.
+
+        질문: {question}
+
+        참고 정보:
+        {context}
+
+        참고 정보에 답이 없는 경우 "제공된 문서에서 해당 정보를 찾을 수 없습니다."라고 답변하세요.
+        """
+
+        self.prompt = PromptTemplate.from_template(template)
+
+        # 체인 구성
+        self.chain = (
+                {"context": self._retrieve, "question": RunnablePassthrough()}
+                | self.prompt
+                | self.llm
+                | StrOutputParser()
+        )
+        print("간단한 RAG 체인 초기화 완료")
+
+    def _retrieve(self, query):
+        """문서 검색"""
+        try:
+            docs = self.vector_store.similarity_search(query, k=3)
+            return "\n\n".join(doc.page_content for doc in docs)
+        except Exception as e:
+            print(f"검색 중 오류: {e}")
+            return "문서 검색 중 오류가 발생했습니다."
+
+    def run(self, query):
+        """쿼리 처리"""
+        try:
+            return self.chain.invoke(query)
+        except Exception as e:
+            print(f"실행 중 오류: {e}")
+            return f"오류 발생: {str(e)}"

vector_store.py

@@ -0,0 +1,235 @@
+"""
+개선된 벡터 스토어 모듈 - Milvus 설정 최적화
+"""
+from typing import List, Dict, Any, Optional
+import uuid
+from langchain.schema import Document
+
+# 벡터 스토어 임포트
+try:
+    # 최신 버전 임포트
+    from langchain_milvus import Milvus
+    from langchain_community.vectorstores import FAISS
+    from langchain_huggingface import HuggingFaceEmbeddings
+    MODERN_IMPORTS = True
+    print("최신 langchain 패키지 임포트 성공")
+except ImportError:
+    # 이전 버전 임포트
+    from langchain_community.vectorstores import Milvus, FAISS
+    from langchain_community.embeddings import HuggingFaceEmbeddings
+    MODERN_IMPORTS = False
+    print("레거시 langchain_community 패키지 사용")
+
+from config import MILVUS_HOST, MILVUS_PORT, MILVUS_COLLECTION, EMBEDDING_MODEL
+
+class VectorStore:
+    def __init__(self, use_milvus: bool = True):
+        """
+        벡터 스토어 초기화
+
+        Args:
+            use_milvus: Milvus 사용 여부 (False이면 FAISS 사용)
+        """
+        self.use_milvus = use_milvus
+
+        # 임베딩 모델 설정
+        print(f"임베딩 모델 로드 중: {EMBEDDING_MODEL}")
+        model_kwargs = {
+            "device": "cpu",
+            "trust_remote_code": True  # 원격 코드 실행 허용 (필수)
+        }
+        encode_kwargs = {"normalize_embeddings": True}
+
+        self.embeddings = HuggingFaceEmbeddings(
+            model_name=EMBEDDING_MODEL,
+            model_kwargs=model_kwargs,
+            encode_kwargs=encode_kwargs
+        )
+        self.vector_store = None
+
+        print(f"임베딩 모델 초기화 완료: {EMBEDDING_MODEL}")
+
+    def init_milvus(self) -> Milvus:
+        """
+        Milvus 벡터 스토어 초기화
+
+        Returns:
+            Milvus 벡터 스토어 인스턴스
+        """
+        connection_args = {
+            "host": MILVUS_HOST,
+            "port": MILVUS_PORT,
+        }
+
+        # 벡터 검색 인덱스 파라미터 (FLAT 인덱스 및 코사인 유사도 메트릭)
+        index_params = {
+            "index_type": "FLAT",  # 정확도 우선 FLAT 인덱스
+            "metric_type": "COSINE",  # 코사인 유사도 (정규화된 벡터에 적합)
+            "params": {}  # FLAT 인덱스에는 추가 파라미터 없음
+        }
+
+        return Milvus(
+            embedding_function=self.embeddings,
+            collection_name=MILVUS_COLLECTION,
+            connection_args=connection_args,
+            index_params=index_params
+        )
+
+    def init_faiss(self) -> FAISS:
+        """
+        FAISS 벡터 스토어 초기화 (로컬 대체용)
+
+        Returns:
+            FAISS 벡터 스토어 인스턴스
+        """
+        return FAISS.from_documents([], self.embeddings)
+
+    def create_or_load(self, documents: Optional[List[Document]] = None) -> Any:
+        """
+        벡터 스토어 생성 또는 로드
+
+        Args:
+            documents: 저장할 문서 리스트 (None이면 빈 스토어 생성)
+
+        Returns:
+            벡터 스토어 인스턴스
+        """
+        if self.use_milvus:
+            if documents:
+                # 문서가 제공된 경우 새 컬렉션 생성
+                try:
+                    # 연결 설정
+                    connection_args = {
+                        "host": MILVUS_HOST,
+                        "port": MILVUS_PORT,
+                    }
+
+                    # 검색 인덱스 설정
+                    index_params = {
+                        "index_type": "FLAT",  # 정확도 우선
+                        "metric_type": "COSINE",  # 코사인 유사도
+                        "params": {}
+                    }
+
+                    print(f"Milvus 컬렉션 생성: {MILVUS_COLLECTION} (기존 컬렉션 삭제)")
+
+                    # 문서로부터 Milvus 컬렉션 생성
+                    self.vector_store = Milvus.from_documents(
+                        documents=documents,
+                        embedding=self.embeddings,
+                        collection_name=MILVUS_COLLECTION,
+                        connection_args=connection_args,
+                        index_params=index_params,
+                        drop_old=True  # 기존 컬렉션 삭제 (재구축)
+                    )
+
+                    print(f"Milvus 컬렉션 생성 완료: {len(documents)}개 문서 인덱싱됨")
+
+                except Exception as e:
+                    print(f"Milvus 컬렉션 생성 실패: {e}")
+                    # 대체 방안으로 FAISS 사용
+                    print("대체 방안으로 FAISS 사용")
+                    self.use_milvus = False
+                    self.vector_store = FAISS.from_documents(documents, self.embeddings)
+            else:
+                # 기존 컬렉션 로드
+                try:
+                    self.vector_store = self.init_milvus()
+                except Exception as e:
+                    print(f"Milvus 컬렉션 로드 실패: {e}")
+                    # 대체 방안으로 FAISS 사용
+                    print("대체 방안으로 FAISS 사용")
+                    self.use_milvus = False
+                    self.vector_store = self.init_faiss()
+        else:
+            # FAISS 사용
+            if documents:
+                print(f"FAISS 인덱스 생성: {len(documents)}개 문서")
+                self.vector_store = FAISS.from_documents(documents, self.embeddings)
+                print("FAISS 인덱스 생성 완료")
+            else:
+                self.vector_store = self.init_faiss()
+                print("빈 FAISS 인덱스 초기화 완료")
+
+        return self.vector_store
+
+    def add_documents(self, documents: List[Document]) -> None:
+        """
+        벡터 스토어에 문서 추가
+
+        Args:
+            documents: 추가할 문서 리스트
+        """
+        if self.vector_store is None:
+            self.create_or_load(documents)
+        else:
+            if self.use_milvus:
+                self.vector_store.add_documents(documents)
+            else:
+                self.vector_store.add_documents(documents)
+
+    def similarity_search(self, query: str, k: int = 5) -> List[Document]:
+        """
+        벡터 유사도 검색 수행
+
+        Args:
+            query: 검색 쿼리
+            k: 반환할 결과 수
+
+        Returns:
+            유사도가 높은 문서 리스트
+        """
+        if self.vector_store is None:
+            raise ValueError("벡터 스토어가 초기화되지 않았습니다.")
+
+        print(f"검색 쿼리: '{query}', 상위 {k}개 결과 요청")
+        results = self.vector_store.similarity_search(query, k=k)
+        print(f"검색 완료: {len(results)}개 결과 찾음")
+
+        return results
+
+    def save_local(self, path: str = "faiss_index") -> None:
+        """
+        FAISS 인덱스 로컬 저장 (Milvus 사용 안 할 경우)
+
+        Args:
+            path: 저장 경로
+        """
+        if not self.use_milvus and self.vector_store is not None:
+            self.vector_store.save_local(path)
+            print(f"FAISS 인덱스 로컬 저장 완료: {path}")
+
+    """
+    FAISS 역직렬화 허용 설정이, 포함된 벡터 스토어 코드
+    """
+
+    # vector_store.py 파일에서 load_local 메서드 수정
+
+    def load_local(self, path: str = "faiss_index") -> None:
+        """
+        FAISS 인덱스 로컬 로드 (Milvus 사용 안 할 경우)
+
+        Args:
+            path: 로드할 인덱스 경로
+        """
+        if not self.use_milvus:
+            try:
+                print(f"FAISS 인덱스 로드 중: {path}")
+
+                # 역직렬화 허용 옵션 추가 (보안 경고 확인 필요)
+                self.vector_store = FAISS.load_local(
+                    path,
+                    self.embeddings,
+                    allow_dangerous_deserialization=True  # 역직렬화 허용
+                )
+                print(f"FAISS 인덱스 로드 완료: {path}")
+            except Exception as e:
+                print(f"FAISS 인덱스 로드 실패: {e}")
+
+                # 오류 세부 정보 출력
+                import traceback
+                traceback.print_exc()
+
+                # 새 인덱스 초기화
+                self.vector_store = self.init_faiss()
+                print("새 FAISS 인덱스 초기화됨")

voice_rag_app.py

@@ -0,0 +1,670 @@
+"""
+음성인식 기능이 추가된 RAG 챗봇 앱
+"""
+import os
+import time
+import tempfile
+from typing import List, Dict, Tuple, Any, Optional
+import hashlib
+import pickle
+import json
+
+# 기존 임포트
+from config import PDF_DIRECTORY, CHUNK_SIZE, CHUNK_OVERLAP, LLM_MODEL
+from optimized_document_processor import OptimizedDocumentProcessor
+from vector_store import VectorStore
+from langchain.schema import Document
+
+# 클로바 STT 모듈 임포트
+from clova_stt import ClovaSTT
+
+# 안전한 임포트
+try:
+    from rag_chain import RAGChain
+
+    RAG_CHAIN_AVAILABLE = True
+except ImportError:
+    print("RAG 체인 모듈을 로드할 수 없습니다.")
+    RAG_CHAIN_AVAILABLE = False
+
+
+class VoiceRAGChatApp:
+    """
+    음성인식 기능이 추가된 RAG 챗봇 애플리케이션
+    """
+
+    def __init__(self):
+        """
+        음성인식 RAG 챗봇 애플리케이션 초기화
+        """
+        # 데이터 디렉토리 정의
+        self.pdf_directory = PDF_DIRECTORY
+        self.cache_directory = "cached_data"
+        self.index_file = os.path.join(self.cache_directory, "file_index.json")
+        self.chunks_dir = os.path.join(self.cache_directory, "chunks")
+        self.vector_index_dir = os.path.join(self.cache_directory, "vector_index")
+
+        # 디렉토리 생성
+        os.makedirs(self.pdf_directory, exist_ok=True)
+        os.makedirs(self.cache_directory, exist_ok=True)
+        os.makedirs(self.chunks_dir, exist_ok=True)
+        os.makedirs(self.vector_index_dir, exist_ok=True)
+
+        print(f"PDF 문서 디렉토리: '{self.pdf_directory}'")
+        print(f"캐시 디렉토리: '{self.cache_directory}'")
+
+        # 컴포넌트 초기화
+        self.document_processor = OptimizedDocumentProcessor(
+            chunk_size=CHUNK_SIZE,
+            chunk_overlap=CHUNK_OVERLAP
+        )
+
+        # 벡터 저장소 초기화
+        self.vector_store = VectorStore(use_milvus=False)
+
+        # 문서 인덱스 로드
+        self.file_index = self._load_file_index()
+
+        # 기본 변수 초기화
+        self.documents = []
+        self.processed_files = []
+        self.is_initialized = False
+
+        # 클로바 STT 클라이언트 초기화
+        self.stt_client = ClovaSTT()
+        print("음성인식(STT) 기능이 초기화되었습니다.")
+
+        # 시작 시 자동으로 문서 로드 및 처리
+        print("문서 자동 로드 및 처리 시작...")
+        self.auto_process_documents()
+
+    def _load_file_index(self) -> Dict[str, Dict[str, Any]]:
+        """
+        파일 인덱스 로드
+
+        Returns:
+            파일 경로 -> 메타데이터 매핑
+        """
+        if os.path.exists(self.index_file):
+            try:
+                with open(self.index_file, 'r', encoding='utf-8') as f:
+                    return json.load(f)
+            except Exception as e:
+                print(f"인덱스 파일 로드 실패: {e}")
+                return {}
+        return {}
+
+    def _save_file_index(self) -> None:
+        """
+        파일 인덱스 저장
+        """
+        with open(self.index_file, 'w', encoding='utf-8') as f:
+            json.dump(self.file_index, f, ensure_ascii=False, indent=2)
+
+    def _calculate_file_hash(self, file_path: str) -> str:
+        """
+        파일 해시 계산
+
+        Args:
+            file_path: 파일 경로
+
+        Returns:
+            MD5 해시값
+        """
+        hasher = hashlib.md5()
+        with open(file_path, 'rb') as f:
+            buf = f.read(65536)
+            while len(buf) > 0:
+                hasher.update(buf)
+                buf = f.read(65536)
+        return hasher.hexdigest()
+
+    def _is_file_processed(self, file_path: str) -> bool:
+        """
+        파일이 이미 처리되었고 변경되지 않았는지 확인
+
+        Args:
+            file_path: 파일 경로
+
+        Returns:
+            처리 여부
+        """
+        if file_path not in self.file_index:
+            return False
+
+        # 현재 해시값 계산
+        current_hash = self._calculate_file_hash(file_path)
+
+        # 저장된 해시값과 비교
+        if self.file_index[file_path]['hash'] != current_hash:
+            print(f"파일 변경 감지: {file_path}")
+            return False
+
+        # 청크 파일 존재 확인
+        chunks_path = self.file_index[file_path]['chunks_path']
+        if not os.path.exists(chunks_path):
+            return False
+
+        return True
+
+    def _get_chunks_path(self, file_hash: str) -> str:
+        """
+        청크 파일 경로 생성
+
+        Args:
+            file_hash: 파일 해시값
+
+        Returns:
+            청크 파일 경로
+        """
+        return os.path.join(self.chunks_dir, f"{file_hash}.pkl")
+
+    def _save_chunks(self, file_path: str, chunks: List[Document]) -> None:
+        """
+        청크 데이터 저장
+
+        Args:
+            file_path: 원본 파일 경로
+            chunks: 문서 청크 리스트
+        """
+        # 해시 계산
+        file_hash = self._calculate_file_hash(file_path)
+
+        # 청크 파일 경로
+        chunks_path = self._get_chunks_path(file_hash)
+
+        # 청크 데이터 저장
+        with open(chunks_path, 'wb') as f:
+            pickle.dump(chunks, f)
+
+        # 인덱스 업데이트
+        self.file_index[file_path] = {
+            'hash': file_hash,
+            'chunks_path': chunks_path,
+            'last_processed': time.time(),
+            'chunks_count': len(chunks)
+        }
+
+        # 인덱스 저장
+        self._save_file_index()
+
+        print(f"청크 저장 완료: {file_path} ({len(chunks)}개 청크)")
+
+    def _load_chunks(self, file_path: str) -> List[Document]:
+        """
+        저장된 청크 데이터 로드
+
+        Args:
+            file_path: 파일 경로
+
+        Returns:
+            문서 청크 리스트
+        """
+        chunks_path = self.file_index[file_path]['chunks_path']
+        with open(chunks_path, 'rb') as f:
+            chunks = pickle.load(f)
+
+        print(f"청크 로드 완료: {file_path} ({len(chunks)}개 청크)")
+        return chunks
+
+    def _process_pdf_file(self, file_path: str) -> List[Document]:
+        """
+        PDF 파일 처리 - docling 실패 시 PyPDFLoader 사용
+
+        Args:
+            file_path: 처리할 PDF 파일 경로
+
+        Returns:
+            처리된 문서 청크 리스트
+        """
+        try:
+            print(f"docling으로 처리 시도: {file_path}")
+
+            # docling 사용 시도
+            try:
+                # 10초 타임아웃 설정 (옵션)
+                import signal
+
+                def timeout_handler(signum, frame):
+                    raise TimeoutError("docling 처리 시간 초과")
+
+                # 리눅스/맥에서만 작동 (윈도우에서는 무시됨)
+                try:
+                    signal.signal(signal.SIGALRM, timeout_handler)
+                    signal.alarm(60)  # 60초 타임아웃
+                except:
+                    pass
+
+                # docling으로 처리 시도
+                chunks = self.document_processor.process_pdf(file_path, use_docling=True)
+
+                # 타임아웃 취소
+                try:
+                    signal.alarm(0)
+                except:
+                    pass
+
+                return chunks
+
+            except Exception as e:
+                # docling 오류 확인
+                error_str = str(e)
+                if "Invalid code point" in error_str or "RuntimeError" in error_str:
+                    print(f"docling 처리 오류 (코드 포인트 문제): {error_str}")
+                    print("PyPDFLoader로 대체합니다.")
+                else:
+                    print(f"docling 처리 오류: {error_str}")
+                    print("PyPDFLoader로 대체합니다.")
+
+                # PyPDFLoader로 대체
+                try:
+                    return self.document_processor.process_pdf(file_path, use_docling=False)
+                except Exception as inner_e:
+                    print(f"PyPDFLoader 처리 오류: {inner_e}")
+                    raise  # 두 방법 모두 실패하면 예외 발생
+
+        except Exception as e:
+            print(f"PDF 처리 중 심각한 오류: {e}")
+            # 빈 청크라도 반환하여 전체 처리가 중단되지 않도록 함
+            return []
+
+    def auto_process_documents(self) -> str:
+        """
+        documents 폴더의 PDF 파일 자동 처리
+
+        Returns:
+            처리 결과 메시지
+        """
+        try:
+            start_time = time.time()
+
+            # PDF 파일 목록 수집
+            pdf_files = []
+            for filename in os.listdir(self.pdf_directory):
+                if filename.lower().endswith('.pdf'):
+                    pdf_files.append(os.path.join(self.pdf_directory, filename))
+
+            if not pdf_files:
+                return f"'{self.pdf_directory}' 폴더에 PDF 파일이 없습니다."
+
+            print(f"발견된 PDF 파일: {len(pdf_files)}개")
+
+            # 폴더 내 PDF 파일 처리
+            new_files = []
+            updated_files = []
+            cached_files = []
+            failed_files = []
+            all_chunks = []
+
+            for file_path in pdf_files:
+                if self._is_file_processed(file_path):
+                    # 캐시에서 청크 로드
+                    chunks = self._load_chunks(file_path)
+                    all_chunks.extend(chunks)
+                    cached_files.append(file_path)
+                    self.processed_files.append(os.path.basename(file_path))
+                else:
+                    # 새 파일 또는 변경된 파일 처리
+                    print(f"처리 중: {file_path}")
+
+                    try:
+                        # 개선된 PDF 처리 메서드 사용
+                        chunks = self._process_pdf_file(file_path)
+
+                        if chunks:  # 청크가 있는 경우에만 저장
+                            # 청크 저장
+                            self._save_chunks(file_path, chunks)
+
+                            all_chunks.extend(chunks)
+                            if file_path in self.file_index:
+                                updated_files.append(file_path)
+                            else:
+                                new_files.append(file_path)
+
+                            self.processed_files.append(os.path.basename(file_path))
+                        else:
+                            print(f"'{file_path}' 처리 실패: 추출된 청크 없음")
+                            failed_files.append(file_path)
+                    except Exception as e:
+                        print(f"'{file_path}' 처리 중 오류: {e}")
+                        failed_files.append(file_path)
+
+            # 모든 청크 저장
+            self.documents = all_chunks
+
+            processing_time = time.time() - start_time
+            print(f"문서 처리 완료: {len(all_chunks)}개 청크, {processing_time:.2f}초")
+
+            # 벡터 인덱스 저장 경로 확인
+            if os.path.exists(self.vector_index_dir) and any(os.listdir(self.vector_index_dir)):
+                # 기존 벡터 인덱스 로드
+                try:
+                    print("저장된 벡터 인덱스 로드 중...")
+                    vector_store_loaded = self.vector_store.load_local(self.vector_index_dir)
+
+                    # 인덱스 로드 성공 확인
+                    if self.vector_store.vector_store is not None:
+                        # 새 문서나 변경된 문서가 있으면 인덱스 업데이트
+                        if new_files or updated_files:
+                            print("벡터 인덱스 업데이트 중...")
+                            self.vector_store.add_documents(self.documents)
+
+                        print("벡터 인덱스 로드 완료")
+                    else:
+                        print("벡터 인덱스를 로드했으나 유효하지 않음, 새로 생성합니다.")
+                        self.vector_store.create_or_load(self.documents)
+
+                except Exception as e:
+                    print(f"벡터 인덱스 로드 실패, 새로 생성합니다: {e}")
+                    # 오류 상세 정보 출력
+                    import traceback
+                    traceback.print_exc()
+
+                    # 새 벡터 인덱스 생성
+                    self.vector_store.create_or_load(self.documents)
+            else:
+                # 새 벡터 인덱스 생성
+                print("새 벡터 인덱스 생성 중...")
+                self.vector_store.create_or_load(self.documents)
+
+            # 벡터 인덱스 저장
+            if self.vector_store and self.vector_store.vector_store is not None:
+                try:
+                    print(f"벡터 인덱스 저장 중: {self.vector_index_dir}")
+                    save_result = self.vector_store.save_local(self.vector_index_dir)
+                    print(f"벡터 인덱스 저장 완료: {self.vector_index_dir}")
+                except Exception as e:
+                    print(f"벡터 인덱스 저장 실패: {e}")
+                    # 오류 상세 정보 출력
+                    import traceback
+                    traceback.print_exc()
+            else:
+                print("벡터 인덱스가 초기화되지 않아 저장하지 않습니다.")
+
+            # RAG 체인 초기화
+            if RAG_CHAIN_AVAILABLE:
+                self.rag_chain = RAGChain(self.vector_store)
+                self.is_initialized = True
+
+                total_time = time.time() - start_time
+
+                status_message = (
+                    f"문서 처리 완료!\n"
+                    f"- 처리된 파일: {len(self.processed_files)}개\n"
+                    f"- 캐시된 파일: {len(cached_files)}개\n"
+                    f"- 새 파일: {len(new_files)}개\n"
+                    f"- 업데이트된 파일: {len(updated_files)}개\n"
+                    f"- 실패한 파일: {len(failed_files)}개\n"
+                    f"- 총 청크 수: {len(self.documents)}개\n"
+                    f"- 처리 시간: {total_time:.2f}초\n"
+                    f"이제 질문할 준비가 되었습니다!"
+                )
+
+                print(status_message)
+                return status_message
+            else:
+                return "RAG 체인을 초기화할 수 없습니다. 필요한 라이브러리가 설치되어 있는지 확인하세요."
+
+        except Exception as e:
+            error_message = f"문서 처리 중 오류 발생: {str(e)}"
+            print(error_message)
+            import traceback
+            traceback.print_exc()
+            return error_message
+
+    def reset_cache(self) -> str:
+        """
+        캐시 초기화
+
+        Returns:
+            결과 메시지
+        """
+        try:
+            # 청��� 파일 삭제
+            for filename in os.listdir(self.chunks_dir):
+                file_path = os.path.join(self.chunks_dir, filename)
+                if os.path.isfile(file_path):
+                    os.remove(file_path)
+
+            # 인덱스 초기화
+            self.file_index = {}
+            self._save_file_index()
+
+            # 벡터 인덱스 삭제
+            for filename in os.listdir(self.vector_index_dir):
+                file_path = os.path.join(self.vector_index_dir, filename)
+                if os.path.isfile(file_path):
+                    os.remove(file_path)
+
+            self.documents = []
+            self.processed_files = []
+            self.is_initialized = False
+
+            return "캐시가 초기화되었습니다. 다음 실행 시 모든 문서가 다시 처리됩니다."
+        except Exception as e:
+            return f"캐시 초기화 중 오류 발생: {str(e)}"
+
+    def process_query(self, query: str, chat_history: List[Tuple[str, str]]) -> Tuple[str, List[Tuple[str, str]]]:
+        """
+        사용자 쿼리 처리
+
+        Args:
+            query: 사용자 질문
+            chat_history: 대화 기록
+
+        Returns:
+            응답 및 업데이트된 대화 기록
+        """
+        if not query:  # 비어있는 쿼리 처리
+            return "", chat_history
+
+        if not self.is_initialized:
+            response = "문서 로드가 초기화되지 않았습니다. 자동 로드를 시도합니다."
+            chat_history.append((query, response))
+
+            # 자동 로드 시도
+            try:
+                self.auto_process_documents()
+                if not self.is_initialized:
+                    response = "문서를 로드할 수 없습니다. 'documents' 폴더에 PDF 파일이 있는지 확인하세요."
+                    chat_history.append((query, response))
+                    return "", chat_history
+            except Exception as e:
+                response = f"문서 로드 중 오류 발생: {str(e)}"
+                chat_history.append((query, response))
+                return "", chat_history
+
+        try:
+            # RAG 체인 실행 및 응답 생성
+            start_time = time.time()
+            response = self.rag_chain.run(query)
+            end_time = time.time()
+
+            query_time = end_time - start_time
+            print(f"쿼리 처리 시간: {query_time:.2f}초")
+
+            chat_history.append((query, response))
+            return "", chat_history
+        except Exception as e:
+            error_msg = f"오류 발생: {str(e)}"
+            chat_history.append((query, error_msg))
+            return "", chat_history
+
+    def process_voice_query(self, audio, chat_history: List[Tuple[str, str]]) -> Tuple[str, List[Tuple[str, str]]]:
+        """
+        음성 쿼리 처리
+
+        Args:
+            audio: 녹음된 오디오 데이터
+            chat_history: 대화 기록
+
+        Returns:
+            응답 및 업데이트된 대화 기록
+        """
+        if audio is None:
+            return "", chat_history
+
+        try:
+            # 임시 파일에 오디오 저장
+            with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file:
+                temp_path = temp_file.name
+                temp_file.write(audio)
+
+            print(f"[STT] 임시 오디오 파일 생성: {temp_path}")
+
+            # STT 실행
+            result = self.stt_client.recognize_file(temp_path)
+
+            # 임시 파일 삭제
+            try:
+                os.unlink(temp_path)
+                print("[STT] 임시 오디오 파일 삭제됨")
+            except Exception as e:
+                print(f"[STT] 임시 파일 삭제 실패: {e}")
+
+            # STT 결과 처리
+            if "error" in result:
+                error_msg = f"음성인식 오류: {result.get('error')}"
+                print(f"[STT] {error_msg}")
+                chat_history.append(("음성 메시지", error_msg))
+                return "", chat_history
+
+            # 인식된 텍스트 추출
+            recognized_text = result.get("text", "")
+            if not recognized_text:
+                error_msg = "음성을 인식할 수 없습니다. 다시 시도해주세요."
+                print("[STT] 인식된 텍스트 없음")
+                chat_history.append(("음성 메시지", error_msg))
+                return "", chat_history
+
+            print(f"[STT] 인식된 텍스트: {recognized_text}")
+
+            # 인식된 텍스트로 쿼리 처리 (음성 메시지 접두어 추가)
+            return self.process_query(f"🎤 {recognized_text}", chat_history)
+
+        except Exception as e:
+            error_msg = f"음성 처리 중 오류 발생: {str(e)}"
+            print(f"[STT] {error_msg}")
+            chat_history.append(("음성 메시지", error_msg))
+            return "", chat_history
+
+    def launch_app(self) -> None:
+        """
+        음성인식 기능이 추가된 Gradio 앱 실행
+        """
+        import gradio as gr
+
+        with gr.Blocks(title="음성인식 기능이 추가된 PDF 문서 기반 RAG 챗봇") as app:
+            gr.Markdown("# 음성인식 기능이 추가된 PDF 문서 기반 RAG 챗봇")
+            gr.Markdown(f"* 사용 중인 LLM 모델: **{LLM_MODEL}**")
+            gr.Markdown(f"* PDF 문서 폴더: **{self.pdf_directory}**")
+            gr.Markdown("* 네이버 클로바 음성인식 API 통합")
+
+            with gr.Row():
+                with gr.Column(scale=1):
+                    # 문서 상태 섹션
+                    status_box = gr.Textbox(
+                        label="문서 처리 상태",
+                        value=f"처리된 문서 ({len(self.processed_files)}개): {', '.join(self.processed_files)}",
+                        lines=5,
+                        interactive=False
+                    )
+
+                    # 캐시 관리 버튼
+                    refresh_button = gr.Button("문서 새로 읽기", variant="primary")
+                    reset_button = gr.Button("캐시 초기화", variant="stop")
+
+                    # 처리된 파일 정보
+                    with gr.Accordion("캐시 세부 정보", open=False):
+                        file_info = ""
+                        for file_path, info in self.file_index.items():
+                            file_info += f"- {os.path.basename(file_path)}: {info['chunks_count']}개 청크\n"
+
+                        cache_info = gr.Textbox(
+                            label="캐시된 파일 정보",
+                            value=file_info or "캐시된 파일이 없습니다.",
+                            lines=5,
+                            interactive=False
+                        )
+
+                with gr.Column(scale=2):
+                    # 채팅 인터페이스
+                    chatbot = gr.Chatbot(
+                        label="대화 내용",
+                        bubble_full_width=False,
+                        height=500,
+                        show_copy_button=True
+                    )
+
+                    with gr.Tabs() as input_tabs:
+                        # 텍스트 입력 탭
+                        with gr.Tab("텍스트 입력"):
+                            # 텍스트 입력과 전송 버튼을 수평으로 배치
+                            with gr.Row():
+                                query_box = gr.Textbox(
+                                    label="질문",
+                                    placeholder="처리된 문서 내용에 대해 질문하세요...",
+                                    lines=2,
+                                    scale=4
+                                )
+                                submit_btn = gr.Button("전송", variant="primary", scale=1)
+
+                        # 음성 입력 탭
+                        with gr.Tab("음성 입력"):
+                            audio_input = gr.Audio(
+                                label="마이크 입력",
+                                sources=["microphone"],
+                                type="bytes",
+                                format="wav"
+                            )
+                            voice_submit_btn = gr.Button("음성 질문 전송", variant="primary")
+
+                    clear_chat_button = gr.Button("대화 초기화")
+
+            # 이벤트 핸들러 설정
+            refresh_button.click(
+                fn=self.auto_process_documents,
+                inputs=[],
+                outputs=[status_box]
+            )
+
+            reset_button.click(
+                fn=lambda: (self.reset_cache(), self.auto_process_documents()),
+                inputs=[],
+                outputs=[status_box]
+            )
+
+            # 텍스트 전송 버튼 클릭 이벤트
+            submit_btn.click(
+                fn=self.process_query,
+                inputs=[query_box, chatbot],
+                outputs=[query_box, chatbot]
+            )
+
+            # 엔터키 입력 이벤트
+            query_box.submit(
+                fn=self.process_query,
+                inputs=[query_box, chatbot],
+                outputs=[query_box, chatbot]
+            )
+
+            # 음성 전송 버튼 클릭 이벤트
+            voice_submit_btn.click(
+                fn=self.process_voice_query,
+                inputs=[audio_input, chatbot],
+                outputs=[audio_input, chatbot]
+            )
+
+            # 대화 초기화 버튼
+            clear_chat_button.click(
+                fn=lambda: [],
+                outputs=[chatbot]
+            )
+
+        # 앱 실행
+        app.launch(share=False)
+
+
+if __name__ == "__main__":
+    app = VoiceRAGChatApp()
+    app.launch_app()