Introduction à Qwen3-ASR-1.7B et l'écosystème Ascend
Le modèle Qwen3-ASR-1.7B d'Alibaba Cloud représente une avancée significative en matière de reconnaissance vocale automatique (ASR), offrant des capacités de compréhension sémantique améliorées par rapport à ses prédécesseurs. Ce guide détaillé vous accompagnera dans le processus d'intégration et d'optimisation de ce modèle sur les plateformes matérielles Huawei Ascend, en tirant parti de l'environnement de développement et d'exécution CANN (Compute Architecture for Neural Networks).
À travers ce tutoriel, vous explorerez les étapes essentielles pour :
- Mettre en place un environnement de développement Huawei CANN.
- Acquérir et préparer les ressources du modèle Qwen3-ASR-1.7B.
- Effectuer la conversion et l'optimisation du modèle pour Ascend.
- Réaliser des évaluations pratiques de transcription vocale.
Prérequis techniques et installation des dépendances
Configuration matérielle requise
Pour garantir un fonctionnement optimal du modèle Qwen3-ASR-1.7B, les spécifications matérielles minimales sont les suivantes :
- Processeur IA Huawei Ascand (par exemple, 310P ou 910B).
- Mémoire système : au moins 32 Go.
- Espace disque disponible : 50 Go.
- Processeur hôte supportant les instructions AVX.
Installation des dépendances logicielles
Commencez par installer les paquets système et Python indispensables :
# Mise à jour des dépôts du gestionnaire de paquets
sudo apt-get update
# Installation des utilitaires de base
sudo apt-get install -y python3.8 python3-pip git wget curl
# Création d'un environnement virtuel Python pour l'ASR
python3.8 -m venv env_asr_qwen
source env_asr_qwen/bin/activate
# Mise à jour de pip et installation des bibliothèques Python
pip install --upgrade pip
pip install numpy==1.21.6 torch==1.10.0 transformers==4.26.0
Installation de la suite d'outils CANN
Téléchargez et installez la boîte à outils CANN de Huawei. Assurez-vous de choisir la version appropriée pour votre architecture :
# Téléchargement du kit d'outils CANN (ajustez le lien et la version si nécessaire)
wget https://ascend-repo.obs.cn-east-2.myhuaweicloud.com/CANN/6.3.RC1/ubuntu-aarch64/Ascend-cann-toolkit_6.3.RC1_linux-aarch64.run
# Attribution des droits d'exécution
chmod +x Ascend-cann-toolkit_6.3.RC1_linux-aarch64.run
# Lancement de l'installation de CANN
./Ascend-cann-toolkit_6.3.RC1_linux-aarch64.run --install
Après l'installation, configurez les variables d'environnement en ajoutant la ligne suivante à votre fichier ~/.bashrc et en le sourçant :
echo "source /usr/local/Ascend/ascend-toolkit/set_env.sh" >> ~/.bashrc
source ~/.bashrc
Acquisition et conversion du modèle
Téléchargement des fichiers du modèle
Récupérez les fichiers du modèle Qwen3-ASR-1.7B depuis la source officielle. Créez un répertoire dédié pour organiser vos modèles :
# Création du répertoire de stockage des modèles
mkdir -p ~/modeles_asr/qwen3-asr-1.7b
cd ~/modeles_asr/qwen3-asr-1.7b
# Téléchargement des archives du modèle (exemple, remplacez par le lien réel)
wget https://example.com/modeles/qwen3-asr-1.7b/fichiers_modele.zip
unzip fichiers_modele.zip
Transformation du format du modèle
Le modèle doit être converti au format OM (Offline Model) pour être exécuté efficacement sur l'architecture Ascend. Cela implique généralement une étape intermédiaire en ONNX.
Utilisez un script Python pour exporter le modèle au format ONNX :
# export_onnx.py
import torch
from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor
# Chemin vers les fichiers du modèle téléchargé
chemin_modele_local = "~/modeles_asr/qwen3-asr-1.7b"
# Chargement du modèle et du processeur pré-entraînés
modele_qwen = AutoModelForSpeechSeq2Seq.from_pretrained(chemin_modele_local)
processeur_audio = AutoProcessor.from_pretrained(chemin_modele_local)
# Exportation du modèle vers le format ONNX
# L'entrée est un tenseur simulant 1 seconde d'audio à 16kHz
torch.onnx.export(
modele_qwen,
torch.randn(1, 16000), # Exemple d'entrée : batch_size=1, 16000 échantillons
"qwen3_asr_1_7b.onnx",
opset_version=13,
input_names=['signal_audio'],
output_names=['resultats_transcription'],
dynamic_axes={
'signal_audio': {0: 'batch_size', 1: 'sequence_length'},
'resultats_transcription': {0: 'batch_size', 1: 'output_length'}
}
)
print("Modèle exporté au format ONNX : qwen3_asr_1_7b.onnx")
Ensuite, utilisez l'outil ATC (Ascend Tensor Compiler) pour convertir le fichier ONNX en OM :
atc --model=qwen3_asr_1_7b.onnx \
--framework=5 \
--output=qwen3_asr_1_7b \
--soc_version=Ascend310P3 \
--input_format=ND \
--input_shape="signal_audio:1,16000" \
--log=info \
--output_type=FP16
Déploiement et exécution
Développement du script d'inférence
Écrivez un script Python pour gérer le processus d'inférence avec le modèle OM. Ce script utilisera l'API ACL (Ascend Computing Language) pour interagir avec le matériel Ascend.
# inference_ascend.py
import numpy as np
import acl
import acl_dvpp
import acl_util
from transformers import AutoProcessor
class ServiceTranscriptionAscend:
"""
Service de transcription audio utilisant le modèle Qwen3-ASR-1.7B sur Ascend.
"""
def __init__(self, chemin_modele_om, chemin_processeur_transformers="qwen3-asr-1.7b"):
self.chemin_modele_om = chemin_modele_om
self.processeur_audio = AutoProcessor.from_pretrained(chemin_processeur_transformers)
self.appareil_id = 0
self.modele_id = None
self.contexte = None
self.flux = None
self.description_entrees = None
self.description_sorties = None
self.initialiser_acl()
def initialiser_acl(self):
"""Initialise l'environnement ACL et charge le modèle OM."""
ret = acl.init()
if ret != acl.ACL_SUCCESS:
raise RuntimeError(f"Échec de l'initialisation ACL: {ret}")
ret = acl.rt.set_device(self.appareil_id)
if ret != acl.ACL_SUCCESS:
raise RuntimeError(f"Échec de la définition du périphérique {self.appareil_id}: {ret}")
self.contexte, ret = acl.rt.create_context(self.appareil_id)
if ret != acl.ACL_SUCCESS:
raise RuntimeError(f"Échec de la création du contexte ACL: {ret}")
self.flux, ret = acl.rt.create_stream()
if ret != acl.ACL_SUCCESS:
raise RuntimeError(f"Échec de la création du flux ACL: {ret}")
self.modele_id, ret = acl.mdl.load_from_file(self.chemin_modele_om)
if ret != acl.ACL_SUCCESS:
raise RuntimeError(f"Échec du chargement du modèle {self.chemin_modele_om}: {ret}")
self.description_entrees = acl.mdl.get_input_desc(self.modele_id, 0) # Assumons une seule entrée
self.description_sorties = acl.mdl.get_output_desc(self.modele_id, 0) # Assumons une seule sortie
def _preparer_entree(self, donnees_brutes_audio):
"""Prétraite les données audio et les transfère sur le périphérique Ascend."""
# Le processeur retourne déjà un tableau numpy après normalisation
inputs = self.processeur_audio(
donnees_brutes_audio,
sampling_rate=16000,
return_tensors="np"
).input_values
# Assurer que l'entrée est de type float32 pour Ascend
inputs = inputs.astype(np.float32)
# Allouer la mémoire sur le périphérique
taille_entree = inputs.size * inputs.itemsize # taille en octets
buffer_entree_dev = acl.rt.malloc(taille_entree, acl.ACL_MEM_MALLOC_HUGE_FIRST)
if buffer_entree_dev == 0:
raise RuntimeError("Échec de l'allocation mémoire pour l'entrée sur le périphérique.")
# Copier les données de l'hôte vers le périphérique
ret = acl.rt.memcpy(
buffer_entree_dev, taille_entree,
inputs.tobytes(), taille_entree,
acl.ACL_MEMCPY_HOST_TO_DEVICE
)
if ret != acl.ACL_SUCCESS:
raise RuntimeError("Échec de la copie des données d'entrée vers le périphérique.")
# Créer la description de l'entrée pour le modèle
acl_entrees = acl.mdl.create_input()
acl.mdl.add_dataset_buffer(acl_entrees, buffer_entree_dev, taille_entree)
return acl_entrees, buffer_entree_dev
def _obtenir_resultats(self, acl_sorties, buffer_sortie_dev):
"""Récupère les résultats du périphérique Ascend et les post-traite."""
# Récupérer le buffer de sortie du dataset ACL
taille_sortie = acl.mdl.get_dataset_buffer_size(acl_sorties, 0)
ptr_sortie_dev = acl.mdl.get_dataset_buffer(acl_sorties, 0)
# Allouer la mémoire sur l'hôte pour la copie
buffer_sortie_host = acl.rt.malloc_host(taille_sortie)
if buffer_sortie_host == 0:
raise RuntimeError("Échec de l'allocation mémoire pour la sortie sur l'hôte.")
# Copier les données du périphérique vers l'hôte
ret = acl.rt.memcpy(
buffer_sortie_host, taille_sortie,
ptr_sortie_dev, taille_sortie,
acl.ACL_MEMCPY_DEVICE_TO_HOST
)
if ret != acl.ACL_SUCCESS:
raise RuntimeError("Échec de la copie des données de sortie vers l'hôte.")
# Convertir le buffer en tableau numpy (assumons int64 pour les ids de tokens)
resultats_tokens = np.frombuffer(
acl.util.get_host_ptr_as_bytes(buffer_sortie_host, taille_sortie),
dtype=np.int64
)
# La forme de sortie attendue est (batch_size, sequence_length)
# Il faut potentiellement redimensionner si le modèle n'a pas une taille fixe
# Pour cet exemple, supposons que c'est une sortie simple de tokens
acl.rt.free_host(buffer_sortie_host)
return resultats_tokens
def transcrire_audio(self, donnees_audio_brutes):
"""
Transcrit un segment audio.
:param donnees_audio_brutes: Tableau numpy (int16 ou float32) des échantillons audio.
:return: Texte transcrit.
"""
acl_entrees, buffer_entree_dev = self._preparer_entree(donnees_audio_brutes)
# Créer les structures de données pour la sortie
acl_sorties = acl.mdl.create_output()
taille_sortie_max = acl.mdl.get_output_size_by_index(self.description_sorties, 0)
buffer_sortie_dev = acl.rt.malloc(taille_sortie_max, acl.ACL_MEM_MALLOC_HUGE_FIRST)
if buffer_sortie_dev == 0:
raise RuntimeError("Échec de l'allocation mémoire pour la sortie sur le périphérique.")
acl.mdl.add_dataset_buffer(acl_sorties, buffer_sortie_dev, taille_sortie_max)
# Exécuter l'inférence du modèle
ret = acl.mdl.execute(self.modele_id, acl_entrees, acl_sorties)
if ret != acl.ACL_SUCCESS:
raise RuntimeError(f"Échec de l'exécution du modèle: {ret}")
# Synchroniser le flux pour s'assurer que l'exécution est terminée
ret = acl.rt.synchronize_stream(self.flux)
if ret != acl.ACL_SUCCESS:
raise RuntimeError(f"Échec de la synchronisation du flux: {ret}")
# Récupérer et post-traiter les résultats
tokens_transcrits = self._obtenir_resultats(acl_sorties, buffer_sortie_dev)
texte_final = self.processeur_audio.decode(tokens_transcrits, skip_special_tokens=True)
# Libérer les ressources ACL
acl.mdl.destroy_dataset(acl_entrees)
acl.mdl.destroy_dataset(acl_sorties)
acl.rt.free(buffer_entree_dev)
acl.rt.free(buffer_sortie_dev)
return texte_final
def __del__(self):
"""Nettoie les ressources ACL à la destruction de l'objet."""
if self.modele_id:
acl.mdl.unload(self.modele_id)
if self.flux:
acl.rt.destroy_stream(self.flux)
if self.contexte:
acl.rt.destroy_context(self.contexte)
acl.rt.reset_device(self.appareil_id)
acl.finalize()
# Exemple d'utilisation
if __name__ == "__main__":
import soundfile as sf
import os
# Créer un fichier audio de test si non existant
if not os.path.exists("audio_test.wav"):
print("Création d'un fichier audio de test...")
from pydub import AudioSegment
from pydub.generators import Sine
# Générer un son sinusoïdal pour l'exemple
audio = Sine(440).to_audio_segment(duration=1000) # 1 seconde de son à 440 Hz
audio = audio.set_frame_rate(16000).set_channels(1).export("audio_test.wav", format="wav")
print("Fichier audio de test 'audio_test.wav' créé.")
# Chargement d'un fichier audio de test
audio_brut, freq_ech = sf.read("audio_test.wav")
if freq_ech != 16000:
print(f"Avertissement: Fréquence d'échantillonnage audio ({freq_ech} Hz) différente de 16000 Hz. Re-échantillonnage recommandé.")
# Pour une vraie application, il faudrait re-échantillonner
# Initialisation du service de transcription
service_asr = ServiceTranscriptionAscend("qwen3_asr_1_7b.om", "~/modeles_asr/qwen3-asr-1.7b")
# Exécution de la transcription
texte_transcrit = service_asr.transcrire_audio(audio_brut)
print(f"Résultat de la transcription : {texte_transcrit}")
del service_asr # S'assure que les ressources ACL sont libérées
Validation par des tests
Préparez un échantillon audio au format WAV et exécutez le script d'inférence pour valider le déploiement.
# Créez un répertoire pour les fichiers audio de test
mkdir -p audio_tests
# Placez votre fichier audio_sample.wav dans ce répertoire
# Exemple: mv mon_audio.wav audio_tests/audio_sample.wav
# Exécutez le script de transcription
python inference_ascend.py
Conseils d'optimisation des performances
Stratégies d'optimisation du modèle
Pour maximiser les performances de Qwen3-ASR-1.7B sur Ascend, considérez les points suivants :
- Traitement par lots (Batch Processing) : Traitez simultanément plusieurs segments audio pour améliorer l'utilisation du processeur IA.
- Gestion de la mémoire : Configurez l'allocation mémoire avec soin pour prévenir les copies inutiles et les débordements.
- Pipeline d'exécution : Chevauchez les phases de prétraitement des données et d'inférence du modèle.
# Exemple de traitement par lots
def traitement_audio_par_lots(chemins_audio, taille_lot=4, service_asr_instance):
"""
Traite une liste de fichiers audio par lots.
"""
resultats_lots = []
for i in range(0, len(chemins_audio), taille_lot):
chemins_batch = chemins_audio[i:i+taille_lot]
# Dans une implémentation réelle, cela nécessiterait un prétraitement batch et une inférence batch
# Pour cet exemple simple, nous appelons la fonction individuelle
for chemin_audio in chemins_batch:
audio_data, _ = sf.read(chemin_audio)
transcription = service_asr_instance.transcrire_audio(audio_data)
resultats_lots.append((chemin_audio, transcription))
return resultats_lots
Résolution des problèmes couarnts
- Erreur "Mémoire insuffisante" :
- Solution : Réduisez la taille des lots d'inférence ou optimisez la consommation mémoire de votre script.
- Lenteur de l'inférence :
- Solution : Activez les optimisations de pipeline de l'AI Core via les options ATC, ou ajustez le format d'entrée.
- Dégradation de la précision de reconnaissance :
- Solution : Vérifiez attentivement le processus de conversion du modèle pour s'assurer qu'aucune information n'est perdue (par exemple, utilisez
--output_type=FP32si FP16 dégrade trop la précision).
- Solution : Vérifiez attentivement le processus de conversion du modèle pour s'assurer qu'aucune information n'est perdue (par exemple, utilisez
Applications pratiques
Reconnaissance vocale en temps réel
Le modèle peut être intégré dans des systèmes de reconnaissance vocale en temps réel. Voici un prototype :
# asr_temps_reel.py
import pyaudio
import numpy as np
from collections import deque
import time
class SystemeASRTempsReel:
"""
Implémente la reconnaissance vocale en temps réel avec un moteur ASR Ascend.
"""
def __init__(self, service_asr_obj):
self.service_asr = service_asr_obj
self.freq_ech = 16000
self.taille_segment = self.freq_ech * 2 # 2 secondes d'audio par segment
self.buffer_audio = deque(maxlen=self.freq_ech * 10) # Buffer de 10 secondes
self.p_audio = pyaudio.PyAudio()
self.flux_audio = self.p_audio.open(format=pyaudio.paInt16,
channels=1,
rate=self.freq_ech,
input=True,
frames_per_buffer=1024)
print("Démarrage de l'écoute audio...")
def ecouter_et_transcrire(self):
"""
Capture l'audio en continu et déclenche la transcription par segments.
"""
try:
while True:
donnees_brutes = self.flux_audio.read(1024, exception_on_overflow=False)
echantillons = np.frombuffer(donnees_brutes, dtype=np.int16)
self.buffer_audio.extend(echantillons)
if len(self.buffer_audio) >= self.taille_segment:
segment_a_traiter = np.array(list(self.buffer_audio)[:self.taille_segment])
self.buffer_audio.clear() # Vider le buffer après traitement
if segment_a_traiter.size > 0:
debut_inf = time.time()
resultat = self.service_asr.transcrire_audio(segment_a_traiter)
fin_inf = time.time()
print(f"Transcription en temps réel ({round(fin_inf - debut_inf, 2)}s): {resultat}")
except KeyboardInterrupt:
print("\nArrêt de l'écoute.")
finally:
self.flux_audio.stop_stream()
self.flux_audio.close()
self.p_audio.terminate()
# Exemple d'exécution
if __name__ == "__main__":
# Assurez-vous que le modèle OM est prêt et le processeur Transformers est accessible
# Pour cet exemple, le chemin du processeur est le même que celui utilisé pour l'export ONNX
asr_moteur = ServiceTranscriptionAscend("qwen3_asr_1_7b.om", "~/modeles_asr/qwen3-asr-1.7b")
systeme_rt = SystemeASRTempsReel(asr_moteur)
systeme_rt.ecouter_et_transcrire()
Traitement de fichiers audio par lots
Pour les scénarios nécessitant le traitement d'un grand volume de fichiers audio, un script shell peut automatiser le processus :
#!/bin/bash
# script_traitement_batch.sh
REP_ENTREE="audios_source"
REP_SORTIE="textes_transcrits"
CHEMIN_MODELE_OM="qwen3_asr_1_7b.om"
CHEMIN_PROCESSEUR_TRANSFORMERS="~/modeles_asr/qwen3-asr-1.7b"
mkdir -p $REP_SORTIE
echo "Démarrage du traitement par lots..."
for fichier_audio in $REP_ENTREE/*.wav; do
if [ -f "$fichier_audio" ]; then
nom_base=$(basename "$fichier_audio" .wav)
echo "Traitement de : $fichier_audio"
# Exécute le script Python d'inférence pour chaque fichier
# Et redirige la sortie vers un fichier texte
python inference_ascend.py --audio "$fichier_audio" --model "$CHEMIN_MODELE_OM" --processor "$CHEMIN_PROCESSEUR_TRANSFORMERS" > "$REP_SORTIE/${nom_base}.txt"
echo "Transcription complétée pour : $fichier_audio -> $REP_SORTIE/${nom_base}.txt"
fi
done
echo "Tous les fichiers audio ont été traités !"