Пишем персонального AI-ассистента на Python

Современные голосовые помощники это мощные приложения, сочетающие обработку речи, машинное обучение и интеграцию с внешними API. В этой статье мы разберём, как создать базовый проект персонального ассистента на Python, используя библиотеки whisper, webrtcvad, gTTS и другие. Наш ассистент будет:

• Слушать микрофон

• Определять начало и конец речи с помощью VAD (Voice Activity Detection)

• Преобразовывать речь в текст через модель Whisper

• Отправлять запросы на локальный LLM для генерации ответа

• Читать ответ вслух с помощью gTTS

• Начинать/останавливать запись по клавише пробел

Проект может служить как началом для экспериментов, так и для прототипирования реальных решений.

🔧 Установка зависимостей

Перед запуском убедитесь, что установлены все необходимые библиотеки:

pip install numpy sounddevice keyboard whisper torch webrtcvad requests colorama gTTS

Также потребуется локальный сервер с LLM, например LM Studio, слушающий по адресу http://localhost:1234.

🎤 Обработка звука и запись голоса

Для работы с аудио используется библиотека sounddevice. Мы создаём поток записи с частотой 16 кГц и ожидаем нажатие пробела — это наш триггер для начала/остановки записи.

def record_audio():     global recording     print("Нажмите Пробел для начала записи...")     with sd.InputStream(samplerate=SAMPLE_RATE, channels=CHANNELS, dtype=DTYPE, callback=callback):         while True:             if keyboard.is_pressed('space'):                 toggle_recording()                 while keyboard.is_pressed('space'):                     pass             time.sleep(0.1)

Каждый фрагмент аудио добавляется в буфер, который затем анализируется с помощью VAD (webrtcvad) для определения наличия речи.

🗣️ Распознавание речи с помощью Whisper

Whisper — одна из популярных моделей распознавания речи. Мы используем её через библиотеку whisper, загружая модель medium и используя графический ускоритель (GPU), если он доступен.

model = whisper.load_model("medium").to(device)

После окончания речи (определяется по паузам) фрагмент передаётся в модель:

result = model.transcribe(audio_float, language="ru", verbose=None) text = result["text"].strip()

💬 Генерация ответа от ИИ

Для генерации ответа используем, например, локально установленную модель google/gemma-3-4B через приложение LM Studio , которое позволяет запускать LLM-модели локально на нашей машине и создавать совместимый с OpenAI API сервер.

После загрузки модели google/gemma-3-4b в LM Studio, вы запускаем её в режиме сервера. HTTP-сервер принимает JSON-запросы по адресу http://localhost:1234/v1/chat/completions. Таким образом, наш Python-скрипт отправляет туда текстовый запрос, и получает готовый ответ от модели:

def generate_response(text):     data = {         "messages": [{"role": "user", "content": text}],     }     response = requests.post("http://localhost:1234/v1/chat/completions", json=data)     return response.json()['choices'][0]['message']['content']

Этот подход позволяет работать с мощной ИИ-моделью без выхода в интернет, сохраняя конфиденциальность данных и обеспечивая приемлемую скорость работы (зависит от мощности процессора и графической карты). Убедитесь, что в LM Studio вы выбрали корректную модель и нажали кнопку Run locally или Start server , чтобы скрипт мог с ней взаимодействовать.

🎧 Преобразование текста в речь (TTS)

Для озвучивания ответа используется библиотека gTTS (Google Text-to-Speech). Она проста в использовании и отлично подходит для начального уровня, (модуль gTTS_module.py):

import io, os, contextlib from gtts import gTTS import pygame from threading import Thread import keyboard  # Для отслеживания клавиш  # Глобальная переменная для остановки воспроизведения _playing = False  def text_to_speech_withEsc(text: str, lang: str = 'ru'):     """     Преобразует текст в речь и воспроизводит его.     Остановка возможна нажатием клавиши Esc.     """     try:         # Генерация аудио в память         tts = gTTS(text=text, lang=lang)         fp = io.BytesIO()         tts.write_to_fp(fp)         fp.seek(0)          # Инициализация Pygame и загрузка аудио из памяти         pygame.mixer.init()         pygame.mixer.music.load(fp)         pygame.mixer.music.play()          # Воспроизводим, пока не закончится или не нажмут Esc         while pygame.mixer.music.get_busy():             if keyboard.is_pressed('esc'):                 pygame.mixer.music.stop()                 print("Воспроизведение остановлено (Esc)")                 break          pygame.mixer.quit()      except Exception as e:         print(f"Ошибка при озвучивании: {e}")     finally:         pass

🖌️ Цветовая схема и интерфейс

Чтобы вывод был удобнее читать, мы применяем цвета с помощью colorama. Вы можете выбрать между светлой и тёмной темой оформления:

THEMES = {     "light": {         "user": Fore.BLUE,         "assistant": Fore.LIGHTBLACK_EX,         ...     },     "dark": {         "user": Fore.CYAN,         "assistant": Fore.LIGHTGREEN_EX,         ...     } }

Также добавлена анимация «думающего» ассистента во время генерации ответа:

loading_animation(duration=1, text="Генерация ответа...")

🚀 Запуск и работа

Запустите LLM сервер и запустите основной скрипт, нажмите Пробел — и задайте вопрос. Ассистент его распознает, отправит в модель, получит ответ и прочитает его вам вслух (файл pers_assist.py).

import numpy as np import sounddevice as sd import keyboard import whisper import threading import time import torch import webrtcvad import requests from colorama import Fore, Style, init import time import sounddevice as sd import re import gTTS_module2   # Инициализация colorama init(autoreset=True)  # === Цветовые схемы === THEMES = {     "light": {         "user": Fore.BLUE,         "assistant": Fore.LIGHTBLACK_EX,         "thinking": Fore.MAGENTA,         "background": Style.BRIGHT,         "prompt": "Светлая"     },     "dark": {         "user": Fore.CYAN,         "assistant": Fore.LIGHTGREEN_EX,         "thinking": Fore.YELLOW,         "background": Style.DIM,         "prompt": "Тёмная"     } }  THEME = THEMES["light"] #print(f"\n✅ Установлена {THEME['prompt']} тема\n")    # --- Настройки --- SAMPLE_RATE = 16000 CHANNELS = 1 DTYPE = np.int16  SEGMENT_DURATION = 0.02  # 20 мс для VAD SEGMENT_SAMPLES = int(SAMPLE_RATE * SEGMENT_DURATION)  MIN_SPEECH_CHUNKS = 10     # минимум фрагментов с голосом подряд SILENCE_TIMEOUT = 1.5      # секунд ожидания перед новой строкой    #['tiny.en', 'tiny', 'base.en', 'base', 'small.en', 'small', 'medium.en', 'medium', 'large-v1', 'large-v2', 'large-v3', 'large', 'large-v3-turbo', 'turbo']   # --- Инициализация модели Whisper с поддержкой CUDA --- device = "cuda" if torch.cuda.is_available() else "cpu" #print(f"[Используется устройство]: {device.upper()}") model = whisper.load_model("medium").to(device)  #можно так: model = whisper.load_model("small", device="cpu")   # --- Инициализация VAD --- vad = webrtcvad.Vad() vad.set_mode(3 )  # чувствительность 0 - высокая, 3 - низкая    def is_speech(frame_bytes):     try:         return vad.is_speech(frame_bytes, SAMPLE_RATE)     except:          return False    # --- Глобальные переменные --- recording = False audio_buffer = [] buffer_index = 0  lock = threading.Lock() last_speech_time = None  # --- Callback записи --- def callback(indata, frames, time, status):     if recording:         with lock:             audio_buffer.extend(indata.copy().flatten())  # --- Управление записью --- def record_audio():     global recording     print("Нажмите Пробел для начала записи...")     with sd.InputStream(samplerate=SAMPLE_RATE, channels=CHANNELS, dtype=DTYPE, callback=callback):         while True:             if keyboard.is_pressed('space'):                 toggle_recording()                 while keyboard.is_pressed('space'):                     pass             time.sleep(0.1)  def toggle_recording():     global recording, audio_buffer, buffer_index     global speech_segment, speech_started, new_line_pending, current_pause, last_speech_time      recording = not recording     if recording:         print("\n[Запись началась...]")         audio_buffer.clear()         buffer_index = 0          # Сброс состояния VAD         speech_segment = []            speech_started = False             new_line_pending = False         current_pause = 0.0         last_speech_time = time.time()  # ← обновляем время начала      else:         print("[Запись остановлена.]")    def generate_response(text):     data = {      "messages": [         {"role": "user", "content": text}            ],     #"temperature": 0.0,        # минимальная случайность     #"max_tokens": 10,          # минимум токенов для ответа     #"stream": False,           # отключает потоковую передачу     #"stop": ["\n"]             # остановка после первой строки      }      response = requests.post(         "http://localhost:1234/v1/chat/completions",         json=data     )     assist_reply = response.json()['choices'][0]['message']['content']     # Удаляем теги вместе с содержимым между ними     #cleaned_text = re.sub(r'\<think\>.*?<\</think\>', '', assist_reply, flags=re.DOTALL)     #print("Ответ ассистента:", assist_reply)      return assist_reply   # === Анимация загрузки === def loading_animation(duration=1 , text="Думаю"):     symbols = [ '⣷', '⣯', '⣟', '⡿', '⢿', '⣻', '⣽','⣾']     end_time = time.time() + duration     idx = 0     while time.time() < end_time:         print(f"\r{THEME['thinking']}[{symbols[idx % len(symbols)]}] {text}{Style.RESET_ALL}", end="")         idx += 1         time.sleep(0.1)     print(" " * (len(text) + 6), end="\r")  # Очистка строки    def process_stream():     global last_speech_time, buffer_index     global speech_segment, speech_started, new_line_pending, current_pause     global recording         while True:         if not recording:             time.sleep(0.5)             continue         question_text = ""                 with lock:             available = len(audio_buffer)          while buffer_index + SEGMENT_SAMPLES <= available:             segment = audio_buffer[buffer_index:buffer_index + SEGMENT_SAMPLES]             buffer_index += SEGMENT_SAMPLES              segment_np = np.array(segment, dtype=np.int16)             frame_bytes = segment_np.tobytes()              try:                 is_silence = not is_speech(frame_bytes)                  if not is_silence:                     speech_segment.extend(segment)                     speech_started = True                     new_line_pending = False                     last_speech_time = time.time()  # ← обновляем время речи                 elif speech_started:                     current_pause = time.time() - last_speech_time                      if current_pause > SILENCE_TIMEOUT:                         if speech_segment:                             # Распознаём и выводим                             audio_float = np.array(speech_segment, dtype=np.float32) / 32768.0                             result = model.transcribe(audio_float, language="ru", verbose=None)                              text = result["text"].strip()                             if text.startswith("Редактор субтитров"): # баг whisper, реакция на шум                                 text = ""                                 continue                             question_text += " " + text                                 if text:                                 print(f"{THEME['user']}Вы: {Style.RESET_ALL}{text}" , end=" ", flush=True)                                                              speech_segment = []                          print()  # новая строка                         speech_segment = []                         speech_started = False                         new_line_pending = False                          # Генерация ответа                         loading_animation(text="Генерация ответа...")                         #print(f"\r{THEME['thinking']}[{symbols[idx % len(symbols)]}] {text}{Style.RESET_ALL}", end="")                         #print(f"{THINKING_COLOR}Генерация ответа...{RESET}", end="\r")                         response = generate_response(question_text)                          print(f"{THEME['assistant']}Ассистент: {response}{Style.RESET_ALL}")                         question_text = ""                          recording = False                         gTTS_module2.text_to_speech_withEsc(response)                         recording = True                    except Exception as e:                  print(f"[Ошибка]: {e}")         time.sleep(0.05)  # --- Точка входа --- if __name__ == "__main__":     print("[Voice-assisient приложение запущено.]")     threading.Thread(target=record_audio, daemon=True).start()     threading.Thread(target=process_stream, daemon=True).start()      try:         while True:             time.sleep(1)     except KeyboardInterrupt:         print("\nВыход.")

✅ Заключение

Полный код проекта доступен на github. Созданный нами голосовой ассистент — это пилотный проект, который можно развивать в сторону полноценного AI-ассистента для дома или офиса. Он объединяет несколько технологий: обработку звука, модели машинного обучения и работу с API. Проект может стать основой для тех кто интересуется темой создания персональных ассистентов.