Les messages vocaux dans les canaux d'équipe peuvent poser des défis pour la communication rapide, notamment lors de la relecture des enregistrements pour des résumés écrits. Ce guide explique comment utiliser le modèle léger Qwen3-ASR-0.6B pour créer un service de reconnaissance vocale automatisé, puis l'intégrer à des robots Feishu ou WeCom afin de convertir les messages audio en texte sans intervention manuelle.

Avantages de Qwen3-ASR-0.6B pour les déploiements d'entreprise

Ce modèle offre plusieurs caractéristiques adoptées aux besoins professionnels :

• Support multilingue : reconnaît 52 langues et dialectes, incluant des variantes régionales du chinois et d'autres langues majeures.
• Installation légère : avec 600 millions de paramètres, il s'exécute efficacement sur des serveurs cloud standard sans GPU dédié.
• Performances optimisées : conçu pour une faible latence et une haute concurrence, grâce à une architecture de base avancée.
• Formats compatibles : prend en charge les formats audio courants tels que WAV, MP3 et M4A, avec une taille maximale de fichier de 100 Mo.

Déploiement du service de reconnaissance vocale

Préparation de l'environnement

Assurez-vous d'avoir un serveur Linux avec Python 3.9 et FFmpeg installés. Créez un répertoire de projet :

# Mise à jour du système et installation des dépendances
sudo apt update && sudo apt install -y python3.9 python3.9-venv ffmpeg
mkdir -p /opt/asr-service && cd /opt/asr-service

Script d'installation automatisé

Un script simplifié pour configurer le service :

#!/bin/bash
# setup_asr.sh

# Création de l'environnement virtuel
python3.9 -m venv env
source env/bin/activate

# Installation des bibliothèques
pip install fastapi uvicorn python-multipart torch torchaudio

# Structure du projet
mkdir -p app logs webui

# Fichier principal de l'API
cat > app/server.py << 'EOF'
from fastapi import FastAPI, File, UploadFile, Form
from fastapi.middleware.cors import CORSMiddleware
import torch
import torchaudio
from typing import Optional

application = FastAPI(title="Service ASR Qwen3")

application.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# Charger le modèle au démarrage
@app.on_event("startup")
async def initialize_model():
    global recognizer_model, audio_processor
    # Code de chargement du modèle à implémenter
    print("Modèle initialisé")

@app.get("/status")
async def check_status():
    return {"available": True, "gpu_enabled": torch.cuda.is_available()}

@app.post("/transcribe")
async def process_audio(
    audio: UploadFile = File(...),
    lang: Optional[str] = Form("auto")
):
    # Logique de transcription
    return {"transcript": "Texte extrait", "language": lang}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(application, host="0.0.0.0", port=8000)
EOF

# Interface web simple
cat > webui/index.html << 'EOF'


    <h1>Transcription audio</h1>
    <input accept="audio/*" id="inputFile" type="file"></input>
    <button onclick="startTranscription()">Lancer</button>
    <div id="output"></div>
    <script>
        async function startTranscription() {
            const file = document.getElementById('inputFile').files[0];
            if (!file) return;
            const formData = new FormData();
            formData.append('audio', file);
            const response = await fetch('/transcribe', { method: 'POST', body: formData });
            const result = await response.json();
            document.getElementById('output').innerText = result.transcript;
        }
    </script>


EOF

# Script de démarrage
echo '#!/bin/bash' > launch.sh
echo 'source env/bin/activate' >> launch.sh
echo 'cd app && python server.py' >> launch.sh
chmod +x launch.sh

echo "Installation terminée. Exécuter ./launch.sh pour démarrer."

Exécutez ce script pour installer le service, puis lancez-le avec ./launch.sh.

Intégration avec les robots de messagerie

Connexion à un robot Feishu

Pour automatiser la transcription dans Feishu, créez un robot personnalisé dans les paramètres du groupe et notez l'URL du webhook. Ensuite, développez un serveur intermédiaire :

# feishu_bot.py
from flask import Flask, request, jsonify
import requests
import tempfile
import os

service = Flask(__name__)
ASR_ENDPOINT = "http://localhost:8000/transcribe"
FEISHU_HOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/VOTRE_CLE"

@service.route('/feishu', methods=['POST'])
def handle_feishu():
    payload = request.json
    if payload.get("type") == "url_verification":
        return jsonify({"challenge": payload["challenge"]})
    if payload["event"]["message_type"] == "audio":
        audio_url = payload["event"]["audio_key"]
        audio_data = requests.get(audio_url).content
        temp_file = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False)
        temp_file.write(audio_data)
        temp_file.close()
        text = transcribe_file(temp_file.name)
        os.unlink(temp_file.name)
        send_response(text)
    return jsonify({"status": "ok"})

def transcribe_file(path):
    with open(path, 'rb') as f:
        resp = requests.post(ASR_ENDPOINT, files={"audio": f})
        return resp.json().get("transcript", "")

def send_response(message):
    requests.post(FEISHU_HOOK, json={
        "msg_type": "text",
        "content": {"text": f"Transcription : {message}"}
    })

if __name__ == '__main__':
    service.run(port=5000)

Configurez l'événement im.message.receive_v1 dans la plateforme Feishu pour rediriger vers ce serveur.

Intégration avec WeCom

Pour WeCom, le processus est similaire mais utilise l'API webhook du groupe :

# wecom_integration.py
from flask import Flask, request, jsonify
import requests
import os
from werkzeug.utils import secure_filename

app = Flask(__name__)
ASR_API = "http://localhost:8000/transcribe"
WECOM_WEBHOOK = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=VOTRE_CLE"

@app.route('/wecom', methods=['POST'])
def wecom_handler():
    data = request.json
    if data.get('msgtype') == 'voice':
        media_id = data['voice']['media_id']
        # Téléchargement du fichier audio via l'API WeCom
        audio_path = retrieve_audio(media_id)
        transcript = transcribe_audio(audio_path)
        os.remove(audio_path)
        post_to_wecom(transcript)
    return jsonify({"code": 0})

def retrieve_audio(identifier):
    # Logique pour récupérer le média depuis WeCom
    # Implémentation dépendante de l'access_token
    return "/tmp/temp_audio.mp3"

def transcribe_audio(file):
    with open(file, 'rb') as f:
        response = requests.post(ASR_API, files={"audio": f})
        return response.json().get("transcript", "")

def post_to_wecom(text):
    requests.post(WECOM_WEBHOOK, json={
        "msgtype": "text",
        "text": {"content": f"Résultat : {text}"}
    })

if __name__ == '__main__':
    app.run(port=5001)

Alternative simple : utiliser l'upload direct de fichiers via une route dédiée dans l'application pour les équipes sans accès API avancé.

Applications et optimisations

Cas d'utilisation pratiques

La transcription automatique peut être appliquée à :

• Comptes-rendus de réunions : enregistrement et conversion en texte pour une analyse rapide.
• Service client : transcription des appels pour la formation et l'amélioration de la qualité.

Améliorations de performance

Pour augmenter l'efficacité :

# Activation GPU si disponible
import torch
if torch.cuda.is_available():
    model = model.to('cuda')

# Traitement par lots
def batch_transcribe(files):
    return [transcribe(f) for f in files]

# Gestion asynchrone
import asyncio
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=4)

async def async_transcribe(path):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(executor, transcribe, path)

Robustesse et surveillance

Implémentez des mécanismes de rejeu et de journalisation :

import logging
import time

logging.basicConfig(level=logging.INFO, filename='asr.log')
logger = logging.getLogger(__name__)

def retry_operation(func, attempts=3):
    for i in range(attempts):
        try:
            return func()
        except Exception as e:
            logger.warning(f"Tentative {i+1} échouée: {e}")
            time.sleep(2 ** i)
    raise Exception("Échec après plusieurs tentatives")