Bot Architecture with Whisper.cpp and aiogram 3.x

Building a Telegram Bot for Spanish Speech Recognition with Whisper.cpp and FSM

A Telegram bot designed to help users practice Spanish uses the local whisper.cpp engine for speech recognition without relying on cloud APIs. The system evaluates pronunciation against reference phrases, supports conversational interactions via LLMs, and runs efficiently on VPS environments with limited resources. By integrating async aiogram 3.x with the whisper.cpp binary, the bot handles multiple requests in parallel.

Key components include audio conversion to 16kHz mono WAV format, calling whisper-cli in a separate process, and a word-by-word comparison algorithm for scoring accuracy.

Non-blocking Integration of Whisper.cpp

Using standard subprocess.run() blocks the aiogram event loop. The solution? Use asyncio.create_subprocess_exec for asynchronous execution and asyncio.to_thread to offload heavy pydub operations.

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. Convert audio to Whisper-compatible format (16kHz, mono, WAV)
        # Offload pydub processing to a thread to avoid blocking the event 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. Run whisper.cpp asynchronously
        process = await asyncio.create_subprocess_exec(
            "/root/whisper.cpp/build/bin/whisper-cli", # Path to binary
            "-m", "/root/whisper.cpp/models/ggml-tiny.bin", # Tiny model
            "-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. Read result
        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:
        # Cleanup temporary files
        for f in [wav_path, txt_path, file_path]:
            if os.path.exists(f):
                os.remove(f)

This code ensures real-time speech recognition without freezing the bot. The ggml-tiny.bin model was chosen for its speed on CPU.

Pronunciation Scoring Algorithm

Simple string comparison is unreliable due to Whisper’s common errors in articles and endings. Instead, we use a word-by-word positional scoring algorithm:

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

    # Compare words by position
    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])
    
    # Formula: (matches / reference length) * 100
    accuracy = int((matches / len(expected_words)) * 100)
    
    return max(0, min(100, accuracy))

Example outcomes:

• Full match: 100%
• Missing last word: 80–90%
• No match: 0–20%

The success threshold is set at 70%. Future improvements will incorporate Levenshtein distance for intra-word error detection.

FSM for State Management and Routing

aiogram.fsm state machine separates two modes: free conversation and pronunciation training.

from aiogram.fsm.state import State, StatesGroup

class UserStates(StatesGroup):
    waiting_for_ai_dialog = State()  # Free chat mode

A universal voice message handler:

@dp.message(F.voice)
async def universal_voice_handler(message: Message, state: FSMContext):
    current_state = await state.get_state()
    
    # BRANCH 1: AI Conversation
    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

    # BRANCH 2: Pronunciation Training (Lessons)
    # ... logic to fetch reference phrase, recognize, and score

Tech Stack and Data Storage

• STT: whisper.cpp (ggml-tiny.bin, Spanish language)
• TTS: gTTS + pydub
• LLM: External API
• Storage: JSON files (users.json, lessons.json, phrases.json)

Admin commands (/addphrase, /stats) enable dynamic content updates. Scalable deployment via PostgreSQL and Docker.

Key Advantages:

• Asynchronous whisper.cpp integration prevents bot freezing.
• Word-level comparison delivers accurate pronunciation feedback without cloud dependencies.
• FSM enables easy expansion of modes without rewriting handlers.
• Local processing ensures privacy and cost efficiency.
• The tiny.bin model runs fast on CPU with minimal latency.

— Editorial Team