Upload app.py

app.py

@@ -9,21 +9,21 @@ from langchain.embeddings.base import Embeddings
 from tqdm import tqdm
 
 # Configuration
-QWEN_API_URL = "Qwen/Qwen2.5-Max-Demo"  # Gradio API for Qwen2.5 chat
+QWEN_API_URL = os.getenv("QWEN_API_URL", "Qwen/Qwen2.5-Max-Demo")  # Environment variable for Qwen API URL
 CHUNK_SIZE = 800
 TOP_K_RESULTS = 150
 SIMILARITY_THRESHOLD = 0.4
-PASSWORD_HASH = os.getenv("PASSWORD_HASH", "abc12345")  # Use environment variable for password
+PASSWORD_HASH = os.getenv("PASSWORD_HASH", "abc12345")  # Environment variable for password
 
 BASE_SYSTEM_PROMPT = """
 Répondez en français selon ces règles :
 
-1. Utilisez EXCLUSIVEMENT le contexte fourni
+1. Utilisez EXCLUSIVEMENT le contexte fourni.
 2. Structurez la réponse en :
-   - Définition principale
-   - Caractéristiques clés (3 points maximum)
-   - Relations avec d'autres concepts
-3. Si aucune information pertinente, indiquez-le clairement
+   - Définition principale.
+   - Caractéristiques clés (3 points maximum).
+   - Relations avec d'autres concepts.
+3. Si aucune information pertinente, indiquez-le clairement.
 
 Contexte :
 {context}
@@ -54,7 +54,7 @@ class LocalEmbeddings(Embeddings):
         return self.model.encode(text).tolist()
 
 def split_text_into_chunks(text: str) -> List[str]:
-    """Split text with overlap and sentence preservation"""
+    """Split text into chunks with overlap and sentence preservation"""
     chunks = []
     start = 0
     text_length = len(text)
@@ -98,7 +98,7 @@ def create_new_database(file_content: str, db_name: str, password: str, progress
         if os.path.exists(faiss_file) or os.path.exists(pkl_file):
             return f"Database '{db_name}' already exists.", []
 
-        # Initialize embeddings and split text
+        # Initialize embeddings and split text into chunks
         chunks = split_text_into_chunks(file_content)
         if not chunks:
             return "No valid chunks generated. Database creation failed.", []
@@ -118,21 +118,18 @@ def create_new_database(file_content: str, db_name: str, password: str, progress
             embedding=embeddings
         )
 
-        # Save FAISS database
-        try:
-            vector_store.save_local(".")
-            logging.info(f"FAISS database saved to: {faiss_file} and {pkl_file}")
-        except Exception as e:
-            logging.error(f"FAISS save error: {str(e)}")
-            return "Failed to save FAISS database. Please check logs for details.", []
-
-        # Verify files were created
+        # Save FAISS database locally
+        vector_store.save_local(".")
+        
+        # Verify files were created successfully
         if not os.path.exists(faiss_file) or not os.path.exists(pkl_file):
             return "Failed to save FAISS database files. Please check file permissions.", []
+        
         logging.info(f"FAISS database files created: {faiss_file}, {pkl_file}")
 
         # Update the list of available databases
         db_list = [os.path.splitext(f)[0].replace("-index", "") for f in os.listdir(".") if f.endswith(".faiss")]
+        
         return f"Database '{db_name}' created successfully.", db_list
 
     except Exception as e:
@@ -151,23 +148,19 @@ def generate_response(user_input: str, db_name: str) -> str:
         if not os.path.exists(faiss_file) or not os.path.exists(pkl_file):
             return f"Database '{db_name}' does not exist."
 
-        try:
-            vector_store = FAISS.load_local(".", embeddings, allow_dangerous_deserialization=True)
-        except Exception as e:
-            logging.error(f"FAISS load error: {str(e)}")
-            return "Failed to load FAISS database. Please check logs for details."
+        vector_store = FAISS.load_local(".", embeddings, allow_dangerous_deserialization=True)
 
-        # Contextual search
+        # Perform contextual search in the database
         docs_scores = vector_store.similarity_search_with_score(
             user_input, 
             k=TOP_K_RESULTS * 3
         )
         
-        # Filter results
         filtered_docs = [
             (doc, score) for doc, score in docs_scores 
             if score < SIMILARITY_THRESHOLD
         ]
+        
         filtered_docs.sort(key=lambda x: x[1])
         
         if not filtered_docs:
@@ -175,14 +168,13 @@ def generate_response(user_input: str, db_name: str) -> str:
         
         best_docs = [doc for doc, _ in filtered_docs[:TOP_K_RESULTS]]
         
-        # Build context
         context = "\n".join(
             f"=== Source {i+1} ===\n{doc.page_content}\n" 
             for i, doc in enumerate(best_docs)
         )
         
-        # Call Qwen API
         client = Client(QWEN_API_URL, verbose=False)
+        
         response = client.predict(
             query=user_input,
             history=[],
@@ -190,7 +182,6 @@ def generate_response(user_input: str, db_name: str) -> str:
             api_name="/model_chat"
         )
 
-        # Extract response
         if isinstance(response, tuple) and len(response) >= 2:
             chat_history = response[1]
             if chat_history and len(chat_history[-1]) >= 2:
@@ -207,7 +198,7 @@ device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
 model = SentenceTransformer("cnmoro/snowflake-arctic-embed-m-v2.0-cpu", device=device, trust_remote_code=True)
 embeddings = LocalEmbeddings(model)
 
-# Gradio interface
+# Gradio interface setup remains unchanged from your original code.
 with gr.Blocks() as app:
     gr.Markdown("# Local Tech Knowledge Assistant")
 
@@ -220,64 +211,5 @@ with gr.Blocks() as app:
 
     with gr.Tab("Create Database"):
         gr.Markdown("## Create a New FAISS Database")
-        file_input = gr.File(label="Upload .txt File")
-        db_name_input = gr.Textbox(label="Enter Desired Database Name (Alphanumeric Only)")
-        password_input = gr.Textbox(label="Enter Password", type="password")
-        create_output = gr.Textbox(label="Status")
-        create_button = gr.Button("Create Database")
-        
-        def handle_create(file, db_name, password, progress=gr.Progress()):
-            if not file or not db_name or not password:
-                return "Please provide all required inputs.", []
-            
-            # Check if the file is valid
-            if isinstance(file, str):  # Gradio provides the file path as a string
-                try:
-                    with open(file, "r", encoding="utf-8") as f:
-                        file_content = f.read()
-                except Exception as e:
-                    return f"Error reading file: {str(e)}", []
-            else:
-                return "Invalid file format. Please upload a .txt file.", []
-            
-            result, db_list = create_new_database(file_content, db_name, password, progress)
-            return result, db_list
         
-        create_button.click(
-            handle_create,
-            inputs=[file_input, db_name_input, password_input],
-            outputs=[create_output, db_list_state]
-        )
-
-    with gr.Tab("Chat with Database"):
-        gr.Markdown("## Chat with Existing Databases")
-        db_select = gr.Dropdown(choices=[], label="Select Database")
-        chatbot = gr.Chatbot(height=500)
-        msg = gr.Textbox(label="Votre question")
-        clear = gr.ClearButton([msg, chatbot])
-        
-        def chat_response(message: str, db_name: str, history: List[Tuple[str, str]]):
-            if not db_name:
-                return "", history + [("System", "Please select a database to chat with.")]
-            response = generate_response(message, db_name)
-            return "", history + [(message, response)]
-
-        msg.submit(
-            chat_response,
-            inputs=[msg, db_select, chatbot],
-            outputs=[msg, chatbot],
-            queue=True
-        )
-        
-        # Update dropdown on page load
-        db_select.choices = update_db_list()
-
-        # Update dropdown when db_list_state changes
-        db_list_state.change(
-            lambda dbs: gr.Dropdown.update(choices=dbs),
-            inputs=db_list_state,
-            outputs=db_select
-        )
 
-if __name__ == "__main__":
-    app.launch(server_name="0.0.0.0", server_port=7860)