Architektura botu s Whisper.cpp a aiogram 3.x

Architektura Telegramového robota pro rozpoznávání španělské řeči pomocí Whisper.cpp a FSM

Telegramový robot pro procvičování španělštiny využívá lokální Whisper.cpp pro rozpoznávání řeči bez cloudových API. Systém hodnotí výslovnost na základě vzorových frází, podporuje rozhovory prostřednictvím LLM a běží na VPS s omezenými prostředky. Integrace asynchronního aiogram 3.x s binárním whisper.cpp umožňuje paralelní zpracování požadavků.

Klíčové komponenty: konverze zvuku do formátu 16 kHz mono WAV, volání whisper-cli v samostatném procesu, algoritmus slovního porovnání pro hodnocení.

Integrování Whisper.cpp bez blokování event loopu

Standardní subprocess.run() blokuje cyklus událostí aiogram. Řešením je použití asyncio.create_subprocess_exec pro asynchronní spuštění a asyncio.to_thread pro konverzi pomocí 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. Konverze do formátu pochopitelného Whisper (16kHz, mono, wav)
        # Těžkou operaci pydub přesuneme do samostatného vlákna, aby neblokovala 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 soubor nebyl vytvořen")

        # 2. Asynchronní spuštění whisper.cpp
        process = await asyncio.create_subprocess_exec(
            "/root/whisper.cpp/build/bin/whisper-cli", # Cesta k binárnímu souboru
            "-m", "/root/whisper.cpp/models/ggml-tiny.bin", # Model 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 selhal: {stderr.decode()}")
            return ""

        # 3. Čtení výsledku
        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"[KRITICKÁ CHYBA] {e}")
        return ""
    finally:
        # Vyčištění dočasných souborů
        for f in [wav_path, txt_path, file_path]:
            if os.path.exists(f):
                os.remove(f)

Tento kód zajišťuje rozpoznávání bez zablokování robota. Model ggml-tiny.bin byl vybrán kvůli rychlosti na CPU.

Algoritmus hodnocení výslovnosti

Jednoduché řetězcové porovnání není spolehlivé kvůli chybám Whisper v částicích a příponách. Implementován je slovový analýza s pozicionálním vážením:

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

    # Porovnáváme slova podle pozice
    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])
    
    # Formule: (shody / délka vzoru) * 100
    accuracy = int((matches / len(expected_words)) * 100)
    
    return max(0, min(100, accuracy))

Příklady fungování algoritmu:

• Úplné shodné: 100 %.
• Přeskočení posledního slova: 80–90 %.
• Nesrovnalost: 0–20 %.

Prah úspěchu je 70 %. Plánuje se vylepšení pomocí vzdálenosti Levenshteina pro vnitroslovové chyby.

FSM pro správu stavů a směrování

Stroj stavů aiogram.fsm odděluje režimy: svobodný rozhovor a procvičování výslovnosti.

from aiogram.fsm.state import State, StatesGroup

class UserStates(StatesGroup):
    waiting_for_ai_dialog = State()  # Režim svobodného rozhovoru

Univerzální handler pro hlasové zprávy:

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

    # VĚTEV 2: Procvičování výslovnosti (Leccióny)
    # ... logika získání vzorové fráze, rozpoznání a hodnocení

Technologický stack a uchování dat

• STT: whisper.cpp (ggml-tiny.bin, jazyk es).
• TTS: gTTS + pydub.
• LLM: externí API.
• Uchování: JSON (users.json, lessons.json, phrases.json).

Administrační příkazy (/addphrase, /stats) pro dynamické naplnění obsahu. Možnost škálování na PostgreSQL a Docker.

Důležité:

• Asynchronní integrace whisper.cpp zabrání zablokování robota.
• Slovové porovnání poskytuje přesné hodnocení výslovnosti bez cloudových služeb.
• FSM umožňuje rozšíření režimů bez refaktoringu handlerů.
• Lokální zpracování zajišťuje soukromí a úsporu.
• Model tiny.bin běží na CPU s nízkou latencí.

— Editorial Team