Голосовой бот в Telegram через Whisper

MAATRIX

Пользователь присылает боту голосовое сообщение вместо того, чтобы печатать — а бот должен понять, что там сказано, и ответить текстом. Задача звучит просто, но состоит из трёх независимых кусков: приём файла из Telegram, конвертация формата и распознавание речи. Ниже — рабочая связка Aiogram + ffmpeg + Whisper, с кодом, который можно вставить в свой проект, и честным разбором того, сколько это займёт по времени.

Архитектура: из чего состоит голосовой бот

Схема простая, но каждое звено имеет значение:

1. Пользователь записывает голосовое сообщение в Telegram — оно приходит боту в формате .ogg (кодек Opus).
2. Aiogram получает объект Voice, бот скачивает файл на диск сервера.
3. .ogg конвертируется в .wav через ffmpeg — Whisper с ним работает надёжнее, чем напрямую с Opus.
4. .wav передаётся в Whisper, который возвращает текст.
5. Текст отправляется пользователю обратно в чат.

Каждый шаг — это отдельная точка отказа: файл может не скачаться, ffmpeg — не найтись в PATH, Whisper — упасть на слишком длинной записи. Дальше разберём каждый шаг с рабочим кодом.

Сам Whisper в этой статье не устанавливаем — это отдельная тема, и в блоге есть пошаговая инструкция: Whisper локально на своём сервере. Здесь предполагается, что модель уже стоит и её можно вызвать функцией transcribe().

Ловим голосовое сообщение в Aiogram

В Aiogram 3.x голосовые сообщения ловятся через фильтр F.voice. У объекта message.voice есть file_id, duration (в секундах) и file_size — их стоит проверять до скачивания, чтобы не тащить на сервер получасовую запись:

from aiogram import Router, F
from aiogram.types import Message

router = Router()

MAX_DURATION = 120  # секунд, ограничение на длину голосового

@router.message(F.voice)
async def handle_voice(message: Message):
    voice = message.voice

    if voice.duration > MAX_DURATION:
        await message.answer(
            f"Сообщение длиннее {MAX_DURATION} секунд, пришлите короче."
        )
        return

    await message.bot.send_chat_action(message.chat.id, "typing")

    file = await message.bot.get_file(voice.file_id)
    ogg_path = f"/tmp/voices/{voice.file_id}.ogg"
    await message.bot.download_file(file.file_path, destination=ogg_path)

    # дальше — конвертация и распознавание

Директорию /tmp/voices/ нужно создать заранее (os.makedirs("/tmp/voices", exist_ok=True) при старте бота) и не забыть чистить — иначе за пару недель активного использования диск на VPS забьётся временными файлами. Проще всего удалять .ogg и .wav сразу после того, как текст ушёл пользователю.

send_chat_action с типом "typing" — не декорация, а часть UX: пока идёт конвертация и распознавание, пользователь видит «печатает…» и не думает, что бот завис.

Конвертация .ogg в .wav через ffmpeg

Telegram присылает голосовые в Opus-контейнере .ogg — Whisper умеет читать его напрямую через ffmpeg под капотом, но на практике конвертация в .wav с нужной частотой дискретизации даёт стабильнее результат и меньше сюрпризов с версиями библиотек. Whisper обучен на аудио с частотой 16 кГц, моно — этим и ограничиваем вывод ffmpeg:

import asyncio

async def convert_to_wav(ogg_path: str, wav_path: str) -> None:
    process = await asyncio.create_subprocess_exec(
        "ffmpeg", "-y",
        "-i", ogg_path,
        "-ar", "16000",
        "-ac", "1",
        "-f", "wav",
        wav_path,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    _, stderr = await process.communicate()

    if process.returncode != 0:
        raise RuntimeError(f"ffmpeg упал: {stderr.decode(errors='ignore')}")

Ключевые флаги:

Используем asyncio.create_subprocess_exec, а не subprocess.run — обычный синхронный вызов заблокирует event loop бота, и пока идёт конвертация, он не сможет обрабатывать сообщения от других пользователей. Для асинхронного бота это критично уже при паре одновременных запросов.

Если ffmpeg не установлен на сервере, create_subprocess_exec бросит FileNotFoundError — на Ubuntu/Debian ставится одной командой: apt install -y ffmpeg.

Передача записи в Whisper

Дальше — сам вызов распознавания. Если используется библиотека openai-whisper (или faster-whisper — по опыту она заметно экономнее по памяти на CPU-серверах), вызов оборачивается в отдельную функцию, которую тоже стоит выполнять не в основном потоке event loop:

import whisper

model = whisper.load_model("small")  # base/small/medium — компромисс скорость/точность

def transcribe_sync(wav_path: str) -> str:
    result = model.transcribe(wav_path, language="ru")
    return result["text"].strip()

async def transcribe(wav_path: str) -> str:
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, transcribe_sync, wav_path)

run_in_executor уводит тяжёлую CPU-задачу в отдельный поток, чтобы она не блокировала обработку остальных сообщений бота — то же соображение, что и с ffmpeg. Параметр language="ru" можно убрать, если бот многоязычный: Whisper сам определит язык, но это добавляет времени на распознавание и иногда ошибается на коротких фразах.

Выбор размера модели — компромисс, который каждый решает под свою нагрузку. base и small работают быстрее и требуют меньше памяти, medium точнее на нечёткой дикции и фоновом шуме, но заметно медленнее на CPU. Если VPS без GPU, для чат-бота с короткими голосовыми small обычно достаточно.

Собираем обработчик целиком

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

import os

@router.message(F.voice)
async def handle_voice(message: Message):
    voice = message.voice
    if voice.duration > MAX_DURATION:
        await message.answer(f"Сообщение длиннее {MAX_DURATION} секунд, пришлите короче.")
        return

    await message.bot.send_chat_action(message.chat.id, "typing")

    file_id = voice.file_id
    ogg_path = f"/tmp/voices/{file_id}.ogg"
    wav_path = f"/tmp/voices/{file_id}.wav"

    try:
        file = await message.bot.get_file(file_id)
        await message.bot.download_file(file.file_path, destination=ogg_path)

        await convert_to_wav(ogg_path, wav_path)
        text = await transcribe(wav_path)

        if not text:
            await message.answer("Не удалось разобрать речь — попробуйте ещё раз, ближе к микрофону.")
            return

        await message.answer(f"Распознано:\n{text}")

    except Exception as e:
        await message.answer("Не получилось обработать голосовое сообщение.")
        # логировать e в общий лог бота
    finally:
        for path in (ogg_path, wav_path):
            if os.path.exists(path):
                os.remove(path)

Блок finally тут не косметика — без него временные файлы копятся при каждой ошибке распознавания, а не только при успешной обработке.

Задержка: честно о том, сколько ждать

Голосовой бот не отвечает мгновенно, и это стоит закладывать в UX заранее, а не объяснять пользователям постфактум. Обработка складывается из трёх задержек подряд: скачивание файла из Telegram, конвертация ffmpeg (доли секунды на короткой записи) и само распознавание Whisper — самый долгий этап. На CPU-сервере без GPU распознавание 10–15-секундного голосового моделью small обычно занимает несколько секунд; на medium — заметно дольше. Точные цифры сильно зависят от процессора VPS, длины записи и загрузки сервера в моменте — не берите цифры из чужих статей как гарантию для своего железа, проверяйте на своём.

Что реально помогает со стороны UX:

• send_chat_action("typing") перед началом обработки — Telegram сам гасит индикатор через ~5 секунд, поэтому его нужно обновлять повторно, если обработка идёт дольше;
• очередь на несколько одновременных голосовых — если бот активно используется, лучше сразу ограничить число параллельных распознаваний (например, через asyncio.Semaphore), иначе несколько долгих запросов подряд посадят CPU и увеличат задержку для всех;
• если бот стоит на слабом тарифе, для стабильной задержки в разумных пределах может понадобиться более мощный CPU — здесь помогает статья про то, сколько RAM нужно для Telegram-бота и отдельно про ресурсы под Whisper по моделям.

Если задержка критична (например, бот для голосовых заметок в реальном времени), стоит рассмотреть faster-whisper вместо оригинальной реализации — она использует более эффективный рантайм CTranslate2 и на CPU обычно работает быстрее при сравнимом качестве, хотя точный выигрыш опять же зависит от железа и модели.

Частые вопросы