Как сделать Telegram-бота для расшифровки аудио на новой модели GigaAM-v3 от Сбера

Почему именно GigaAM-v3 и Telegram бот?

В мире speech-to-text (STT) моделей появился новый игрок — GigaAM-v3 от Сбера. Эта модель показывает впечатляющие результаты для русского языка, особенно в условиях шума и различных акцентов. Но как обычному пользователю получить к ней доступ? API-доступ ограничен, а локальный запуск требует серьезных ресурсов.

Telegram бот становится идеальным решением: он доступен с любого устройства, не требует установки дополнительного ПО, а главное — позволяет централизованно управлять вычислительными ресурсами. В отличие от мультиагентных систем, которые могут быть излишни для простых задач, наш бот будет выполнять одну функцию, но делать это максимально эффективно.

Архитектура решения

Прежде чем писать код, давайте разберемся, как будет работать наша система:

1Подготовка окружения

Начнем с установки необходимых зависимостей. Нам понадобится Python 3.9+ и CUDA для работы с GPU.

# Создаем виртуальное окружение
python -m venv venv
source venv/bin/activate  # для Windows: venv\Scripts\activate

# Устанавливаем базовые зависимости
pip install torch torchaudio --index-url https://download.pytorch.org/whl/cu118
pip install python-telegram-bot==20.7
pip install redis rq
pip install soundfile pydub
pip install minio  # для облачного хранилища

2Получение и настройка GigaAM-v3

Модель GigaAM-v3 доступна через Hugging Face Hub. Для доступа может потребоваться авторизация.

from transformers import Wav2Vec2Processor, Wav2Vec2ForCTC
import torch

# Загрузка модели и процессора
model_name = "ai-forever/gigaam-v3"
processor = Wav2Vec2Processor.from_pretrained(model_name)
model = Wav2Vec2ForCTC.from_pretrained(model_name)

# Перенос модели на GPU если доступен
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model.to(device)
model.eval()  # переключаем в режим инференса

Для работы с различными форматами аудио нам понадобится функция конвертации:

import io
import soundfile as sf
from pydub import AudioSegment

def convert_audio(input_audio, target_sr=16000):
    """Конвертирует аудио в формат для GigaAM-v3"""
    # Загружаем аудио через pydub (поддерживает много форматов)
    if isinstance(input_audio, bytes):
        audio = AudioSegment.from_file(io.BytesIO(input_audio))
    else:
        audio = AudioSegment.from_file(input_audio)
    
    # Конвертируем в моно, 16kHz, 16-bit
    audio = audio.set_frame_rate(target_sr)
    audio = audio.set_channels(1)
    audio = audio.set_sample_width(2)  # 16-bit
    
    # Экспортируем в raw PCM
    buffer = io.BytesIO()
    audio.export(buffer, format="wav")
    buffer.seek(0)
    
    # Загружаем через soundfile для получения numpy массива
    data, samplerate = sf.read(buffer)
    return data, samplerate

3Создание Telegram бота

Создаем бота через @BotFather в Telegram и получаем токен. Основной код бота:

from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
import logging

# Настройка логирования
logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    level=logging.INFO
)
logger = logging.getLogger(__name__)

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Обработчик команды /start"""
    welcome_text = """👋 Привет! Я бот для расшифровки аудио.

Отправьте мне голосовое сообщение или аудиофайл, и я преобразую его в текст с помощью модели GigaAM-v3 от Сбера.

Поддерживаемые форматы:
• Голосовые сообщения Telegram
• Аудиофайлы (MP3, WAV, OGG, M4A)
• Видеосообщения (извлеку аудио)

Максимальная длина: 5 минут"""
    await update.message.reply_text(welcome_text)

async def handle_audio(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Обработка аудиосообщений"""
    try:
        # Отправляем сообщение о начале обработки
        status_msg = await update.message.reply_text(
            "🎧 Получил аудио. Начинаю обработку..."
        )
        
        # Получаем файл
        if update.message.voice:
            file = await update.message.voice.get_file()
        elif update.message.audio:
            file = await update.message.audio.get_file()
        elif update.message.video_note:
            file = await update.message.video_note.get_file()
        else:
            await update.message.reply_text("Пожалуйста, отправьте аудио или голосовое сообщение.")
            return
        
        # Скачиваем файл
        audio_bytes = await file.download_as_bytearray()
        
        # Добавляем задачу в очередь
        job = audio_queue.enqueue(
            process_audio_task,
            audio_bytes,
            update.effective_user.id,
            update.message.message_id,
            status_msg.message_id
        )
        
        await status_msg.edit_text(
            f"✅ Аудио добавлено в очередь. Позиция: {len(audio_queue) + 1}"
        )
        
    except Exception as e:
        logger.error(f"Error handling audio: {e}")
        await update.message.reply_text("❌ Произошла ошибка при обработке аудио.")

def main():
    """Запуск бота"""
    # Токен бота (хранить в переменных окружения!)
    TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
    
    # Создаем приложение
    application = Application.builder().token(TOKEN).build()
    
    # Регистрируем обработчики
    application.add_handler(CommandHandler("start", start))
    application.add_handler(MessageHandler(
        filters.VOICE | filters.AUDIO | filters.VIDEO_NOTE,
        handle_audio
    ))
    
    # Запускаем бота
    application.run_polling(allowed_updates=Update.ALL_TYPES)

if __name__ == "__main__":
    main()

4Система очередей с Redis и RQ

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

import redis
from rq import Queue
from rq.job import Job

# Подключение к Redis
redis_conn = redis.Redis(
    host=os.getenv("REDIS_HOST", "localhost"),
    port=int(os.getenv("REDIS_PORT", 6379)),
    password=os.getenv("REDIS_PASSWORD", None),
    decode_responses=False
)

# Создаем очередь
audio_queue = Queue("audio_processing", connection=redis_conn)

def process_audio_task(audio_bytes, user_id, message_id, status_msg_id):
    """Фоновая задача для обработки аудио"""
    try:
        # Конвертируем аудио
        audio_array, samplerate = convert_audio(audio_bytes)
        
        # Подготавливаем входные данные для модели
        inputs = processor(
            audio_array,
            sampling_rate=samplerate,
            return_tensors="pt",
            padding=True
        ).to(device)
        
        # Прямой проход через модель
        with torch.no_grad():
            logits = model(inputs.input_values).logits
        
        # Декодируем предсказания
        predicted_ids = torch.argmax(logits, dim=-1)
        transcription = processor.batch_decode(predicted_ids)[0]
        
        # Отправляем результат через Telegram API
        send_result_via_telegram(
            user_id,
            message_id,
            status_msg_id,
            transcription
        )
        
        return {"status": "success", "text_length": len(transcription)}
        
    except Exception as e:
        logger.error(f"Error in audio processing task: {e}")
        # Отправляем сообщение об ошибке
        send_error_via_telegram(user_id, message_id, status_msg_id, str(e))
        return {"status": "error", "error": str(e)}

5Запуск воркеров и мониторинг

Запускаем воркеров для обработки задач из очереди:

# Запуск воркера (отдельный процесс)
rq worker audio_processing --url redis://localhost:6379

Для мониторинга очереди можно использовать rq-dashboard:

pip install rq-dashboard
rq-dashboard

Dashboard будет доступен по адресу http://localhost:9181 и покажет все задачи, их статусы и ошибки.

Деплой и масштабирование

Для продакшена рекомендую использовать Docker и оркестратор:

# Dockerfile для бота
FROM python:3.10-slim

WORKDIR /app

# Устанавливаем системные зависимости
RUN apt-get update && apt-get install -y \
    ffmpeg \
    libsndfile1 \
    && rm -rf /var/lib/apt/lists/*

# Копируем зависимости
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Копируем код
COPY . .

# Запускаем бота
CMD ["python", "bot.py"]

Docker Compose для локального тестирования:

version: '3.8'

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

  bot:
    build: .
    depends_on:
      - redis
    environment:
      - TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
      - REDIS_HOST=redis
    restart: unless-stopped

  worker:
    build: .
    command: rq worker audio_processing --url redis://redis:6379
    depends_on:
      - redis
    environment:
      - REDIS_HOST=redis
    deploy:
      replicas: 2  # Запускаем 2 воркера
    restart: unless-stopped

  dashboard:
    image: python:3.10-slim
    command: >
      sh -c "pip install rq-dashboard && 
             rq-dashboard --url redis://redis:6379"
    ports:
      - "9181:9181"
    depends_on:
      - redis

volumes:
  redis_data:

Оптимизация производительности

GigaAM-v3 — большая модель. Вот несколько советов по оптимизации:

• Кэширование модели: Загружайте модель один раз при запуске воркера
• Пакетная обработка: Обрабатывайте несколько коротких аудиофайлов вместе
• Квантование: Используйте torch.quantization для уменьшения размера модели
• GPU memory: Настройте очистку кэша CUDA между задачами

# Пример кэширования модели в глобальной переменной
_global_model = None
_global_processor = None

def get_model():
    global _global_model, _global_processor
    if _global_model is None:
        _global_model = Wav2Vec2ForCTC.from_pretrained("ai-forever/gigaam-v3")
        _global_model.to(device)
        _global_model.eval()
        
        _global_processor = Wav2Vec2Processor.from_pretrained("ai-forever/gigaam-v3")
    return _global_model, _global_processor

Возможные ошибки и их решение

FAQ

Создание Telegram-бота для расшифровки аудио на GigaAM-v3 — отличный пример практического применения современных AI-моделей. Этот проект сочетает в себе передовые технологии speech-to-text, асинхронную обработку задач и удобный интерфейс для пользователей. Как и в случае с попытками решить сложные математические задачи через ML, здесь мы видим, как специализированные модели меняют подход к решению повседневных задач.

Для дальнейшей оптимизации рассмотрите использование NPU в AI MAX 395 для аппаратного ускорения или экспериментируйте с квантованием модели для уменьшения требований к памяти.