Привет, Хабр!
База знаний в поддержке почти всегда отстаёт от реальности: знания появляются в звонках, остаются в умах операторов, а в документацию попадают выборочно и с задержкой. При этом сами звонки уже записываются и транскрибируются.
В этой статье разберём инженерный способ автоматически извлекать из этого потока проблемы и решения и вести актуальную базу знаний с помощью Python, МТС Exolve и LLM.
Архитектура решения
Система использует записываемые и транскрибируемые звонки техподдержки как источник данных. После завершения разговора сервис запрашивает транскрипцию, извлекает из неё суть обращения и проверяет её на дубликаты.
Результат автоматически записывается в базу знаний. Чтобы не раздувать контекст нейросети и контролировать стоимость, при сравнении учитываются только последние 50 записей.
Основные компоненты
МТС Exolve — записывает звонки, делает транскрипцию и отправляет вебхук о готовности расшифровки
Flask Webhook API — принимает вебхук, отвечает 200 OK и запускает обработку звонка асинхронно
LLM llama-3.3-70b-instruct от MWS GPT — извлекает topic, solution, is_solved и делает семантическую проверку на дубликаты
Дедупликация — сравнивает новую тему с существующими
Google Sheets в роли хранилища знаний — хранит тему, решение, частоту и долю решенных обращений. Удобно для MVP и быстрого анализа данных
Пререквизит
Для примера используем Python 3.9+ и набор библиотек для приёма вебхуков, запросов к API и работы с Google Sheets.
pip install flask requests python-dotenv gspread google-auth
Ключи и токены хранятся в переменных окружения. Для записи данных в таблицу используется сервисный аккаунт Google и файл credentials.json.
Шаг 1. Получаем транскрипцию и готовим данные
Пайплайн запускается с вебхука от МТС Exolve о готовности транскрипции. В обработчике /exolve/webhook мы проверяем data.get("type") == "trc", извлекаем uid звонка и сразу отвечаем 200 OK. Чтобы не держать вебхук и не упираться в таймауты, дальнейшая обработка запускается асинхронно в отдельном потоке через threading.
Далее функция get_call_data() запрашивает транскрипцию через метод GetTranscribation по uid звонка, извлекает chunks и собирает единый текст диалога. Роли сохраняются по channel_tag: Client для канала 1 и Support для канала 2. На выходе получаем строку dialogue_text, готовую для передачи в LLM.
# data_provider.py
import os
import requests
from dotenv import load_dotenv
load_dotenv()
def get_call_data(call_uid: str) -> dict | None:
"""
Забирает транскрипцию звонка из MTS Exolve.
Возвращает словарь с текстом диалога и длительностью.
"""
url = "https://api.exolve.ru/statistics/call-record/v1/GetTranscribation"
token = os.getenv("EXOLVE_API_KEY")
try:
resp = requests.post(url, json={"uid": int(call_uid)}, headers={"Authorization": f"Bearer {token}"})
resp.raise_for_status()
data = resp.json()
record = data.get('transcribation', [{}])[0]
chunks = record.get('chunks', [])
if not chunks:
return None
# Формируем текст для LLM с разделением по ролям
dialogue_lines = []
for chunk in chunks:
# channel_tag: 1 — Клиент, 2 — Оператор
role = "Client" if chunk['channel_tag'] == 1 else "Support"
dialogue_lines.append(f"{role}: {chunk['text']}")
return {
"dialogue_text": "\n".join(dialogue_lines),
"duration": record.get("duration", 0)
}
except Exception as e:
print(f"❌ Ошибка получения данных МТС Exolve: {e}")
return NoneШаг 2. Мозг системы: MWS GPT
Превращаем диалог в структурированные данные с помощью нейросети. Для анализа используем open-source модель llama-3.3-70b-instruct через API MWS GPT.
Для ориентира по качеству можно посмотреть результаты по MMLU — бенчмарку общего понимания текста и знаний из разных доменов. В агрегированных сравнениях llama-3.3-70b-instruct показывает сопоставимый уровень с GPT-4o от OpenAI, отставая на несколько процентов, что достаточно для задач семантического анализа диалогов.
Модель решает две задачи:
Извлекает проблему, решение и статус
Проверяет новизну темы
Для повышения стабильности и воспроизводимости мы используем низкие значения temperature и top_p.
Работу с API выносим в отдельную функцию-обёртку. Она принимает системный промпт и текст диалога, отправляет запрос к модели и возвращает ответ.
# ai_engine.py
import os
import json
import requests
MTS_AI_URL = "https://api.gpt.mws.ru/v1/chat/completions" # Пример URL
MTS_AI_TOKEN = os.getenv("MTS_AI_TOKEN")
def query_llm(system_prompt: str, user_text: str) -> str:
headers = {
"Authorization": f"Bearer {MTS_AI_TOKEN}",
"Content-Type": "application/json"
}
payload = {
"model": "llama-3.3-70b-instruct",
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_text}
],
"temperature": 0.1, # Минимальная креативность для стабильности
"top_p": 0.1,
"max_tokens": 1000
}
resp = requests.post(MTS_AI_URL, json=payload, headers=headers)
resp.raise_for_status()
return resp.json()['choices'][0]['message']['content']Извлекаем проблему, решение и статус
В модель передаём текст разговора с размеченными ролями Client и Support, и системный промпт с чётко заданными правилами обработки. Мы явно просим игнорировать приветствия и разговорную воду, извлечь суть обращения, кратко сформулировать решение и определить, была ли проблема решена к концу разговора.
Формат ответа заранее ограничен строгим JSON, чтобы результат можно было автоматически обрабатывать.
def extract_knowledge(dialogue_text: str) -> dict | None:
system_prompt = """
Ты — аналитик базы знаний. Твоя задача:
1. Игнорировать приветствия и 'воду'.
2. Сформулировать 'Topic' (суть проблемы в 3-5 слов).
3. Сформулировать 'Solution' (краткое решение, которое дал оператор).
4. Определить 'Is_Solved' (True/False): решена ли проблема клиента в конце диалога?
Верни ТОЛЬКО валидный JSON формата:
{
"topic": "...",
"solution": "...",
"is_solved": true/false
}
"""
try:
response_text = query_llm(system_prompt, dialogue_text)
# Чистим ответ, если LLM добавила markdown-теги
clean_json = response_text.replace("```json", "").replace("```", "").strip()
return json.loads(clean_json)
except Exception as e:
print(f"⚠️ Ошибка AI Extraction: {e}")
return None
Семантическая дедупликация
После извлечения темы мы проверяем, является ли она новой или уже встречалась ранее. Для этого берём список существующих тем напрямую из базы знаний в Google Sheets — из колонки Topic, где каждая строка соответствует отдельной проблеме.
Если новая тема описывает уже известную проблему другими словами, модель возвращает точное название существующей темы, иначе — маркер NEW. Такой подход позволяет избежать строковых алгоритмов нечеткого поиска и работать с дубликатами на уровне смысла, а не символов.
def check_duplicate_with_ai(new_topic: str, existing_topics: list[str]) -> str | None:
"""
Возвращает название существующей темы, если найден дубликат, или None.
"""
if not existing_topics:
return None
topics_to_check = existing_topics[-30:]
topics_list_str = "\n".join([f"- {t}" for t in topics_to_check])
system_prompt = f"""
У тебя есть список существующих тем обращений:
{topics_list_str}
Пользователь прислал новую тему: "{new_topic}".
Является ли новая тема смысловым дубликатом одной из существующих?
1. Если ДА — верни ТОЧНОЕ название те��ы из списка.
2. Если НЕТ — верни слово "NEW".
Отвечай только названием или словом NEW. Не используй кавычки и знаки препинания.
"""
verdict = query_llm(system_prompt, "Сравни темы.")
# Чистим ответ LLM: убираем случайные кавычки, точки и лишние пробелы
clean_verdict = verdict.strip().replace('"', '').replace("'", "").rstrip('.')
# Сравниваем с учетом регистра (case-insensitive)
for topic in existing_topics:
if clean_verdict.lower() == topic.lower():
return topic
return NoneБаза знаний может содержать тысячи записей и отправлять их все в промпт — дорого и неэффективно. В коде выше мы применили стратегию скользящего окна и отправляем на проверку только 30 последних актуальных тем. Для промышленного решения для организации базу знаний и поиска по ней потребуется иной подход.
Шаг 3. Обновление базы знаний в Google Sheets
Google Sheets используем как простой вариант для наглядности, который позволяет работать с данными всей команде и выполнять базовый анализ без отдельного интерфейса.
Каждая строка таблицы — это одна тема обращения. В ней хранятся:
Topic — сама тема
Solution — последнее актуальное решение
Frequency — частота обращений
Solved — доля успешно решённых запросов
Вся логика работы с таблицей собрана в классе KnowledgeRegistry с несколькими методами:
get_existing_topics()
Читает колонку Topic и возвращает список уже известных тем. Этот список используется для проверки дубликатов на предыдущем шагеadd_new_issue()
Если тема новая, метод добавляет строку в таблицу, записывая Topic и Solution, и задаёт Frequency = 1. Значение Solved % задаётся по флагу is_solvedupdate_issue_stats()
Если тема уже существует, метод находит соответствующую строку, увеличивает Frequency и пересчитывает Solved %. История отдельных звонков при этом не хранится — обновляются только агрегированные значения
В результате база знаний обновляется автоматически: новые темы появляются по мере обработки звонков, а по существующим накапливается статистика частоты и успешности решений.
# sheets_manager.py
import gspread
class KnowledgeRegistry:
def __init__(self, credentials_path='credentials.json', sheet_name='Support_KB_Drafts'):
self.gc = gspread.service_account(filename=credentials_path)
self.sh = self.gc.open(sheet_name).sheet1
def get_existing_topics(self) -> list[str]:
"""Возвращает список всех тем из колонки A."""
return self.sh.col_values(1)[1:] # Пропускаем заголовок
def add_new_issue(self, data: dict):
"""Добавляет новую строку."""
# Структура колонок: [Topic, Solution, Freq, Solved %]
row = [
data['topic'],
data['solution'],
1, # Начальная частота (Frequency)
100 if data['solved'] else 0 # Процент решений
]
self.sh.append_row(row)
print(f"✅ Добавлена новая тема: {data['topic']}")
def update_issue_stats(self, topic: str, new_data: dict):
"""Обновляет статистику по существующей теме."""
try:
# Находим строку с нужной темой
cell = self.sh.find(topic)
row_num = cell.row
# ОПТИМИЗАЦИЯ: Читаем строку целиком за один запрос API
# Ожидаем порядок в таблице: [Topic, Solution, Freq, Solved %]
row_values = self.sh.row_values(row_num)
# Индексы в списке (Python start 0): Col C (Freq)=2, Col D (Solved)=3
current_freq = int(row_values[2])
current_solved = float(row_values[3])
# Пересчитываем средние значения
new_freq = current_freq + 1
is_solved_int = 100 if new_data['solved'] else 0
new_avg_solved = (current_solved * current_freq + is_solved_int) / new_freq
# ОПТИМИЗАЦИЯ: Обновляем ячейки C и D одной пачкой
self.sh.update(
range_name=f"C{row_num}:D{row_num}",
values=[[new_freq, round(new_avg_solved, 1)]]
)
print(f"🔄 Обновлена тема: {topic}. Частота: {new_freq}")
except Exception as e:
print(f"❌ Ошибка обновления таблицы: {e}")
Шаг 4. Сборка пайплайна и оркестрация
Собираем все предыдущие части вместе. Вся логика сосредоточена в функции process_call_logic(), которая вызывается из обработчика вебхука и последовательно выполняет шаги обработки одного звонка — от получения транскрипции до обновления базы знаний.
Внутри process_call_logic() происходит следующее:
Получаем транскрипцию через get_call_data(call_uid). Если данных нет — завершаем обработку
Извлекаем структуру обращения через extract_knowledge(dialogue_text). Если модель не вернула результат — завершаем обработку
Подключаемся к Google Sheets через KnowledgeRegistry() и забираем список тем get_existing_topics()
Проверяем на дубли через check_duplicate_with_ai(), передавая в сравнение последние 30 тем
Обновляем базу знаний: если найден дубликат — вызываем update_issue_stats(), иначе — add_new_issue()
Пайплайн запускается асинхронно, чтобы не блокировать обработчик вебхука и не упираться в таймауты платформы.
# main_logic.py
from data_provider import get_call_data
from ai_engine import extract_knowledge, check_duplicate_with_ai
from sheets_manager import KnowledgeRegistry
def process_call_logic(call_uid: str):
print(f"🚀 Начало обработки звонка {call_uid}")
# 1. Получаем данные и метрики
call_data = get_call_data(call_uid)
if not call_data: return
# 2. Извлекаем смысл через AI
knowledge = extract_knowledge(call_data['dialogue_text'])
if not knowledge: return
print(f"🤖 AI выделил тему: {knowledge['topic']}")
# 3. Работаем с таблицей
kb = KnowledgeRegistry()
existing_topics = kb.get_existing_topics()
# 4. Проверяем на дубликаты через AI
# (Берем последние 30 тем для экономии токенов и контекста)
recent_topics = existing_topics[-30:]
duplicate_topic = check_duplicate_with_ai(knowledge['topic'], recent_topics)
stats_data = {
"topic": knowledge['topic'],
"solution": knowledge['solution'],
"solved": knowledge['is_solved']
}
if duplicate_topic:
# Это дубль — обновляем статистику существующей темы
kb.update_issue_stats(duplicate_topic, stats_data)
else:
# Это новая проблема — создаем запись
kb.add_new_issue(stats_data)Связка с внешним событием реализована через Flask-сервер. Он принимает вебхук от платформы, фильтрует нужный тип события и запускает обработку в фоне:
# app.py
import threading
from flask import Flask, request, jsonify
from main_logic import process_call_logic
app = Flask(__name__)
@app.route('/exolve/webhook', methods=['POST'])
def handle_webhook():
data = request.json
if data.get('type') == 'trc': # Transcription Ready
uid = data.get('uid')
# Fire-and-forget: запускаем обработку в фоне
threading.Thread(target=process_call_logic, args=(uid,)).start()
return jsonify({"status": "ok"}), 200
if __name__ == '__main__':
app.run(port=5000)Результаты
После запуска база знаний начнёт наполняться по мере обработки звонков. В Google Sheets формируется живая база знаний с проблемами и решениями, которые обновляются без ручного разбора диалогов и участия операторов.
Важно, что система работает реактивно: она не требует предварительной классификации тем и формируются из живых формулировок клиентов и эволюционируют вместе с продуктом.
Ограничения подхода
Мы сделали сознательно минималистичный вариант:
Дедупликация по скользящему окну хорошо ловит волны инцидентов, но будет пропускать редкие или давние проблемы
Google Sheets подходит для MVP, но не для сложных запросов и высокой нагрузки
В статье используется llama-3.3-70b-instruct — сильная open-source модель с результатами на уровне GPT-4-класса в задачах семантического анализа. При этом модели продолжают быстро развиваться, и замена модели или уточнение промптов может дать дополнительный прирост качества без изменения архитектуры
Эти ограничения не критичны на старте, но их важно понимать до того, как система станет источником управленческих решений.
Идеи для развития
Если использовать этот подход как основу, дальнейшее развитие выглядит довольно прямолинейно:
Двухэтапная дедупликация: embeddings для поиска кандидатов → LLM для финального сопоставления
Переход на базу данных
Обратная интеграция в поддержку — подсказки операторам по часто встречающимся темам во время звонка
Использование в RAG: база знаний как источник для ботов и внутренних ассистентов.
Код доступен в репозитории.
