Architektura bota z Whisper.cpp i aiogram 3.x

Architektura bota Telegram do rozpoznawania mowy hiszpańskiej z Whisper.cpp i FSM

Bot Telegram do nauki hiszpańskiego wykorzystuje lokalny Whisper.cpp do rozpoznawania mowy bez korzystania z chmurowych API. System ocenia wymowę na podstawie wzorcowych fraz, obsługuje rozmowy za pomocą LLM i działa na VPS z ograniczonymi zasobami. Integracja asynchroniczna aiogram 3.x z binarnym whisper.cpp zapewnia równoległą przetwarzanie żądań.

Kluczowe komponenty: konwersja dźwięku do formatu WAV mono 16 kHz, wywołanie whisper-cli w oddzielnym procesie, algorytm porównania słów do oceny dokładności.

Integracja Whisper.cpp bez blokowania event loop

Standardowa funkcja subprocess.run() blokuje pętlę zdarzeń aiogram. Rozwiązaniem jest asyncio.create_subprocess_exec do uruchamiania asynchronicznego oraz asyncio.to_thread do konwersji z wykorzystaniem 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. Konwersja do formatu zrozumiałego przez Whisper (16kHz, mono, wav)
        # Przenosimy ciężką operację pydub do osobnego wątku, by nie blokować pętli
        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("Plik WAV nie został utworzony")

        # 2. Asynchroniczne uruchomienie whisper.cpp
        process = await asyncio.create_subprocess_exec(
            "/root/whisper.cpp/build/bin/whisper-cli", # Ścieżka do pliku binarnego
            "-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"[BŁĄD] Whisper nie powiódł się: {stderr.decode()}")
            return ""

        # 3. Odczyt wyniku
        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"[KRYTYCZNY BŁĄD] {e}")
        return ""
    finally:
        # Czyszczenie plików tymczasowych
        for f in [wav_path, txt_path, file_path]:
            if os.path.exists(f):
                os.remove(f)

Ten kod zapewnia rozpoznawanie bez zawieszenia bota. Wybrano model ggml-tiny.bin ze względu na szybkość działania na CPU.

Algorytm oceny wymowy

Proste porównanie ciągów znaków jest niepewne z powodu błędów Whisper w artykułach i końcówkach. Zaimplementowano analizę słów z wagą pozycyjną:

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

    # Porównujemy słowa według pozycji
    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])
    
    # Wzór: (zgodne / długość wzorca) * 100
    accuracy = int((matches / len(expected_words)) * 100)
    
    return max(0, min(100, accuracy))

Przykłady działania algorytmu:

• Pełne pokrycie: 100%.
• Pominięcie ostatniego słowa: 80–90%.
• Brak zgodności: 0–20%.

Prog dla sukcesu to 70%. Planowane ulepszenie z odległością Levenshteina do poprawy błędów wewnątrz słów.

FSM do zarządzania stanami i routingu

Maszyna stanów aiogram.fsm dzieli tryby: swobodna rozmowa i trening wymowy.

from aiogram.fsm.state import State, StatesGroup

class UserStates(StatesGroup):
    waiting_for_ai_dialog = State()  # Tryb swobodnej rozmowy

Uniwersalny handler wiadomości głosowych:

@dp.message(F.voice)
async def universal_voice_handler(message: Message, state: FSMContext):
    current_state = await state.get_state()
    
    # GAŁĘŹ 1: Rozeznanie z 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

    # GAŁĘŹ 2: Trening wymowy (Lekcje)
    # ... logika pobierania wzorcowej frazy, rozpoznawania i oceny

Stos technologiczny i przechowywanie danych

• STT: whisper.cpp (ggml-tiny.bin, język es).
• TTS: gTTS + pydub.
• LLM: zewnętrzne API.
• Przechowywanie: JSON (users.json, lessons.json, phrases.json).

Komendy admina (/addphrase, /stats) do dynamicznego wypełniania treści. Skalowanie do PostgreSQL i Docker.

Co jest ważne:

• Asynchroniczna integracja whisper.cpp zapobiega blokowaniu bota.
• Porównanie słów daje precyzyjną ocenę wymowy bez usług chmurowych.
• FSM pozwala na rozbudowę trybów bez refaktoryzacji handlerów.
• Lokalne przetwarzanie zapewnia prywatność i oszczędność kosztów.
• Model tiny.bin działa na CPU z niskim opóźnieniem.

— Editorial Team