Arquitectura de Bot con Whisper.cpp y aiogram 3.x

Creando un Bot de Telegram para Reconocimiento de Voz en Español con Whisper.cpp y FSM

Un bot de Telegram diseñado para ayudar a los usuarios a practicar español utiliza el motor local whisper.cpp para el reconocimiento de voz sin depender de APIs en la nube. El sistema evalúa la pronunciación frente a frases de referencia, permite interacciones conversacionales mediante LLMs y funciona eficientemente en entornos VPS con recursos limitados. Al integrar aiogram 3.x asíncrono con el binario whisper.cpp, el bot maneja múltiples solicitudes al mismo tiempo.

Los componentes clave incluyen la conversión de audio a formato WAV mono de 16 kHz, la llamada a whisper-cli en un proceso separado y un algoritmo de comparación palabra por palabra para medir la precisión.

Integración No Bloqueante de Whisper.cpp

Usar subprocess.run() estándar bloquea el bucle de eventos de aiogram. La solución? Utilizar asyncio.create_subprocess_exec para ejecución asíncrona y asyncio.to_thread para desviar operaciones pesadas de pydub.

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. Convertir audio al formato compatible con Whisper (16kHz, mono, WAV)
        # Desviar el procesamiento de pydub a un hilo para evitar bloquear el bucle de eventos
        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("Archivo WAV no creado")

        # 2. Ejecutar whisper.cpp de forma asíncrona
        process = await asyncio.create_subprocess_exec(
            "/root/whisper.cpp/build/bin/whisper-cli", # Ruta al binario
            "-m", "/root/whisper.cpp/models/ggml-tiny.bin", # Modelo pequeño
            "-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 falló: {stderr.decode()}")
            return ""

        # 3. Leer el resultado
        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"[ERROR CRÍTICO] {e}")
        return ""
    finally:
        # Limpiar archivos temporales
        for f in [wav_path, txt_path, file_path]:
            if os.path.exists(f):
                os.remove(f)

Este código garantiza el reconocimiento de voz en tiempo real sin congelar el bot. El modelo ggml-tiny.bin fue elegido por su velocidad en CPU.

Algoritmo de Puntuación de Pronunciación

La comparación de cadenas simples es poco confiable debido a los errores comunes de Whisper en artículos y terminaciones. En su lugar, usamos un algoritmo de puntuación posicional palabra por palabra:

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

    # Comparar palabras por posición
    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])
    
    # Fórmula: (coincidencias / longitud de referencia) * 100
    accuracy = int((matches / len(expected_words)) * 100)
    
    return max(0, min(100, accuracy))

Resultados ejemplares:

• Coincidencia completa: 100%
• Falta la última palabra: 80–90%
• Sin coincidencia: 0–20%

El umbral de éxito se establece en el 70%. Mejoras futuras incorporarán la distancia de Levenshtein para detectar errores dentro de las palabras.

FSM para Gestión de Estados y Enrutamiento

La máquina de estados aiogram.fsm separa dos modos: conversación libre y entrenamiento de pronunciación.

from aiogram.fsm.state import State, StatesGroup

class UserStates(StatesGroup):
    waiting_for_ai_dialog = State()  # Modo chat libre

Un manejador universal para mensajes de voz:

@dp.message(F.voice)
async def universal_voice_handler(message: Message, state: FSMContext):
    current_state = await state.get_state()
    
    # RAMA 1: Conversación con IA
    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

    # RAMA 2: Entrenamiento de Pronunciación (Lecciones)
    # ... lógica para obtener frase de referencia, reconocer y puntuar

Pila Tecnológica y Almacenamiento de Datos

• STT: whisper.cpp (ggml-tiny.bin, idioma español)
• TTS: gTTS + pydub
• LLM: API externa
• Almacenamiento: Archivos JSON (users.json, lessons.json, phrases.json)

Comandos de administrador (/addphrase, /stats) permiten actualizaciones dinámicas de contenido. Implementación escalable mediante PostgreSQL y Docker.

Ventajas Clave:

• Integración asíncrona de whisper.cpp evita que el bot se congele.
• Comparación a nivel de palabra ofrece retroalimentación precisa sobre pronunciación sin dependencias en la nube.
• FSM permite ampliar fácilmente nuevos modos sin reescribir manejadores.
• Procesamiento local garantiza privacidad y eficiencia de costos.
• El modelo tiny.bin funciona rápido en CPU con latencia mínima.

— Editorial Team