Emotion Classification Agent

EmotionLabel = Literal[
    "CALM",              # Relaxed, no urgency
    "RELIEVED",          # Positive, situation improving
    "TENSE",             # Mild anxiety
    "DISTRESSED",        # Significant distress
    "HIGHLY_DISTRESSED", # Extreme distress/panic
]

EMOTION_PROVIDER=heuristic

async def _analyze_emotion_heuristic(
    transcript: Optional[str],
    distress: float,
) -> Dict[str, Any]:
    """
    Rule-based classifier using distress + text cues.
    
    Returns:
        {
            "label": "HIGHLY_DISTRESSED",
            "intensity": 0.85,
            "sentiment": "negative",
            "distress_input": 0.75,
            "has_transcript": True,
            "source": "heuristic"
        }
    """

EMOTION_PROVIDER=deepgram
DEEPGRAM_API_KEY=your_key_here

async def _analyze_emotion_deepgram(
    transcript: Optional[str], distress: float
) -> Optional[dict]:
    """
    Uses Deepgram Text Intelligence (sentiment) on the transcript.
    
    API endpoint: https://api.deepgram.com/v1/read?sentiment=true
    """

EMOTION_PROVIDER=openai
OPENAI_API_KEY=sk-...
OPENAI_MODEL_EMOTION=gpt-5-nano  # or gpt-4o-mini

async def _analyze_emotion_openai(
    transcript: Optional[str], distress: float
) -> Optional[dict]:
    """
    Ask OpenAI to refine the emotion label given transcript + distress.
    Returns None on any error so caller can safely fall back.
    """

async def analyze_emotion(
    transcript: Optional[str],
    distress: float,
) -> Optional[Dict[str, Any]]:
    """
    Public entrypoint. Chooses provider based on EMOTION_PROVIDER env var.
    
    Args:
        transcript: Text from speech-to-text (can be None)
        distress: Audio distress score (0.0-1.0)
        
    Returns:
        {
            "label": EmotionLabel,
            "intensity": float,  # 0.0-1.0
            "sentiment": "positive" | "neutral" | "negative",
            "distress_input": float,  # Original distress score
            "has_transcript": bool
        }
    """

from app.agents.emotion import analyze_emotion

# Analyze with transcript and distress score
emotion = await analyze_emotion(
    transcript="I need help, someone's been shot!",
    distress=0.8
)

print(emotion)
# {
#   "label": "HIGHLY_DISTRESSED",
#   "intensity": 0.85,
#   "sentiment": "negative",
#   "distress_input": 0.8,
#   "has_transcript": True,
#   "source": "heuristic"
# }

# When transcript is not yet available
emotion = await analyze_emotion(
    transcript=None,
    distress=0.6
)
# Will classify based solely on audio distress

# In NLP track
async def process_call(transcript: str, audio_distress: float):
    # Get emotion alongside other analyses
    emotion, service = await asyncio.gather(
        analyze_emotion(transcript, audio_distress),
        classify_service_and_tags(transcript, audio_distress)
    )
    
    return {
        "emotion": emotion,
        "service": service,
        "summary": await generate_summary(transcript, service['category'], service['tags'])
    }

life_threatening = [
    "shot", "shooting", "stabbed", "stabbing",
    "can't breathe", "not breathing",
    "overdose", "heart attack", "unconscious",
    "bleeding out", "heavy bleeding",
    "suicide", "kill myself"
]

if any(k in txt for k in life_threatening):
    # Person reporting life-threatening emergency should be
    # classified as distressed even if voice sounds calm
    # (shock, numbness, disassociation)
    sentiment = "negative"
    label = "HIGHLY_DISTRESSED"
    intensity = max(intensity, 0.8)

neg_keywords = [
    "angry", "upset", "emergency", "hurt",
    "scared", "afraid", "terrible", "panic",
    "fuck", "fucking", "crazy", "going through it"
]

pos_keywords = [
    "thank you", "thanks", "great",
    "awesome", "happy", "appreciate"
]

headers = {
    "Authorization": f"Token {DEEPGRAM_API_KEY}",
    "Content-Type": "application/json",
}

payload = {"text": transcript}

async with httpx.AsyncClient(timeout=10) as c:
    r = await c.post(
        "https://api.deepgram.com/v1/read?sentiment=true&language=en",
        headers=headers,
        json=payload,
    )

j = r.json()
results = j.get("results", {})
sentiments = results.get("sentiments", {})
avg = sentiments.get("average", {})

sentiment = str(avg.get("sentiment", "neutral")).lower()
confidence = float(avg.get("sentiment_score", distress))

crisis_keywords = [
    "shot", "shoot", "bleeding", "die", "dying",
    "kill myself", "kill me", "suicide",
    "hanging", "overdose", "jump off",
    "people are dying"
]
crisis_hit = any(k in text_l for k in crisis_keywords)

if sentiment == "negative":
    if crisis_hit:
        if distress >= 0.4:
            label = "HIGHLY_DISTRESSED"
        else:
            label = "DISTRESSED"
        distress_intensity = max(distress, 0.7)
    else:
        if distress >= 0.7:
            label = "HIGHLY_DISTRESSED"
        elif distress >= 0.4:
            label = "DISTRESSED"
        else:
            label = "TENSE"

system_msg = (
    "You are an assistant that classifies a short emergency phone call "
    "transcript into a coarse emotional state. "
    "You MUST respond with a single JSON object only.\n\n"
    "Fields:\n"
    "  - label: one of CALM, RELIEVED, TENSE, DISTRESSED, HIGHLY_DISTRESSED\n"
    "  - sentiment: one of positive, neutral, negative\n"
    "  - intensity: float between 0 and 1 (overall emotional intensity)\n"
)

user_msg = {
    "role": "user",
    "content": (
        "Transcript:\n"
        f"{transcript}\n\n"
        f"Audio distress_score (0–1): {distress:.3f}\n\n"
        "Return ONLY JSON, like:\n"
        '{"label":"DISTRESSED","sentiment":"negative","intensity":0.82}'
    ),
}

payload = {
    "model": OPENAI_MODEL_EMOTION,
    "temperature": 0,  # Deterministic output
    "response_format": {"type": "json_object"},
    "messages": [
        {"role": "system", "content": system_msg},
        user_msg,
    ],
}

async with httpx.AsyncClient(timeout=10) as c:
    r = await c.post(
        "https://api.openai.com/v1/chat/completions",
        headers={"Authorization": f"Bearer {OPENAI_API_KEY}"},
        json=payload,
    )

# Choose provider (default: heuristic)
EMOTION_PROVIDER=heuristic  # or "deepgram" or "openai"

# API Keys (only needed for respective providers)
OPENAI_API_KEY=sk-...
DEEPGRAM_API_KEY=...

# Model selection (OpenAI only)
OPENAI_MODEL_EMOTION=gpt-5-nano  # or "gpt-4o-mini"

# In emotion.py
EMOTION_PROVIDER = os.getenv("EMOTION_PROVIDER", "heuristic").lower()

if provider == "deepgram":
    return await _analyze_emotion_deepgram(transcript, distress)
elif provider == "openai":
    return await _analyze_emotion_openai(transcript, distress)
else:
    return await _analyze_emotion_heuristic(transcript, distress)

try:
    emotion = await analyze_emotion(transcript, distress)
    if emotion is None:
        # API provider failed, use default
        emotion = {
            "label": "UNKNOWN",
            "intensity": distress,
            "sentiment": "neutral"
        }
except Exception as e:
    print(f"[emotion] Unexpected error: {e}")
    emotion = {"label": "UNKNOWN", "intensity": 0.5}

if r.status_code >= 400:
    print(
        "[emotion][deepgram] HTTP",
        r.status_code,
        "body:",
        r.text[:500],
    )
    return None

except Exception as e:
    print("[emotion][openai] error:", e)
    return None

import pytest
from app.agents.emotion import analyze_emotion

@pytest.mark.asyncio
async def test_emotion_heuristic_calm():
    emotion = await analyze_emotion(
        transcript="Everything is fine, thanks",
        distress=0.1
    )
    assert emotion['label'] == 'CALM'
    assert emotion['intensity'] < 0.3

@pytest.mark.asyncio
async def test_emotion_life_threatening():
    # Should detect high distress even with low audio score
    emotion = await analyze_emotion(
        transcript="Someone's been shot",
        distress=0.2  # Low distress (shock)
    )
    assert emotion['label'] == 'HIGHLY_DISTRESSED'
    assert emotion['intensity'] >= 0.8

@pytest.mark.asyncio
async def test_emotion_no_transcript():
    # Should work with audio only
    emotion = await analyze_emotion(
        transcript=None,
        distress=0.7
    )
    assert emotion['label'] in ['DISTRESSED', 'HIGHLY_DISTRESSED']

@pytest.mark.asyncio
async def test_emotion_with_service():
    """Test emotion + service classification together"""
    transcript = "My chest hurts really bad"
    distress = 0.6
    
    emotion = await analyze_emotion(transcript, distress)
    service = classify_service_and_tags(transcript, distress)
    
    assert emotion['label'] in ['TENSE', 'DISTRESSED']
    assert service['category'] == 'EMS'
    assert 'CARDIAC_EVENT' in service['tags']