Привет, Хабр! Меня зовут Vlad, я начинающий Python-разработчик и энтузиаст изучения языков.
Недавно я столкнулся с классической проблемой полиглота-самоучки: учебники дают теорию, аудиокурсы — пассивное восприятие, но нет главного — обратной связи по произношению. Репетиторы дороги, а разговорные клубы требуют уровня, которого у меня еще не было.
Я решил закрыть эту боль кодом. Моя цель была амбициозной: создать Telegram-бота, который:
Слушает голосовые сообщения и распознает речь без дорогих облачных API.
Оценивает точность произношения в процентах, сравнивая с эталоном.
Поддерживает живой диалог через LLM, исправляя ошибки на лету.
Работает быстро и экономно на слабом 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() является блокирующей операцией. Если один пользователь отправит голосовое, весь бот «зависнет» на время транскрибации (2-5 секунд) для всех остальных.
Решение: asyncio.to_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))
Логика работы:
Если пользователь сказал всё верно → 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}%") # Логика ответа пользователю...
Такая архитектура позволяет легко масштабировать бота: добавляя новые состояния, мы можем внедрять новые режимы обучения (например, «Диктант» или «Перевод»), не переписывая базовые хендлеры.
Для хранения прогресса пользователей и списка фраз я использовал JSON-файлы. Это простое решение для старта, которое легко мигрировать на SQLite или PostgreSQL в будущем.
users.json: ID, имя, дата регистрации, флаг Premium, настройки уведомлений.
lessons.json: Структура уроков (список фраз).
phrases.json: База фраз дня для ежедневной рассылки.
Админ-панель реализована через команды (/addphrase, /stats) с проверкой ADMIN_ID. Это позволяет наполнять контент «на лету» без перезагрузки бота.
Заключение и планы
В результате получился работающий инструмент, который уже помогает мне и первым тестировщикам ставить произношение.
Что проект дал мне как разработчику:
Опыт работы с асинхронностью и внешними процессами в Python.
Понимание работы STT/LLM интеграций.
Навык проектирования FSM для сложных диалоговых сценариев.
Планы развития:
Внедрение расстояния Левенштейна для более мягкой оценки.
Генерация иллюстраций к словам через Stable Diffusion (визуальное запоминание).
Переезд на PostgreSQL и Docker-контейнеризацию
Код проекта открыт для обсуждения.
Буду рад вашим комментариям, критике архитектуры и предложениям по улучшению алгоритмов!
Попробовать бота: https://t.me/Spanish1_Vladd_bot
Спасибо за внимание! ¡Hasta luego!
