diff --git a/configs/samo_emotion_detection_config.yaml b/configs/samo_emotion_detection_config.yaml new file mode 100644 index 000000000..cd1ca8d05 --- /dev/null +++ b/configs/samo_emotion_detection_config.yaml @@ -0,0 +1,186 @@ +# SAMO-DL Emotion Detection Configuration +# Optimized parameters for journal entry emotion analysis + +# Model Configuration +model: + name: "bert-base-uncased" # Robust BERT model for emotion understanding + device: null # Auto-detect (CPU/GPU) + +# Emotion Detection Parameters +emotion_detection: + # Number of emotion categories (27 GoEmotions + neutral) + num_emotions: 28 + + # Prediction threshold for binary classification + prediction_threshold: 0.6 # Updated from 0.5 for better calibration + + # Temperature scaling for calibrated predictions + temperature: 1.0 + + # Top-k emotions to return per prediction + top_k: 5 + +# Model Architecture +architecture: + # BERT configuration + hidden_dropout_prob: 0.3 # Dropout for BERT hidden layers + attention_probs_dropout_prob: 0.3 # Dropout for attention layers + + # Classification head configuration + classifier_dropout_prob: 0.5 # Dropout for classification layers + + # Freezing strategy + freeze_bert_layers: 6 # Number of BERT layers to freeze initially + + # Class balancing + use_class_weights: true # Enable class weight balancing + +# Training Configuration +training: + # Batch sizes + train_batch_size: 16 + eval_batch_size: 32 + + # Learning rates + bert_learning_rate: 2e-5 # Lower LR for BERT (fine-tuning) + classifier_learning_rate: 5e-4 # Higher LR for classification head + + # Training epochs + num_epochs: 10 + warmup_steps: 100 + + # Gradient settings + max_grad_norm: 1.0 + gradient_accumulation_steps: 1 + + # Early stopping + early_stopping_patience: 3 + early_stopping_threshold: 0.01 + +# Data Processing +data: + # Text processing + max_length: 512 # Maximum sequence length + truncation: true + padding: "max_length" + + # Data augmentation + enable_augmentation: false # Disable for now, can be enabled later + + # Validation split + validation_split: 0.2 + test_split: 0.1 + +# Evaluation Configuration +evaluation: + # Metrics to compute + metrics: + - "precision" + - "recall" + - "f1_micro" + - "f1_macro" + - "accuracy" + + # Evaluation threshold (lowered to capture more predictions) + threshold: 0.2 + + # Top-k evaluation + top_k_evaluation: true + top_k_values: [1, 3, 5] + +# Logging and Monitoring +logging: + level: "INFO" + log_interval: 100 # Log every N steps + save_interval: 1000 # Save checkpoint every N steps + + # TensorBoard logging + enable_tensorboard: true + log_dir: "logs/emotion_detection" + +# Model Saving +model_saving: + # Save directory + save_dir: "models/emotion_detection" + + # Save best model based on metric + save_best_metric: "f1_macro" + + # Save checkpoints + save_checkpoints: true + checkpoint_interval: 1 # Save every N epochs + +# Performance Optimization +performance: + # Mixed precision training + use_amp: true # Automatic Mixed Precision + + # Data loading + num_workers: 4 + pin_memory: true + + # Memory optimization + gradient_checkpointing: false # Can be enabled for memory savings + + # Inference optimization + use_torchscript: false # Can be enabled for faster inference + +# SAMO-Specific Optimizations +samo_optimizations: + # Journal entry specific settings + journal_entry_mode: true + + # Emotional context awareness + context_awareness: true + + # Multi-label prediction + multi_label_mode: true + + # Confidence calibration + calibration_enabled: true + + # Emotion intensity scaling + intensity_scaling: true + +# Error Handling +error_handling: + # Retry settings + max_retries: 3 + retry_delay: 1.0 + + # Fallback behavior + fallback_to_cpu: true + graceful_degradation: true + + # Logging errors + log_errors: true + error_log_file: "logs/emotion_detection_errors.log" + +# Security and Privacy +security: + # Input sanitization + sanitize_input: true + + # Output filtering + filter_sensitive_emotions: false # Can be enabled for privacy + + # Rate limiting + rate_limit_requests: 1000 # Requests per minute + + # Data privacy + anonymize_predictions: false # Can be enabled for privacy + +# Development and Debugging +development: + # Debug mode + debug_mode: false + + # Verbose logging + verbose: false + + # Test mode + test_mode: false + + # Profiling + enable_profiling: false + profile_steps: 100 diff --git a/src/models/emotion_detection/__pycache__/emotion_labels.cpython-38.pyc b/src/models/emotion_detection/__pycache__/emotion_labels.cpython-38.pyc new file mode 100644 index 000000000..0e25ab4a1 Binary files /dev/null and b/src/models/emotion_detection/__pycache__/emotion_labels.cpython-38.pyc differ diff --git a/src/models/emotion_detection/__pycache__/samo_bert_emotion_classifier.cpython-38.pyc b/src/models/emotion_detection/__pycache__/samo_bert_emotion_classifier.cpython-38.pyc new file mode 100644 index 000000000..9df7157b7 Binary files /dev/null and b/src/models/emotion_detection/__pycache__/samo_bert_emotion_classifier.cpython-38.pyc differ diff --git a/src/models/emotion_detection/config.py b/src/models/emotion_detection/config.py new file mode 100644 index 000000000..6af30a7c9 --- /dev/null +++ b/src/models/emotion_detection/config.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python3 +""" +Configuration module for SAMO Emotion Detection System + +This module provides centralized configuration management for the emotion +detection system, ensuring consistency across training and evaluation. +""" + +from typing import Dict, Any +from dataclasses import dataclass + +# Default configuration values +DEFAULT_CONFIG = { + "model": { + "name": "bert-base-uncased", + "num_emotions": 28, + "hidden_dropout_prob": 0.3, + "classifier_dropout_prob": 0.5, + "freeze_bert_layers": 6, + "temperature": 1.0, + }, + "training": { + "batch_size": 16, + "learning_rate": 2e-5, + "num_epochs": 10, + "weight_decay": 0.01, + }, + "evaluation": { + "threshold": 0.2, # Default evaluation threshold + "top_k": 5, + }, + "prediction": { + "threshold": 0.6, # Default prediction threshold + "max_length": 512, + } +} + + +def get_evaluation_threshold() -> float: + """Get current evaluation threshold from global config.""" + return _config.evaluation_threshold + + +def get_prediction_threshold() -> float: + """Get current prediction threshold from global config.""" + return _config.prediction_threshold + + +@dataclass +class EmotionDetectionConfig: + """Configuration class for emotion detection system.""" + + # Model configuration + model_name: str = "bert-base-uncased" + num_emotions: int = 28 + hidden_dropout_prob: float = 0.3 + classifier_dropout_prob: float = 0.5 + freeze_bert_layers: int = 6 + temperature: float = 1.0 + + # Training configuration + batch_size: int = 16 + learning_rate: float = 2e-5 + num_epochs: int = 10 + weight_decay: float = 0.01 + + # Evaluation configuration + evaluation_threshold: float = 0.2 + top_k: int = 5 + + # Prediction configuration + prediction_threshold: float = 0.6 + max_length: int = 512 + + @classmethod + def from_dict(cls, config_dict: Dict[str, Any]) -> 'EmotionDetectionConfig': + """Create config from dictionary.""" + return cls(**config_dict) + + def to_dict(self) -> Dict[str, Any]: + """Convert config to dictionary.""" + return { + "model_name": self.model_name, + "num_emotions": self.num_emotions, + "hidden_dropout_prob": self.hidden_dropout_prob, + "classifier_dropout_prob": self.classifier_dropout_prob, + "freeze_bert_layers": self.freeze_bert_layers, + "temperature": self.temperature, + "batch_size": self.batch_size, + "learning_rate": self.learning_rate, + "num_epochs": self.num_epochs, + "weight_decay": self.weight_decay, + "evaluation_threshold": self.evaluation_threshold, + "top_k": self.top_k, + "prediction_threshold": self.prediction_threshold, + "max_length": self.max_length, + } + + +def get_default_config() -> EmotionDetectionConfig: + """Get default configuration.""" + return EmotionDetectionConfig() + + +def get_config_from_dict(config_dict: Dict[str, Any]) -> EmotionDetectionConfig: + """Get configuration from dictionary with defaults.""" + default_config = get_default_config() + + # Update with provided values + for key, value in config_dict.items(): + if hasattr(default_config, key): + setattr(default_config, key, value) + + return default_config + + +# Global configuration instance +_config = get_default_config() + + +def get_config() -> EmotionDetectionConfig: + """Get current configuration.""" + return _config + + +def update_config(config_dict: Dict[str, Any]) -> None: + """Update global configuration.""" + global _config + _config = get_config_from_dict(config_dict) + + +def reset_config() -> None: + """Reset to default configuration.""" + global _config + _config = get_default_config() diff --git a/src/models/emotion_detection/emotion_labels.py b/src/models/emotion_detection/emotion_labels.py new file mode 100644 index 000000000..a846d7c90 --- /dev/null +++ b/src/models/emotion_detection/emotion_labels.py @@ -0,0 +1,345 @@ +#!/usr/bin/env python3 +""" +Emotion Labels for SAMO-DL Emotion Detection + +This module defines the emotion categories and labels used by the +SAMO emotion detection system, based on the GoEmotions dataset. + +The GoEmotions dataset includes 27 emotion categories plus neutral, +providing comprehensive coverage of emotional states for journal analysis. +""" + +from typing import List, Dict, Tuple + +# GoEmotions emotion categories (27 emotions + neutral = 28 total) +GOEMOTIONS_EMOTIONS = [ + "admiration", # 0 + "amusement", # 1 + "anger", # 2 + "annoyance", # 3 + "approval", # 4 + "caring", # 5 + "confusion", # 6 + "curiosity", # 7 + "desire", # 8 + "disappointment", # 9 + "disapproval", # 10 + "disgust", # 11 + "embarrassment", # 12 + "excitement", # 13 + "fear", # 14 + "gratitude", # 15 + "grief", # 16 + "joy", # 17 + "love", # 18 + "nervousness", # 19 + "optimism", # 20 + "pride", # 21 + "realization", # 22 + "relief", # 23 + "remorse", # 24 + "sadness", # 25 + "surprise", # 26 + "neutral", # 27 +] + +# Emotion categories grouped by valence (positive, negative, neutral) +EMOTION_VALENCE_GROUPS = { + "positive": [ + "admiration", "amusement", "approval", "caring", "curiosity", + "desire", "excitement", "gratitude", "joy", "love", + "optimism", "pride", "realization", "relief" + ], + "negative": [ + "anger", "annoyance", "confusion", "disappointment", "disapproval", + "disgust", "embarrassment", "fear", "grief", "nervousness", + "remorse", "sadness" + ], + "neutral": [ + "neutral" + ] +} + +# Emotion categories grouped by arousal (high, medium, low) +EMOTION_AROUSAL_GROUPS = { + "high": [ + "anger", "excitement", "fear", "joy", "nervousness", "surprise" + ], + "medium": [ + "amusement", "annoyance", "confusion", "curiosity", "desire", + "disappointment", "disgust", "embarrassment", "gratitude", + "love", "optimism", "pride", "relief", "remorse", "sadness" + ], + "low": [ + "admiration", "approval", "caring", "grief", "realization", "neutral" + ] +} + +# Emotion categories grouped by dominance (high, medium, low) +EMOTION_DOMINANCE_GROUPS = { + "high": [ + "anger", "approval", "disapproval", "pride", "realization" + ], + "medium": [ + "admiration", "amusement", "annoyance", "caring", "curiosity", + "desire", "excitement", "gratitude", "joy", "love", "optimism", "relief" + ], + "low": [ + "confusion", "disappointment", "disgust", "embarrassment", "fear", + "grief", "nervousness", "remorse", "sadness", "surprise", "neutral" + ] +} + +# Emotion intensity levels (for future enhancement) +EMOTION_INTENSITY_LEVELS = { + "very_low": 0.0, + "low": 0.25, + "medium": 0.5, + "high": 0.75, + "very_high": 1.0 +} + +# Emotion descriptions for better understanding +EMOTION_DESCRIPTIONS = { + "admiration": "A feeling of respect and approval for someone or something", + "amusement": "A feeling of being entertained or finding something funny", + "anger": "A strong feeling of displeasure and hostility", + "annoyance": "A feeling of slight anger or irritation", + "approval": "A feeling of agreement with or support for something", + "caring": "A feeling of concern and kindness for others", + "confusion": "A feeling of being puzzled or unclear about something", + "curiosity": "A strong desire to know or learn something", + "desire": "A strong feeling of wanting something", + "disappointment": "A feeling of sadness because something didn't meet expectations", + "disapproval": "A feeling of disagreement with or opposition to something", + "disgust": "A strong feeling of revulsion or repugnance", + "embarrassment": "A feeling of self-consciousness or shame", + "excitement": "A feeling of great enthusiasm and eagerness", + "fear": "An unpleasant emotion caused by the threat of danger or pain", + "gratitude": "A feeling of thankfulness and appreciation", + "grief": "Deep sorrow, especially caused by someone's death", + "joy": "A feeling of great pleasure and happiness", + "love": "An intense feeling of deep affection", + "nervousness": "A feeling of anxiety or unease", + "optimism": "A feeling of hopefulness and confidence about the future", + "pride": "A feeling of satisfaction in one's achievements", + "realization": "A moment of sudden understanding or awareness", + "relief": "A feeling of reassurance and relaxation", + "remorse": "A feeling of deep regret for a wrong committed", + "sadness": "A feeling of sorrow and unhappiness", + "surprise": "A feeling of astonishment or amazement", + "neutral": "A state of being neither positive nor negative" +} + +# Emotion synonyms for better text matching +EMOTION_SYNONYMS = { + "admiration": ["respect", "esteem", "reverence", "veneration"], + "amusement": ["entertainment", "fun", "delight", "merriment"], + "anger": ["rage", "fury", "wrath", "irritation", "madness"], + "annoyance": ["irritation", "bother", "vexation", "aggravation"], + "approval": ["endorsement", "support", "agreement", "acceptance"], + "caring": ["concern", "compassion", "empathy", "kindness"], + "confusion": ["bewilderment", "perplexity", "puzzlement", "disorientation"], + "curiosity": ["inquisitiveness", "interest", "wonder", "inquiry"], + "desire": ["want", "wish", "longing", "yearning", "craving"], + "disappointment": ["letdown", "dismay", "discouragement", "frustration"], + "disapproval": ["disagreement", "opposition", "objection", "dissent"], + "disgust": ["revulsion", "repugnance", "loathing", "abhorrence"], + "embarrassment": ["shame", "humiliation", "self-consciousness", "awkwardness"], + "excitement": ["enthusiasm", "eagerness", "anticipation", "thrill"], + "fear": ["anxiety", "worry", "dread", "terror", "panic"], + "gratitude": ["thankfulness", "appreciation", "recognition", "acknowledgment"], + "grief": ["sorrow", "mourning", "anguish", "heartache"], + "joy": ["happiness", "delight", "elation", "bliss", "cheerfulness"], + "love": ["affection", "adoration", "fondness", "devotion"], + "nervousness": ["anxiety", "unease", "tension", "apprehension"], + "optimism": ["hopefulness", "confidence", "positivity", "cheerfulness"], + "pride": ["satisfaction", "accomplishment", "achievement", "honor"], + "realization": ["understanding", "awareness", "insight", "comprehension"], + "relief": ["reassurance", "comfort", "ease", "relaxation"], + "remorse": ["regret", "guilt", "penitence", "contrition"], + "sadness": ["sorrow", "melancholy", "gloom", "despair", "unhappiness"], + "surprise": ["astonishment", "amazement", "shock", "wonder"], + "neutral": ["indifferent", "impartial", "unbiased", "objective"] +} + + +def get_emotion_index(emotion: str) -> int: + """ + Get the index of an emotion in the GoEmotions list. + + Args: + emotion: Emotion name + + Returns: + Index of the emotion (0-27) + + Raises: + ValueError: If emotion is not found + """ + try: + return GOEMOTIONS_EMOTIONS.index(emotion.lower()) + except ValueError as e: + raise ValueError(f"Emotion '{emotion}' not found in GoEmotions list") from e + + +def get_emotion_name(index: int) -> str: + """ + Get the emotion name from its index. + + Args: + index: Emotion index (0-27) + + Returns: + Emotion name + + Raises: + IndexError: If index is out of range + """ + if 0 <= index < len(GOEMOTIONS_EMOTIONS): + return GOEMOTIONS_EMOTIONS[index] + raise IndexError(f"Index {index} out of range for GoEmotions list") + + +def get_emotions_by_valence(valence: str) -> List[str]: + """ + Get emotions by valence group. + + Args: + valence: Valence group ('positive', 'negative', 'neutral') + + Returns: + List of emotions in the valence group + """ + return EMOTION_VALENCE_GROUPS.get(valence, []) + + +def get_emotions_by_arousal(arousal: str) -> List[str]: + """ + Get emotions by arousal group. + + Args: + arousal: Arousal group ('high', 'medium', 'low') + + Returns: + List of emotions in the arousal group + """ + return EMOTION_AROUSAL_GROUPS.get(arousal, []) + + +def get_emotions_by_dominance(dominance: str) -> List[str]: + """ + Get emotions by dominance group. + + Args: + dominance: Dominance group ('high', 'medium', 'low') + + Returns: + List of emotions in the dominance group + """ + return EMOTION_DOMINANCE_GROUPS.get(dominance, []) + + +def get_emotion_description(emotion: str) -> str: + """ + Get description of an emotion. + + Args: + emotion: Emotion name + + Returns: + Description of the emotion + """ + return EMOTION_DESCRIPTIONS.get(emotion.lower(), "No description available") + + +def get_emotion_synonyms(emotion: str) -> List[str]: + """ + Get synonyms for an emotion. + + Args: + emotion: Emotion name + + Returns: + List of synonyms for the emotion + """ + return EMOTION_SYNONYMS.get(emotion.lower(), []) + + +def get_all_emotions() -> List[str]: + """ + Get all emotion names. + + Returns: + List of all emotion names + """ + return GOEMOTIONS_EMOTIONS.copy() + + +def get_emotion_count() -> int: + """ + Get total number of emotions. + + Returns: + Number of emotions (28) + """ + return len(GOEMOTIONS_EMOTIONS) + + +def validate_emotion(emotion: str) -> bool: + """ + Check if an emotion is valid. + + Args: + emotion: Emotion name to validate + + Returns: + True if emotion is valid, False otherwise + """ + return emotion.lower() in GOEMOTIONS_EMOTIONS + + +def get_emotion_statistics() -> Dict[str, int]: + """ + Get statistics about emotion categories. + + Returns: + Dictionary with emotion statistics + """ + return { + "total_emotions": len(GOEMOTIONS_EMOTIONS), + "positive_emotions": len(EMOTION_VALENCE_GROUPS["positive"]), + "negative_emotions": len(EMOTION_VALENCE_GROUPS["negative"]), + "neutral_emotions": len(EMOTION_VALENCE_GROUPS["neutral"]), + "high_arousal_emotions": len(EMOTION_AROUSAL_GROUPS["high"]), + "medium_arousal_emotions": len(EMOTION_AROUSAL_GROUPS["medium"]), + "low_arousal_emotions": len(EMOTION_AROUSAL_GROUPS["low"]), + } + + +if __name__ == "__main__": + # Test the emotion labels module + print("๐Ÿงช Testing Emotion Labels Module") + print("=" * 50) + + print(f"Total emotions: {get_emotion_count()}") + print(f"All emotions: {get_all_emotions()}") + + print(f"\nPositive emotions: {get_emotions_by_valence('positive')}") + print(f"Negative emotions: {get_emotions_by_valence('negative')}") + print(f"Neutral emotions: {get_emotions_by_valence('neutral')}") + + print(f"\nHigh arousal emotions: {get_emotions_by_arousal('high')}") + print(f"Medium arousal emotions: {get_emotions_by_arousal('medium')}") + print(f"Low arousal emotions: {get_emotions_by_arousal('low')}") + + print("\nEmotion descriptions:") + for emotion_name in ["joy", "sadness", "anger", "fear"]: + print(f" {emotion_name}: {get_emotion_description(emotion_name)}") + + print(f"\nEmotion synonyms for 'joy': {get_emotion_synonyms('joy')}") + print(f"Emotion synonyms for 'sadness': {get_emotion_synonyms('sadness')}") + + print(f"\nEmotion statistics: {get_emotion_statistics()}") + + print("\nโœ… Emotion labels module test completed successfully!") diff --git a/src/models/emotion_detection/enhanced_bert_classifier.py b/src/models/emotion_detection/enhanced_bert_classifier.py new file mode 100644 index 000000000..0e76adcdf --- /dev/null +++ b/src/models/emotion_detection/enhanced_bert_classifier.py @@ -0,0 +1,532 @@ +""" +Enhanced BERT-based Emotion Classifier for SAMO Deep Learning. + +This module provides an enhanced BERT-based multi-label emotion classification model +with improved error handling, performance optimizations, and advanced features. +""" + +import logging +import warnings +from typing import Optional, Union, List, Dict, Tuple, Any +from dataclasses import dataclass +from contextlib import contextmanager + +import numpy as np +import torch +import torch.nn as nn +from transformers import AutoConfig, AutoModel, AutoTokenizer + +from .emotion_labels import GOEMOTIONS_EMOTIONS + +# Configure logging +logger = logging.getLogger(__name__) + +# Suppress warnings for cleaner output +warnings.filterwarnings("ignore", category=UserWarning, module="transformers") + + +@dataclass +class EmotionPrediction: + """Structured emotion prediction result.""" + emotions: Dict[str, float] + primary_emotion: str + confidence: float + emotional_intensity: str + top_k_emotions: List[Tuple[str, float]] + prediction_metadata: Dict[str, Any] + + +class EnhancedBERTEmotionClassifier(nn.Module): + """Enhanced BERT-based emotion classifier with advanced features. + + Features: + - Robust error handling and recovery + - Performance optimizations (mixed precision, caching) + - Advanced emotion analysis (intensity, confidence scoring) + - Model management utilities + - Comprehensive logging and monitoring + """ + + def __init__( + self, + model_name: str = "bert-base-uncased", + num_emotions: int = 28, + hidden_dropout_prob: float = 0.3, + classifier_dropout_prob: float = 0.5, + freeze_bert_layers: int = 0, + temperature: float = 1.0, + class_weights: Optional[torch.Tensor] = None, + use_mixed_precision: bool = True, + cache_embeddings: bool = False, + max_sequence_length: int = 512, + ) -> None: + """Initialize enhanced BERT emotion classifier. + + Args: + model_name: Hugging Face model name + num_emotions: Number of emotion categories + hidden_dropout_prob: Dropout rate for BERT hidden layers + classifier_dropout_prob: Dropout rate for classification head + freeze_bert_layers: Number of BERT layers to freeze initially + temperature: Temperature scaling parameter for calibration + class_weights: Optional class weights for imbalanced data + use_mixed_precision: Enable mixed precision training/inference + cache_embeddings: Cache BERT embeddings for repeated inputs + max_sequence_length: Maximum input sequence length + """ + super().__init__() + + self.model_name = model_name + self.num_emotions = num_emotions + self.hidden_dropout_prob = hidden_dropout_prob + self.classifier_dropout_prob = classifier_dropout_prob + self.freeze_bert_layers = freeze_bert_layers + self.temperature = temperature + self.prediction_threshold = 0.6 + self.class_weights = class_weights + self.emotion_labels = GOEMOTIONS_EMOTIONS[:num_emotions] + self.use_mixed_precision = use_mixed_precision + self.cache_embeddings = cache_embeddings + self.max_sequence_length = max_sequence_length + + # Device setup with fallback + self.device = self._setup_device() + + # Initialize model components + self._initialize_bert_model() + self._initialize_classifier() + self._initialize_utilities() + + # Move to device + self.to(self.device) + + @staticmethod + def _setup_device() -> torch.device: + """Setup device with fallback handling.""" + try: + if torch.cuda.is_available(): + device = torch.device("cuda") + logger.info("Using CUDA device: %s", torch.cuda.get_device_name()) + elif hasattr(torch.backends, 'mps') and torch.backends.mps.is_available(): + device = torch.device("mps") + logger.info("Using MPS device (Apple Silicon)") + else: + device = torch.device("cpu") + logger.info("Using CPU device") + return device + except Exception as e: + logger.warning("Device setup failed, falling back to CPU: %s", e) + return torch.device("cpu") + + def _initialize_bert_model(self) -> None: + """Initialize BERT model with error handling.""" + try: + config = AutoConfig.from_pretrained(self.model_name) + config.hidden_dropout_prob = self.hidden_dropout_prob + config.attention_probs_dropout_prob = self.hidden_dropout_prob + + self.bert = AutoModel.from_pretrained(self.model_name, config=config) + self.bert_hidden_size = config.hidden_size + + logger.info("BERT model loaded: %s", self.model_name) + except Exception as e: + logger.error("Failed to load BERT model: %s", e) + raise RuntimeError(f"BERT model initialization failed: {e}") from e + + def _initialize_classifier(self) -> None: + """Initialize classification head.""" + self.classifier = nn.Sequential( + nn.Dropout(self.classifier_dropout_prob), + nn.Linear(self.bert_hidden_size, self.bert_hidden_size), + nn.ReLU(), + nn.Dropout(self.classifier_dropout_prob), + nn.Linear(self.bert_hidden_size, self.num_emotions), + ) + + self.temperature = nn.Parameter(torch.ones(1) * self.temperature) + self._init_classification_layers() + + # Freeze BERT layers if specified + if self.freeze_bert_layers > 0: + self._freeze_bert_layers(self.freeze_bert_layers) + + def _initialize_utilities(self) -> None: + """Initialize utility components.""" + # Embedding cache for repeated inputs + if self.cache_embeddings: + self._embedding_cache = {} + + # Performance tracking + self._inference_count = 0 + self._total_inference_time = 0.0 + + # Error tracking + self._error_count = 0 + self._last_error = None + + def _init_classification_layers(self) -> None: + """Initialize classification layers with proper weight initialization.""" + for module in self.classifier: + if isinstance(module, nn.Linear): + nn.init.xavier_uniform_(module.weight) + nn.init.zeros_(module.bias) + + def _freeze_bert_layers(self, num_layers: int) -> None: + """Freeze the first num_layers of BERT.""" + if num_layers <= 0: + return + + # Freeze embeddings + for param in self.bert.embeddings.parameters(): + param.requires_grad = False + + # Freeze specified number of encoder layers + for i in range(min(num_layers, len(self.bert.encoder.layer))): + for param in self.bert.encoder.layer[i].parameters(): + param.requires_grad = False + + logger.info("Froze %s BERT layers", num_layers) + + def forward(self, input_ids: torch.Tensor, attention_mask: torch.Tensor) -> torch.Tensor: + """Forward pass with error handling and optimizations.""" + try: + # Get BERT outputs + bert_outputs = self.bert(input_ids=input_ids, attention_mask=attention_mask) + pooled_output = bert_outputs.pooler_output + + # Classification head + logits = self.classifier(pooled_output) + + # Apply temperature scaling + logits = logits / self.temperature + + return logits + + except Exception as e: + logger.exception("Forward pass failed") + self._error_count += 1 + self._last_error = str(e) + raise RuntimeError(f"Model forward pass failed: {e}") from e + + @contextmanager + def inference_mode(self): + """Context manager for inference mode with optimizations.""" + was_training = self.training + self.eval() + try: + with torch.no_grad(): + if self.use_mixed_precision: + with torch.cuda.amp.autocast(): + yield + else: + yield + finally: + if was_training: + self.train() + + def predict_emotions( + self, + texts: Union[str, List[str]], + top_k: int = 5, + return_metadata: bool = True, + batch_size: int = 32 + ) -> Union[EmotionPrediction, List[EmotionPrediction]]: + """Predict emotions for input text(s) with enhanced features. + + Args: + texts: Input text or list of texts + top_k: Number of top emotions to return + return_metadata: Whether to return prediction metadata + batch_size: Batch size for processing multiple texts + + Returns: + EmotionPrediction or list of EmotionPrediction objects + """ + start_time = torch.cuda.Event(enable_timing=True) if torch.cuda.is_available() else None + if start_time: + start_time.record() + + try: + # Handle single text input + if isinstance(texts, str): + return self._predict_single_text(texts, top_k, return_metadata) + + # Handle multiple texts + return self._predict_batch_texts(texts, top_k, return_metadata, batch_size) + + except Exception as e: + logger.exception("Emotion prediction failed") + self._error_count += 1 + self._last_error = str(e) + raise RuntimeError(f"Emotion prediction failed: {e}") from e + finally: + if start_time: + end_time = torch.cuda.Event(enable_timing=True) + end_time.record() + torch.cuda.synchronize() + inference_time = start_time.elapsed_time(end_time) / 1000.0 + self._update_performance_metrics(inference_time) + + def _predict_single_text( + self, + text: str, + top_k: int, + return_metadata: bool + ) -> EmotionPrediction: + """Predict emotions for a single text.""" + if not text or not text.strip(): + return self._create_empty_prediction(return_metadata) + + # Tokenize input + tokenizer = self._get_tokenizer() + inputs = tokenizer( + text, + truncation=True, + padding=True, + max_length=self.max_sequence_length, + return_tensors="pt" + ) + + input_ids = inputs["input_ids"].to(self.device) + attention_mask = inputs["attention_mask"].to(self.device) + + # Get predictions + with self.inference_mode(): + logits = self.forward(input_ids, attention_mask) + probabilities = torch.sigmoid(logits).cpu().numpy()[0] + + # Process results + return self._process_prediction_results( + probabilities, top_k, return_metadata, text + ) + + def _predict_batch_texts( + self, + texts: List[str], + top_k: int, + return_metadata: bool, + batch_size: int + ) -> List[EmotionPrediction]: + """Predict emotions for multiple texts in batches.""" + results = [] + + for i in range(0, len(texts), batch_size): + batch_texts = texts[i:i + batch_size] + batch_results = self._process_batch(batch_texts, top_k, return_metadata) + results.extend(batch_results) + + return results + + def _process_batch( + self, + texts: List[str], + top_k: int, + return_metadata: bool + ) -> List[EmotionPrediction]: + """Process a batch of texts.""" + # Filter out empty texts + valid_texts = [text for text in texts if text and text.strip()] + if not valid_texts: + return [self._create_empty_prediction(return_metadata) for _ in texts] + + # Tokenize batch + tokenizer = self._get_tokenizer() + inputs = tokenizer( + valid_texts, + truncation=True, + padding=True, + max_length=self.max_sequence_length, + return_tensors="pt" + ) + + input_ids = inputs["input_ids"].to(self.device) + attention_mask = inputs["attention_mask"].to(self.device) + + # Get predictions + with self.inference_mode(): + logits = self.forward(input_ids, attention_mask) + probabilities = torch.sigmoid(logits).cpu().numpy() + + # Process results + results = [] + for i, prob in enumerate(probabilities): + result = self._process_prediction_results( + prob, top_k, return_metadata, valid_texts[i] + ) + results.append(result) + + return results + + def _process_prediction_results( + self, + probabilities: np.ndarray, + top_k: int, + return_metadata: bool, + text: str + ) -> EmotionPrediction: + """Process prediction results into structured format.""" + # Create emotion dictionary + emotions = { + self.emotion_labels[i]: float(prob) + for i, prob in enumerate(probabilities) + } + + # Get top-k emotions + top_k_indices = np.argsort(probabilities)[-top_k:][::-1] + top_k_emotions = [ + (self.emotion_labels[i], float(probabilities[i])) + for i in top_k_indices + ] + + # Primary emotion and confidence + primary_idx = np.argmax(probabilities) + primary_emotion = self.emotion_labels[primary_idx] + confidence = float(probabilities[primary_idx]) + + # Emotional intensity + emotional_intensity = self._calculate_emotional_intensity(probabilities) + + # Metadata + metadata = {} + if return_metadata: + metadata = { + "text_length": len(text), + "prediction_threshold": self.prediction_threshold, + "temperature": float(self.temperature.item()), + "model_name": self.model_name, + "num_emotions": self.num_emotions, + "max_confidence": float(np.max(probabilities)), + "confidence_std": float(np.std(probabilities)), + } + + return EmotionPrediction( + emotions=emotions, + primary_emotion=primary_emotion, + confidence=confidence, + emotional_intensity=emotional_intensity, + top_k_emotions=top_k_emotions, + prediction_metadata=metadata + ) + + @staticmethod + def _calculate_emotional_intensity(probabilities: np.ndarray) -> str: + """Calculate emotional intensity based on prediction distribution.""" + max_prob = np.max(probabilities) + prob_std = np.std(probabilities) + + if max_prob >= 0.8 and prob_std >= 0.3: + return "very_high" + if max_prob >= 0.7 and prob_std >= 0.2: + return "high" + if max_prob >= 0.5 and prob_std >= 0.1: + return "moderate" + if max_prob >= 0.3: + return "low" + return "very_low" + + def _create_empty_prediction(self, return_metadata: bool) -> EmotionPrediction: + """Create empty prediction for invalid inputs.""" + emotions = {emotion: 0.0 for emotion in self.emotion_labels} + metadata = {"error": "empty_input"} if return_metadata else {} + + return EmotionPrediction( + emotions=emotions, + primary_emotion="neutral", + confidence=0.0, + emotional_intensity="very_low", + top_k_emotions=[("neutral", 1.0)], + prediction_metadata=metadata + ) + + def _get_tokenizer(self): + """Get or create tokenizer with caching.""" + if not hasattr(self, '_tokenizer'): + try: + self._tokenizer = AutoTokenizer.from_pretrained(self.model_name) + logger.info("Tokenizer loaded: %s", self.model_name) + except Exception as e: + logger.error("Failed to load tokenizer: %s", e) + raise RuntimeError(f"Tokenizer loading failed: {e}") from e + return self._tokenizer + + def _update_performance_metrics(self, inference_time: float) -> None: + """Update performance tracking metrics.""" + self._inference_count += 1 + self._total_inference_time += inference_time + + def get_performance_metrics(self) -> Dict[str, Any]: + """Get performance metrics.""" + avg_inference_time = ( + self._total_inference_time / self._inference_count + if self._inference_count > 0 else 0.0 + ) + + return { + "total_inferences": self._inference_count, + "total_inference_time": self._total_inference_time, + "average_inference_time": avg_inference_time, + "error_count": self._error_count, + "error_rate": self._error_count / max(self._inference_count, 1), + "last_error": self._last_error, + } + + def get_model_info(self) -> Dict[str, Any]: + """Get comprehensive model information.""" + total_params = sum(p.numel() for p in self.parameters()) + trainable_params = sum(p.numel() for p in self.parameters() if p.requires_grad) + + return { + "model_name": self.model_name, + "num_emotions": self.num_emotions, + "total_parameters": total_params, + "trainable_parameters": trainable_params, + "frozen_parameters": total_params - trainable_params, + "device": str(self.device), + "use_mixed_precision": self.use_mixed_precision, + "max_sequence_length": self.max_sequence_length, + "prediction_threshold": self.prediction_threshold, + "temperature": float(self.temperature.item()), + } + + def count_parameters(self) -> int: + """Count total parameters.""" + return sum(p.numel() for p in self.parameters()) + + def count_frozen_parameters(self) -> int: + """Count frozen parameters.""" + return sum(p.numel() for p in self.parameters() if not p.requires_grad) + + def save_model(self, path: str) -> None: + """Save model with error handling.""" + try: + torch.save({ + 'model_state_dict': self.state_dict(), + 'model_config': { + 'model_name': self.model_name, + 'num_emotions': self.num_emotions, + 'hidden_dropout_prob': self.hidden_dropout_prob, + 'classifier_dropout_prob': self.classifier_dropout_prob, + 'freeze_bert_layers': self.freeze_bert_layers, + 'temperature': float(self.temperature.item()), + } + }, path) + logger.info("Model saved to: %s", path) + except Exception as e: + logger.error("Failed to save model: %s", e) + raise RuntimeError(f"Model saving failed: {e}") from e + + @classmethod + def load_model(cls, path: str, device: Optional[str] = None) -> 'EnhancedBERTEmotionClassifier': + """Load model with error handling.""" + try: + checkpoint = torch.load(path, map_location=device or 'cpu') + model_config = checkpoint['model_config'] + + model = cls(**model_config) + model.load_state_dict(checkpoint['model_state_dict']) + + logger.info("Model loaded from: %s", path) + return model + except Exception as e: + logger.error("Failed to load model: %s", e) + raise RuntimeError(f"Model loading failed: {e}") from e diff --git a/src/models/emotion_detection/enhanced_config.py b/src/models/emotion_detection/enhanced_config.py new file mode 100644 index 000000000..8c7de589a --- /dev/null +++ b/src/models/emotion_detection/enhanced_config.py @@ -0,0 +1,757 @@ +""" +Enhanced Configuration Manager for SAMO Emotion Detection. + +This module provides robust configuration loading, validation, and management +for the emotion detection system with comprehensive error handling and fallbacks. +""" + +import logging +import yaml +from pathlib import Path +from typing import Dict, Any, Optional, Union, List +from dataclasses import dataclass, field +from enum import Enum + +logger = logging.getLogger(__name__) + + +class LogLevel(Enum): + """Logging levels.""" + DEBUG = "DEBUG" + INFO = "INFO" + WARNING = "WARNING" + ERROR = "ERROR" + CRITICAL = "CRITICAL" + + +class DeviceType(Enum): + """Device types.""" + AUTO = "auto" + CPU = "cpu" + CUDA = "cuda" + MPS = "mps" + + +@dataclass +class ModelConfig: + """Model configuration parameters.""" + name: str = "bert-base-uncased" + device: Optional[str] = None + use_mixed_precision: bool = True + cache_embeddings: bool = False + max_sequence_length: int = 512 + + +@dataclass +class EmotionDetectionConfig: + """Emotion detection configuration parameters.""" + num_emotions: int = 28 + prediction_threshold: float = 0.6 + temperature: float = 1.0 + top_k: int = 5 + + +@dataclass +class ArchitectureConfig: + """Model architecture configuration.""" + hidden_dropout_prob: float = 0.3 + attention_probs_dropout_prob: float = 0.3 + classifier_dropout_prob: float = 0.5 + freeze_bert_layers: int = 6 + use_class_weights: bool = True + + +@dataclass +class TrainingConfig: + """Training configuration parameters.""" + train_batch_size: int = 16 + eval_batch_size: int = 32 + bert_learning_rate: float = 2e-5 + classifier_learning_rate: float = 5e-4 + num_epochs: int = 10 + warmup_steps: int = 100 + max_grad_norm: float = 1.0 + gradient_accumulation_steps: int = 1 + early_stopping_patience: int = 3 + early_stopping_threshold: float = 0.01 + + +@dataclass +class DataConfig: + """Data processing configuration.""" + max_length: int = 512 + truncation: bool = True + padding: str = "max_length" + enable_augmentation: bool = False + validation_split: float = 0.2 + test_split: float = 0.1 + + +@dataclass +class EvaluationConfig: + """Evaluation configuration.""" + metrics: List[str] = field(default_factory=lambda: [ + "precision", "recall", "f1_micro", "f1_macro", "accuracy" + ]) + threshold: float = 0.2 + top_k_evaluation: bool = True + top_k_values: List[int] = field(default_factory=lambda: [1, 3, 5]) + + +@dataclass +class LoggingConfig: + """Logging configuration.""" + level: str = "INFO" + log_interval: int = 100 + save_interval: int = 1000 + enable_tensorboard: bool = True + log_dir: str = "logs/emotion_detection" + + +@dataclass +class ModelSavingConfig: + """Model saving configuration.""" + save_dir: str = "models/emotion_detection" + save_best_metric: str = "f1_macro" + save_checkpoints: bool = True + checkpoint_interval: int = 1 + + +@dataclass +class PerformanceConfig: + """Performance optimization configuration.""" + use_amp: bool = True + num_workers: int = 4 + pin_memory: bool = True + gradient_checkpointing: bool = False + use_torchscript: bool = False + + +@dataclass +class SAMOOptimizationsConfig: + """SAMO-specific optimizations.""" + journal_entry_mode: bool = True + context_awareness: bool = True + multi_label_mode: bool = True + calibration_enabled: bool = True + intensity_scaling: bool = True + + +@dataclass +class ErrorHandlingConfig: + """Error handling configuration.""" + max_retries: int = 3 + retry_delay: float = 1.0 + fallback_to_cpu: bool = True + graceful_degradation: bool = True + log_errors: bool = True + error_log_file: str = "logs/emotion_detection_errors.log" + + +@dataclass +class SecurityConfig: + """Security and privacy configuration.""" + sanitize_input: bool = True + filter_sensitive_emotions: bool = False + rate_limit_requests: int = 1000 + anonymize_predictions: bool = False + + +@dataclass +class DevelopmentConfig: + """Development and debugging configuration.""" + debug_mode: bool = False + verbose: bool = False + test_mode: bool = False + enable_profiling: bool = False + profile_steps: int = 100 + + +@dataclass +class EnhancedEmotionDetectionConfig: + """Enhanced configuration container for emotion detection.""" + model: ModelConfig = field(default_factory=ModelConfig) + emotion_detection: EmotionDetectionConfig = field( + default_factory=EmotionDetectionConfig + ) + architecture: ArchitectureConfig = field( + default_factory=ArchitectureConfig + ) + training: TrainingConfig = field(default_factory=TrainingConfig) + data: DataConfig = field(default_factory=DataConfig) + evaluation: EvaluationConfig = field(default_factory=EvaluationConfig) + logging: LoggingConfig = field(default_factory=LoggingConfig) + model_saving: ModelSavingConfig = field( + default_factory=ModelSavingConfig + ) + performance: PerformanceConfig = field( + default_factory=PerformanceConfig + ) + samo_optimizations: SAMOOptimizationsConfig = field( + default_factory=SAMOOptimizationsConfig + ) + error_handling: ErrorHandlingConfig = field( + default_factory=ErrorHandlingConfig + ) + security: SecurityConfig = field(default_factory=SecurityConfig) + development: DevelopmentConfig = field( + default_factory=DevelopmentConfig + ) + + +class EnhancedConfigManager: + """Enhanced configuration manager with validation and fallbacks.""" + + def __init__(self, config_path: Optional[Union[str, Path]] = None): + """Initialize configuration manager. + + Args: + config_path: Path to configuration file. If None, uses default. + """ + self.config_path = config_path + self.config = self._load_config() + + def _load_config(self) -> EnhancedEmotionDetectionConfig: + """Load configuration with comprehensive error handling.""" + if self.config_path is None: + self.config_path = self._find_default_config() + + if self.config_path is None or not Path(self.config_path).exists(): + logger.warning("No configuration file found, using defaults") + return self._create_default_config() + + try: + with open(self.config_path, 'r', encoding='utf-8') as f: + config_data = yaml.safe_load(f) or {} + + logger.info("Configuration loaded from: %s", self.config_path) + return self._parse_config(config_data) + + except yaml.YAMLError as e: + logger.error("YAML parsing error: %s", e) + logger.warning("Using default configuration due to YAML error") + return self._create_default_config() + except Exception as e: + logger.error("Configuration loading failed: %s", e) + logger.warning("Using default configuration due to loading error") + return self._create_default_config() + + @staticmethod + def _find_default_config() -> Optional[Path]: + """Find default configuration file.""" + possible_paths = [ + Path("configs/samo_emotion_detection_config.yaml"), + Path("configs/emotion_detection_config.yaml"), + Path("emotion_detection_config.yaml"), + ] + + for path in possible_paths: + if path.exists(): + return path + + return None + + @staticmethod + def _create_default_config() -> EnhancedEmotionDetectionConfig: + """Create default configuration.""" + logger.info("Creating default configuration") + return EnhancedEmotionDetectionConfig() + + def _parse_config(self, config_data: Dict[str, Any]) -> EnhancedEmotionDetectionConfig: + """Parse configuration data into structured format.""" + try: + # Parse each section with validation + model_config = self._parse_model_config( + config_data.get("model", {}) + ) + emotion_config = self._parse_emotion_detection_config( + config_data.get("emotion_detection", {}) + ) + architecture_config = self._parse_architecture_config( + config_data.get("architecture", {}) + ) + training_config = self._parse_training_config( + config_data.get("training", {}) + ) + data_config = self._parse_data_config( + config_data.get("data", {}) + ) + evaluation_config = self._parse_evaluation_config( + config_data.get("evaluation", {}) + ) + logging_config = self._parse_logging_config( + config_data.get("logging", {}) + ) + model_saving_config = self._parse_model_saving_config( + config_data.get("model_saving", {}) + ) + performance_config = self._parse_performance_config( + config_data.get("performance", {}) + ) + samo_optimizations = self._parse_samo_optimizations( + config_data.get("samo_optimizations", {}) + ) + error_handling = self._parse_error_handling( + config_data.get("error_handling", {}) + ) + security_config = self._parse_security_config( + config_data.get("security", {}) + ) + development_config = self._parse_development_config( + config_data.get("development", {}) + ) + + return EnhancedEmotionDetectionConfig( + model=model_config, + emotion_detection=emotion_config, + architecture=architecture_config, + training=training_config, + data=data_config, + evaluation=evaluation_config, + logging=logging_config, + model_saving=model_saving_config, + performance=performance_config, + samo_optimizations=samo_optimizations, + error_handling=error_handling, + security=security_config, + development=development_config, + ) + except Exception as e: + logger.error("Configuration parsing failed: %s", e) + logger.warning("Using default configuration due to parsing error") + return self._create_default_config() + + def _parse_model_config(self, data: Dict[str, Any]) -> ModelConfig: + """Parse model configuration with validation.""" + return ModelConfig( + name=self._validate_string( + data.get("name", "bert-base-uncased"), "model.name" + ), + device=self._validate_device(data.get("device")), + use_mixed_precision=self._validate_bool( + data.get("use_mixed_precision", True), "model.use_mixed_precision" + ), + cache_embeddings=self._validate_bool( + data.get("cache_embeddings", False), "model.cache_embeddings" + ), + max_sequence_length=self._validate_positive_int( + data.get("max_sequence_length", 512), "model.max_sequence_length" + ), + ) + + def _parse_emotion_detection_config(self, data: Dict[str, Any]) -> EmotionDetectionConfig: + """Parse emotion detection configuration with validation.""" + return EmotionDetectionConfig( + num_emotions=self._validate_positive_int( + data.get("num_emotions", 28), "emotion_detection.num_emotions" + ), + prediction_threshold=self._validate_float_range( + data.get("prediction_threshold", 0.6), 0.0, 1.0, + "emotion_detection.prediction_threshold" + ), + temperature=self._validate_positive_float( + data.get("temperature", 1.0), "emotion_detection.temperature" + ), + top_k=self._validate_positive_int( + data.get("top_k", 5), "emotion_detection.top_k" + ), + ) + + def _parse_architecture_config(self, data: Dict[str, Any]) -> ArchitectureConfig: + """Parse architecture configuration with validation.""" + return ArchitectureConfig( + hidden_dropout_prob=self._validate_float_range(data.get("hidden_dropout_prob", 0.3), 0.0, 1.0, "architecture.hidden_dropout_prob"), + attention_probs_dropout_prob=self._validate_float_range(data.get("attention_probs_dropout_prob", 0.3), 0.0, 1.0, "architecture.attention_probs_dropout_prob"), + classifier_dropout_prob=self._validate_float_range(data.get("classifier_dropout_prob", 0.5), 0.0, 1.0, "architecture.classifier_dropout_prob"), + freeze_bert_layers=self._validate_non_negative_int(data.get("freeze_bert_layers", 6), "architecture.freeze_bert_layers"), + use_class_weights=self._validate_bool(data.get("use_class_weights", True), "architecture.use_class_weights"), + ) + + def _parse_training_config(self, data: Dict[str, Any]) -> TrainingConfig: + """Parse training configuration with validation.""" + return TrainingConfig( + train_batch_size=self._validate_positive_int(data.get("train_batch_size", 16), "training.train_batch_size"), + eval_batch_size=self._validate_positive_int(data.get("eval_batch_size", 32), "training.eval_batch_size"), + bert_learning_rate=self._validate_positive_float(data.get("bert_learning_rate", 2e-5), "training.bert_learning_rate"), + classifier_learning_rate=self._validate_positive_float(data.get("classifier_learning_rate", 5e-4), "training.classifier_learning_rate"), + num_epochs=self._validate_positive_int(data.get("num_epochs", 10), "training.num_epochs"), + warmup_steps=self._validate_non_negative_int(data.get("warmup_steps", 100), "training.warmup_steps"), + max_grad_norm=self._validate_positive_float(data.get("max_grad_norm", 1.0), "training.max_grad_norm"), + gradient_accumulation_steps=self._validate_positive_int(data.get("gradient_accumulation_steps", 1), "training.gradient_accumulation_steps"), + early_stopping_patience=self._validate_positive_int(data.get("early_stopping_patience", 3), "training.early_stopping_patience"), + early_stopping_threshold=self._validate_positive_float(data.get("early_stopping_threshold", 0.01), "training.early_stopping_threshold"), + ) + + def _parse_data_config(self, data: Dict[str, Any]) -> DataConfig: + """Parse data configuration with validation.""" + return DataConfig( + max_length=self._validate_positive_int( + data.get("max_length", 512), "data.max_length" + ), + truncation=self._validate_bool( + data.get("truncation", True), "data.truncation" + ), + padding=self._validate_string( + data.get("padding", "max_length"), "data.padding" + ), + enable_augmentation=self._validate_bool( + data.get("enable_augmentation", False), "data.enable_augmentation" + ), + validation_split=self._validate_float_range( + data.get("validation_split", 0.2), 0.0, 1.0, "data.validation_split" + ), + test_split=self._validate_float_range( + data.get("test_split", 0.1), 0.0, 1.0, "data.test_split" + ), + ) + + def _parse_evaluation_config(self, data: Dict[str, Any]) -> EvaluationConfig: + """Parse evaluation configuration with validation.""" + return EvaluationConfig( + metrics=self._validate_list( + data.get("metrics", ["precision", "recall", "f1_micro", "f1_macro", "accuracy"]), + "evaluation.metrics" + ), + threshold=self._validate_float_range( + data.get("threshold", 0.2), 0.0, 1.0, "evaluation.threshold" + ), + top_k_evaluation=self._validate_bool( + data.get("top_k_evaluation", True), "evaluation.top_k_evaluation" + ), + top_k_values=self._validate_list( + data.get("top_k_values", [1, 3, 5]), "evaluation.top_k_values" + ), + ) + + def _parse_logging_config(self, data: Dict[str, Any]) -> LoggingConfig: + """Parse logging configuration with validation.""" + return LoggingConfig( + level=self._validate_log_level( + data.get("level", "INFO"), "logging.level" + ), + log_interval=self._validate_positive_int( + data.get("log_interval", 100), "logging.log_interval" + ), + save_interval=self._validate_positive_int( + data.get("save_interval", 1000), "logging.save_interval" + ), + enable_tensorboard=self._validate_bool( + data.get("enable_tensorboard", True), "logging.enable_tensorboard" + ), + log_dir=self._validate_string( + data.get("log_dir", "logs/emotion_detection"), "logging.log_dir" + ), + ) + + def _parse_model_saving_config(self, data: Dict[str, Any]) -> ModelSavingConfig: + """Parse model saving configuration with validation.""" + return ModelSavingConfig( + save_dir=self._validate_string( + data.get("save_dir", "models/emotion_detection"), "model_saving.save_dir" + ), + save_best_metric=self._validate_string( + data.get("save_best_metric", "f1_macro"), "model_saving.save_best_metric" + ), + save_checkpoints=self._validate_bool( + data.get("save_checkpoints", True), "model_saving.save_checkpoints" + ), + checkpoint_interval=self._validate_positive_int( + data.get("checkpoint_interval", 1), "model_saving.checkpoint_interval" + ), + ) + + def _parse_performance_config(self, data: Dict[str, Any]) -> PerformanceConfig: + """Parse performance configuration with validation.""" + return PerformanceConfig( + use_amp=self._validate_bool( + data.get("use_amp", True), "performance.use_amp" + ), + num_workers=self._validate_non_negative_int( + data.get("num_workers", 4), "performance.num_workers" + ), + pin_memory=self._validate_bool( + data.get("pin_memory", True), "performance.pin_memory" + ), + gradient_checkpointing=self._validate_bool( + data.get("gradient_checkpointing", False), "performance.gradient_checkpointing" + ), + use_torchscript=self._validate_bool( + data.get("use_torchscript", False), "performance.use_torchscript" + ), + ) + + def _parse_samo_optimizations(self, data: Dict[str, Any]) -> SAMOOptimizationsConfig: + """Parse SAMO optimizations configuration with validation.""" + return SAMOOptimizationsConfig( + journal_entry_mode=self._validate_bool( + data.get("journal_entry_mode", True), "samo_optimizations.journal_entry_mode" + ), + context_awareness=self._validate_bool( + data.get("context_awareness", True), "samo_optimizations.context_awareness" + ), + multi_label_mode=self._validate_bool( + data.get("multi_label_mode", True), "samo_optimizations.multi_label_mode" + ), + calibration_enabled=self._validate_bool( + data.get("calibration_enabled", True), "samo_optimizations.calibration_enabled" + ), + intensity_scaling=self._validate_bool( + data.get("intensity_scaling", True), "samo_optimizations.intensity_scaling" + ), + ) + + def _parse_error_handling(self, data: Dict[str, Any]) -> ErrorHandlingConfig: + """Parse error handling configuration with validation.""" + return ErrorHandlingConfig( + max_retries=self._validate_non_negative_int( + data.get("max_retries", 3), "error_handling.max_retries" + ), + retry_delay=self._validate_positive_float( + data.get("retry_delay", 1.0), "error_handling.retry_delay" + ), + fallback_to_cpu=self._validate_bool( + data.get("fallback_to_cpu", True), "error_handling.fallback_to_cpu" + ), + graceful_degradation=self._validate_bool( + data.get("graceful_degradation", True), "error_handling.graceful_degradation" + ), + log_errors=self._validate_bool( + data.get("log_errors", True), "error_handling.log_errors" + ), + error_log_file=self._validate_string( + data.get("error_log_file", "logs/emotion_detection_errors.log"), + "error_handling.error_log_file" + ), + ) + + def _parse_security_config(self, data: Dict[str, Any]) -> SecurityConfig: + """Parse security configuration with validation.""" + return SecurityConfig( + sanitize_input=self._validate_bool( + data.get("sanitize_input", True), "security.sanitize_input" + ), + filter_sensitive_emotions=self._validate_bool( + data.get("filter_sensitive_emotions", False), + "security.filter_sensitive_emotions" + ), + rate_limit_requests=self._validate_positive_int( + data.get("rate_limit_requests", 1000), "security.rate_limit_requests" + ), + anonymize_predictions=self._validate_bool( + data.get("anonymize_predictions", False), "security.anonymize_predictions" + ), + ) + + def _parse_development_config(self, data: Dict[str, Any]) -> DevelopmentConfig: + """Parse development configuration with validation.""" + return DevelopmentConfig( + debug_mode=self._validate_bool( + data.get("debug_mode", False), "development.debug_mode" + ), + verbose=self._validate_bool( + data.get("verbose", False), "development.verbose" + ), + test_mode=self._validate_bool( + data.get("test_mode", False), "development.test_mode" + ), + enable_profiling=self._validate_bool( + data.get("enable_profiling", False), "development.enable_profiling" + ), + profile_steps=self._validate_positive_int( + data.get("profile_steps", 100), "development.profile_steps" + ), + ) + + # Validation methods + @staticmethod + def _validate_string(value: Any, field_name: str) -> str: + """Validate string value.""" + if not isinstance(value, str): + logger.warning( + "Invalid string value for %s: %s, using default", field_name, value + ) + return "" + return value + + @staticmethod + def _validate_bool(value: Any, field_name: str) -> bool: + """Validate boolean value.""" + if not isinstance(value, bool): + logger.warning( + "Invalid boolean value for %s: %s, using default", field_name, value + ) + return False + return value + + @staticmethod + def _validate_positive_int(value: Any, field_name: str) -> int: + """Validate positive integer value.""" + try: + int_val = int(value) + if int_val <= 0: + logger.warning( + "Non-positive integer for %s: %s, using default", field_name, value + ) + return 1 + return int_val + except (ValueError, TypeError): + logger.warning( + "Invalid integer for %s: %s, using default", field_name, value + ) + return 1 + + @staticmethod + def _validate_non_negative_int(value: Any, field_name: str) -> int: + """Validate non-negative integer value.""" + try: + int_val = int(value) + if int_val < 0: + logger.warning( + "Negative integer for %s: %s, using default", field_name, value + ) + return 0 + return int_val + except (ValueError, TypeError): + logger.warning( + "Invalid integer for %s: %s, using default", field_name, value + ) + return 0 + + @staticmethod + def _validate_positive_float(value: Any, field_name: str) -> float: + """Validate positive float value.""" + try: + float_val = float(value) + if float_val <= 0: + logger.warning( + "Non-positive float for %s: %s, using default", field_name, value + ) + return 1.0 + return float_val + except (ValueError, TypeError): + logger.warning( + "Invalid float for %s: %s, using default", field_name, value + ) + return 1.0 + + @staticmethod + def _validate_float_range( + value: Any, min_val: float, max_val: float, field_name: str + ) -> float: + """Validate float value within range.""" + try: + float_val = float(value) + if not min_val <= float_val <= max_val: + logger.warning( + "Float out of range for %s: %s, using default", field_name, value + ) + return (min_val + max_val) / 2 + return float_val + except (ValueError, TypeError): + logger.warning( + "Invalid float for %s: %s, using default", field_name, value + ) + return (min_val + max_val) / 2 + + @staticmethod + def _validate_device(value: Any) -> Optional[str]: + """Validate device value.""" + if value is None: + return None + if not isinstance(value, str): + logger.warning("Invalid device value: %s, using auto", value) + return None + valid_devices = ["auto", "cpu", "cuda", "mps"] + if value.lower() not in valid_devices: + logger.warning("Invalid device: %s, using auto", value) + return None + return value.lower() + + @staticmethod + def _validate_log_level(value: Any, field_name: str) -> str: + """Validate log level value.""" + if not isinstance(value, str): + logger.warning( + "Invalid log level for %s: %s, using default", field_name, value + ) + return "INFO" + valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] + if value.upper() not in valid_levels: + logger.warning( + "Invalid log level for %s: %s, using default", field_name, value + ) + return "INFO" + return value.upper() + + @staticmethod + def _validate_list(value: Any, field_name: str) -> List: + """Validate list value.""" + if not isinstance(value, list): + logger.warning("Invalid list for %s: %s, using default", field_name, value) + return [] + return value + + def get_config(self) -> EnhancedEmotionDetectionConfig: + """Get the current configuration.""" + return self.config + + @staticmethod + def update_config(updates: Dict[str, Any]) -> None: + """Update configuration with new values.""" + try: + # This would need more sophisticated merging logic + # For now, just log the attempt + logger.info("Configuration update requested: %s", updates) + except Exception as e: + logger.error("Configuration update failed: %s", e) + + def save_config(self, path: Optional[Union[str, Path]] = None) -> None: + """Save current configuration to file.""" + if path is None: + path = self.config_path or "configs/samo_emotion_detection_config.yaml" + + try: + # Convert config to dictionary and save as YAML + config_dict = self._config_to_dict() + with open(path, 'w', encoding='utf-8') as f: + yaml.dump(config_dict, f, default_flow_style=False, indent=2) + logger.info("Configuration saved to: %s", path) + except Exception as e: + logger.error("Failed to save configuration: %s", e) + + def _config_to_dict(self) -> Dict[str, Any]: + """Convert configuration to dictionary.""" + # This would need proper serialization logic + # For now, return a basic structure + return { + "model": { + "name": self.config.model.name, + "device": self.config.model.device, + "use_mixed_precision": self.config.model.use_mixed_precision, + "cache_embeddings": self.config.model.cache_embeddings, + "max_sequence_length": self.config.model.max_sequence_length, + }, + "emotion_detection": { + "num_emotions": self.config.emotion_detection.num_emotions, + "prediction_threshold": ( + self.config.emotion_detection.prediction_threshold + ), + "temperature": self.config.emotion_detection.temperature, + "top_k": self.config.emotion_detection.top_k, + }, + # Add other sections as needed + } + + +def create_enhanced_config_manager( + config_path: Optional[Union[str, Path]] = None +) -> EnhancedConfigManager: + """Create an enhanced configuration manager. + + Args: + config_path: Path to configuration file + + Returns: + Enhanced configuration manager instance + """ + return EnhancedConfigManager(config_path) diff --git a/src/models/emotion_detection/samo_bert_emotion_classifier.py b/src/models/emotion_detection/samo_bert_emotion_classifier.py new file mode 100644 index 000000000..b92ab994c --- /dev/null +++ b/src/models/emotion_detection/samo_bert_emotion_classifier.py @@ -0,0 +1,521 @@ +#!/usr/bin/env python3 +""" +SAMO-Enhanced BERT Emotion Classifier + +This module provides an enhanced BERT-based emotion classification model +optimized for journal entries and emotional text processing in the SAMO-DL system. + +Key Features: +- BERT-base-uncased backbone for robust text understanding +- Multi-label emotion classification (27 emotions + neutral) +- Temperature scaling for calibrated predictions +- Dropout regularization to prevent overfitting +- Comprehensive error handling and logging +- SAMO-specific optimizations for journal entries +""" + +import logging +import warnings +from typing import Optional, Union, List, Dict, Tuple + +import numpy as np +import torch +import torch.nn as nn +from torch.utils.data import Dataset, DataLoader +import torch.nn.functional as F +from sklearn.metrics import f1_score, precision_recall_fscore_support +from transformers import AutoConfig, AutoModel, AutoTokenizer + +from .emotion_labels import GOEMOTIONS_EMOTIONS + +# Configure logging +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +# Suppress warnings for cleaner output +warnings.filterwarnings("ignore", category=UserWarning) + + +class SAMOBERTEmotionClassifier(nn.Module): + """ + SAMO-enhanced BERT emotion classifier for multi-label emotion detection. + + Architecture: + - BERT-base-uncased backbone + - Two-layer classification head for non-linear feature combination + - Sigmoid activation for independent emotion predictions + - Temperature scaling for calibrated predictions + - Dropout regularization to prevent overfitting + """ + + def __init__( + self, + model_name: str = "bert-base-uncased", + num_emotions: int = 28, # 27 emotions + neutral + config: Optional[Dict] = None, + ) -> None: + """ + Initialize SAMO BERT emotion classifier. + + Args: + model_name: Hugging Face model name + num_emotions: Number of emotion categories (27 + neutral) + config: Optional configuration dictionary + """ + super().__init__() + + # Set default config + default_config = { + "hidden_dropout_prob": 0.3, + "classifier_dropout_prob": 0.5, + "freeze_bert_layers": 6, + "temperature": 1.0, + } + + if config is None: + config = default_config + else: + config = {**default_config, **config} + + self.model_name = model_name + self.num_emotions = num_emotions + self.hidden_dropout_prob = config["hidden_dropout_prob"] + self.classifier_dropout_prob = config["classifier_dropout_prob"] + self.freeze_bert_layers = config["freeze_bert_layers"] + self.temperature = nn.Parameter(torch.ones(1) * config["temperature"]) + self.class_weights = None + self.prediction_threshold = 0.6 # Updated from 0.5 to 0.6 based on calibration + + # Load BERT model and tokenizer + self.config = AutoConfig.from_pretrained(model_name) + self.config.hidden_dropout_prob = self.hidden_dropout_prob + self.config.attention_probs_dropout_prob = self.hidden_dropout_prob + + self.bert = AutoModel.from_pretrained(model_name, config=self.config) + self.tokenizer = AutoTokenizer.from_pretrained(model_name) + + self.bert_hidden_size = self.config.hidden_size + + # Classification head + self.classifier = nn.Sequential( + nn.Dropout(self.classifier_dropout_prob), + nn.Linear(self.bert_hidden_size, self.bert_hidden_size), + nn.ReLU(), + nn.Dropout(self.classifier_dropout_prob), + nn.Linear(self.bert_hidden_size, num_emotions), + ) + + # Initialize classification layers + self._init_classification_layers() + + # Freeze BERT layers if specified + if self.freeze_bert_layers > 0: + self._freeze_bert_layers(self.freeze_bert_layers) + + # Set device + self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + self.to(self.device) + + logger.info("โœ… SAMO BERT Emotion Classifier initialized on %s", self.device) + + def _init_classification_layers(self) -> None: + """Initialize classification layers with proper weight initialization.""" + for module in self.classifier: + if isinstance(module, nn.Linear): + nn.init.xavier_uniform_(module.weight) + nn.init.zeros_(module.bias) + + def _set_bert_layers_grad(self, num_layers: int, requires_grad: bool) -> None: + """Set gradient requirements for BERT layers.""" + if num_layers <= 0: + return + + # Set embeddings + for param in self.bert.embeddings.parameters(): + param.requires_grad = requires_grad + + # Set encoder layers + for layer_idx in range(min(num_layers, len(self.bert.encoder.layer))): + for param in self.bert.encoder.layer[layer_idx].parameters(): + param.requires_grad = requires_grad + + action = "Unfrozen" if requires_grad else "Frozen" + logger.info("%s %s BERT layers", action, num_layers) + + def _freeze_bert_layers(self, num_layers: int) -> None: + """Freeze the first num_layers of BERT.""" + self._set_bert_layers_grad(num_layers, False) + + def unfreeze_bert_layers(self, num_layers: int) -> None: + """Unfreeze the first num_layers of BERT.""" + self._set_bert_layers_grad(num_layers, True) + + def forward( + self, + input_ids: torch.Tensor, + attention_mask: torch.Tensor, + token_type_ids: Optional[torch.Tensor] = None, + ) -> torch.Tensor: + """ + Forward pass through the BERT emotion classifier. + + Args: + input_ids: Token IDs from tokenizer + attention_mask: Attention mask for padding + token_type_ids: Token type IDs (optional) + + Returns: + Logits for emotion classification + """ + # Get BERT outputs + bert_outputs = self.bert( + input_ids=input_ids, + attention_mask=attention_mask, + token_type_ids=token_type_ids, + ) + + # Use [CLS] token representation for classification + # Check if pooler_output exists, fallback to first token if not + if bert_outputs.pooler_output is not None: + pooled_output = bert_outputs.pooler_output + else: + # Fallback to first token ([CLS]) hidden state + pooled_output = bert_outputs.last_hidden_state[:, 0, :] + + # Pass through classification head + logits = self.classifier(pooled_output) + + # Apply temperature scaling + logits = logits / self.temperature + + return logits + + def predict_emotions( + self, + texts: Union[str, List[str]], + threshold: float = None, + top_k: Optional[int] = None, + batch_size: int = 32, + ) -> Dict[str, Union[List[List[str]], List[List[float]], List[List[float]]]]: + """ + Predict emotions for given texts. + + Args: + texts: Single text or list of texts + threshold: Prediction threshold (uses default if None) + top_k: Return top-k emotions per text + batch_size: Batch size for processing + + Returns: + Dictionary with emotions, probabilities, and predictions + """ + if threshold is None: + threshold = self.prediction_threshold + + if isinstance(texts, str): + texts = [texts] + + self.eval() + all_emotions = [] + all_probabilities = [] + all_predictions = [] + + with torch.no_grad(): + for batch_idx in range(0, len(texts), batch_size): + batch_texts = texts[batch_idx : batch_idx + batch_size] + + # Tokenize batch + encoded = self.tokenizer( + batch_texts, + padding=True, + truncation=True, + max_length=512, + return_tensors="pt", + ) + + # Move to device + input_ids = encoded["input_ids"].to(self.device) + attention_mask = encoded["attention_mask"].to(self.device) + token_type_ids = encoded.get("token_type_ids", None) + if token_type_ids is not None: + token_type_ids = token_type_ids.to(self.device) + + # Get predictions + logits = self.forward(input_ids, attention_mask, token_type_ids) + probabilities = torch.sigmoid(logits) + + # Apply threshold + predictions = (probabilities > threshold).float() + + # Get top-k if specified + if top_k is not None: + _, top_k_indices = torch.topk(probabilities, top_k, dim=1) + predictions = torch.zeros_like(probabilities) + predictions.scatter_(1, top_k_indices, 1.0) + + # Convert to lists + batch_predictions = predictions.cpu().numpy() + batch_probabilities = probabilities.cpu().numpy() + + # Get emotion names for predictions + for pred in batch_predictions: + emotions = [ + GOEMOTIONS_EMOTIONS[i] for i, p in enumerate(pred) if p > 0 + ] + all_emotions.append(emotions) + + all_probabilities.extend(batch_probabilities.tolist()) + all_predictions.extend(batch_predictions.tolist()) + + return { + "emotions": all_emotions, + "probabilities": all_probabilities, + "predictions": all_predictions, + } + + def set_temperature(self, temperature: float) -> None: + """Set temperature scaling parameter.""" + self.temperature.data.fill_(temperature) + logger.info("Set temperature to %s", temperature) + + def count_parameters(self) -> int: + """Count total number of parameters.""" + return sum(p.numel() for p in self.parameters()) + + def count_frozen_parameters(self) -> int: + """Count number of frozen parameters.""" + return sum(p.numel() for p in self.parameters() if not p.requires_grad) + + +class WeightedBCELoss(nn.Module): + """Weighted Binary Cross Entropy Loss for multi-label emotion classification.""" + + def __init__( + self, + class_weights: Optional[torch.Tensor] = None, + reduction: str = "mean", + ) -> None: + """ + Initialize weighted BCE loss. + + Args: + class_weights: Class weights for balancing loss + reduction: Loss reduction method + """ + super().__init__() + self.class_weights = class_weights + self.reduction = reduction + + def forward(self, logits: torch.Tensor, targets: torch.Tensor) -> torch.Tensor: + """ + Compute weighted BCE loss. + + Args: + logits: Model predictions + targets: Ground truth labels + + Returns: + Weighted BCE loss + """ + # Compute BCE loss with logits for numerical stability + bce_loss = F.binary_cross_entropy_with_logits( + logits, targets.float(), reduction="none" + ) + + # Apply class weights if provided + if self.class_weights is not None: + bce_loss = bce_loss * self.class_weights.unsqueeze(0) + + # Apply reduction + if self.reduction == "mean": + return bce_loss.mean() + if self.reduction == "sum": + return bce_loss.sum() + return bce_loss + + +class EmotionDataset(Dataset): + """Dataset for emotion classification.""" + + def __init__( + self, + texts: List[str], + labels: List[List[int]], + tokenizer: AutoTokenizer, + max_length: int = 512, + ) -> None: + """ + Initialize emotion dataset. + + Args: + texts: List of text samples + labels: List of label lists (multi-label) + tokenizer: BERT tokenizer + max_length: Maximum sequence length + """ + self.texts = texts + self.labels = labels + self.tokenizer = tokenizer + self.max_length = max_length + + def __len__(self) -> int: + """Return dataset length.""" + return len(self.texts) + + def __getitem__(self, idx: int) -> Dict[str, torch.Tensor]: + """Get item at index.""" + sample_text = self.texts[idx] + labels = self.labels[idx] + + # Tokenize text + encoding = self.tokenizer( + sample_text, + truncation=True, + padding="max_length", + max_length=self.max_length, + return_tensors="pt", + ) + + # Convert labels to tensor + label_tensor = torch.tensor(labels, dtype=torch.float) + + return { + "input_ids": encoding["input_ids"].squeeze(0), + "attention_mask": encoding["attention_mask"].squeeze(0), + "token_type_ids": encoding.get("token_type_ids", None), + "labels": label_tensor, + } + + +def create_samo_bert_emotion_classifier( + model_name: str = "bert-base-uncased", + num_emotions: int = 28, + class_weights: Optional[np.ndarray] = None, + freeze_bert_layers: int = 6, +) -> Tuple[SAMOBERTEmotionClassifier, WeightedBCELoss]: + """ + Create SAMO BERT emotion classifier with loss function. + + Args: + model_name: Hugging Face model name + num_emotions: Number of emotion categories + class_weights: Optional class weights for imbalanced data + freeze_bert_layers: Number of BERT layers to freeze + + Returns: + Tuple of (model, loss_function) + """ + # Create model with only the parameters that override defaults + config = { + "freeze_bert_layers": freeze_bert_layers, + } + + emotion_classifier = SAMOBERTEmotionClassifier( + model_name=model_name, + num_emotions=num_emotions, + config=config, + ) + + # Convert class weights to tensor if provided and move to model device + class_weights_tensor = None + if class_weights is not None: + class_weights_tensor = torch.tensor(class_weights, dtype=torch.float) + class_weights_tensor = class_weights_tensor.to(emotion_classifier.device) + + # Create loss function + loss_function = WeightedBCELoss(class_weights=class_weights_tensor) + + return emotion_classifier, loss_function + + +def evaluate_emotion_classifier( + emotion_model: SAMOBERTEmotionClassifier, + dataloader: DataLoader, + device: torch.device, +) -> Dict[str, float]: + """ + Evaluate emotion classifier performance. + + Args: + emotion_model: Trained emotion classifier + dataloader: Data loader for evaluation + device: Device to run evaluation on + + Returns: + Dictionary with evaluation metrics + """ + from .config import get_evaluation_threshold + threshold = get_evaluation_threshold() + emotion_model.eval() + all_predictions = [] + all_targets = [] + + with torch.no_grad(): + for batch in dataloader: + input_ids = batch["input_ids"].to(device) + attention_mask = batch["attention_mask"].to(device) + token_type_ids = batch.get("token_type_ids", None) + if token_type_ids is not None: + token_type_ids = token_type_ids.to(device) + targets = batch["labels"].to(device) + + # Get predictions + logits = emotion_model(input_ids, attention_mask, token_type_ids) + probabilities = torch.sigmoid(logits) + predictions = (probabilities > threshold).float() + + all_predictions.append(predictions.cpu().numpy()) + all_targets.append(targets.cpu().numpy()) + + # Concatenate all batches + all_predictions = np.concatenate(all_predictions, axis=0) + all_targets = np.concatenate(all_targets, axis=0) + + # Calculate metrics + precision, recall, f1, _ = precision_recall_fscore_support( + all_targets, all_predictions, average="micro", zero_division=0 + ) + + macro_f1 = f1_score(all_targets, all_predictions, average="macro", zero_division=0) + + return { + "precision": precision, + "recall": recall, + "f1_micro": f1, + "f1_macro": macro_f1, + } + + +if __name__ == "__main__": + # Test the emotion classifier + print("๐Ÿงช Testing SAMO BERT Emotion Classifier") + print("=" * 50) + + try: + # Create model + print("1. Creating SAMO BERT Emotion Classifier...") + model, loss_fn = create_samo_bert_emotion_classifier() + print(f"โœ… Model created with {model.count_parameters():,} parameters") + print(f" Frozen parameters: {model.count_frozen_parameters():,}") + + # Test prediction + print("\n2. Testing emotion prediction...") + test_texts = [ + "I am so happy today! This is amazing!", + "I feel really sad and disappointed about this situation.", + "I'm feeling anxious and worried about the future.", + ] + + results = model.predict_emotions(test_texts, threshold=0.3) + + for i, text in enumerate(test_texts): + print(f"\nText: {text}") + print(f"Emotions: {results['emotions'][i]}") + top_probs = [f'{p:.3f}' for p in results['probabilities'][i][:5]] + print(f"Top probabilities: {top_probs}") + + print("\nโœ… SAMO BERT Emotion Classifier test completed successfully!") + + except Exception as e: + print(f"โŒ Error testing emotion classifier: {e}") + raise diff --git a/test_samo_emotion_detection_enhanced.py b/test_samo_emotion_detection_enhanced.py new file mode 100644 index 000000000..522135050 --- /dev/null +++ b/test_samo_emotion_detection_enhanced.py @@ -0,0 +1,290 @@ +#!/usr/bin/env python3 +""" +Enhanced standalone test for SAMO Emotion Detection Model + +This script tests the enhanced BERT emotion detection model with comprehensive +testing including edge cases, performance benchmarks, and error handling. +""" + +import sys +import logging +import time +from pathlib import Path + +# Add src to path for standalone testing +sys.path.insert(0, str(Path(__file__).parent / "src")) + +from models.emotion_detection.enhanced_bert_classifier import EnhancedBERTEmotionClassifier +from models.emotion_detection.enhanced_config import create_enhanced_config_manager +from models.emotion_detection.emotion_labels import get_all_emotions, get_emotion_description + +logger = logging.getLogger(__name__) + + +def test_model_initialization(): + """Test enhanced model initialization.""" + print("1. Initializing Enhanced SAMO BERT Emotion Classifier...") + try: + model = EnhancedBERTEmotionClassifier( + model_name="bert-base-uncased", + num_emotions=28, + use_mixed_precision=True, + cache_embeddings=True + ) + print("โœ… Enhanced classifier initialized successfully") + return model + except Exception as e: + print(f"โŒ Model initialization failed: {e}") + return None + + +def test_model_info(model): + """Test model information display.""" + print("\n2. Checking enhanced model information...") + try: + info = model.get_model_info() + print(f" Model: {info['model_name']}") + print(f" Total parameters: {info['total_parameters']:,}") + print(f" Trainable parameters: {info['trainable_parameters']:,}") + print(f" Device: {info['device']}") + print(f" Mixed precision: {info['use_mixed_precision']}") + print(f" Max sequence length: {info['max_sequence_length']}") + return info + except Exception as e: + print(f"โŒ Model info failed: {e}") + return None + + +def test_configuration_system(): + """Test enhanced configuration system.""" + print("\n3. Testing enhanced configuration system...") + try: + config_manager = create_enhanced_config_manager() + config = config_manager.get_config() + + print(f" Model name: {config.model.name}") + print(f" Num emotions: {config.emotion_detection.num_emotions}") + print(f" Prediction threshold: {config.emotion_detection.prediction_threshold}") + print(f" Mixed precision: {config.model.use_mixed_precision}") + print("โœ… Configuration system working") + return config_manager + except Exception as e: + print(f"โŒ Configuration test failed: {e}") + return None + + +def test_emotion_predictions(model): + """Test emotion predictions with various inputs.""" + print("\n4. Testing emotion predictions...") + + test_texts = [ + "I am so happy and excited about this new opportunity!", + "I feel really sad and disappointed about what happened.", + "I'm feeling anxious and worried about the future.", + "I'm grateful and thankful for all the support I've received.", + "", # Empty text + "This is a very long text that should test the maximum sequence length handling and truncation capabilities of the model. " * 20, # Very long text + ] + + try: + for i, text in enumerate(test_texts, 1): + print(f"\n Test {i}: {text[:50]}{'...' if len(text) > 50 else ''}") + + if not text.strip(): + print(" (Empty text - testing edge case)") + + prediction = model.predict_emotions( + text, + top_k=3, + return_metadata=True + ) + + print(f" Primary emotion: {prediction.primary_emotion}") + print(f" Confidence: {prediction.confidence:.3f}") + print(f" Intensity: {prediction.emotional_intensity}") + print(f" Top emotions: {prediction.top_k_emotions[:3]}") + + if prediction.prediction_metadata: + print(f" Text length: {prediction.prediction_metadata.get('text_length', 'N/A')}") + + print("โœ… Emotion predictions working") + return True + except Exception as e: + print(f"โŒ Emotion prediction test failed: {e}") + return False + + +def test_batch_predictions(model): + """Test batch prediction capabilities.""" + print("\n5. Testing batch predictions...") + + batch_texts = [ + "I'm feeling great today!", + "This is really frustrating.", + "I'm so grateful for everything.", + "I feel overwhelmed by all this work.", + "I'm excited about the new project!" + ] + + try: + start_time = time.time() + predictions = model.predict_emotions( + batch_texts, + top_k=2, + return_metadata=True, + batch_size=2 + ) + end_time = time.time() + + print(f" Processed {len(batch_texts)} texts in {end_time - start_time:.3f}s") + + for i, prediction in enumerate(predictions): + print(f" Text {i+1}: {prediction.primary_emotion} ({prediction.confidence:.3f})") + + print("โœ… Batch predictions working") + return True + except Exception as e: + print(f"โŒ Batch prediction test failed: {e}") + return False + + +def test_performance_metrics(model): + """Test performance tracking.""" + print("\n6. Testing performance metrics...") + + try: + # Run some predictions to generate metrics + test_texts = ["I'm happy", "I'm sad", "I'm excited"] * 5 + + for text in test_texts: + model.predict_emotions(text) + + metrics = model.get_performance_metrics() + + print(f" Total inferences: {metrics['total_inferences']}") + print(f" Average inference time: {metrics['average_inference_time']:.3f}s") + print(f" Error count: {metrics['error_count']}") + print(f" Error rate: {metrics['error_rate']:.3f}") + + print("โœ… Performance metrics working") + return True + except Exception as e: + print(f"โŒ Performance metrics test failed: {e}") + return False + + +def test_error_handling(model): + """Test error handling capabilities.""" + print("\n7. Testing error handling...") + + try: + # Test with None input + try: + model.predict_emotions(None) + except Exception as e: + print(f" โœ… Handled None input: {type(e).__name__}") + + # Test with very long text + very_long_text = "This is a test. " * 1000 + prediction = model.predict_emotions(very_long_text) + print(f" โœ… Handled very long text: {len(very_long_text)} chars") + + # Test with special characters + special_text = "I'm feeling ๐Ÿ˜Š๐ŸŽ‰ excited! @#$%^&*()" + prediction = model.predict_emotions(special_text) + print(f" โœ… Handled special characters: {prediction.primary_emotion}") + + print("โœ… Error handling working") + return True + except Exception as e: + print(f"โŒ Error handling test failed: {e}") + return False + + +def test_emotion_labels(): + """Test emotion labels functionality.""" + print("\n8. Testing emotion labels...") + + try: + all_emotions = get_all_emotions() + print(f" Total emotions: {len(all_emotions)}") + print(f" Sample emotions: {all_emotions[:5]}") + + # Test emotion descriptions + for emotion in all_emotions[:3]: + description = get_emotion_description(emotion) + print(f" {emotion}: {description}") + + print("โœ… Emotion labels working") + return True + except Exception as e: + print(f"โŒ Emotion labels test failed: {e}") + return False + + +def main(): + """Main test function.""" + print("๐Ÿง  Testing Enhanced SAMO Emotion Detection Model") + print("=" * 60) + + # Initialize logging + logging.basicConfig(level=logging.INFO) + + try: + # Test model initialization + model = test_model_initialization() + if model is None: + print("\nโŒ Model initialization failed, cannot continue") + return False + + # Test model information + model_info = test_model_info(model) + if model_info is None: + print("\nโŒ Model info test failed") + return False + + # Test configuration system + config_manager = test_configuration_system() + if config_manager is None: + print("\nโŒ Configuration test failed") + return False + + # Test emotion predictions + if not test_emotion_predictions(model): + print("\nโŒ Emotion prediction test failed") + return False + + # Test batch predictions + if not test_batch_predictions(model): + print("\nโŒ Batch prediction test failed") + return False + + # Test performance metrics + if not test_performance_metrics(model): + print("\nโŒ Performance metrics test failed") + return False + + # Test error handling + if not test_error_handling(model): + print("\nโŒ Error handling test failed") + return False + + # Test emotion labels + if not test_emotion_labels(): + print("\nโŒ Emotion labels test failed") + return False + + print("\n๐ŸŽ‰ All enhanced emotion detection tests passed!") + print("โœ… Enhanced model is ready for production use") + + return True + + except Exception as e: + logger.exception("Test suite failed") + print(f"\nโŒ Test suite failed: {e}") + return False + + +if __name__ == "__main__": + success = main() + sys.exit(0 if success else 1) diff --git a/test_samo_emotion_detection_standalone.py b/test_samo_emotion_detection_standalone.py new file mode 100644 index 000000000..d49521f4a --- /dev/null +++ b/test_samo_emotion_detection_standalone.py @@ -0,0 +1,320 @@ +#!/usr/bin/env python3 +""" +Standalone test for SAMO Emotion Detection Model + +This script tests the BERT emotion detection model independently +to ensure it works correctly before API integration. +""" + +import sys +from pathlib import Path + +# Add src to path +sys.path.insert(0, str(Path(__file__).parent / "src")) + +from models.emotion_detection.samo_bert_emotion_classifier import create_samo_bert_emotion_classifier +from models.emotion_detection.emotion_labels import get_all_emotions, get_emotion_description + +# Module-level constants for test data +SAMPLE_TEST_TEXTS = [ + "I am so happy and excited about this amazing opportunity!", + "I feel really sad and disappointed about what happened today.", + "I'm feeling anxious and worried about the upcoming presentation.", +] + +def test_model_initialization(): + """Test model initialization and basic info.""" + print("1. Initializing SAMO BERT Emotion Classifier...") + model, loss_fn = create_samo_bert_emotion_classifier() + print("โœ… Classifier initialized successfully") + return model, loss_fn + + +def test_model_info(model): + """Test model information display.""" + print("\n2. Checking model information...") + total_params = model.count_parameters() + frozen_params = model.count_frozen_parameters() + trainable_params = total_params - frozen_params + + print(f" Total parameters: {total_params:,}") + print(f" Frozen parameters: {frozen_params:,}") + print(f" Trainable parameters: {trainable_params:,}") + print(f" Device: {model.device}") + return trainable_params + + +def test_emotion_labels(): + """Test emotion labels functionality.""" + print("\n3. Testing emotion labels...") + all_emotions = get_all_emotions() + print(f" Total emotions: {len(all_emotions)}") + print(f" Sample emotions: {all_emotions[:5]}...") + return all_emotions + + +def test_emotion_predictions(model, all_emotions): + """Test emotion prediction on sample texts with edge cases.""" + print("\n4. Testing emotion prediction...") + test_texts = [ + "I am so happy and excited about this amazing opportunity!", + "I feel really sad and disappointed about what happened today.", + "I'm feeling anxious and worried about the upcoming presentation.", + "I love spending time with my family and friends.", + "I'm angry and frustrated with this situation.", + "I feel grateful and thankful for all the support I've received.", + "I'm confused and don't understand what's going on.", + "I feel proud of my accomplishments and achievements.", + # Edge cases + "", # Empty string + "Je suis trรจs heureux aujourd'hui!", # Non-English text + "I feel both happy and sad about this situation.", # Mixed emotions + "This is a very long text that should test the maximum sequence length handling and truncation capabilities of the model. " * 20, # Very long text + "I'm feeling ๐Ÿ˜Š๐ŸŽ‰ excited! @#$%^&*()", # Special characters and emojis + ] + + print(f" Testing {len(test_texts)} sample texts including edge cases...") + + for i, text in enumerate(test_texts, 1): + print(f"\n Text {i}: {text[:50]}{'...' if len(text) > 50 else ''}") + + # Handle empty text case + if not text.strip(): + print(" (Empty text - testing edge case)") + try: + results = model.predict_emotions(text, threshold=0.3, top_k=3) + emotions = results['emotions'][0] + probabilities = results['probabilities'][0] + print(f" Detected emotions: {emotions}") + print(" โœ… Empty text handled gracefully") + except Exception as e: + print(f" โŒ Empty text caused error: {e}") + continue + + # Get predictions + try: + results = model.predict_emotions(text, threshold=0.3, top_k=3) + + emotions = results['emotions'][0] + probabilities = results['probabilities'][0] + + print(f" Detected emotions: {emotions}") + + # Validate results + assert isinstance(emotions, list), "Emotions should be a list" + assert isinstance(probabilities, list), "Probabilities should be a list" + assert len(probabilities) == len(all_emotions), f"Expected {len(all_emotions)} probabilities, got {len(probabilities)}" + + # Show top probabilities + top_indices = sorted(range(len(probabilities)), + key=lambda i, probs=probabilities: probs[i], reverse=True)[:5] + print(" Top probabilities:") + for idx in top_indices: + emotion_name = all_emotions[idx] + prob = probabilities[idx] + print(f" {emotion_name}: {prob:.3f}") + + print(" โœ… Prediction successful") + + except Exception as e: + print(f" โŒ Prediction failed: {e}") + raise + + +def test_batch_predictions(model, test_texts): + """Test batch prediction functionality with assertions.""" + print("\n5. Testing batch prediction...") + batch_results = model.predict_emotions(test_texts[:3], threshold=0.3) + + # Validate output structure + _validate_batch_output_structure(batch_results) + + # Validate prediction counts + _validate_batch_prediction_counts(batch_results) + + # Validate emotion and probability data + _validate_batch_emotion_data(batch_results) + + print(f" Batch size: {len(batch_results['emotions'])}") + print(f" All predictions successful: {len(batch_results['emotions']) == 3}") + print(" โœ… Batch prediction assertions passed") + + +def _validate_batch_output_structure(batch_results): + """Validate the structure of batch prediction results.""" + assert isinstance(batch_results, dict), "Batch results should be a dictionary." + assert "emotions" in batch_results, "'emotions' key missing in batch results." + assert "probabilities" in batch_results, "'probabilities' key missing in batch results." + assert "predictions" in batch_results, "'predictions' key missing in batch results." + assert isinstance(batch_results["emotions"], list), "'emotions' should be a list." + assert isinstance(batch_results["probabilities"], list), "'probabilities' should be a list." + assert isinstance(batch_results["predictions"], list), "'predictions' should be a list." + + +def _validate_batch_prediction_counts(batch_results): + """Validate prediction counts in batch results.""" + assert len(batch_results["emotions"]) == 3, "Batch size should be 3." + assert len(batch_results["probabilities"]) == 3, "Probabilities count should be 3." + assert len(batch_results["predictions"]) == 3, "Predictions count should be 3." + + +def _validate_batch_emotion_data(batch_results): + """Validate emotion and probability data in batch results.""" + # Validate emotion lists + for i, emotion_list in enumerate(batch_results["emotions"]): + assert isinstance(emotion_list, list), f"Prediction {i} should be a list of emotions." + for emotion in emotion_list: + assert isinstance(emotion, str), f"Emotion should be a string in prediction {i}." + + # Validate probability ranges + for i, prob_list in enumerate(batch_results["probabilities"]): + assert isinstance(prob_list, list), f"Probabilities {i} should be a list." + for prob in prob_list: + assert 0.0 <= prob <= 1.0, f"Probability {prob} out of range in prediction {i}." + + +def test_temperature_scaling(model): + """Test temperature scaling functionality.""" + print("\n6. Testing temperature scaling...") + original_temp = model.temperature.item() + + model.set_temperature(0.5) # Lower temperature = more confident + results_cold = model.predict_emotions("I am very happy!", threshold=0.3) + + model.set_temperature(2.0) # Higher temperature = less confident + results_hot = model.predict_emotions("I am very happy!", threshold=0.3) + + model.set_temperature(original_temp) # Reset + + print(f" Cold temperature (0.5): {len(results_cold['emotions'][0])} emotions") + print(f" Hot temperature (2.0): {len(results_hot['emotions'][0])} emotions") + + +def test_prediction_thresholds(model): + """Test different prediction thresholds.""" + print("\n7. Testing different prediction thresholds...") + test_text = "I feel both happy and sad about this situation." + + thresholds = [0.1, 0.3, 0.5, 0.7] + num_emotions = [] + for threshold in thresholds: + results = model.predict_emotions(test_text, threshold=threshold) + emotions = results['emotions'][0] + num_emotions.append(len(emotions)) + print(f" Threshold {threshold}: {len(emotions)} emotions - {emotions}") + + # Assert that increasing the threshold does not increase the number of detected emotions + for i in range(1, len(num_emotions)): + assert num_emotions[i] <= num_emotions[i-1], ( + f"Number of emotions at threshold {thresholds[i]} ({num_emotions[i]}) " + f"should not be greater than at threshold {thresholds[i-1]} ({num_emotions[i-1]})" + ) + + print(" โœ… Threshold behavior assertions passed") + + +def test_emotion_descriptions(): + """Test emotion descriptions functionality.""" + print("\n8. Testing emotion descriptions...") + sample_emotions = ["joy", "sadness", "anger", "fear", "love"] + + # Use list comprehension to avoid loop in test + descriptions = [ + f" {emotion}: {get_emotion_description(emotion)}" + for emotion in sample_emotions + ] + + # Print all descriptions at once + print("\n".join(descriptions)) + + +def run_all_tests(): + """Run all emotion classifier tests.""" + model, _ = test_model_initialization() # loss_fn is not used in this function + trainable_params = test_model_info(model) + all_emotions = test_emotion_labels() + test_emotion_predictions(model, all_emotions) + test_batch_predictions(model, SAMPLE_TEST_TEXTS) + test_temperature_scaling(model) + test_prediction_thresholds(model) + test_emotion_descriptions() + return model, all_emotions, trainable_params + + +def test_emotion_classifier(): + """Test the SAMO emotion detection classifier functionality.""" + print("๐Ÿงช Testing SAMO Emotion Detection Model") + print("=" * 50) + + try: + model, all_emotions, trainable_params = run_all_tests() + + print("\nโœ… SAMO Emotion Detection Model test completed successfully!") + print(f" Model is ready for integration with {len(all_emotions)} emotion categories") + print(f" Device: {model.device}") + print(f" Trainable parameters: {trainable_params:,}") + + except Exception as e: + print(f"โŒ Error testing emotion classifier: {e}") + import traceback + traceback.print_exc() + raise + +def test_performance(): + """Test model performance on various text lengths with assertions.""" + print("\n๐Ÿš€ Testing Performance Characteristics") + print("=" * 50) + + try: + model, _ = create_samo_bert_emotion_classifier() + + # Test with different text lengths and edge cases + test_cases = [ + ("Short text", "I am happy!"), + ("Medium text", "I am feeling really happy and excited about this new opportunity that has come my way."), + ("Long text", "I am feeling incredibly happy and excited about this amazing new opportunity that has come my way. This is something I've been waiting for a long time, and I can't believe it's finally happening. I'm also a bit nervous about the challenges ahead, but I'm confident that I can handle them with the support of my friends and family."), + ("Empty text", ""), + ("Very long text", "This is a test. " * 1000), # Very long text + ("Special characters", "I'm feeling ๐Ÿ˜Š๐ŸŽ‰ excited! @#$%^&*()"), + ("Non-English", "Je suis trรจs heureux aujourd'hui!"), + ] + + for name, text in test_cases: + print(f"\n{name}:") + print(f" Length: {len(text)} characters") + + import time + start_time = time.time() + + try: + results = model.predict_emotions(text, threshold=0.3) + end_time = time.time() + + processing_time = end_time - start_time + emotions = results['emotions'][0] + + # Assertions for performance test + assert processing_time < 10.0, f"Processing time {processing_time:.3f}s too slow for {name}" + assert isinstance(emotions, list), f"Emotions should be a list for {name}" + assert len(emotions) >= 0, f"Emotions count should be non-negative for {name}" + + print(f" Processing time: {processing_time:.3f}s") + print(f" Detected emotions: {emotions}") + print(f" Emotions count: {len(emotions)}") + print(f" โœ… {name} handled successfully") + + except Exception as e: + print(f" โŒ Error processing {name}: {e}") + # For empty text, this might be expected behavior + if name == "Empty text": + print(f" โš ๏ธ Empty text error may be expected: {e}") + else: + raise + + except Exception as e: + print(f"โŒ Error in performance test: {e}") + raise + +if __name__ == "__main__": + test_emotion_classifier() + test_performance()