import os from gpt_index import SimpleDirectoryReader, GPTSimpleVectorIndex, LLMPredictor, PromptHelper from langchain.chat_models import ChatOpenAI import gradio as gr import speech_recognition as sr import openai import logging import openai from transformers import GPTJForCausalLM, GPT2Tokenizer import numpy as np import soundfile as sf import tempfile import os import boto3 from gradio import Interface, components as gr from gradio import Interface import io from scipy.io import wavfile from google.cloud import speech import pyttsx3 from nltk.tokenize import sent_tokenize import nltk nltk.download('punkt') import langchain.schema print(dir(langchain.schema)) logging.basicConfig(level=logging.INFO) os.environ["OPENAI_API_KEY"] def construct_index(directory_path): max_input_size = 4096 num_outputs = 512 max_chunk_overlap = 20 chunk_size_limit = 2048 prompt_helper = PromptHelper(max_input_size, num_outputs, max_chunk_overlap, chunk_size_limit=chunk_size_limit) llm_predictor = LLMPredictor(llm=ChatOpenAI(temperature=0.7, model_name="gpt-3.5-turbo", max_tokens=num_outputs)) documents = SimpleDirectoryReader(directory_path).load_data() index = GPTSimpleVectorIndex(documents, llm_predictor=llm_predictor, prompt_helper=prompt_helper) index.save_to_disk('index.json') return index def transcribe_audio(audio, service="Google"): sampling_rate, audio_data = audio # unpack the tuple if audio_data.ndim > 1: audio_data = np.mean(audio_data, axis=1) print(type(audio_data), audio_data) fp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) fp.close() text = "" try: sf.write(fp.name, audio_data, sampling_rate) r = sr.Recognizer() with sr.AudioFile(fp.name) as source: audio_data = r.record(source) if service == "Google": try: text = r.recognize_google(audio_data) except sr.RequestError as e: print(f"Could not request results from Google Speech Recognition service; {e}") except sr.UnknownValueError: print("Google Speech Recognition could not understand audio") text = sent_tokenize(text) elif service == "Whisper": try: with open(fp.name, "rb") as audio_file: transcript = openai.Audio.transcribe("whisper-1", audio_file) print(transcript) conversation = [{"role": "user", "content": transcript["text"]}] response = openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=conversation ) print(response) text = transcript["text"] except Exception as e: print("Error with Whisper Service:", str(e)) text = sent_tokenize(text) finally: os.unlink(fp.name) return text def get_gpt_response(input_text): try: # Check that input_text is not empty if not input_text: return "No input provided.", "", "", "", "" conversation = [ {"role": "system", "content": "You are an experienced medical consultant who provides a SOAP note based on the conversation provided."}, {"role": "user", "content": input_text} ] response = openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=conversation ) gpt_response = response['choices'][0]['message']['content'] # Parse the GPT response into SOAP components if all(keyword in gpt_response for keyword in ["Subjective:", "Objective:", "Assessment:", "Plan:"]): s_index = gpt_response.find('Subjective:') o_index = gpt_response.find('Objective:') a_index = gpt_response.find('Assessment:') p_index = gpt_response.find('Plan:') subjective = gpt_response[s_index:o_index].replace('Subjective:', '').strip() objective = gpt_response[o_index:a_index].replace('Objective:', '').strip() assessment = gpt_response[a_index:p_index].replace('Assessment:', '').strip() plan = gpt_response[p_index:].replace('Plan:', '').strip() return subjective, objective, assessment, plan, "" else: return "", "", "", "", gpt_response except Exception as e: print(f"Error in get_gpt_response: {e}") return "", "", "", "", "" def chatbot(input_text, input_voice, transcription_service, patient_name=None): # Check if patient_name is in index index = GPTSimpleVectorIndex.load_from_disk('index.json') if patient_name: # Only do the check if patient_name is not None and not an empty string patient_names = [doc['name'] for doc in index.documents] # Assuming each document is a dictionary with a 'name' field if patient_name and patient_name not in patient_names: return "Patient not found in index.", "", "", "", "", "", "", "", "", "", "", input_text # Fill the rest of the outputs with empty strings if input_voice is not None: input_text = transcribe_audio(input_voice, transcription_service) # Get a response from GPT-3.5-turbo gpt_subjective, gpt_objective, gpt_assessment, gpt_plan, gpt_general = get_gpt_response(input_text) index = GPTSimpleVectorIndex.load_from_disk('index.json') response_index = index.query(input_text, response_mode="compact") soap_response = response_index.response patient_name = soap_response.split(' ')[1] if 'Subjective:' in soap_response else 'General' patient_file_path = os.path.join('/home/user/app/Docs', f"{patient_name}.txt") if all(keyword in soap_response for keyword in ["Subjective:", "Objective:", "Assessment:", "Plan:"]): s_index = soap_response.find('Subjective:') o_index = soap_response.find('Objective:') a_index = soap_response.find('Assessment:') p_index = soap_response.find('Plan:') subjective = soap_response[s_index:o_index].replace('Subjective:', '').strip() objective = soap_response[o_index:a_index].replace('Objective:', '').strip() assessment = soap_response[a_index:p_index].replace('Assessment:', '').strip() plan = soap_response[p_index:].replace('Plan:', '').strip() with open(patient_file_path, "a") as f: f.write(f"Subjective: {subjective}\nObjective: {objective}\nAssessment: {assessment}\nPlan: {plan}\n\n") output = [subjective, objective, assessment, plan, ""] else: with open(patient_file_path, "a" , encoding='utf-8') as f: f.write(f"General: {soap_response}\n\n") output = ["", "", "", "", soap_response] return *output, gpt_subjective, gpt_objective, gpt_assessment, gpt_plan, gpt_general, input_text # return the transcribed text and the GPT response #return *output, gpt_subjective, gpt_objective, gpt_assessment, gpt_plan, output[4] + gpt_general, input_text(this to merge the general (none SOAP) from Index and GPT) from gradio import Interface from gradio.inputs import Textbox, Audio, Radio from gradio.outputs import Textbox interface = Interface( fn=chatbot, inputs=[ Textbox(label="Enter your text"), Audio(source="microphone", type="numpy", label="Speak Something"), Radio(["Google", "Whisper"], label="Choose a transcription service") ], outputs=[ Textbox(label="Subjective"), Textbox(label="Objective"), Textbox(label="Assessment"), Textbox(label="Plan"), Textbox(label="General"), Textbox(label="GPT Subjective"), Textbox(label="GPT Objective"), Textbox(label="GPT Assessment"), Textbox(label="GPT Plan"), Textbox(label="GPT General"), Textbox(label="Transcribed Text"), # window for the transcribed text ], ) index = construct_index('/home/user/app/Docs') interface.launch()