scipio avatar

Learn AI Series (#97) - Speaker Recognition and Diarization

scipio

Published: 14 Jun 2026 › Updated: 14 Jun 2026Learn AI Series (#97) - Speaker Recognition and Diarization

Learn AI Series (#97) - Speaker Recognition and Diarization

Learn AI Series (#97) - Speaker Recognition and Diarization

variant-a-12-green.png

What will I learn

  • You will learn speaker verification: confirming "is this the person they claim to be?";
  • speaker identification: determining "who is speaking?" from a set of known speakers;
  • speaker embeddings: d-vectors and x-vectors that capture vocal identity in fixed-size vectors;
  • speaker diarization: segmenting multi-speaker audio into "who spoke when?";
  • voice activity detection (VAD): finding where speech occurs in audio;
  • building a complete speaker verification system from scratch using cosine similarity on Mel spectrograms;
  • combining ASR with diarization for speaker-attributed meeting transcripts;
  • evaluation metrics: EER for verification and DER for diarization.

Requirements

  • A working modern computer running macOS, Windows or Ubuntu;
  • An installed Python 3(.11+) distribution;
  • The ambition to learn AI and machine learning.

Difficulty

  • Beginner

Curriculum (of the Learn AI Series):

Learn AI Series (#97) - Speaker Recognition and Diarization

Solutions to Episode #96 Exercises

Exercise 1: Melody complexity analyzer -- categorize intervals (unison/step/skip/leap), compute pitch range, and measure contour complexity (direction changes per interval). Result: C major scale = 100% steps, zero contour changes. Twinkle Twinkle mixes steps and leaps with several direction changes. Random walk produces high contour complexity because each interval direction is essentially a coin flip.

Exercise 2: Chord progression generator with Markov transitions and consonance scoring. The V->I resolution appears as the strongest cadence -- the transition matrix gives V a 55% probability of resolving to I, reflecting how dominant-tonic resolution is the most fundamental harmonic motion. Major triads (I, IV, V) score higher consonance than the diminished triad (vii).

Exercise 3: Rhythmic pattern evaluator measuring IOI statistics, swing ratio, and tempo stability across metronome, swing, human drummer, and random patterns. The metronome has CV=0 (perfect regularity) and swing ratio=1.0. The swing pattern has ratio near 2.0. The human drummer has small but nonzero CV from Gaussian timing jitter. The random pattern has the highest CV and lowest tempo stability.

On to today's episode

Here we go! We've been deep in the audio AI domain over the past five episodes: how machines hear sound (#92), transcribing speech to text (#93), synthesizing speech from text (#94), classifying environmental sounds (#95), and generating music from scratch (#96). All of those dealt with understanding what's in the audio -- the words, the sounds, the melodies. Today we're shifting focus to understanding who is in the audio.

Speaker recognition is the audio equivalent of face recognition (episode #88). Just as FaceNet learns a compact embedding vector for each face, speaker recognition models learn embedding vectors for voices. Two audio clips from the same person produce similar vectors; clips from different people produce different vectors. The math is almost identical -- cosine similarity, angular margin losses, the whole apparatus we already built for faces. The difference is the input representation: spectrograms and Mel filterbanks instead of pixel arrays.

This is one of those topics where the practical applications are everywhere -- from "Hey Siri, is that actually me talking?" to meeting transcription systems that figure out which participant said what. The underlying techniques are elegant and connect back to nearly everything we've covered so far. Let's get into it ;-)

Speaker verification vs speaker identification

Before we write any code, let me clarify the two main tasks in speaker recognition. They sound similar but they're fundamentally different problems.

Speaker verification (SV) answers a yes/no question: "Is this the person they claim to be?" You have a stored voiceprint (an embedding vector from enrollment) and a new audio sample. Compute the similarity. If it's above a threshold -- accept. Below -- reject. This is a binary classification problem (same speaker or different speaker), and it's what banks use for phone authentication, what smart speakers use for personalization, and what forensic analysts use for voice matching.

Speaker identification (SI) answers a who question: "Which of these N known speakers is talking?" You have a database of enrolled speakers with their voiceprints. Given a new audio clip, find the closest match. This is an N-way classification problem, and it's what meeting transcription systems and surveillance systems use.

Both tasks rely on the same foundation: speaker embeddings -- fixed-dimensional vectors that capture the essential characteristics of a voice while being invariant to what's being said, background noise, and recording conditions. The quality of your embeddings determines the quality of everything downstream:

import numpy as np


class SpeakerTaskDemo:
    """Demonstrate the difference between
    speaker verification (1:1) and speaker
    identification (1:N)."""

    def __init__(self):
        self.rng = np.random.RandomState(42)

    def simulate_embeddings(self,
                             n_speakers=5,
                             dim=192,
                             n_samples=3):
        """Create synthetic speaker
        embeddings with realistic
        intra/inter speaker distances."""
        speakers = {}
        for i in range(n_speakers):
            center = self.rng.randn(dim)
            center = center / np.linalg.norm(
                center)
            samples = []
            for _ in range(n_samples):
                noise = self.rng.randn(
                    dim) * 0.15
                emb = center + noise
                emb = emb / np.linalg.norm(
                    emb)
                samples.append(emb)
            speakers[f"spk_{i}"] = samples
        return speakers

    def cosine_sim(self, a, b):
        return float(np.dot(a, b) / (
            np.linalg.norm(a)
            * np.linalg.norm(b)))

    def verification(self, speakers):
        """1:1 -- is this the claimed
        speaker?"""
        print("=== Speaker Verification ===")
        # Same speaker comparison
        s0 = speakers["spk_0"]
        same = self.cosine_sim(
            s0[0], s0[1])
        # Different speaker comparison
        diff = self.cosine_sim(
            s0[0],
            speakers["spk_1"][0])
        threshold = 0.7
        print(f"Same speaker sim: "
              f"{same:.3f} "
              f"({'ACCEPT' if same > threshold else 'REJECT'})")
        print(f"Diff speaker sim: "
              f"{diff:.3f} "
              f"({'ACCEPT' if diff > threshold else 'REJECT'})")
        print(f"Threshold: {threshold}")

    def identification(self, speakers):
        """1:N -- who is this?"""
        print("\n=== Speaker Identification ===")
        # Take a sample from spk_2
        query = speakers["spk_2"][2]
        # Compare against all enrolled
        # speakers (using first sample
        # as enrollment)
        print(f"Query: spk_2 sample #3")
        best_score = -1
        best_match = None
        for name, samples in (
                speakers.items()):
            sim = self.cosine_sim(
                query, samples[0])
            match = " <-- MATCH" if (
                sim > 0.7
                and sim > best_score) else ""
            if sim > best_score:
                best_score = sim
                best_match = name
            print(f"  vs {name}: "
                  f"{sim:.3f}{match}")
        print(f"Identified as: "
              f"{best_match} "
              f"(score: {best_score:.3f})")

    def run(self):
        speakers = (
            self.simulate_embeddings())
        self.verification(speakers)
        self.identification(speakers)

        # Show intra vs inter variance
        print("\n=== Distance Analysis ===")
        intra = []
        inter = []
        names = list(speakers.keys())
        for n in names:
            samps = speakers[n]
            for i in range(len(samps)):
                for j in range(
                        i+1, len(samps)):
                    intra.append(
                        self.cosine_sim(
                            samps[i],
                            samps[j]))
        for i in range(len(names)):
            for j in range(
                    i+1, len(names)):
                inter.append(
                    self.cosine_sim(
                        speakers[names[i]][0],
                        speakers[names[j]][0]))
        print(f"Intra-speaker sim: "
              f"{np.mean(intra):.3f} "
              f"(+/- {np.std(intra):.3f})")
        print(f"Inter-speaker sim: "
              f"{np.mean(inter):.3f} "
              f"(+/- {np.std(inter):.3f})")
        print(f"Separation gap: "
              f"{np.mean(intra) - np.mean(inter):.3f}")


demo = SpeakerTaskDemo()
demo.run()

The key insight from the distance analysis: for speaker recognition to work, the intra-speaker similarity (same person, different utterances) must be significantly higher than the inter-speaker similarity (different people). This is the same separation criterion that drives all metric learning (episode #63 on embeddings). The wider the gap between intra and inter distances, the easier it is to set a reliable threshold for verification or identification.

Speaker embeddings: d-vectors and x-vectors

A speaker embedding model takes a variable-length audio segment and produces a fixed-size vector that captures vocal identity. Two segments from the same speaker should produce similar vectors; segments from different speakers should produce distant vectors. This is conceptually identical to how FaceNet (episode #88) maps face images to embedding vectors -- same problem, different modality.

D-vectors (Variani et al., 2014; Wan et al., 2018) use the simplest possible approach: process Mel spectrogram frames through an LSTM (episode #49), then average the frame-level hidden states to get a single utterance-level vector:

import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np


class DVectorModel(nn.Module):
    """D-vector: LSTM-based speaker
    embedding extractor."""

    def __init__(self, n_mels=40,
                 hidden=256,
                 embed_dim=256,
                 n_layers=3):
        super().__init__()
        self.lstm = nn.LSTM(
            n_mels, hidden, n_layers,
            batch_first=True)
        self.proj = nn.Linear(
            hidden, embed_dim)

    def forward(self, mel_frames):
        """mel_frames: (batch, time,
        n_mels)"""
        output, _ = self.lstm(mel_frames)
        # Average pooling over time
        embedding = output.mean(dim=1)
        embedding = self.proj(embedding)
        return F.normalize(
            embedding, dim=1)


# Demo
model = DVectorModel()
# Simulate 2 utterances: 3 sec
# and 5 sec at ~100 frames/sec
utt_short = torch.randn(1, 300, 40)
utt_long = torch.randn(1, 500, 40)

emb_short = model(utt_short)
emb_long = model(utt_long)
print(f"Short utterance embedding: "
      f"{emb_short.shape}")
print(f"Long utterance embedding: "
      f"{emb_long.shape}")
print(f"Both produce {emb_short.shape[1]}"
      f"-dim vectors regardless of "
      f"input length")

total = sum(p.numel()
            for p in model.parameters())
print(f"Parameters: {total:,}")

The averaging over time is what makes d-vectors length-invariant -- a 2-second clip and a 10-second clip both produce a 256-dimensional vector. Simple and effective, but the averaging loses temporal information. A speaker who emphasizes certain sounds differently at the beginning versus end of an utterance -- that nuance gets averaged away.

X-vectors (Snyder et al., 2018) replaced the LSTM with a TDNN (Time-Delay Neural Network) -- essentially 1D convolutions at different temporal strides -- followed by a statistics pooling layer that computes both the mean AND standard deviation across time:

class XVectorModel(nn.Module):
    """X-vector: TDNN + statistics
    pooling for speaker embeddings."""

    def __init__(self, n_mels=40,
                 embed_dim=512):
        super().__init__()
        # TDNN layers with increasing
        # temporal context
        self.tdnn1 = nn.Conv1d(
            n_mels, 512, 5, padding=2)
        self.bn1 = nn.BatchNorm1d(512)
        self.tdnn2 = nn.Conv1d(
            512, 512, 3,
            dilation=2, padding=2)
        self.bn2 = nn.BatchNorm1d(512)
        self.tdnn3 = nn.Conv1d(
            512, 512, 3,
            dilation=3, padding=3)
        self.bn3 = nn.BatchNorm1d(512)
        self.tdnn4 = nn.Conv1d(
            512, 512, 1)
        self.bn4 = nn.BatchNorm1d(512)
        self.tdnn5 = nn.Conv1d(
            512, 1500, 1)
        self.bn5 = nn.BatchNorm1d(1500)

        # Stats pooling: mean + std
        # = 1500 * 2 = 3000 dim
        self.embed = nn.Linear(
            3000, embed_dim)

    def forward(self, mel):
        """mel: (batch, n_mels, time)"""
        x = F.relu(self.bn1(
            self.tdnn1(mel)))
        x = F.relu(self.bn2(
            self.tdnn2(x)))
        x = F.relu(self.bn3(
            self.tdnn3(x)))
        x = F.relu(self.bn4(
            self.tdnn4(x)))
        x = F.relu(self.bn5(
            self.tdnn5(x)))

        # Statistics pooling
        mean = x.mean(dim=2)
        std = x.std(dim=2)
        pooled = torch.cat(
            [mean, std], dim=1)

        embedding = self.embed(pooled)
        return F.normalize(
            embedding, dim=1)


model = XVectorModel()
mel = torch.randn(2, 40, 300)
emb = model(mel)
print(f"Input: {mel.shape}")
print(f"Output: {emb.shape}")
total = sum(p.numel()
            for p in model.parameters())
print(f"Parameters: {total:,}")
print(f"\nThe statistics pooling "
      f"captures BOTH the mean "
      f"spectral profile")
print(f"AND the variability -- "
      f"how much a speaker's voice "
      f"fluctuates over time.")

The statistics pooling is the key innovation here. By computing the standard deviation across time in addition to the mean, the model captures not just what the voice sounds like on average but how much it varies. Some speakers have very consistent vocal characteristics across an utterance; others modulate pitch and energy dramatically. That variation pattern is itself an identity signal. None the less, the dilation factors in the TDNN layers (2, 3) give later layers progressively wider temporal context without increasing the number of parameters -- same idea as dilated convolutions in WaveNet (episode #94).

Modern speaker embeddings: ECAPA-TDNN

Modern speaker embedding models like ECAPA-TDNN (Desplanques et al., 2020) improve on x-vectors with three additions: channel attention (Squeeze-and-Excitation blocks that let the network focus on the most informative frequency channels), residual connections (so gradients flow through deep networks), and multi-layer feature aggregation (combining features from multiple TDNN layers before pooling, rather than using only the last layer):

class ChannelAttention(nn.Module):
    """Squeeze-and-Excitation block."""
    def __init__(self, ch, r=8):
        super().__init__()
        self.pool = nn.AdaptiveAvgPool1d(1)
        self.fc = nn.Sequential(
            nn.Linear(ch, ch // r),
            nn.ReLU(),
            nn.Linear(ch // r, ch),
            nn.Sigmoid())

    def forward(self, x):
        w = self.pool(x).squeeze(-1)
        return x * self.fc(w).unsqueeze(-1)


class AttentiveStatsPool(nn.Module):
    """Attention-weighted stats pool."""
    def __init__(self, ch):
        super().__init__()
        self.attn = nn.Sequential(
            nn.Conv1d(ch, 128, 1),
            nn.ReLU(),
            nn.Conv1d(128, ch, 1),
            nn.Softmax(dim=2))

    def forward(self, x):
        w = self.attn(x)
        mean = (x * w).sum(dim=2)
        var = ((x**2*w).sum(dim=2)
               - mean**2)
        std = torch.sqrt(
            var.clamp(min=1e-9))
        return torch.cat(
            [mean, std], dim=1)


class SimpleECAPATDNN(nn.Module):
    def __init__(self, n_mels=80,
                 embed_dim=192):
        super().__init__()
        self.layer1 = nn.Conv1d(
            n_mels, 512, 5, padding=2)
        self.bn1 = nn.BatchNorm1d(512)
        self.se1 = ChannelAttention(512)
        self.layer2 = nn.Conv1d(
            512, 512, 3, dilation=2,
            padding=2)
        self.bn2 = nn.BatchNorm1d(512)
        self.se2 = ChannelAttention(512)
        self.layer3 = nn.Conv1d(
            512, 512, 3, dilation=3,
            padding=3)
        self.bn3 = nn.BatchNorm1d(512)
        self.se3 = ChannelAttention(512)
        self.mfa = nn.Conv1d(
            512*3, 1536, 1)
        self.bn_mfa = nn.BatchNorm1d(1536)
        self.asp = AttentiveStatsPool(1536)
        self.embed = nn.Linear(
            1536*2, embed_dim)

    def forward(self, mel):
        x1 = self.se1(F.relu(
            self.bn1(self.layer1(mel))))
        x2 = self.se2(F.relu(
            self.bn2(self.layer2(x1))))
        x3 = self.se3(F.relu(
            self.bn3(self.layer3(x2))))
        cat = F.relu(self.bn_mfa(
            self.mfa(torch.cat(
                [x1, x2, x3], dim=1))))
        pooled = self.asp(cat)
        return F.normalize(
            self.embed(pooled), dim=1)


model = SimpleECAPATDNN()
mel = torch.randn(2, 80, 300)
emb = model(mel)
total = sum(p.numel()
            for p in model.parameters())
print(f"Input: {mel.shape}")
print(f"Output: {emb.shape}")
print(f"Parameters: {total:,}")

The attentive statistics pooling learns which frames are most informative for speaker identity. Frames with clear speech get high attention; noisy or silent frames get low attention. This makes the model robust to variable-length inputs with noisy segments.

These models are trained with angular margin losses like AAM-Softmax -- the same ArcFace loss from episode #88 for face recognition. The loss pushes same-speaker embeddings closer on the unit hypersphere while enforcing angular margin between different speakers. Training datasets like VoxCeleb (7,000+ celebrities, 1M+ utterances from YouTube) provide the scale needed to generalize.

Voice Activity Detection: where is speech?

Before you can identify speakers, you need to know where speech actually occurs. Voice Activity Detection (VAD) classifies each audio frame as speech or non-speech. It sounds trivial -- and for clean studio audio, it practically is. But in real-world recordings with background music, air conditioning hum, keyboard typing, and other non-speech sounds, VAD becomes a serious preprocessing challenge:

import numpy as np


class SimpleVAD:
    """Energy-based Voice Activity
    Detection with adaptive threshold."""

    def __init__(self, sr=16000,
                 frame_ms=25,
                 hop_ms=10):
        self.sr = sr
        self.frame_len = int(
            sr * frame_ms / 1000)
        self.hop_len = int(
            sr * hop_ms / 1000)

    def compute_energy(self, audio):
        """Frame-level energy in dB."""
        n_frames = (
            (len(audio) - self.frame_len)
            // self.hop_len + 1)
        energy = np.zeros(n_frames)
        for i in range(n_frames):
            start = i * self.hop_len
            frame = audio[
                start:start+self.frame_len]
            rms = np.sqrt(
                np.mean(frame ** 2))
            energy[i] = 20 * np.log10(
                max(rms, 1e-10))
        return energy

    def detect(self, audio,
               energy_thresh_db=-30,
               min_speech_ms=200,
               min_silence_ms=100):
        """Detect speech segments."""
        energy = self.compute_energy(
            audio)
        speech = energy > energy_thresh_db

        hop_ms = (self.hop_len / self.sr
                  * 1000)
        min_speech = int(
            min_speech_ms / hop_ms)
        min_silence = int(
            min_silence_ms / hop_ms)

        # Fill short gaps
        result = speech.copy()
        in_gap = False
        gap_start = 0
        for i in range(len(result)):
            if not result[i] and (
                    not in_gap):
                gap_start = i
                in_gap = True
            elif result[i] and in_gap:
                if i - gap_start < (
                        min_silence):
                    result[
                        gap_start:i] = True
                in_gap = False

        # Remove short segments
        in_seg = False
        seg_start = 0
        for i in range(len(result)):
            if result[i] and not in_seg:
                seg_start = i
                in_seg = True
            elif not result[i] and in_seg:
                if i - seg_start < (
                        min_speech):
                    result[
                        seg_start:i] = False
                in_seg = False

        return result, energy

    def extract_segments(self, vad):
        """Convert frame-level VAD to
        time segments."""
        segments = []
        in_seg = False
        start = 0
        for i in range(len(vad)):
            if vad[i] and not in_seg:
                start = i
                in_seg = True
            elif not vad[i] and in_seg:
                t_start = (
                    start * self.hop_len
                    / self.sr)
                t_end = (
                    i * self.hop_len
                    / self.sr)
                segments.append(
                    (t_start, t_end))
                in_seg = False
        if in_seg:
            t_start = (
                start * self.hop_len
                / self.sr)
            t_end = (
                len(vad) * self.hop_len
                / self.sr)
            segments.append(
                (t_start, t_end))
        return segments

    def run(self):
        rng = np.random.RandomState(42)
        sr = self.sr
        dur = 5.0
        t = np.arange(int(sr * dur)) / sr
        audio = rng.randn(len(t)) * 0.001

        # Two speech segments
        for start, end, f0 in [
                (0.5, 1.8, 150),
                (2.5, 4.0, 180)]:
            s = slice(int(start*sr),
                      int(end*sr))
            audio[s] += 0.3 * np.sin(
                2*np.pi*f0*t[s])
            audio[s] += 0.15 * np.sin(
                2*np.pi*f0*2*t[s])

        vad, energy = self.detect(audio)
        segs = self.extract_segments(vad)
        print(f"Audio: {dur}s")
        print(f"Detected speech segments:")
        for start, end in segs:
            print(f"  {start:.2f}s - "
                  f"{end:.2f}s")
        print(f"Expected: ~0.5-1.8s "
              f"and ~2.5-4.0s")


vad = SimpleVAD()
vad.run()

VAD might seem like a solved problem, but it's absolutly critical for downstream quality. Without it, silence and background noise get processed as speech, which wastes compute and (worse) contaminates speaker embeddings with non-speech features. Production systems like Silero VAD use small neural networks (just a few megabytes) trained on massive amounts of audio with precise speech/non-speech labels. They run in real-time on a CPU, processing hours of audio in seconds.

Building a speaker verification system from scratch

Let's put the pieces together and build a complete speaker verification system. We'll synthesize multi-speaker audio, extract simple spectral features, compute embeddings, and verify speakers using cosine similarity:

import numpy as np


class SpeakerVerificationSystem:
    """Complete speaker verification
    pipeline from scratch."""

    def __init__(self, sr=16000,
                 n_mels=40,
                 embed_dim=64):
        self.sr = sr
        self.n_mels = n_mels
        self.embed_dim = embed_dim
        self.rng = np.random.RandomState(
            42)

    def synthesize_speaker(self,
                            f0_range,
                            formants,
                            duration=2.0,
                            n_utterances=5):
        """Synthetic speech-like audio."""
        utterances = []
        for _ in range(n_utterances):
            n = int(self.sr * duration)
            t = np.arange(n) / self.sr
            f0 = self.rng.uniform(
                *f0_range)
            signal = np.zeros(n)
            for f, a in formants:
                signal += a * np.sin(
                    2*np.pi*f*t)
            for h in range(2, 6):
                signal += (0.3/h) * np.sin(
                    2*np.pi*f0*h*t)
            signal *= (0.5 + 0.5 * np.sin(
                2*np.pi*4.5*t))
            signal += self.rng.randn(
                n) * 0.02
            utterances.append(signal)
        return utterances

    def mel_spectrogram(self, audio):
        """Simplified Mel spectrogram."""
        n_fft = 512
        hop = 160
        n_frames = (
            len(audio) - n_fft) // hop
        spec = np.zeros(
            (self.n_mels, n_frames))
        window = np.hanning(n_fft)
        for i in range(n_frames):
            start = i * hop
            frame = audio[
                start:start + n_fft]
            fft_mag = np.abs(
                np.fft.rfft(
                    frame * window))
            # Simple Mel binning
            n_bins = len(fft_mag)
            mel_bins = np.linspace(
                0, n_bins,
                self.n_mels + 1,
                dtype=int)
            for m in range(self.n_mels):
                lo = mel_bins[m]
                hi = mel_bins[m + 1]
                if hi > lo:
                    spec[m, i] = np.mean(
                        fft_mag[lo:hi])
        return np.log(spec + 1e-9)

    def extract_embedding(self, mel):
        """Simple embedding: mean + std
        of mel features, projected."""
        mean = np.mean(mel, axis=1)
        std = np.std(mel, axis=1)
        features = np.concatenate(
            [mean, std])
        # Random projection (in a real
        # system this is a learned
        # network)
        proj = self.rng.randn(
            self.embed_dim,
            len(features)) * 0.1
        emb = proj @ features
        norm = np.linalg.norm(emb)
        return emb / max(norm, 1e-9)

    def cosine_sim(self, a, b):
        return float(
            np.dot(a, b)
            / (np.linalg.norm(a)
               * np.linalg.norm(b)))

    def run(self):
        speaker_configs = {
            'Alice': {'f0_range': (180, 220),
                'formants': [(800,.4),
                    (1200,.3), (2500,.15)]},
            'Bob': {'f0_range': (100, 130),
                'formants': [(600,.5),
                    (1000,.25), (2200,.1)]},
            'Carol': {'f0_range': (200, 240),
                'formants': [(850,.35),
                    (1300,.28), (2700,.12)]},
            'Dave': {'f0_range': (90, 115),
                'formants': [(550,.45),
                    (950,.2), (2000,.1)]},
        }

        # Generate utterances
        speakers = {}
        for name, cfg in (
                speaker_configs.items()):
            utts = self.synthesize_speaker(
                cfg['f0_range'],
                cfg['formants'])
            speakers[name] = utts

        # Extract embeddings
        embeddings = {}
        for name, utts in (
                speakers.items()):
            embs = []
            for utt in utts:
                mel = self.mel_spectrogram(
                    utt)
                emb = self.extract_embedding(
                    mel)
                embs.append(emb)
            embeddings[name] = embs

        # Enrollment: use first 3
        # utterances per speaker
        enrolled = {}
        for name, embs in (
                embeddings.items()):
            enrolled[name] = np.mean(
                embs[:3], axis=0)
            enrolled[name] /= (
                np.linalg.norm(
                    enrolled[name]))

        # Verification trials
        print("=== Verification Trials ===")
        print(f"{'Claim':>8} {'Actual':>8}"
              f" {'Score':>7}"
              f" {'Result':>8}")
        print("-" * 36)

        trials = [
            ('Alice', 'Alice', 3),
            ('Alice', 'Alice', 4),
            ('Bob', 'Bob', 3),
            ('Alice', 'Carol', 3),
            ('Bob', 'Dave', 4),
            ('Carol', 'Bob', 3),
        ]
        threshold = 0.6
        for claim, actual, idx in trials:
            test_emb = embeddings[
                actual][idx]
            score = self.cosine_sim(
                enrolled[claim], test_emb)
            accept = score > threshold
            correct = (
                (accept and
                 claim == actual)
                or (not accept and
                    claim != actual))
            print(f"{claim:>8} {actual:>8}"
                  f" {score:>7.3f}"
                  f" {'ACCEPT' if accept else 'REJECT':>8}"
                  f" {'OK' if correct else 'ERR'}")


system = SpeakerVerificationSystem()
system.run()

In a real system, the "random projection" would be replaced by a trained neural network (d-vector LSTM or ECAPA-TDNN), but the pipeline structure is identical: synthesize/capture audio, extract Mel spectrogram, compute embedding, compare with cosine similarity. The threshold (0.6 in our demo) is the critical operating parameter -- too low and impostors get accepted (high false acceptance rate), too high and legitimate speakers get rejected (high false rejection rate).

Speaker diarization: who spoke when?

Diarization segments a multi-speaker recording into speaker-homogenous regions. The output is a timeline showing which speaker is talking at each moment, without necessarily knowing who the speakers are (they're labeled as Speaker 0, Speaker 1, etc.). Think meeting recordings, podcast transcriptions, or courtroom audio analysis.

The traditional diarization pipeline looks like this:

Audio -> VAD -> Segmentation -> Embeddings -> Clustering -> Speaker labels

Let's build one:

import numpy as np


class SpeakerDiarizer:
    """Pipeline-based speaker
    diarization system."""

    def __init__(self, sr=16000,
                 seg_dur=1.5,
                 seg_hop=0.75):
        self.sr = sr
        self.seg_dur = seg_dur
        self.seg_hop = seg_hop
        self.rng = np.random.RandomState(
            42)

    def generate_meeting(self):
        """Simulate a 20s meeting
        with 3 speakers."""
        dur = 20.0
        n = int(self.sr * dur)
        audio = self.rng.randn(n) * 0.005
        profiles = {
            0: (120, [500, 1000, 2200]),
            1: (200, [800, 1200, 2600]),
            2: (150, [650, 1100, 2400])}
        timeline = [
            (0, 0.5, 3.2), (1, 3.5, 6.0),
            (0, 6.3, 8.5), (2, 8.8, 12.0),
            (1, 12.3, 14.5),
            (2, 14.8, 17.0),
            (0, 17.2, 19.5)]
        for spk, start, end in timeline:
            f0, fmts = profiles[spk]
            s = int(start * self.sr)
            e = int(end * self.sr)
            t = np.arange(e-s) / self.sr
            sig = sum(0.2 * np.sin(
                2*np.pi*f*t) for f in fmts)
            for h in range(2, 5):
                sig += (0.15/h) * np.sin(
                    2*np.pi*f0*h*t)
            sig *= (0.5 + 0.5 * np.sin(
                2*np.pi*3*t))
            audio[s:e] += sig
        return audio, timeline

    def segment_audio(self, audio):
        """Split audio into overlapping
        segments."""
        seg_samples = int(
            self.seg_dur * self.sr)
        hop_samples = int(
            self.seg_hop * self.sr)
        segments = []
        starts = []
        i = 0
        while i + seg_samples <= (
                len(audio)):
            seg = audio[
                i:i + seg_samples]
            # Skip near-silent segments
            if np.sqrt(np.mean(
                    seg**2)) > 0.01:
                segments.append(seg)
                starts.append(
                    i / self.sr)
            i += hop_samples
        return segments, starts

    def extract_features(self, segment):
        """Simple spectral features
        as speaker embedding proxy."""
        n_fft = 512
        hop = 160
        n_frames = (
            len(segment) - n_fft) // hop
        if n_frames < 1:
            return np.zeros(80)
        spec = np.zeros(
            (40, n_frames))
        w = np.hanning(n_fft)
        for i in range(n_frames):
            s = i * hop
            fft = np.abs(np.fft.rfft(
                segment[s:s+n_fft] * w))
            bins = np.linspace(
                0, len(fft), 41,
                dtype=int)
            for m in range(40):
                if bins[m+1] > bins[m]:
                    spec[m, i] = np.mean(
                        fft[bins[m]:
                            bins[m+1]])
        spec = np.log(spec + 1e-9)
        mean = np.mean(spec, axis=1)
        std = np.std(spec, axis=1)
        return np.concatenate(
            [mean, std])

    def cluster_embeddings(self, embs,
                            n_clusters=3):
        """Simple K-means clustering."""
        # Initialize with K-means++
        centers = [embs[0].copy()]
        for _ in range(1, n_clusters):
            dists = np.array([
                min(np.linalg.norm(
                    e - c)**2
                    for c in centers)
                for e in embs])
            probs = dists / dists.sum()
            idx = self.rng.choice(
                len(embs), p=probs)
            centers.append(
                embs[idx].copy())
        centers = np.array(centers)

        # Iterate
        for _ in range(50):
            # Assign
            labels = np.array([
                np.argmin([
                    np.linalg.norm(e - c)
                    for c in centers])
                for e in embs])
            # Update
            new_centers = np.array([
                embs[labels == k].mean(
                    axis=0)
                if (labels == k).any()
                else centers[k]
                for k in range(
                    n_clusters)])
            if np.allclose(
                    centers, new_centers):
                break
            centers = new_centers

        return labels

    def run(self):
        audio, gt_timeline = (
            self.generate_meeting())
        print(f"Audio: {len(audio)/self.sr:.1f}s")
        print(f"\nGround truth:")
        for spk, s, e in gt_timeline:
            print(f"  Speaker {spk}: "
                  f"{s:.1f}s - {e:.1f}s")

        # Pipeline
        segments, starts = (
            self.segment_audio(audio))
        embs = np.array([
            self.extract_features(seg)
            for seg in segments])
        labels = self.cluster_embeddings(
            embs, n_clusters=3)

        print(f"\nDiarization output "
              f"({len(segments)} segments):")
        prev_label = -1
        for i, (start, label) in (
                enumerate(
                    zip(starts, labels))):
            end = start + self.seg_dur
            if label != prev_label:
                print(f"  Speaker_{label}: "
                      f"{start:.1f}s - "
                      f"{end:.1f}s")
                prev_label = label

        # Simple DER at 100ms resolution
        gt_l = np.zeros(200)
        for spk, s, e in gt_timeline:
            gt_l[int(s*10):int(e*10)] = (
                spk + 1)
        pr_l = np.zeros(200)
        for start, label in zip(
                starts, labels):
            s = int(start * 10)
            e = min(int((start +
                self.seg_dur) * 10), 200)
            pr_l[s:e] = label + 1
        speech = (gt_l > 0).sum()
        confusion = ((gt_l > 0)
            & (pr_l > 0)
            & (gt_l != pr_l)).sum()
        missed = ((gt_l > 0)
            & (pr_l == 0)).sum()
        fa = ((gt_l == 0)
            & (pr_l > 0)).sum()
        der = ((confusion + missed + fa)
               / max(speech, 1))
        print(f"\nDER: {der:.1%} "
              f"(confusion={confusion}, "
              f"missed={missed}, "
              f"fa={fa})")


diarizer = SpeakerDiarizer()
diarizer.run()

The Diarization Error Rate (DER) is the standard evaluation metric. It has three components: missed speech (speech that was labeled as silence), false alarm (silence that was labeled as speech), and speaker confusion (speech that was correctly detected but assigned to the wrong speaker). State-of-the-art systems achieve 3-7% DER on benchmark datasets like AMI and CALLHOME.

Production diarization systems like pyannote.audio use end-to-end neural approaches that handle overlapping speech (two people talking simultaneously) -- a scenario where our clustering-based pipeline fails because a single time frame contains features from multiple speakers. The pyannote system uses a segmentation network trained to predict speaker activity per-frame for each possible speaker, followed by a neural clustering step.

Combining ASR with diarization

The ultimate goal for meeting transcription is answering: who said what when? This requires combining speech recognition (episode #93) with speaker diarization. The approach is straightforward: run ASR to get word-level timestamps, run diarization to get speaker-level timestamps, then match each word to its speaker based on time overlap:

import numpy as np


class ASRDiarizationMerger:
    """Merge ASR transcripts with
    speaker diarization labels."""

    def __init__(self):
        self.rng = np.random.RandomState(
            42)

    def simulate_asr_output(self):
        """Simulated Whisper output."""
        words = [
            ('Okay', .5, .8),
            ('let', .85, .95),
            ('me', .95, 1.05),
            ('start', 1.1, 1.4),
            ('the', 1.45, 1.55),
            ('meeting', 1.6, 2.0),
            ('Thanks', 3.5, 3.8),
            ('for', 3.85, 3.95),
            ('joining', 4.0, 4.3),
            ('everyone', 4.35, 4.8),
            ('I', 6.3, 6.4),
            ('have', 6.4, 6.6),
            ('the', 6.65, 6.75),
            ('updates', 6.8, 7.2),
            ('ready', 7.25, 7.5)]
        return [{'word': w, 'start': s,
                 'end': e}
                for w, s, e in words]

    def simulate_diarization(self):
        """Simulated pyannote output."""
        return [
            ('SPEAKER_00', 0.3, 2.2),
            ('SPEAKER_01', 3.3, 5.0),
            ('SPEAKER_00', 6.1, 7.8)]

    def assign_speakers(self, words,
                          diar):
        """Map each word to its speaker
        based on timestamp overlap."""
        result = []
        for w in words:
            w_mid = (
                w['start'] + w['end']
                ) / 2
            speaker = "UNKNOWN"
            for spk, s, e in diar:
                if s <= w_mid <= e:
                    speaker = spk
                    break
            result.append({
                'speaker': speaker,
                'word': w['word'],
                'start': w['start']})
        return result

    def format_transcript(self, merged):
        """Group words by speaker."""
        lines = []
        cur_spk = None
        words = []
        for item in merged:
            if item['speaker'] != cur_spk:
                if words:
                    lines.append(
                        (cur_spk,
                         ' '.join(words)))
                cur_spk = item['speaker']
                words = [item['word']]
            else:
                words.append(item['word'])
        if words:
            lines.append(
                (cur_spk,
                 ' '.join(words)))
        return lines

    def run(self):
        words = (
            self.simulate_asr_output())
        diar = (
            self.simulate_diarization())

        print("=== Speaker-Attributed "
              "Transcript ===\n")
        merged = self.assign_speakers(
            words, diar)
        lines = self.format_transcript(
            merged)
        for spk, text in lines:
            print(f"[{spk}]: {text}")

        print(f"\n--- Stats ---")
        print(f"Total words: {len(words)}")
        spk_counts = {}
        for item in merged:
            spk = item['speaker']
            spk_counts[spk] = (
                spk_counts.get(spk, 0)
                + 1)
        for spk, count in sorted(
                spk_counts.items()):
            print(f"  {spk}: "
                  f"{count} words")


merger = ASRDiarizationMerger()
merger.run()

This is exactly how modern meeting transcription services work -- Whisper (or another ASR model from episode #93) provides the "what was said" with precise timestamps, and pyannote (or another diarizer) provides the "who was speaking." The merger assigns each word to the speaker who was active at that word's midpoint timestamp.

The main challenge is alignment: ASR and diarization models sometimes disagree about exact transition boundaries. A word might start during one speaker's turn and end during another's (during a rapid speaker change). Production systems handle this with more sophisticated overlap resolution, but the midpoint heuristic works surprisingly well in practice.

Evaluation metrics: EER and DER

Let's build proper evaluation metrics for both speaker verification and diarization:

import numpy as np


class SpeakerMetrics:
    """Compute EER for verification
    and DER for diarization."""

    def __init__(self):
        self.rng = np.random.RandomState(
            42)

    def compute_eer(self, pos_scores,
                     neg_scores):
        """Equal Error Rate: threshold
        where FAR == FRR."""
        thresholds = np.linspace(
            0, 1, 1000)
        best_diff = float('inf')
        eer = 0.0
        eer_thresh = 0.0
        for t in thresholds:
            # False Rejection Rate
            frr = np.mean(
                np.array(pos_scores) < t)
            # False Acceptance Rate
            far = np.mean(
                np.array(neg_scores) >= t)
            diff = abs(far - frr)
            if diff < best_diff:
                best_diff = diff
                eer = (far + frr) / 2
                eer_thresh = t
        return eer, eer_thresh

    def run(self):
        # Simulate verification scores
        pos = self.rng.normal(
            0.82, 0.08, 200)
        neg = self.rng.normal(
            0.35, 0.12, 200)

        eer, thresh = self.compute_eer(
            pos, neg)
        print(f"=== Speaker Verification ===")
        print(f"EER: {eer:.2%}")
        print(f"Threshold: {thresh:.3f}")
        print(f"Positive mean: "
              f"{np.mean(pos):.3f}")
        print(f"Negative mean: "
              f"{np.mean(neg):.3f}")

        # Compare systems
        print(f"\n=== System Comparison ===")
        systems = {
            'Basic x-vector': (
                self.rng.normal(.78,.1,200),
                self.rng.normal(.4,.15,200)),
            'ECAPA-TDNN': (
                self.rng.normal(.88,.05,200),
                self.rng.normal(.25,.08,200)),
            'ECAPA + LM': (
                self.rng.normal(.92,.04,200),
                self.rng.normal(.2,.06,200))}
        print(f"{'System':<16} {'EER':>8}")
        print("-" * 26)
        for name, (p, n) in (
                systems.items()):
            e, _ = self.compute_eer(p, n)
            print(f"{name:<16} {e:>7.2%}")


metrics = SpeakerMetrics()
metrics.run()

EER (Equal Error Rate) is the threshold where the False Acceptance Rate (impostors accepted) equals the False Rejection Rate (genuine speakers rejected). Modern systems achieve <1% EER on VoxCeleb benchmarks, which means that at the operating threshold, fewer than 1 in 100 imposter attempts succeed AND fewer than 1 in 100 genuine speakers are locked out.

The system comparison shows how architectural improvements translate to EER gains: basic x-vectors might achieve 5-8% EER, ECAPA-TDNN pushes to 1-3%, and adding score normalization or language-model-based resegmentation (for diarization) pushes further below 1%.

Samengevat

  • Speaker recognition maps voice segments to identity embeddings; verification asks "same person?" (1:1) while identification asks "which person?" (1:N) -- both rely on cosine similarity between embedding vectors;
  • d-vectors (LSTM + average pooling) and x-vectors (TDNN + statistics pooling) are the two foundational embedding paradigms; ECAPA-TDNN adds channel attention, multi-layer feature aggregation, and attentive statistics pooling;
  • training uses angular margin losses (AAM-Softmax / ArcFace) on large speaker datasets like VoxCeleb (7,000+ celebrities, 1M+ utterances) -- same loss family we saw for face recognition in episode #88;
  • VAD (Voice Activity Detection) identifies speech vs non-speech frames using energy + zero-crossing analysis or small neural networks -- critical preprocessing that prevents noise from contaminating speaker embeddings;
  • diarization segments multi-speaker audio into "who spoke when" using the pipeline: VAD -> segmentation -> embedding extraction -> clustering; DER (Diarization Error Rate) measures missed speech + false alarm + speaker confusion;
  • combining Whisper (ASR from episode #93) with diarization produces speaker-attributed transcripts for meeting recordings -- map each word to its speaker using timestamp overlap;
  • EER (Equal Error Rate) for verification is the threshold where false acceptance equals false rejection -- modern ECAPA-TDNN systems achieve <1% EER on standard benchmarks.

There's more to explore in the audio domain -- understanding natural language intent from spoken utterances, enhancing noisy audio recordings, and combining audio and visual streams into unified representations. The human voice carries a remarkable amount of information beyond just the words.

Exercises

Exercise 1: Build a speaker enrollment and verification simulator. Create a class SpeakerEnrollment that generates 6 synthetic speakers with distinct f0 ranges (80-250 Hz) and formant patterns, produces 5 enrollment + 3 test utterances each (1.5s, sr=16000), computes spectral embeddings (Mel spectrogram -> mean+std -> L2 normalize), averages enrollment embeddings per speaker, then runs all 108 verification trials (every test vs every enrolled speaker). Sweep thresholds 0.0-1.0 to find the EER. Print the EER, threshold, and a table of same-speaker vs different-speaker scores.

Exercise 2: Build a speaker change detector. Create a class SpeakerChangeDetector that generates 15s of audio with 4 segments (A: 0-3.5s, B: 3.5-7s, A: 7-10.5s, C: 10.5-15s). Use sliding windows (1.5s, hop 0.25s) to extract spectral features, compute cosine distance between consecutive windows, and detect peaks (local maxima above mean + 1.5*std). A change is "detected" if a peak falls within +/-0.5s of the true boundary. Print detected vs ground truth change points, precision, and recall.

Exercise 3: Build a multi-speaker overlap detector. Create a class OverlapDetector that generates 12s of audio: A alone (0-3s), A+B overlap (3-5s), B alone (5-8s), B+C overlap (8-10s), C alone (10-12s). Analyze energy in 4 frequency bands per frame, compute an overlap score (0-1) based on active band count and total energy. Compare predicted overlap frames to ground truth (3-5s and 8-10s), print precision, recall, F1, and average overlap score in overlap vs non-overlap regions.

Bedankt en tot de volgende keer!

scipio@scipio

Leave Learn AI Series (#97) - Speaker Recognition and Diarization to:

Written by

Does it matter who's right, or who's left?

Read more #stem posts

Best Posts From scipio

We have not curated any of scipio's posts yet. But you can encourage our curation team to review posts by visiting them regularly and by referring other readers. Because we give priority to frequently read content.

More Posts From scipio