Привет, Хабр!
Выгорание операторов — распространенная проблема в кол-центрах. По разным оценкам, текучесть персонала здесь достигает 40–45%, а средний срок работы составляет 8–12 месяцев. Это приводит к дополнительным расходам на обучение, росту нагрузки на команду и снижению качества сервиса. При этом заметные изменения в поведении сотрудников обычно фиксируются слишком поздно — когда проблема уже стала системной.
Я Катя Саяпина, менеджер продукта МТС Exolve. В этом материале разберу способ раннего обнаружения таких изменений. Он опирается на статистические отклонения в поведении оператора и дополняет прямое общение с сотрудниками и сбор обратной связи в команде. Мы создадим на Python сервис, который объединит Telegram-бота, API МТС Exolve и LLM, развернутую на платформе MWS GPT.
Архитектура решения
Для отслеживания аномалий мы будем анализировать структуру и ритм диалога, сравнивать текущий день с историческими данными конкретного оператора и автоматически сигнализировать о нетипичных отклонениях. Архитектура строится вокруг простого ежедневного запуска — сервис работает как ночной аналитик, который собирает данные за прошедший день и формирует отчет.
Система будет работать по следующим шагам:
Получение данных
Скрипт запрашивает у API МТС Exolve транскрипции всех звонков за последние 24 часа. Формат данных включает сегменты речи, время и признак говорящего.Расчет метрик
Для каждог�� звонка вычисляются шесть статистических показателей, которые описывают ритм разговора: доля речи, латентность, доля тишины и другие параметры поведения оператора.Хранение истории
Метрики сохраняются в локальную базу SQLite. Этого достаточно для десятков тысяч записей и удобного получения выборок за 7–30 дней.Анализ отклонений
Для каждого оператора система берет его норму за последние две недели и передает ее и текущие значения в MWS GPT, которая дает оценку наличия риска. Такой подход учитывает индивидуальные особенности и снижает количество ложных срабатываний.Формирование отчета
При обнаружении отклонений сервис строит небольшой график с динамикой и сопровождает текстовым сообщением. Визуализация помогает быстро оценить, разовое ли это событие или начало тренда.Алертинг
Итоговый отчет отправляется в Telegram-бот. В сообщении содержится краткое описание проблемы и прикрепленный график.
Для этого нам потребуется один Python-скрипт, небольшая база данных и Telegram-бот — этого достаточно, чтобы ежедневно отсылать сигналы о состоянии команды и оперативно реагировать на изменения.
Как искать аномалии
Главная задача — понять, вел ли себя оператор сегодня так же, как обычно, или его поведение заметно изменилось. Для этого будем отслеживать сдвиги в привычном стиле общения оператора.
Если растет задержка ответа, увеличивается длительная тишина, меняется доля речи, появляются перебивания или снижается темп диалога, это может говорить об усталости и перегрузке. Такие сигналы не заменяют личную работу с сотрудником, но помогают увидеть изменения заранее, пока они не начали влиять на сервис.
Мы делаем два шага:
Формируем эталонные значения
Для каждого оператора агрегируем метрики за выбранный период, например, 10–14 дней, и получаем диапазон типичных значений.Сравниваем с текущим днем
После обработки звонков считаем усредненные показатели за сутки и проверяем, какие из них вышли за привычный диапазон выше заданного порога.
Набор метрик
Чтобы уловить изменения в поведении, система рассчитывает шесть показателей:
Доля речи оператора.
Задержка ответа на реплики клиента, измеренная по 95‑му перцентилю. Это значение, длиннее которого оказываются лишь 5% самых редких пауз. Такой подход позволяет учитывать почти все реакции оператора, но игнорировать единичные выбросы и фиксировать именно устойчивые изменения в скорости ответа.
Доля пауз дольше 1,5 секунд.
Интенсивность диалога — число смен говорящего в минуту.
Задержку перед первым ответом оператора на приветствие клиента.
Доля перебиваний — доля времени, когда оператор и клиент говорят одновременно.
Эти метрики — не стандарт, а рабочая гипотеза, основанная на практике, и пригодная для прикладного мониторинга. Они дают компактное описание стиля общения оператора и позволяют видеть сдвиги, которые сложно заметить при разборе отдельных звонков вручную.
Шаг 1. Подготовка окружения и сбор данных
Снача��а нужно настроить окружение и научиться забирать звонки из API МТС Exolve. Нам понадобятся requests для API-запросов, python-dotenv для конфигурации и schedule для периодического запуска.
pip install requests python‑dotenv schedule numpy matplotlib langchain‑community
Зачем они нужны:
requests — отправлять запросы в МТС Exolve;
python-dotenv — хранить токены в .env;
schedule — запускать скрипт раз в сутки;
numpy — считать статистику;
matplotlib — строить графики для алертов.
Для хранения исторических данных используем простую базу SQLite. При первом запуске скрипт автоматически развернет базу данных и создаст таблицу для хранения метрик. Полный код, как и весь проект, можно найти в этом репозитории на GitHub.
Шаг 2. Расчет метрик
Теперь переходим к основе решения — функции, которая из одного звонка делает компактный профиль поведения оператора.
В МТС Exolve есть речевая аналитика: она автоматически считает метрики по речи, молчанию, перебиваниям, формирует семантическое резюме разговора и классифицирует фразы. Такой инструмент закрывает большинство практических задач. Но в нашем примере важно полностью контролировать логику расчетов, поэтому мы используем только текстовую расшифровку звонка и считаем показатели самостоятельно.
Внутри JSON-объекта с транскрипцией звонка есть:
duration — длительность звонка в секундах;
chunks — список фрагментов речи с полями:
channel_tag — кто говорит (1 — клиент, 2 — оператор),
start_time и end_time — границы фрагмента в секундах.
На их основе функция calculate_metrics:
проверяет, что в звонке есть данные и ненулевая длительность;
разделяет реплики клиента и оператора;
проходит по всем паузам между фрагментами;
считает шесть метрик по следующей логике:
доля речи оператора atr и интенсивность диалога в сменах говорящего в минуту tpm вычисляются простым делением: длительность речи оператора делится на общую, а количество реплик — на время;
для скорости реакции по 95-му перцентилю p95_latency и доли «мертвой» тишины dead_air_ratio мы итерируемся по паузам между репликами. Паузы между клиентом и оператором попадают в latency, а все паузы длиннее 1,5 секунд — в dead-air;
задержку перед первым ответом first_response_time — это пауза между самой первой репликой клиента и первым ответом оператора;
для долей перебиваний agent_overlap и client_overlap мы вложенным циклом находим пересечения. Если реплика оператора началась позже реплики клиента, но наложилась на нее — значит, сотрудник перебил клиента. И наоборот. Мы считаем эти показатели раздельно.
Результат — один словарь, который можно сразу сохранять в базу.
# metrics_calculator.py import numpy as np def calculate_metrics(call_data: dict) -> dict | None: """ Принимает JSON одного звонка и возвращает словарь с шестью метриками. """ chunks = call_data.get("chunks", []) if not chunks: return None call_duration_seconds = call_data.get("duration", 0) if call_duration_seconds == 0: return None agent_chunks = [c for c in chunks if c.get('channel_tag') == 2] client_chunks = [c for c in chunks if c.get('channel_tag') == 1] # 1. Расчет ATR (Agent Talk Ratio) agent_speech_duration = sum(c['end_time'] - c['start_time'] for c in agent_chunks) total_speech_duration = sum(c['end_time'] - c['start_time'] for c in chunks) atr = agent_speech_duration / total_speech_duration if total_speech_duration > 0 else 0 # 2. Расчет TPM (Turns Per Minute) tpm = len(chunks) / (call_duration_seconds / 60) if call_duration_seconds > 0 else 0 # 3. Расчет Response Latency (p95) и Dead-air latencies = [] dead_air_duration = 0 DEAD_AIR_THRESHOLD = 1.5 for i in range(1, len(chunks)): prev_chunk = chunks[i - 1] current_chunk = chunks[i] pause = current_chunk['start_time'] - prev_chunk['end_time'] if pause < 0: continue if prev_chunk['channel_tag'] == 1 and current_chunk['channel_tag'] == 2: latencies.append(pause) if pause > DEAD_AIR_THRESHOLD: dead_air_duration += pause p95_latency = np.percentile(latencies, 95) if latencies else 0 dead_air_ratio = dead_air_duration / call_duration_seconds if call_duration_seconds > 0 else 0 # 4. Расчет First Response Time first_response_time = -1 if client_chunks and agent_chunks: first_client_chunk = client_chunks[0] # Ищем первый ответ оператора ПОСЛЕ первой реплики клиента first_agent_response = next((ac for ac in agent_chunks if ac['start_time'] > first_client_chunk['end_time']), None) if first_agent_response: first_response_time = first_agent_response['start_time'] - first_client_chunk['end_time'] # 5. Расчет Overlap Ratio (кто кого перебил) agent_overlap = 0 client_overlap = 0 for ac in agent_chunks: for cc in client_chunks: overlap = max(0, min(ac['end_time'], cc['end_time']) - max(ac['start_time'], cc['start_time'])) if overlap > 0: if ac['start_time'] > cc['start_time']: agent_overlap += overlap else: client_overlap += overlap agent_ratio = agent_overlap / call_duration_seconds if call_duration_seconds > 0 else 0 client_ratio = client_overlap / call_duration_seconds if call_duration_seconds > 0 else 0 return { "atr": round(atr, 2), "p95_latency": round(p95_latency, 2), "dead_air_ratio": round(dead_air_ratio, 2), "tpm": round(tpm, 2), "first_response_time": round(first_response_time, 2), "agent_overlap_ratio": round(agent_ratio, 2), "client_overlap_ratio": round(client_ratio, 2) }
Функция специально написана максимально прямолинейно: без сторонних зависимостей, только базовая работа со списками и числами. Если в звонке нет данных или длительность равна нулю, она возвращает None, и такой звонок можно просто пропустить при обработке.
Шаг 3. Поиск аномалий
Теперь, когда у нас есть метрики за текущий день и норма оператора, мы не будем задавать жесткие пороговые правила вручную. Вместо этого передадим решение LLM. Для модели gpt-oss-120b от OpenAI формируем текстовый промпт со всей статистикой и просим ее выступить в роли аналитика, который оценивает наличие отклонений и степень риска.
Вот как выглядит функция, которая обращается к MWS GPT за оценкой:
# ai_analyzer.py import os import requests import json def get_ai_verdict(manager_id, today_metrics, baseline_metrics): """Отправляет метрики в MWS GPT для экспертной оценки.""" token = os.getenv("MTS_AI_API_KEY") url = "https://api.gpt.mws.ru/v1/chat/completions" # Формируем промпт с цифрами prompt = f""" Ты — аналитик колл-центра. Оцени риск выгорания оператора {manager_id}. Сравни его показатели за сегодня с его личной нормой (среднее за 14 дней). 1. Скорость ответа (p95 Latency): - Сегодня: {today_metrics['p95_latency']:.2f}с - Норма: {baseline_metrics.get('avg_latency', 0):.2f}с 2. Доля тишины (Dead-air): - Сегодня: {today_metrics['dead_air_ratio'] * 100:.1f}% - Норма: {baseline_metrics.get('avg_dead_air', 0):.1f}% Если показатели сильно хуже нормы (рост задержки или молчания), это высокий риск. Верни JSON с двумя полями: 1. "risk_level": "Low", "Medium" или "High". 2. "reason": "Краткое объяснение для руководителя (1 предложение на русском)". """ payload = { "model": "gpt-oss-120b", "messages": [{"role": "user", "content": prompt}], "temperature": 0.1, "response_format": {"type": "json_object"} } try: resp = requests.post(url, json=payload, headers={"Authorization": f"Bearer {token}"}) resp.raise_for_status() ai_response = resp.json()['choices'][0]['message']['content'] return json.loads(ai_response) except Exception as e: print(f"Ошибка AI: {e}") return {"risk_level": "Unknown", "reason": "Ошибка анализа"}
Шаг 4. Отправка сообщения в Telegram
На этом этапе мы формируем сигнал о возможном выгорании. Из дневных метрик и истории по оператору ищем отклонения от его обычного поведения и отправляем руководителю понятное уведомление — текст плюс график.
Логика работы:
для каждого оператора считаем среднее значение метрики за последние 14 дней;
считаем среднее значение за текущий день;
сравниваем текущий показатель с нормой и, если отклонение выше порога (например, +50% по p95_latency), считаем это аномалией;
учитываем оценку от MWS GPT: если risk_level высокий, подтверждаем сигнал;
строим график метрики за последние 14 дней вместе с текущим значением и отправляем его в Telegram.
График пишем сразу в память через BytesIO, чтобы не создавать временные файлы, и передаем в Telegram как вложение.
# chart_generator.py import io import matplotlib.pyplot as plt def create_anomaly_chart(dates: list, values: list, baseline: float, anomaly_value: float, metric_name: str) -> io.BytesIO: """Строит график динамики метрики и сохраняет его в байтовый буфер.""" plt.style.use('seaborn-v0_8-whitegrid') fig, ax = plt.subplots(figsize=(10, 5), dpi=100) ax.plot(dates, values, marker='o', linestyle='-', label='Динамика за 14 дней') ax.axhline(y=baseline, color='grey', linestyle='--', label=f'Норма ({baseline:.2f})') ax.scatter(dates[-1], anomaly_value, color='red', s=100, zorder=5, label='Аномалия сегодня!') ax.set_title(f'Аномалия по метрике: {metric_name}', fontsize=16) ax.set_ylabel('Значение метрики') ax.tick_params(axis='x', labelrotation=45) ax.legend() fig.tight_layout() # Сохраняем график в буфер памяти buf = io.BytesIO() fig.savefig(buf, format='png') buf.seek(0) plt.close(fig) return buf
На вход этой функции приходят подготовленные данные — списки дат и значений, эталонное и текущее значение. На выходе — готовый PNG в памяти, который можно сразу отправлять в Telegram.
Если аномалия найдена, мы формируем и отправляем сообщение в Telegram. Для этого понадобится токен Telegram-бота и ID чата, которые нужно добавить в ваш .env файл.
Функция send_telegram_alert работает в двух режимах:
если передан image_buffer — отправляет график с подписью;
если нет — отправляет только текст.
# telegram_alerter import os import requests import io def escape_markdown_v2(text: str) -> str: """Экранирует специальные символы для Telegram MarkdownV2.""" escape_chars = r'_*[]()~`>#+-=|{}.!' return ''.join(f'\\{char}' if char in escape_chars else char for char in text) def send_telegram_alert(message: str, image_buffer: io.BytesIO = None): """Отправляет сообщение и/или изображение в Telegram.""" TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN") CHAT_ID = os.getenv("TELEGRAM_CHAT_ID") if not TELEGRAM_TOKEN or not CHAT_ID: return try: # Экранируем сообщение перед отправкой safe_message = escape_markdown_v2(message) if image_buffer: url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendPhoto" files = {'photo': ('anomaly_chart.png', image_buffer, 'image/png')} data = {'chat_id': CHAT_ID, 'caption': safe_message, 'parse_mode': 'MarkdownV2'} requests.post(url, files=files, data=data, timeout=10) else: url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage" payload = {"chat_id": CHAT_ID, "text": safe_message, "parse_mode": "MarkdownV2"} requests.post(url, json=payload, timeout=10) print("✅ Алерт успешно отправлен в Telegram.") except Exception as e: print(f"❌ Ошибка отправки в Telegram: {e}")
В итоге руководитель получает сигнал: какая метрика ушла из привычного диапазона, в какую сторону и насколько, плюс наглядный график с историей.
В итоге у нас получился рабочий сервис, который строит поведенческий профиль по звонкам и отслеживает его отклонения.
Это снижает вероятность того, что изменения в поведении оператора останутся незамеченными, и дает возможность реагировать на проблему до того, как она превратится в текучку, жалобы и просадку качества сервиса.
Возможности для развития:
Web-дашборд. Добавить простой интерфейс на Dash/Streamlit для просмотра динамики метрик по операторам и периодам.
Динамический поиск аномалий. Заменить фиксированные пороги на статистические методы. Например, Z-оценка, межквартильный размах и другие.
Фиксация позитивныхе отклонений. Отмечать не только ухудшения, но и устойчивые улучшения показателей — для поощрения и обмена опытом.
