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("Qwen/Qwen2-1.5b-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 therapy. Can you help me with some exercises or techniques to manage my stress and emotions?" 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": "Shorten and refine this response in bullet list."}, {"role": "user", "content": response}] result = client.chat_completion(messages, max_tokens=512, 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 concisely speaking empathetic Dialectical Behaviour Therapist assistant. You politely guide users through DBT exercises based on the given DBT book. You must say one thing at a time and ask follow-up questions to continue the chat." 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=1024, temperature=0.7, top_p=0.9) 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 based on a DBT exercise book that is publicly available. " "We are not medical practitioners, and the use of this chatbot is at your own responsibility." ) chatbot = gr.Chatbot() with gr.Row(): txt_input = gr.Textbox( show_label=False, placeholder="Type your message here...", lines=1 ) 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 , fufferng fro cold and caugh for last one week, mild headeache, ocassional 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)