Bot-Architektur mit Whisper.cpp und aiogram 3.x

Telegram-Bot für spanische Spracherkennung mit Whisper.cpp und FSM erstellen

Ein Telegram-Bot, der Nutzern beim Lernen der spanischen Sprache hilft, nutzt die lokale Whisper.cpp-Engine zur Spracherkennung, ohne auf Cloud-APIs angewiesen zu sein. Das System bewertet die Aussprache anhand Referenzphrasen, unterstützt konversationelle Interaktionen über KI-Modelle und läuft effizient in Umgebungen mit begrenzten Ressourcen wie VPS. Durch die Integration von async aiogram 3.x mit dem whisper.cpp-Binärfile verarbeitet der Bot mehrere Anfragen gleichzeitig.

Wichtige Komponenten sind die Umwandlung von Audio in das 16 kHz mono WAV-Format, die Ausführung von whisper-cli in einem separaten Prozess sowie ein Wort-für-Wort-Vergleichsalgorithmus zur Bewertung der Genauigkeit.

Nicht-blockierende Integration von Whisper.cpp

Die Verwendung von subprocess.run() blockiert die aiogram-Ereignisschleife. Die Lösung? Einsatz von asyncio.create_subprocess_exec für asynchrone Ausführung und asyncio.to_thread, um rechenintensive pydub-Operationen auszulagern.

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. Konvertiere Audio in Whisper-kompatibles Format (16kHz, mono, WAV)
        # pydub-Verarbeitung in einem Thread auslagern, um die Ereignisschleife nicht zu blockieren
        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-Datei wurde nicht erstellt")

        # 2. Führe whisper.cpp asynchron aus
        process = await asyncio.create_subprocess_exec(
            "/root/whisper.cpp/build/bin/whisper-cli", # Pfad zur Binärdatei
            "-m", "/root/whisper.cpp/models/ggml-tiny.bin", # Tiny-Modell
            "-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 fehlgeschlagen: {stderr.decode()}")
            return ""

        # 3. Ergebnis lesen
        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"[KRITISCHER FEHLER] {e}")
        return ""
    finally:
        # Temporäre Dateien löschen
        for f in [wav_path, txt_path, file_path]:
            if os.path.exists(f):
                os.remove(f)

Dieser Code gewährleistet eine Echtzeit-Spracherkennung ohne, dass der Bot einfriert. Das ggml-tiny.bin-Modell wurde wegen seiner Geschwindigkeit auf der CPU ausgewählt.

Aussprachebewertungsalgorithmus

Eine einfache Zeichenkettenvergleich ist aufgrund typischer Fehler von Whisper bei Artikeln und Endungen unzuverlässig. Stattdessen verwenden wir einen wortbasierten Positionsscore-Algorithmus:

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

    # Wörter nach Position vergleichen
    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])
    
    # Formel: (Treffer / Länge der Referenz) * 100
    accuracy = int((matches / len(expected_words)) * 100)
    
    return max(0, min(100, accuracy))

Beispiel-Ergebnisse:

• Vollständige Übereinstimmung: 100%
• Letztes Wort fehlt: 80–90%
• Keine Übereinstimmung: 0–20%

Der Erfolgs-Schwellenwert liegt bei 70 %. Zukünftige Verbesserungen werden den Levenshtein-Abstand für Fehler innerhalb von Wörtern nutzen.

FSM für Zustandsmanagement und Routing

aiogram.fsm ermöglicht die Trennung zweier Modi: freie Konversation und Aussprachetraining.

from aiogram.fsm.state import State, StatesGroup

class UserStates(StatesGroup):
    waiting_for_ai_dialog = State()  # Freier Chat-Modus

Ein universeller Handler für Sprachnachrichten:

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

    # ZWEIG 2: Aussprachetraining (Lektionen)
    # ... Logik zum Abrufen der Referenzphrase, Erkennen und Bewerten

Technologiestack und Datenhaltung

• STT: whisper.cpp (ggml-tiny.bin, spanisch)
• TTS: gTTS + pydub
• LLM: Externe API
• Speicherung: JSON-Dateien (users.json, lessons.json, phrases.json)

Admin-Befehle (/addphrase, /stats) ermöglichen dynamische Inhaltsaktualisierungen. Skalierbare Bereitstellung via PostgreSQL und Docker.

Wesentliche Vorteile:

• Asynchrone Integration von whisper.cpp verhindert das Einfrieren des Bots.
• Wortbasierte Vergleichslogik liefert präzise Ausspracheproben ohne Cloud-Abhängigkeiten.
• FSM erlaubt problemlose Erweiterung von Modi ohne Handler-Neuimplementierung.
• Lokale Verarbeitung garantiert Datenschutz und Kosteneffizienz.
• Das tiny.bin-Modell läuft schnell auf der CPU mit minimaler Latenz.

— Editorial Team