Update app.py

prompt become static , pdf loaded by default

app.py

@@ -1,6 +1,6 @@
 import os
 import time
-import fitz
+import fitz # PyMuPDF
 import faiss
 import pickle
 import numpy as np
@@ -16,21 +16,20 @@ from sentence_transformers import SentenceTransformer
 import gradio as gr
 
 # Define the ML_prompt (as it was in your notebook)
+# This prompt will now be hardcoded and not exposed to the user
 ML_prompt = """
 نقش ات:
 تو دستیار هوش مصنوعی من برای امتحان یادگیری ماشین هستی
 این امتحان تمرکز روی مفاهیم تیوری یادگیری ماشین داره
 منبع درس کتاب بیشاپ هست
-
 لحن صحبت کردن ات:
 تو استاد دانشگاه هستی و کسایی که باهات چت می کنن دانشجوهات اند
 """
-# api_key = os.getenv("google_api_key")
 
 class GeminiRAG:
     def __init__(self, api_key: str, model_name: str = "models/gemini-2.0-flash",
                  embed_model_name: str = "all-MiniLM-L6-v2", # Using a common SentenceTransformer model
-                 instruction_prompt: str = ML_prompt,
+                 instruction_prompt: str = ML_prompt, # Prompt is passed here
                  vectorstore_dir: str = "vectorstore"): # Use a directory within the app for persistence
 
         if not api_key:
@@ -62,19 +61,26 @@ class GeminiRAG:
         self.load_vectorstore()
 
     def _split_into_sentences(self, text: str) -> List[str]:
+        # Improved sentence splitting for better chunking
         sentences = re.split(r'(?<=[.!?])\s+', text)
         return [s.strip() for s in sentences if s.strip()]
 
     def load_document(self, pdf_path: str) -> List[str]:
-        doc = fitz.open(pdf_path)
-        page_contents = []
-        for page_num in range(len(doc)):
-            page = doc.load_page(page_num)
-            text = page.get_text()
-            if text.strip():
-                page_contents.append(text.strip())
-        doc.close()
-        return page_contents
+        print(f"Loading document from: {pdf_path}")
+        try:
+            doc = fitz.open(pdf_path)
+            page_contents = []
+            for page_num in range(len(doc)):
+                page = doc.load_page(page_num)
+                text = page.get_text()
+                if text.strip():
+                    page_contents.append(text.strip())
+            doc.close()
+            print(f"Successfully extracted {len(page_contents)} pages from {pdf_path}")
+            return page_contents
+        except Exception as e:
+            print(f"Error loading PDF {pdf_path}: {e}")
+            raise # Re-raise the exception to be caught higher up
 
     def add_document(self, parent_chunks: List[str]):
         new_sentence_chunks = []
@@ -107,7 +113,7 @@ class GeminiRAG:
 
         retrieved_parent_doc_indices = set()
         for idx in I[0]:
-            if idx < len(self.sentence_chunks):
+            if idx < len(self.sentence_chunks): # Ensure index is within bounds
                 parent_idx = self.sentence_to_parent_map[idx]
                 retrieved_parent_doc_indices.add(parent_idx)
 
@@ -115,7 +121,7 @@ class GeminiRAG:
         sorted_parent_indices = sorted(list(retrieved_parent_doc_indices))
 
         for parent_idx in sorted_parent_indices:
-            if parent_idx < len(self.parent_documents):
+            if parent_idx < len(self.parent_documents): # Ensure index is within bounds
                 context_parts.append(self.parent_documents[parent_idx])
 
         context = "\n\n---\\n\\n".join(context_parts)
@@ -123,17 +129,15 @@ class GeminiRAG:
         if not context.strip():
             return "No relevant information found in the knowledge base."
 
+        # The instruction prompt is now self.instruction_prompt which is set at init
         prompt = f"""
-              ### instruction prompt : (explanation : this text is your guideline don't mention it on response)
-              {self.instruction_prompt}
-
-              Use the following context to answer the question.\n
-              Context:\n
-              {context}\n
-
-              Question: {query}\n
-
-              Answer:"""
+             ### instruction prompt : (explanation : this text is your guideline don't mention it on response)
+             {self.instruction_prompt}
+             Use the following context to answer the question.\n
+             Context:\n
+             {context}\n
+             Question: {query}\n
+             Answer:"""
 
         for attempt in range(3):
             try:
@@ -142,28 +146,44 @@ class GeminiRAG:
             except InternalServerError as e:
                 print(f"Error: {e}. Retrying in 5 seconds...")
                 time.sleep(5)
-        raise Exception("Failed to generate after 3 retries.")
+            except Exception as e: # Catch other potential errors from API call
+                print(f"An unexpected error occurred during API call: {e}. Retrying in 5 seconds...")
+                time.sleep(5)
+        raise Exception("Failed to generate after 3 retries due to persistent errors.")
 
     def save_vectorstore(self):
-        faiss.write_index(self.index, self.vectorstore_faiss_path)
-        with open(self.vectorstore_data_path, "wb") as f:
-            pickle.dump({
-                'sentence_chunks': self.sentence_chunks,
-                'parent_documents': self.parent_documents,
-                'sentence_to_parent_map': self.sentence_to_parent_map
-            }, f)
-        print(f"Vectorstore saved to {self.vectorstore_faiss_path} and {self.vectorstore_data_path}")
+        try:
+            faiss.write_index(self.index, self.vectorstore_faiss_path)
+            with open(self.vectorstore_data_path, "wb") as f:
+                pickle.dump({
+                    'sentence_chunks': self.sentence_chunks,
+                    'parent_documents': self.parent_documents,
+                    'sentence_to_parent_map': self.sentence_to_parent_map
+                }, f)
+            print(f"Vectorstore saved to {self.vectorstore_faiss_path} and {self.vectorstore_data_path}")
+        except Exception as e:
+            print(f"Error saving vectorstore: {e}")
 
     def load_vectorstore(self):
         if os.path.exists(self.vectorstore_faiss_path) and os.path.exists(self.vectorstore_data_path):
-            self.index = faiss.read_index(self.vectorstore_faiss_path)
-            with open(self.vectorstore_data_path, "rb") as f:
-                data = pickle.load(f)
-                self.sentence_chunks = data['sentence_chunks']
-                self.parent_documents = data['parent_documents']
-                self.sentence_to_parent_map = data['sentence_to_parent_map']
-            print("📦 Loaded vectorstore.")
-            return True
+            try:
+                self.index = faiss.read_index(self.vectorstore_faiss_path)
+                with open(self.vectorstore_data_path, "rb") as f:
+                    data = pickle.load(f)
+                    self.sentence_chunks = data['sentence_chunks']
+                    self.parent_documents = data['parent_documents']
+                    self.sentence_to_parent_map = data['sentence_to_parent_map']
+                print("📦 Loaded vectorstore.")
+                return True
+            except Exception as e:
+                print(f"Error loading vectorstore: {e}")
+                # If loading fails, it's better to start fresh
+                self.index = faiss.IndexFlatL2(self.embedder.get_sentence_embedding_dimension())
+                self.sentence_chunks = []
+                self.parent_documents = []
+                self.sentence_to_parent_map = []
+                print("⚠️ Failed to load vectorstore, initializing a new one.")
+                return False
         print("ℹ️ No saved vectorstore found.")
         return False
 
@@ -172,86 +192,69 @@ class GeminiRAG:
 # Get API key from environment variable
 api_key = os.getenv("google_api_key")
 if not api_key:
-    raise ValueError("GEMINI_API_KEY environment variable not set. Please set it in Hugging Face Space secrets.")
+    print("Warning: GEMINI_API_KEY environment variable not set. Please set it in Hugging Face Space secrets.")
+
 
 # Initialize the RAG system globally for the Gradio app
-rag_instance = GeminiRAG(api_key=api_key)
+# The ML_prompt is passed during initialization and is then part of the rag_instance state
+rag_instance = GeminiRAG(api_key=api_key, instruction_prompt=ML_prompt) # Pass the prompt here
+
+# --- Load the predefined PDF at startup ---
+PDF_PATH = "MLT.pdf" # Assumes MLT.pdf is in the same directory as this script, or specify full path
+VECTORSTORE_BUILT_FLAG = os.path.join(rag_instance.vectorstore_dir, "vectorstore_built_flag.txt")
+
+
+if not rag_instance.load_vectorstore(): # Try to load existing
+    print(f"Attempting to load and process {PDF_PATH}...")
+    if os.path.exists(PDF_PATH):
+        try:
+            chunks = rag_instance.load_document(PDF_PATH)
+            if chunks:
+                rag_instance.add_document(chunks)
+                rag_instance.save_vectorstore()
+                with open(VECTORSTORE_BUILT_FLAG, "w") as f:
+                    f.write("Vectorstore built successfully.")
+                print("Initial PDF processed and vectorstore saved.")
+            else:
+                print(f"Warning: No text extracted from {PDF_PATH}. Please check the PDF content.")
+        except Exception as e:
+            print(f"Fatal Error: Could not process {PDF_PATH} at startup: {e}")
+    else:
+        print(f"Error: {PDF_PATH} not found. Please ensure the PDF file is in the correct directory.")
+
 
 def respond(
     message: str,
     history: list[list[str]], # Gradio Chatbot history format
-    system_message: str, # From additional_inputs
+    # Removed system_message from inputs as it's no longer user-configurable
     max_tokens: int, # From additional_inputs (not directly used by RAG but kept for interface consistency)
     temperature: float, # From additional_inputs (not directly used by RAG)
     top_p: float, # From additional_inputs (not directly used by RAG)
 ):
-    # The `system_message` from Gradio can be used to dynamically update the RAG's instruction prompt
-    # For this example, we'll keep the ML_prompt fixed, but you could add logic here:
-    # rag_instance.instruction_prompt = system_message
+    # The instruction prompt is now handled internally by rag_instance
+    # No need to access a system_message input here
+
+    if not rag_instance.sentence_chunks:
+        yield "Knowledge base is empty. Please ensure the PDF was loaded correctly at startup."
+        return
 
     try:
-        # Call your RAG system's ask_question method
-        # The top_k parameter can be exposed in Gradio's additional_inputs if needed
         response = rag_instance.ask_question(message)
-        # Gradio ChatInterface expects a generator for streaming or a direct string for non-streaming
-        yield response # Yield the full response, as ask_question does not stream token by token
+        yield response
     except Exception as e:
         yield f"❌ An error occurred: {e}"
 
-def upload_and_process_documents(files):
-    if not files:
-        return "Please upload PDF files to process."
-
-    # Re-initialize RAG instance to clear previous data and rebuild with new documents
-    # This is a simple approach; for more complex scenarios, you might want to append
-    # or manage different knowledge bases.
-    print("Rebuilding knowledge base with new documents...")
-    try:
-        # Re-initialize to clear previous data
-        global rag_instance
-        rag_instance = GeminiRAG(api_key=api_key)
-    except Exception as e:
-        return f"Error re-initializing RAG: {e}"
-
-    success_count = 0
-    error_files = []
-    for file_obj in files:
-        file_path = file_obj.name # Gradio passes a NamedTemporaryFile object
-        print(f"Processing {file_path}")
-        try:
-            chunks = rag_instance.load_document(file_path)
-            rag_instance.add_document(chunks)
-            success_count += 1
-        except Exception as e:
-            error_files.append(f"{os.path.basename(file_path)}: {e}")
-
-    rag_instance.save_vectorstore()
-
-    status_message = f"Successfully loaded and embedded {success_count} document(s)."
-    if error_files:
-        status_message += f"\nErrors occurred with: {'; '.join(error_files)}"
-    return status_message
-
-
 # Define the Gradio ChatInterface
 with gr.Blocks() as demo:
     gr.Markdown("# Gemini RAG Chatbot for ML Theory")
-    gr.Markdown("Upload your PDF documents, and then ask questions about the content. Ensure your `GEMINI_API_KEY` is set as a Space Secret.")
-
-    with gr.Row():
-        file_output = gr.Textbox(label="Upload Status", interactive=False)
-        upload_button = gr.UploadButton(
-            label="Upload PDF Documents",
-            file_types=["pdf"],
-            file_count="multiple"
-        )
-        upload_button.upload(upload_and_process_documents, inputs=upload_button, outputs=file_output)
-
-    # The ChatInterface component simplifies the chat UI setup
+    gr.Markdown(f"This chatbot is powered by {PDF_PATH}. Ensure your `GEMINI_API_KEY` is set as a Space Secret.")
+
+    # No file upload section anymore
+
     chat_interface_component = gr.ChatInterface(
         respond,
         additional_inputs=[
-            gr.Textbox(value=ML_prompt, label="System message", info="This sets the fixed role for the AI."),
+            # Removed the Textbox for system_message
             gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens", info="Not directly used by RAG model."),
             gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature", info="Not directly used by RAG model."),
             gr.Slider(
@@ -265,16 +268,14 @@ with gr.Blocks() as demo:
         ],
         chatbot=gr.Chatbot(height=400),
         textbox=gr.Textbox(placeholder="Ask me about Machine Learning Theory!", container=False, scale=7),
-        # clear_btn="Clear Chat",
         submit_btn="Send",
-        # Set examples for quick testing
+        # Update examples as the system_message input is no longer present
         examples=[
-            ["درمورد boosting بهم بگو", ML_prompt, 512, 0.7, 0.95],
-            ["انواع رگرسیون را توضیح بده", ML_prompt, 512, 0.7, 0.95],
-            ["شبکه های عصبی چیستند؟", ML_prompt, 512, 0.7, 0.95]
+            ["درمورد boosting بهم بگو", 512, 0.7, 0.95],
+            ["انواع رگرسیون را توضیح بده", 512, 0.7, 0.95],
+            ["شبکه های عصبی چیستند؟", 512, 0.7, 0.95]
         ]
     )
-    
 
 
 if __name__ == "__main__":