Разработка AI-системы детекции стресса/агрессии в голосе клиента

Разработка AI-системы детекции стресса/агрессии в голосе клиента

Акустическая детекция стресса и агрессии работает без анализа слов — только по характеристикам голоса: темп речи, частота основного тона (F0), энергия, тремор. Это позволяет реагировать за 2–3 секунды, до того как человек произнесёт угрозу.

Акустические маркеры стресса и агрессии

import librosa
import numpy as np
from dataclasses import dataclass

@dataclass
class EmotionalAcoustics:
    f0_mean: float      # средняя частота (агрессия: рост >20%)
    f0_range: float     # диапазон тона (стресс: сужение)
    f0_std: float       # вариабельность
    speaking_rate: float # темп (стресс: ускорение или замедление)
    energy_mean: float  # громкость (агрессия: значительный рост)
    jitter: float       # тремор голоса (стресс: рост)
    shimmer: float      # неравномерность амплитуды
    hnr: float          # гармоническое отношение (стресс: снижение)

def extract_stress_features(audio: np.ndarray, sr: int = 16000) -> EmotionalAcoustics:
    # Основной тон
    f0, voiced_flag, _ = librosa.pyin(audio, fmin=75, fmax=500, sr=sr)
    f0_voiced = f0[voiced_flag & ~np.isnan(f0)]

    # Энергия
    rms = librosa.feature.rms(y=audio, frame_length=2048, hop_length=512)[0]

    # Spectral features для HNR proxy
    zcr = librosa.feature.zero_crossing_rate(audio)[0]

    return EmotionalAcoustics(
        f0_mean=float(np.mean(f0_voiced)) if len(f0_voiced) > 0 else 0,
        f0_range=float(np.ptp(f0_voiced)) if len(f0_voiced) > 0 else 0,
        f0_std=float(np.std(f0_voiced)) if len(f0_voiced) > 0 else 0,
        speaking_rate=estimate_speaking_rate(audio, sr),
        energy_mean=float(np.mean(rms)),
        jitter=estimate_jitter(f0_voiced),
        shimmer=estimate_shimmer(rms),
        hnr=float(1.0 / (np.mean(zcr) + 1e-8))
    )

Классификатор на ML

from sklearn.ensemble import GradientBoostingClassifier
import joblib

class StressAggressionClassifier:
    LABELS = {0: "neutral", 1: "stressed", 2: "aggressive"}

    def __init__(self, model_path: str):
        self.model = joblib.load(model_path)
        self.baseline = {}  # персональный baseline первых 10 секунд

    def classify(
        self,
        features: EmotionalAcoustics,
        baseline: EmotionalAcoustics = None
    ) -> dict:
        feat_vector = self._to_vector(features)

        if baseline:
            # Нормализуем относительно персонального baseline
            base_vector = self._to_vector(baseline)
            feat_vector = (feat_vector - base_vector) / (base_vector + 1e-8)

        proba = self.model.predict_proba([feat_vector])[0]
        label_id = np.argmax(proba)

        return {
            "label": self.LABELS[label_id],
            "confidence": float(proba[label_id]),
            "probabilities": {self.LABELS[i]: float(p) for i, p in enumerate(proba)}
        }

Обучение на датасете эмоций

Используем RAVDESS (английский) + EMOVO + кастомные записи с разметкой. Для русского — RESD (Russian Emotional Speech Dataset) или собственная разметка.

Производительность после обучения: accuracy ~78–85% на 3 классах (neutral / stressed / aggressive).

Интеграция в поток звонка

CLASSIFICATION_WINDOW_SEC = 3.0  # анализируем каждые 3 секунды

async def continuous_emotion_monitoring(call_id: str, audio_stream):
    classifier = StressAggressionClassifier("models/stress_model.pkl")
    baseline = None
    buffer = bytearray()

    async for chunk in audio_stream:
        buffer.extend(chunk)
        if len(buffer) >= 16000 * CLASSIFICATION_WINDOW_SEC * 2:
            audio = np.frombuffer(buffer, dtype=np.int16).astype(np.float32) / 32768.0
            features = extract_stress_features(audio)

            # Первые 10 сек — устанавливаем baseline
            if baseline is None and len(buffer) < 160000:
                baseline = features
                buffer = bytearray()
                continue

            result = classifier.classify(features, baseline)
            if result["label"] == "aggressive" and result["confidence"] > 0.75:
                await trigger_aggression_alert(call_id, result)

            buffer = bytearray()

Сроки: классификатор на готовом датасете — 2–3 недели. Сбор датасета и обучение с нуля — 2–3 месяца.