如何基于WhisperX Speaker Diarization实现跨音频一致说话人识别

问题 #

我用WhisperX的Speaker Diarization转录固定4位嘉宾的系列播客，生成带时间戳和说话人标注的文本。但单集处理时，不同集的说话人标签（比如SPEAKER_00、SPEAKER_01）无法对应——比如第400集的SPEAKER_00和第423集的SPEAKER_00可能是不同嘉宾。

我目前想到两个思路：

• 把所有音频合并成大文件处理后拆分结果
• 写单独脚本分析语音片段，跨文件匹配说话人

有没有更优的方案实现数百个播客音频的一致说话人识别？当前使用的代码如下：

# -*- coding: utf-8 -*-
"""WhisperX_Speaker_Diarization.ipynb

Automatically generated by Colab.

Original file is located at
    https://colab.research.google.com/drive/1IHum-j2AOjVOs_ZoqJ5yBUjf1kI4SLmt

    pip install --q git+https://github.com/m-bain/whisperx.git

Run with:
    python3 Collab\ Notebooks/whisperx_speaker_diarization.py > output/output.log 
"""
import whisperx
import gc
from dotenv import load_dotenv
import os
import json
import time

# Load environment variables from .env file
load_dotenv()

# Get Hugging Face token from environment variable
huggingface_token = os.getenv("HUGGINGFACE_TOKEN")

device = "cuda"
batch_size = 4 # reduce if low on GPU mem
compute_type = "int8" # change from "float16" to "int8" if low on GPU mem (may reduce accuracy)

audio_file = "audio/short_MEGA64_PODCAST_483.mp3"

audio = whisperx.load_audio(audio_file)

model = whisperx.load_model("large-v2", device, compute_type=compute_type)

result = model.transcribe(audio, batch_size=batch_size)
print(result["segments"]) # before alignment

# delete model if low on GPU resources
# import gc; gc.collect(); torch.cuda.empty_cache(); del model

# 2. Align whisper output
model_a, metadata = whisperx.load_align_model(language_code=result["language"], device=device)
result = whisperx.align(result["segments"], model_a, metadata, audio, device, return_char_alignments=False)

result

diarize_model = whisperx.DiarizationPipeline(use_auth_token=huggingface_token,
                                             device=device)

diarize_segments = diarize_model(audio, min_speakers=1, max_speakers=8)

diarize_segments

diarize_segments.speaker.unique()

result = whisperx.assign_word_speakers(diarize_segments, result)
print(diarize_segments)
# print(result["segments"]) # segments are now assigned speaker IDs

# Save the result to a JSON file with a unique filename
timestamp = int(time.time() * 1000)
audio_filename = os.path.basename(audio_file).split('.')[0]
output_filename = f"speaker_timestamps_{audio_filename}_{timestamp}.json"

with open(output_filename, 'w') as f:
    json.dump(result, f)

print(f"Results saved to {output_filename}")

更优解决方案 #

1. 基于说话人嵌入的全局匹配（推荐） #

WhisperX底层依赖的pyannote.audio可以提取说话人的语音嵌入（代表语音特征的向量），核心思路是建立一个全局的嘉宾特征库，再将单集的说话人特征与库中匹配，实现统一标签：

• 步骤1：构建全局嘉宾特征库
  从已处理的播客中，为每个说话人抽取1-2分钟清晰的语音片段，用pyannote的嵌入模型计算每个嘉宾的平均嵌入向量，保存为全局特征库（比如用Numpy文件存储）。
• 步骤2：单集处理时匹配全局特征
  处理新集时，提取当前所有说话人的嵌入向量，与全局库中的向量计算余弦相似度，找到最匹配的嘉宾标签，替换原有的SPEAKER_XX临时标签。

这种方案GPU资源占用可控，适合批量处理数百集，且匹配准确率较高。

2. Few-Shot说话人验证（若有已知样本） #

如果有嘉宾的已知语音样本（比如单独的自我介绍录音、某一集里明确标注的片段），可以直接用pyannote的Few-Shot验证功能，在处理每一集时，用已知样本和当前集的说话人片段做匹配，直接映射到对应嘉宾标签，无需提前构建全局库。

3. 优化合并拆分方案（不推荐） #

如果坚持用合并音频的思路，需注意：

• 合并时给每个原音频之间添加5秒以上的静音，方便后续识别拆分边界
• 处理后根据静音位置拆分结果，避免说话人标签跨集混淆
  但该方案对GPU内存要求极高，数百集的合并文件体积过大，实际操作可行性低。

代码修改示例（全局匹配方案） #

在原有代码基础上添加说话人嵌入匹配逻辑：

# 新增依赖导入
from pyannote.audio import Model
from pyannote.audio.pipelines import SpeakerVerification
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

# 预加载说话人嵌入模型（需Hugging Face权限）
embedding_model = Model.from_pretrained(
    "pyannote/speaker-diarization-3.1",
    use_auth_token=huggingface_token
)
verification_pipeline = SpeakerVerification(embedding_model)

# 假设已提前构建全局嘉宾嵌入库（示例）
# 实际使用时需从已处理的播客中提取并保存
global_embeddings = {
    "Guest_A": np.load("guest_a_embedding.npy"),
    "Guest_B": np.load("guest_b_embedding.npy"),
    "Guest_C": np.load("guest_c_embedding.npy"),
    "Guest_D": np.load("guest_d_embedding.npy")
}

def map_speaker_labels(diarize_segments, global_embeddings, audio):
    speaker_embeddings = {}
    # 提取当前集每个说话人的嵌入向量
    for speaker in diarize_segments.speaker.unique():
        # 取该说话人第一个清晰的语音片段
        speaker_segments = diarize_segments[diarize_segments.speaker == speaker]
        start = speaker_segments.iloc[0]["start"]
        end = speaker_segments.iloc[0]["end"]
        # 提取对应音频片段（pyannote默认16kHz采样率）
        speaker_audio = audio[int(start*16000):int(end*16000)]
        # 计算嵌入
        embedding = verification_pipeline.embed(speaker_audio)
        speaker_embeddings[speaker] = embedding
    
    # 匹配全局嵌入，生成标签映射
    label_mapping = {}
    for local_speaker, local_emb in speaker_embeddings.items():
        max_sim = -1
        best_match = None
        for global_label, global_emb in global_embeddings.items():
            sim = cosine_similarity([local_emb], [global_emb])[0][0]
            if sim > max_sim:
                max_sim = sim
                best_match = global_label
        label_mapping[local_speaker] = best_match
    
    # 替换原标签
    diarize_segments["speaker"] = diarize_segments["speaker"].map(label_mapping)
    return diarize_segments

# 在原有代码生成diarize_segments后调用
diarize_segments = map_speaker_labels(diarize_segments, global_embeddings, audio)

# 后续继续执行assign_word_speakers和保存逻辑
result = whisperx.assign_word_speakers(diarize_segments, result)

内容的提问来源于stack exchange，提问作者Martin