Нейросеть мне в помощь или как я сделал телеграм бота, который умеет переводить песни

Однажды, когда я искал эффективное решение для преобразования речи в текст (транскрибации), чтобы применить его в своем проекте умной колонки, обнаружил интересное решение под названием Whisper от широко известной компании Open AI. К сожалению, Whisper не подошел для реализации в моем проекте по «аппаратным» причинам, но его функционал отпечатался в моей душе. Прошло время и меня посетила идея: «Почему бы не разработать телеграмм бота, куда бы пользователь мог отправлять аудиофайл, а в ответ получал текстовую расшифровку и перевод (песни) на родной язык». В этой статье я расскажу о реализации данной идеи и Whisper в этом проекте займет одну из ключевых функций.

❯ Чем же меня так впечатлил Whisper?

Во первых, что такое Whisper?

Whisper — это универсальная модель распознавания речи. Она обучена на большом наборе данных разнообразной аудиоинформации и является многофункциональной моделью, способной выполнять мультиязычное распознавание речи, перевод речи и идентификацию языка.

Ну и само название модели «Whisper» переводится как шепот, что само по себе намекает о качестве распознавания речи.

Чтобы не быть голословным, давайте запустим базовый пример использования и посмотрим вывод:

import whisper model = whisper.load_model("base")  # load audio and pad/trim it to fit 30 seconds audio = whisper.load_audio("audio.mp3") audio = whisper.pad_or_trim(audio)  # make log-Mel spectrogram and move to the same device as the model mel = whisper.log_mel_spectrogram(audio).to(model.device)  # detect the spoken language _, probs = model.detect_language(mel) print(f"Detected language: {max(probs, key=probs.get)}")  # decode the audio options = whisper.DecodingOptions() result = whisper.decode(model, mel, options)  # print the recognized text print(result)

В результате выполнения скрипта мы получим следующий вывод в формате json:

{'text': ' Солнечное излучение (большой текст обрезал).',  'segments': [ {'id': 0, 'seek': 0, 'start': 0.0, 'end': 4.84, 'text': ' Солнечное излучение является одним из самых мощных источников энергии в Селенной.', 'tokens': [50364, 2933, 20470, 4310, 6126, 3943, 41497, 5627, 29755, 50096, 3943, 37241, 39218, 5783, 12410, 20483, 22122, 40804, 7347, 740, 2933, 21180, 5007, 13, 50606], 'temperature': 0.0, 'avg_logprob': -0.1954750242687407, 'compression_ratio': 2.0371621621621623, 'no_speech_prob': 0.0006262446404434741}, {'id': 1, 'seek': 0, 'start': 4.84, 'end': 8.4, 'text': ' Оно поступает на Землю в виде видимого и инфракрасного света,', 'tokens': [50606, 3688, 1234, 43829, 3310, 1470, 42604, 6578, 740, 12921, 38273, 2350, 1006, 6635, 3619, 481, 1272, 17184, 4699, 4155, 20513, 11, 50784], 'temperature': 0.0, 'avg_logprob': -0.1954750242687407, 'compression_ratio': 2.0371621621621623, 'no_speech_prob': 0.0006262446404434741}, {'id': 2, 'seek': 0, 'start': 8.4, 'end': 10.56, 'text': ' а также в виде космических лучей.', 'tokens': [50784, 2559, 16584, 740, 12921, 31839, 919, 38911, 15525, 2345, 13, 50892], 'temperature': 0.0, 'avg_logprob': -0.1954750242687407, 'compression_ratio': 2.0371621621621623, 'no_speech_prob': 0.0006262446404434741},  {'id': 3, 'seek': 0, 'start': 10.56, 'end': 15.8, 'text': ' Солнечное излучение имеет мощность примерно в квадратный метр поверхности Земли.', 'tokens': [50892, 2933, 20470, 4310, 6126, 3943, 41497, 5627, 33761, 39218, 9930, 37424, 740, 35350, 2601, 11157, 4441, 18791, 481, 44397, 13975, 42604, 1675, 13, 51154], 'temperature': 0.0, 'avg_logprob': -0.1954750242687407, 'compression_ratio': 2.0371621621621623, 'no_speech_prob': 0.0006262446404434741}, {'id': 4, 'seek': 0, 'start': 15.8, 'end': 19.44, 'text': ' Это достаточно для производства электроэнергии и других применений.', 'tokens': [51154, 6684, 28562, 5561, 28685, 12115, 31314, 9938, 7570, 489, 19480, 7347, 1006, 31211, 31806, 1008, 17271, 13, 51336], 'temperature': 0.0, 'avg_logprob': -0.1954750242687407, 'compression_ratio': 2.0371621621621623, 'no_speech_prob': 0.0006262446404434741} ], 'language': 'ru'} 

Как можно видеть из содержания вывода, Whisper выводит довольно информативный результат транскрибации, с разбивкой текста на сегменты и указанием временных меток, языка и другой информации. Используя эту информацию, мы можем сформировать удобный и красивый текстовой файл расшифровки аудио.

❯ Улучшения приходят в процессе творчества

Изначально планировалось реализовать только текстовую расшифровку аудиофайла(песни), но потом мелькнула мысль: «Почему бы не реализовать перевод текста, отличного от родного языка? Ведь так можно узнать смысл любимой песни всего за пару минут.»
Для реализации этой идеи, я решил пойти простым путём: использовать API сервисов онлайн переводчиков. К сожалению, а может и к счастью, это была моя ошибка, ввиду ограничения запросов на данные API. Ну что ж, подумал я, меньше внешних сервисов — больше свободы, будем использовать локальные алгоритмы перевода на базе нейронных сетей.

❯ Оффлайн перевод с помощью нейронных сетей

Трансформеры нам помогут! Не долго размышляя, решил использовать библиотеку Transformers, которая позволяет применять в своих проектах большое количество моделей, полный список можно посмотреть на сайте проекта.

Для оффлайн перевода будем использовать модели Helsinki-NLP от Группы исследований языковых технологий Хельсинкского университета.

Пример Python скрипта для оффлайн перевода:

from transformers import pipeline import torch  device = "cuda:0" if torch.cuda.is_available() else "cpu" torch_dtype = torch.float16 if torch.cuda.is_available() else torch.float32 # устанавливаем количество потоков для torch torch.set_num_threads(4) #Перевод с любого языка на английский translator   = pipeline("translation", model="Helsinki-NLP/opus-mt-mul-en") text         = "Какой-то текст на любом языке" translation  = translator(text) print(translation[0]['translation_text']) 

В примере указан мультиязычный перевод на английский. К сожалению, у Helsinki-NLP нет мультиязычных моделей с переводом на русский язык, поэтому воспользуемся трюком большинства генеративных ИИ и применим двойной перевод, то есть в нашем случае, с английского языка на русский, используя модель Helsinki-NLP/opus-mt-en-ru.

❯ Ну что ж, давайте собирать бота

Учитывая все свои размышления по функциональности бота, получился следующий результат, для удобного восприятия кода, я разделил скрипт на несколько файлов:

import neural_process import file_process import config from telegram import ForceReply, Update from telegram.ext import Application, CommandHandler, MessageHandler, filters, CallbackContext, ContextTypes   # Функция, которая вызывается при отправке файла async def handle_file(update: Update, context: CallbackContext):     audio = update.message.audio     print(audio.mime_type)     # Сохраняем аудиофайл локально и получаем путь сохранения     file = await context.bot.get_file(audio.file_id)     f_name = audio.file_name     local_file_path = file_process.save_audio_file(f_name, file.file_path)     # Отправляем сообщение о успешном сохранении аудиофайла     await update.message.reply_text(         f"Подождите немного, я в ускоренном темпе прослушаю аудиофайл {f_name} и отправлю вам текстовую "         f"расшифровку с переводом."     )     # проверяем файл на длительность     status = True     status_dur = True     try:         file_dur = file_process.file_duration_check(local_file_path)         if file_dur > 600:             status_dur = False     except Exception as e:         print(f"Возникла ошибка: {e}")         status_dur = False      # Получаем перевод      trance_text = 'в эту переменную сохраняется текст транскрибации с переводом'     if status_dur:         try:             start_time = time.time()             trance_text = neural_process.final_process(local_file_path, f_name)         except Exception as e:             print(f"Возникла ошибка: {e}")             status = False     else:         status = False         await update.message.reply_text(             f"Сожалею, но возникла ошибка обработки файла. Длительность файла не должна превышать десять минут."         )      if status:         # Сохраняем вывод в текстовый файл         tx_file_path = file_process.save_text_to_file(f_name, trance_text)         end_time = time.time()         process_time = round(end_time - start_time, 2)         await update.message.reply_text(             f"Перевод готов! Ловите файл с переводом. Затраченное время: {process_time} сек.  "         )         # Отправляем текстовый файл         with open(tx_file_path, "rb") as document:             await context.bot.send_document(chat_id=update.message.chat_id, document=document)      else:         file_process.delete_file(local_file_path)         if status_dur:             await update.message.reply_text(                 f"Сожалею, но возникла ошибка обработки файла. Пожалуйста, убедитесь что файл имеет правильный аудиоформат."             )  # Функция для команды /start async def start(update: Update, context: CallbackContext) -> None:     await update.message.reply_text(         'Привет! Добро пожаловать! Нравится песня, но ты не знаешь о чем она? Я могу перевести песню с любого языка '         'на русский за считанные секунды, просто скинь аудио файл в чат. ')   # Функция для команды / help async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:     """Send a message when the command /help is issued."""     user = update.effective_user     await update.message.reply_html(         rf"{user.mention_html()}, я могу перевести песню с любого языка на русский, просто скинь аудио файл в чат.",         reply_markup=ForceReply(selective=False),     )  # Запускаем бота def main():     # Токен  бота     application = Application.builder().token(config.tg_key).build()     # Регистрируем обработчик для команды /start     application.add_handler(CommandHandler("start", start))     application.add_handler(CommandHandler("help", help_command))     # Регистрируем обработчик для приема файлов     application.add_handler(MessageHandler(filters.AUDIO, handle_file))     # Запускаем бота     application.run_polling()     # Останавливаем бота при нажатии Ctrl+C     application.idle()   if __name__ == '__main__':     main() 

import requests import os from pydub.utils import mediainfo   def delete_file(file_path):     try:         os.remove(file_path)         print(f"Файл {file_path} успешно удален.")     except FileNotFoundError:         print(f"Файл {file_path} не найден.")     except Exception as e:         print(f"Произошла ошибка при удалении файла {file_path}: {e}")   def save_audio_file(file_name, file_link):     response = requests.get(file_link)     # Сохраняем аудиофайл локально для последующей обработки     sound_folder = 'sound'     local_file_path = f"{sound_folder}/{file_name}"      if not os.path.exists(sound_folder):         # Если папки нет, создаем её         os.makedirs(sound_folder)      with open(local_file_path, 'wb') as local_file:         local_file.write(response.content)     return local_file_path   def save_text_to_file(file_name, trance_text):     text_folder = 'output'     local_file_path = f"{text_folder}/{file_name.replace('.', '_') + '.txt'}"      if not os.path.exists(text_folder):         # Если папки нет, создаем её         os.makedirs(text_folder)     # Сохраняем вывод в текстовый файл     with open(local_file_path, "w", encoding="utf-8") as output_file:         output_file.write(trance_text)      return local_file_path   def file_duration_check(file_path):     audio_info = mediainfo(file_path)     duration = float(audio_info['duration'])     print(duration)     return duration 

from transformers import pipeline import torch import whisper import file_process  device = "cuda:0" if torch.cuda.is_available() else "cpu" torch_dtype = torch.float16 if torch.cuda.is_available() else torch.float32 # устанавливаем количество потоков для torch torch.set_num_threads(4)   def sound_to_text(audios):     model = whisper.load_model("base")     # load audio and pad/trim it to fit 30 seconds     audio = whisper.load_audio(audios)     audio = whisper.pad_or_trim(audio)     # make log-Mel spectrogram and move to the same device as the model     mel = whisper.log_mel_spectrogram(audio).to(model.device)     # detect the spoken language     _, probs = model.detect_language(mel)     lang = max(probs, key=probs.get)     print(f"Detected language: {max(probs, key=probs.get)}")     # decode the audio     result = model.transcribe(audios, fp16=False, language=lang)     print(result)     return result['segments'], lang   def final_process(file, file_name):     print(f"Сохранен в директории: {file}")     raw, detected_lang = sound_to_text(file)     translator = pipeline("translation", model="Helsinki-NLP/opus-mt-mul-en")     translator2 = pipeline("translation", model="Helsinki-NLP/opus-mt-en-ru")     text = f"Перевод аудиофайла: {file_name} \n"     text += f"В файле используется {get_language_name(detected_lang)} язык. \n"     for segment in raw:         text += "-------------------- \n"         text += f"ID элемента: {segment['id']} Начало: {int(segment['start'])} --- Конец: {int(segment['end'])} \n"         text += f"Исходный текст:{segment['text']} \n"         if detected_lang == 'en':             text_en = segment['text']             translation2 = translator2(text_en)             text += f"Перевод: {translation2[0]['translation_text']} \n"         elif detected_lang == 'ru':             text += ""         else:             translation = translator(segment['text'])             text_en = translation[0]['translation_text']             translation2 = translator2(text_en)             text += f"Перевод: {translation2[0]['translation_text']} \n"      file_process.delete_file(file)     print(text)     return text   def get_language_name(code):     languages = {         'ru': 'русский',         'en': 'английский',         'zh': 'китайский',         'es': 'испанский',         'ar': 'арабский',         'he': 'иврит',         'hi': 'хинди',         'bn': 'бенгальский',         'pt': 'португальский',         'fr': 'французский',         'de': 'немецкий',         'ja': 'японский',         'pa': 'панджаби',         'jv': 'яванский',         'te': 'телугу',         'ms': 'малайский',         'ko': 'корейский',         'vi': 'вьетнамский',         'ta': 'тамильский',         'it': 'итальянский',         'tr': 'турецкий',         'uk': 'украинский',         'pl': 'польский',     }     return languages.get(code, 'неизвестный язык') 

Для запуска скрипта, нам необходимо установить следующие пакеты, используя команду:

pip install transformers, torch, accelerate, sentencepiece, sacremoses, python-telegram-bot, openai-whisper, pydub 

❯ Деплоим бота

Чтобы всё заработало, не деплойте по пятницам.

Наш телеграмм бот, который работает с использованием алгоритмов ML, достаточно требователен к аппаратным ресурсам. Да, вы можете его использовать на своем ПК, но целесообразнее использовать облачные сервисы для подобных целей. Для размещения своего телеграмм бота, я решил воспользоваться облачной инфраструктурой от компании Timeweb Cloud. Итак, приступим к созданию нашего облачного сервиса.

Для создания нашего облачного сервиса, нам необходимо войти в панель управления под вашей учетной записью. Если учетная запись не создана, то регистрация учетной записи займет всего пару кликов. Я, например, для регистрации и авторизации воспользовался аккаунтом Google.
Создаем новый сервис, выбираем пункт «Облачный сервер»:

Далее нам предстоит выбрать операционную систему для нашего сервера, я выбрал Debian 11:

Выбираем расположение нашего сервера:

Выбираем конфигурацию сервера, как видно на изображении, для начальных тестов я выбрал конфигурацию с четырьмя ядрами CPU и 8 ГБ оперативной памяти:

Вот и все основные опции, которые нам нужны для создания сервера, остальные опции выбираете/указываете по желанию. Чтобы создать сервер, необходимо нажать кнопку «Заказать», которая располагается справа.

Процесс создания сервера:

Сервер создан:

После успешного создания сервера, перейдя в дашборд, с правой стороны вы увидите параметры для удаленного подключения к созданному серверу:

Всё выполняется очень быстро, на создание сервера ушло пару минут.

❯ Подключаемся, настраиваем сервер для размещения бота

Далее мы будем использовать терминал для подключения к нашему облачному сервису.
Запускаем терминал и подключаемся к нашему серверу, после подключения нам необходимо проверить и установить доступные обновления:

apt-get update & upgrade

Запускать скрипты с root правами — как минимум, это дурной тон, поэтому создаем нового пользователя с именем tg_bot:

adduser tg_bot

и пропишем его в группу sudo:

usermod -aG sudo tg_bot

Затем нам нужно выйти из учетной записи root и подключиться под новым пользователем tg_bot, и далее все операции мы будем выполнять под этой учетной записью.

Проверим обновления Python3 и установим, если имеются:

sudo apt update sudo apt install python3

Установим менеджер пакетов:

sudo apt install python3-pip

Так как мы работаем с медиафайлами, нам необходимо установить набор библиотек FFmpeg:

sudo apt install ffmpeg

Для оценки производительности, на всякий случай установим системный монитор htop:

sudo apt install htop

Устанавливаем необходимые зависимости, для работы нашего бота:

pip install transformers, torch, accelerate, sentencepiece, sacremoses, python-telegram-bot, openai-whisper, pydub

Создаем рабочую директорию для нашего бота:

mkdir tg_bot_scripts

Затем переходим в созданную директорию, с помощью команды:

cd tg_bot_scripts

Далее нам нужно создать файлы Python нашего бота, с помощью команд:

nano tg_bot.py

В открывшемся текстовом редакторе вставляем код из статьи и сохраняем, аналогично поступаем и с другими файлами.

nano neural_process.py

nano file_process.py

nano config.py

файл config.py содержит только одну переменную, которая содержит секретный ключ телеграм-бота.

tg_key = 'Ваш секретный ключ'

Или можно воспользоваться альтернативным способом, клонировав код моего репозитория GitHub.

Чтобы это реализовать, нам нужно установить Git:

 sudo apt install git 

После установки Git, выполнить следующую команду:

 git clone https://github.com/VGCH/music_translate_AI_bot 

Далее нам необходимо переименовать созданную при клонировании Git репозитория папку music_translate_AI_bot с помощью команды:

 mv music_translate_AI_bot tg_bot_scripts

После этого важно не забыть добавить токен телеграмм бота:

 cd tg_bot_scripts  nano config.py 

Настало время тестового запуска, запустим нашего бота с помощью команды:

pyton3 tg_bot.py

Если бот запустился и работает нормально, то нам необходимо создать системный сервис для автоматического запуска нашего бота. Создадим файл нашего сервиса с помощью команды:

sudo nano /etc/systemd/system/AI_tg_bot.service

Копируем в открывшийся текстовый редактор следующее содержимое и сохраняем файл:

[Unit] Description=My AI Bot service After=network.target  [Service] WorkingDirectory=/home/tg_bot/tg_bot_scripts/ User=tg_bot Type=simple Restart=always ExecStart=/usr/bin/python3 /home/tg_bot/tg_bot_scripts/tg_bot.py > /dev/null  [Install] WantedBy=multi-user.target 

Чтобы добавить созданный сервис в автозагрузку, используйте следующую команду:

 sudo systemctl enable AI_tg_bot.service 

Чтобы запустить сервис используйте команду:

 sudo systemctl start AI_tg_bot.service 

Чтобы остановить:

 sudo systemctl stop AI_tg_bot.service 

❯ Изменение конфигурации сервера

В процессе отладки бота, я решил изменить конфигурацию своего облачного сервера с четырехъядерного CPU на восьмиядерный. Захожу в панель управления Timeweb Cloud, выбрав свой сервер, перехожу на вкладку «конфигурация»:

И выбираю необходимую конфигурацию:

Менее минуты и сервер с новой конфигурацией в работе. Еще мне понравилось то, что оплата облачного сервера выполняется только за фактические часы работы и вы в любой момент можете изменить конфигурацию, без необходимости вноса оплаты за месяц вперед. В панели управления вы можете видеть ваш текущий баланс и дату окончания средств при текущей конфигурации сервера (или серверов).

При изменении количества ядер CPU на облачном сервере, так же необходимо в скрипте обработки neural_process.py изменить количество потоков для torch, в моем случае на восемь.

torch.set_num_threads(8)

После изменения конфигурации сервера, можно понаблюдать нагрузку при обработке аудиофайла нашим телеграмм ботом с помощью команды:

htop

❯ Заключение

В итоге у нас получился интересный эксперимент реализации телеграм-бота с применением алгоритмов машинного обучения для распознавания речи и перевода текста.
Спасибо вам за уделенное время и внимание. Есть вопросы, критика, осуждение? Добро пожаловать в комментарии. 🙂

Полезные ссылки:

• Облачные сервисы Timeweb Cloud — это реферальная ссылка, которая может помочь поддержать меня и мои проекты.
• Мой GitHub с исходным кодом проекта.

Возможно, захочется почитать и это:

• ➤ Моя б̶е̶з̶умная колонка или бюджетный DIY голосового ассистента для умного дома
• ➤ Большая дыра в безопасности Android: почему смартфоны с зеленым роботом подвержены кражам?
• ➤ Как проклятие невидимой стены ждало меня 20 лет
• ➤ Разрабатываем полезное браузерное расширение легким движением руки
• ➤ «Дюна» – от Герберта до Вильнёва