Привет, Хабр! Это Катя Саяпина, менеджер продукта МТС Exolve.
Самую честную обратную связь бизнес получает не из опросов, а из живых разговоров — когда клиент сам звонит и рассказывает, что его раздражает, что не работает или чего не хватает. Мы хотим извлекать эту ценность автоматически.
Сегодня покажу, как собрать простую систему фонового анализа звонков. Она забирает расшифровки разговоров через API МТС Exolve, отправляет их в GigaChat для обработки, а результаты сохраняет в базу SQLite.
Архитектура и точка входа
Предполагаю, что скрипт и все зависимости уже установлены, а ключи API прописаны в файле .env. Сосредоточимся на том, как он работает под капотом.
Прежде чем нырять в код, взглянем на общую схему. Она проста и соответствует классическому ETL-подходу (Extract, Transform, Load):
Extract — получаем готовые транскрипции звонков из МТС Exolve API.
Transform — отправляем текст в GigaChat и преобразуем его в структурированные данные, а именно списки проблем и предложений.
Load — сохраняем эти структурированные данные в локальную базу SQLite для последующего анализа.
Точкой входа в наш скрипт служит стандартный блок if name == "__main__":. Мы используем argparse для создания простого интерфейса командной строки, который запускает разные части функциональности.
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Инструмент для управления звонками и их анализом.")
parser.add_argument(
"command",
choices=["create-resource", "make-call", "analyze"],
help="Действие для выполнения: \n"
"'create-resource' - создать ID ресурса для коллбэков. \n"
"'make-call' - совершить звонок. \n"
"'analyze' - проанализировать завершенные звонки."
)
args = parser.parse_args()
if args.command == "create-resource":
create_callback_resource()
elif args.command == "make-call":
make_call()
elif args.command == "analyze":
process_and_analyze_calls()
Основная магия происходит в команде analyze, которая запускает функцию process_and_analyze_calls. Давайте проследим её логику шаг за шагом.
Шаг 1: Извлечение данных из МТС Exolve
Первая задача — получить данные для анализа. Мы обращаемся к эндпоинту МТС Exolve GetSpeechAnalyticsList. Это мощный метод, который возвращает не просто аудиозаписи, а уже готовые результаты речевой аналитики, в том числе полную текстовую расшифровку, разбитую по участникам диалога. Нам не нужно самостоятельно заниматься распознаванием речи.
# main.py - внутри функции process_and_analyze_calls()
def process_and_analyze_calls():
url = "https://api.exolve.ru/statistics/call-record/v1/GetSpeechAnalyticsList"
headers = {"Authorization": f"Bearer {EXOLVE_API_KEY}", "Content-Type": "application/json"}
date_to = datetime.now(timezone.utc)
date_from = date_to - timedelta(hours=5) # Забираем данные за последние 5 часов
payload = {"date_from": date_from.strftime('%Y-%m-%dT%H:%M:%SZ'), "date_to": date_to.strftime('%Y-%m-%dT%H:%M:%SZ'), "limit": 100}
print("--- Запрашиваю список готовых аналитик из Exolve ---")
response = requests.post(url, headers=headers, json=payload)
# ...
Запрашиваем аналитику за последние пять часов. Но что именно мы получаем в ответе? Посмотрим на упрощённую структуру JSON, которую возвращает МТС Exolve API для одного звонка:
{
"speech_analytics": [
{
"call_id": 123456789,
"transcription": {
"status": "SUCCESS",
"phrases": [
{
"channel_tag": 1,
"text": "Здравствуйте, компания X, оператор Анна, слушаю вас."
},
{
"channel_tag": 2,
"text": "Добрый день. У меня проблема с доставкой, курьер опоздал на три часа!"
}
]
}
}
]
}
Ключевое для нас здесь — массив phrases. Поле channel_tag чётко отделяет реплики оператора (1) от реплик клиента (2), чтобы мы могли анализировать только слова клиента.
Шаг 2: Анализ и структурирование текста
Превращаем сырой текст диалога в полезные выводы.
Интеграция с LLM через Pydantic
Чтобы гарантировать, что LLM вернёт данные в нужном формате, мы описываем этот формат с помощью класса Pydantic. Это, по сути, схема, которой модель обязана следовать.
from pydantic import BaseModel, Field
from typing import ListЭта схема — наш «контракт» с моделью. Но чтобы она выполнила его максимально точно, нужен качественный промпт. Простая инструкция «извлеки проблемы» будет работать, но для повышения стабильности и уменьшения «галлюцинаций» модели стоит применить более продвинутые техники промпт-инжиниринга.
Давайте улучшим системный промпт — добавим в него несколько ключевых элементов:
Назначение роли. Мы говорим модели, кем она должна быть («Ты — ведущий аналитик...»). Это помогает ей лучше понять контекст и стиль ответа.
Уточнение задачи. Чётко прописываем, что нужно делать («Из реплик клиента...»), а чего делать не нужно («Игнорируй приветствия...»).
Негативные инструкции. Прямо запрещаем модели додумывать («Не придумывай ничего от себя...»).
Пример (Few-shot prompting): даём короткий пример прямо в инструкции, чтобы модель наглядно увидела, какого результата от неё ожидаем.
Вот как будет выглядеть улучшенная версия системного промпта, которую мы поместим в функцию analyze_text_with_gigachat:
# Улучшенный системный промпт внутри функции analyze_text_with_gigachat
system_prompt = (
"Ты — ведущий аналитик в отделе контроля качества. "
"Твоя задача — беспристрастно проанализировать текстовую расшифровку диалога между оператором и клиентом. "
"Из реплик клиента извлеки только конкретные проблемы, с которыми он столкнулся, и его предложения по улучшению сервиса. "
"Если проблем или предложений нет, верни пустые списки. "
"Не придумывай ничего от себя и игнорируй общие фразы. "
"Пример: из фразы 'Здравствуйте, курьер опоздал, но приложение у вас удобное, добавьте туда еще оплату по СБП' нужно извлечь: "
"Проблема: 'курьер опоздал'. Предложение: 'добавить оплату по СБП в приложении'."
)
# Этот system_prompt будет использоваться при вызове модели
messages = [
SystemMessage(content=system_prompt),
HumanMessage(content=dialog_text),
]
Такой подход превращает взаимодействие с LLM в управляемый инженерный процесс и даёт более предсказуемые и качественные результаты на больших объёмах данных.
Далее мы используем langchain_gigachat для вызова модели. Метод .with_structured_output(schema=AnalysisResult) — это ключевой элемент, который «заставляет» GigaChat вернуть JSON, строго соответствующий нашему классу.
giga = GigaChat(credentials=GIGACHAT_API_KEY, verify_ssl_certs=False)
structured_giga = giga.with_structured_output(schema=AnalysisResult)
def analyze_text_with_gigachat(dialog_text: str):
"""Принимает текст диалога и возвращает объект AnalysisResult."""
# ...
messages = [
SystemMessage(content="Ты — умный ассистент... Извлеки из слов клиента проблемы и предложения."),
HumanMessage(content=dialog_text),
]
return structured_giga.invoke(messages)
На выходе получаем не строку, а полноценный объект AnalysisResult, с которым удобно работать дальше.
Шаг 3: Загрузка данных в SQLite
Последний этап — сохранение извлечённых данных. Проходим циклом по звонкам, полученным от МТС Exolve. Чтобы не обрабатывать один и тот же звонок дважды, мы ведём учёт call_id в отдельной таблице processed_calls.
Для каждого нового звонка:
Склеиваем все фразы из транскрипции в один большой текст.
Передаём этот текст в функцию analyze_text_with_gigachat.
Итерируемся по полям .problems и .suggestions полученного объекта.
Сохраняем каждую проблему и каждое предложение в таблицу call_analysis.
for analytics in analytics_list:
call_id = analytics.get('call_id')
if call_id and call_id not in processed_ids:
# ... (склеиваем текст из phrases)
analysis_result = analyze_text_with_gigachat(full_text)
if analysis_result:
# Сохраняем проблемы
for problem in analysis_result.problems:
cursor.execute("INSERT INTO call_analysis (call_id, type, content) VALUES (?, ?, ?)",
(call_id, 'problem', problem))
# Сохраняем предложения
for suggestion in analysis_result.suggestions:
cursor.execute("INSERT INTO call_analysis (call_id, type, content) VALUES (?, ?, ?)",
(call_id, 'suggestion', suggestion))
# Помечаем звонок как обработанный
cursor.execute("INSERT INTO processed_calls (uid) VALUES (?)", (call_id,))
conn.commit()
Теперь, когда у нас есть база данных со структурированными инсайтами, извлечь из неё пользу — дело техники. Простейший отчёт можно получить двумя SQL-запросами.
Запрос № 1: Топ-5 проблем по частоте упоминаний
SELECT content, COUNT(content) as mentions
FROM call_analysis
WHERE type = 'problem'
GROUP BY content
ORDER BY mentions DESC
LIMIT 5;
Пример ответа:
Топ-5 проблем:
Курьер опоздал на три часа (упоминаний: 15).
Не могу найти, как отменить заказ на сайте (упоминаний: 9).
Запрос № 2: Список уникальных предложений
SELECT DISTINCT content
FROM call_analysis
WHERE type = 'suggestion';
Эти запросы превращают разрозненные данные в результат, который можно показать продуктовой команде.
Новые предложения от клиентов:
Добавьте, пожалуйста, оплату через СБП в приложении.
Сделайте SMS-уведомление, когда курьер выехал.
Заключение
Мы разработали решение, которое автоматизирует трудоёмкую задачу в клиентском сервисе. Вместо прослушивания вручную или проведения исследований с таким решением можно автоматически с максимальной выборкой быть в курсе всех жалоб и предложений от клиентов.
Как это можно развивать
Интегрировать в процессы компании. Например, выделять продукт, о котором идёт речь во время звонка, и отправлять сводку жалоб и предложений релевантной команде. Или добавлять список предложений в трекер задач продуктовой команде на доисследования.
Подключить текстовые коммуникации: мессенджеры, чаты и электронную почту.
Автоматизировать запуск скрипта, чтобы он срабатывал мгновенно после каждого завершённого звонка, а не по расписанию.
Визуализировать статистику с помощью Grafana или сделать простой веб-интерфейс на Flask, чтобы менеджеры могли сами в реальном времени смотреть отчёты.
Если есть вопросы и идеи о том, что ещё мы могли бы разобрать в статье или добавить в существующий код для вас, — пишите в комментах. А это — код на гитхабе.