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

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

Система автоматически обнаруживает критические ситуации в звонках: нарастающий негатив, угрозы, жалобы, риск оттока — и немедленно уведомляет супервайзера для вмешательства в реальном времени.

Детектируемые события

from enum import Enum

class EscalationTrigger(Enum):
    SUSTAINED_ANGER = "sustained_anger"        # злость >30 сек
    THREAT_LEGAL = "threat_legal"              # угроза судом/жалобой
    CHURN_RISK = "churn_risk"                  # намерение уйти
    COMPLAINT_ESCALATION = "complaint_esc"    # жалоба нарастает
    OPERATOR_CONFUSION = "operator_confused"   # оператор теряется
    SILENCE_ISSUE = "silence"                  # долгое молчание
    PROFANITY = "profanity"                    # нецензурная лексика

Детектор с накопленным контекстом

class NegativeEscalationDetector:
    def __init__(self, call_id: str):
        self.call_id = call_id
        self.sentiment_history = []
        self.consecutive_negative = 0
        self.escalated = False

    LEGAL_THREAT_PATTERNS = [
        "роспотребнадзор", "прокуратура", "суд", "иск",
        "жалоба", "претензия", "юрист", "адвокат"
    ]
    CHURN_PATTERNS = [
        "расторгну договор", "уйду", "другой оператор",
        "больше не буду", "отключусь"
    ]

    async def process_utterance(self, text: str, speaker: str) -> list[dict]:
        triggers = []

        if speaker != "customer":
            return []

        text_lower = text.lower()

        # Юридические угрозы
        if any(p in text_lower for p in self.LEGAL_THREAT_PATTERNS):
            triggers.append({
                "trigger": EscalationTrigger.THREAT_LEGAL,
                "severity": "critical",
                "evidence": text
            })

        # Риск оттока
        if any(p in text_lower for p in self.CHURN_PATTERNS):
            triggers.append({
                "trigger": EscalationTrigger.CHURN_RISK,
                "severity": "high",
                "evidence": text
            })

        # Накопленный негатив
        sentiment = await analyze_sentiment(text)
        if sentiment["label"] == "NEGATIVE":
            self.consecutive_negative += 1
        else:
            self.consecutive_negative = 0

        if self.consecutive_negative >= 3:
            triggers.append({
                "trigger": EscalationTrigger.SUSTAINED_ANGER,
                "severity": "high",
                "evidence": f"{self.consecutive_negative} негативных реплик подряд"
            })

        if triggers and not self.escalated:
            await self.notify_supervisor(triggers)

        return triggers

    async def notify_supervisor(self, triggers: list[dict]):
        self.escalated = True
        await notification_service.send({
            "call_id": self.call_id,
            "urgency": max(t["severity"] for t in triggers),
            "triggers": triggers,
            "operator_id": self.operator_id,
            "call_url": f"/monitor/calls/{self.call_id}"
        })
        # Push уведомление + звуковой сигнал на рабочем месте супервайзера
        await websocket_manager.notify_supervisors(self.call_id, triggers)

Дашборд супервайзера

Real-time список активных звонков с индикаторами тональности. При эскалации — карточка поднимается наверх с красной меткой и историей диалога.

Сроки: детектор негатива + уведомления — 3–4 недели. Дашборд супервайзера — +2–3 недели.