Spaces:
Runtime error
Runtime error
from pydantic import NoneStr | |
import os | |
import mimetypes | |
import validators | |
import requests | |
import tempfile | |
import gradio as gr | |
import openai | |
import re | |
import json | |
from transformers import pipeline | |
import matplotlib.pyplot as plt | |
import plotly.express as px | |
import pandas as pd | |
class SentimentAnalyzer: | |
def __init__(self): | |
# self.model="facebook/bart-large-mnli" | |
openai.api_key=os.getenv("OPENAI_API_KEY") | |
def analyze_sentiment(self, text): | |
prompt = f""" Your task is find the top 3 setiments : <labels = positive, negative, neutral> and it's sentiment score for the Mental Healthcare Doctor Chatbot and patient conversation text.\ | |
your are analyze the text and provide the output in the following json order: \"\"\" | |
<\{result['labels'][0]: result['scores'][0], result['labels'][1]: result['scores'][1], result['labels'][2]: result['scores'][2] \}>\"\"\" \ | |
analyze the text : '''{text}''' | |
""" | |
response = openai.Completion.create( | |
model="text-davinci-003", | |
prompt=prompt, | |
temperature=1, | |
max_tokens=60, | |
top_p=1, | |
frequency_penalty=0, | |
presence_penalty=0 | |
) | |
# Extract the generated text | |
sentiment_scores_str = response.choices[0].text.strip() | |
print(sentiment_scores_str) | |
return sentiment_scores_str | |
def emotion_analysis(self,text): | |
prompt = f""" Your task is find the top 3 emotion : <Sadness, Happiness, Joy, Fear, Disgust, Anger> and it's emotion score for the Mental Healthcare Doctor Chatbot and patient conversation text.\ | |
your are analyze the text and provide the output in the following list format heigher to lower order: ["emotion1","emotion2","emotion3"][score1,score2,score3]''' [with top 1 result having the highest score] | |
The scores should be in the range of 0.0 to 1.0, where 1.0 represents the highest intensity of the emotion.\ | |
analyze the text : '''{text}''' | |
""" | |
response = openai.Completion.create( | |
model="text-davinci-003", | |
prompt=prompt, | |
temperature=1, | |
max_tokens=60, | |
top_p=1, | |
frequency_penalty=0, | |
presence_penalty=0 | |
) | |
message = response.choices[0].text.strip().replace("\n","") | |
return message | |
def analyze_sentiment_for_graph(self, text): | |
prompt = f""" Your task is find the top 3 setiments : <labels = positive, negative, neutral> and it's sentiment score for the Mental Healthcare Doctor Chatbot and patient conversation text.\ | |
your are analyze the text and provide the output in the following json format heigher to lower order: '''["label1","label2","label3"][score1,score2,score3]''' \ | |
analyze the text : '''{text}''' | |
""" | |
response = openai.Completion.create( | |
model="text-davinci-003", | |
prompt=prompt, | |
temperature=1, | |
max_tokens=60, | |
top_p=1, | |
frequency_penalty=0, | |
presence_penalty=0 | |
) | |
# Extract the generated text | |
sentiment_scores = response.choices[0].text.strip() | |
start_index = sentiment_scores.find("[") | |
end_index = sentiment_scores.find("]") | |
list1_text = sentiment_scores[start_index + 1: end_index] | |
list2_text = sentiment_scores[end_index + 2:-1] | |
sentiment = list(map(str.strip, list1_text.split(","))) | |
scores = list(map(float, list2_text.split(","))) | |
score_dict={"Labels": sentiment, "Score": scores} | |
print(score_dict) | |
return score_dict | |
def emotion_analysis_for_graph(self,text): | |
start_index = text.find("[") | |
end_index = text.find("]") | |
list1_text = text[start_index + 1: end_index] | |
list2_text = text[end_index + 2:-1] | |
emotions = list(map(str.strip, list1_text.split(","))) | |
scores = list(map(float, list2_text.split(","))) | |
score_dict={"Emotion": emotions, "Score": scores} | |
print(score_dict) | |
return score_dict | |
class Summarizer: | |
def __init__(self): | |
openai.api_key=os.getenv("OPENAI_API_KEY") | |
def generate_summary(self, text): | |
model_engine = "text-davinci-003" | |
prompt = f"""summarize the following conversation delimited by triple backticks. write within 30 words.```{text}``` """ | |
completions = openai.Completion.create( | |
engine=model_engine, | |
prompt=prompt, | |
max_tokens=60, | |
n=1, | |
stop=None, | |
temperature=0.5, | |
) | |
message = completions.choices[0].text.strip() | |
return message | |
history_state = gr.State() | |
summarizer = Summarizer() | |
sentiment = SentimentAnalyzer() | |
class LangChain_Document_QA: | |
def __init__(self): | |
openai.api_key=os.getenv("OPENAI_API_KEY") | |
def _add_text(self,history, text): | |
history = history + [(text, None)] | |
history_state.value = history | |
return history,gr.update(value="", interactive=False) | |
def _agent_text(self,history, text): | |
response = text | |
history[-1][1] = response | |
history_state.value = history | |
return history | |
def _chat_history(self): | |
history = history_state.value | |
formatted_history = " " | |
for entry in history: | |
customer_text, agent_text = entry | |
formatted_history += f"Patient: {customer_text}\n" | |
if agent_text: | |
formatted_history += f"Mental Healthcare Doctor Chatbot: {agent_text}\n" | |
return formatted_history | |
def _display_history(self): | |
formatted_history=self._chat_history() | |
summary=summarizer.generate_summary(formatted_history) | |
return summary | |
def _display_graph(self,sentiment_scores): | |
df = pd.DataFrame(sentiment_scores) | |
fig = px.bar(df, x='Score', y='Labels', orientation='h', labels={'Score': 'Score', 'Labels': 'Sentiment'}) | |
fig.update_layout(height=500, width=200) | |
return fig | |
def _display_graph_emotion(self,customer_emotion_score): | |
fig = px.pie(customer_emotion_score, values='Score', names='Emotion', title='Emotion Distribution', hover_data=['Score']) | |
#fig.update_traces(texttemplate='Emotion', textposition='outside') | |
fig.update_layout(height=500, width=200) | |
return fig | |
def _history_of_chat(self): | |
history = history_state.value | |
formatted_history = "" | |
client="" | |
agent="" | |
for entry in history: | |
customer_text, agent_text = entry | |
client+=customer_text | |
formatted_history += f"Patient: {customer_text}\n" | |
if agent_text: | |
agent+=agent_text | |
formatted_history += f"Mental Healthcare Doctor Chatbot: {agent_text}\n" | |
return client,agent | |
def _suggested_answer(self,text): | |
try: | |
history = self._chat_history() | |
try: | |
file_path = "patient_details.json" | |
with open(file_path) as file: | |
patient_details = json.load(file) | |
except: | |
pass | |
prompt = f"""Analyse the patient json If asked for information take it from {patient_details} \ | |
you first get patient details : <get name,age,gender,contact,address from patient> if not match patient json information start new chat else match patient \ | |
json information ask previous: <description,symptoms,diagnosis,treatment talk about patient> As an empathic AI Mental Healthcare Doctor Chatbot, provide effective solutions to patients' mental health concerns. \ | |
first start the conversation ask existing patient or new patient. if new patient get name,age,gender,contact,address from the patient and start. \ | |
if existing customer get name,age,gender,contact,address details and start the chat about existing issues and current issues. \ | |
if patient say thanking tone message to end the conversation with a thanking greeting when the patient expresses gratitude. \ | |
Chat History:['''{history}'''] | |
Patient: ['''{text}'''] | |
Perform as Mental Healthcare Doctor Chatbot | |
""" | |
response = openai.Completion.create( | |
model="text-davinci-003", | |
prompt=prompt, | |
temperature=0, | |
max_tokens=500, | |
top_p=1, | |
frequency_penalty=0, | |
presence_penalty=0.6, | |
) | |
message = response.choices[0].text.strip() | |
if ":" in message: | |
message = re.sub(r'^.*:', '', message) | |
return message.strip() | |
except: | |
return "How can I help you?" | |
def _text_box(self,customer_emotion,customer_sentiment_score): | |
sentiment_data = customer_sentiment_score | |
emotion_data = customer_emotion | |
sentiment_str = ', '.join([f'{label}: {score}' for label, score in zip(sentiment_data['Labels'], sentiment_data['Score'])]) | |
emotion_str = ', '.join([f'{emotion}: {score}' for emotion, score in zip(emotion_data['Emotion'], emotion_data['Score'])]) | |
return f"Sentiment: {sentiment_str}, Emotion: {emotion_str}" | |
def _on_sentiment_btn_click(self): | |
client=self._history_of_chat() | |
customer_emotion=sentiment.emotion_analysis(client) | |
customer_sentiment_score = sentiment.analyze_sentiment_for_graph(client) | |
scores=self._text_box(customer_emotion,customer_sentiment_score) | |
customer_fig=self._display_graph(customer_sentiment_score) | |
customer_fig.update_layout(title="Sentiment Analysis",width=800) | |
customer_emotion_score = sentiment.emotion_analysis_for_graph(customer_emotion) | |
customer_emotion_fig=self._display_graph_emotion(customer_emotion_score) | |
customer_emotion_fig.update_layout(title="Emotion Analysis",width=800) | |
return scores,customer_fig,customer_emotion_fig | |
def clear_func(self): | |
history_state.clear() | |
def gradio_interface(self): | |
with gr.Blocks(css="style.css",theme=gr.themes.Glass()) as demo: | |
with gr.Row(): | |
gr.HTML("""<center><img class="image" src="https://www.syrahealth.com/images/SyraHealth_Logo_Dark.svg" alt="Image" width="210" height="210"></center> | |
""") | |
with gr.Row(): | |
gr.HTML("""<center><h1>AI Mental Healthcare ChatBot</h1></center>""") | |
chatbot = gr.Chatbot([], elem_id="chatbot").style(height=360) | |
with gr.Row(): | |
with gr.Column(scale=0.8): | |
txt = gr.Textbox( | |
show_label=False, | |
placeholder="Patient").style(container=False) | |
with gr.Column(scale=0.2): | |
emptyBtn = gr.Button("🧹 Clear") | |
with gr.Row(): | |
with gr.Column(scale=0.80): | |
txt3 =gr.Textbox( | |
show_label=False, | |
placeholder="AI Healthcare Suggesstion").style(container=False) | |
with gr.Column(scale=0.20, min_width=0): | |
button=gr.Button(value="🚀send") | |
with gr.Row(): | |
with gr.Column(scale=0.50): | |
txt4 =gr.Textbox( | |
show_label=False, | |
lines=4, | |
placeholder="Summary").style(container=False) | |
with gr.Column(scale=0.50): | |
txt5 =gr.Textbox( | |
show_label=False, | |
lines=4, | |
placeholder="Sentiment").style(container=False) | |
with gr.Row(): | |
with gr.Column(scale=0.50, min_width=0): | |
end_btn=gr.Button(value="End") | |
with gr.Column(scale=0.50, min_width=0): | |
Sentiment_btn=gr.Button(value="📊",callback=self._on_sentiment_btn_click) | |
with gr.Row(): | |
gr.HTML("""<center><h1>Sentiment and Emotion Score Graph</h1></center>""") | |
with gr.Row(): | |
with gr.Column(scale=1, min_width=0): | |
plot =gr.Plot(label="Patient", size=(500, 600)) | |
with gr.Row(): | |
with gr.Column(scale=1, min_width=0): | |
plot_3 =gr.Plot(label="Patient_Emotion", size=(500, 600)) | |
txt_msg = txt.submit(self._add_text, [chatbot, txt], [chatbot, txt]) | |
txt_msg.then(lambda: gr.update(interactive=True), None, [txt]) | |
txt.submit(self._suggested_answer,txt,txt3) | |
button.click(self._agent_text, [chatbot,txt3], chatbot) | |
end_btn.click(self._display_history, [], txt4) | |
emptyBtn.click(self.clear_func,[],[]) | |
emptyBtn.click(lambda: None, None, chatbot, queue=False) | |
Sentiment_btn.click(self._on_sentiment_btn_click,[],[txt5,plot,plot_3]) | |
demo.title = "AI Mental Healthcare ChatBot" | |
demo.launch() | |
document_qa =LangChain_Document_QA() | |
document_qa.gradio_interface() |