Update app.py

app.py

@@ -1,64 +1,504 @@
+import os
 import gradio as gr
-from huggingface_hub import InferenceClient
-
-"""
-For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference
-"""
-client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
-
-
-def respond(
-    message,
-    history: list[tuple[str, str]],
-    system_message,
-    max_tokens,
-    temperature,
-    top_p,
-):
-    messages = [{"role": "system", "content": system_message}]
-
-    for val in history:
-        if val[0]:
-            messages.append({"role": "user", "content": val[0]})
-        if val[1]:
-            messages.append({"role": "assistant", "content": val[1]})
-
-    messages.append({"role": "user", "content": message})
-
-    response = ""
-
-    for message in client.chat_completion(
-        messages,
-        max_tokens=max_tokens,
-        stream=True,
-        temperature=temperature,
-        top_p=top_p,
-    ):
-        token = message.choices[0].delta.content
-
-        response += token
-        yield response
-
-
-"""
-For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface
-"""
-demo = gr.ChatInterface(
-    respond,
-    additional_inputs=[
-        gr.Textbox(value="You are a friendly Chatbot.", label="System message"),
-        gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens"),
-        gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"),
-        gr.Slider(
-            minimum=0.1,
-            maximum=1.0,
-            value=0.95,
-            step=0.05,
-            label="Top-p (nucleus sampling)",
-        ),
-    ],
-)
+import pandas as pd
+import numpy as np
+from pathlib import Path
+import time
+import hashlib
+from datetime import datetime
+import torch
+from PIL import Image
 
+# 必要なライブラリをインポート
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from langchain.embeddings import HuggingFaceEmbeddings
+from langchain.vectorstores import Chroma
+from langchain.chains import RetrievalQA
+from langchain.prompts import PromptTemplate
+from langchain.llms import HuggingFacePipeline
+from langchain.schema import Document
+from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
+from transformers import T5ForConditionalGeneration, T5Tokenizer
 
+# 条件付きインポート（ローカル環境とHugging Face Spacesの両方に対応）
+try:
+    import fitz  # PyMuPDF
+    PYMUPDF_AVAILABLE = True
+except ImportError:
+    PYMUPDF_AVAILABLE = False
+    print("PyMuPDFが利用できません。PDFファイルはテキスト抽出のみで処理されます。")
+
+try:
+    import easyocr
+    EASYOCR_AVAILABLE = True
+except ImportError:
+    EASYOCR_AVAILABLE = False
+    print("EasyOCRが利用できません。OCR機能は無効化されます。")
+
+try:
+    import cv2
+    CV2_AVAILABLE = True
+except ImportError:
+    CV2_AVAILABLE = False
+    print("OpenCVが利用できません。画像処理機能は制限されます。")
+
+class ManualChatbot:
+    def __init__(self, docs_dir="./manuals"):
+        """手順書チャットボットの初期化"""
+        self.docs_dir = docs_dir
+        self.vectorstore = None  # ベクトルデータベースの初期化
+        self.file_hashes = {}  # ファイルのハッシュ値を保持する辞書
+        self.last_update_check = None  # 最後に更新をチェックした時間
+        self.processing_status = "未初期化"
+
+        # ディレクトリが存在しなければ作成
+        os.makedirs(docs_dir, exist_ok=True)
+        os.makedirs("./chroma_db", exist_ok=True)
+        
+        # ファイルハッシュの記録ファイルパス
+        self.hash_file_path = os.path.join(os.path.dirname(docs_dir), "file_hashes.json")
+        
+        # OCRの初期化（可能な場合）
+        if EASYOCR_AVAILABLE:
+            self.reader = easyocr.Reader(['ja', 'en'])  # 日本語と英語に対応
+            print("EasyOCRを初期化しました")
+        else:
+            self.reader = None
+        
+        # 要約用の T5 モデル準備（モデルサイズを小さくしてHF Spacesでの動作に最適化）
+        self.summarizer_model = None
+        self.summarizer_tokenizer = None
+        
+        # ハッシュ読み込み
+        self._load_file_hashes()
+
+    def _load_file_hashes(self):
+        """保存されたファイルハッシュを読み込む"""
+        if os.path.exists(self.hash_file_path):
+            try:
+                import json
+                with open(self.hash_file_path, 'r') as f:
+                    self.file_hashes = json.load(f)
+                print(f"{len(self.file_hashes)}件のファイルハッシュを読み込みました")
+            except Exception as e:
+                print(f"ファイルハッシュの読み込みに失敗しました: {str(e)}")
+                self.file_hashes = {}
+        else:
+            self.file_hashes = {}
+
+    def _save_file_hashes(self):
+        """ファイルハッシュを保存する"""
+        try:
+            import json
+            with open(self.hash_file_path, 'w') as f:
+                json.dump(self.file_hashes, f)
+            print(f"{len(self.file_hashes)}件のファイルハッシュを保存しました")
+        except Exception as e:
+            print(f"ファイルハッシュの保存に失敗しました: {str(e)}")
+
+    def _get_file_hash(self, file_path):
+        """ファイルのMD5ハッシュを計算する"""
+        hash_md5 = hashlib.md5()
+        with open(file_path, "rb") as f:
+            for chunk in iter(lambda: f.read(4096), b""):
+                hash_md5.update(chunk)
+        return hash_md5.hexdigest()
+
+    def process_uploaded_files(self, files):
+        """
+        Gradioからアップロードされたファイルを処理する
+        :param files: アップロードされたファイルのリスト
+        :return: 処理状況を示すメッセージ
+        """
+        if not files:
+            return "ファイルがアップロードされていません"
+        
+        self.processing_status = "処理中..."
+        
+        # 新しく追加されたファイルを一時的に保存し処理する
+        file_paths = []
+        for file in files:
+            if file is None:
+                continue
+                
+            # ファイル拡張子を確認
+            filename = file.name
+            file_ext = os.path.splitext(filename)[1].lower()
+            
+            if file_ext not in ['.pdf', '.xlsx', '.xls', '.png', '.jpg', '.jpeg']:
+                continue
+                
+            # ファイルを保存する
+            save_path = os.path.join(self.docs_dir, os.path.basename(filename))
+            with open(save_path, 'wb') as f:
+                f.write(file.read())
+            
+            file_paths.append(save_path)
+        
+        if not file_paths:
+            self.processing_status = "サポートされているファイルがありませんでした"
+            return "サポートされているファイルがありませんでした（.pdf, .xlsx, .xls, .png, .jpg, .jpeg）"
+        
+        # ファイルを処理して知識ベースを更新
+        self.update_knowledge_base(file_paths)
+        
+        self.processing_status = "準備完了"
+        return f"{len(file_paths)}個のファイルが処理され、知識ベースに追加されました"
+
+    def update_knowledge_base(self, file_paths):
+        """
+        指定したファイルから新しいデータを読み込み、インデックスを更新する
+        :param file_paths: 更新したファイルのパス一覧（リスト）
+        """
+        print(f"{len(file_paths)}件のファイルを処理します...")
+        
+        new_documents = []
+        for file_path in file_paths:
+            if file_path.lower().endswith(".pdf"):
+                new_documents.extend(self._process_pdf(file_path))
+            elif file_path.lower().endswith((".xlsx", ".xls")):
+                new_documents.extend(self._process_excel(file_path))
+            elif file_path.lower().endswith((".png", ".jpg", ".jpeg")):
+                new_documents.extend(self._process_image(file_path))
+        
+        if not new_documents:
+            print("処理対象のドキュメントがありませんでした")
+            return
+        
+        print(f"{len(new_documents)}件のドキュメントを処理しました")
+        
+        # テキスト分割
+        text_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=1000,
+            chunk_overlap=200,
+            separators=["\n\n", "\n", "。", "、", " ", ""]
+        )
+        
+        chunks = text_splitter.split_documents(new_documents)
+        print(f"{len(chunks)}個のテキストチャンクに分割しました")
+        
+        # 埋め込みモデルの初期化
+        embeddings = HuggingFaceEmbeddings(
+            model_name="intfloat/multilingual-e5-base",  # 軽量化のためbaseモデルを使用
+            model_kwargs={'device': 'cpu'}  # Spacesでは常にCPUを使用
+        )
+        
+        # 既存のベクトルストアが存在する場合は追加、なければ新規作成
+        if self.vectorstore is None:
+            self.vectorstore = Chroma.from_documents(
+                documents=chunks,
+                embedding=embeddings,
+                persist_directory="./chroma_db"
+            )
+        else:
+            # 既存のベクトルストアに新しいドキュメントを追加
+            self.vectorstore.add_documents(chunks)
+        
+        # ベクトルストアを保存
+        self.vectorstore.persist()
+        
+        # もしQAチェーンがなければ初期化
+        if not hasattr(self, 'qa_chain') or self.qa_chain is None:
+            self._initialize_qa_chain()
+        else:
+            # QAチェーンを更新���れた検索エンジンで更新
+            self.qa_chain.retriever = self.vectorstore.as_retriever(search_kwargs={"k": 3})
+        
+        print("知識ベースを更新しました！")
+
+    def _process_pdf(self, file_path):
+        """PDFファイルを処理してドキュメントを返す"""
+        try:
+            # PyMuPDFが利用可能な場合
+            if PYMUPDF_AVAILABLE:
+                doc = fitz.open(file_path)
+                all_text = ""
+                
+                for page_num, page in enumerate(doc):
+                    text = page.get_text()
+                    all_text += f"--- Page {page_num + 1} ---\n{text}\n\n"
+                
+                # OCRが必要か確認（テキストが少ない場合）
+                if len(all_text.strip()) < 100 and EASYOCR_AVAILABLE and self.reader:
+                    all_text = self.extract_text_from_pdf_with_ocr(file_path)
+                
+                return [Document(page_content=all_text, metadata={"source": file_path})]
+            else:
+                # 簡易処理（PyMuPDFが利用できない場合）
+                # 注意: この場合はPDFの内容を適切に抽出できない可能性がある
+                return [Document(page_content=f"PDF file: {os.path.basename(file_path)}", 
+                                metadata={"source": file_path})]
+        except Exception as e:
+            print(f"PDFファイルの処理中にエラーが発生しました ({file_path}): {str(e)}")
+            return []
+
+    def _process_excel(self, file_path):
+        """Excelファイルを処理してドキュメントを返す"""
+        try:
+            # Pandas でExcelを読み込む
+            dfs = pd.read_excel(file_path, sheet_name=None)
+            documents = []
+            
+            for sheet_name, df in dfs.items():
+                # NaN値を空文字列に変換
+                df = df.fillna('')
+                
+                # 各行をテキストに変換
+                for idx, row in df.iterrows():
+                    content = f"Sheet: {sheet_name}, Row: {idx}\n"
+                    for col in df.columns:
+                        content += f"{col}: {row[col]}\n"
+                    
+                    doc = Document(
+                        page_content=content,
+                        metadata={"source": file_path, "sheet": sheet_name, "row": idx}
+                    )
+                    documents.append(doc)
+            
+            return documents
+        except Exception as e:
+            print(f"Excelファイルの処理中にエラーが発生しました ({file_path}): {str(e)}")
+            return []
+
+    def _process_image(self, file_path):
+        """画像ファイルを処理してドキュメントを返す"""
+        try:
+            if not EASYOCR_AVAILABLE or not self.reader:
+                return [Document(
+                    page_content=f"画像ファイル: {os.path.basename(file_path)} (OCR未対応)",
+                    metadata={"source": file_path}
+                )]
+            
+            img = Image.open(file_path)
+            
+            # EasyOCRで画像からテキストを抽出
+            result = self.reader.readtext(np.array(img))
+            
+            # 抽出されたテキストを結合
+            text = "\n".join([detection[1] for detection in result])
+            
+            if not text.strip():
+                text = f"画像ファイル: {os.path.basename(file_path)} (テキスト検出なし)"
+            
+            # ドキュメントとしてリストに追加
+            return [Document(page_content=text, metadata={"source": file_path})]
+        except Exception as e:
+            print(f"画像ファイルの処理中にエラーが発生しました ({file_path}): {str(e)}")
+            return []
+
+    def _initialize_qa_chain(self):
+        """QAチェーンを初期化する"""
+        try:
+            # LLMの初期化（小さいモデルを使用）
+            model_name = "cyberagent/open-calm-small"  # 日本語対応の小さいモデル
+            
+            tokenizer = AutoTokenizer.from_pretrained(model_name)
+            model = AutoModelForCausalLM.from_pretrained(model_name)
+            
+            pipe = pipeline(
+                "text-generation",
+                model=model, 
+                tokenizer=tokenizer,
+                max_new_tokens=300,
+                temperature=0.7,
+                do_sample=True,
+                device="cpu"  # Spaces環境ではCPU使用
+            )
+            
+            local_llm = HuggingFacePipeline(pipeline=pipe)
+            
+            # プロンプトテンプレート
+            template = """
+            次の手順書データを使って質問に答えてください。
+
+            ### 手順書データ:
+            {context}
+
+            ### 質問:
+            {question}
+
+            ### 回答:
+            """
+            
+            prompt = PromptTemplate(
+                template=template,
+                input_variables=["context", "question"]
+            )
+            
+            # QAチェーンの作成
+            self.qa_chain = RetrievalQA.from_chain_type(
+                llm=local_llm,
+                chain_type="stuff",
+                retriever=self.vectorstore.as_retriever(search_kwargs={"k": 3}),
+                chain_type_kwargs={"prompt": prompt},
+                return_source_documents=True
+            )
+            
+            print("QAチェーンを初期化しました")
+        except Exception as e:
+            print(f"QAチェーンの初期化中にエラーが発生しました: {str(e)}")
+            self.qa_chain = None
+
+    def extract_text_from_pdf_with_ocr(self, pdf_path):
+        """PDFファイルからテキストを抽出し、必要に応じてOCRを適用する"""
+        if not PYMUPDF_AVAILABLE or not EASYOCR_AVAILABLE or not self.reader:
+            return f"PDF: {os.path.basename(pdf_path)} (OCR未対応)"
+            
+        doc = fitz.open(pdf_path)
+        full_text = ""
+        
+        for page_num, page in enumerate(doc):
+            # テキストの抽出を試みる
+            text = page.get_text()
+            
+            # テキストが少ない場合はOCRを適用する
+            if len(text.strip()) < 50:  # 少ないテキストの閾値
+                # ページを画像として抽出
+                pix = page.get_pixmap()
+                img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
+                img_np = np.array(img)
+                
+                # EasyOCRを使用してテキスト抽出
+                result = self.reader.readtext(img_np)
+                ocr_text = "\n".join([detection[1] for detection in result])
+                
+                # OCRテキストを使用
+                text = ocr_text if ocr_text.strip() else text
+            
+            full_text += f"--- Page {page_num + 1} ---\n{text}\n\n"
+            
+        return full_text
+
+    def ask(self, question):
+        """質問をボットに問いかけ、回答と参照ソースを取得する"""
+        if not hasattr(self, 'qa_chain') or self.qa_chain is None:
+            return "チャットボットがまだ初期化されていません。ファイルをアップロードしてください。", ""
+        
+        try:
+            result = self.qa_chain.invoke({"query": question})
+            
+            # 回答の取得
+            if "result" in result:
+                answer = result["result"]
+            else:
+                return "回答を生成できませんでした。", ""
+            
+            # 参照ソースの取得
+            source_documents = result.get("source_documents", [])
+            sources_text = ""
+            
+            if source_documents:
+                sources_text = "参照ソース:\n"
+                for i, doc in enumerate(source_documents, 1):
+                    source = doc.metadata.get("source", "不明")
+                    filename = os.path.basename(source)
+                    sources_text += f"{i}. {filename}\n"
+            
+            return answer, sources_text
+        
+        except Exception as e:
+            return f"エラーが発生しました: {str(e)}", ""
+
+    def load(self):
+        """保存済みのベクトルストアを読み込む"""
+        if os.path.exists("./chroma_db"):
+            try:
+                embeddings = HuggingFaceEmbeddings(
+                    model_name="intfloat/multilingual-e5-base",
+                    model_kwargs={'device': 'cpu'}
+                )
+                
+                self.vectorstore = Chroma(
+                    persist_directory="./chroma_db",
+                    embedding_function=embeddings
+                )
+                
+                # QAチェーンを初期化
+                self._initialize_qa_chain()
+                
+                self.processing_status = "準備完了"
+                return "保存済みの知識ベースを読み込みました"
+            except Exception as e:
+                self.processing_status = "エラー"
+                return f"知識ベースの読み込みに失敗しました: {str(e)}"
+        else:
+            self.processing_status = "初期化待ち"
+            return "知識ベースが見つかりません。ファイルをアップロードしてください。"
+
+# Gradioインターフェースの作成
+def create_interface():
+    # チャットボットのインスタンスを作成
+    bot = ManualChatbot(docs_dir="./manuals")
+    
+    # 保存済みデータがあれば読み込む
+    load_status = bot.load()
+    
+    # Gradioインターフェース
+    with gr.Blocks(title="手順書RAGチャットボット") as demo:
+        gr.Markdown("# 手順書RAGチャットボット")
+        gr.Markdown("PDFやExcel、画像ファイルをアップロードして、それらの内容に関する質問に答��ます。")
+        
+        with gr.Tab("ファイルアップロード"):
+            upload_files = gr.File(file_count="multiple", label="PDFやExcel、画像ファイルをアップロード")
+            upload_button = gr.Button("処理開始")
+            status_output = gr.Textbox(label="ステータス", value=load_status)
+            
+            upload_button.click(
+                fn=bot.process_uploaded_files,
+                inputs=[upload_files],
+                outputs=[status_output]
+            )
+        
+        with gr.Tab("チャット"):
+            chatbot = gr.Chatbot(label="会話")
+            msg = gr.Textbox(label="質問を入力してください")
+            clear = gr.Button("クリア")
+            
+            def respond(message, chat_history):
+                if not message.strip():
+                    return chat_history
+                
+                # ボットに質問する
+                bot_response, sources = bot.ask(message)
+                
+                # 回答とソース情報を組み合わせる
+                full_response = bot_response
+                if sources:
+                    full_response += f"\n\n{sources}"
+                
+                # チャット履歴を更新する
+                chat_history.append((message, full_response))
+                return "", chat_history
+            
+            msg.submit(respond, [msg, chatbot], [msg, chatbot])
+            clear.click(lambda: None, None, chatbot, queue=False)
+        
+        with gr.Tab("使い方"):
+            gr.Markdown("""
+            ## 使い方
+            
+            1. **ファイルアップロード**タブで、PDFファイル、Excelファイル、または画像ファイルをアップロードします。
+            2. **処理開始**ボタンをクリックして、ファイルを処理します。
+            3. 処理が完了したら**チャット**タブに移動します。
+            4. 質問を入力して、手順書の内容に基づいた回答を得ることができます。
+            
+            ## サポートしているファイル形式
+            
+            - PDF (.pdf)
+            - Excel (.xlsx, .xls)
+            - 画像ファイル (.png, .jpg, .jpeg)
+            
+            ## 注意事項
+            
+            - 大きなファイルの処理には時間がかかる場合があります。
+            - 画像からのテキスト抽出（OCR）は言語によって精度が異なります。
+            - 回答は参照元のドキュメントに基づいて生成されるため、データが不十分な場合は正確な回答ができない場合があります。
+            """)
+    
+    return demo
+
+# Hugging Face Spacesで実行する場合のエントリーポイント
 if __name__ == "__main__":
-    demo.launch()
+    # Gradioインターフェースを作成して起動
+    demo = create_interface()
+    demo.launch()