Spaces:
Sleeping
Sleeping
| 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 emotion_analysis(self,text): | |
| prompt = f""" Your task is find the top 3 emotion for this converstion {text}: <Sadness, Happiness, Fear, Disgust, Anger> and it's emotion score for the Mental Healthcare Doctor Chatbot and patient conversation text.\ | |
| you 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 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. | |
| """ | |
| response = openai.Completion.create( | |
| model="text-davinci-003", | |
| prompt=prompt, | |
| temperature=0, | |
| 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 setiments for this converstion {text} : <labels = positive, negative, neutral> and it's sentiment score for the Mental Healthcare Doctor Chatbot and patient conversation text.\ | |
| you are analyze the text and provide the output in the following json format heigher to lower order: '''["label1","label2","label3"][score1,score2,score3]''' | |
| """ | |
| response = openai.Completion.create( | |
| model="text-davinci-003", | |
| prompt=prompt, | |
| temperature=0, | |
| 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={"Sentiment": 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='Sentiment', 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,history, text): | |
| try: | |
| history_list = 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_list}'''] | |
| 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) | |
| history[-1][1] = message.strip() | |
| history_state.value = history | |
| return history | |
| except: | |
| history[-1][1] = "How can I help you?" | |
| history_state.value = history | |
| return history | |
| def _text_box(self,customer_emotion,customer_sentiment_score): | |
| sentiment_str = ', '.join([f'{label}: {score}' for label, score in zip(customer_sentiment_score['Sentiment'], customer_sentiment_score['Score'])]) | |
| #emotion_str = ', '.join([f'{emotion}: {score}' for emotion, score in zip(customer_emotion['Emotion'], customer_emotion['Score'])]) | |
| return f"Sentiment: {sentiment_str},\nEmotion: {customer_emotion}" | |
| 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='freddyaboulton/test-blue') as demo: | |
| with gr.Row(): | |
| gr.HTML("""<center class="darkblue" text-align:center;padding:30px;'><center> | |
| <center><h1 class ="center" style="color:#fff">ADOPLE AI</h1></center> | |
| <br><center><h1 style="color:#fff">AI Mental Healthcare ChatBot</h1></center>""") | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| with gr.Row(): | |
| chatbot = gr.Chatbot([], elem_id="chatbot").style(height=360) | |
| with gr.Row(): | |
| with gr.Column(scale=0.90): | |
| txt = gr.Textbox(show_label=False,placeholder="Patient").style(container=False) | |
| with gr.Column(scale=0.10): | |
| emptyBtn = gr.Button("🧹 Clear") | |
| with gr.Accordion("Conversational AI Analytics", open = False): | |
| 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]).then( | |
| self._suggested_answer, [chatbot,txt],chatbot) | |
| txt_msg.then(lambda: gr.update(interactive=True), None, [txt]) | |
| # txt.submit(self._suggested_answer, [chatbot,txt],chatbot) | |
| # 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() |