Привет, Хабр!
В этой статье разберём, как настроить полный сценарий: от вебхуков в МТС Exolve до автоматической оценки звонков с помощью GigaChat и LangChain. По разным исследованиям, менеджеры по качеству тратят до 60% рабочего времени на прослушивание диалогов и при этом успевают проверять лишь 5–10% звонков. Мы соберём сервис на Python, который автоматически обрабатывает каждый звонок, расшифровывает аудио, прогоняет диалог через модель и возвращает структурированный JSON по чек-листу оценки оператора. Такой подход снижает ручную нагрузку и даёт воспроизводимую оценку в реальном времени.
Архитектура и точка входа
Для обработки звонков в реальном времени нам нужен простой событийный контур. Вместо того чтобы периодически выгружать данные, мы будем реагировать на уведомления от МТС Exolve сразу. Под это подходит небольшое Flask-приложение, которое принимает вебхуки и передаёт управление в основную логику.
Схема решения соответствует классическому ETL-подходу, только онлайн:
Extract. Вебхуки МТС Exolve дают два ключевых сигнала. Событие начала звонка передаёт uid и sip_id, а событие готовой транскрипции (type: "trc") сообщает, что аудио уже обработано и текст можно запрашивать.
Transform. После сигнала мы получаем транскрипт по uid, анализируем речь оператора с помощью GigaChat и приводим результат к строгой Pydantic-схеме. Это обеспечивает предсказуемую структуру данных.
Load. Итог собираем в JSON: добавляем ID звонка, менеджера, оценку по чек-листу и отдаём дальше — в базу, дашборд или внешний сервис.
Чтобы не задерживать платформу, тяжёлая обработка запускается в отдельном потоке. Эндпоинт отвечает 200 OK сразу, а сетевые запросы и вызовы модели выполняются в фоне. В примере используется простой словарь для хранения связки uid → sip_id, но в продакшене лучше применить Redis или другое in-memory хранилище.
Перед началом работы установите зависимости:
pip install Flask requests python-dotenv langchain langchain-gigachat pydantic
и создайте файл .env для ключей МТС Exolve и GigaChat.
Extract: настраиваем вебхуки и ловим сигналы
Точка входа в систему — HTTP-эндпоинт, который принимает вебхуки от МТС Exolve. Для обработки звонков нам достаточно двух событий из их полного жизненного цикла.
Начало звонка (type: "b"). Это первое уведомление, которое приходит почти мгновенно. В нём содержится связка uid с ID звонка и sip_id оператора. Мы фиксируем эту информацию, чтобы знать, какой сотрудник вёл разговор.
Готовность транскрипции (type: "trc"). Это событие приходит после завершения звонка и сообщает, что аудио обработано и текстовую расшифровку можно запрашивать по тому же uid. Это наш сигнал переходить к анализу.
Такой двухшаговый подход позволяет сначала сохранить участников звонка, а затем, когда текст готов, запускать основную обработку.
Фронт этого шага реализован на небольшом Flask-сервисе. Важно, что мы не держим МТС Exolve в ожидании. Эндпоинт отвечает 200 OK сразу, а ресурсоёмкая логика — запрос транскрипции и работа модели — выполняется в отдельном потоке. Благодаря этому сервис остаётся отзывчивым даже при высокой нагрузке.
from flask import Flask, request, jsonify import threading app = Flask(__name__) # Временное хранилище. В production-системе лучше использовать Redis. call_manager_storage = {} storage_lock = threading.Lock() @app.route('/exolve-webhook', methods=['POST']) def handle_webhook(): data = request.json event_type = data.get('type') uid = data.get('uid') if event_type == 'b' and uid: # Begin - начало звонка sip_id = data.get('sip_id') if sip_id: with storage_lock: call_manager_storage[str(uid)] = sip_id print(f"📞 Звонок {uid} начат менеджером {sip_id}. Связка сохранена.") elif event_type == 'trc' and uid: # Call Transcribation Ready print(f"📄 Транскрипция для звонка {uid} готова. Запускаю обработку...") thread = threading.Thread(target=process_call, args=(str(uid),)) thread.start() return jsonify({"status": "ok"}), 200
Таким образом, у нас получился простой и надёжный асинхронный приёмник событий. Сервис в реальном времени принимает вебхуки МТС Exolve, фиксирует ключевые идентификаторы uid и sip_id и запускает обработку в фоне, не блокируя входящие запросы. Теперь переходим ко второму шагу — запросим по uid саму транскрипцию разговора и преобразуем её в структурированные данные.
Transform: превращаем речь в структурированные данные
После сигнала о готовности транскрипции запрашиваем текст разговора по uid. Метод GetTranscribation в API МТС Exolve возвращает массив фрагментов с указанием, кто говорит в каждом из них. Для анализа нам нужны только реплики оператора, поэтому мы отбираем элементы с channel_tag = 2 и объединяем их в один текст. Получив чистую речь менеджера, мы можем передавать её в модель для оценки.
import requests import os from dotenv import load_dotenv load_dotenv() def get_transcript_by_id(call_id: str) -> str | None: """Делает точечный запрос к Exolve API за транскрипцией одного звонка.""" url = "https://api.exolve.ru/statistics/call-record/v1/GetTranscribation" headers = {"Authorization": f"Bearer {os.getenv('EXOLVE_API_KEY')}"} # Важно: этот метод ожидает 'uid' в теле запроса payload = {"uid": int(call_id)} try: response = requests.post(url, headers=headers, json=payload, timeout=20) response.raise_for_status() transcribation_data = response.json().get("transcribation", []) if not transcribation_data: return None # Ответ - это массив, берем первый элемент chunks = transcribation_data[0].get("chunks", []) # channel_tag: 1 — звонящий, 2 — отвечающий. Нам нужен отвечающий. manager_speech = "\n".join([chunk['text'] for chunk in chunks if chunk.get('channel_tag') == 2]) return manager_speech except requests.exceptions.RequestException as e: print(f"Ошибка получения транскрипции для {call_id}: {e}") return None
Анализ текста с помощью LLM
Когда у нас есть речь менеджера, следующий шаг — превратить её в формальную оценку по чек-листу. Запрос типа «оцени диалог» даёт слишком свободный и непредсказуемый ответ, поэтому нам нужна строгая структура. Для этого используем связку Pydantic и LangChain.
Pydantic позволяет описать форму будущего результата в виде Python-класса: какие поля должны быть в JSON, какие у них типы и как они называются.
LangChain использует эту схему и автоматически формирует подсказку для модели, заставляя её возвращать ответ в нужном формате. Таким образом мы избегаем вариативности и получаем стабильный структурированный результат.
Такой подход превращает вызов LLM в понятную инженерную операцию: модель анализирует текст, но выдаёт строго описанные данные. Ниже показано, как задаётся схема и запускается анализ.
from pydantic import BaseModel, Field from typing import List from langchain_core.messages import HumanMessage, SystemMessage from langchain_gigachat import GigaChat class CheckListItem(BaseModel): """Описывает результат проверки по одному критерию из чек-листа.""" criterion: str = Field(description="Название критерия, например, 'Приветствие'") passed: bool = Field(description="Выполнен ли критерий (true/false)") comment: str = Field(description="Краткий комментарий, почему принято такое решение") class ScriptComplianceCheck(BaseModel): """Итоговый результат проверки звонка на соответствие скрипту.""" overall_summary: str = Field(description="Общий вывод по звонку в одном предложении") checklist: List[CheckListItem] = Field(description="Список проверок по каждому критерию") def analyze_manager_speech(manager_transcript: str) -> ScriptComplianceCheck | None: """Анализирует речь менеджера и возвращает структурированную оценку.""" system_prompt = ( "Ты — система контроля качества работы оператора в колл-центре. " "Тебе даётся транскрипт телефонного разговора. " "Твоя задача — по каждому пункту чеклиста установить бинарно: выполнен критерий или нет. " "Оценивай строго по смыслу, а не по наличию конкретных слов. " "Если информация отсутствует или сомнительна — ставь 'false'. " "Твой ответ ДОЛЖЕН быть в формате JSON, соответствующем предоставленной схеме." ) human_prompt = f""" Проанализируй реплики менеджера по следующему чек-листу. **ЧЕКЛИСТ И КРИТЕРИИ** 1. **Приветствие:** Оператор должен поздороваться и назвать своё имя. * Пример TRUE: «Здравствуйте, меня зовут Анна». * Пример FALSE: «Алло, слушаю». 2. **Идентификация компании:** Оператор должен назвать компанию или свою роль (например, "служба поддержки"). * Пример TRUE: «...компания "Рога и копыта"». * Пример FALSE: Оператор представился только по имени. 3. **Выявление потребности:** Оператор должен задать уточняющий вопрос, чтобы понять задачу клиента. * Пример TRUE: «Подскажите, чем могу помочь?» * Пример FALSE: Перешёл сразу к ответу, не уточнив запрос. 4. **Призыв к действию (CTA):** Оператор должен подвести итог и корректно попрощаться. * Пример TRUE: «Итак, мы сменили вам тариф. Хорошего дня». * Пример FALSE: Просто положил трубку. Вот транскрипция реплик менеджера для анализа: --- {manager_transcript} --- """ try: giga = GigaChat(credentials=os.getenv("GIGACHAT_API_KEY"), verify_ssl_certs=False) structured_giga = giga.with_structured_output(schema=ScriptComplianceCheck) messages = [SystemMessage(content=system_prompt), HumanMessage(content=human_prompt)] result = structured_giga.invoke(messages) return result except Exception as e: print(f"Ошибка при анализе LLM: {e}") return None
Таким образом мы закрываем самый насыщенный этап — Transform. Теперь сервис умеет не только забирать транскрипт, но и превращать его в предсказуемый Pydantic-объект с оценкой по чек-листу. Формат строго задан, структура стабильна, и результат можно спокойно отправлять дальше по конвейеру.
Следующий шаг самый простой: сформировать итоговый JSON и подготовить данные к передаче во внешние системы.
Load: готовим итоговый JSON
Берём структурированный результат анализа и добавляем к нему служебные данные: ID звонка, оператора и итоговый балл. Из этого формируется JSON-отчёт, который можно отправить в любую внешнюю систему — хранилище, дашборд или внутренний API. Ниже приведена небольшая функция, которая собирает итоговый объект и выводит его в консоль.
import json from datetime import datetime def save_result_as_json(call_id: str, manager_id: str, analysis_result: ScriptComplianceCheck): """Формирует и выводит итоговый JSON-объект в консоль.""" passed_count = sum(1 for item in analysis_result.checklist if item.passed) score = int((passed_count / len(analysis_result.checklist)) * 100) if analysis_result.checklist else 0 final_report = { "processed_at": datetime.now().isoformat(), "call_id": call_id, "manager_id": manager_id, "score": score, "summary": analysis_result.overall_summary, "checklist_details": analysis_result.model_dump().get('checklist', []) } print("\n--- ГОТОВЫЙ РЕЗУЛЬТАТ (JSON) ---") print(json.dumps(final_report, indent=2, ensure_ascii=False)) print("--------------------------------\n")
На этом основная логика завершена. У нас есть все части процесса: прием событий, получение транскрипции, анализ с помощью LLM и сбор итогового отчёта. Осталось объединить шаги в единый процесс — функцию, которая будет вызываться при готовности транскрипции и проходить весь путь от uid до итогового JSON.
Собираем всё вместе
Оставшийся шаг — объединить отдельные функции в единый рабочий процесс. За это отвечает оркестратор process_call: он запускается в отдельном потоке после вебхука о готовности транскрипции и последовательно выполняет три операции — получает текст разговора, анализирует речь оператора и формирует итоговый JSON. На каждом шаге есть простые проверки, чтобы не продолжать обработку при ошибках.
Финальный блок if name == '__main__': запускает Flask-сервер, который принимает вебхуки от МТС Exolve. В продакшене для него можно использовать Gunicorn или другой WSGI-сервер.
def process_call(call_id_str: str): """Основная логика: достать ID, получить транскрипт, проанализировать, сохранить.""" with storage_lock: manager_id = call_manager_storage.pop(call_id_str, "N/A") # Шаг 1: Получаем текст transcript_text = get_transcript_by_id(call_id_str) if not transcript_text: return # Шаг 2: Анализируем analysis_result = analyze_manager_speech(transcript_text) if not analysis_result: return # Шаг 3: Сохраняем save_result_as_json(call_id_str, manager_id, analysis_result) if __name__ == '__main__': # Для production используйте Gunicorn или другой WSGI-сервер app.run(host='0.0.0.0', port=5003)
Что в итоге
После запуска сервиса и первого звонка вы увидите в консоли итоговый отчет — структурированный JSON с оценкой диалога. Он уже готов к использованию: можно сохранять его в базу, передавать во внутреннюю систему или включать в последующую аналитику. Ниже приведён пример результата для звонка, в котором оператор нарушил все критерии чек-листа.
{ "processed_at": "2025-11-08T21:20:06.228057", "call_id": "268224", "manager_id": "sip-operator-2", "score": 0, "summary": "Ни один из критериев не выполнен правильно. Разговор не содержит необходимых элементов общения с клиентом.", "checklist_details": [ { "criterion": "Приветствие", "passed": false, "comment": "Нет приветствия и имени оператора" }, { "criterion": "Идентификация компании", "passed": false, "comment": "Нет идентификации компании или роли оператора" }, { "criterion": "Выявление потребности", "passed": false, "comment": "Нет выявленного запроса клиента" }, { "criterion": "Призыв к действию (CTA)", "passed": false, "comment": "Нет завершения разговора с подведением итогов и прощанием" } ] }
Заключение
Мы собрали решение, которое превращает рутинную задачу контроля в автоматизированный процесс. Выходной JSON — это универсальный формат, который можно интегрировать с чем угодно: от простой записи в базу данных до отправки в сложные BI-системы вроде Grafana или Power BI для построения дашбордов.
Варианты развития:
Подключить дашборд для отслеживания динамики показателей
Добавить игровые механики и рейтинги операторов
Рекомендовать сотрудникам обучающие материалы по ��езультатам анализа
Отправлять короткие отчёты супервизору и подключать его в сложные кейсы
Полный код проекта доступен в репозитории на гитхаб. Будем рады вопросам и предложениям в комментариях.
