Speaker Diarization
Identify and separate multiple speakers from a single audio stream.
Overview
Speaker diarization answers "who spoke when" in multi-speaker recordings. This solution uses PyAnnote.audio's pretrained models to classify audio segments by speaker, enabling downstream processing like role assignment and transcript attribution.
Key capability: Identify 2-10 speakers in real-world audio without prior voice samples.
Problem
Multi-speaker recordings are common in business contexts:
- Meeting recordings with 3-8 participants
- Interview audio with interviewer and interviewee
- Podcast episodes with host and guests
- Call center recordings with agent and customer
Challenge: Standard transcription returns a single text stream without speaker attribution. Manual speaker labeling is time-consuming and error-prone.
Requirement: Automatically segment audio by speaker with high accuracy, even with overlapping speech.
Technical Approach
Model Selection
| Model | Accuracy | Speed | GPU Required | Selected |
|---|---|---|---|---|
| PyAnnote 3.1 | 92% F1 | 40s/10min | Optional | ✅ |
| Whisper diarization | 85% F1 | 60s/10min | Recommended | |
| AWS Transcribe | 88% F1 | 45s/10min | N/A (cloud) | |
| Google Speech-to-Text | 87% F1 | 50s/10min | N/A (cloud) |
Decision: PyAnnote offers best accuracy with local deployment option.
Architecture
┌─────────────────────────────────────────────────────────────┐
│ Input │
├─────────────────────────────────────────────────────────────┤
│ Audio file (.mp3, .wav) + Optional: expected speaker count │
└───────────────────────────┬─────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ PyAnnote Pipeline │
├─────────────────────────────────────────────────────────────┤
│ 1. Voice Activity Detection (VAD) │
│ └─ Identify speech vs. silence │
│ │
│ 2. Speaker Embedding Extraction │
│ └─ 256-dim vector per segment │
│ │
│ 3. Clustering │
│ └─ Group embeddings by speaker │
│ │
│ 4. Resegmentation │
│ └─ Refine boundaries │
└───────────────────────────┬─────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ Output │
├─────────────────────────────────────────────────────────────┤
│ [ │
│ { "start": 0.0, "end": 5.2, "speaker": "SPEAKER_00" }, │
│ { "start": 5.2, "end": 10.5, "speaker": "SPEAKER_01" }, │
│ ... │
│ ] │
└─────────────────────────────────────────────────────────────┘
Implementation Details
- HuggingFace Authentication: PyAnnote requires token for model download
- GPU Acceleration: 4x speedup on CUDA devices
- Speaker Count: Can specify or let model auto-detect (2-10)
- Minimum Duration: Filter out segments < 0.5s for noise reduction
Code Sample
from pyannote.audio import Pipeline
import torch
from typing import Optional
class SpeakerDiarizer:
"""
Speaker diarization using PyAnnote pretrained models.
Requirements:
pip install pyannote.audio torch
Usage:
diarizer = SpeakerDiarizer(hf_token="your_token")
segments = diarizer.diarize("meeting.mp3", num_speakers=4)
"""
def __init__(self, hf_token: str, model: str = "pyannote/speaker-diarization-3.1"):
"""
Initialize diarization pipeline.
Args:
hf_token: HuggingFace access token (required)
model: Model identifier on HuggingFace Hub
"""
self.pipeline = Pipeline.from_pretrained(
model,
use_auth_token=hf_token
)
# Use GPU if available
if torch.cuda.is_available():
self.pipeline.to(torch.device("cuda"))
self.device = "cuda"
else:
self.device = "cpu"
def diarize(
self,
audio_path: str,
num_speakers: Optional[int] = None,
min_speakers: int = 2,
max_speakers: int = 10,
min_duration: float = 0.5
) -> list[dict]:
"""
Perform speaker diarization on audio file.
Args:
audio_path: Path to audio file
num_speakers: Exact number of speakers (if known)
min_speakers: Minimum speakers for auto-detection
max_speakers: Maximum speakers for auto-detection
min_duration: Minimum segment duration in seconds
Returns:
List of segments with speaker labels:
[
{"start": 0.0, "end": 5.2, "speaker": "SPEAKER_00", "duration": 5.2},
{"start": 5.2, "end": 10.5, "speaker": "SPEAKER_01", "duration": 5.3},
...
]
"""
# Run diarization
diarization = self.pipeline(
audio_path,
num_speakers=num_speakers,
min_speakers=min_speakers,
max_speakers=max_speakers
)
# Convert to list format
segments = []
for turn, _, speaker in diarization.itertracks(yield_label=True):
duration = turn.end - turn.start
if duration >= min_duration:
segments.append({
"start": round(turn.start, 3),
"end": round(turn.end, 3),
"speaker": speaker,
"duration": round(duration, 3)
})
return self._merge_adjacent(segments)
def _merge_adjacent(
self,
segments: list[dict],
gap_threshold: float = 0.3
) -> list[dict]:
"""
Merge adjacent segments from same speaker.
Args:
segments: List of segments
gap_threshold: Maximum gap to merge (seconds)
Returns:
Merged segment list
"""
if not segments:
return segments
merged = [segments[0].copy()]
for seg in segments[1:]:
prev = merged[-1]
gap = seg["start"] - prev["end"]
if seg["speaker"] == prev["speaker"] and gap <= gap_threshold:
# Extend previous segment
prev["end"] = seg["end"]
prev["duration"] = round(prev["end"] - prev["start"], 3)
else:
merged.append(seg.copy())
return merged
def get_speaker_stats(self, segments: list[dict]) -> dict:
"""
Calculate statistics per speaker.
Returns:
{
"SPEAKER_00": {"duration": 120.5, "segments": 15, "percentage": 45.2},
"SPEAKER_01": {"duration": 98.3, "segments": 12, "percentage": 36.8},
...
}
"""
stats = {}
total_duration = sum(s["duration"] for s in segments)
for seg in segments:
speaker = seg["speaker"]
if speaker not in stats:
stats[speaker] = {"duration": 0, "segments": 0}
stats[speaker]["duration"] += seg["duration"]
stats[speaker]["segments"] += 1
for speaker in stats:
stats[speaker]["duration"] = round(stats[speaker]["duration"], 2)
stats[speaker]["percentage"] = round(
stats[speaker]["duration"] / total_duration * 100, 1
) if total_duration > 0 else 0
return stats
# Example usage
if __name__ == "__main__":
import os
diarizer = SpeakerDiarizer(
hf_token=os.environ["HF_TOKEN"]
)
segments = diarizer.diarize(
"meeting_recording.mp3",
num_speakers=3 # Known: 3 participants
)
print(f"Found {len(segments)} segments")
stats = diarizer.get_speaker_stats(segments)
for speaker, data in stats.items():
print(f"{speaker}: {data['duration']}s ({data['percentage']}%)")
Performance Metrics
| Metric | Value | Test Conditions |
|---|---|---|
| Diarization Error Rate (DER) | 8.2% | AMI meeting corpus |
| F1 Score | 0.92 | NAATI exam audio |
| Processing Speed | 0.25x real-time | CPU (M1 Mac) |
| Processing Speed | 0.06x real-time | GPU (RTX 3080) |
| Memory Usage | 2.1 GB | Peak during inference |
| Min Audio Duration | 10s | For reliable results |
References:
Use Cases
| Industry | Application |
|---|---|
| 🎙️ Podcasting | Attribute speech to host vs. guest for editing |
| 📞 Call Centers | Separate agent and customer for quality analysis |
| 🏢 Meetings | Assign speaking time per participant |
| 🎓 Education | Distinguish instructor from students |
| 🏛️ Legal | Identify speakers in depositions and hearings |
Integration Patterns
With Transcription
# Combine diarization with transcription
segments = diarizer.diarize("audio.mp3")
transcript = transcriber.transcribe("audio.mp3")
# Assign words to speakers
for word in transcript["words"]:
word["speaker"] = find_speaker_at_time(segments, word["start"])
With LLM Analysis
# Use diarization for role classification
segments = diarizer.diarize("interview.mp3", num_speakers=2)
# LLM assigns semantic roles
roles = llm.classify_roles(
segments=segments,
possible_roles=["interviewer", "interviewee"]
)
Limitations
| Limitation | Mitigation |
|---|---|
| Overlapping speech | Post-process with higher time resolution |
| Similar voices (twins) | Combine with content analysis |
| Background noise | Pre-filter with VAD or noise reduction |
| Very short utterances (<0.5s) | Lower min_duration threshold carefully |
Related Projects
- Multilingual Audio Processing Platform — Full case study using this module