| # import gradio as gr | |
| # import os | |
| # from langchain_community.document_loaders import PyPDFLoader | |
| # from langchain.text_splitter import RecursiveCharacterTextSplitter | |
| # from langchain.vectorstores import FAISS | |
| # from langchain.embeddings import HuggingFaceEmbeddings | |
| # from langchain.chains import RetrievalQA | |
| # from langchain_community.llms import HuggingFacePipeline | |
| # from transformers import pipeline | |
| # from langchain.chains.question_answering import load_qa_chain | |
| # from langchain.prompts import PromptTemplate | |
| # # Global variables to store the model components | |
| # qa = None | |
| # model_loaded = False | |
| # def initialize_model(): | |
| # """Initialize the AI model components""" | |
| # global qa, model_loaded | |
| # if model_loaded: | |
| # return "Model already loaded" | |
| # try: | |
| # # Step 1: Load PDF (you need to upload your PDF to the Hugging Face Space) | |
| # pdf_path = "basava formulaity.pdf" # Make sure to upload this file to your space | |
| # if not os.path.exists(pdf_path): | |
| # return f"Error: PDF file not found at {pdf_path}. Please upload your PDF to the space." | |
| # loader = PyPDFLoader(pdf_path) | |
| # documents = loader.load() | |
| # # Step 2: Split Text into Chunks | |
| # text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=100) | |
| # chunks = text_splitter.split_documents(documents) | |
| # # Step 3: Generate Embeddings and Store in FAISS | |
| # embedding_model = HuggingFaceEmbeddings(model_name="mixedbread-ai/mxbai-embed-large-v1") | |
| # db = FAISS.from_documents(chunks, embedding_model) | |
| # retriever = db.as_retriever() | |
| # # Step 4: Load Local QA Model | |
| # flan_pipeline = pipeline( | |
| # "text2text-generation", | |
| # model="NousResearch/Llama-3.2-1B", | |
| # max_length=1024, | |
| # do_sample=True, | |
| # temperature=0.7, | |
| # top_p=0.9, | |
| # repetition_penalty=1.2 | |
| # ) | |
| # local_llm = HuggingFacePipeline(pipeline=flan_pipeline) | |
| # # Step 5: Define custom prompt template | |
| # prompt_template = PromptTemplate( | |
| # input_variables=["context", "question"], | |
| # template=""" | |
| # You are a helpful veterinary assistant. Based only on the following context, answer the question accurately, do not hallucinate try to keep your language simple and be as precise as possible dont use medical terminologies. | |
| # Context: | |
| # {context} | |
| # Question: | |
| # {question} | |
| # Answer:""" | |
| # ) | |
| # # Step 6: Load QA chain with the custom prompt | |
| # qa_chain = load_qa_chain(llm=local_llm, chain_type="stuff", prompt=prompt_template) | |
| # # Step 7: Build the RetrievalQA pipeline | |
| # qa = RetrievalQA( | |
| # combine_documents_chain=qa_chain, | |
| # retriever=retriever, | |
| # return_source_documents=False | |
| # ) | |
| # model_loaded = True | |
| # return "✅ Model loaded successfully!" | |
| # except Exception as e: | |
| # return f"❌ Error loading model: {str(e)}" | |
| # def analyze_cat_symptoms(query): | |
| # """ | |
| # Analyze cat symptoms using the loaded AI model | |
| # """ | |
| # global qa, model_loaded | |
| # if not model_loaded: | |
| # init_result = initialize_model() | |
| # if "Error" in init_result or "❌" in init_result: | |
| # return init_result | |
| # try: | |
| # if not query.strip(): | |
| # return "Please provide cat symptoms and information for analysis." | |
| # # Query the model | |
| # result = qa({"query": query}) | |
| # # Extract the answer | |
| # raw_output = result["result"] | |
| # # Remove everything before "Answer:" and keep only the actual answer | |
| # if "Answer:" in raw_output: | |
| # final_answer = raw_output.split("Answer:", 1)[-1].strip() | |
| # else: | |
| # final_answer = raw_output.strip() | |
| # # Add disclaimer | |
| # disclaimer = "\n\n⚠️ **Important Disclaimer:** This AI analysis is for informational purposes only and should not replace professional veterinary care. If your cat shows severe symptoms or you're concerned about their health, please consult a veterinarian immediately." | |
| # return final_answer + disclaimer | |
| # except Exception as e: | |
| # return f"❌ Error analyzing symptoms: {str(e)}" | |
| # # Create Gradio interface | |
| # def create_interface(): | |
| # with gr.Blocks(title="Cat Health Analyzer", theme=gr.themes.Soft()) as interface: | |
| # gr.Markdown(""" | |
| # # 🐱 Cat Health AI Analyzer | |
| # **AI-powered veterinary assistant for cat health analysis** | |
| # Enter detailed information about your cat and their symptoms for analysis. | |
| # """) | |
| # with gr.Row(): | |
| # with gr.Column(scale=2): | |
| # query_input = gr.Textbox( | |
| # label="Cat Information & Symptoms", | |
| # placeholder="""Example: | |
| # Cat Information: | |
| # - Name: Fluffy | |
| # - Age: 3 years | |
| # - Breed: Persian | |
| # - Weight: 4.5 kg | |
| # - Gender: Female | |
| # Observed Symptoms: sneezing, watery eyes, loss of appetite | |
| # Symptom Duration: 2-3 days | |
| # Symptom Severity: Moderate | |
| # Additional Notes: Cat has been hiding under the bed and not playing as usual. | |
| # What could be wrong with my cat and what should I do?""", | |
| # lines=15, | |
| # max_lines=20 | |
| # ) | |
| # analyze_btn = gr.Button("🔍 Analyze Symptoms", variant="primary", size="lg") | |
| # with gr.Column(scale=2): | |
| # output = gr.Textbox( | |
| # label="AI Analysis & Recommendations", | |
| # lines=15, | |
| # max_lines=25, | |
| # show_copy_button=True | |
| # ) | |
| # # Add example queries | |
| # gr.Markdown("### 📝 Example Queries") | |
| # examples = gr.Examples( | |
| # examples=[ | |
| # ["My cat (Whiskers, 5 years old, male) has been vomiting for 2 days and won't eat. He's hiding under the bed. What could be wrong?"], | |
| # ["My female cat (Luna, 2 years, Siamese, 3.5kg) has watery eyes, sneezing, and seems lethargic. Symptoms started 3 days ago. Should I be worried?"], | |
| # ["My 8-year-old cat has been drinking more water than usual and urinating frequently. He's also lost some weight recently. What might this indicate?"] | |
| # ], | |
| # inputs=[query_input], | |
| # outputs=[output], | |
| # fn=analyze_cat_symptoms, | |
| # cache_examples=False | |
| # ) | |
| # analyze_btn.click( | |
| # fn=analyze_cat_symptoms, | |
| # inputs=[query_input], | |
| # outputs=[output] | |
| # ) | |
| # # Add footer | |
| # gr.Markdown(""" | |
| # --- | |
| # **Note:** This tool uses AI to provide general guidance based on veterinary knowledge. | |
| # Always consult with a qualified veterinarian for proper diagnosis and treatment. | |
| # """) | |
| # return interface | |
| # # Initialize model on startup | |
| # print("🚀 Starting Cat Health AI Analyzer...") | |
| # print("📚 Loading AI model components...") | |
| # # Create and launch the interface | |
| # if __name__ == "__main__": | |
| # interface = create_interface() | |
| # interface.launch( | |
| # server_name="0.0.0.0", | |
| # server_port=7860, | |
| # share=False | |
| # ) | |
| #######################ALMOST PERFFF | |
| # import gradio as gr | |
| # import os | |
| # from langchain_community.document_loaders import PyPDFLoader | |
| # from langchain.text_splitter import RecursiveCharacterTextSplitter | |
| # from langchain.vectorstores import FAISS | |
| # from langchain.embeddings import HuggingFaceEmbeddings | |
| # from langchain.chains import RetrievalQA | |
| # from langchain_community.llms import HuggingFacePipeline | |
| # from transformers import pipeline, AutoModelForCausalLM, AutoTokenizer | |
| # from langchain.chains.question_answering import load_qa_chain | |
| # from langchain.prompts import PromptTemplate | |
| # import torch | |
| # # Global variables to store the model components | |
| # qa = None | |
| # model_loaded = False | |
| # def initialize_model(): | |
| # """Initialize the AI model components""" | |
| # global qa, model_loaded | |
| # if model_loaded: | |
| # return "Model already loaded" | |
| # try: | |
| # # Step 1: Load PDF | |
| # pdf_path = "basava formulaity.pdf" | |
| # if not os.path.exists(pdf_path): | |
| # return f"Error: PDF file not found at {pdf_path}. Please upload your PDF to the space." | |
| # loader = PyPDFLoader(pdf_path) | |
| # documents = loader.load() | |
| # # Step 2: Split Text into Chunks (optimized parameters) | |
| # text_splitter = RecursiveCharacterTextSplitter( | |
| # chunk_size=800, | |
| # chunk_overlap=200, | |
| # separators=["\n\n", "\n", ".", " ", ""] | |
| # ) | |
| # chunks = text_splitter.split_documents(documents) | |
| # # Step 3: Generate Embeddings and Store in FAISS (more efficient model) | |
| # embedding_model = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2") | |
| # db = FAISS.from_documents(chunks, embedding_model) | |
| # retriever = db.as_retriever(search_kwargs={"k": 4}) # Retrieve 4 relevant chunks | |
| # # Step 4: Load TinyLlama model (more efficient than Llama-3.2-1B) | |
| # model_id = "TinyLlama/TinyLlama-1.1B-Chat-v1.0" | |
| # tokenizer = AutoTokenizer.from_pretrained(model_id) | |
| # # Check if GPU is available | |
| # device = 0 if torch.cuda.is_available() else -1 | |
| # model = AutoModelForCausalLM.from_pretrained( | |
| # model_id, | |
| # torch_dtype=torch.float16 if device == 0 else torch.float32, | |
| # device_map="auto" if device == 0 else None, | |
| # low_cpu_mem_usage=True | |
| # ) | |
| # # Create text generation pipeline | |
| # text_gen_pipeline = pipeline( | |
| # "text-generation", | |
| # model=model, | |
| # tokenizer=tokenizer, | |
| # max_new_tokens=256, | |
| # temperature=0.7, | |
| # device=device | |
| # ) | |
| # local_llm = HuggingFacePipeline(pipeline=text_gen_pipeline) | |
| # # Step 5: Define improved prompt template | |
| # prompt_template = PromptTemplate( | |
| # input_variables=["context", "question"], | |
| # template=""" | |
| # You are a helpful veterinary assistant. Use only the context to answer. Explain briefly. | |
| # Context: | |
| # {context} | |
| # Question: | |
| # {question} | |
| # Answer:""" | |
| # ) | |
| # # Step 6: Load QA chain with the custom prompt | |
| # qa_chain = load_qa_chain(llm=local_llm, chain_type="stuff", prompt=prompt_template) | |
| # # Step 7: Build the RetrievalQA pipeline | |
| # qa = RetrievalQA( | |
| # combine_documents_chain=qa_chain, | |
| # retriever=retriever, | |
| # return_source_documents=False | |
| # ) | |
| # model_loaded = True | |
| # return "✅ Model loaded successfully!" | |
| # except Exception as e: | |
| # return f"❌ Error loading model: {str(e)}" | |
| # def analyze_cat_symptoms(query): | |
| # """ | |
| # Analyze cat symptoms using the loaded AI model | |
| # """ | |
| # global qa, model_loaded | |
| # if not model_loaded: | |
| # init_result = initialize_model() | |
| # if "Error" in init_result or "❌" in init_result: | |
| # return init_result | |
| # try: | |
| # if not query.strip(): | |
| # return "Please provide cat symptoms and information for analysis." | |
| # # Query the model | |
| # result = qa({"query": query}) | |
| # # Extract the answer | |
| # raw_output = result["result"] | |
| # # Clean up the output | |
| # if "Answer:" in raw_output: | |
| # final_answer = raw_output.split("Answer:", 1)[-1].strip() | |
| # else: | |
| # final_answer = raw_output.strip() | |
| # # Add disclaimer | |
| # disclaimer = "\n\n⚠️ **Important Disclaimer:** This AI analysis is for informational purposes only and should not replace professional veterinary care. If your cat shows severe symptoms or you're concerned about their health, please consult a veterinarian immediately." | |
| # return final_answer + disclaimer | |
| # except Exception as e: | |
| # return f"❌ Error analyzing symptoms: {str(e)}" | |
| # # Create Gradio interface | |
| # def create_interface(): | |
| # with gr.Blocks(title="Cat Health Analyzer", theme=gr.themes.Soft()) as interface: | |
| # gr.Markdown(""" | |
| # # 🐱 Cat Health AI Analyzer | |
| # **AI-powered veterinary assistant for cat health analysis** | |
| # Enter detailed information about your cat and their symptoms for analysis. | |
| # """) | |
| # with gr.Row(): | |
| # with gr.Column(scale=2): | |
| # query_input = gr.Textbox( | |
| # label="Cat Information & Symptoms", | |
| # placeholder="""Example: | |
| # Cat Information: | |
| # - Name: Fluffy | |
| # - Age: 3 years | |
| # - Breed: Persian | |
| # - Weight: 4.5 kg | |
| # - Gender: Female | |
| # Observed Symptoms: sneezing, watery eyes, loss of appetite | |
| # Symptom Duration: 2-3 days | |
| # Symptom Severity: Moderate | |
| # Additional Notes: Cat has been hiding under the bed and not playing as usual. | |
| # What could be wrong with my cat and what should I do?""", | |
| # lines=15, | |
| # max_lines=20 | |
| # ) | |
| # analyze_btn = gr.Button("🔍 Analyze Symptoms", variant="primary", size="lg") | |
| # with gr.Column(scale=2): | |
| # output = gr.Textbox( | |
| # label="AI Analysis & Recommendations", | |
| # lines=15, | |
| # max_lines=25, | |
| # show_copy_button=True | |
| # ) | |
| # # Add example queries | |
| # gr.Markdown("### 📝 Example Queries") | |
| # examples = gr.Examples( | |
| # examples=[ | |
| # ["My cat (Whiskers, 5 years old, male) has been vomiting for 2 days and won't eat. He's hiding under the bed. What could be wrong?"], | |
| # ["My female cat (Luna, 2 years, Siamese, 3.5kg) has watery eyes, sneezing, and seems lethargic. Symptoms started 3 days ago. Should I be worried?"], | |
| # ["My 8-year-old cat has been drinking more water than usual and urinating frequently. He's also lost some weight recently. What might this indicate?"] | |
| # ], | |
| # inputs=[query_input], | |
| # outputs=[output], | |
| # fn=analyze_cat_symptoms, | |
| # cache_examples=False | |
| # ) | |
| # analyze_btn.click( | |
| # fn=analyze_cat_symptoms, | |
| # inputs=[query_input], | |
| # outputs=[output] | |
| # ) | |
| # # Add footer | |
| # gr.Markdown(""" | |
| # --- | |
| # **Note:** This tool uses AI to provide general guidance based on veterinary knowledge. | |
| # Always consult with a qualified veterinarian for proper diagnosis and treatment. | |
| # """) | |
| # return interface | |
| # # Initialize model on startup | |
| # print("🚀 Starting Cat Health AI Analyzer...") | |
| # print("📚 Loading AI model components...") | |
| # # Preload the model during startup | |
| # initialize_model() | |
| # # Create and launch the interface | |
| # if __name__ == "__main__": | |
| # interface = create_interface() | |
| # interface.launch( | |
| # server_name="0.0.0.0", | |
| # server_port=7860, | |
| # share=False | |
| # ) | |
| ##################################ALMOST PERFFFFFF | |
| ############################## PRELOADED IMBEDDING fast but not quite it | |
| # import gradio as gr | |
| # import os | |
| # from langchain_community.vectorstores import FAISS | |
| # from langchain_community.embeddings import HuggingFaceEmbeddings | |
| # from langchain.chains import RetrievalQA | |
| # from langchain_community.llms import HuggingFacePipeline | |
| # from transformers import pipeline, AutoModelForCausalLM, AutoTokenizer | |
| # from langchain.chains.question_answering import load_qa_chain | |
| # from langchain.prompts import PromptTemplate | |
| # import torch | |
| # import threading | |
| # import time | |
| # # Global variables to store the model components | |
| # qa = None | |
| # model_loaded = False | |
| # loading_status = "Not started" | |
| # def load_model_async(): | |
| # """Load the model asynchronously in the background""" | |
| # global qa, model_loaded, loading_status | |
| # try: | |
| # loading_status = "Loading embeddings..." | |
| # # Step 1: Load pre-computed FAISS embeddings (fastest part) | |
| # embedding_files = ["index.faiss", "index.pkl"] | |
| # # Check if embedding files exist | |
| # for file in embedding_files: | |
| # if not os.path.exists(file): | |
| # loading_status = f"Error: {file} not found" | |
| # return | |
| # # Initialize embedding model (lightweight) | |
| # embedding_model = HuggingFaceEmbeddings( | |
| # model_name="sentence-transformers/all-MiniLM-L6-v2", | |
| # model_kwargs={'device': 'cpu'}, # Keep embeddings on CPU to save GPU memory | |
| # encode_kwargs={'normalize_embeddings': True} | |
| # ) | |
| # # Load the pre-computed FAISS vector store | |
| # db = FAISS.load_local(".", embedding_model, allow_dangerous_deserialization=True) | |
| # retriever = db.as_retriever(search_kwargs={"k": 3}) # Reduced from 4 to 3 for speed | |
| # loading_status = "Loading language model..." | |
| # # Step 2: Load a smaller, faster model for better performance | |
| # model_id = "microsoft/DialoGPT-small" # Much faster than TinyLlama | |
| # # Check device availability | |
| # device = "cuda" if torch.cuda.is_available() else "cpu" | |
| # # Load tokenizer | |
| # tokenizer = AutoTokenizer.from_pretrained(model_id) | |
| # if tokenizer.pad_token is None: | |
| # tokenizer.pad_token = tokenizer.eos_token | |
| # # Load model with optimizations | |
| # model = AutoModelForCausalLM.from_pretrained( | |
| # model_id, | |
| # torch_dtype=torch.float16 if device == "cuda" else torch.float32, | |
| # device_map="auto" if device == "cuda" else None, | |
| # low_cpu_mem_usage=True, | |
| # use_cache=True # Enable KV cache for faster inference | |
| # ) | |
| # # Create optimized text generation pipeline | |
| # text_gen_pipeline = pipeline( | |
| # "text-generation", | |
| # model=model, | |
| # tokenizer=tokenizer, | |
| # max_new_tokens=150, # Reduced for faster generation | |
| # temperature=0.6, | |
| # do_sample=True, | |
| # pad_token_id=tokenizer.eos_token_id, | |
| # device=0 if device == "cuda" else -1, | |
| # batch_size=1 # Single batch for consistent performance | |
| # ) | |
| # local_llm = HuggingFacePipeline(pipeline=text_gen_pipeline) | |
| # loading_status = "Setting up QA chain..." | |
| # # Step 3: Simplified prompt template for faster processing | |
| # prompt_template = PromptTemplate( | |
| # input_variables=["context", "question"], | |
| # template="""Based on the veterinary information provided, give a brief answer. | |
| # Context: {context} | |
| # Question: {question} | |
| # Answer:""" | |
| # ) | |
| # # Step 4: Load QA chain | |
| # qa_chain = load_qa_chain( | |
| # llm=local_llm, | |
| # chain_type="stuff", | |
| # prompt=prompt_template, | |
| # verbose=False # Disable verbose logging for speed | |
| # ) | |
| # # Step 5: Build the RetrievalQA pipeline | |
| # qa = RetrievalQA( | |
| # combine_documents_chain=qa_chain, | |
| # retriever=retriever, | |
| # return_source_documents=False # Skip source docs for speed | |
| # ) | |
| # model_loaded = True | |
| # loading_status = "✅ Ready! Model loaded successfully." | |
| # except Exception as e: | |
| # loading_status = f"❌ Error: {str(e)}" | |
| # model_loaded = False | |
| # def get_loading_status(): | |
| # """Get current loading status""" | |
| # return loading_status | |
| # def initialize_model(): | |
| # """Initialize or check model status""" | |
| # global model_loaded | |
| # if model_loaded: | |
| # return "✅ Model is ready!" | |
| # return loading_status | |
| # def analyze_cat_symptoms(query): | |
| # """ | |
| # Analyze cat symptoms with optimized processing | |
| # """ | |
| # global qa, model_loaded | |
| # if not model_loaded: | |
| # return f"⏳ Model is still loading. Status: {loading_status}\n\nPlease wait a moment and try again." | |
| # try: | |
| # if not query.strip(): | |
| # return "Please provide cat symptoms and information for analysis." | |
| # # Pre-process query for better results | |
| # processed_query = f"Cat health question: {query}" | |
| # # Query with timeout protection | |
| # result = qa({"query": processed_query}) | |
| # # Extract and clean the answer | |
| # raw_output = result["result"] | |
| # # Clean up the output more aggressively | |
| # lines = raw_output.split('\n') | |
| # cleaned_lines = [] | |
| # for line in lines: | |
| # line = line.strip() | |
| # if line and not line.startswith('Context:') and not line.startswith('Question:'): | |
| # if 'Answer:' in line: | |
| # line = line.split('Answer:', 1)[-1].strip() | |
| # cleaned_lines.append(line) | |
| # final_answer = ' '.join(cleaned_lines).strip() | |
| # # Fallback if answer is too short or empty | |
| # if len(final_answer) < 20: | |
| # final_answer = "Based on the symptoms you've described, I recommend consulting with a veterinarian for proper diagnosis and treatment. The symptoms could indicate various conditions that require professional evaluation." | |
| # # Add disclaimer | |
| # disclaimer = "\n\n⚠️ **Important:** This is AI guidance only. Always consult a veterinarian for proper diagnosis and treatment." | |
| # return final_answer + disclaimer | |
| # except Exception as e: | |
| # return f"❌ Analysis error: {str(e)}\n\nPlease try rephrasing your question." | |
| # def check_embedding_status(): | |
| # """Check embedding files and model status""" | |
| # embedding_files = ["index.faiss", "index.pkl"] | |
| # status_lines = [] | |
| # for file in embedding_files: | |
| # if os.path.exists(file): | |
| # file_size = round(os.path.getsize(file) / (1024*1024), 2) # Size in MB | |
| # status_lines.append(f"✅ {file} ({file_size} MB)") | |
| # else: | |
| # status_lines.append(f"❌ {file} (missing)") | |
| # status_lines.append(f"🤖 Model Status: {loading_status}") | |
| # return "\n".join(status_lines) | |
| # # Create Gradio interface | |
| # def create_interface(): | |
| # with gr.Blocks(title="Cat Health Analyzer - Ultra Fast", theme=gr.themes.Soft()) as interface: | |
| # gr.Markdown(""" | |
| # # 🐱 Cat Health AI Analyzer (Ultra-Fast Edition) | |
| # **Optimized AI veterinary assistant with background model loading** | |
| # ⚡ The model loads automatically in the background for instant responses! | |
| # """) | |
| # # Status display that updates automatically | |
| # with gr.Row(): | |
| # status_display = gr.Textbox( | |
| # label="📊 System Status", | |
| # value=check_embedding_status(), | |
| # lines=4, | |
| # interactive=False | |
| # ) | |
| # with gr.Row(): | |
| # with gr.Column(scale=2): | |
| # query_input = gr.Textbox( | |
| # label="Cat Symptoms & Information", | |
| # placeholder="""Quick example: | |
| # My 3-year-old cat has been sneezing and has watery eyes for 2 days. Not eating much. What could this be?""", | |
| # lines=8, | |
| # max_lines=15 | |
| # ) | |
| # with gr.Row(): | |
| # analyze_btn = gr.Button("🔍 Analyze Now", variant="primary", size="lg") | |
| # refresh_btn = gr.Button("🔄 Refresh Status", variant="secondary") | |
| # with gr.Column(scale=2): | |
| # output = gr.Textbox( | |
| # label="AI Analysis", | |
| # lines=12, | |
| # max_lines=20, | |
| # show_copy_button=True | |
| # ) | |
| # # Quick examples for faster testing | |
| # gr.Markdown("### 🚀 Quick Examples") | |
| # examples = gr.Examples( | |
| # examples=[ | |
| # ["My cat is vomiting and won't eat. What should I do?"], | |
| # ["My cat has watery eyes and is sneezing. Is this serious?"], | |
| # ["My older cat is drinking more water and urinating frequently."] | |
| # ], | |
| # inputs=[query_input], | |
| # outputs=[output], | |
| # fn=analyze_cat_symptoms, | |
| # cache_examples=False | |
| # ) | |
| # # Event handlers | |
| # analyze_btn.click( | |
| # fn=analyze_cat_symptoms, | |
| # inputs=[query_input], | |
| # outputs=[output] | |
| # ) | |
| # refresh_btn.click( | |
| # fn=check_embedding_status, | |
| # inputs=[], | |
| # outputs=[status_display] | |
| # ) | |
| # # Auto-refresh status using a separate thread | |
| # def auto_refresh_status(): | |
| # while True: | |
| # time.sleep(3) | |
| # if model_loaded: | |
| # break | |
| # # This is a simplified approach - in a real app you'd use | |
| # # Gradio's built-in update mechanisms | |
| # # Start the auto-refresh thread | |
| # refresh_thread = threading.Thread(target=auto_refresh_status, daemon=True) | |
| # refresh_thread.start() | |
| # # Footer with performance info | |
| # gr.Markdown(""" | |
| # --- | |
| # **⚡ Performance Features:** | |
| # - 🔄 Background model loading | |
| # - 📁 Pre-computed embeddings | |
| # - 🚀 Optimized inference pipeline | |
| # - 💾 Memory-efficient processing | |
| # **Note:** This AI provides general guidance only. Always consult a veterinarian for medical decisions. | |
| # """) | |
| # return interface | |
| # # Start background model loading immediately | |
| # print("🚀 Starting Ultra-Fast Cat Health AI Analyzer...") | |
| # print("📁 Checking embedding files...") | |
| # # Check files on startup | |
| # embedding_status = check_embedding_status() | |
| # print(embedding_status) | |
| # if "❌" in embedding_status: | |
| # print("⚠️ Warning: Missing embedding files. Upload index.faiss and index.pkl files.") | |
| # loading_status = "❌ Missing embedding files" | |
| # else: | |
| # print("✅ Embedding files found! Starting background model loading...") | |
| # # Start loading model in background thread | |
| # loading_thread = threading.Thread(target=load_model_async, daemon=True) | |
| # loading_thread.start() | |
| # # Launch interface immediately | |
| # if __name__ == "__main__": | |
| # interface = create_interface() | |
| # interface.launch( | |
| # server_name="0.0.0.0", | |
| # server_port=7860, | |
| # share=False, | |
| # show_error=True | |
| # ) | |
| ######################### IMBEDDING LATEST | |
| # import gradio as gr | |
| # import os | |
| # from langchain_community.vectorstores import FAISS | |
| # from langchain_community.embeddings import HuggingFaceEmbeddings | |
| # from langchain.chains import RetrievalQA | |
| # from langchain_community.llms import HuggingFacePipeline | |
| # from transformers import pipeline, AutoModelForCausalLM, AutoTokenizer | |
| # from langchain.chains.question_answering import load_qa_chain | |
| # from langchain.prompts import PromptTemplate | |
| # import torch | |
| # import threading | |
| # # Global variables to store the model components | |
| # qa = None | |
| # model_loaded = False | |
| # loading_status = "Not started" | |
| # def load_model_async(): | |
| # """Load the model asynchronously in the background""" | |
| # global qa, model_loaded, loading_status | |
| # try: | |
| # loading_status = "Loading embeddings..." | |
| # # Step 1: Load pre-computed FAISS embeddings (fastest part) | |
| # embedding_files = ["index.faiss", "index.pkl"] | |
| # # Check if embedding files exist | |
| # for file in embedding_files: | |
| # if not os.path.exists(file): | |
| # loading_status = f"Error: {file} not found" | |
| # return | |
| # # Initialize embedding model (lightweight) | |
| # embedding_model = HuggingFaceEmbeddings( | |
| # model_name="sentence-transformers/all-MiniLM-L6-v2", | |
| # model_kwargs={'device': 'cpu'}, # Keep embeddings on CPU to save GPU memory | |
| # encode_kwargs={'normalize_embeddings': True} | |
| # ) | |
| # # Load the pre-computed FAISS vector store | |
| # db = FAISS.load_local(".", embedding_model, allow_dangerous_deserialization=True) | |
| # retriever = db.as_retriever(search_kwargs={"k": 3}) # Reduced from 4 to 3 for speed | |
| # loading_status = "Loading language model..." | |
| # # Step 2: Load TinyLlama model as requested | |
| # model_id = "TinyLlama/TinyLlama-1.1B-Chat-v1.0" | |
| # # Check device availability | |
| # device = "cuda" if torch.cuda.is_available() else "cpu" | |
| # # Load tokenizer | |
| # tokenizer = AutoTokenizer.from_pretrained(model_id) | |
| # if tokenizer.pad_token is None: | |
| # tokenizer.pad_token = tokenizer.eos_token | |
| # # Load model with optimizations | |
| # model = AutoModelForCausalLM.from_pretrained( | |
| # model_id, | |
| # torch_dtype=torch.float16 if device == "cuda" else torch.float32, | |
| # device_map="auto" if device == "cuda" else None, | |
| # low_cpu_mem_usage=True, | |
| # use_cache=True # Enable KV cache for faster inference | |
| # ) | |
| # # Create optimized text generation pipeline | |
| # text_gen_pipeline = pipeline( | |
| # "text-generation", | |
| # model=model, | |
| # tokenizer=tokenizer, | |
| # max_new_tokens=256, # Back to original value as requested | |
| # temperature=0.7, | |
| # do_sample=True, | |
| # pad_token_id=tokenizer.eos_token_id, | |
| # device=0 if device == "cuda" else -1, | |
| # batch_size=1 # Single batch for consistent performance | |
| # ) | |
| # local_llm = HuggingFacePipeline(pipeline=text_gen_pipeline) | |
| # loading_status = "Setting up QA chain..." | |
| # # Step 3: Simplified prompt template for faster processing | |
| # prompt_template = PromptTemplate( | |
| # input_variables=["context", "question"], | |
| # template="""Based on the veterinary information provided, give a brief answer. | |
| # Context: {context} | |
| # Question: {question} | |
| # Answer:""" | |
| # ) | |
| # # Step 4: Load QA chain | |
| # qa_chain = load_qa_chain( | |
| # llm=local_llm, | |
| # chain_type="stuff", | |
| # prompt=prompt_template, | |
| # verbose=False # Disable verbose logging for speed | |
| # ) | |
| # # Step 5: Build the RetrievalQA pipeline | |
| # qa = RetrievalQA( | |
| # combine_documents_chain=qa_chain, | |
| # retriever=retriever, | |
| # return_source_documents=False # Skip source docs for speed | |
| # ) | |
| # model_loaded = True | |
| # loading_status = "✅ Ready! Model loaded successfully." | |
| # except Exception as e: | |
| # loading_status = f"❌ Error: {str(e)}" | |
| # model_loaded = False | |
| # def get_loading_status(): | |
| # """Get current loading status""" | |
| # return loading_status | |
| # def initialize_model(): | |
| # """Initialize or check model status""" | |
| # global model_loaded | |
| # if model_loaded: | |
| # return "✅ Model is ready!" | |
| # return loading_status | |
| # def analyze_cat_symptoms(query): | |
| # """ | |
| # Analyze cat symptoms with optimized processing | |
| # """ | |
| # global qa, model_loaded | |
| # if not model_loaded: | |
| # return f"⏳ Model is still loading. Status: {loading_status}\n\nPlease wait a moment and try again." | |
| # try: | |
| # if not query.strip(): | |
| # return "Please provide cat symptoms and information for analysis." | |
| # # Pre-process query for better results | |
| # processed_query = f"Cat health question: {query}" | |
| # # Query with timeout protection | |
| # result = qa({"query": processed_query}) | |
| # # Extract and clean the answer | |
| # raw_output = result["result"] | |
| # # Clean up the output more aggressively | |
| # lines = raw_output.split('\n') | |
| # cleaned_lines = [] | |
| # for line in lines: | |
| # line = line.strip() | |
| # if line and not line.startswith('Context:') and not line.startswith('Question:'): | |
| # if 'Answer:' in line: | |
| # line = line.split('Answer:', 1)[-1].strip() | |
| # cleaned_lines.append(line) | |
| # final_answer = ' '.join(cleaned_lines).strip() | |
| # # Fallback if answer is too short or empty | |
| # if len(final_answer) < 20: | |
| # final_answer = "Based on the symptoms you've described, I recommend consulting with a veterinarian for proper diagnosis and treatment. The symptoms could indicate various conditions that require professional evaluation." | |
| # # Add disclaimer | |
| # disclaimer = "\n\n⚠️ **Important:** This is AI guidance only. Always consult a veterinarian for proper diagnosis and treatment." | |
| # return final_answer + disclaimer | |
| # except Exception as e: | |
| # return f"❌ Analysis error: {str(e)}\n\nPlease try rephrasing your question." | |
| # def check_embedding_status(): | |
| # """Check embedding files and model status""" | |
| # embedding_files = ["index.faiss", "index.pkl"] | |
| # status_lines = [] | |
| # for file in embedding_files: | |
| # if os.path.exists(file): | |
| # file_size = round(os.path.getsize(file) / (1024*1024), 2) # Size in MB | |
| # status_lines.append(f"✅ {file} ({file_size} MB)") | |
| # else: | |
| # status_lines.append(f"❌ {file} (missing)") | |
| # status_lines.append(f"🤖 Model Status: {loading_status}") | |
| # return "\n".join(status_lines) | |
| # # Create Gradio interface | |
| # def create_interface(): | |
| # with gr.Blocks(title="Cat Health Analyzer - Ultra Fast", theme=gr.themes.Soft()) as interface: | |
| # gr.Markdown(""" | |
| # # 🐱 Cat Health AI Analyzer (Ultra-Fast Edition) | |
| # **Optimized AI veterinary assistant with background model loading** | |
| # ⚡ The model loads automatically in the background for instant responses! | |
| # """) | |
| # # Status display that updates automatically | |
| # with gr.Row(): | |
| # status_display = gr.Textbox( | |
| # label="📊 System Status", | |
| # value=check_embedding_status(), | |
| # lines=4, | |
| # interactive=False | |
| # ) | |
| # with gr.Row(): | |
| # with gr.Column(scale=2): | |
| # query_input = gr.Textbox( | |
| # label="Cat Symptoms & Information", | |
| # placeholder="""Quick example: | |
| # My 3-year-old cat has been sneezing and has watery eyes for 2 days. Not eating much. What could this be?""", | |
| # lines=8, | |
| # max_lines=15 | |
| # ) | |
| # with gr.Row(): | |
| # analyze_btn = gr.Button("🔍 Analyze Now", variant="primary", size="lg") | |
| # refresh_btn = gr.Button("🔄 Refresh Status", variant="secondary") | |
| # with gr.Column(scale=2): | |
| # output = gr.Textbox( | |
| # label="AI Analysis", | |
| # lines=12, | |
| # max_lines=20, | |
| # show_copy_button=True | |
| # ) | |
| # # Quick examples for faster testing | |
| # gr.Markdown("### 🚀 Quick Examples") | |
| # examples = gr.Examples( | |
| # examples=[ | |
| # ["My cat is vomiting and won't eat. What should I do?"], | |
| # ["My cat has watery eyes and is sneezing. Is this serious?"], | |
| # ["My older cat is drinking more water and urinating frequently."] | |
| # ], | |
| # inputs=[query_input], | |
| # outputs=[output], | |
| # fn=analyze_cat_symptoms, | |
| # cache_examples=False | |
| # ) | |
| # # Event handlers | |
| # analyze_btn.click( | |
| # fn=analyze_cat_symptoms, | |
| # inputs=[query_input], | |
| # outputs=[output] | |
| # ) | |
| # refresh_btn.click( | |
| # fn=check_embedding_status, | |
| # inputs=[], | |
| # outputs=[status_display] | |
| # ) | |
| # # Footer with performance info | |
| # gr.Markdown(""" | |
| # --- | |
| # **⚡ Performance Features:** | |
| # - 🔄 Background model loading | |
| # - 📁 Pre-computed embeddings | |
| # - 🚀 Optimized inference pipeline | |
| # - 💾 Memory-efficient processing | |
| # **Note:** This AI provides general guidance only. Always consult a veterinarian for medical decisions. | |
| # """) | |
| # return interface | |
| # # Start background model loading immediately | |
| # print("🚀 Starting Ultra-Fast Cat Health AI Analyzer...") | |
| # print("📁 Checking embedding files...") | |
| # # Check files on startup | |
| # embedding_status = check_embedding_status() | |
| # print(embedding_status) | |
| # if "❌" in embedding_status: | |
| # print("⚠️ Warning: Missing embedding files. Upload index.faiss and index.pkl files.") | |
| # loading_status = "❌ Missing embedding files" | |
| # else: | |
| # print("✅ Embedding files found! Starting background model loading...") | |
| # # Start loading model in background thread | |
| # loading_thread = threading.Thread(target=load_model_async, daemon=True) | |
| # loading_thread.start() | |
| # # Launch interface immediately | |
| # if __name__ == "__main__": | |
| # interface = create_interface() | |
| # interface.launch( | |
| # server_name="0.0.0.0", | |
| # server_port=7860, | |
| # share=False, | |
| # show_error=True | |
| # ) | |
| ################################## MORE OPTIMIZATION and embedding optimization IRFANNN UnicodeError | |
| import os | |
| import shutil | |
| import gradio as gr | |
| from langchain_community.document_loaders import PyPDFLoader | |
| from langchain.text_splitter import RecursiveCharacterTextSplitter | |
| from langchain.vectorstores import FAISS | |
| from langchain.embeddings import HuggingFaceEmbeddings | |
| from langchain.chains import RetrievalQA | |
| from langchain.chains.question_answering import load_qa_chain | |
| from langchain.prompts import PromptTemplate | |
| from langchain_community.llms import HuggingFacePipeline | |
| from transformers import pipeline | |
| import torch | |
| # ----------------------------- | |
| # Config (can be changed in Space "Variables & secrets") | |
| # ----------------------------- | |
| EMBEDDINGS_MODEL = os.getenv("EMBEDDINGS_MODEL", "mixedbread-ai/mxbai-embed-large-v1") | |
| GENERATION_MODEL = os.getenv("GENERATION_MODEL", "google/flan-t5-base") | |
| MAX_NEW_TOKENS = int(os.getenv("MAX_NEW_TOKENS", "256")) | |
| TEMPERATURE = float(os.getenv("TEMPERATURE", "0.2")) | |
| HF_TOKEN = os.getenv("HF_TOKEN", None) | |
| # Where to save the FAISS index on disk | |
| INDEX_DIR = os.getenv("INDEX_DIR", "faiss_index") | |
| # ----------------------------- | |
| # Globals kept in memory | |
| # ----------------------------- | |
| db_state = {"db": None, "retriever": None} | |
| qa_state = {"qa": None} | |
| embeddings_state = {"emb": None} | |
| def get_embeddings(): | |
| if embeddings_state["emb"] is None: | |
| embeddings_state["emb"] = HuggingFaceEmbeddings(model_name=EMBEDDINGS_MODEL) | |
| return embeddings_state["emb"] | |
| def try_load_index(): | |
| """Try loading an existing FAISS index from disk on startup or when user opens the app.""" | |
| emb = get_embeddings() | |
| if os.path.isdir(INDEX_DIR) and os.path.exists(os.path.join(INDEX_DIR, "index.faiss")): | |
| try: | |
| db = FAISS.load_local(INDEX_DIR, emb, allow_dangerous_deserialization=True) | |
| db_state["db"] = db | |
| db_state["retriever"] = db.as_retriever() | |
| return f"✅ Loaded saved FAISS index from disk.\n- index: `{INDEX_DIR}`\n- embeddings: `{EMBEDDINGS_MODEL}`" | |
| except Exception as e: | |
| return f"⚠️ Found an index on disk but failed to load it: {e}" | |
| return f"ℹ️ No saved index found yet. Upload a PDF and click **Build index**.\n- expected index dir: `{INDEX_DIR}`\n- embeddings: `{EMBEDDINGS_MODEL}`" | |
| def build_index(pdf_path: str): | |
| """Load a PDF, split text, build FAISS index with HuggingFaceEmbeddings, then save to disk.""" | |
| if not pdf_path or not os.path.exists(pdf_path): | |
| raise gr.Error("Please upload a valid PDF file.") | |
| loader = PyPDFLoader(pdf_path) | |
| documents = loader.load() | |
| splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=100) | |
| chunks = splitter.split_documents(documents) | |
| emb = get_embeddings() | |
| db = FAISS.from_documents(chunks, emb) | |
| # save/overwrite local index | |
| if os.path.isdir(INDEX_DIR): | |
| shutil.rmtree(INDEX_DIR, ignore_errors=True) | |
| db.save_local(INDEX_DIR) | |
| db_state["db"] = db | |
| db_state["retriever"] = db.as_retriever() | |
| return f"✅ FAISS index built and saved to disk.\n- index: `{INDEX_DIR}`\n- chunks: {len(chunks)}\n- embeddings: `{EMBEDDINGS_MODEL}`" | |
| def make_llm(): | |
| """Create a Transformers pipeline and wrap it as a LangChain LLM.""" | |
| device = 0 if torch.cuda.is_available() else -1 | |
| task = "text2text-generation" if "flan" in GENERATION_MODEL.lower() else "text-generation" | |
| text_gen = pipeline( | |
| task, | |
| model=GENERATION_MODEL, | |
| device=device, | |
| token=HF_TOKEN, | |
| max_new_tokens=MAX_NEW_TOKENS, | |
| temperature=TEMPERATURE, | |
| do_sample=TEMPERATURE > 0, | |
| ) | |
| return HuggingFacePipeline(pipeline=text_gen) | |
| def make_qa_chain(): | |
| """Build the RetrievalQA chain with a custom prompt similar to your Kaggle notebook.""" | |
| if db_state["retriever"] is None: | |
| # Try to load saved index if user forgot to click "Build index" after a restart | |
| msg = try_load_index() | |
| if db_state["retriever"] is None: | |
| raise gr.Error("Please upload a PDF and click 'Build index' first. " + msg) | |
| llm = make_llm() | |
| prompt_template = PromptTemplate( | |
| input_variables=["context", "question"], | |
| template=( | |
| "You are a helpful veterinary assistant. Answer the question using ONLY the information in the context below. " | |
| "Explain the answer in **simple, clear language** that any pet owner can understand. " | |
| "Avoid medical jargon unless absolutely necessary, and if you must use a term, explain it briefly. " | |
| "Keep the answer short and easy to follow (2–4 sentences). '\n\n" | |
| "Context:\n{context}\n\n" | |
| "Question:\n{question}\n\n" | |
| "Simple answer:" | |
| ), | |
| ) | |
| qa_chain = load_qa_chain(llm=llm, chain_type="stuff", prompt=prompt_template) | |
| qa = RetrievalQA( | |
| combine_documents_chain=qa_chain, | |
| retriever=db_state["retriever"], | |
| return_source_documents=True, # keep sources for display | |
| ) | |
| qa_state["qa"] = qa | |
| return "✅ LLM & QA chain are ready." | |
| # ---------- helpers to display what the retriever returns ---------- | |
| def format_sources(docs_with_scores, show_text=True, max_chars=700): | |
| """ | |
| docs_with_scores: list[ (Document, score) ] as returned by similarity_search_with_score | |
| """ | |
| if not docs_with_scores: | |
| return "_No retrieved chunks (check index + model match)._" | |
| blocks = [] | |
| for i, (doc, score) in enumerate(docs_with_scores, 1): | |
| meta = doc.metadata or {} | |
| page = meta.get("page", "N/A") | |
| src = meta.get("source", "") | |
| text = (doc.page_content or "").strip().replace("\n", " ") | |
| if show_text and len(text) > max_chars: | |
| text = text[:max_chars] + "…" | |
| blocks.append( | |
| f"**Hit {i}** — score: `{score:.4f}` — page `{page}` \n" | |
| f"`{src}`" + (f"\n> {text}" if show_text else "") | |
| ) | |
| return "\n\n".join(blocks) | |
| def ask_question(question: str, k: int, show_chunks: bool): | |
| if not question or not question.strip(): | |
| return "Please type a question.", "" | |
| if qa_state["qa"] is None: | |
| raise gr.Error("Click 'Init model' after building/loading the index.") | |
| # 1) Preview: what the index returns (same FAISS db behind the retriever) | |
| # Note: lower score is better for some metrics; FAISS returns distance; we just display raw. | |
| try: | |
| docs_with_scores = db_state["db"].similarity_search_with_score(question, k=k) | |
| except Exception as e: | |
| docs_with_scores = [] | |
| preview = f"_Could not preview retrieved chunks: {e}_" | |
| else: | |
| preview = format_sources(docs_with_scores, show_text=show_chunks) | |
| # 2) Run QA chain (this will also retrieve internally, but we already previewed) | |
| result = qa_state["qa"]({"query": question}) | |
| raw = (result.get("result") or "").strip() | |
| if "Answer:" in raw: | |
| raw = raw.split("Answer:", 1)[-1].strip() | |
| # Combine the final answer + what was retrieved | |
| sources_header = f"### Retrieved from index `{INDEX_DIR}` (k={k}, embeddings=`{EMBEDDINGS_MODEL}`)\n\n" | |
| return raw, sources_header + preview | |
| def reset_states(): | |
| db_state["db"] = None | |
| db_state["retriever"] = None | |
| qa_state["qa"] = None | |
| return "State cleared. Upload a PDF again or rely on the saved index.", "", 3, True | |
| # --------------------------------- | |
| # Gradio UI | |
| # --------------------------------- | |
| with gr.Blocks(title="RAG Khidmat") as demo: | |
| gr.Markdown( | |
| """ | |
| # RAG Khidmat (PDF Q&A) — persistent index | |
| This Space caches the FAISS index on disk so it survives app restarts.<br> | |
| **Flow:** 1) Upload a PDF → 2) **Build index** (saves to disk) → 3) **Init model** → 4) Ask questions. | |
| """ | |
| ) | |
| with gr.Row(): | |
| pdf = gr.File(label="Upload PDF", file_types=[".pdf"], type="filepath") | |
| with gr.Row(): | |
| build_btn = gr.Button("🔨 Build index (save)") | |
| init_btn = gr.Button("⚙️ Init model") | |
| load_btn = gr.Button("📦 Load saved index") | |
| clear_btn = gr.Button("♻️ Reset") | |
| status = gr.Markdown(try_load_index()) | |
| with gr.Row(): | |
| question = gr.Textbox(label="Ask a question", placeholder="Type your question here...", scale=4) | |
| topk = gr.Slider(minimum=1, maximum=10, value=3, step=1, label="Top-K") | |
| show_chunks = gr.Checkbox(value=True, label="Show retrieved chunk text") | |
| ask_btn = gr.Button("🤖 Answer") | |
| answer = gr.Markdown("") | |
| sources = gr.Markdown("") | |
| ask_btn.click( | |
| fn=ask_question, | |
| inputs=[question, topk, show_chunks], | |
| outputs=[answer, sources], | |
| api_name="/ask" # 🔑 Exposes API endpoint: /run/ask | |
| ) | |
| build_btn.click(fn=build_index, inputs=[pdf], outputs=[status]) | |
| init_btn.click(fn=make_qa_chain, inputs=None, outputs=[status]) | |
| load_btn.click(fn=lambda: try_load_index(), inputs=None, outputs=[status]) | |
| # ask_btn.click(fn=ask_question, inputs=[question, topk, show_chunks], outputs=[answer, sources]) | |
| clear_btn.click(fn=reset_states, inputs=None, outputs=[status, answer, topk, show_chunks]) | |
| if __name__ == "__main__": | |
| demo.queue().launch() | |