Spaces:
Sleeping
Sleeping
File size: 6,942 Bytes
f1ef7e2 | 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 | """
Audio escalation scoring for call-center sentiment analysis.
This module combines emotion probabilities, audio features, timeline behavior,
and prediction confidence into one audio escalation score.
The score is designed for dashboard triage:
- Low: emotionally calm or low-risk
- Medium: some negative/emotional signal
- High: strong anger/stress or worsening emotional pattern
- Critical: intense negative emotional state with strong voice/timeline risk
The score is not meant to replace human QA. It prioritizes calls that need review.
"""
from dataclasses import dataclass
from typing import Optional
import numpy as np
from src.sentiment_config import IntensityLevel, SentimentShift
from src.sentiment_schema import AudioFeatureSummary, EmotionProbabilities, SentimentSegment
@dataclass(frozen=True)
class EscalationScoreBreakdown:
"""
Detailed score components used for debugging/reporting.
This is useful later if we want to show why a call was marked high risk.
"""
emotion_risk: float
voice_risk: float
timeline_risk: float
uncertainty_adjustment: float
final_score: float
def level_to_score(level: IntensityLevel) -> float:
"""
Convert Low / Medium / High feature levels into numeric risk values.
"""
if level == IntensityLevel.HIGH:
return 1.0
if level == IntensityLevel.MEDIUM:
return 0.5
if level == IntensityLevel.LOW:
return 0.0
return 0.0
def sentiment_shift_to_score(shift: SentimentShift) -> float:
"""
Convert sentiment shift into numeric risk.
Worsened = higher risk.
Improved = lower risk.
Mixed = medium/high because the call was emotionally unstable.
"""
if shift == SentimentShift.WORSENED:
return 1.0
if shift == SentimentShift.MIXED:
return 0.7
if shift == SentimentShift.UNCHANGED:
return 0.3
if shift == SentimentShift.IMPROVED:
return 0.0
return 0.0
def calculate_emotion_risk(probabilities: EmotionProbabilities) -> float:
"""
Calculate emotion-only risk.
Anger and stress/fear are weighted strongly because they are important
escalation indicators in call-center conversations.
"""
emotion_risk = (
0.35 * probabilities.anger
+ 0.25 * probabilities.stress_probability()
+ 0.25 * probabilities.negative_probability()
+ 0.15 * probabilities.fear
)
return float(np.clip(emotion_risk, 0.0, 1.0))
def calculate_voice_risk(audio_features: AudioFeatureSummary) -> float:
"""
Calculate risk from interpretable audio features.
High loudness, high pitch variability, fast speech, frequent pauses, and
long silence can all indicate emotional escalation or poor call handling.
"""
vocal_intensity_score = level_to_score(audio_features.vocal_intensity)
pitch_variability_score = level_to_score(audio_features.pitch_variability)
speech_rate_score = level_to_score(audio_features.speech_rate)
pause_frequency_score = level_to_score(audio_features.pause_frequency)
long_silence_score = 1.0 if audio_features.long_silence_detected else 0.0
voice_risk = (
0.30 * vocal_intensity_score
+ 0.25 * pitch_variability_score
+ 0.20 * speech_rate_score
+ 0.15 * pause_frequency_score
+ 0.10 * long_silence_score
)
return float(np.clip(voice_risk, 0.0, 1.0))
def calculate_timeline_risk(
timeline: list[SentimentSegment],
emotional_volatility: IntensityLevel,
audio_sentiment_shift: SentimentShift,
) -> float:
"""
Calculate risk from the sentiment timeline.
For short clips, this will usually be close to the single segment risk.
For long calls, it captures peak emotion, average emotion, volatility, and
whether the emotional state improved or worsened.
"""
if not timeline:
average_segment_risk = 0.0
peak_segment_risk = 0.0
else:
segment_risks = [segment.risk_score for segment in timeline]
average_segment_risk = float(np.mean(segment_risks))
peak_segment_risk = float(np.max(segment_risks))
volatility_score = level_to_score(emotional_volatility)
shift_score = sentiment_shift_to_score(audio_sentiment_shift)
timeline_risk = (
0.35 * average_segment_risk
+ 0.30 * peak_segment_risk
+ 0.20 * volatility_score
+ 0.15 * shift_score
)
return float(np.clip(timeline_risk, 0.0, 1.0))
def calculate_uncertainty_adjustment(
uncertain_prediction: bool,
prediction_confidence: float,
) -> float:
"""
Calculate score adjustment for uncertain predictions.
If prediction is uncertain, we slightly reduce the final automated score
because the model is not confident. The separate uncertain_prediction flag
should still tell the dashboard to review the result manually.
"""
if not uncertain_prediction:
return 1.0
if prediction_confidence < 0.50:
return 0.85
return 0.90
def calculate_audio_escalation_score(
probabilities: EmotionProbabilities,
audio_features: AudioFeatureSummary,
timeline: Optional[list[SentimentSegment]] = None,
emotional_volatility: IntensityLevel = IntensityLevel.UNKNOWN,
audio_sentiment_shift: SentimentShift = SentimentShift.UNKNOWN,
prediction_confidence: float = 1.0,
uncertain_prediction: bool = False,
) -> EscalationScoreBreakdown:
"""
Calculate final audio escalation score.
The final score combines:
- emotion probabilities
- voice feature risk
- timeline risk
- uncertainty adjustment
Returns:
EscalationScoreBreakdown with component values and final score.
"""
if timeline is None:
timeline = []
emotion_risk = calculate_emotion_risk(probabilities)
voice_risk = calculate_voice_risk(audio_features)
timeline_risk = calculate_timeline_risk(
timeline=timeline,
emotional_volatility=emotional_volatility,
audio_sentiment_shift=audio_sentiment_shift,
)
uncertainty_adjustment = calculate_uncertainty_adjustment(
uncertain_prediction=uncertain_prediction,
prediction_confidence=prediction_confidence,
)
# Emotion is still the main signal, but voice/timeline features make the
# score more useful for call-center triage.
combined_score = (
0.50 * emotion_risk
+ 0.25 * voice_risk
+ 0.25 * timeline_risk
)
final_score = combined_score * uncertainty_adjustment
return EscalationScoreBreakdown(
emotion_risk=float(np.clip(emotion_risk, 0.0, 1.0)),
voice_risk=float(np.clip(voice_risk, 0.0, 1.0)),
timeline_risk=float(np.clip(timeline_risk, 0.0, 1.0)),
uncertainty_adjustment=float(np.clip(uncertainty_adjustment, 0.0, 1.0)),
final_score=float(np.clip(final_score, 0.0, 1.0)),
) |