Привет! Меня зовут Иван Володин, я разработчик DD Planet, и я задался целью сделать для себя максимально удобный скрипт для набора текста речью.

Содержание

• Введение

• Определимся с целями

• Выбор нейронки

• Первые шаги

• Первые проблемы

• Вторая итерация

• Проблемы второй итерации

• Третья итерация

• Заключение

Введение

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

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

Продукта который позволял бы, как в андроиде, делать это одной кнопкой (+ через локальную ллм) — нет.

Лучшим выходом из моей ситуации было создать свое минималистичное решение, и вот как это было:

Определимся с целями

Мы хотим использовать свой voice2text (real-time перевод аудио в текст) в самых разных приложениях, во всех, где можно вводить что угодно с клавиатуры. Поэтому ставим себе требование — распознанные слова должны сами печататься в активное текстовое поле любого вида.

Поток с микрофона мы будем отправлять в нейросеть, запущенную локально. Изначально был план использовать API от OpenAI, но self-host дает нам больше преимуществ, ведь так наш voice2text сможет работать без интернета, без проблем с конфиденциальностью и бесплатно.

Интерфейса у нас не будет, скрипт будет работать в headless режиме и запускаться автоматически при запуске ПК.

Для схожести с оригиналом включать/выключать микрофон будем по дефолту на самую верхнюю правую кнопку TKL клавиатуры — Pause. Благо, она редко используется в других приложениях.

Выбор нейронки

Основных предложений для локального запуска на обычном домашнем железе лидера два: Vosk и Whisper. Vosk более легковесный, запускается на CPU и из коробки поддерживает стриминг потока из микрофона. Whisper поддерживает куда больше языков и имеет заметно большее разнообразие моделей.

Для наших целей лучше подойдет Vosk, так как нас нас интересует быстрая потоковая обработка речи, а не постанализ аудио на любом языке. Немаловажный аргумент — при распознавании русской речи лучше всего себя показал Vosk.

Первые шаги

ТЗ определен, переходим к коду. Пример работы Vosk от нейросети:
Пример работы Vosk от нейросети:

from vosk import Model, KaldiRecognizer
import sounddevice as sd
import json

model = Model(r"C:\путь\к\vosk-model-small-ru-0.22")
rec = KaldiRecognizer(model, 16000)

def callback(indata, frames, time, status):
    if status:
        print(status)
    # Для RawInputStream преобразуем напрямую
    if rec.AcceptWaveform(bytes(indata)):
        result = json.loads(rec.Result())
        print("Распознано:", result.get("text", ""))
    else:
        partial = json.loads(rec.PartialResult())
        print("Промежуточно:", partial.get("partial", ""))

with sd.RawInputStream(samplerate=16000, blocksize=8000, dtype='int16',
                       channels=1, callback=callback):
    print("Начало записи, говорите...")
    while True:
        sd.sleep(1000)

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

Чтобы избежать этого эффекта при вводе текста, нужно учесть один момент:

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

Теперь можно переходить к реализации этой логики.

Для полной симуляции ввода с клавиатуры будем использовать WinAPI. Такой подход позволит нам отправлять события нажатия клавиш напрямую в операционную систему, что освобождает нас от проблем поиска активных текстовых полей вручную и отправки данных в них.

user32 = ctypes.WinDLL('user32', use_last_error=True)

INPUT_KEYBOARD   = 1
KEYEVENTF_KEYUP  = 0x0002
KEYEVENTF_UNICODE = 0x0004
VK_BACK = 0x08

class KEYBDINPUT(ctypes.Structure):
    _fields_ = (
        ("wVk",       ctypes.wintypes.WORD),
        ("wScan",     ctypes.wintypes.WORD),
        ("dwFlags",   ctypes.wintypes.DWORD),
        ("time",      ctypes.wintypes.DWORD),
        ("dwExtraInfo", ctypes.POINTER(ctypes.wintypes.LONG)),
    )

class INPUT(ctypes.Structure):
    _fields_ = (
        ("type", ctypes.wintypes.DWORD),
        ("ki",   KEYBDINPUT),
        ("pad",  ctypes.wintypes.BYTE * 8),
    )

def make_input_pair(wVk: int, wScan: int) -> list:
    """Создаёт массив из двух INPUT объектов: keydown + keyup."""
    keydown = INPUT(
        type=INPUT_KEYBOARD,
        ki=KEYBDINPUT(wVk=wVk, wScan=wScan, dwFlags=KEYEVENTF_UNICODE, time=0, dwExtraInfo=None)
    )
    keyup = INPUT(
        type=INPUT_KEYBOARD,
        ki=KEYBDINPUT(wVk=wVk, wScan=wScan, dwFlags=KEYEVENTF_UNICODE | KEYEVENTF_KEYUP, time=0, dwExtraInfo=None)
    )
    return [keydown, keyup]

def send_text(text: str):
    """Печатает unicode-текст как будто пользователь набирает его."""
    if not text:
        return
    arr = [inp for ch in text for inp in make_input_pair(0, ord(ch))]
    # Превращаем список в C-массив и отправляем
    user32.SendInput(len(arr), ctypes.byref((INPUT * len(arr))(*arr)), ctypes.sizeof(INPUT))

def send_backspaces(n: int):
    """Нажимает Backspace n раз (keydown+keyup)."""
    if n <= 0:
        return
    arr = [inp for inp in make_input_pair(VK_BACK, 0) * n]
    user32.SendInput(len(arr), ctypes.byref((INPUT * len(arr))(*arr)), ctypes.sizeof(INPUT))

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

printed_text = ""  # что мы уже вывели «клавиатурой»

def apply_text(new_text: str):
    """Сравнивает new_text с printed_text и вносит минимальные изменения на экране."""
    global printed_text
    if new_text == printed_text:
        return

    # Находим длину общего префикса
    a, b = printed_text, new_text
    i = 0
    max_i = min(len(a), len(b))
    # быстрый посимвольный поиск LCP
    while i < max_i and a[i] == b[i]:
        i += 1

    # Удаляем хвост старого текста
    to_delete = len(a) - i
    if to_delete > 0:
        send_backspaces(to_delete)

    # Допечатываем хвост нового текста
    to_type = b[i:]
    if to_type:
        send_text(to_type)

    printed_text = new_text

По аналогии с примером, загружаем Vosk и распознанный им текст передаём в apply_text

SMALL_MODEL_PATH = 'C:\\путь\\к\\vosk-model-small-ru-0.22'
current_model = Model(SMALL_MODEL_PATH)
recognizer = KaldiRecognizer(current_model, 16000)

def process_text(txt: str, reset_printed=False):
    """Извлекает текст из результата и применяет его к экрану."""
    global printed_text
    if txt:
        txt += " "
        apply_text(txt)
        if reset_printed:
            printed_text = ""

def callback(indata, frames, time, status):
    if status:
        print(status)
    if not is_listening.is_set():
        return
    if recognizer.AcceptWaveform(bytes(indata)):
        process_text(json.loads(recognizer.Result()).get("text", ""), reset_printed=True)
    else:
        process_text(json.loads(recognizer.PartialResult()).get("partial", ""))

def audio_raw_input_stream():
    try:
        with sd.RawInputStream(samplerate=16000, blocksize=8000, dtype='int16',
                            channels=1, callback=callback):
            while is_listening.is_set():
                sd.sleep(1)
    except:
        pass

Теперь, когда все готово, остается реализовать только главный цикл приложения. В нем мы будем следить за состоянием клавиши Pause:

• При нажатии — менять флаг is_listening,

• Если запись активна, останавливать поток audio_thread с функцией audio_raw_input_stream,

• Если запись выключена — наоборот, запускать audio_thread. 

Работа с audio_raw_input_stream в отдельном потоке необходима по двум причинам:

• Чтобы поток с микрофона не читал данные непрерывно, когда запись не ведется,

• И чтобы при отключении микрофона не «падал» главный цикл приложения.

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

# Коды клавиш для лучшей читаемости
VK_PAUSE = 0x13

is_listening = Event()
audio_thread = None

if __name__ == "__main__":
    try:
        last_pause_state = False
        
        while True:
            single_pause_state = user32.GetAsyncKeyState(VK_PAUSE) & 0x8000
            
            # Обработка одиночного Pause (используем VK_PAUSE)
            if single_pause_state and not last_pause_state:
                is_listening.set() if not is_listening.is_set() else is_listening.clear()
                if is_listening.is_set():
                    audio_thread = Thread(target=audio_raw_input_stream)
                    audio_thread.start()
                print(f"Режим распознавания: {'ВКЛ' if is_listening.is_set() else 'ВЫКЛ'}")
            
            # Сохраняем состояние для следующей итерации
            last_pause_state = single_pause_state
            
            time.sleep(0.05)

    finally:
        if audio_thread is not None:
            audio_thread.join()

Скрипт готов! Первую итерацию можно запускать и использовать. Чтобы запускать скрипт фоном, используем .pyw (или ярлык, в котором прямо укажем открытие через pythonw.exe). Чтобы скрипт запускался автоматически при старте ПК, переместим его в Автозагрузку: C:\Users\BathDuck\AppData\Roaming\Microsoft\Windows\Start Menu\Programs\Startup

Весь код первой итерации

Определить, активно ли сейчас распознавание речи, можно по иконке микрофона в трее — она отображает текущее состояние.

Первые проблемы

Первые проблемы, с которыми мы сталкиваемся, используя свое решение — vosk-small распознает русский текст недостаточно точно (20-25% ошибок, это слишком неудобно). Особенно, если речь не дикторская и микрофон не стоит близко к говорящему.

Решим проблему самым простым способом — возьмем модель потяжелее: vosk-model-ru-0.42. Работает она в разы точнее, но запускается несколько минут.

Реализация, при которой после запуска ПК необходимо будет ждать несколько минут, пока тяжелая модель запустится и сможет работать — не очень user-friendly, поэтому решим эту проблему следующим образом: сначала запустим small модель и распознавать текст будем ей, в этот же момент в отдельном потоке поставим грузиться тяжелую. Как только она загрузится, поменяем их местами. Так мы сможем и распознавать текст сразу с момента запуска ПК, и повысить точность распознавания настолько быстро, насколько это возможно.

Вторая итерация

Изменим инициализацию моделей под новую логику

# Пути к моделям
SMALL_MODEL_PATH = 'C:\\путь\\к\\vosk-model-small-ru-0.22'
LARGE_MODEL_PATH = 'C:\\путь\\к\\vosk-model-ru-0.42'

# Инициализация моделей
small_model = Model(SMALL_MODEL_PATH)
large_model = None

# Текущая модель и распознаватель
current_model = small_model
recognizer = KaldiRecognizer(current_model, 16000)

def load_large_model():
    global large_model, current_model, recognizer, only_small_mode
    print("Загрузка более совершенной модели...")
    large_model = Model(LARGE_MODEL_PATH)
    current_model = large_model
    recognizer = KaldiRecognizer(current_model, 16000)
    print("Более совершенная модель загружена!")

И в основном потоке запустим отдельный поток, загружающий тяжёлую модель.

# Коды клавиш для лучшей читаемости
VK_PAUSE = 0x13

is_listening = Event()
audio_thread = None
model_loader_thread = Thread(target=load_large_model, daemon=True)

if __name__ == "__main__":
    try:
        model_loader_thread.start()
        ...

Весь код второй итерации

Проблемы второй версии

После долгого использования второй версии на практике все больше начинает смущать тот факт, что наш скрипт после инициализации тяжелой модели занимает 5-6 ГБ ОЗУ. В повседневной работе это не большая проблема, но при запуске других требовательных к ОЗУ приложений, наш фоновый скрипт может им мешать.

Вместо того, чтобы каждый раз убивать наше приложение в Диспетчере задач, изменим наш скрипт так, чтобы при нажатии на другую клавишу (Ctrl+Pause) тяжелая модель выгружалась из памяти, а распознавала текст снова small версия.

Заодно, так как наш скрипт не будет работать корректно, если мы запустим несколько его инстансов, мы запретим запуск более одного экземпляра нашей программы с помощью WinAPI мьютексов.

Третья итерация

Добавим логику удаления тяжёлой модели из памяти.

def load_large_model():
    global large_model, current_model, recognizer, only_small_mode
    print("Загрузка более совершенной модели...")
    large_model = Model(LARGE_MODEL_PATH)
    if only_small_mode:
        free_large_model()
        return
    current_model = large_model
    recognizer = KaldiRecognizer(current_model, 16000)
    print("Более совершенная модель загружена и активирована!")

def free_large_model():
    global large_model
    if large_model is not None:
        try:
            large_model.free()
        except AttributeError:
            pass
        large_model = None
        gc.collect()

only_small_mode = False
def unload_large_model():
    global large_model, current_model, recognizer, audio_thread
    print("Удаление из памяти более совершенной модели...")
    current_model = small_model
    recognizer = KaldiRecognizer(current_model, 16000)
    free_large_model()

И используем её в главном потоке.

# Коды клавиш для лучшей читаемости
VK_PAUSE = 0x13
VK_CANCEL = 0x03  # Код для Ctrl+Pause

# Состояние программы
is_listening = Event()

# Запускаем потоки
audio_thread = None
model_loader_thread = Thread(target=load_large_model, daemon=True)

if __name__ == "__main__":
    try:
        # Проверка уникальности экземпляра
        mutex_name = "Global\\VoskSpeechRecognitionUniqueMutex"
        mutex = ctypes.windll.kernel32.CreateMutexW(None, False, mutex_name)
        last_error = ctypes.windll.kernel32.GetLastError()
        if last_error == 183:  # ERROR_ALREADY_EXISTS
            print("Программа уже запущена! Завершение.")
            ctypes.windll.kernel32.CloseHandle(mutex)
            exit(0)

        print("Нажмите Pause/Break или Scroll Lock для включения/выключения режима распознавания...")
        print("Нажмите Ctrl+Pause для выхода из программы.")
        
        model_loader_thread.start()
        
        last_pause_state = False
        last_ctrl_pause_state = False
        while True:
            # Для Ctrl+Pause используем VK_CANCEL вместо VK_PAUSE
            ctrl_pause_state = user32.GetAsyncKeyState(VK_CANCEL) & 0x8000
            
            # Для одиночной клавиши Pause используем VK_PAUSE
            single_pause_state = user32.GetAsyncKeyState(VK_PAUSE) & 0x8000
            
            # Обработка Ctrl+Pause (используем VK_CANCEL)
            if ctrl_pause_state and not last_ctrl_pause_state:
                print("Обнаружено нажатие Ctrl+Pause")
                is_listening.clear()
                only_small_mode = not only_small_mode
                if only_small_mode:
                    unload_large_model()
                else:
                    model_loader_thread = Thread(target=load_large_model, daemon=True)
                    model_loader_thread.start()
            
            # Обработка одиночного Pause (используем VK_PAUSE)
            if single_pause_state and not last_pause_state:
                is_listening.set() if not is_listening.is_set() else is_listening.clear()
                if is_listening.is_set():
                    audio_thread = Thread(target=audio_raw_input_stream)
                    audio_thread.start()
                print(f"Режим распознавания: {'ВКЛ' if is_listening.is_set() else 'ВЫКЛ'}")
            
            last_pause_state = single_pause_state
            last_ctrl_pause_state = ctrl_pause_state
            
            time.sleep(0.05)

    finally:
        if audio_thread is not None:
            audio_thread.join()
        if mutex:
            ctypes.windll.kernel32.CloseHandle(mutex)

Весь код третьей итерации

Заключение

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

Также мы добавили возможность управлять количеством памяти, которое использует нейронка с помощью переключения между разными распознавателями.

На будущее есть планы добавить пунктуацию (через vosk-recasepunc-ru-0.22), поддержку интеграции пользовательских слов в словарь Vosk и переключение на онлайн-режим распознавания.