Испанский в кармане: Архитектура Telegram-бота с локальным Whisper.cpp, AI-диалогами и оценкой произношения

Меня зовут Vlad, я начинающий Python-разработчик и энтузиаст изучения языков.

Недавно я столкнулся с классической проблемой полиглота-самоучки: учебники дают теорию, аудиокурсы — пассивное восприятие ^[1], но нет главного — обратной связи по произношению. Репетиторы дороги, а разговорные клубы требуют уровня, которого у меня еще не было.

Я решил закрыть эту боль ^[2] кодом. Моя цель была амбициозной: создать Telegram-бота, который:

1. Слушает голосовые сообщения и распознает речь без дорогих облачных API.

2. Оценивает точность произношения в процентах, сравнивая с эталоном.

3. Поддерживает живой диалог через LLM, исправляя ошибки ^[3] на лету.

4. Работает быстро и экономно на слабом VPS.

В этой статье я подробно разберу архитектуру проекта, покажу, как интегрировать бинарный whisper.cpp в асинхронный aiogram 3.x, реализую алгоритм оценки речи и расскажу про управление состояниями (FSM). Под капотом — Python, нейросети и немного магии.

Выбор стека и архитектура

Главным требованием была экономия ресурсов и приватность данных. Отправлять голос пользователей в платные API (Google Speech, Azure) при масштабировании стало бы дорого, а локальные модели на Python (типа speech_recognition с обертками) часто медленные.

Решение:

Использовать скомпилированный C++ инструмент whisper.cpp как внешний процесс, вызываемый из Python. Это дает скорость даже на CPU и полную изоляцию.

Стек технологий: Язык: Python 3.10+ Фреймворк: aiogram 3.x (асинхронность обязательна). STT (Speech-to-Text): whisper.cpp (модель ggml-tiny.bin для скорости). TTS (Text-to-Speech): gTTS + pydub для конвертации. LLM: Внешний API (для диалогов). Хранение: JSON-файлы (для простоты старта) + FSM для сессий.

Интеграция Whisper.cpp: Борьба с блокировкой Event Loop

Первая техническая проблема возникла сразу. aiogram работает асинхронно, а запуск внешнего бинарного файла через стандартный subprocess.run ^[4]() является блокирующей операцией. Если один пользователь отправит голосовое, весь бот «зависнет» на время транскрибации (2-5 секунд) для всех остальных.

Решение: asyncio.to ^[5]_thread и create_subprocess_exec

Чтобы не блокировать цикл событий, тяжелые операции (конвертация аудио и вызов Whisper) были вынесены в отдельные потоки или обернуты в асинхронные процессы.

Вот реализация функции распознавания с правильной асинхронностью:

python

import asyncio
from pydub import AudioSegment
import subprocess
import os

async def recognize_voice_async(file_path: str) -> str:
    wav_path = "temp_voice.wav"
    txt_path = wav_path + ".txt"
    
    try:
        # 1. Конвертация в формат, понятный Whisper (16kHz, mono, wav)
        # Выносим тяжелую операцию pydub в поток, чтобы не блокировать loop
        audio = AudioSegment.from_file(file_path)
        await asyncio.to_thread(
            lambda: audio.set_frame_rate(16000)
                     .set_channels(1)
                     .set_sample_width(2)
                     .export(wav_path, format="wav")
        )

        if not os.path.exists(wav_path):
            raise FileNotFoundError("WAV file not created")

        # 2. Асинхронный запуск whisper.cpp
        process = await asyncio.create_subprocess_exec(
            "/root/whisper.cpp/build/bin/whisper-cli", # Путь к бинарнику
            "-m", "/root/whisper.cpp/models/ggml-tiny.bin", # Модель tiny
            "-f", wav_path,
            "--language", "es",
            "--output-txt",
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        
        stdout, stderr = await process.communicate()

        if process.returncode != 0:
            print(f"[ERROR] Whisper failed: {stderr.decode()}")
            return ""

        # 3. Чтение результата
        if os.path.exists(txt_path):
            with open(txt_path, "r", encoding="utf-8") as f:
                text = f.read().strip()
            return text
        return ""

    except Exception as e:
        print(f"[CRITICAL ERROR] {e}")
        return ""
    finally:
        # Очистка временных файлов
        for f in [wav_path, txt_path, file_path]:
            if os.path.exists(f):
                os.remove(f)

Такой подход позволяет боту обрабатывать сотни запросов параллельно, пока идет транскрибация для отдельных пользователей.

3. Алгоритм оценки произношения: Не просто сравнение строк

Получив текст от Whisper, нужно сравнить его с эталонной фразой из урока. Простое сравнение if text == expected не подходит: Whisper может ошибиться в артиклях, падежах или проглотить окончания, даже если пользователь сказал правильно.

Я реализовал алгоритм пословного сравнения с позиционным weighting.

def calculate_accuracy(expected: str, recognized: str) -> int:
    if not recognized:
        return 0
    
    expected_words = expected.lower().split()
    recognized_words = recognized.lower().split()
    
    if not expected_words:
        return 0

    # Сравниваем слова по позициям
    min_len = min(len(expected_words), len(recognized_words))
    matches = sum(1 for i in range(min_len) if expected_words[i] == recognized_words[i])
    
    # Формула: (совпадения / длина эталона) * 100
    # Мы делим на длину эталона, чтобы штрафовать за пропущенные слова
    accuracy = int((matches / len(expected_words)) * 100)
    
    # Ограничиваем диапазон от 0 до 100
    return max(0, min(100, accuracy))

Логика ^[6] работы:

Если пользователь сказал всё верно → 100%.
Если пропустил последнее слово → ~80-90% (зависит от длины фразы).
Если сказал совсем другое → 0-20%.
Порог успешности установлен на 70%. Если результат выше, бот хвалит пользователя (¡Muy bien!), иначе мягко просит повторить. В будущих версиях планирую внедрить расстояние Левенштейна для учета опечаток распознавания внутри слов.

4. Управление состояниями (FSM) и роутинг логики

Бот должен понимать контекст: пользователь хочет просто поболтать с ИИ или отрабатывает конкретную фразу из урока? Для этого используется машина состояний (aiogram.fsm).

Структура состояний

from aiogram.fsm.state import State, StatesGroup

class UserStates(StatesGroup):
    waiting_for_ai_dialog = State()  # Режим свободного диалога
    # В будущем можно добавить: waiting_for_translation, etc.

Универсальный хендлер голосовых сообщений

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

@dp.message(F.voice)
async def universal_voice_handler(message: Message, state: FSMContext):
    current_state = await state.get_state()
    
    # ВЕТКА 1: Диалог с ИИ
    if current_state == "UserStates:waiting_for_ai_dialog":
        from handlers.voice_ai import handle_ai_dialog_voice
        await handle_ai_dialog_voice(message, bot, state)
        return

    # ВЕТКА 2: Тренировка произношения (Уроки)
    user_id = str(message.from_user.id)
    # Здесь должна быть логика получения текущего урока из БД/JSON
    # user_data = get_user_progress(user_id) 
    # expected_phrase = lessons[user_data['lesson']][user_data['phrase']]
    
    # Скачиваем файл и распознаем
    file = await bot.get_file(message.voice.file_id)
    local_path = await bot.download_file(file.file_path, "user_voice.ogg")
    
    text = await recognize_voice_async(str(local_path))
    
    if not text.strip():
        await message.answer("❌ Не удалось распознать речь. Попробуйте громче!")
        return

    # Считаем точность (пример)
    # accuracy = calculate_accuracy(expected_phrase, text)
    # await message.answer(f"Вы сказали: {text}nТочность: {accuracy}%")
    
    # Логика ответа пользователю...

Такая архитектура позволяет легко масштабировать бота: добавляя новые состояния, мы можем внедрять новые режимы обучения ^[7] (например, «Диктант» или «Перевод»), не переписывая базовые хендлеры.

Для хранения прогресса пользователей и списка фраз я использовал JSON-файлы. Это простое решение для старта, которое легко мигрировать на SQLite или PostgreSQL в будущем.

• users.json: ID, имя, дата регистрации, флаг Premium, настройки уведомлений.

• lessons.json: Структура уроков (список фраз).

• phrases.json: База фраз дня для ежедневной рассылки.

Админ-панель реализована через команды (/addphrase, /stats) с проверкой ADMIN_ID. Это позволяет наполнять контент «на лету» без перезагрузки бота.

Заключение и планы

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

Что проект дал мне как разработчику:

• Опыт ^[8] работы с асинхронностью и внешними процессами в Python.

• Понимание работы STT/LLM интеграций.

• Навык проектирования FSM для сложных диалоговых сценариев.

Планы развития:

• Внедрение расстояния Левенштейна для более мягкой оценки.

• Генерация иллюстраций к словам через Stable Diffusion (визуальное запоминание ^[9]).

• Переезд на PostgreSQL и Docker-контейнеризацию

Код проекта открыт для обсуждения.

Буду рад вашим комментариям, критике архитектуры и предложениям по улучшению алгоритмов!

Попробовать бота: @Spanish1_Vladd_bot ^[10]
Спасибо за внимание ^[11]! ¡Hasta luego!