Голосовые помощники вроде Алисы и Siri прочно вошли в нашу жизнь. Но что, если создать своего собственного цифрового помощника, который будет работать полностью офлайн, не отправляя ваши данные в облако, и которого можно научить именно тому, что нужно вам? Сегодня мы разберем, как написать такого ассистента на Python, используя открытые библиотеки.
Архитектура ассистента «Шустрик»
Представленный код — это прототип голосового помощника по имени «Шустрик». Его ключевая особенность — модульность и использование локальных моделей для синтеза речи (Silero), что обеспечивает приватность и скорость отклика.
Ассистент состоит из нескольких логических блоков:
-
Голосовой ввод (
Voiceвmain.py): Распознавание речи черезspeech_recognitionи Google Speech Recognition API (онлайн). -
Обработка намерений (
Queryвquery_request.py): Машинное обучение на базеscikit-learnдля определения цели команды (узнать время, погоду, выполнить поиск и т.д.). -
Голосовой вывод (
Silero_вsiler_audio.py): Локальный синтез русской речи с помощью модели Silero от SberAI. Это «голос» ассистента. -
Выполнение задач:
-
Поиск в Google через Selenium (
Search_google). -
Получение погоды через OpenWeatherMap API (
pyowm). -
Определение времени (
datetime).
-
-
Управление потоком: Многопоточность (
threading) позволяет ассистенту постоянно «слушать» фон, не блокируя работу.
Как это работает? Пошаговый разбор
-
Запуск и калибровка: При старте ассистент калибрует микрофон, чтобы отфильтровать фоновый шум.
-
Ожидание ключевого слова: Основной цикл непрерывно записывает короткие аудиофрагменты и пытается распознать в них слово «Шустрик». Это активирует бота.
-
Распознавание команды: После ключевого слова ассистент записывает следующую фразу (до 8 секунд) и отправляет ее на распознавание в Google.
-
Понимание намерения: Распознанный текст передается классификатору. Он сравнивает его с заранее подготовленными примерами из
data.json(например, «сколько времени», «который час») и определяет интент — тип запроса (time,weather,search). -
Выполнение и ответ:
-
Время: Ассистент получает текущий час и минуту, преобразует числа в слова («пять», «сорок») и отдает текст модели Silero для озвучки.
-
Погода: Делает запрос к API погоды, формирует фразу и озвучивает ее
-
Поиск в Google: Самая сложная задача. Запускается «невидимый» браузер (Selenium), который выполняет поисковый запрос, парсит первые результаты (заголовок, ссылка, описание) и выводит их в консоль. В будущем их тоже можно зачитывать.
-
Голосовой ответ: Текст ответа передается в модель Silero, которая генерирует аудиопоток и проигрывает его через системные динамики.
-
Технологический стек: сильные стороны и скрытые сложности
-
Silero TTS: Главная «фишка». Локальный синтез без задержек на сеть. Однако модель требует значительных вычислительных ресурсов при первом запуске (загрузка весов) и занимает место на диске.
-
Scikit-learn для NLP: Использование
TfidfVectorizerс n-граммами символов и логистической регрессии — классический и эффективный подход для простых классификаторов с небольшим набором данных. Все обучение происходит в момент запуска на основеdata.json. -
Selenium для парсинга: Позволяет работать с динамически загружаемым контентом (как страницы Google). Это мощно, но хрупко: любое изменение верстки Google сломает парсер. Кроме того, это тяжеловесное решение, требующее установки Chrome и драйвера.
-
Распознавание речи от Google: В данном коде используется онлайн-распознавание, что является узким местом с точки зрения приватности и требует подключения к интернету. Альтернатива — локальные модели, например, от Vosk.
main.py
import time
import speech_recognition as sr
import threading
from datetime import datetime
from siler_audio import Silero_
from num2words import num2words
from query_request import Query
from pyowm import OWM
from dotenv import load_dotenv
import os
from search_google import Search_google
audio_silero = Silero_()
qr = Query()
load_dotenv()
class Voice:
def __init__(self):
self.recognizer = sr.Recognizer()
self.microphone = sr.Microphone()
self.is_listening = False
self.listening_thread = None
self.calibrate_microphone()
self.google = Search_google()
def speak(self, text):
print(f"[speak] {text}")
audio_silero.silero_tts_basic(text)
def stop(self):
print("[stop] Остановка...")
self.is_listening = False
def calibrate_microphone(self):
print("Калибровка микрофона...")
try:
with self.microphone as source:
self.recognizer.adjust_for_ambient_noise(source, duration=2)
print("Калибровка завершена!")
return True
except Exception as e:
print(f"Ошибка калибровки микрофона: {e}")
return False
def listen(self):
try:
with self.microphone as source:
print("Слушаю...")
# Увеличиваем timeout и phrase_time_limit для лучшего распознавания
audio = self.recognizer.listen(source, timeout=10, phrase_time_limit=8)
print("Распознаю речь...")
text = self.recognizer.recognize_google(audio, language='ru-RU')
print(f"Распознано: {text}")
return text.lower()
except sr.WaitTimeoutError:
return ""
except sr.UnknownValueError:
print("Речь не распознана")
return ""
except Exception as e:
print(f"Ошибка слушания: {e}")
return ""
def process_command(self, command):
if not command:
return
if not command.startswith('шустрик'):
return
print(f"Команда: {command}")
command = command.replace('шустрик', '')
if qr.get_intent(command) == 'greeting':
self.speak("Привет! Рад вас слышать!")
elif qr.get_intent(command) == "search":
ss = self.google.search(command)
sss = []
for i, res in enumerate(ss, 1):
if res.get('description') == '':
continue
sss.append({
'title': res.get('title', 'Без заголовка'),
'link': res.get('url', ''),
'description': res.get('description', '')
})
print(sss)
elif qr.get_intent(command) == 'time':
h = datetime.now().strftime("%H")
m = datetime.now().strftime("%M")
a_h = num2words(h, lang='ru')
a_m = num2words(m, lang='ru')
self.speak(f"Сейчас {a_h} {a_m}")
elif qr.get_intent(command) == 'weather':
own = OWM(os.getenv('API_KEY_WEATHER'))
mgr = own.weather_manager()
obs = mgr.weather_at_place('Москва,RU')
weather = obs.weather
res = f'Температура: {num2words(round(weather.temperature('celsius')['temp']), lang='ru')} градусов. Влажность: {num2words(round(weather.humidity), lang='ru')}%'
self.speak(f"Сейчас {res}")
elif qr.get_intent(command) == 'farewell':
self.speak("До свидания! Выключаюсь.")
self.is_listening = False
else:
self.speak("Пока не понимаю эту команду.")
def listening_loop(self):
print("Цикл прослушивания запущен")
self.speak("Ассистент запущен. Говорите команды")
while self.is_listening:
command = self.listen()
if command and command.strip():
self.process_command(command)
time.sleep(0.5)
def start_listening(self):
if self.is_listening:
print("Уже слушаем...")
return
self.is_listening = True
self.listening_thread = threading.Thread(target=self.listening_loop)
self.listening_thread.daemon = True
self.listening_thread.start()
print("Прослушивание запущено")
if __name__ == "__main__":
v = Voice()
try:
v.start_listening()
print("Ассистент активен. Нажмите Ctrl+C для остановки.")
while v.is_listening:
time.sleep(1)
except KeyboardInterrupt:
print("\nОстановка пользователем")
finally:
v.stop()
print("Ассистент завершил работу")
siler_audio.py
import torch
import sounddevice as sd
class Silero_:
def __init__(self):
self.device = torch.device('cpu')
self.model, self.text = torch.hub.load(
repo_or_dir='snakers4/silero-models',
model='silero_tts',
language='ru',
speaker='v3_1_ru'
)
self.model.to(self.device)
self.sample_rate = 48000
def silero_tts_basic(self, text, speaker='aidar'):
audio = self.model.apply_tts(
text=text,
speaker=speaker,
sample_rate=self.sample_rate,
put_accent=True,
put_yo=True
)
sd.play(audio, self.sample_rate)
sd.wait()
search_google.py
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.keys import Keys
from selenium.common.exceptions import TimeoutException, StaleElementReferenceException
import time
class Search_google:
def __init__(self):
self.driver = None
def setup_driver(self):
chrome_options = Options()
chrome_options.add_argument("--no-sandbox")
chrome_options.add_argument("--disable-dev-shm-usage")
chrome_options.add_argument("--disable-blink-features=AutomationControlled")
chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
chrome_options.add_experimental_option('useAutomationExtension', False)
chrome_options.add_argument(
'user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36')
chrome_options.page_load_strategy = 'eager'
self.driver = webdriver.Chrome(options=chrome_options)
return self.driver
def search(self, query, max_results=10, timeout=5):
try:
if not self.driver:
self.setup_driver()
wait = WebDriverWait(self.driver, timeout)
self.driver.get('https://www.google.com')
search_box = wait.until(EC.presence_of_element_located((By.NAME, "q")))
search_box.clear()
search_box.send_keys(query)
search_box.send_keys(Keys.RETURN)
wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, 'div.g, div.MjjYud')))
results = []
last_count = 0
start_time = time.time()
while len(results) < max_results and (time.time() - start_time) < timeout:
try:
result_elements = self.driver.find_elements(By.CSS_SELECTOR, 'div.g, div.MjjYud')
if len(result_elements) > last_count:
for i in range(last_count, min(len(result_elements), max_results)):
try:
result = self.extract_result_info(result_elements[i])
if result and result not in results:
results.append(result)
except StaleElementReferenceException:
continue
last_count = len(result_elements)
if len(results) >= 3:
break
time.sleep(0.3)
except Exception as e:
print(f"Ошибка при сборе результатов: {e}")
break
if len(results) < max_results:
self.scroll_for_more_results(max_results - len(results))
result_elements = self.driver.find_elements(By.CSS_SELECTOR, 'div.g, div.MjjYud')
for i in range(len(results), min(len(result_elements), max_results)):
try:
result = self.extract_result_info(result_elements[i])
if result and result not in results:
results.append(result)
except (StaleElementReferenceException, IndexError):
continue
return results[:max_results]
except Exception as e:
print(f"Ошибка при поиске: {e}")
return []
def extract_result_info(self, result_element):
try:
result_data = {}
try:
title_elem = result_element.find_element(By.CSS_SELECTOR, 'h3, .DKV0Md, .LC20lb')
result_data['title'] = title_elem.text
except:
result_data['title'] = ''
try:
link_elem = result_element.find_element(By.CSS_SELECTOR, 'a')
result_data['url'] = link_elem.get_attribute('href')
except:
result_data['url'] = ''
try:
desc_elem = result_element.find_element(By.CSS_SELECTOR, '.VwiC3b, .IsZvec, .MUxGbd')
result_data['description'] = desc_elem.text[:200]
except:
result_data['description'] = ''
return result_data
except Exception as e:
print(f"Ошибка при извлечении данных: {e}")
return None
def scroll_for_more_results(self, count_needed):
try:
for _ in range(count_needed // 5 + 1):
self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
time.sleep(0.5)
try:
more_results = self.driver.find_elements(By.CSS_SELECTOR,
'.RVQdVd, .mye4qd, a[aria-label^="More results"]')
if more_results:
more_results[0].click()
time.sleep(1)
except:
pass
except Exception as e:
print(f"Ошибка при прокрутке: {e}")
def close(self):
if self.driver:
self.driver.quit()
self.driver = None
def __del__(self):
self.close()
query_request.py
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.svm import LinearSVC
import json
class Query:
config = None
def __init__(self):
with open('data.json', 'r', encoding='utf-8') as f:
Query.config = json.load(f)
self.vectorizer = TfidfVectorizer(analyzer="char", ngram_range=(2, 3))
self.classifier_probability = LogisticRegression()
self.classifier = LinearSVC()
self.prepare_corpus()
def get_intent(self, request):
best_intent = self.classifier.predict(self.vectorizer.transform([request]))[0]
index_of_best_intent = list(self.classifier_probability.classes_).index(best_intent)
probabilities = self.classifier_probability.predict_proba(self.vectorizer.transform([request]))[0]
best_intent_probability = probabilities[index_of_best_intent]
if best_intent_probability > 0.157:
return best_intent
def prepare_corpus(self):
corpus = []
target_vector = []
for intent_name, intent_data in Query.config["intents"].items():
for example in intent_data["examples"]:
corpus.append(example)
target_vector.append(intent_name)
training_vector = self.vectorizer.fit_transform(corpus)
self.classifier_probability.fit(training_vector, target_vector)
self.classifier.fit(training_vector, target_vector)
data.json
{
"intents": {
"greeting": {
"examples": [
"привет", "здравствуй", "добрый день", "доброе утро", "добрый вечер",
"здравствуйте", "приветствую", "доброго времени суток", "приветик",
"хай", "здарова", "салют", "приветствую вас", "рад вас видеть",
"приветствую тебя", "здорово", "доброго здоровья", "моё почтение",
"ку", "йоу", "хеллоу", "шалом", "чао",
"привет, как дела", "здравствуйте, как поживаете", "добрый день, как ты",
"доброго дня", "разрешите поприветствовать", "имею честь приветствовать",
"приветствую вас, уважаемый", "здравствуйте, коллеги",
"доброе утро, друзья", "добрый день, господа", "добрый вечер, дамы и господа"
]
},
"farewell": {
"examples": [
"пока", "до свидания", "увидимся", "до встречи", "прощай",
"всего доброго", "всего хорошего", "до скорого", "чао", "покеда",
"бывай", "счастливо", "до завтра", "до следующего раза",
"всего наилучшего", "до свидания, всего хорошего", "разрешите попрощаться",
"с уважением", "приятного дня", "хорошего дня", "удачи",
"до новых встреч", "до скорой встречи",
"пока, удачи", "до встречи, будь здоров", "прощай, счастливо",
"до свидания, хорошего вечера", "пока, хорошего дня",
"гыг", "давай", "бай", "чмоки", "адью", "ариведерчи",
"гуд бай", "покедова", "всего", "досвидос"
]
},
"time": {
"examples": [
"время", "дата", "который час", "сколько времени", "текущее время",
"точное время", "время сейчас", "дата сегодня", "какое сегодня число",
"какой сегодня день", "текущая дата", "время и дата",
"часы", "тайм", "времени", "дату", "день", "месяц", "год",
"число", "день недели", "сейчас время", "местное время",
"скажи время", "подскажи, который час", "можно узнать время",
"интересует текущее время", "хочу знать время", "нужно знать время",
"какое сейчас время", "сколько время", "сколько сейчас времени",
"время московское", "время по Москве", "время в моём городе",
"какой год", "какой месяц", "какое число", "какой день недели",
"сегодняшняя дата", "текущий год", "текущий месяц"
]
},
"weather": {
"examples": [
"погода", "погода сейчас", "прогноз погоды", "какая погода",
"погода сегодня", "погода за окном", "температура", "погодные условия",
"погода в москве", "погода в городе", "погода в моём городе",
"погода на улице", "погода за окном", "погода сегодня в москве",
"погода в спб", "погода в питере", "погода в лондоне",
"температура сейчас", "сколько градусов", "градусы на улице",
"осадки", "будет ли дождь", "будет ли снег", "солнечно ли сегодня",
"ветрено ли", "влажность", "давление", "погода на завтра",
"прогноз на неделю", "погода на выходные",
"скажи погоду", "как погода", "что с погодой", "какая сейчас погода",
"хочу узнать погоду", "нужен прогноз погоды", "сообщи о погоде",
"интересует погода", "подскажи погоду", "можно узнать погоду",
"погода утром", "погода днём", "погода вечером", "погода ночью",
"температура днём", "температура ночью", "ощущается как",
"по ощущениям", "погода по ощущениям"
]
},
"search": {
"examples": [
"поиск", "найди", "найти", "ищи", "погугли", "загугли", "поищи в сети", "посмотри в интернете"
]
}
}
}
requirements.txt
antlr4-python3-runtime==4.9.3
certifi==2025.11.12
cffi==2.0.0
charset-normalizer==3.4.4
colorama==0.4.6
comtypes==1.4.13
docopt==0.6.2
filelock==3.20.0
fsspec==2025.10.0
geojson==3.2.0
idna==3.11
Jinja2==3.1.6
joblib==1.5.2
MarkupSafe==3.0.3
mpmath==1.3.0
networkx==3.6
num2words==0.5.14
numpy==2.3.5
omegaconf==2.3.0
pocketsphinx==5.0.4
PyAudio==0.2.14
pycparser==2.23
pyowm==3.5.0
pypiwin32==223
PySocks==1.7.1
python-dotenv==1.2.1
pyttsx3==2.99
pywin32==311
PyYAML==6.0.3
requests==2.32.5
scikit-learn==1.7.2
scipy==1.16.3
setuptools==80.9.0
silero==0.5.2
sounddevice==0.5.3
SpeechRecognition==3.14.4
srt==3.5.3
sympy==1.14.0
threadpoolctl==3.6.0
torch==2.9.1
torchaudio==2.9.1
tqdm==4.67.1
typing_extensions==4.15.0
urllib3==2.5.0
vosk==0.3.45
websockets==15.0.1
Ссылка на github yemorkovin/asistant
Автор: Евгений Морковин
0 комментариев
Оставьте комментарий
Комментарии