24.11.2025 10:39
Katner 0 130 Источник

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

В этой статье разберём, как настроить полный сценарий: от вебхуков в МТС 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. Для обработки звонков нам достаточно двух событий из их полного жизненного цикла.

  1. Начало звонка (type: "b"). Это первое уведомление, которое приходит почти мгновенно. В нём содержится связка uid с ID звонка и sip_id оператора. Мы фиксируем эту информацию, чтобы знать, какой сотрудник вёл разговор.

  2. Готовность транскрипции (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 для построения дашбордов.

Варианты развития:

  • Подключить дашборд для отслеживания динамики показателей

  • Добавить игровые механики и рейтинги операторов

  • Рекомендовать сотрудникам обучающие материалы по результатам анализа

  • Отправлять короткие отчёты супервизору и подключать его в сложные кейсы

Полный код проекта доступен в репозитории на гитхаб. Будем рады вопросам и предложениям в комментариях.

Комментарии (0)