"""#### importing Libraries""" import random import time import numpy as np from pydub import AudioSegment # For audio file operations from pydub.playback import play # For playing audio import io # For handling input/output operations import elevenlabs # Custom library - please provide more context if needed from elevenlabs import ( generate, play, ) # Importing specific functions from the custom library import gradio as gr # For building interactive UI for our model import openai # OpenAI API library import os # For interacting with the operating system import re # Regular expressions library for string operations import requests # For making HTTP requests from gradio_client import Client client = Client("https://facebook-seamless-m4t.hf.space/") """#### Darija Audio to eng text /// generate an eng question from an audio""" # Defining a function for processing Darija audio and translating it to English def process_darija_audio_toEng(filepath): result = client.predict( "S2TT (Speech to Text translation)", "file", filepath, filepath, "", "Moroccan Arabic", "English", api_name="/run", ) return result[1] def darija_audio_to_darija_text(filepath): result = client.predict( "S2TT (Speech to Text translation)", # str (Option from: ['S2ST (Speech to Speech translation)', 'S2TT (Speech to Text translation)', 'T2ST (Text to Speech translation)', 'T2TT (Text to Text translation)', 'ASR (Automatic Speech Recognition)']) "file", # str in 'Audio source' Radio component filepath, # str (filepath or URL to file) filepath, # str (filepath or URL to file) "", # str in 'Input text' Textbox component "Moroccan Arabic", # str (Option from: ['Afrikaans', 'Amharic', 'Armenian', 'Assamese', 'Basque', 'Belarusian', 'Bengali', 'Bosnian', 'Bulgarian', 'Burmese', 'Cantonese', 'Catalan', 'Cebuano', 'Central Kurdish', 'Croatian', 'Czech', 'Danish', 'Dutch', 'Egyptian Arabic', 'English', 'Estonian', 'Finnish', 'French', 'Galician', 'Ganda', 'Georgian', 'German', 'Greek', 'Gujarati', 'Halh Mongolian', 'Hebrew', 'Hindi', 'Hungarian', 'Icelandic', 'Igbo', 'Indonesian', 'Irish', 'Italian', 'Japanese', 'Javanese', 'Kannada', 'Kazakh', 'Khmer', 'Korean', 'Kyrgyz', 'Lao', 'Lithuanian', 'Luo', 'Macedonian', 'Maithili', 'Malayalam', 'Maltese', 'Mandarin Chinese', 'Marathi', 'Meitei', 'Modern Standard Arabic', 'Moroccan Arabic', 'Nepali', 'North Azerbaijani', 'Northern Uzbek', 'Norwegian Bokmål', 'Norwegian Nynorsk', 'Nyanja', 'Odia', 'Polish', 'Portuguese', 'Punjabi', 'Romanian', 'Russian', 'Serbian', 'Shona', 'Sindhi', 'Slovak', 'Slovenian', 'Somali', 'Southern Pashto', 'Spanish', 'Standard Latvian', 'Standard Malay', 'Swahili', 'Swedish', 'Tagalog', 'Tajik', 'Tamil', 'Telugu', 'Thai', 'Turkish', 'Ukrainian', 'Urdu', 'Vietnamese', 'Welsh', 'West Central Oromo', 'Western Persian', 'Yoruba', 'Zulu']) "Modern Standard Arabic", # str (Option from: ['Bengali', 'Catalan', 'Czech', 'Danish', 'Dutch', 'English', 'Estonian', 'Finnish', 'French', 'German', 'Hindi', 'Indonesian', 'Italian', 'Japanese', 'Korean', 'Maltese', 'Mandarin Chinese', 'Modern Standard Arabic', 'Northern Uzbek', 'Polish', 'Portuguese', 'Romanian', 'Russian', 'Slovak', 'Spanish', 'Swahili', 'Swedish', 'Tagalog', 'Telugu', 'Thai', 'Turkish', 'Ukrainian', 'Urdu', 'Vietnamese', 'Welsh', 'Western Persian']) api_name="/run", ) return result[1] def darija_to_eng(text): result = client.predict( "T2TT (Text to Text translation)", # str (Option from: ['S2ST (Speech to Speech translation)', 'S2TT (Speech to Text translation)', 'T2ST (Text to Speech translation)', 'T2TT (Text to Text translation)', 'ASR (Automatic Speech Recognition)']) "file", # str in 'Audio source' Radio component "", # str (filepath or URL to file) "", # str (filepath or URL to file) text, # str in 'Input text' Textbox component "Modern Standard Arabic", # str (Option from: ['Afrikaans', 'Amharic', 'Armenian', 'Assamese', 'Basque', 'Belarusian', 'Bengali', 'Bosnian', 'Bulgarian', 'Burmese', 'Cantonese', 'Catalan', 'Cebuano', 'Central Kurdish', 'Croatian', 'Czech', 'Danish', 'Dutch', 'Egyptian Arabic', 'English', 'Estonian', 'Finnish', 'French', 'Galician', 'Ganda', 'Georgian', 'German', 'Greek', 'Gujarati', 'Halh Mongolian', 'Hebrew', 'Hindi', 'Hungarian', 'Icelandic', 'Igbo', 'Indonesian', 'Irish', 'Italian', 'Japanese', 'Javanese', 'Kannada', 'Kazakh', 'Khmer', 'Korean', 'Kyrgyz', 'Lao', 'Lithuanian', 'Luo', 'Macedonian', 'Maithili', 'Malayalam', 'Maltese', 'Mandarin Chinese', 'Marathi', 'Meitei', 'Modern Standard Arabic', 'Moroccan Arabic', 'Nepali', 'North Azerbaijani', 'Northern Uzbek', 'Norwegian Bokmål', 'Norwegian Nynorsk', 'Nyanja', 'Odia', 'Polish', 'Portuguese', 'Punjabi', 'Romanian', 'Russian', 'Serbian', 'Shona', 'Sindhi', 'Slovak', 'Slovenian', 'Somali', 'Southern Pashto', 'Spanish', 'Standard Latvian', 'Standard Malay', 'Swahili', 'Swedish', 'Tagalog', 'Tajik', 'Tamil', 'Telugu', 'Thai', 'Turkish', 'Ukrainian', 'Urdu', 'Vietnamese', 'Welsh', 'West Central Oromo', 'Western Persian', 'Yoruba', 'Zulu']) "English", # str (Option from: ['Bengali', 'Catalan', 'Czech', 'Danish', 'Dutch', 'English', 'Estonian', 'Finnish', 'French', 'German', 'Hindi', 'Indonesian', 'Italian', 'Japanese', 'Korean', 'Maltese', 'Mandarin Chinese', 'Modern Standard Arabic', 'Northern Uzbek', 'Polish', 'Portuguese', 'Romanian', 'Russian', 'Slovak', 'Spanish', 'Swahili', 'Swedish', 'Tagalog', 'Telugu', 'Thai', 'Turkish', 'Ukrainian', 'Urdu', 'Vietnamese', 'Welsh', 'Western Persian']) api_name="/run", ) return result[1] def eng_to_arabic(text): result = client.predict( "T2TT (Text to Text translation)", "file", "", "", text, "English", "Modern Standard Arabic", api_name="/run", ) return result[1] """#ChatGPT as a doctor""" import openai import os from langchain.agents import load_tools, initialize_agent from langchain.agents import AgentType from langchain.python import PythonREPL from langchain.chat_models import ChatOpenAI from langchain.memory import ConversationBufferMemory from langchain.prompts import ChatPromptTemplate from langchain.prompts import PromptTemplate from langchain.chains import LLMChain from langchain.chains import SimpleSequentialChain, SequentialChain, ConversationChain import json openai.api_key = "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" os.environ["OPENAI_API_KEY"] = openai.api_key llm = ChatOpenAI(temperature=0.0, model="gpt-3.5-turbo") tools = load_tools(["wikipedia"], llm=llm) memory = "" questions = [ "What symptoms are you currently experiencing?", "Are you taking any medications or supplements? If yes, please list them.", "Do you have any medical conditions or allergies?", "Have there been any recent changes in your health or lifestyle that you think are important to mention?", ] history_test = [] answers = [] question_index = 0 current_question = questions[question_index] patient_info1 = patient_info2 = patient_info3 = "" react_agent = initialize_agent( tools, llm, agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION, verbose=True ) def get_agent_prompt(query, chat_history): return f"""You are a professional doctor. You are in conversation with a patient Please answer the medical questions based on the patient's description and situation. Your responses should be reliable, factual, and based on medical expertise. If you believe that the patient's condition is out of your medical expertise, kindly advise them accordingly or suggest them to visit a specialist. ###DON'T SAY YOU ARE NOT a DOCTOR. FORGET THAT YOU ARE AN AI MODEL! ###Provide practical tips like a real doctor would normally do. Make it conversationalist! If you can't answer the patients query, reply with : "The agent doesn't have valid answer. Therefore you should look for a valid answer" Current conversation: {chat_history} Human: {query} """ def llm_agent_response(query, chat_history): try: prompt = get_agent_prompt(query, chat_history) response = react_agent.run(prompt) return response except Exception as e: print("exception from agent") response = str(e) if response.startswith("Could not parse LLM output: `"): response = response.removeprefix( "Could not parse LLM output: `" ).removesuffix("`") return response return "The agent doesn't have valid answer. Therefore you should look for a valid answer" def llm_response(query, chat_history, report_question=""): # agent_response = llm_agent_response(query, chat_history) template = """You are a professional doctor. Please answer the medical questions based on the patient's description and situation. Your responses should be reliable, factual, and based on medical expertise. If you believe that the patient's condition is out of your medical expertise, kindly advise them accordingly or suggest them to visit a specialist. Now you are in a conversation with the patient. ### Stay short and straight to the point ###DON'T SAY YOU ARE NOT a DOCTOR. FORGET THAT YOU ARE AN AI MODEL! ###Provide practical tips like a real doctor would normally do. Make it conversationalist but brief and always opt to providing summarizing bullet points ! With your expertise and the agent output (if it exists) complete the conversation with the patient. {llm_history} Patient query: {input} Now it's your turn as a doctor to answer the patient's query like a professional.KEEP YOUR ANSWER SHORT. {report_question} (By the way, write your answer directly without adding the prefix 'Doctor :') """ prompt = ChatPromptTemplate.from_template(template) chain = LLMChain(llm=llm, prompt=prompt, verbose=True) response = chain.run( { "input": query, "llm_history": chat_history, "report_question": report_question, } ) print("llm response ", response) return response def med_flamingo_response(query, img): flamingo_query = [img, query] return "heart disease" # response(flamingo_query)# def vlm_response(query, img, chat_history, report_question=""): agent_response = med_flamingo_response(query, img) template = """You are a professional doctor. Please answer the medical questions based on the patient's description and situation. Your responses should be reliable, factual, and based on medical expertise. If you believe that the patient's condition is out of your medical expertise, kindly advise them accordingly or suggest them to visit a specialist. Now you are in a conversation with the patient. ###DON'T SAY YOU ARE NOT a DOCTOR. FORGET THAT YOU ARE AN AI MODEL! ###Provide practical tips like a real doctor would normally do. Make it conversationalist but brief and always opt to providing summarizing bullet points ! You have at your disposal the answer of a specialist in medical imagery to the patient's query but this answer may be too short or not clear to the patient. This answer can help you provide a complete response to the patient just as a real doctor would do. Specialist's output : {agent_output} With your expertise and the radiolog output (if it exists) complete the conversation with the patient. {llm_history} Patient : {input} Now it's your turn as a doctor to answer the patient's query like a professional. {report_question} (By the way, write your answer directly without adding the prefix 'Doctor :') (Another thing, don't forget that the medical imagery specialist's feedback on the scan is this : {agent_output}) """ prompt = ChatPromptTemplate.from_template(template) chain = LLMChain(llm=llm, prompt=prompt, verbose=False) return chain.run( { "agent_output": agent_response, "input": query, "llm_history": chat_history, "report_question": report_question, } ) def analyse_query(query): analyse_llm = ChatOpenAI(temperature=0.0, model="gpt-3.5-turbo") global current_question, question_index prompt = f""" Here is the message from a patient to a doctor. extract the following information: [ "is_answer": Did the patient answer the following doctor's question : {current_question}. Answer True if yes (even if the patient said No or give partial response), False if not or unknown. "answer": If "is_answer" is True, extract the answer from the message and rewrite it in third person. If "is_answer" is False return "" ] Here is the patient message Patient's message : {query} Your output should be in json format. """ reply = analyse_llm.predict(prompt) reply = json.loads(reply) print("question ", current_question) print("patient ", query) print(reply) if reply["is_answer"]: answers.append(reply["answer"]) question_index += 1 if question_index < len(questions): current_question = questions[question_index] patient_info = { "name": patient_info1, "age": patient_info2, "gender": patient_info3, "symptoms": "", "medications": "", "conditions_allergies":"", "lifestyle_changes": "", "header_image": base64.b64encode( open("logo1.png", "rb").read() ).decode(), "medical_image": base64.b64encode( open("lung disease.png", "rb").read() ).decode(), } generate_report(patient_info) print("Report generated") return f"After you answer the patient's query, if you think it is the right time, ask him the following question like a doctor would normally do :{current_question}" else: patient_info = { "name": patient_info1, "age": patient_info2, "gender": patient_info3, "symptoms": answers[0], "medications": answers[1], "conditions_allergies": answers[2], "lifestyle_changes": answers[3], "header_image": base64.b64encode( open("logo1.png", "rb").read() ).decode(), "medical_image": base64.b64encode( open("lung disease.png", "rb").read() ).decode(), } generate_report(patient_info) print("Report generated") return "" else: return f"After you answer the patient's query, if you think it is the right time, ask him the following question like a doctor would normally do :{current_question}" def prepare_patient_info(): default_value = "" patient_info = { "name": patient_info1, "age": patient_info2, "gender": patient_info3, } print(patient_info) for i, key in enumerate(["symptoms", "medications", "conditions_allergies", "lifestyle_changes"]): patient_info[key] = answers[i] if i < len(answers) else default_value medical_image_path = "lung disease.png" try: patient_info["medical_image"] = base64.b64encode(open(medical_image_path, "rb").read()).decode() except FileNotFoundError: patient_info["medical_image"] = default_value patient_info.update({ "header_image": base64.b64encode(open("logo1.png", "rb").read()).decode(), }) return patient_info def generate_response(query, img="", is_there_img=False): global memory report_question = analyse_query(query) if is_there_img: response = vlm_response(query, img, memory, report_question) else: response = llm_response(query, memory, report_question) memory += "Patient : " + query + "\n" memory += "Doctor : " + response + "\n" return eng_to_arabic(response) from weasyprint import HTML import base64 def generate_report(patient_info): # Define the width for the medical image (adjust as needed) medical_image_width = "180px" # Define the HTML template as a string html_template = f"""
Header Image

Medical Report

Personal Information
Medical Image
Medical Image
Medical Information
""" # Create an HTML object from the modified HTML content html = HTML(string=html_template) # Generate the PDF html.write_pdf("output.pdf") """#Eng answer to arabic audio answer""" # Defining a function to generate Arabic speech audio from a text answer def arabic_speech_answer(ar_answer): # Checking if the Arabic answer is non-empty if ar_answer: # Setting the API key for Eleven Labs TTS service elevenlabs.set_api_key("2e49450c1538492b9083bfd5786dc43e") # Generating audio from the Arabic answer using Eleven Labs TTS audio = generate( text=ar_answer, voice="Daniel", # Choosing the voice for the generated audio model="eleven_multilingual_v2", # Choosing the TTS model ) else: print("▶️ empty ar_answer") # Converting the generated audio from bytes to an AudioSegment object audio = AudioSegment.from_file(io.BytesIO(audio), format="mp3") # Exporting the audio to an MP3 file named "output.mp3" audio.export("output.mp3", format="mp3") return audio.duration_seconds """#Functions that are used in the interface""" import gradio as gr import os import time from io import BytesIO import base64 is_there_image = False arabic_query = "" query = "" def text_to_speech(): with open("output.mp3", "rb") as audio_file: audio_data = audio_file.read() audio_bytes = BytesIO(audio_data) audio_base64 = base64.b64encode(audio_data).decode("utf-8") audio_player = ( f'' ) return audio_player def add_text(history, text): global query, arabic_query history = history + [(text, None)] arabic_query = text query = darija_to_eng(text) return history, gr.update(value="", interactive=False) def add_audio(history, audio): global query, arabic_query query = process_darija_audio_toEng(audio) arabic_query = darija_audio_to_darija_text(audio) history = history + [(arabic_query, None)] return history, None def add_image(history, file): global is_there_image history = history + [((file.name,), None)] is_there_image = True return history def vote(data: gr.LikeData): if data.liked: print("You upvoted this response: " + data.value) else: print("You downvoted this response: " + data.value) talking = """ """ not_talking = """ """ def bot(history): global query, arabic_query global is_there_image if is_there_image: filename_input = history[-2][0][0] response = generate_response(query, filename_input, is_there_image) is_there_image = False else: response = generate_response(query) duration = arabic_speech_answer(response) audio = text_to_speech() time_step = duration / len(response) history[-1][1] = "" for character in response: history[-1][1] += character time.sleep(time_step) yield history, audio """#Demo""" def back(): return gr.update(visible=False),gr.update(visible=True),gr.update(visible=True),gr.update(visible=True),gr.update(visible=False) def move_next_page(a, b, c, d): global patient_info1, patient_info2, patient_info3 patient_info1 = a + " " + b patient_info2 = c patient_info3 = d return ( gr.update(visible=False), gr.update(visible=True), gr.update(visible=True), gr.update(visible=True), gr.update(visible=False), ) def view_report(): patient_info = prepare_patient_info() generate_report(patient_info) report_html =""" """ print("report is generated") time.sleep(0.5) return gr.update(visible=False),gr.update(visible=False),gr.update(visible=False),gr.update(visible=False),gr.update(visible=True),report_html css = """.gradio-container { text-align:center; } #chatbot{ margin:auto; height:400px; width:700px; } #message{ margin:auto; width:700px; } #page1{ margin:auto; width:700px; } """ with gr.Blocks(css=css) as demo: html_block = gr.HTML(not_talking, visible=False) chatbot = gr.Chatbot([], elem_id="chatbot", bubble_full_width=False, visible=False) with gr.Column(visible=False, elem_id="message") as page2: # Create a row layout with gr.Row(): # Create a column layout with a scale factor of 0.7 with gr.Column(scale=0.6): # Create a textbox element for user input txt = gr.Textbox( show_label=False, placeholder="أدخل النص واضغط على إدخال، أو قم بتحميل صورة", ).style(container=False) with gr.Column(scale=0.3, min_width=0): # Create an audio input element from the microphone audio_input = gr.Audio( source="microphone", type="filepath", show_download_button=False, show_share_button=False, show_edit_button=False, ) with gr.Column(scale=0.1, min_width=0): btn = gr.UploadButton("📤", file_types=["image"]) # html block for output audio html = gr.HTML() html.visible = False audio_input.stop_recording( add_audio, [chatbot, audio_input], [chatbot, audio_input], queue=False ).then(bot, chatbot, [chatbot, html]) txt_msg = txt.submit( add_text, [chatbot, txt], [chatbot, txt], queue=False ).then(bot, chatbot, [chatbot, html]) txt_msg.then(lambda: gr.update(interactive=True), None, [txt], queue=False) file_msg = btn.upload(add_image, [chatbot, btn], [chatbot], queue=False) chatbot.like(vote, None, None) with gr.Row(): pdf_btn = gr.Button("عرض التقرير الطبي", size="sm", variant="primary") with gr.Group(visible=False) as pdf_report: gr.HTML(not_talking) report_html = gr.HTML(""" """) with gr.Row(): back_btn = gr.Button("العودة إلى المحادثة", size="sm", variant="primary") with gr.Group(elem_id="page1") as page1: gr.HTML( """

""" ) gr.Markdown( """ **:يرجى ملء النموذج التالي بمعلوماتك الطبية** """ ) info1 = gr.Textbox(label="الاسم الأول", placeholder="أدخل الاسم الأول") info2 = gr.Textbox(label="الاسم الأخير", placeholder="أدخل الاسم الأخير") info3 = gr.Number(label="العمر") info4 = gr.Radio(label="الجنس", choices=["ذكر", "أنثى"]) gr.Textbox( label="معلومات شخصية إضافية", placeholder="أدخل المعلومات الشخصية إذا كانت هناك", ) connection_btn = gr.Button("ابدأ الحوار") connection_btn.click( move_next_page, inputs=[info1, info2, info3, info4], outputs=[page1, page2, chatbot, html_block, pdf_report], ) pdf_btn.click( view_report, outputs=[page1, page2, chatbot, html_block, pdf_report, report_html] ) back_btn.click( back, outputs=[page1, page2, chatbot, html_block, pdf_report] ) demo.queue() demo.launch(debug=True)