File size: 8,086 Bytes
3f871dd
303fc54
 
3f871dd
 
 
 
 
 
 
303fc54
3f871dd
 
303fc54
 
3f871dd
303fc54
 
 
3f871dd
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
303fc54
 
 
 
3f871dd
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
303fc54
3f871dd
90a9f0d
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
import gradio as gr
import torch
import cv2
import speech_recognition as sr
from groq import Groq
import os
import time
import base64
from io import BytesIO
from gtts import gTTS
import tempfile

# Set device
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
print(f"Using device: {device}")

# Clear GPU memory if using GPU
if torch.cuda.is_available():
    torch.cuda.empty_cache()

# Grok API client with API key (stored as environment variable for security)
GROQ_API_KEY = os.getenv("GROQ_API_KEY", "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")  # Replace with your key or use env variable
try:
    client = Groq(api_key=GROQ_API_KEY)
    print("Grok client initialized successfully")
except Exception as e:
    print(f"Error initializing Groq client: {str(e)}")
    raise

# Functions
def predict_text_emotion(text):
    prompt = f"The user has entered text '{text}' classify user's emotion as happy or sad or anxious or angry. Respond in only one word."
    try:
        completion = client.chat.completions.create(
            model="llama-3.2-90b-vision-preview",
            messages=[{"role": "user", "content": prompt}],
            temperature=1,
            max_completion_tokens=64,
            top_p=1,
            stream=False,
            stop=None,
        )
        return completion.choices[0].message.content
    except Exception as e:
        return f"Error with Grok API: {str(e)}"

def transcribe_audio(audio_path):
    r = sr.Recognizer()
    with sr.AudioFile(audio_path) as source:
        audio_text = r.listen(source)
    try:
        text = r.recognize_google(audio_text)
        return text
    except sr.UnknownValueError:
        return "I didn’t catch that—could you try again?"
    except sr.RequestError:
        return "Speech recognition unavailable—try typing instead."

def capture_webcam_frame():
    cap = cv2.VideoCapture(0)
    if not cap.isOpened():
        return None
    start_time = time.time()
    while time.time() - start_time < 2:
        ret, frame = cap.read()
        if ret:
            _, buffer = cv2.imencode('.jpg', frame)
            img_base64 = base64.b64encode(buffer).decode('utf-8')
            img_url = f"data:image/jpeg;base64,{img_base64}"
            cap.release()
            return img_url
    cap.release()
    return None

def detect_facial_emotion():
    img_url = capture_webcam_frame()
    if not img_url:
        return "neutral"
    try:
        completion = client.chat.completions.create(
            model="llama-3.2-90b-vision-preview",
            messages=[
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": "Identify user's facial emotion into happy or sad or anxious or angry. Respond in one word only"},
                        {"type": "image_url", "image_url": {"url": img_url}}
                    ]
                }
            ],
            temperature=1,
            max_completion_tokens=20,
            top_p=1,
            stream=False,
            stop=None,
        )
        emotion = completion.choices[0].message.content.strip().lower()
        if emotion not in ["happy", "sad", "anxious", "angry"]:
            return "neutral"
        return emotion
    except Exception as e:
        print(f"Error with Grok facial detection: {str(e)}")
        return "neutral"

def generate_response(user_input, emotion):
    prompt = f"The user is feeling {emotion}. They said: '{user_input}'. Respond in a friendly caring manner with the user so the user feels being loved."
    try:
        completion = client.chat.completions.create(
            model="llama-3.2-90b-vision-preview",
            messages=[{"role": "user", "content": prompt}],
            temperature=1,
            max_completion_tokens=64,
            top_p=1,
            stream=False,
            stop=None,
        )
        return completion.choices[0].message.content
    except Exception as e:
        return f"Error with Groq API: {str(e)}"

def text_to_speech(text):
    try:
        tts = gTTS(text=text, lang='en', slow=False)
        # Create a temporary file to store the audio
        with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as temp_audio:
            tts.save(temp_audio.name)
            return temp_audio.name
    except Exception as e:
        print(f"Error generating speech: {str(e)}")
        return None

# Chat function for Gradio with voice output
def chat_function(input_type, text_input, audio_input, chat_history):
    if input_type == "text" and text_input:
        user_input = text_input
    elif input_type == "voice" and audio_input:
        user_input = transcribe_audio(audio_input)
    else:
        return chat_history, "Please provide text or voice input.", gr.update(value=text_input), None

    text_emotion = predict_text_emotion(user_input)
    if not chat_history:
        gr.Info("Please look at the camera for emotion detection...")
        facial_emotion = detect_facial_emotion()
    else:
        facial_emotion = "neutral"

    emotions = [e for e in [text_emotion, facial_emotion] if e and e != "neutral"]
    combined_emotion = emotions[0] if emotions else "neutral"

    response = generate_response(user_input, combined_emotion)
    chat_history.append({"role": "user", "content": user_input})
    chat_history.append({"role": "assistant", "content": response})

    audio_output = text_to_speech(response)
    return chat_history, f"Detected Emotion: {combined_emotion}", "", audio_output

# Custom CSS for better styling
css = """
<style>
    .chatbot .message-user {
        background-color: #e3f2fd;
        border-radius: 10px;
        padding: 10px;
        margin: 5px 0;
    }
    .chatbot .message-assistant {
        background-color: #c8e6c9;
        border-radius: 10px;
        padding: 10px;
        margin: 5px 0;
    }
    .input-container {
        padding: 10px;
        background-color: #f9f9f9;
        border-radius: 10px;
        margin-top: 10px;
    }
</style>
"""

# Build the Gradio interface
with gr.Blocks(theme=gr.themes.Soft(), css=css) as app:
    gr.Markdown(
        """
        # Multimodal Mental Health AI Agent
        Chat with our empathetic AI designed to support you by understanding your emotions through text and facial expressions.
        """
    )

    with gr.Row():
        with gr.Column(scale=1):
            emotion_display = gr.Textbox(label="Emotion", interactive=False, placeholder="Detected emotion will appear here")
        
        with gr.Column(scale=3):
            chatbot = gr.Chatbot(label="Conversation History", height=500, type="messages", elem_classes="chatbot")

    with gr.Row(elem_classes="input-container"):
        input_type = gr.Radio(["text", "voice"], label="Input Method", value="text")
        text_input = gr.Textbox(label="Type Your Message", placeholder="How are you feeling today?", visible=True)
        audio_input = gr.Audio(type="filepath", label="Record Your Message", visible=False)
        submit_btn = gr.Button("Send", variant="primary")
        clear_btn = gr.Button("Clear Chat", variant="secondary")
        audio_output = gr.Audio(label="Assistant Response", type="filepath", interactive=False, autoplay=True)

    # Dynamic visibility based on input type
    def update_visibility(input_type):
        return gr.update(visible=input_type == "text"), gr.update(visible=input_type == "voice")

    input_type.change(fn=update_visibility, inputs=input_type, outputs=[text_input, audio_input])

    # Submit action with voice output
    submit_btn.click(
        fn=chat_function,
        inputs=[input_type, text_input, audio_input, chatbot],
        outputs=[chatbot, emotion_display, text_input, audio_output]
    )

    # Clear chat and audio
    clear_btn.click(
        lambda: ([], "", "", None),
        inputs=None,
        outputs=[chatbot, emotion_display, text_input, audio_output]
    )

# Launch the app (for local testing)
if __name__ == "__main__":
    app.launch(server_name="0.0.0.0", server_port=7860)