Содержание
- Описание проблемы
- Основные функции бота:
- Технологический стек:
- Подробный разбор функциональности
- 1. Инициализация и настройка
- 2. Обработка команды /start
- 3. Прием и обработка аудиофайлов
- 4. Транскрибация аудио
- 5. Структурирование и анализ транскрипции
- 6. Сохранение результатов
- 7. Отправка результатов пользователю
- 8. Дополнительные функции
- Преимущества решения
- Развертывание на сервере TimeWeb Cloud в Нидерландах
- Python код
Описание проблемы
Традиционный подход к оценке качества звонков менеджеров часто включает в себя:
- Прослушивание записей звонков супервайзерами
- Ручное заполнение оценочных форм
- Предоставление обратной связи менеджерам
Этот процесс трудоемкий, субъективный и не масштабируемый при росте компании. В Микро-компаниях прослушивание звонков, как правило, задача далеко не на пережнем плане — из-за своей трудоемкости.
Создал Telegram-бота, который автоматизирует процесс анализа звонков, используя современные технологии искусственного интеллекта и обработки естественного языка.
Основные функции бота:
- Прием аудиофайлов звонков через Telegram
- Транскрибация аудио в текст
- Структурирование диалога
- Анализ качества разговора по заданным критериям
- Предоставление подробного отчета
Технологический стек:
- Python
- Telegram Bot API
- OpenAI API (GPT и Whisper)
- Google Docs API
- PostgreSQL
- Асинхронное программирование (asyncio)
Подробный разбор функциональности
1. Инициализация и настройка
Бот использует библиотеку python-telegram-bot для взаимодействия с Telegram API. Настраивается логирование для отслеживания работы бота.
2. Обработка команды /start
При первом взаимодействии пользователя с ботом:
- Отправляется приветственное сообщение
- Данные пользователя сохраняются в базу данных PostgreSQL
- Администратор уведомляется о новом пользователе
3. Прием и обработка аудиофайлов
Когда пользователь отправляет аудиофайл:
- Проверяется формат и размер файла
- Файл сохраняется локально
- Проверяется лимит использования (5 раз в сутки на пользователя)
4. Транскрибация аудио
Используется модель Whisper от OpenAI для преобразования аудио в текст.
5. Структурирование и анализ транскрипции
Два запроса к GPT модели OpenAI:
- Структурирование диалога (определение реплик менеджера и клиента)
- Анализ качества разговора по заданным критериям:
- Начало разговора
- Описание продукта и предложение
- Обработка возражений
- Заключение сделки
- Постпродажное обслуживание
- Эмоциональный контекст
- Интеграция с CRM
6. Сохранение результатов
Структурированный текст и анализ добавляются в Google Документ с использованием Google Docs API.
7. Отправка результатов пользователю
Бот отправляет структурированную транскрипцию пользователю в Telegram.
8. Дополнительные функции
- Рассылка сообщений всем пользователям (доступно только администратору)
- Ограничение использования бота (5 раз в сутки на пользователя)
Преимущества решения
- Автоматизация: исключает необходимость ручного прослушивания звонков.
- Объективность: использование AI для анализа обеспечивает непредвзятую оценку.
- Масштабируемость: может обрабатывать большое количество звонков одновременно.
- Быстрая обратная связь: результаты анализа доступны практически мгновенно.
- Детальный анализ: охватывает множество аспектов качества разговора.
- Интеграция: результаты сохраняются в Google Docs для дальнейшего анализа и обучения.
Развертывание на сервере TimeWeb Cloud в Нидерландах
Для обеспечения бесперебойной работы бота без необходимости использования VPN в России, было принято решение развернуть его на сервере TimeWeb Cloud, расположенном в Нидерландах. Этот подход позволяет обойти географические ограничения и обеспечить стабильную работу сервиса.
Процесс развертывания включает следующие шаги:
- Регистрация на платформе TimeWeb Cloud
- Выбор опции «Облачные серверы» и создание нового сервера
- Выбор локации «Нидерланды» для размещения сервера
- Выбор операционной системы (рекомендуется Ubuntu 20.04)
- Выбор тарифного плана (для начала подойдет минимальный тариф)
- Создание сервера и сохранение данных для доступа
Такой подход к развертыванию обеспечивает несколько ключевых преимуществ:
- Обход географических ограничений: размещение в Нидерландах позволяет беспрепятственно использовать API OpenAI.
- Стабильность работы: облачный сервер обеспечивает высокую доступность и надежность сервиса.
- Масштабируемость: при необходимости легко увеличить мощность сервера.
- Безопасность: использование выделенного сервера повышает уровень безопасности и контроля над данными.
Python код
import os
import logging
import openai
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
import aiohttp
from aiohttp import TCPConnector
import asyncio
import csv
from google.oauth2 import service_account
from googleapiclient.discovery import build
client = openai.AsyncClient(api_key="Cюда вставить свой openai API")
# Настройка логирования
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)
user_uses = {}
admin_state = {}
async def db_connect():
try:
connection = await asyncpg.connect(user='Имя пользователя', password='пароль', database='название базы данных', host='localhost')
logger.info("Успешное подключение к базе данных")
return connection
except Exception as e:
logger.error(f"Ошибка подключения к базе данных: {e}")
raise
# Токен, полученный от BotFather
TOKEN = 'Cюда вставить токен от BotFather '
# Директория для сохранения аудиофайлов
AUDIO_FILES_DIR = 'audio_files'
if not os.path.exists(AUDIO_FILES_DIR):
os.makedirs(AUDIO_FILES_DIR)
# Ограничения Whisper
MAX_FILE_SIZE_MB = 25
SUPPORTED_FORMATS = ['mp3', 'mp4', 'mpeg', 'mpga', 'm4a', 'wav', 'webm']
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
full_name = update.effective_user.full_name
telegram_id = update.effective_user.id
username = update.effective_user.username
"""Отправляет приветственное сообщение при команде /start."""
user_full_name = update.effective_user.full_name
await update.message.reply_text(f"Здравствуйте,{user_full_name}! Я система распознавания и анализа речи “Хатипыч”. Слушаю звонки ваших менеджеров и провожу анализ зввонка.\n\n "
"Обновления SRAR Хатипыч публикуются в телеграм-канале: https://t.me/+4i7ik4mtATBjMTky. Настоятельно рекомендую подписаться!\n\n"
"Ограничения: максимальный размер файла - 25 МБ, поддерживаемые форматы: mp3, mp4, mpeg, mpga, m4a, wav, webm.\n\n "
"Использование ограничено до 5 раз в сутки на пользователя.\n\n" "Пришлите мне запись звонка для анализа\n\n")
await notify_admin(context, f"Новый пользователь: {user_full_name} (@{update.effective_user.username})")
async def notify_admin(context: ContextTypes.DEFAULT_TYPE, message: str):
admin_chat_id = 'Ваш chat_id администратора' # Замените на chat_id администратора или пользователя
try:
await context.bot.send_message(chat_id=admin_chat_id, text=message)
except Exception as e:
logger.error(f"Ошибка при отправке сообщения администратору: {e}")
async def save_user_data_if_not_exists(telegram_id, username, full_name):
conn = await db_connect()
try:
user_exists = await conn.fetchval('SELECT EXISTS(SELECT 1 FROM users WHERE telegram_id = $1)', telegram_id)
if not user_exists:
await conn.execute('INSERT INTO users(telegram_id, username, full_name) VALUES ($1, $2, $3)', telegram_id, username, full_name)
logger.info(f"Новый пользователь сохранен: {full_name} (@{username})")
# Отправляем уведомление администратору
await notify_admin(context, f"Новый пользователь: {full_name} (@{username})")
except Exception as e:
logger.error(f"Ошибка при сохранении данных пользователя: {e}")
finally:
await conn.close()
async def check_usage_and_update(user_id):
if user_id not in user_uses:
user_uses[user_id] = 1
else:
if user_uses[user_id] >= 5:
return False
user_uses[user_id] += 1
return True
async def save_audio(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
# Добавление проверки в начале save_audio, этот блок можно убрать
if not await check_usage_and_update(update.effective_user.id):
await update.message.reply_text("Извините, вы достигли лимита использования в сутки. Я готов проверять ваши звонки ежедневно 24/7 без выходных, у меня не болит голова и не вбывает похмелья. Еще моя продуктивность не зависит от настроения. \n\n Меня можно нанять. Свяжитесь с моим хозяином. Контакты Динара можно найти в его телеграм канале. ")
return
audio_file = await context.bot.get_file(update.message.audio.file_id)
file_size_mb = audio_file.file_size / 1024 / 1024
file_extension = os.path.splitext(audio_file.file_path)[1]
if file_size_mb > MAX_FILE_SIZE_MB or file_extension[1:] not in SUPPORTED_FORMATS:
await update.message.reply_text("Извините, файл не подходит по параметрам.")
return
file_url = audio_file.file_path
logger.info(f"Попытка загрузки файла по URL: {file_url}")
try:
async with aiohttp.ClientSession(connector=TCPConnector(ssl=False)) as session:
async with session.get(file_url) as resp:
if resp.status == 200:
file_name = f"{audio_file.file_unique_id}{file_extension}"
file_path = os.path.join(AUDIO_FILES_DIR, file_name) # Определение file_path
with open(file_path, 'wb') as fd:
while True:
chunk = await resp.content.read(1024)
if not chunk:
break
fd.write(chunk)
logger.info(f'Аудио сохранено: {file_path}')
await update.message.reply_text('Ваш аудиофайл сохранён и готов к обработке!')
# Использование OpenAI API для транскрибации аудиофайла
audio_file = open(file_path, "rb")
try:
transcription = await client.audio.transcriptions.create(
model="whisper-1",
file=audio_file,
response_format="text"
)
transcription_text = transcription
# Вызов функции для структурирования и проверки транскрипции
analysis_text, structured_text = await structure_and_check_transcription(transcription_text)
document_id = 'Ваш ID гугл документа'
await append_to_google_doc(document_id, structured_text, analysis_text)
# Отправка ответа в телеграм бота
await update.message.reply_text(f'Транскрипция: {structured_text}')
except Exception as e:
logger.error(f"Произошла ошибка при вызове OpenAI API: {e}")
await update.message.reply_text('Произошла ошибка при вызове OpenAI API. Пожалуйста, попробуйте еще раз позже.')
else:
logger.error(f"Ошибка загрузки файла: статус {resp.status}")
await update.message.reply_text('К сожалению, не удалось загрузить файл.')
except Exception as e:
logger.error(f"Произошла ошибка при загрузке файла: {e}")
await update.message.reply_text('Произошла ошибка при обработке вашего файла.')
await notify_admin(context, f"Успешно обработан аудиофайл для пользователя {update.effective_user.full_name} (@{update.effective_user.username})")
async def structure_and_check_transcription(transcription_text):
try:
# Формулировка запросов для структурирования и анализа транскрипции
prompt_structure = f"Структурируйте следующий текст транскрипции в формат беседы менеджера и клиента. Определи имена собеседников. Не используй обобщения. Исправь орфографичеие ошибки при необходимости: {transcription_text}"
prompt_analyze = f"Проанализируйте предоставленный аудиозапись или текст разговора менеджера с клиентом. Анализируйет разговор менеджера максимально строго, от вашего анализа зависит прибыль компании. Проверьте выполнение скрипта продаж по следующим критериям 1. Начало разговора Было ли приветствие теплым и профессиональным? Представил ли себя менеджер и компанию? Установил ли менеджер потребность клиента в начале разговора? 2. Описание продукта и предложение Дал ли менеджер ясное и краткое описание продукта? Объяснил ли менеджер, как продукт может решить проблему клиента? Были ли представлены ключевые преимущества продукта? 3. Обработка возражений Как менеджер отреагировал на возражения клиента по цене? Были ли использованы стратегии для преодоления возражений по функционалу? Предложил ли менеджер альтернативные решения или компромиссы? 4. Заключение сделки Проявил ли менеджер инициативу для заключения сделки? Был ли призыв к действию четким и убедительным? Как менеджер завершил разговор? Было ли предложено дальнейшее действие? 5. Постпродажное обслуживание Попросил ли менеджер обратную связь у клиента? Были ли предложены дополнительные услуги или продукты для улучшения клиентского опыта? Особое внимание уделите Эмоциональному контексту разговора Была ли реакция менеджера адаптирована к эмоциональному состоянию клиента? Интеграции с CRM Занесена ли информация о разговоре в CRM систему корректно? Индивидуальным рекомендациям для улучшения Предложите конкретные советы для улучшения работы менеджера на основе анализа разговора"
# Выполнение запроса к модели для структурирования транскрипции
response_structure = await client.chat.completions.create(
model="gpt-3.5-turbo-0125",
messages=[{"role": "system", "content": prompt_structure}],
)
structured_text = response_structure.choices[0].message.content.strip()
# Выполнение запроса к модели для анализа структурированного текста
response_analyze = await client.chat.completions.create(
model="gpt-3.5-turbo-0125",
messages=[{"role": "system", "content": prompt_analyze + structured_text}],
)
analysis_text = response_analyze.choices[0].message.content.strip()
return analysis_text, structured_text
except Exception as e:
logger.error(f"Произошла ошибка при выполнении структурирования и анализа транскрипции: {e}")
return None, None
async def send_broadcast_message(bot, message_text):
conn = await db_connect()
try:
subscribers = await conn.fetch('SELECT telegram_id FROM users')
logger.info("Получен список подписчиков для рассылки")
except Exception as e:
logger.error(f"Ошибка при получении списка подписчиков: {e}")
return 0, 0, []
finally:
await conn.close()
success_count = 0
failed_count = 0
failed_subscribers = []
for record in subscribers:
subscriber_id = record['telegram_id']
try:
await bot.send_message(chat_id=subscriber_id, text=message_text)
success_count += 1
except Exception as e:
logger.error(f"Не удалось отправить сообщение {subscriber_id}: {e}")
failed_count += 1
failed_subscribers.append(subscriber_id)
logger.info(f"Рассылка завершена. Успешно: {success_count}, Не удалось: {failed_count}")
return success_count, failed_count, failed_subscribers
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
telegram_id = update.effective_user.id
user_message = update.message.text
chat_id = update.effective_chat.id
username = update.effective_user.username
full_name = update.effective_user.full_name
logger.info(f"Получено сообщение от {username}: {user_message}")
# Логика для обработки текста рассылки от администратора
if admin_state.get(telegram_id) == "awaiting_message_text":
if telegram_id == ВАШ Телеграм ID:
success_count, failed_count, failed_subscribers = await send_broadcast_message(context.bot, user_message)
admin_state[telegram_id] = None # Сброс состояния
report_message = f"Рассылка выполнена.\nУспешно отправлено сообщений: {success_count}.\nНе удалось отправить сообщений: {failed_count}."
if failed_count > 0:
report_message += f"\nНе удалось отправить следующим пользователям: {', '.join(failed_subscribers)}"
await update.message.reply_text(report_message)
return
# Сохраняем данные пользователя, если они еще не были сохранены
await save_user_data_if_not_exists(chat_id, telegram_id, username, full_name)
async def sendall_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
telegram_id = update.effective_user.id
if telegram_id == ВАШ Телеграм ID: # Замените на ваш Telegram ID администратора
admin_state[telegram_id] = "awaiting_message_text"
await update.message.reply_text("Пожалуйста, отправьте текст для рассылки.")
else:
await update.message.reply_text("У вас нет прав для выполнения этой команды.")
# Функция для добавления текста в Google Документ
async def append_to_google_doc(document_id, structured_text, analysis_text):
# Укажите путь к файлу учетных данных сервисной учетной записи
SERVICE_ACCOUNT_FILE = 'ссылка на файл от google'
SCOPES = ['https://www.googleapis.com/auth/documents']
credentials = service_account.Credentials.from_service_account_file(
SERVICE_ACCOUNT_FILE, scopes=SCOPES)
service = build('docs', 'v1', credentials=credentials)
# Структура запроса для добавления текста
requests = [
{
'insertText': {
'location': {
'index': 1,
},
'text': f"Структурированная транскрипция:\n{structured_text}\n\nРезультаты анализ звонка:\n{analysis_text}\n\n"
}
}
]
try:
# Выполняем запрос к Google Docs API для добавления текста
result = service.documents().batchUpdate(documentId=document_id, body={'requests': requests}).execute()
logger.info(f'Текст добавлен в документ. Document ID:{document_id}')
except Exception as e:
logger.error(f"Ошибка при добавлении в документ: {e}", exc_info=True)
def main():
"""Запуск бота."""
application = Application.builder().token(TOKEN).build()
application.add_handler(CommandHandler("start", start))
application.add_handler(CommandHandler('sendall', sendall_command))
application.add_handler(MessageHandler(filters.AUDIO, save_audio))
application.run_polling()
if __name__ == '__main__':
main()
Пишите ваши вопросы в моем тг канале, либо здесь в комментариях
