import gradio as gr from huggingface_hub import InferenceClient from typing import List, Tuple import fitz # PyMuPDF from sentence_transformers import SentenceTransformer, util import numpy as np import faiss client = InferenceClient("meta-llama/Meta-Llama-3-8B-Instruct") # Placeholder for the app's state class MyApp: def __init__(self) -> None: self.documents = [] self.embeddings = None self.index = None self.load_pdf("static/" "medical1.pdf") self.build_vector_db() def load_pdf(self, file_path: str) -> None: """Extracts text from a PDF file and stores it in the app's documents.""" doc = fitz.open(file_path) self.documents = [] for page_num in range(len(doc)): page = doc[page_num] text = page.get_text() self.documents.append({"page": page_num + 1, "content": text}) print("PDF processed successfully!") def build_vector_db(self) -> None: """Builds a vector database using the content of the PDF.""" model = SentenceTransformer('all-MiniLM-L6-v2') self.embeddings = model.encode([doc["content"] for doc in self.documents]) self.index = faiss.IndexFlatL2(self.embeddings.shape[1]) self.index.add(np.array(self.embeddings)) print("Vector database built successfully!") def search_documents(self, query: str, k: int = 3) -> List[str]: """Searches for relevant documents using vector similarity.""" model = SentenceTransformer('all-MiniLM-L6-v2') query_embedding = model.encode([query]) D, I = self.index.search(np.array(query_embedding), k) results = [self.documents[i]["content"] for i in I[0]] return results if results else ["No relevant documents found."] app = MyApp() def preprocess_input(user_input: str) -> str: """Preprocesses user input to enhance it for better context.""" if "therapy" in user_input.lower() or "exercise" in user_input.lower(): return "I am looking for guidance on treatment and health issues" return user_input def preprocess_response(response: str) -> str: """Preprocesses the response to make it more polished.""" response = response.strip() response = response.replace("\n\n", "\n") response = response.replace(" ,", ",") response = response.replace(" .", ".") response = " ".join(response.split()) return response def shorten_response(response: str) -> str: """Uses the Zephyr model to shorten and refine the response.""" messages = [{"role": "system", "content": "give a zeroshot clasiification on 'patient', 'student', 'kin and kinds' and give responce acoordingly. use vector data for disease related information."}, {"role": "user", "content": response}] result = client.chat_completion(messages, max_tokens=412, temperature=0.5, top_p=0.9) return result.choices[0].message['content'].strip() def respond(message: str, history: List[Tuple[str, str]]): system_message = "you are a doctor practicing in india and assume all the patients are from northeast India, follow strict rules of NMC india and also a professor so think and answer. Instructions: Chief Complaints and Patient History: chief Complaints(Co): List the chief complaints of the patient(Pt). Patient History(Hpi): Ask questions to gather the patient's history that will help increase the confidence score of the diagnosis. Lab Tests(Lx): Basic Lab Tests: Provide a limited number(up to 4) of the most probable basic laboratory tests for the given case. Order of Tests: List the basic lab tests in the order of their probability and relevance to the chief complaints and patient history. Advanced Lab Tests: For each basic test, indicate which advanced laboratory tests should be ordered if the result is positive or abnormal. Table Format: Present this information in a table with three columns: 'Basic Lab Tests' 'Positive or Abnormal Results' 'Advanced Lab Tests' Differential Diagnosis(Dx): Suggestions: Suggest differential diagnoses with confidence scores for each diagnosis on a scale of 0 to 10. Table Format: Present the differential diagnosis in a table with two columns: 'Differential Diagnosis' 'Confidence Score(0-10)' Treatment Plan(Tx): Structured Plan: Create a structured treatment plan for the provided case. Table Format: Present the plan in a table with the following columns: 'Time Frame'(e.g., 'Day X' to 'Day Y') 'Treatment Goals' 'Intervention Strategies' 'Medication Type' 'Drug Form' 'Drug Name' 'Medication Regimen' 'Duration'(length of each Drug Name usage) 'Contingency Plans" messages = [{"role": "system", "content": system_message}] for val in history: if val[0]: messages.append({"role": "user", "content": val[0]}) if val[1]: messages.append({"role": "assistant", "content": val[1]}) # Preprocess user input preprocessed_message = preprocess_input(message) messages.append({"role": "user", "content": preprocessed_message}) # RAG - Retrieve relevant documents retrieved_docs = app.search_documents(preprocessed_message) context = "\n".join(retrieved_docs) if context.strip(): messages.append({"role": "system", "content": "Relevant documents: " + context}) response = client.chat_completion(messages, max_tokens=600, temperature=0.2, top_p=0.7) response_content = "".join([choice.message['content'] for choice in response.choices if 'content' in choice.message]) polished_response = preprocess_response(response_content) shortened_response = shorten_response(polished_response) history.append((message, shortened_response)) return history, "" with gr.Blocks() as demo: gr.Markdown("# 🧘‍♀️ **Dialectical Behaviour Therapy**") gr.Markdown( "‼️Disclaimer: This chatbot is for educational purpose that is publicly available. " "Use of this chatbot is at your own responsibility." ) chatbot = gr.Chatbot(label="OPD Assistance") with gr.Row(): txt_input = gr.Textbox( show_label=False, placeholder="Type your message here...", lines=1 ) # Bind the Enter key press to the submit function txt_input.submit(fn=respond, inputs=[txt_input, chatbot], outputs=[chatbot, txt_input]) with gr.Row(): submit_btn = gr.Button("Submit", scale=1) refresh_btn = gr.Button("Refresh Chat", scale=1, variant="secondary") example_questions = [ ["What is a cataract, and what type of information is available regarding it?"], ["What is the history and what are the symptoms of truncus arteriosus?"], ["I am going through Jaundice due to Hepatitis E, my stomach has gone too much weak, at time a lot of acidity. I was in hospital for 5 days but now feeling much better then before and my Total bilirubin has come down to 5 and liver enzymes are around 1400. Body is too much fluctuating, at times i feel better and at times i feel pathetic. I am on complete rest, at times i walk around else not much of physical work. I am taking monolive but i can see Liv52 is more famous, should i change that?"], ["My son is 1 yr 8 months old...he is not walking on his own...so dr suggested MRI n MRI report shows .ventricular system appears mildly prominent,normal in shape position and signal morphology,cisternal and sulcal spaces are mildly prominent,small faint hyperintense areas noted in bilateral periventricular frontal and Pareto occipital regions.rests things are normal....pls tell me wat does it means"], ["Age: 56 years old Gender: Male, Presenting Complaints: presents with complaints of worsening shortness of breath associated with a dry cough over the last seven days. He experiences increased dyspnea upon climbing stairs and performing mild physical activities around the house. Additionally, he reports generalized fatigue and occasional dizziness during episodes of severe breathlessness. There is no history of orthopnoea, paroxysmal nocturnal dyspnea, or lower extremity edema."], ["female 35 , sufferings from cold and cough for last one week, mild headache, occasional Fever"], ["What are some distress tolerance skills?"], ] gr.Examples(examples=example_questions, inputs=[txt_input]) submit_btn.click(respond, [txt_input, chatbot], [chatbot, txt_input]) refresh_btn.click(lambda: [], None, chatbot) if __name__ == "__main__": demo.launch(share=True)