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 | |
class SentimentAnalyzer: | |
def __init__(self): | |
self.model="facebook/bart-large-mnli" | |
openai.api_key=os.getenv("OPENAI_API_KEY") | |
def analyze_sentiment(self, text): | |
pipe = pipeline("zero-shot-classification", model=self.model) | |
label=["positive","negative","neutral"] | |
result = pipe(text, label) | |
sentiment_scores= {result['labels'][0]:result['scores'][0],result['labels'][1]:result['scores'][1],result['labels'][2]:result['scores'][2]} | |
sentiment_scores_str = f"Positive: {sentiment_scores['positive']:.2f}, Neutral: {sentiment_scores['neutral']:.2f}, Negative: {sentiment_scores['negative']:.2f}" | |
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 of the text.\ | |
your are analyze the text and provide the output in the following format: \{emotions: scores\} [with top 3 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","") | |
print(message) | |
return message | |
def analyze_sentiment_for_graph(self, text): | |
pipe = pipeline("zero-shot-classification", model=self.model) | |
label=["positive", "negative", "neutral"] | |
result = pipe(text, label) | |
sentiment_scores = { | |
result['labels'][0]: result['scores'][0], | |
result['labels'][1]: result['scores'][1], | |
result['labels'][2]: result['scores'][2] | |
} | |
return sentiment_scores | |
def emotion_analysis_for_graph(self,text): | |
list_of_emotion=text.split(":") | |
label=list_of_emotion[0] | |
score=list_of_emotion[1] | |
score_dict={ | |
label:float(score) | |
} | |
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"Psycotherapist Bot: {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): | |
labels = sentiment_scores.keys() | |
scores = sentiment_scores.values() | |
fig = px.bar(x=scores, y=labels, orientation='h', color=labels, color_discrete_map={"Negative": "red", "Positive": "green", "Neutral": "gray"}) | |
fig.update_traces(texttemplate='%{x:.2f}%', 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"""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. | |
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> | |
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): | |
customer_score = ", ".join([f"{key}: {value:.2f}" for key, value in customer_sentiment_score.items()]) | |
return f"customer_emotion:{customer_emotion}\nCustomer_sentiment_score:{customer_score}" | |
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=770) | |
customer_emotion_score = sentiment.emotion_analysis_for_graph(customer_emotion) | |
customer_emotion_fig=self._display_graph(customer_emotion_score) | |
customer_emotion_fig.update_layout(title="Emotion Analysis",width=770) | |
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",elem_classes="height") | |
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=0.50, min_width=0): | |
plot =gr.Plot(label="Patient", size=(500, 600)) | |
with gr.Column(scale=0.50, 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() |