Пишем голосового ассистента на Python
Введение
Технологии в области машинного обучения за последний год развиваются с потрясающей скоростью. Всё больше компаний делятся своими наработками, тем самым открывая новые возможности для создания умных цифровых помощников.
В рамках данной статьи я хочу поделиться своим опытом реализации голосового ассистента и предложить вам несколько идей для того, чтобы сделать его ещё умнее и полезнее.
Что умеет мой голосовой ассистент?
Шаг 1. Обработка голосового ввода
Начнём с того, что научимся обрабатывать голосовой ввод. Нам потребуется микрофон и пара установленных библиотек: PyAudio и SpeechRecognition.
Подготовим main-метод:
import speech_recognition
if __name__ == "__main__":
# инициализация инструментов распознавания и ввода речи
recognizer = speech_recognition.Recognizer()
microphone = speech_recognition.Microphone()
while True:
# старт записи речи с последующим выводом распознанной речи
voice_input = record_and_recognize_audio()
print(voice_input)
Теперь создадим метод для записи и распознавания речи. Для онлайн-распознавания нам потребуется Google, поскольку он имеет высокое качество распознавания на большом количестве языков.
def record_and_recognize_audio(*args: tuple):
"""
Запись и распознавание аудио
"""
with microphone:
recognized_data = ""
# регулирование уровня окружающего шума
recognizer.adjust_for_ambient_noise(microphone, duration=2)
try:
print("Listening...")
audio = recognizer.listen(microphone, 5, 5)
except speech_recognition.WaitTimeoutError:
print("Can you check if your microphone is on, please?")
return
# использование online-распознавания через Google
try:
print("Started recognition...")
recognized_data = recognizer.recognize_google(audio, language="ru").lower()
except speech_recognition.UnknownValueError:
pass
# в случае проблем с доступом в Интернет происходит выброс ошибки
except speech_recognition.RequestError:
print("Check your Internet Connection, please")
return recognized_data
А что делать, если нет доступа в Интернет? Можно воспользоваться решениями для offline-распознавания. Мне лично безумно понравился проект Vosk.
На самом деле, необязательно внедрять offline-вариант, если он вам не нужен. Мне просто хотелось показать оба способа в рамках статьи, а вы уже выбирайте, исходя из своих требований к системе (например, по количеству доступных языков распознавания бесспорно лидирует Google).
Теперь, внедрив offline-решение и добавив в проект нужные языковые модели, при отсутствии доступа к сети у нас автоматически будет выполняться переключение на offline-распознавание.
Замечу, что для того, чтобы не нужно было два раза повторять одну и ту же фразу, я решила записывать аудио с микрофона во временный wav-файл, который будет удаляться после каждого распознавания.
Таким образом, полученный код выглядит следующим образом:
from vosk import Model, KaldiRecognizer # оффлайн-распознавание от Vosk
import speech_recognition # распознавание пользовательской речи (Speech-To-Text)
import wave # создание и чтение аудиофайлов формата wav
import json # работа с json-файлами и json-строками
import os # работа с файловой системой
def record_and_recognize_audio(*args: tuple):
"""
Запись и распознавание аудио
"""
with microphone:
recognized_data = ""
# регулирование уровня окружающего шума
recognizer.adjust_for_ambient_noise(microphone, duration=2)
try:
print("Listening...")
audio = recognizer.listen(microphone, 5, 5)
with open("microphone-results.wav", "wb") as file:
file.write(audio.get_wav_data())
except speech_recognition.WaitTimeoutError:
print("Can you check if your microphone is on, please?")
return
# использование online-распознавания через Google
try:
print("Started recognition...")
recognized_data = recognizer.recognize_google(audio, language="ru").lower()
except speech_recognition.UnknownValueError:
pass
# в случае проблем с доступом в Интернет происходит попытка
# использовать offline-распознавание через Vosk
except speech_recognition.RequestError:
print("Trying to use offline recognition...")
recognized_data = use_offline_recognition()
return recognized_data
def use_offline_recognition():
"""
Переключение на оффлайн-распознавание речи
:return: распознанная фраза
"""
recognized_data = ""
try:
# проверка наличия модели на нужном языке в каталоге приложения
if not os.path.exists("models/vosk-model-small-ru-0.4"):
print("Please download the model from:\n"
"https://alphacephei.com/vosk/models and unpack as 'model' in the current folder.")
exit(1)
# анализ записанного в микрофон аудио (чтобы избежать повторов фразы)
wave_audio_file = wave.open("microphone-results.wav", "rb")
model = Model("models/vosk-model-small-ru-0.4")
offline_recognizer = KaldiRecognizer(model, wave_audio_file.getframerate())
data = wave_audio_file.readframes(wave_audio_file.getnframes())
if len(data) > 0:
if offline_recognizer.AcceptWaveform(data):
recognized_data = offline_recognizer.Result()
# получение данных распознанного текста из JSON-строки
# (чтобы можно было выдать по ней ответ)
recognized_data = json.loads(recognized_data)
recognized_data = recognized_data["text"]
except:
print("Sorry, speech service is unavailable. Try again later")
return recognized_data
if __name__ == "__main__":
# инициализация инструментов распознавания и ввода речи
recognizer = speech_recognition.Recognizer()
microphone = speech_recognition.Microphone()
while True:
# старт записи речи с последующим выводом распознанной речи
# и удалением записанного в микрофон аудио
voice_input = record_and_recognize_audio()
os.remove("microphone-results.wav")
print(voice_input)
Возможно, вы спросите «А зачем поддерживать offline-возможности?»
Я считаю, что всегда стоит учитывать, что пользователь может быть отрезан от сети. В таком случае, голосовой ассистент всё еще может быть полезным, если использовать его как разговорного бота или для решения ряда простых задач, например, посчитать что-то, порекомендовать фильм, помочь сделать выбор кухни, сыграть в игру и т.д.
Шаг 2. Конфигурация голосового ассистента
Поскольку наш голосовой ассистент может иметь пол, язык речи, ну и по классике, имя, то давайте выделим под эти данные отдельный класс, с которым будем работать в дальнейшем.
Для того, чтобы задать нашему ассистенту голос, мы воспользуемся библиотекой для offline-синтеза речи pyttsx3. Она автоматически найдет голоса, доступные для синтеза на нашем компьютере в зависимости от настроек операционной системы (поэтому, возможно, что у вас могут быть доступны другие голоса и вам нужны будут другие индексы).
Также добавим в в main-метод инициализацию синтеза речи и отдельный метод для её проигрывания. Чтобы убедиться, что всё работает, сделаем небольшую проверку на то, что пользователь с нами поздоровался, и выдадим ему обратное приветствие от ассистента:
from vosk import Model, KaldiRecognizer # оффлайн-распознавание от Vosk
import speech_recognition # распознавание пользовательской речи (Speech-To-Text)
import pyttsx3 # синтез речи (Text-To-Speech)
import wave # создание и чтение аудиофайлов формата wav
import json # работа с json-файлами и json-строками
import os # работа с файловой системой
class VoiceAssistant:
"""
Настройки голосового ассистента, включающие имя, пол, язык речи
"""
name = ""
sex = ""
speech_language = ""
recognition_language = ""
def setup_assistant_voice():
"""
Установка голоса по умолчанию (индекс может меняться в
зависимости от настроек операционной системы)
"""
voices = ttsEngine.getProperty("voices")
if assistant.speech_language == "en":
assistant.recognition_language = "en-US"
if assistant.sex == "female":
# Microsoft Zira Desktop - English (United States)
ttsEngine.setProperty("voice", voices[1].id)
else:
# Microsoft David Desktop - English (United States)
ttsEngine.setProperty("voice", voices[2].id)
else:
assistant.recognition_language = "ru-RU"
# Microsoft Irina Desktop - Russian
ttsEngine.setProperty("voice", voices[0].id)
def play_voice_assistant_speech(text_to_speech):
"""
Проигрывание речи ответов голосового ассистента (без сохранения аудио)
:param text_to_speech: текст, который нужно преобразовать в речь
"""
ttsEngine.say(str(text_to_speech))
ttsEngine.runAndWait()
def record_and_recognize_audio(*args: tuple):
"""
Запись и распознавание аудио
"""
with microphone:
recognized_data = ""
# регулирование уровня окружающего шума
recognizer.adjust_for_ambient_noise(microphone, duration=2)
try:
print("Listening...")
audio = recognizer.listen(microphone, 5, 5)
with open("microphone-results.wav", "wb") as file:
file.write(audio.get_wav_data())
except speech_recognition.WaitTimeoutError:
print("Can you check if your microphone is on, please?")
return
# использование online-распознавания через Google
# (высокое качество распознавания)
try:
print("Started recognition...")
recognized_data = recognizer.recognize_google(audio, language="ru").lower()
except speech_recognition.UnknownValueError:
pass
# в случае проблем с доступом в Интернет происходит
# попытка использовать offline-распознавание через Vosk
except speech_recognition.RequestError:
print("Trying to use offline recognition...")
recognized_data = use_offline_recognition()
return recognized_data
def use_offline_recognition():
"""
Переключение на оффлайн-распознавание речи
:return: распознанная фраза
"""
recognized_data = ""
try:
# проверка наличия модели на нужном языке в каталоге приложения
if not os.path.exists("models/vosk-model-small-ru-0.4"):
print("Please download the model from:\n"
"https://alphacephei.com/vosk/models and unpack as 'model' in the current folder.")
exit(1)
# анализ записанного в микрофон аудио (чтобы избежать повторов фразы)
wave_audio_file = wave.open("microphone-results.wav", "rb")
model = Model("models/vosk-model-small-ru-0.4")
offline_recognizer = KaldiRecognizer(model, wave_audio_file.getframerate())
data = wave_audio_file.readframes(wave_audio_file.getnframes())
if len(data) > 0:
if offline_recognizer.AcceptWaveform(data):
recognized_data = offline_recognizer.Result()
# получение данных распознанного текста из JSON-строки
# (чтобы можно было выдать по ней ответ)
recognized_data = json.loads(recognized_data)
recognized_data = recognized_data["text"]
except:
print("Sorry, speech service is unavailable. Try again later")
return recognized_data
if __name__ == "__main__":
# инициализация инструментов распознавания и ввода речи
recognizer = speech_recognition.Recognizer()
microphone = speech_recognition.Microphone()
# инициализация инструмента синтеза речи
ttsEngine = pyttsx3.init()
# настройка данных голосового помощника
assistant = VoiceAssistant()
assistant.name = "Alice"
assistant.sex = "female"
assistant.speech_language = "ru"
# установка голоса по умолчанию
setup_assistant_voice()
while True:
# старт записи речи с последующим выводом распознанной речи
# и удалением записанного в микрофон аудио
voice_input = record_and_recognize_audio()
os.remove("microphone-results.wav")
print(voice_input)
# отделение комманд от дополнительной информации (аргументов)
voice_input = voice_input.split(" ")
command = voice_input[0]
if command == "привет":
play_voice_assistant_speech("Здравствуй")
На самом деле, здесь бы хотелось самостоятельно научиться писать синтезатор речи, однако моих знаний здесь не будет достаточно. Если вы можете подсказать хорошую литературу, курс или интересное документированное решение, которое поможет разобраться в этой теме глубоко — пожалуйста, напишите в комментариях.
Шаг 3. Обработка команд
Теперь, когда мы «научились» распознавать и синтезировать речь с помощью просто божественных разработок наших коллег, можно начать изобретать свой велосипед для обработки речевых команд пользователя : D
В моём случае я использую мультиязычные варианты хранения команд, поскольку у меня в демонстрационном проекте не так много событий, и меня устраивает точность определения той или иной команды. Однако, для больших проектов я рекомендую разделить конфигурации по языкам.
Для хранения команд я могу предложить два способа.
1 способ
Можно использовать прекрасный JSON-подобный объект, в котором хранить намерения, сценарии развития, ответы при неудавшихся попытках (такие часто используются для чат-ботов). Выглядит это примерно вот так:
config = {
"intents": {
"greeting": {
"examples": ["привет", "здравствуй", "добрый день",
"hello", "good morning"],
"responses": play_greetings
},
"farewell": {
"examples": ["пока", "до свидания", "увидимся", "до встречи",
"goodbye", "bye", "see you soon"],
"responses": play_farewell_and_quit
},
"google_search": {
"examples": ["найди в гугл",
"search on google", "google", "find on google"],
"responses": search_for_term_on_google
},
},
"failure_phrases": play_failure_phrase
}
Такой вариант подойдёт тем, кто хочет натренировать ассистента на то, чтобы он отвечал на сложные фразы. Более того, здесь можно применить NLU-подход и создать возможность предугадывать намерение пользователя, сверяя их с теми, что уже есть в конфигурации.
Подробно этот способ мы его рассмотрим на 5 шаге данной статьи. А пока обращу ваше внимание на более простой вариант
2 способ
Можно взять упрощенный словарь, у которого в качестве ключей будет hashable-тип tuple, а в виде значений будут названия методов, которые будут выполняться. Для коротких команд подойдёт вот такой вариант:
commands = {
("hello", "hi", "morning", "привет"): play_greetings,
("bye", "goodbye", "quit", "exit", "stop", "пока"): play_farewell_and_quit,
("search", "google", "find", "найди"): search_for_term_on_google,
("video", "youtube", "watch", "видео"): search_for_video_on_youtube,
("wikipedia", "definition", "about", "определение", "википедия"): search_for_definition_on_wikipedia,
("translate", "interpretation", "translation", "перевод", "перевести", "переведи"): get_translation,
("language", "язык"): change_language,
("weather", "forecast", "погода", "прогноз"): get_weather_forecast,
}
Для его обработки нам потребуется дополнить код следующим образом:
def execute_command_with_name(command_name: str, *args: list):
"""
Выполнение заданной пользователем команды и аргументами
:param command_name: название команды
:param args: аргументы, которые будут переданы в метод
:return:
"""
for key in commands.keys():
if command_name in key:
commands[key](*args)
else:
pass # print("Command not found")
if __name__ == "__main__":
# инициализация инструментов распознавания и ввода речи
recognizer = speech_recognition.Recognizer()
microphone = speech_recognition.Microphone()
while True:
# старт записи речи с последующим выводом распознанной речи
# и удалением записанного в микрофон аудио
voice_input = record_and_recognize_audio()
os.remove("microphone-results.wav")
print(voice_input)
# отделение комманд от дополнительной информации (аргументов)
voice_input = voice_input.split(" ")
command = voice_input[0]
command_options = [str(input_part) for input_part in voice_input[1:len(voice_input)]]
execute_command_with_name(command, command_options)
В методы будут передаваться дополнительные аргументы, сказанные после командного слова. То есть, если сказать фразу »видео милые котики», команда »видео» вызовет метод search_for_video_on_youtube () с аргументом »милые котики» и выдаст вот такой результат:
Пример такого метода с обработкой входящих аргументов:
def search_for_video_on_youtube(*args: tuple):
"""
Поиск видео на YouTube с автоматическим открытием ссылки на список результатов
:param args: фраза поискового запроса
"""
if not args[0]: return
search_term = " ".join(args[0])
url = "https://www.youtube.com/results?search_query=" + search_term
webbrowser.get().open(url)
# для мультиязычных голосовых ассистентов лучше создать
# отдельный класс, который будет брать перевод из JSON-файла
play_voice_assistant_speech("Here is what I found for " + search_term + "on youtube")
Ну вот и всё! Основной функционал бота готов. Далее вы можете до бесконечности улучшать его различными методами. Моя реализация с подробными комментариями доступна на моём GitHub.
Ниже мы рассмотрим ряд улучшений, чтобы сделать нашего ассистента ещё умнее.
Шаг 4. Добавление мультиязычности
Чтобы научить нашего ассистента работать с несколькими языковыми моделями, будет удобнее всего организовать небольшой JSON-файл с простой структурой:
{
"Can you check if your microphone is on, please?": {
"ru": "Пожалуйста, проверь, что микрофон включен",
"en": "Can you check if your microphone is on, please?"
},
"What did you say again?": {
"ru": "Пожалуйста, повтори",
"en": "What did you say again?"
},
}
В моём случае я использую переключение между русским и английским языком, поскольку мне для этого доступны модели для распознавания речи и голоса для синтеза речи. Язык будет выбран в зависимости от языка речи самого голосового ассистента.
Для того, чтобы получать перевод мы можем создать отдельный класс с методом, который будет возвращать нам строку с переводом:
class Translation:
"""
Получение вшитого в приложение перевода строк для
создания мультиязычного ассистента
"""
with open("translations.json", "r", encoding="UTF-8") as file:
translations = json.load(file)
def get(self, text: str):
"""
Получение перевода строки из файла на нужный язык (по его коду)
:param text: текст, который требуется перевести
:return: вшитый в приложение перевод текста
"""
if text in self.translations:
return self.translations[text][assistant.speech_language]
else:
# в случае отсутствия перевода происходит вывод сообщения
# об этом в логах и возврат исходного текста
print(colored("Not translated phrase: {}".format(text), "red"))
return text
В main-методе до цикла объявим наш переводчик таким образом: translator = Translation ()
Теперь при проигрывании речи ассистента мы сможем получить перевод следующим образом:
play_voice_assistant_speech(translator.get(
"Here is what I found for {} on Wikipedia").format(search_term))
Как видно из примера выше, это работает даже для тех строк, которые требуют вставки дополнительных аргументов. Таким образом можно переводить «стандартные» наборы фраз для ваших ассистентов.
Шаг 5. Немного машинного обучения
А теперь вернёмся к характерному для большинства чат-ботов варианту с JSON-объектом для хранения команд из нескольких слов, о котором я упоминала в пункте 3. Для работы с ним нам нужно будет добавить пару методов:
def prepare_corpus():
"""
Подготовка модели для угадывания намерения пользователя
"""
corpus = []
target_vector = []
for intent_name, intent_data in config["intents"].items():
for example in intent_data["examples"]:
corpus.append(example)
target_vector.append(intent_name)
training_vector = vectorizer.fit_transform(corpus)
classifier_probability.fit(training_vector, target_vector)
classifier.fit(training_vector, target_vector)
def get_intent(request):
"""
Получение наиболее вероятного намерения в зависимости от запроса пользователя
:param request: запрос пользователя
:return: наиболее вероятное намерение
"""
best_intent = classifier.predict(vectorizer.transform([request]))[0]
index_of_best_intent = list(classifier_probability.classes_).index(best_intent)
probabilities = classifier_probability.predict_proba(vectorizer.transform([request]))[0]
best_intent_probability = probabilities[index_of_best_intent]
# при добавлении новых намерений стоит уменьшать этот показатель
if best_intent_probability > 0.57:
return best_intent
А также немного модифицировать main-метод, добавив инициализацию переменных для подготовки модели и изменив цикл на версию, соответствующую новой конфигурации:
# подготовка корпуса для распознавания запросов пользователя с некоторой вероятностью
# (поиск похожих)
vectorizer = TfidfVectorizer(analyzer="char", ngram_range=(2, 3))
classifier_probability = LogisticRegression()
classifier = LinearSVC()
prepare_corpus()
while True:
# старт записи речи с последующим выводом распознанной речи
# и удалением записанного в микрофон аудио
voice_input = record_and_recognize_audio()
if os.path.exists("microphone-results.wav"):
os.remove("microphone-results.wav")
print(colored(voice_input, "blue"))
# отделение команд от дополнительной информации (аргументов)
if voice_input:
voice_input_parts = voice_input.split(" ")
# если было сказано одно слово - выполняем команду сразу
# без дополнительных аргументов
if len(voice_input_parts) == 1:
intent = get_intent(voice_input)
if intent:
config["intents"][intent]["responses"]()
else:
config["failure_phrases"]()
# в случае длинной фразы - выполняется поиск ключевой фразы
# и аргументов через каждое слово,
# пока не будет найдено совпадение
if len(voice_input_parts) > 1:
for guess in range(len(voice_input_parts)):
intent = get_intent((" ".join(voice_input_parts[0:guess])).strip())
if intent:
command_options = [voice_input_parts[guess:len(voice_input_parts)]]
config["intents"][intent]["responses"](*command_options)
break
if not intent and guess == len(voice_input_parts)-1:
config["failure_phrases"]()
Однако, такой способ сложнее контролировать: он требует постоянной проверки того, что та или иная фраза всё ещё верно определяется системой как часть того или иного намерения. Поэтому данным способом стоит пользоваться с аккуратностью (либо экспериментировать с самой моделью).
Заключение
На этом мой небольшой туториал подошёл к концу.
Мне будет приятно, если вы поделитесь со мной в комментариях известными вам open-source решениями, которые можно внедрить в данный проект, а также вашими идеями касательно того, какие ещё online и offline-функции можно реализовать.
Документированные исходники моего голосового ассистента в двух вариантах можно найти здесь.
