Spaces:
Sleeping
Sleeping
| import torch | |
| import gradio as gr | |
| import torch.nn.functional as F | |
| from transformers import AutoModelForSequenceClassification, AutoTokenizer | |
| from fastapi import FastAPI, Request, BackgroundTasks | |
| import os | |
| import uvicorn | |
| import threading | |
| import tempfile | |
| import logging | |
| from datetime import datetime | |
| import sys | |
| import random | |
| # Create temporary directories with proper permissions | |
| temp_dir = tempfile.mkdtemp() | |
| temp_cache_dir = os.path.join(temp_dir, 'cache') | |
| temp_flagged_dir = os.path.join(temp_dir, 'flagged') | |
| temp_log_dir = os.path.join(temp_dir, 'logs') | |
| # Create necessary directories | |
| os.makedirs(temp_cache_dir, exist_ok=True) | |
| os.makedirs(temp_flagged_dir, exist_ok=True) | |
| os.makedirs(temp_log_dir, exist_ok=True) | |
| # Set Hugging Face Cache Directory | |
| os.environ["TRANSFORMERS_CACHE"] = temp_cache_dir | |
| # Set up logging to use temporary directory | |
| log_file = os.path.join(temp_log_dir, f'chatbot_{datetime.now().strftime("%Y%m%d")}.log') | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format='%(asctime)s - %(levelname)s - %(message)s', | |
| handlers=[ | |
| logging.StreamHandler(sys.stdout), | |
| logging.StreamHandler(sys.stderr) | |
| ] | |
| ) | |
| # Load model function with error handling | |
| def load_model(model_path): | |
| try: | |
| logging.info("Loading model...") | |
| tokenizer = AutoTokenizer.from_pretrained(model_path) | |
| model = AutoModelForSequenceClassification.from_pretrained(model_path) | |
| device = torch.device("cuda" if torch.cuda.is_available() else "cpu") | |
| model.to(device) | |
| logging.info(f"Model loaded successfully on {device}!") | |
| return tokenizer, model, device | |
| except Exception as e: | |
| logging.error(f"Error loading model: {str(e)}") | |
| raise | |
| # GoEmotions dataset labels (full set) | |
| emotion_labels = [ | |
| "admiration", "amusement", "anger", "annoyance", "approval", "caring", | |
| "confusion", "curiosity", "desire", "disappointment", "disapproval", "disgust", | |
| "embarrassment", "excitement", "fear", "gratitude", "grief", "joy", | |
| "love", "nervousness", "optimism", "pride", "realization", "relief", | |
| "remorse", "sadness", "surprise", "neutral" | |
| ] | |
| # Load model & tokenizer | |
| model_path = "fine_tuned_goemotions_1" | |
| try: | |
| tokenizer, model, device = load_model(model_path) | |
| except Exception as e: | |
| logging.critical("Fatal error: Could not load model. Exiting.") | |
| raise | |
| # Conversation Context Tracking | |
| class ConversationContext: | |
| def __init__(self, max_history=5): | |
| self.history = [] | |
| self.mentioned_topics = set() | |
| self.previous_emotions = [] | |
| self.max_history = max_history | |
| def update(self, user_message, detected_emotions, bot_response): | |
| self.history.append((user_message, bot_response)) | |
| if len(self.history) > self.max_history: | |
| self.history.pop(0) | |
| self.previous_emotions.extend(detected_emotions) | |
| self.previous_emotions = self.previous_emotions[-self.max_history:] # Keep recent emotions | |
| # Extract keywords for topic continuity | |
| keywords = extract_keywords(user_message) | |
| self.mentioned_topics.update(keywords) | |
| def get_last_message(self): | |
| return self.history[-1][0] if self.history else None | |
| # Extract key topics for continuity | |
| def extract_keywords(text): | |
| common_words = {"I", "you", "the", "and", "but", "is", "was", "are", "were"} | |
| return {word.lower().strip(".,!?") for word in text.split() if word.lower() not in common_words} | |
| # Simulating typing effect | |
| async def simulate_typing(response, callback, wpm=200): | |
| delay_per_char = 60 / (wpm * 5) | |
| for i in range(len(response) + 1): | |
| await callback(response[:i]) | |
| await asyncio.sleep(delay_per_char * random.uniform(0.5, 1.5)) | |
| # Tone Analysis | |
| def analyze_tone(text): | |
| if "!" in text and "😂" in text: | |
| return "playful" | |
| elif "..." in text: | |
| return "hesitant" | |
| elif "😭" in text or "😢" in text: | |
| return "sad" | |
| return "neutral" | |
| # Emotion Smoothing | |
| def smooth_emotion_transition(prev_emotions, current_emotions): | |
| if prev_emotions and current_emotions: | |
| if "joy" in prev_emotions and "sadness" in current_emotions: | |
| return ["surprise", "concern"] + current_emotions | |
| if "anger" in prev_emotions and "joy" in current_emotions: | |
| return ["relief"] + current_emotions | |
| return current_emotions | |
| # Split user input into sentences | |
| def split_sentences(text): | |
| sentences = re.split(r"(?<=[.!?])\s+", text.strip()) | |
| return [s for s in sentences if s] | |
| # Add Human-like Variations | |
| def add_human_touches(response): | |
| fillers = ["Hmm...", "Let me think...", "I see...", "To be honest,", "You know,"] | |
| if random.random() < 0.3: | |
| return random.choice(fillers) + " " + response | |
| return response | |
| # Introduce small imperfections | |
| def add_imperfections(response, probability=0.15): | |
| if random.random() > probability: | |
| return response | |
| imperfections = [ | |
| lambda r: r.replace(".", "... wait, I meant to say") + r, | |
| lambda r: "Uh..." + r, | |
| lambda r: r[:-1] + "... you know what I mean, right?", | |
| ] | |
| modifier = random.choice(imperfections) | |
| return modifier(response) | |
| # Adaptive Personality | |
| class PersonalityAdapter: | |
| def __init__(self): | |
| self.formality_level = 0.5 | |
| self.verbosity_level = 0.5 | |
| def adapt_to_user(self, user_message): | |
| if len(user_message.split()) > 12: | |
| self.formality_level = min(self.formality_level + 0.05, 1.0) | |
| else: | |
| self.formality_level = max(self.formality_level - 0.05, 0.0) | |
| if "tell me more" in user_message.lower(): | |
| self.verbosity_level = min(self.verbosity_level + 0.1, 1.0) | |
| def modify_response(self, response): | |
| if self.formality_level > 0.7: | |
| response = response.replace("yeah", "yes").replace("gonna", "going to") | |
| if self.verbosity_level < 0.3: | |
| response = response.split(".")[0] | |
| return response | |
| # Emotion prediction function | |
| def predict_emotions(text, threshold=0.3): | |
| try: | |
| model.eval() | |
| inputs = tokenizer(text, return_tensors="pt", truncation=True, padding="max_length").to(device) | |
| with torch.no_grad(): | |
| logits = model(**inputs).logits | |
| probs = F.sigmoid(logits).squeeze().cpu().numpy() | |
| predicted_emotions = [emotion_labels[i] for i, prob in enumerate(probs) if prob > threshold] | |
| if not predicted_emotions: | |
| logging.info("No emotions detected above threshold, defaulting to neutral") | |
| return ["neutral"] | |
| logging.info(f"Detected emotions: {predicted_emotions}") | |
| return predicted_emotions | |
| except Exception as e: | |
| logging.error(f"Error in emotion prediction: {str(e)}") | |
| return ["neutral"] | |
| # Generate empathetic response based on emotions and tone | |
| def generate_response(emotions, tone="neutral", context=None): | |
| try: | |
| if not emotions: | |
| emotions = ["neutral"] | |
| selected_emotion = random.choice(emotions) | |
| emotion_responses = { | |
| "admiration": [ | |
| "That’s amazing! Who or what do you admire the most? 🌟", | |
| "It’s great to be inspired! What qualities do you admire the most?", | |
| "Wow, that sounds incredible! What makes this so special to you?", | |
| "I love hearing about admiration! Who has influenced you the most?", | |
| "Admiration can be powerful! How does this inspire you in your life?" | |
| ], | |
| "amusement": [ | |
| "Haha, that sounds fun! What’s making you laugh? 😆", | |
| "A good laugh is priceless! What happened? 😄", | |
| "I love a good joke! Want to share what’s so funny?", | |
| "Laughter makes everything better! Tell me more!", | |
| "That sounds hilarious! I’d love to hear the story!" | |
| ], | |
| "anger": [ | |
| "I hear you. That sounds really frustrating. Want to talk about it? 😡", | |
| "It's okay to feel angry sometimes. What happened that upset you?", | |
| "That sounds so frustrating! Do you need to vent?", | |
| "I can see why that would make you mad. Do you want to talk it through?", | |
| "Anger is a valid emotion. How can I support you?" | |
| ], | |
| "annoyance": [ | |
| "Ugh, that sounds really annoying. Want to tell me more?", | |
| "Little things can really build up. What’s bothering you?", | |
| "That must be frustrating. What happened?", | |
| "I totally get why that would be irritating. What do you think would help?", | |
| "Sometimes things just get under our skin. What’s on your mind?" | |
| ], | |
| "approval": [ | |
| "That’s great! What do you like most about it? 👍", | |
| "That sounds amazing! What stood out to you?", | |
| "It's wonderful when things go well! What do you approve of?", | |
| "I love positivity! What aspect do you appreciate the most?", | |
| "That’s fantastic! What makes this particularly noteworthy?" | |
| ], | |
| "caring": [ | |
| "That’s so kind of you! The world needs more people like you. ❤️", | |
| "Your empathy is beautiful. What inspired this caring gesture?", | |
| "It’s wonderful to see compassion in action. What motivates you?", | |
| "Your kindness makes a difference. What made you decide to help?", | |
| "Such a thoughtful gesture! How does helping others make you feel?" | |
| ], | |
| "confusion": [ | |
| "Feeling confused? Maybe I can help! What’s unclear? 🤔", | |
| "It’s okay to feel uncertain. What’s causing your confusion?", | |
| "Let’s try to make sense of this together. What’s puzzling you?", | |
| "Confusion is a sign of learning! What are you trying to figure out?", | |
| "Sometimes things just don’t add up. What’s on your mind?" | |
| ], | |
| "curiosity": [ | |
| "Curiosity is the key to discovery! What are you wondering about? 🔍", | |
| "That’s such an interesting thought! What sparked your curiosity?", | |
| "I love curious minds! What do you want to explore more?", | |
| "It’s great to ask questions! What’s been on your mind?", | |
| "Curiosity leads to amazing discoveries! What are you investigating?" | |
| ], | |
| "desire": [ | |
| "That sounds exciting! What do you want the most right now? 😍", | |
| "It’s great to have goals! What’s driving this desire?", | |
| "Having something to strive for is motivating! What’s your plan?", | |
| "That’s a powerful desire! What makes this so important to you?", | |
| "Following your passions is amazing! What’s your next step?" | |
| ], | |
| "disappointment": [ | |
| "I’m sorry you’re feeling this way. Want to talk about it? 😞", | |
| "Disappointment can be tough. What happened?", | |
| "It’s okay to feel let down sometimes. What didn’t go as expected?", | |
| "I understand that must be frustrating. How can I support you?", | |
| "These feelings are valid. Want to share more?" | |
| ], | |
| "disapproval": [ | |
| "That sounds frustrating. What made you feel this way? 👎", | |
| "I can sense your disapproval. What aspects concern you the most?", | |
| "It’s important to uphold our standards. What didn’t meet your expectations?", | |
| "Sometimes things don’t sit right with us. What’s bothering you?", | |
| "I understand why you feel that way. What would you like to see changed?" | |
| ], | |
| "disgust": [ | |
| "Yikes, that doesn’t sound pleasant. What happened? 🤢", | |
| "That must have been awful. Want to talk about what disgusted you?", | |
| "I can understand your reaction. What made you feel this way?", | |
| "Some things are just hard to tolerate. What specifically upset you?", | |
| "That sounds really unpleasant. How are you feeling about it now?" | |
| ], | |
| "embarrassment": [ | |
| "Oh no! Don’t worry, we all have embarrassing moments. Want to share? 😳", | |
| "It’s okay, embarrassing things happen to everyone. What happened?", | |
| "Try not to be too hard on yourself. Want to talk about it?", | |
| "These moments pass with time. How are you feeling now?", | |
| "I get why that would be embarrassing, but you’re not alone in this!" | |
| ], | |
| "excitement": [ | |
| "Woohoo! That’s so exciting! What’s happening? 🎉", | |
| "I can feel your excitement! What’s making your heart race?", | |
| "That sounds thrilling! What are you most looking forward to?", | |
| "Wow, that’s amazing! What’s got you so hyped?", | |
| "How wonderful! What’s the best part about this for you?" | |
| ], | |
| "fear": [ | |
| "It’s okay to feel afraid. What’s worrying you? 😨", | |
| "I understand feeling scared. What’s making you anxious?", | |
| "Fear is a natural response. Would you like to talk about it?", | |
| "You’re not alone. What’s on your mind?", | |
| "Facing fear is tough. Want to share what’s bothering you?" | |
| ], | |
| "gratitude": [ | |
| "That’s wonderful! Gratitude is such a beautiful thing. What are you thankful for? 🙏", | |
| "It’s great to appreciate life’s blessings. What inspired this gratitude?", | |
| "Gratitude can really change our perspective! What made you feel this way?", | |
| "That’s such a positive outlook! What are you most grateful for?", | |
| "It’s great to reflect on the good things in life! What’s on your mind?" | |
| ], | |
| "grief": [ | |
| "I’m so sorry for your loss. It’s okay to grieve. I’m here for you. 💔", | |
| "Grief can be overwhelming. Want to share memories of them?", | |
| "Take your time to process your feelings. I’m here if you want to talk.", | |
| "Loss is never easy. What support do you need right now?", | |
| "I’m here to listen as you go through this. Would talking help?" | |
| ], | |
| "joy": [ | |
| "That’s amazing! Joy is contagious. Want to share why you’re happy? 😊", | |
| "It’s wonderful to hear good news! What’s bringing you joy?", | |
| "Happiness is a gift! What’s making you feel this way?", | |
| "I love hearing about happy moments! What’s bringing you happiness?", | |
| "That sounds delightful! What’s making your day so great?" | |
| ], | |
| "love": [ | |
| "Love is such a beautiful feeling! Who or what are you thinking about? ❤️", | |
| "Love makes the world a better place! What makes this so special?", | |
| "That’s so sweet! What do you cherish the most?", | |
| "Love can be truly powerful. What about this touches your heart?", | |
| "It’s wonderful to feel love! Want to share more?" | |
| ], | |
| "nervousness": [ | |
| "It’s completely normal to feel this way. You’ve got this! 💪", | |
| "I totally understand. Do you want some tips for staying calm?", | |
| "Nerves can mean you care about doing well! What’s on your mind?", | |
| "You’re going to do great! Want to talk through any concerns?", | |
| "Feeling nervous shows this is important to you. How can I support you?" | |
| ], | |
| "optimism": [ | |
| "That’s great! Optimism can change everything. What’s giving you hope today? 🌞", | |
| "Your positive outlook is inspiring! What’s making you feel optimistic?", | |
| "It’s wonderful to feel hopeful! What’s brightening your future?", | |
| "Optimism can be so powerful! What’s keeping you motivated?", | |
| "That’s such a great perspective! What’s making you feel this way?" | |
| ], | |
| "pride": [ | |
| "You should be proud! What accomplishment are you celebrating? 🏆", | |
| "That’s amazing! What achievement has made you feel this way?", | |
| "It’s great to acknowledge your successes! What did you accomplish?", | |
| "You’ve earned this moment of pride! What’s your biggest win?", | |
| "That’s something to be proud of! Want to share your success?" | |
| ], | |
| "realization": [ | |
| "That’s an interesting realization! What made you think of that? 🤯", | |
| "Isn’t it amazing when things click? What led to this insight?", | |
| "Those aha moments are special! What did you realize?", | |
| "It’s exciting when new ideas form! What did you discover?", | |
| "That sounds like a breakthrough! What helped you come to this realization?" | |
| ], | |
| "relief": [ | |
| "Ahh, that must feel good! What happened to make you feel relieved? 😌", | |
| "It’s nice when worry lifts! What brought this relief?", | |
| "That’s wonderful! What made the stress go away?", | |
| "Relief can be so freeing! What changed?", | |
| "I’m glad you’re feeling better! What helped resolve things?" | |
| ], | |
| "remorse": [ | |
| "It’s okay to make mistakes. Want to talk about what’s making you feel remorseful? 😔", | |
| "We all have regrets sometimes. Would you like to share what’s troubling you?", | |
| "Feeling remorse shows that you care. What’s on your mind?", | |
| "It’s brave to acknowledge our regrets. Want to talk about it?", | |
| "Remorse can be heavy to carry. How are you processing this?" | |
| ], | |
| "sadness": [ | |
| "I’m here for you. It sounds like you’re feeling sad. Want to talk about it? 💙", | |
| "It’s okay to feel sad sometimes. Would you like to share what’s hurting?", | |
| "Your feelings are valid. What’s making you feel this way?", | |
| "I’m sorry you’re feeling down. Want to talk about what’s on your mind?", | |
| "Sadness is a natural emotion. What’s been weighing on you?" | |
| ], | |
| "surprise": [ | |
| "Wow! That sounds unexpected. Was it a good surprise? 😲", | |
| "Oh my! What caught you off guard?", | |
| "That must have been quite a moment! What surprised you?", | |
| "Unexpected things can be exciting! What happened?", | |
| "That sounds surprising! How did it make you feel?" | |
| ], | |
| "neutral": [ | |
| "Hey! How’s your day going? 💙", | |
| "What’s on your mind? I’m here to listen. 💙", | |
| "Sometimes it’s nice to just chat. Want to share anything?", | |
| "I’m happy to talk about anything. What’s up?", | |
| "I’m here for you! Tell me what’s on your mind." | |
| ] | |
| } | |
| # Adjust response based on detected tone | |
| if tone == "playful": | |
| follow_up = "Haha, are you joking? 😆" | |
| elif tone == "hesitant": | |
| follow_up = "It sounds like you’re unsure about this?" | |
| elif tone == "sad": | |
| follow_up = "I hear you. It’s okay to feel this way. Do you want to talk about it? 💙" | |
| else: | |
| follow_up = random.choice(emotion_responses.get(selected_emotion, emotion_responses["neutral"])) | |
| # Context-awareness: adjust response if past messages are available | |
| if context and context.previous_emotions: | |
| transition = smooth_emotion_transition(context.previous_emotions, emotions) | |
| return f"{transition} {follow_up}" | |
| return follow_up | |
| except Exception as e: | |
| logging.error(f"Error generating response: {str(e)}") | |
| return "I'm here to listen and support you. Would you like to tell me more? 💙" | |
| # Smooth Emotion Transitions | |
| def smooth_emotion_transition(prev_emotions, current_emotions): | |
| if not prev_emotions: | |
| return current_emotions | |
| if "joy" in prev_emotions and "sadness" in current_emotions: | |
| return "That’s quite a mix of emotions. Let’s talk about it! " | |
| elif "anger" in prev_emotions and "relief" in current_emotions: | |
| return "It sounds like you’ve been through a lot, but I’m glad you’re feeling some relief. " | |
| elif "fear" in prev_emotions and "excitement" in current_emotions: | |
| return "It’s completely normal to feel both excited and a little anxious. That means it’s important to you! " | |
| elif "confusion" in prev_emotions and "realization" in current_emotions: | |
| return "It sounds like things are becoming clearer for you. That’s great! " | |
| elif "nervousness" in prev_emotions and "pride" in current_emotions: | |
| return "You might have started off nervous, but it sounds like you’re feeling more confident now. " | |
| else: | |
| return "" | |
| # Process Multiple Sentences | |
| def process_multiple_sentences(text, context): | |
| sentences = text.split(". ") # Basic sentence splitting | |
| all_emotions = [] | |
| responses = [] | |
| for sentence in sentences: | |
| emotions = predict_emotions(sentence) | |
| all_emotions.extend(emotions) | |
| transition_text = smooth_emotion_transition(context.previous_emotions, emotions) | |
| context.update(sentence, emotions, transition_text) | |
| response = generate_response(emotions, context=context) | |
| responses.append(response) | |
| return merge_responses(responses, all_emotions) | |
| # Merge Responses for Natural Flow | |
| def merge_responses(responses, emotions): | |
| if len(responses) == 1: | |
| return responses[0] | |
| # Define emotion categories | |
| positive_emotions = {"joy", "admiration", "optimism", "pride", "relief", "love", "gratitude", "excitement", "approval", "curiosity", "desire", "realization"} | |
| negative_emotions = {"sadness", "anger", "fear", "disappointment", "remorse", "grief", "disgust", "disapproval", "annoyance"} | |
| neutral_emotions = {"neutral", "confusion", "nervousness", "surprise", "caring"} | |
| # Find the dominant emotion (most frequently occurring) | |
| dominant_emotion = max(set(emotions), key=emotions.count) | |
| # Transition phrases based on emotion types | |
| if any(e in positive_emotions for e in emotions) and any(e in negative_emotions for e in emotions): | |
| transition = "I can see you’re experiencing a mix of emotions. Let’s explore this together. " | |
| elif "anger" in emotions and "relief" in emotions: | |
| transition = "It sounds like you’ve been through a lot, but I’m glad there’s some relief. " | |
| elif "nervousness" in emotions and "excitement" in emotions: | |
| transition = "It sounds like a mix of nerves and excitement—totally normal! You must really care about this. " | |
| elif "fear" in emotions and "excitement" in emotions: | |
| transition = "It’s completely natural to feel both excited and a little scared. That just means it matters to you! " | |
| elif "disappointment" in emotions and "approval" in emotions: | |
| transition = "I hear that there were some ups and downs in this experience. Let’s talk about both sides. " | |
| elif "confusion" in emotions and "realization" in emotions: | |
| transition = "It sounds like you were uncertain at first, but now things are becoming clearer. " | |
| else: | |
| transition = "" | |
| # Merge responses with varied connectors | |
| if len(responses) == 2: | |
| return transition + " ".join(responses) | |
| else: | |
| # Use varied sentence connectors to avoid repetition | |
| connectors = ["On another note,", "At the same time,", "Also,", "One more thing,", "Additionally,"] | |
| merged_responses = [responses[0]] + [f"{random.choice(connectors)} {r}" for r in responses[1:]] | |
| return transition + " ".join(merged_responses) | |
| # Chatbot Response Function | |
| def chatbot_response(text, context): | |
| try: | |
| if not text.strip(): | |
| return "I'm here for you. Feel free to share your thoughts." | |
| emotions = predict_emotions(text) | |
| tone = analyze_tone(text) | |
| # Process multi-sentence input | |
| if ". " in text: | |
| return process_multiple_sentences(text, context) | |
| # Generate chatbot response | |
| response = generate_response(emotions, tone, context) | |
| return response | |
| except Exception as e: | |
| logging.error(f"Error in chatbot response: {str(e)}") | |
| return "I'm here to listen and support you. Would you like to tell me more? 💙" | |
| # Gradio interface setup | |
| chatbot = gr.Interface( | |
| fn=chatbot_response, | |
| inputs=gr.Textbox(placeholder="Share your thoughts..."), | |
| outputs=gr.Textbox(), | |
| title="🤖 EmoChatbot - AI-Powered Emotion Detection", | |
| description="💬 Type a message and let the chatbot analyze your emotions and provide empathetic responses!", | |
| theme="compact", | |
| flagging_dir=temp_flagged_dir, | |
| flagging_options=[], | |
| css=".gradio-container {max-width: 800px; margin: auto}" | |
| ) | |
| # FastAPI setup | |
| app = FastAPI(title="EmoChatbot API") | |
| # Store user chat history | |
| user_sessions = {} | |
| def home(): | |
| return { | |
| "message": "EmoChatbot API is running", | |
| "endpoints": { | |
| "POST /dialogflow": "Dialogflow webhook for emotion detection", | |
| "POST /test": "Test endpoint" | |
| } | |
| } | |
| async def test_api(request: Request): | |
| try: | |
| data = await request.json() | |
| return {"status": "success", "message": "API is working!", "data": data} | |
| except Exception as e: | |
| logging.error(f"Error in test endpoint: {str(e)}") | |
| return {"status": "error", "message": str(e)} | |
| async def dialogflow_webhook(request: Request, background_tasks: BackgroundTasks): | |
| try: | |
| data = await request.json() | |
| user_text = data["queryResult"]["queryText"] | |
| user_id = data["session"] | |
| # Update chat history | |
| if user_id not in user_sessions: | |
| user_sessions[user_id] = [] | |
| user_sessions[user_id].append(user_text) | |
| user_sessions[user_id] = user_sessions[user_id][-5:] # Keep last 5 messages | |
| # Process emotions and generate response | |
| emotions = predict_emotions(user_text) | |
| response_text = generate_response(emotions) | |
| # Log interaction | |
| logging.info(f"Dialogflow - Session: {user_id} | Input: {user_text} | Emotions: {emotions}") | |
| return {"fulfillmentText": response_text} | |
| except Exception as e: | |
| logging.error(f"Error in dialogflow webhook: {str(e)}") | |
| return { | |
| "fulfillmentText": "I'm having trouble processing your message right now. Could you try again? 💙" | |
| } | |
| # Graceful shutdown handler | |
| def cleanup(): | |
| try: | |
| # Clean up all temporary directories | |
| import shutil | |
| shutil.rmtree(temp_dir, ignore_errors=True) | |
| logging.info("Cleanup completed successfully") | |
| except Exception as e: | |
| logging.error(f"Error during cleanup: {str(e)}") | |
| # Launch function with error handling | |
| def launch_gradio(): | |
| try: | |
| chatbot.launch(share=True, server_port=7861) | |
| except Exception as e: | |
| logging.error(f"Error launching Gradio: {str(e)}") | |
| raise | |
| if __name__ == "__main__": | |
| try: | |
| # Start Gradio in a separate thread | |
| gradio_thread = threading.Thread(target=launch_gradio) | |
| gradio_thread.start() | |
| # Run FastAPI server | |
| uvicorn.run(app, host="0.0.0.0", port=7860) | |
| except Exception as e: | |
| logging.critical(f"Application failed to start: {str(e)}") | |
| finally: | |
| cleanup() | |