Разработка AI-системы анализа тональности клиента в реальном времени (Live Sentiment)

Разработка AI-системы анализа тональности клиента в реальном времени (Live Sentiment)

Live Sentiment Analysis непрерывно оценивает эмоциональное состояние клиента в ходе звонка и выводит индикатор на экран оператора или супервайзера. Позволяет реагировать на изменение тональности до того, как клиент положит трубку.

Подходы к real-time анализу тональности

Текстовый анализ транскрипта (задержка 300–500 мс):

from transformers import pipeline
import asyncio

sentiment_analyzer = pipeline(
    "sentiment-analysis",
    model="blanchefort/rubert-base-cased-sentiment-rurewiews",
    tokenizer="blanchefort/rubert-base-cased-sentiment-rurewiews"
)

async def analyze_utterance_sentiment(text: str) -> dict:
    result = sentiment_analyzer(text[:512])[0]  # ограничиваем длину
    return {
        "label": result["label"],  # POSITIVE | NEGATIVE | NEUTRAL
        "score": result["score"],
        "text": text
    }

Акустический анализ голоса (без транскрибации, задержка <100 мс):

import librosa
import numpy as np

def extract_acoustic_features(audio_chunk: bytes, sr: int = 16000) -> dict:
    """Извлекаем просодические признаки для классификации эмоций"""
    audio = np.frombuffer(audio_chunk, dtype=np.int16).astype(np.float32) / 32768.0

    # Основной тон (F0) — маркер эмоционального состояния
    f0, _ = librosa.pyin(audio, fmin=80, fmax=400, sr=sr)
    f0_mean = np.nanmean(f0)
    f0_std = np.nanstd(f0)

    # Темп речи
    tempo, _ = librosa.beat.beat_track(y=audio, sr=sr)

    # Энергия
    rms = librosa.feature.rms(y=audio)[0]
    energy_mean = np.mean(rms)

    return {
        "f0_mean": float(f0_mean) if not np.isnan(f0_mean) else 0,
        "f0_std": float(f0_std) if not np.isnan(f0_std) else 0,
        "energy": float(energy_mean),
        "tempo": float(tempo)
    }

Fusion: текст + акустика

async def combined_sentiment(text: str, audio: bytes) -> dict:
    text_sentiment, acoustic_features = await asyncio.gather(
        analyze_utterance_sentiment(text),
        asyncio.get_event_loop().run_in_executor(
            None, extract_acoustic_features, audio
        )
    )

    # Высокая энергия + негативный текст = злость
    # Низкая энергия + негативный текст = расстройство
    emotion = classify_emotion(text_sentiment, acoustic_features)
    return {
        "sentiment": text_sentiment["label"],
        "emotion": emotion,
        "confidence": text_sentiment["score"],
        "acoustic_signals": acoustic_features
    }

WebSocket для real-time UI

@app.websocket("/sentiment-stream/{call_id}")
async def sentiment_stream(websocket: WebSocket, call_id: str):
    await websocket.accept()
    async for event in get_call_events(call_id):
        if event["type"] == "customer_utterance":
            sentiment = await analyze_utterance_sentiment(event["text"])
            await websocket.send_json({
                "timestamp": event["timestamp"],
                "text": event["text"],
                **sentiment
            })

Сроки: текстовый sentiment в реальном времени — 2–3 недели. Акустический + fusion — 4–6 недель.