"""Emotion detection using a local transformer model with lexicon fallback.""" import os from typing import Dict # --------------------------------------------------------------------------- # Lexicon fallback (offline, zero-dependency) # --------------------------------------------------------------------------- EMOTION_KEYWORDS = { "anxious": { "worried", "nervous", "scared", "stress", "stressed", "panic", "afraid", "deadline", }, "angry": {"angry", "mad", "furious", "annoyed", "irritated", "rage", "unfair"}, "sad": {"sad", "hurt", "grief", "down", "depressed", "empty"}, "lonely": {"lonely", "alone", "isolated", "unseen"}, "ashamed": {"ashamed", "embarrassed", "guilty", "worthless"}, "curious": {"curious", "wonder", "why", "how", "learn", "explore", "question"}, "joyful": {"happy", "joy", "grateful", "glad"}, "hopeful": {"hope", "hopeful", "possible", "better"}, "confused": {"confused", "lost", "stuck", "unclear", "messy"}, "overwhelmed": {"overwhelmed", "too", "much", "can't", "cannot"}, "tired": {"tired", "exhausted", "sleepy", "drained"}, "vulnerable": {"honest", "scared", "hard", "sensitive", "open"}, "excited": {"excited", "amazing", "awesome", "wow", "unbelievable", "huge"}, "focused": {"build", "fix", "ship", "implement", "test", "run", "phase", "ready"}, } EMOTION_DIMENSIONS = { "anxious": (-0.6, 0.8, "grounding,clarity"), "angry": (-0.5, 0.85, "validation,boundaries"), "sad": (-0.75, 0.35, "presence,validation"), "lonely": (-0.7, 0.35, "connection,validation"), "ashamed": (-0.8, 0.6, "low_judgment,reframing"), "curious": (0.35, 0.55, "exploration,questions"), "joyful": (0.8, 0.65, "celebration,continuity"), "hopeful": (0.65, 0.55, "encouragement,planning"), "confused": (-0.2, 0.55, "clarity,structure"), "overwhelmed": (-0.65, 0.9, "grounding,prioritization"), "focused": (0.45, 0.75, "problem_solving,challenge"), "excited": (0.85, 0.8, "momentum,validation"), "tired": (-0.35, 0.2, "rest,simplification"), "vulnerable": (-0.25, 0.5, "care,trust"), "neutral": (0.0, 0.2, "natural"), } # --------------------------------------------------------------------------- # Transformer-based classifier (lazy load) # --------------------------------------------------------------------------- _TRANSFORMER_CLASSIFIER = None _MODEL_TO_LABEL = { "fear": "anxious", "sadness": "sad", "anger": "angry", "joy": "joyful", "surprise": "excited", "disgust": "ashamed", "neutral": "neutral", } _MODEL_VALENCE = { "fear": -0.6, "sadness": -0.7, "anger": -0.5, "joy": 0.8, "surprise": 0.4, "disgust": -0.6, "neutral": 0.0, } _MODEL_AROUSAL = { "fear": 0.85, "sadness": 0.3, "anger": 0.9, "joy": 0.7, "surprise": 0.8, "disgust": 0.6, "neutral": 0.2, } def _load_classifier(): global _TRANSFORMER_CLASSIFIER if _TRANSFORMER_CLASSIFIER is not None: return _TRANSFORMER_CLASSIFIER try: from transformers import pipeline # Suppress noisy download logs os.environ["HF_HUB_DISABLE_SYMLINKS_WARNING"] = "1" _TRANSFORMER_CLASSIFIER = pipeline( "text-classification", model="j-hartmann/emotion-english-distilroberta-base", top_k=None, device="cpu", ) return _TRANSFORMER_CLASSIFIER except Exception as exc: print( f"[emotion] Transformer model unavailable ({exc}), using lexicon fallback." ) _TRANSFORMER_CLASSIFIER = False return _TRANSFORMER_CLASSIFIER def _detect_transformer(text: str) -> Dict: clf = _load_classifier() if clf is False: return {} # Truncate to the last 1500 characters to stay within Roberta 512-token limit text_truncated = text[-1500:] if len(text) > 1500 else text raw = clf(text_truncated)[0] # list of dicts # Sort by score descending raw = sorted(raw, key=lambda x: x["score"], reverse=True) top = raw[0] secondary = raw[1] if len(raw) > 1 else {"label": "neutral", "score": 0.0} label = _MODEL_TO_LABEL.get(str(top["label"]), "neutral") sec_label = _MODEL_TO_LABEL.get(str(secondary["label"]), "neutral") # Weighted valence/arousal from top-3 predictions valence = 0.0 arousal = 0.0 total = 0.0 for item in raw[:3]: w = item["score"] valence += _MODEL_VALENCE.get(item["label"], 0.0) * w arousal += _MODEL_AROUSAL.get(item["label"], 0.0) * w total += w if total > 0: valence /= total arousal /= total intensity = min(1.0, 0.25 + top["score"] * 0.75 + text.count("!") * 0.03) confidence = min(0.95, 0.35 + top["score"] * 0.6) return { "label": label, "secondary": sec_label, "confidence": confidence, "valence": valence, "arousal": arousal, "intensity": intensity, "needs": EMOTION_DIMENSIONS.get(label, EMOTION_DIMENSIONS["neutral"])[2], "detector": "local_transformer_v1", "raw_model": top["label"], } # --------------------------------------------------------------------------- # Lexicon fallback # --------------------------------------------------------------------------- def _detect_lexicon(text: str) -> Dict: words = {word.strip(".,!?;:\"'()[]{}").lower() for word in text.split()} scores = { label: len(words & keywords) for label, keywords in EMOTION_KEYWORDS.items() } label, score = max(scores.items(), key=lambda item: item[1]) if score == 0: label = "neutral" sorted_scores = sorted(scores.items(), key=lambda item: item[1], reverse=True) secondary = ( sorted_scores[1][0] if len(sorted_scores) > 1 and sorted_scores[1][1] else "neutral" ) intensity = min(1.0, 0.25 + score * 0.18 + text.count("!") * 0.05) confidence = 0.2 if score == 0 else min(0.95, 0.35 + score * 0.12) valence, arousal, needs = EMOTION_DIMENSIONS[label] return { "label": label, "secondary": secondary, "confidence": confidence, "valence": valence, "arousal": arousal, "intensity": intensity, "needs": needs, "detector": "offline_lexicon_v1", "raw_model": None, } # --------------------------------------------------------------------------- # Unified API # --------------------------------------------------------------------------- def detect_emotion(text: str) -> Dict: result = _detect_transformer(text) if result is not None: return result return _detect_lexicon(text) def emotion_prompt_hint(emotion): label = emotion.get("label", "neutral") hints = { "anxious": "Use steadier pacing, reduce ambiguity, and separate the next small action from the larger fear.", "angry": "Validate the signal without amplifying heat; look for boundaries, facts, and leverage.", "sad": "Lead with warmth and presence before analysis. Do not rush to fix what first needs witnessing.", "curious": "Lean into exploration, pattern finding, and generative questions.", "excited": "Channel momentum into concrete next steps and keep assumptions checked.", "confused": "Compress the situation into simple handles, then rebuild clarity.", "focused": "Be direct, operational, and concise; prioritize execution.", } return hints.get( label, "Respond naturally; infer the needed mode from the message and context." ) if __name__ == "__main__": samples = [ "I am so worried about the deadline", "I feel happy and grateful today", "That makes me furious", "I feel empty and alone", "I am curious about how this works", "This is too much, I cannot handle it", "The sky is blue", ] for s in samples: r = detect_emotion(s) print( f"{s[:40]:40s} -> {r['label']:10s} (conf={r['confidence']:.2f}, val={r['valence']:.2f}, ar={r['arousal']:.2f}, det={r['detector']})" )