File size: 9,616 Bytes
9366995 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 |
"""
Emotion Evaluator
Analyzes user emotions using j-hartmann/emotion-english-distilroberta-base model.
Calculates negative emotion sum, joy/neutral shift, and tracks emotion change trends.
"""
from typing import List, Dict, Optional
import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer
import torch.nn.functional as F
from evaluators.base import Evaluator
from evaluators.registry import register_evaluator
from custom_types import Utterance, EvaluationResult
from utils.evaluation_helpers import create_numerical_score, create_categorical_score, create_utterance_result
@register_evaluator(
"emotion_analysis",
label="Emotion Analysis",
description="Analyzes user emotions: negative emotion sum, joy/neutral shift, and emotion change trend",
category="Emotion"
)
class EmotionEvaluator(Evaluator):
"""Evaluator for emotion analysis using j-hartmann/emotion-english-distilroberta-base."""
METRIC_NAME = "emotion_analysis"
MODEL_NAME = "j-hartmann/emotion-english-distilroberta-base"
# Emotion labels in the order the model outputs them
EMOTION_LABELS = ["anger", "disgust", "fear", "joy", "neutral", "sadness", "surprise"]
# Negative emotions to sum
NEGATIVE_EMOTIONS = ["anger", "disgust", "fear", "sadness"]
# User role identifiers
USER_ROLES = {"patient", "seeker", "client", "user"}
def __init__(
self,
api_keys: Optional[Dict[str, str]] = None,
api_key: Optional[str] = None,
**kwargs
):
"""
Initialize Emotion Evaluator.
Args:
api_keys: Dict of API keys (not used for local model, kept for interface consistency)
api_key: Single API key (not used for local model, kept for interface consistency)
**kwargs: Additional arguments (ignored)
"""
super().__init__()
self.tokenizer = None
self.model = None
self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
self._load_model()
def _load_model(self):
"""Load the model and tokenizer."""
try:
self.tokenizer = AutoTokenizer.from_pretrained(self.MODEL_NAME)
self.model = AutoModelForSequenceClassification.from_pretrained(self.MODEL_NAME)
self.model.to(self.device)
self.model.eval()
except Exception as e:
raise RuntimeError(f"Failed to load {self.MODEL_NAME}: {e}")
def _predict_emotions(self, text: str) -> Dict[str, float]:
"""
Predict emotion scores for a single text.
Args:
text: The text to analyze
Returns:
Dict mapping emotion labels to their probability scores
"""
# Tokenize
inputs = self.tokenizer(
text,
return_tensors="pt",
truncation=True,
max_length=512,
padding=True
)
inputs = {k: v.to(self.device) for k, v in inputs.items()}
# Predict
with torch.no_grad():
outputs = self.model(**inputs)
scores = F.softmax(outputs.logits, dim=1)
scores = scores.cpu().numpy()[0]
# Map to emotion labels
emotion_scores = dict(zip(self.EMOTION_LABELS, scores))
return emotion_scores
def _calculate_metrics(self, emotion_scores: Dict[str, float]) -> Dict[str, float]:
"""
Calculate negative emotion sum and joy/neutral shift.
Args:
emotion_scores: Dict of emotion label -> probability
Returns:
Dict with 'sum_negative' and 'joy_neutral_shift'
"""
# Sum negative emotions
sum_negative = sum(emotion_scores[emotion] for emotion in self.NEGATIVE_EMOTIONS)
# Joy/neutral shift
joy_neutral_shift = emotion_scores["joy"] - emotion_scores["neutral"]
return {
"sum_negative": sum_negative,
"joy_neutral_shift": joy_neutral_shift,
"emotion_scores": emotion_scores
}
def _analyze_trend(self, all_metrics: List[Dict[str, float]]) -> Dict[str, float]:
"""
Analyze emotion change trend across all user utterances.
Args:
all_metrics: List of metric dicts from all user utterances
Returns:
Dict with trend information
"""
if not all_metrics:
return {
"avg_sum_negative": 0.0,
"avg_joy_neutral_shift": 0.0,
"trend_direction": "neutral"
}
# Calculate averages
avg_sum_negative = sum(m["sum_negative"] for m in all_metrics) / len(all_metrics)
avg_joy_neutral_shift = sum(m["joy_neutral_shift"] for m in all_metrics) / len(all_metrics)
# Determine trend direction
# Compare first half vs second half of conversation
mid_point = len(all_metrics) // 2
if mid_point > 0:
first_half_negative = sum(m["sum_negative"] for m in all_metrics[:mid_point]) / mid_point
second_half_negative = sum(m["sum_negative"] for m in all_metrics[mid_point:]) / (len(all_metrics) - mid_point)
first_half_shift = sum(m["joy_neutral_shift"] for m in all_metrics[:mid_point]) / mid_point
second_half_shift = sum(m["joy_neutral_shift"] for m in all_metrics[mid_point:]) / (len(all_metrics) - mid_point)
# Determine trend
negative_change = second_half_negative - first_half_negative
shift_change = second_half_shift - first_half_shift
if negative_change < -0.1 and shift_change > 0.1:
trend_direction = "improving"
elif negative_change > 0.1 and shift_change < -0.1:
trend_direction = "declining"
else:
trend_direction = "stable"
else:
trend_direction = "neutral"
return {
"avg_sum_negative": avg_sum_negative,
"avg_joy_neutral_shift": avg_joy_neutral_shift,
"trend_direction": trend_direction
}
def execute(self, conversation: List[Utterance], **kwargs) -> EvaluationResult:
"""
Evaluate emotions for each user utterance in the conversation.
Args:
conversation: List of utterances with 'speaker' and 'text'
Returns:
EvaluationResult with per-utterance scores and overall trend
"""
scores_per_utterance = []
user_metrics = [] # Track metrics for trend analysis
for i, utt in enumerate(conversation):
# Only evaluate user utterances
if utt["speaker"].lower() in self.USER_ROLES:
# Predict emotions
emotion_scores = self._predict_emotions(utt["text"])
# Calculate metrics
metrics = self._calculate_metrics(emotion_scores)
user_metrics.append(metrics)
# Create scores for this utterance
# Store both metrics per utterance
scores_per_utterance.append({
"emotion_sum_negative": create_numerical_score(
value=metrics["sum_negative"],
max_value=1.0,
label=self._get_label_for_negative(metrics["sum_negative"])
),
"emotion_joy_neutral_shift": create_numerical_score(
value=metrics["joy_neutral_shift"],
max_value=1.0,
label=self._get_label_for_shift(metrics["joy_neutral_shift"])
)
})
else:
# Not a user utterance, skip
scores_per_utterance.append({})
# Analyze overall trend
trend = self._analyze_trend(user_metrics)
# Create result with both per-utterance scores and overall trend
result = create_utterance_result(conversation, scores_per_utterance)
# Add overall trend information
if user_metrics:
result["overall"] = {
"emotion_avg_sum_negative": create_numerical_score(
value=trend["avg_sum_negative"],
max_value=1.0,
label=trend["trend_direction"]
),
"emotion_avg_joy_neutral_shift": create_numerical_score(
value=trend["avg_joy_neutral_shift"],
max_value=1.0,
label=trend["trend_direction"]
),
"emotion_trend_direction": create_categorical_score(
label=trend["trend_direction"],
confidence=None
)
}
return result
def _get_label_for_negative(self, value: float) -> str:
"""Get label for negative emotion sum."""
if value < 0.2:
return "Low"
elif value < 0.5:
return "Medium"
else:
return "High"
def _get_label_for_shift(self, value: float) -> str:
"""Get label for joy/neutral shift."""
if value > 0.2:
return "Positive"
elif value < -0.2:
return "Negative"
else:
return "Neutral"
|