Разработка AI-системы Speech Analytics для анализа записей звонков

Разработка AI-системы Speech Analytics для анализа записей звонков

Speech Analytics обрабатывает 100% записей звонков, извлекая структурированную информацию: темы, тональность, ключевые слова, паттерны поведения, compliance-нарушения. Заменяет выборочный ручной аудит на полное автоматическое покрытие.

Архитектура Speech Analytics платформы

Recording → STT → NLP Pipeline → Feature Extraction → Storage → Analytics
              ↓        ↓               ↓                              ↓
         Transcript  Entities      Topics/Categories              Dashboard
                    Sentiment      Compliance flags               Reports
                    Intent         Script adherence               Alerts

NLP pipeline пост-обработки

from dataclasses import dataclass
from typing import Optional

@dataclass
class CallAnalysis:
    call_id: str
    transcript: str
    duration: float

    # NLP результаты
    topics: list[str]
    entities: dict          # имена, суммы, даты, продукты
    sentiment_timeline: list[dict]  # тональность по сегментам
    overall_sentiment: str

    # Compliance
    compliance_flags: list[dict]    # нарушения стандартов
    required_phrases_present: dict  # обязательные фразы

    # Качество
    script_adherence_score: float
    professionalism_score: float
    resolution_status: str          # resolved | unresolved | escalated

    # Ключевые моменты
    key_moments: list[dict]         # важные моменты в записи
    action_items: list[str]         # задачи по итогам

class SpeechAnalyticsPipeline:
    async def analyze_call(self, transcript: dict) -> CallAnalysis:
        full_text = self.format_transcript(transcript["turns"])

        # Параллельный запуск всех анализаторов
        results = await asyncio.gather(
            self.extract_topics(full_text),
            self.extract_entities(full_text),
            self.analyze_sentiment_timeline(transcript["turns"]),
            self.check_compliance(full_text, transcript),
            self.evaluate_script_adherence(full_text),
            self.extract_key_moments(transcript),
        )

        return CallAnalysis(
            call_id=transcript["call_id"],
            transcript=full_text,
            duration=transcript["duration"],
            topics=results[0],
            entities=results[1],
            sentiment_timeline=results[2]["timeline"],
            overall_sentiment=results[2]["overall"],
            compliance_flags=results[3],
            required_phrases_present=results[4]["required_phrases"],
            script_adherence_score=results[4]["score"],
            professionalism_score=results[4]["professionalism"],
            resolution_status=self.detect_resolution(full_text),
            key_moments=results[5],
            action_items=await self.extract_action_items(full_text)
        )

Тематический анализ (Topic Modeling)

async def extract_topics(text: str) -> list[str]:
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "system",
            "content": """Определи 1-3 основные темы звонка.
            Выбирай из: оплата, доставка, технические проблемы, возврат,
            жалоба, консультация, продажа, информация.
            Или предложи свою тему. JSON: ["тема1", "тема2"]"""
        }, {"role": "user", "content": text[:3000]}],
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content).get("topics", [])

Compliance мониторинг

REQUIRED_PHRASES = {
    "greeting": ["добрый день", "здравствуйте", "меня зовут"],
    "verification": ["подтвердите", "назовите", "последние 4 цифры"],
    "farewell": ["до свидания", "хорошего дня", "спасибо за обращение"],
    "gdpr_consent": ["вы соглашаетесь", "запись ведётся", "качество обслуживания"],
}

FORBIDDEN_PHRASES = [
    "это не моя проблема", "я не знаю", "не могу помочь",
    "позвоните позже", "перезвоните завтра"
]

def check_compliance(transcript: str) -> dict:
    violations = []
    required_present = {}

    for category, phrases in REQUIRED_PHRASES.items():
        found = any(p in transcript.lower() for p in phrases)
        required_present[category] = found
        if not found:
            violations.append({"type": "missing_required", "category": category})

    for phrase in FORBIDDEN_PHRASES:
        if phrase in transcript.lower():
            violations.append({"type": "forbidden_phrase", "phrase": phrase})

    return {"violations": violations, "required_present": required_present}

Массовый поиск по паттернам

async def search_calls_by_pattern(
    pattern: str,
    date_range: tuple,
    operator_ids: list = None
) -> list[dict]:
    """Полнотекстовый поиск по транскриптам звонков"""
    query = {
        "text": {"$regex": pattern, "$options": "i"},
        "date": {"$gte": date_range[0], "$lte": date_range[1]}
    }
    if operator_ids:
        query["operator_id"] = {"$in": operator_ids}

    return await db.call_analyses.find(query).to_list(100)

Сроки: базовый Speech Analytics с 5 анализаторами — 6–8 недель. Полная платформа с дашбордом — 3–4 месяца.