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, Textbox, Audio, Radio import io from scipy.io import wavfile 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-4", 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): 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) 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 information in the input 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, 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.", "" # Fill the rest of the outputs with empty strings if input_voice is not None: input_text = transcribe_audio(input_voice) # Get a response from GPT-3.5-turbo gpt_subjective, gpt_objective, gpt_assessment, gpt_plan, gpt_general = get_gpt_response(input_text) # Save GPT response to a file gpt_file_path = os.path.join('GPTresponses/', f"{patient_name}.txt") with open(gpt_file_path, "a") as f: f.write(f"Subjective: {gpt_subjective}\nObjective: {gpt_objective}\nAssessment: {gpt_assessment}\nPlan: {gpt_plan}\nGeneral: {gpt_general}\n\n") 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('Docs/', f"{patient_name}.txt") if all(keyword.lower() in soap_response.lower() for keyword in ["subjective:", "objective:", "assessment:", "plan:"]): s_index = soap_response.lower().find('subjective:') o_index = soap_response.lower().find('objective:') a_index = soap_response.lower().find('assessment:') p_index = soap_response.lower().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 = [f"\n\n\u2022 Subjective: {subjective}\n\n\u2022 Objective: {objective}\n\n\u2022 Assessment: {assessment}\n\n\u2022 Plan: {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, f"Subjective: {gpt_subjective}\nObjective: {gpt_objective}\nAssessment: {gpt_assessment}\nPlan: {gpt_plan}", gpt_general, input_text # return the transcribed text and the GPT response from gradio import Interface, Textbox, Audio, Radio from gradio import Interface, Textbox, Audio, Radio interface = Interface( fn=chatbot, inputs=[ Textbox(label="Enter your text"), Audio(source="microphone", type="numpy", label="Speak Something"), ], outputs=[ Textbox(label="SOAP Output"), Textbox(label="General Output"), Textbox(label="GPT SOAP Output"), Textbox(label="GPT General Output"), Textbox(label="Transcribed Text") ], ) index = construct_index('Docs/') interface.launch()