SOAPAssist / app.py
AbeerTrial's picture
Update app.py
09d94f9
raw
history blame
7.56 kB
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-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):
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="Transcribed Text"),
Textbox(label="SOAP Output"),
Textbox(label="General Output"),
Textbox(label="GPT SOAP Output"),
Textbox(label="GPT General Output")
],
)
index = construct_index('Docs/')
interface.launch()