Современные голосовые помощники это мощные приложения, сочетающие обработку речи, машинное обучение и интеграцию с внешними 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. Проект может стать основой для тех кто интересуется темой создания персональных ассистентов.