Delete app.py

app.py

@@ -1,695 +0,0 @@
-# BioGPT Medical Chatbot with Gradio Interface - HUGGING FACE SPACES VERSION
-
-import gradio as gr
-import torch
-import warnings
-import numpy as np
-import os
-import re
-import time
-from datetime import datetime
-from typing import List, Dict, Optional, Tuple
-import json
-
-# Install required packages if not already installed
-try:
-    from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
-    from sentence_transformers import SentenceTransformer
-    import faiss
-except ImportError:
-    print("Installing required packages...")
-    import subprocess
-    import sys
-    
-    packages = [
-        "transformers>=4.21.0",
-        "torch>=1.12.0", 
-        "sentence-transformers",
-        "faiss-cpu",
-        "accelerate",
-        "bitsandbytes",
-        "datasets",
-        "numpy",
-        "sacremoses",
-        "scipy"
-    ]
-    
-    for package in packages:
-        try:
-            subprocess.check_call([sys.executable, "-m", "pip", "install", package])
-        except Exception as e:
-            print(f"Failed to install {package}: {e}")
-    
-    from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
-    from sentence_transformers import SentenceTransformer
-    import faiss
-
-# Suppress warnings
-warnings.filterwarnings('ignore')
-
-class GradioBioGPTChatbot:
-    def __init__(self, use_gpu=False, use_8bit=False):  # Default to CPU for HF Spaces
-        """Initialize BioGPT chatbot for Gradio deployment"""
-        self.device = "cuda" if torch.cuda.is_available() and use_gpu else "cpu"
-        self.use_8bit = use_8bit and torch.cuda.is_available()
-        
-        print(f"🔧 Initializing on device: {self.device}")
-        
-        # Initialize components with error handling
-        try:
-            self.setup_embeddings()
-            self.setup_faiss_index()
-            self.setup_biogpt()
-        except Exception as e:
-            print(f"❌ Initialization error: {e}")
-            self.model = None
-            self.tokenizer = None
-            self.embedding_model = None
-        
-        # Conversation tracking
-        self.conversation_history = []
-        self.knowledge_chunks = []
-        self.is_data_loaded = False
-        
-    def setup_embeddings(self):
-        """Setup medical-optimized embeddings with error handling"""
-        try:
-            print("🔄 Loading embedding model...")
-            self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
-            self.embedding_dim = self.embedding_model.get_sentence_embedding_dimension()
-            self.use_embeddings = True
-            print("✅ Embeddings loaded successfully")
-        except Exception as e:
-            print(f"❌ Embeddings setup failed: {e}")
-            self.embedding_model = None
-            self.embedding_dim = 384
-            self.use_embeddings = False
-
-    def setup_faiss_index(self):
-        """Setup FAISS for vector search with error handling"""
-        try:
-            print("🔄 Setting up FAISS index...")
-            self.faiss_index = faiss.IndexFlatIP(self.embedding_dim)
-            self.faiss_ready = True
-            print("✅ FAISS index ready")
-        except Exception as e:
-            print(f"❌ FAISS setup failed: {e}")
-            self.faiss_index = None
-            self.faiss_ready = False
-
-    def setup_biogpt(self):
-        """Setup BioGPT model with optimizations and fallbacks"""
-        print("🔄 Loading BioGPT model...")
-
-        models_to_try = [
-            "microsoft/BioGPT",  # Smaller version
-            "microsoft/DialoGPT-medium",  # Fallback 1
-            "microsoft/DialoGPT-small",   # Fallback 2
-            "gpt2"  # Final fallback
-        ]
-
-        for model_name in models_to_try:
-            try:
-                print(f"🔄 Trying model: {model_name}")
-
-                self.tokenizer = AutoTokenizer.from_pretrained(model_name)
-                if self.tokenizer.pad_token is None:
-                    self.tokenizer.pad_token = self.tokenizer.eos_token
-
-                if self.device == "cuda" and self.use_8bit:
-                    quantization_config = BitsAndBytesConfig(load_in_8bit=True, llm_int8_threshold=6.0)
-                else:
-                    quantization_config = None
-
-                self.model = AutoModelForCausalLM.from_pretrained(
-                    model_name,
-                    quantization_config=quantization_config,
-                    torch_dtype=torch.float16 if self.device == "cuda" else torch.float32,
-                    device_map="auto" if self.device == "cuda" else None,
-                    trust_remote_code=True,
-                    low_cpu_mem_usage=True
-                )
-
-                if self.device == "cpu":
-                    self.model = self.model.to(self.device)
-
-                print(f"✅ Successfully loaded: {model_name}")
-                return  # Exit after first successful model load
-
-            except Exception as e:
-                print(f"❌ Failed to load {model_name}: {e}")
-                continue
-
-        print("❌ All models failed to load")
-        self.model = None
-        self.tokenizer = None
-
-    def create_medical_chunks(self, text: str, chunk_size: int = 300) -> List[Dict]:
-        """Create medically-optimized text chunks with smaller size for efficiency"""
-        chunks = []
-        
-        # Split by medical sections first
-        medical_sections = self.split_by_medical_sections(text)
-        
-        chunk_id = 0
-        for section in medical_sections:
-            if len(section.split()) > chunk_size:
-                # Split large sections by sentences
-                sentences = re.split(r'[.!?]+', section)
-                current_chunk = ""
-                
-                for sentence in sentences:
-                    sentence = sentence.strip()
-                    if not sentence:
-                        continue
-                        
-                    if len(current_chunk.split()) + len(sentence.split()) < chunk_size:
-                        current_chunk += sentence + ". "
-                    else:
-                        if current_chunk.strip():
-                            chunks.append({
-                                'id': chunk_id,
-                                'text': current_chunk.strip(),
-                                'medical_focus': self.identify_medical_focus(current_chunk)
-                            })
-                            chunk_id += 1
-                        current_chunk = sentence + ". "
-                
-                if current_chunk.strip():
-                    chunks.append({
-                        'id': chunk_id,
-                        'text': current_chunk.strip(),
-                        'medical_focus': self.identify_medical_focus(current_chunk)
-                    })
-                    chunk_id += 1
-            else:
-                if section.strip():  # Don't add empty sections
-                    chunks.append({
-                        'id': chunk_id,
-                        'text': section,
-                        'medical_focus': self.identify_medical_focus(section)
-                    })
-                    chunk_id += 1
-                
-        return chunks
-
-    def split_by_medical_sections(self, text: str) -> List[str]:
-        """Split text by medical sections"""
-        section_patterns = [
-            r'\n\s*(?:SYMPTOMS?|TREATMENT|DIAGNOSIS|CAUSES?|PREVENTION|MANAGEMENT).*?\n',
-            r'\n\s*\d+\.\s+',
-            r'\n\n+'
-        ]
-        
-        sections = [text]
-        for pattern in section_patterns:
-            new_sections = []
-            for section in sections:
-                splits = re.split(pattern, section, flags=re.IGNORECASE)
-                new_sections.extend([s.strip() for s in splits if len(s.strip()) > 50])  # Reduced minimum length
-            sections = new_sections
-            
-        return sections
-
-    def identify_medical_focus(self, text: str) -> str:
-        """Identify the medical focus of a text chunk"""
-        text_lower = text.lower()
-        
-        categories = {
-            'pediatric_symptoms': ['fever', 'cough', 'rash', 'vomiting', 'diarrhea', 'child', 'baby', 'infant'],
-            'treatments': ['treatment', 'therapy', 'medication', 'antibiotics', 'medicine'],
-            'diagnosis': ['diagnosis', 'diagnostic', 'symptoms', 'signs', 'condition'],
-            'emergency': ['emergency', 'urgent', 'serious', 'hospital', 'call doctor'],
-            'prevention': ['prevention', 'vaccine', 'immunization', 'avoid', 'prevent']
-        }
-        
-        for category, keywords in categories.items():
-            if any(keyword in text_lower for keyword in keywords):
-                return category
-                
-        return 'general_medical'
-
-    def load_medical_data_from_file(self, file_path: str) -> Tuple[str, bool]:
-        """Load medical data from uploaded file with better error handling"""
-        if not file_path or not os.path.exists(file_path):
-            return "❌ No file uploaded or file not found.", False
-            
-        try:
-            print(f"🔄 Processing file: {file_path}")
-            
-            # Read file with encoding detection
-            encodings_to_try = ['utf-8', 'utf-8-sig', 'latin-1', 'cp1252']
-            text = None
-            
-            for encoding in encodings_to_try:
-                try:
-                    with open(file_path, 'r', encoding=encoding) as f:
-                        text = f.read()
-                    print(f"✅ File read successfully with {encoding} encoding")
-                    break
-                except UnicodeDecodeError:
-                    continue
-            
-            if text is None:
-                return "❌ Could not read file. Please ensure it's a valid text file.", False
-            
-            if len(text.strip()) < 100:
-                return "❌ File appears to be too short or empty. Please upload a substantial medical text.", False
-                
-            # Create chunks
-            print("🔄 Creating medical chunks...")
-            chunks = self.create_medical_chunks(text)
-            
-            if not chunks:
-                return "❌ No valid medical content found in the file.", False
-                
-            self.knowledge_chunks = chunks
-            print(f"✅ Created {len(chunks)} chunks")
-            
-            # Generate embeddings if available
-            if self.use_embeddings and self.embedding_model and self.faiss_ready:
-                print("🔄 Generating embeddings...")
-                success = self.generate_embeddings_and_index(chunks)
-                if success:
-                    self.is_data_loaded = True
-                    return f"✅ Medical data loaded successfully! {len(chunks)} chunks processed with vector search.", True
-            
-            self.is_data_loaded = True
-            return f"✅ Medical data loaded successfully! {len(chunks)} chunks processed (keyword search mode).", True
-            
-        except Exception as e:
-            print(f"❌ Error processing file: {e}")
-            return f"❌ Error loading file: {str(e)}", False
-
-    def generate_embeddings_and_index(self, chunks: List[Dict]) -> bool:
-        """Generate embeddings and add to FAISS index with error handling"""
-        try:
-            print("🔄 Generating embeddings...")
-            texts = [chunk['text'] for chunk in chunks]
-            
-            # Process in batches to avoid memory issues
-            batch_size = 32
-            all_embeddings = []
-            
-            for i in range(0, len(texts), batch_size):
-                batch_texts = texts[i:i+batch_size]
-                batch_embeddings = self.embedding_model.encode(batch_texts, show_progress_bar=False)
-                all_embeddings.append(batch_embeddings)
-            
-            embeddings = np.vstack(all_embeddings)
-            self.faiss_index.add(embeddings.astype('float32'))
-            print(f"✅ Added {len(embeddings)} embeddings to FAISS index")
-            return True
-            
-        except Exception as e:
-            print(f"❌ Embedding generation failed: {e}")
-            return False
-
-    def retrieve_medical_context(self, query: str, n_results: int = 3) -> List[str]:
-        """Retrieve relevant medical context with fallback"""
-        if not self.knowledge_chunks:
-            return []
-            
-        if self.use_embeddings and self.embedding_model and self.faiss_ready:
-            try:
-                query_embedding = self.embedding_model.encode([query])
-                distances, indices = self.faiss_index.search(query_embedding.astype('float32'), n_results)
-                context_chunks = []
-                for i in indices[0]:
-                    if i != -1 and i < len(self.knowledge_chunks):
-                        context_chunks.append(self.knowledge_chunks[i]['text'])
-                
-                if context_chunks:
-                    return context_chunks
-            except Exception as e:
-                print(f"❌ Embedding search failed: {e}")
-        
-        # Fallback to keyword search
-        return self.keyword_search_medical(query, n_results)
-
-    def keyword_search_medical(self, query: str, n_results: int) -> List[str]:
-        """Medical-focused keyword search"""
-        if not self.knowledge_chunks:
-            return []
-            
-        query_words = set(query.lower().split())
-        chunk_scores = []
-        
-        for chunk_info in self.knowledge_chunks:
-            chunk_text = chunk_info['text']
-            chunk_words = set(chunk_text.lower().split())
-            
-            word_overlap = len(query_words.intersection(chunk_words))
-            base_score = word_overlap / len(query_words) if query_words else 0
-            
-            # Boost medical content
-            medical_boost = 0
-            if chunk_info.get('medical_focus') in ['pediatric_symptoms', 'treatments', 'diagnosis']:
-                medical_boost = 0.3
-                
-            final_score = base_score + medical_boost
-            
-            if final_score > 0:
-                chunk_scores.append((final_score, chunk_text))
-        
-        chunk_scores.sort(reverse=True)
-        return [chunk for _, chunk in chunk_scores[:n_results]]
-
-    def generate_biogpt_response(self, context: str, query: str) -> str:
-        """Generate medical response using loaded model"""
-        if not self.model or not self.tokenizer:
-            return "Medical AI model is not available. Using fallback response based on retrieved context."
-            
-        try:
-            # Simplified prompt for better compatibility
-            prompt = f"Context: {context[:600]}\n\nQuestion: {query}\n\nAnswer:"
-
-            inputs = self.tokenizer(
-                prompt,
-                return_tensors="pt",
-                truncation=True,
-                max_length=512,  # Reduced for efficiency
-                padding=True
-            )
-            
-            if self.device == "cuda":
-                inputs = {k: v.to(self.device) for k, v in inputs.items()}
-            
-            with torch.no_grad():
-                outputs = self.model.generate(
-                    **inputs,
-                    max_new_tokens=100,  # Reduced for efficiency
-                    do_sample=True,
-                    temperature=0.7,
-                    top_p=0.9,
-                    pad_token_id=self.tokenizer.eos_token_id,
-                    repetition_penalty=1.1,
-                    no_repeat_ngram_size=3
-                )
-            
-            full_response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
-            
-            if "Answer:" in full_response:
-                generated_response = full_response.split("Answer:")[-1].strip()
-            else:
-                generated_response = full_response[len(prompt):].strip()
-            
-            return self.clean_medical_response(generated_response) if generated_response else self.fallback_response(context, query)
-            
-        except Exception as e:
-            print(f"❌ Generation failed: {e}")
-            return self.fallback_response(context, query)
-
-    def clean_medical_response(self, response: str) -> str:
-        """Clean and format medical response"""
-        if not response:
-            return "I couldn't generate a specific response. Please consult a healthcare professional."
-            
-        # Remove incomplete sentences and clean up
-        sentences = re.split(r'[.!?]+', response)
-        clean_sentences = []
-        
-        for sentence in sentences:
-            sentence = sentence.strip()
-            if len(sentence) > 15 and not sentence.endswith(('and', 'or', 'but', 'however', 'the', 'a', 'an')):
-                clean_sentences.append(sentence)
-            if len(clean_sentences) >= 2:  # Limit to 2 sentences for clarity
-                break
-        
-        if clean_sentences:
-            cleaned = '. '.join(clean_sentences) + '.'
-        else:
-            cleaned = response[:150] + '...' if len(response) > 150 else response
-            
-        return cleaned
-
-    def fallback_response(self, context: str, query: str) -> str:
-        """Fallback response when model generation fails"""
-        if not context:
-            return "I don't have specific information about this topic in my medical database. Please consult with a healthcare professional."
-            
-        # Extract most relevant sentences from context
-        sentences = [s.strip() for s in context.split('.') if len(s.strip()) > 20]
-        
-        if sentences:
-            # Return first 1-2 most relevant sentences
-            response = sentences[0]
-            if len(sentences) > 1 and len(response) < 100:
-                response += '. ' + sentences[1]
-            response += '.'
-        else:
-            response = context[:200] + '...' if len(context) > 200 else context
-            
-        return response
-
-    def handle_conversational_interactions(self, query: str) -> Optional[str]:
-        """Handle conversational interactions"""
-        query_lower = query.lower().strip()
-        
-        # Greetings
-        if query_lower in ['hello', 'hi', 'hey', 'good morning', 'good afternoon']:
-            if not self.is_data_loaded:
-                return "👋 Hello! I'm your medical AI assistant. Please upload your medical data file first, then ask me any health-related questions!"
-            else:
-                return "👋 Hello again! I'm ready to help. Ask me any medical question related to your uploaded data."
-
-        
-        # Thanks
-        if any(thanks in query_lower for thanks in ['thank you', 'thanks', 'thx', 'appreciate']):
-            return "🙏 You're welcome! Remember to always consult healthcare professionals for medical decisions. Feel free to ask more questions!"
-        
-        # Goodbyes
-        if any(bye in query_lower for bye in ['bye', 'goodbye', 'see you', 'farewell']):
-            return "👋 Goodbye! Take care and stay healthy! 🏥"
-        
-        # Help/About
-        if any(help_word in query_lower for help_word in ['help', 'what can you do', 'how do you work']):
-            return """🤖 **Medical AI Assistant**
-
-I can help with:
-• Medical information and conditions
-• Symptom understanding
-• Treatment information
-• When to seek medical care
-
-**How to use:**
-1. Upload your medical data file
-2. Ask specific medical questions
-3. Get evidence-based information
-
-⚠️ **Important:** I provide educational information only. Always consult healthcare professionals for medical advice."""
-
-        return None
-
-    def chat_interface(self, message: str, history: List[List[str]]) -> Tuple[str, List[List[str]]]:
-        """Main chat interface for Gradio"""
-        if not message.strip():
-            return "", history
-        
-        # Check if data is loaded
-        if not self.is_data_loaded:
-            response = "⚠️ Please upload your medical data file first using the file upload above before asking questions."
-            history.append([message, response])
-            return "", history
-        
-        # Handle conversational interactions
-        conversational_response = self.handle_conversational_interactions(message)
-        if conversational_response:
-            history.append([message, conversational_response])
-            return "", history
-        
-        # Process medical query
-        try:
-            context = self.retrieve_medical_context(message)
-            
-            if not context:
-                response = "I don't have specific information about this topic in my medical database. Please consult with a healthcare professional for personalized medical advice."
-            else:
-                main_context = '\n\n'.join(context)
-                medical_response = self.generate_biogpt_response(main_context, message)
-                response = f"🩺 **Medical Information:** {medical_response}\n\n⚠️ **Important:** This information is for educational purposes only. Always consult with qualified healthcare professionals for medical diagnosis, treatment, and personalized advice."
-            
-            # Add to conversation history
-            self.conversation_history.append({
-                'query': message,
-                'response': response,
-                'timestamp': datetime.now().isoformat()
-            })
-            
-            history.append([message, response])
-            return "", history
-            
-        except Exception as e:
-            print(f"❌ Chat interface error: {e}")
-            error_response = "I encountered an error processing your question. Please try again or consult a healthcare professional."
-            history.append([message, error_response])
-            return "", history
-
-# Initialize the chatbot with error handling
-print("🚀 Initializing Medical AI Assistant...")
-try:
-    chatbot = GradioBioGPTChatbot(use_gpu=False, use_8bit=False)  # CPU-optimized for HF Spaces
-    print("✅ Chatbot initialized successfully")
-except Exception as e:
-    print(f"❌ Chatbot initialization failed: {e}")
-    chatbot = None
-
-def upload_and_process_file(file):
-    """Handle file upload and processing"""
-    if file is None:
-        return "❌ No file uploaded."
-    
-    if chatbot is None:
-        return "❌ Chatbot not initialized properly. Please refresh the page."
-    
-    try:
-        message, success = chatbot.load_medical_data_from_file(file)
-        return message
-    except Exception as e:
-        return f"❌ Error processing file: {str(e)}"
-
-# Create Gradio Interface
-def create_gradio_interface():
-    """Create and launch Gradio interface"""
-    
-    with gr.Blocks(
-        title="🏥 Medical AI Assistant",
-        theme=gr.themes.Soft(),
-        css="""
-        .gradio-container {
-            max-width: 1200px !important;
-        }
-        .chat-message {
-            border-radius: 10px !important;
-        }
-        """
-    ) as demo:
-        
-        gr.HTML("""
-        <div style="text-align: center; padding: 20px;">
-            <h1>🏥 Medical AI Assistant</h1>
-            <p style="font-size: 18px; color: #666;">
-                AI-powered medical information assistant
-            </p>
-            <p style="color: #888;">
-                ⚠️ For educational purposes only. Always consult healthcare professionals for medical advice.
-            </p>
-        </div>
-        """)
-        
-        with gr.Row():
-            with gr.Column(scale=1):
-                gr.HTML("<h3>📁 Upload Medical Data</h3>")
-                file_upload = gr.File(
-                    label="Upload Medical Text File (.txt)",
-                    file_types=[".txt"],
-                    type="filepath"
-                )
-                upload_status = gr.Textbox(
-                    label="Upload Status",
-                    value="📋 Please upload your medical data file to begin...",
-                    interactive=False,
-                    lines=3
-                )
-                
-                gr.HTML("""
-                <div style="margin-top: 20px; padding: 15px; background-color: #f0f8ff; border-radius: 10px;">
-                    <h4>💡 How to Use:</h4>
-                    <ol>
-                        <li>Upload your medical text file (.txt format)</li>
-                        <li>Wait for processing confirmation</li>
-                        <li>Start asking medical questions!</li>
-                    </ol>
-                    
-                    <h4>📝 Example Questions:</h4>
-                    <ul>
-                        <li>"What causes fever in children?"</li>
-                        <li>"How to treat a persistent cough?"</li>
-                        <li>"When should I call the doctor?"</li>
-                        <li>"Signs of dehydration in infants?"</li>
-                    </ul>
-                </div>
-                """)
-            
-            with gr.Column(scale=2):
-                gr.HTML("<h3>💬 Medical Consultation</h3>")
-                chatbot_interface = gr.Chatbot(
-                    label="Medical AI Chat",
-                    height=500,
-                    bubble_full_width=False
-                )
-                
-                msg_input = gr.Textbox(
-                    label="Your Medical Question",
-                    placeholder="Ask me about health topics, symptoms, treatments, or when to seek care...",
-                    lines=2
-                )
-                
-                with gr.Row():
-                    send_btn = gr.Button("🩺 Send Question", variant="primary")
-                    clear_btn = gr.Button("🗑️ Clear Chat", variant="secondary")
-        
-        # Event handlers with error handling
-        def safe_upload_handler(file):
-            try:
-                return upload_and_process_file(file)
-            except Exception as e:
-                return f"❌ Upload error: {str(e)}"
-        
-        def safe_chat_handler(message, history):
-            try:
-                if chatbot is None:
-                    return "", history + [[message, "❌ System error. Please refresh the page."]]
-                return chatbot.chat_interface(message, history)
-            except Exception as e:
-                return "", history + [[message, f"❌ Error: {str(e)}"]]
-        
-        file_upload.change(
-            fn=safe_upload_handler,
-            inputs=[file_upload],
-            outputs=[upload_status]
-        )
-        
-        msg_input.submit(
-            fn=safe_chat_handler,
-            inputs=[msg_input, chatbot_interface],
-            outputs=[msg_input, chatbot_interface]
-        )
-        
-        send_btn.click(
-            fn=safe_chat_handler,
-            inputs=[msg_input, chatbot_interface],
-            outputs=[msg_input, chatbot_interface]
-        )
-        
-        clear_btn.click(
-            fn=lambda: ([], ""),
-            outputs=[chatbot_interface, msg_input]
-        )
-        
-        gr.HTML("""
-        <div style="text-align: center; margin-top: 30px; padding: 20px; background-color: #fff3cd; border-radius: 10px;">
-            <h4>⚠️ Medical Disclaimer</h4>
-            <p>This AI assistant provides educational medical information only and is not a substitute for professional medical advice, diagnosis, or treatment. Always seek the advice of qualified healthcare providers with questions about medical conditions.</p>
-        </div>
-        """)
-    
-    return demo
-
-if __name__ == "__main__":
-    # Create and launch the Gradio interface
-    demo = create_gradio_interface()
-    
-    print("🌐 Launching Gradio interface...")
-    print("📋 Upload your medical data file and start chatting!")
-    
-    # Launch with HF Spaces optimized settings
-    demo.launch(
-        share=False,
-        server_name="0.0.0.0",
-        server_port=7860,
-        show_error=True,
-        show_tips=False,
-        enable_queue=True,
-        max_threads=40
-    )