Créer un bot Telegram pour la reconnaissance vocale espagnole avec Whisper.cpp et FSM
Un bot Telegram conçu pour aider les utilisateurs à pratiquer l'espagnol utilise le moteur local whisper.cpp pour la reconnaissance vocale, sans dépendre d'API cloud. Le système évalue la prononciation par rapport à des phrases de référence, prend en charge les interactions conversationnelles via des LLM, et fonctionne efficacement sur des environnements VPS à ressources limitées. En intégrant aiogram 3.x asynchrone avec le binaire whisper.cpp, le bot traite plusieurs requêtes en parallèle.
Les composants clés incluent la conversion audio au format WAV mono 16 kHz, l'appel de whisper-cli dans un processus séparé, et un algorithme de comparaison mot par mot pour évaluer la précision.
Intégration non bloquante de Whisper.cpp
Utiliser subprocess.run() standard bloque la boucle d'événements aiogram. La solution ? Utiliser asyncio.create_subprocess_exec pour une exécution asynchrone et asyncio.to_thread pour déléguer les opérations lourdes 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 l'audio au format compatible Whisper (16kHz, mono, WAV)
# Déléguer le traitement pydub à un thread pour éviter de bloquer la boucle d'événements
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("Fichier WAV non créé")
# 2. Exécuter whisper.cpp de manière asynchrone
process = await asyncio.create_subprocess_exec(
"/root/whisper.cpp/build/bin/whisper-cli", # Chemin vers le binaire
"-m", "/root/whisper.cpp/models/ggml-tiny.bin", # Modèle 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"[ERREUR] Whisper a échoué : {stderr.decode()}")
return ""
# 3. Lire le résultat
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"[ERREUR CRITIQUE] {e}")
return ""
finally:
# Nettoyage des fichiers temporaires
for f in [wav_path, txt_path, file_path]:
if os.path.exists(f):
os.remove(f)
Ce code garantit une reconnaissance vocale en temps réel sans figer le bot. Le modèle ggml-tiny.bin a été choisi pour sa rapidité sur CPU.
Algorithme d'évaluation de la prononciation
Une comparaison simple de chaînes est peu fiable en raison des erreurs fréquentes de Whisper dans les articles et les terminaisons. Nous utilisons donc un algorithme de scoring positionnel mot par mot :
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
# Comparer les mots par 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])
# Formule : (correspondances / longueur de référence) * 100
accuracy = int((matches / len(expected_words)) * 100)
return max(0, min(100, accuracy))
Exemples de résultats :
- Correspondance complète : 100 %
- Dernier mot manquant : 80–90 %
- Aucune correspondance : 0–20 %
Le seuil de réussite est fixé à 70 %. Des améliorations futures intégreront la distance de Levenshtein pour détecter les erreurs intra-mot.
FSM pour la gestion d'état et le routage
La machine à états aiogram.fsm sépare deux modes : conversation libre et entraînement à la prononciation.
from aiogram.fsm.state import State, StatesGroup
class UserStates(StatesGroup):
waiting_for_ai_dialog = State() # Mode chat libre
Un gestionnaire universel de messages vocaux :
@dp.message(F.voice)
async def universal_voice_handler(message: Message, state: FSMContext):
current_state = await state.get_state()
# BRANCHE 1 : Conversation 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
# BRANCHE 2 : Entraînement à la prononciation (leçons)
# ... logique pour récupérer la phrase de référence, reconnaître et noter
Stack technique et stockage de données
- STT : whisper.cpp (ggml-tiny.bin, langue espagnole)
- TTS : gTTS + pydub
- LLM : API externe
- Stockage : fichiers JSON (users.json, lessons.json, phrases.json)
Des commandes administrateur (/addphrase, /stats) permettent des mises à jour dynamiques du contenu. Déploiement évolutif via PostgreSQL et Docker.
Avantages clés :
- Intégration asynchrone de whisper.cpp empêche le gel du bot.
- Comparaison au niveau mot fournit un retour précis sur la prononciation sans dépendre du cloud.
- FSM permet une extension facile des modes sans réécrire les gestionnaires.
- Traitement local assure confidentialité et efficacité coûts.
- Le modèle tiny.bin tourne rapidement sur CPU avec une latence minimale.
— Editorial Team
Aucun commentaire pour le moment.