Pipeline de Diarisation Speaker Temps Reel : Silero VAD + pyannote + Voxtral

Le Probleme : Qui Parle et Que Dit-il, en Temps Reel ?

La transcription audio est un probleme resolu. Whisper, Voxtral, Deepgram — les modeles STT actuels sont excellents. Mais savoir qui parle en meme temps qu’on transcrit, en streaming, avec une latence acceptable pour une application interactive — c’est un tout autre defi.

Selon les benchmarks du dataset CALLHOME, les meilleurs systemes de diarisation atteignent 5-8% de DER (Diarization Error Rate) en conditions de laboratoire. En conditions reelles — bruit ambiant, chevauchements de parole, micro de telephone — le DER monte facilement a 15-25% pour les approches naives.

Nous avons construit un pipeline complet de diarisation speaker en temps reel qui combine trois briques :

1. Silero VAD pour la detection d’activite vocale (quand quelqu’un parle)
2. pyannote.audio pour l’identification du locuteur (qui parle)
3. Voxtral (Mistral) pour la transcription (ce qui est dit)

Le pipeline fonctionne en streaming bidirectionnel : l’audio entre en continu, et les segments transcrits et attribues sortent au fil de l’eau.

Architecture Globale

Microphone / Fichier audio
        |
        v
+---------------------+
|   SileroVAD         |  -- Detecte les segments de parole
|   (chunks 512 ms)   |     Filtre les silences
+---------------------+
        |  (audio_chunk: np.ndarray)
        v
+---------------------+
|  SpeakerDiarizer    |  -- Extrait embeddings 512D
|  (pyannote.audio)   |     Compare avec reference
+---------------------+
        |  (speaker_id: str, audio: bytes)
        v
+---------------------+
|   VoxtralSTT        |  -- Transcrit le segment
|   (Mistral API)     |     Cache MD5 par segment
+---------------------+
        |
        v
   [{speaker, text, start, end}]

Cette architecture en couches permet de remplacer chaque composant independamment. Silero VAD peut etre remplace par WebRTC VAD. pyannote par un autre modele d’embeddings. Voxtral par Whisper local. L’interface entre composants est simple : np.ndarray audio + metadonnees de timing.

Silero VAD : Le Gardien du Silence

Pourquoi un VAD Avant Tout

Envoyer de l’audio continu a un modele STT est un gaspillage. Le silence, les bruits de fond, les manipulations de materiel — tout ca genere des transcriptions vides ou du bruit. Le VAD filtre en amont :

• Reduction de 60-80% du volume audio envoye au STT
• Elimination des faux segments (claquements, bruits de fond)
• Delimitation naturelle des tours de parole pour la diarisation

L’Implementation

class SileroVAD:
    def __init__(
        self,
        threshold: float = 0.5,
        sample_rate: int = 16000,
        silence_threshold: int = 62
    ):
        self.threshold = threshold
        self.sample_rate = sample_rate
        self.silence_threshold = silence_threshold

        # Charger le modele Silero VAD (~1 Mo, LSTM leger)
        self.model, self.utils = torch.hub.load(
            repo_or_dir='snakers4/silero-vad',
            model='silero_vad',
            force_reload=False,
            trust_repo=True,
            verbose=False
        )
        self.reset()

Le silence_threshold de 62 chunks est un parametre critique. Avec des chunks de 512 samples a 16kHz, chaque chunk fait 32ms. 62 chunks = environ 2 secondes de silence avant de considerer la parole terminee. C’est un compromis entre reactivite et robustesse aux pauses naturelles dans la conversation.

Le Traitement Chunk par Chunk

Le coeur du VAD est la methode process_chunk qui traite l’audio en morceaux de 512 samples :

def process_chunk(self, audio_chunk: np.ndarray) -> Tuple[bool, Optional[np.ndarray]]:
    # Normaliser en float32
    if audio_chunk.dtype == np.int16:
        audio_chunk = audio_chunk.astype(np.float32) / 32768.0

    # Silero attend des chunks de taille fixe
    chunk_size = 512 if self.sample_rate == 16000 else 256

    for i in range(0, len(audio_chunk), chunk_size):
        mini_chunk = audio_chunk[i:i + chunk_size]

        if len(mini_chunk) < chunk_size:
            mini_chunk = np.pad(
                mini_chunk, (0, chunk_size - len(mini_chunk))
            )

        audio_tensor = torch.from_numpy(mini_chunk)
        speech_prob = self.model(audio_tensor, self.sample_rate).item()

        if speech_prob > self.threshold:
            self.speech_detected = True
            self.silence_frames = 0
        else:
            if self.speech_detected:
                self.silence_frames += 1

    # Accumuler l'audio pendant la parole
    if self.speech_detected:
        self.audio_buffer.append(audio_chunk)

    # Si assez de silence apres parole -> termine
    if self.speech_detected and self.silence_frames > self.silence_threshold:
        if len(self.audio_buffer) > 0:
            full_audio = np.concatenate(self.audio_buffer)
            self.reset()
            return (True, full_audio)

    return (False, None)

Le retour est un tuple : (True, audio_complet) quand un segment de parole est termine, (False, None) sinon. Le buffer accumule tous les chunks pendant la parole et les concatene a la fin.

Points Techniques Importants

1. Padding des chunks courts : le dernier mini-chunk est complete par des zeros si necessaire. Silero ne tolere pas les tailles non standard.
2. Reset des etats : self.model.reset_states() reinitialise les etats internes du RNN de Silero entre chaque segment.
3. Le seuil de 0.5 est le defaut recommande. En pratique, on peut descendre a 0.3 pour les voix faibles, mais ca augmente les faux positifs.

Pourquoi Pas WebRTC VAD ?

Silero VAD est un modele neuronal (LSTM). WebRTC VAD est purement statistique (analyse de frequence). En environnement bruyant (air conditionne, clavier, bruits de fond), Silero est nettement plus robuste. Le cout : Silero necessite PyTorch (~500 Mo de dependances), WebRTC est tres leger. Pour un pipeline qui utilise deja PyTorch pour pyannote, le surcout est nul.

AudioChunker : Le Decoupeur de Fichiers

Pour le traitement de fichiers audio (pas du streaming), AudioChunker wrappe le VAD :

@dataclass
class AudioChunk:
    start: float      # Start time in seconds
    end: float        # End time in seconds
    audio_data: torch.Tensor
    sample_rate: int

    @property
    def duration(self) -> float:
        return self.end - self.start


class AudioChunker:
    def __init__(self, vad_threshold: float = 0.5):
        self.vad_threshold = vad_threshold
        self._vad = None  # Lazy initialization

    @property
    def vad(self):
        if self._vad is None:
            from src.voice.vad import SileroVAD
            self._vad = SileroVAD(threshold=self.vad_threshold)
        return self._vad

    def chunk_audio_file(
        self,
        audio_path: str,
        min_chunk_duration: float = 0.5,
        max_silence_duration: float = 0.5,
    ) -> List[AudioChunk]:
        # Decoupe le fichier en segments de parole
        # ...

L’initialisation paresseuse du VAD (_vad = None + @property) evite le torch.hub.load() au moment de l’import. C’est une source connue d’erreurs “broken pipe” dans les subprocesses Python : charger PyTorch dans un subprocess avant le fork provoque des problemes de file descriptors.

SpeakerDiarizer : Identifier Qui Parle

Le Principe : Embeddings + Similarite Cosinus

La diarisation repose sur un concept simple : chaque voix a une “empreinte” (embedding) de 512 dimensions. On compare l’embedding du segment courant avec une reference connue :

class SpeakerDiarizer:
    def __init__(
        self,
        hf_token: str,
        reference_audio: Optional[str] = None,
        similarity_threshold: float = 0.75,
    ):
        self.similarity_threshold = similarity_threshold
        self.vet_embedding = None

        # Authentification HuggingFace
        login(token=hf_token, add_to_git_credential=False)

        # Charger le modele pyannote embedding
        model = Model.from_pretrained("pyannote/embedding")
        self.inference = Inference(model)

Le modele pyannote/embedding est un reseau de neurones entraine pour produire des embeddings speaker-discriminants. Deux segments de la meme personne auront des embeddings proches (similarite cosinus > 0.75), deux personnes differentes seront distantes.

Le Probleme du torch.load en PyTorch 2.6+

Un piege classique avec pyannote : a partir de PyTorch 2.6, torch.load utilise weights_only=True par defaut. pyannote n’est pas compatible. Le fix applique dans le projet :

# Patch pour PyTorch 2.6+
_original_torch_load = torch.load

def _patched_torch_load(*args, **kwargs):
    kwargs['weights_only'] = False
    return _original_torch_load(*args, **kwargs)

torch.load = _patched_torch_load

# Aussi patcher lightning (utilise par pyannote internement)
import lightning.fabric.utilities.cloud_io as cloud_io
cloud_io.torch.load = _patched_torch_load

C’est du monkey-patching. Ca casse la securite de weights_only=True (prevention d’execution de code via pickles malveillants). En production, verifiez la provenance de vos modeles.

Extraction d’Embedding avec Padding

Les segments courts (<500ms) produisent des embeddings instables. Les acquiescements (“oui”, “d’accord”) durent 200-300ms. La solution : repeter l’audio jusqu’a atteindre une duree minimale :

def _pad_audio_for_embedding(
    self, waveform, sample_rate, min_duration=5.0
):
    current_duration = waveform.shape[1] / sample_rate

    if current_duration >= min_duration:
        return waveform

    min_samples = int(min_duration * sample_rate)
    current_samples = waveform.shape[1]
    num_repeats = (
        (min_samples + current_samples - 1) // current_samples
    )

    repeated_waveform = waveform.repeat(1, num_repeats)
    return repeated_waveform[:, :min_samples]

5 secondes de duree minimale, c’est beaucoup. Mais pyannote a besoin de suffisamment de signal pour moyenner les features temporelles. La repetition (plutot que le padding par silence) preserve les caracteristiques spectrales du locuteur. Un segment de 200ms repete 25 fois donne un embedding plus stable qu’un segment de 200ms brut.

L’Extraction Proprement Dite

def _extract_embedding(
    self, audio_input, start_time=None, end_time=None,
    sample_rate=None
):
    if isinstance(audio_input, torch.Tensor):
        waveform = audio_input
        if sample_rate is None:
            sample_rate = 16000
    else:
        waveform, sample_rate = torchaudio.load(audio_input)

    # Extraire le segment temporel si specifie
    if start_time is not None or end_time is not None:
        start_idx = int((start_time or 0.0) * sample_rate)
        end_idx = int(
            (end_time or (waveform.shape[1] / sample_rate))
            * sample_rate
        )
        waveform = waveform[:, start_idx:end_idx]

    # Padding obligatoire
    waveform = self._pad_audio_for_embedding(
        waveform, sample_rate, min_duration=5.0
    )

    # Inference pyannote -- accepte tensor ou fichier
    audio_dict = {
        "waveform": waveform,
        "sample_rate": sample_rate
    }
    embedding_feature = self.inference(audio_dict)

    # Moyenne sur les frames temporelles -> vecteur 512D
    embedding = embedding_feature.data.mean(axis=0)
    return embedding

L’API Inference de pyannote accepte un dictionnaire {waveform, sample_rate}. Cette forme evite les problemes d’arrondi lies aux sauvegardes temporaires en WAV. Le resultat est un vecteur de 512 dimensions.

Identification du Locuteur

def identify_speaker(self, audio_chunk, sample_rate=None):
    chunk_embedding = self._extract_embedding(
        audio_chunk, sample_rate=sample_rate
    )

    if self.vet_embedding is None:
        return {"speaker": "Inconnu", "similarity": 0.0}

    similarity = 1 - cosine(
        self.vet_embedding, chunk_embedding
    )

    if similarity >= self.similarity_threshold:
        return {
            "speaker": "Speaker_A",
            "similarity": float(similarity)
        }
    else:
        return {
            "speaker": "Speaker_B",
            "similarity": float(similarity)
        }

Le seuil de 0.75 est empirique. Pour des voix similaires (meme genre, meme tranche d’age), il faut descendre a 0.65-0.70. Pour des voix tres differentes (homme/femme), 0.80 suffit.

Reference par Fichier Audio

On fournit un echantillon de 5-10 secondes du locuteur principal :

def set_veterinaire_reference(self, audio_path: str):
    if not Path(audio_path).exists():
        raise FileNotFoundError(
            f"Reference audio not found: {audio_path}"
        )

    self.vet_embedding = self._extract_embedding(audio_path)

C’est le calibrage initial. Sans reference, le systeme ne peut pas identifier qui est qui — il faudrait alors passer en mode clustering non supervise (k-means sur les embeddings).

VoxtralSTT : La Transcription via Mistral

Le Client

class VoxtralSTT:
    def __init__(
        self,
        api_key: str = None,
        model: str = "voxtral-mini-latest"
    ):
        self.api_key = api_key or os.getenv("MISTRAL_API_KEY")
        self.client = Mistral(api_key=self.api_key)
        self.default_model = model

    def transcribe(
        self,
        audio_data: bytes,
        language: str = "fr",
        model: str = None,
        use_cache: bool = True
    ) -> str:
        # Transcription via API Mistral
        # ...

Voxtral voxtral-mini-latest est la version legere du modele audio de Mistral. Le choix d’une API cloud plutot que Whisper local est delibere : la qualite de transcription en francais est superieure, et la latence reseau (~200-500ms) est acceptable pour du quasi-temps-reel.

Le Cache Audio

Chaque transcription est mise en cache par hash MD5 de l’audio :

def _get_audio_hash(audio_data: bytes) -> str:
    wav_io = io.BytesIO(audio_data)
    waveform, sample_rate = torchaudio.load(wav_io)

    # Hasher les samples bruts + sample rate
    # (independant du format WAV qui peut varier)
    content = waveform.numpy().tobytes() + str(sample_rate).encode()
    return hashlib.md5(content).hexdigest()

On ne hashe pas les bytes WAV bruts, mais les samples audio + sample rate. Cela rend le hash independant de la version de torchaudio qui peut encoder differemment les headers WAV.

Le cache est un fichier CSV thread-safe :

def _save_to_cache(audio_hash, transcription):
    CACHE_DIR.mkdir(parents=True, exist_ok=True)
    with _cache_lock:
        with open(CACHE_FILE, 'a', encoding='utf-8', newline='') as f:
            writer = csv.DictWriter(
                f, fieldnames=['hash', 'transcription', 'timestamp']
            )
            writer.writerow({
                'hash': audio_hash,
                'transcription': transcription,
                'timestamp': datetime.now().isoformat()
            })

Append-only, lisible par un humain. Pas besoin de SQLite pour un cache de transcriptions.

Voxtral vs Whisper : Le Compromis

Pour un deploiement en ligne avec connexion stable, Voxtral gagne. Pour un deploiement embarque ou hors-ligne, Whisper s’impose.

ConsultationListener : L’Orchestrateur

Le ConsultationListener assemble toutes les briques et ajoute la detection d’intentions par LLM :

class ConsultationListener:
    def __init__(self, mistral_client: Mistral, stt_service):
        self.mistral_client = mistral_client
        self.stt_service = stt_service

        self.is_listening = False
        self.dialogue_history: List[str] = []
        self.detected_actions: Queue[DetectedAction] = Queue()

Le flux complet :

1. Le micro capture l’audio en continu
2. SileroVAD.process_chunk() detecte la fin d’un segment de parole
3. SpeakerDiarizer.identify_speaker() identifie le locuteur
4. VoxtralSTT.transcribe() transcrit le segment
5. Le LLM Mistral analyse le dialogue pour detecter des intentions

Detection d’Intentions par LLM

Au-dela de la transcription, le pipeline detecte des actions implicites dans le dialogue. Le prompt est structure pour extraire des actions specifiques :

@dataclass
class DetectedAction:
    action_type: str     # Ex: "create_prescription"
    parameters: Dict[str, Any]
    confidence: float    # 0.0 to 1.0
    trigger_context: str # Citation du dialogue
    description: str     # Description lisible pour l'UI
    detected_at: datetime = field(default_factory=datetime.now)
    status: str = "pending"  # pending, confirmed, rejected
    segment_id: Optional[int] = None

Le prompt de detection gere explicitement les erreurs de transcription audio :

IMPLICIT_DETECTION_PROMPT = """Tu analyses une conversation pour
detecter des actions.

NORMALISATION DES NOMS:
La transcription audio peut contenir des erreurs. Normalise:
- "Serenia", "Cerenia", "serenia" -> "Cerenia"
- "Dermipred", "dermipred" -> "Dermipred"

DIALOGUE RECENT:
{dialogue}

DERNIER SEGMENT (locuteur: {speaker}):
{last_segment}
"""

Les reponses LLM sont aussi mises en cache par hash MD5 du prompt, evitant les appels redondants en mode replay.

Latences Reelles Mesurees

La latence dominante est le silence threshold (2 secondes d’attente). En pratique, c’est acceptable car les locuteurs humains font naturellement des pauses de 1-2 secondes entre les phrases. Reduire le seuil a 30 chunks (~1 seconde) ameliore la reactivite mais fragmente les phrases longues.

Empreinte Memoire du Pipeline

Le pipeline entier tient sur un GPU de 2 Go. C’est frugal compare a Whisper large (6 Go) ou NeMo (8 Go+).

Les Choix d’Architecture et Leurs Consequences

VAD Avant Diarisation : Le Bon Choix

L’alternative serait d’envoyer tout l’audio au diarizer. Mais :

• pyannote est plus lourd que Silero (GPU vs CPU)
• Le VAD en amont reduit le volume de 60-80%
• Silero fonctionne en streaming chunk-par-chunk ; pyannote fonctionne mieux sur des segments complets

Embeddings + Reference vs Clustering Non Supervise

Notre approche utilise une reference connue et compare par similarite cosinus. L’alternative serait un clustering non supervise (k-means, HDBSCAN) sur les embeddings, mais :

• Le clustering necessite de traiter tout l’audio d’abord (pas compatible streaming)
• On connait le nombre de locuteurs a l’avance (2)
• La reference rend l’identification deterministe et instantanee

Pour des reunions a N locuteurs sans reference, le clustering reste la meilleure approche non supervisee.

Cache CSV vs Base de Donnees

Le choix d’un cache CSV plutot que SQLite est delibere :

• Lisible par un humain sans outil
• Append-only, pas de corruption possible
• Thread-safe avec un simple lock
• Pour un cache de quelques milliers d’entrees, les performances sont largement suffisantes

Pieges en Production

Le Token HuggingFace

pyannote.audio necessite un token HuggingFace avec acces au modele pyannote/embedding. C’est gratuit mais requiert une demande d’acces acceptee. En production, stockez le token dans une variable d’environnement, jamais dans le code.

Drift d’Embedding sur les Longues Sessions

Sur des sessions de plus de 30 minutes, la voix peut deriver (fatigue, hydratation). Mettre a jour l’embedding de reference toutes les 5 minutes en moyennant avec les segments recents classes avec haute confiance (similarite > 0.90) stabilise le systeme.

Chevauchement de Parole

Ni Silero VAD ni pyannote ne gerent nativement les chevauchements (deux personnes parlant simultanement). Un pre-traitement avec un separateur de sources (Demucs, SepFormer) peut etre insere entre le VAD et la diarisation pour les cas critiques.

Conclusion : Un Pipeline Viable en Production

Le trio Silero VAD + pyannote embeddings + Voxtral STT produit un pipeline de diarisation fonctionnel avec une latence de 3-4 secondes. C’est suffisant pour des applications interactives ou le feedback n’a pas besoin d’etre instantane.

Les forces : frugalite GPU (~500 Mo VRAM), streaming natif, identification deterministe par reference, cache intelligent pour le STT.

Les faiblesses : latence du silence threshold, dependance au cloud pour le STT, monkey-patching de torch.load pour pyannote.

Pour aller plus loin : VAD adaptatif (seuil de silence variable selon le contexte), embedding streaming (sans le padding 5s), et migration vers un STT local quand les performances en francais de Whisper v4 ou successeurs seront au niveau de Voxtral.

Le DER mesure sur des conversations a deux locuteurs en francais avec ce pipeline est de l’ordre de 10-12%, ce qui est competitif avec les solutions commerciales et suffisant pour l’exploitation directe des transcriptions.

Un projet similaire ? Contactez Loick Briot : contact@brio-novia.eu