Привет, Хабр!

Выгорание операторов — распространенная проблема в кол-центрах. По разным оценкам, текучесть персонала здесь достигает 40–45%, а средний срок работы составляет 8–12 месяцев. Это приводит к дополнительным расходам на обучение, росту нагрузки на команду и снижению качества сервиса. При этом заметные изменения в поведении сотрудников обычно фиксируются слишком поздно — когда проблема уже стала системной.

Я Катя Саяпина, менеджер продукта МТС Exolve. В этом материале разберу способ раннего обнаружения таких изменений. Он опирается на статистические отклонения в поведении оператора и дополняет прямое общение с сотрудниками и сбор обратной связи в команде. Мы создадим на Python сервис, который объединит Telegram-бота, API МТС Exolve и LLM, развернутую на платформе MWS GPT.

Архитектура решения

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

Система будет работать по следующим шагам:

1. Получение данных
   Скрипт запрашивает у API МТС Exolve транскрипции всех звонков за последние 24 часа. Формат данных включает сегменты речи, время и признак говорящего.

2. Расчет метрик
   Для каждого звонка вычисляются шесть статистических показателей, которые описывают ритм разговора: доля речи, латентность, доля тишины и другие параметры поведения оператора.

3. Хранение истории
   Метрики сохраняются в локальную базу SQLite. Этого достаточно для десятков тысяч записей и удобного получения выборок за 7–30 дней.

4. Анализ отклонений
   Для каждого оператора система берет его норму за последние две недели и передает ее и текущие значения в MWS GPT, которая дает оценку наличия риска. Такой подход учитывает индивидуальные особенности и снижает количество ложных срабатываний.

5. Формирование отчета
   При обнаружении отклонений сервис строит небольшой график с динамикой и сопровождает текстовым сообщением. Визуализация помогает быстро оценить, разовое ли это событие или начало тренда.

6. Алертинг
   Итоговый отчет отправляется в Telegram-бот. В сообщении содержится краткое описание проблемы и прикрепленный график.

Для этого нам потребуется один Python-скрипт, небольшая база данных и Telegram-бот — этого достаточно, чтобы ежедневно отсылать сигналы о состоянии команды и оперативно реагировать на изменения.

Как искать аномалии

Главная задача — понять, вел ли себя оператор сегодня так же, как обычно, или его поведение заметно изменилось. Для этого будем отслеживать сдвиги в привычном стиле общения оператора. 

Если растет задержка ответа, увеличивается длительная тишина, меняется доля речи, появляются перебивания или снижается темп диалога, это может говорить об усталости и перегрузке. Такие сигналы не заменяют личную работу с сотрудником, но помогают увидеть изменения заранее, пока они не начали влиять на сервис.

Мы делаем два шага:

1. Формируем эталонные значения
   Для каждого оператора агрегируем метрики за выбранный период, например, 10–14 дней, и получаем диапазон типичных значений.

2. Сравниваем с текущим днем
   После обработки звонков считаем усредненные показатели за сутки и проверяем, какие из них вышли за привычный диапазон выше заданного порога.

Набор метрик

Чтобы уловить изменения в поведении, система рассчитывает шесть показателей:

1. Доля речи оператора.

2. Задержка ответа на реплики клиента, измеренная по 95‑му перцентилю. Это значение, длиннее которого оказываются лишь 5% самых редких пауз. Такой подход позволяет учитывать почти все реакции оператора, но игнорировать единичные выбросы и фиксировать именно устойчивые изменения в скорости ответа.

3. Доля пауз дольше 1,5 секунд.

4. Интенсивность диалога — число смен говорящего в минуту.

5. Задержку перед первым ответом оператора на приветствие клиента.

6. Доля перебиваний — доля времени, когда оператор и клиент говорят одновременно.

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

Шаг 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-оценка, межквартильный размах и другие.

• Фиксация позитивныхе отклонений. Отмечать не только ухудшения, но и устойчивые улучшения показателей — для поощрения и обмена опытом.