如何用Segmentation替代Diarization实现准确高效的说话人数量估算

原本使用pyannote的Diarization模块估算未知说话人数量，但该方案运行缓慢且冗余。尝试通过音频分段+语音嵌入提取+聚类的方式优化速度，但当说话人数量超过2时，结果与Diarization模块的准确性差距较大，需要优化方案以兼顾准确性和运行效率。

原Diarization方案代码 #

from pyannote.audio import Pipeline
MY_TOKEN = ""  # huggingface_auth_token
audio_file = "my_audio.wav"
pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization@2.1", use_auth_token=MY_TOKEN)
output = pipeline(audio_file, min_speakers=2, max_speakers=10)
results = []
for turn, _, speaker in list(output.itertracks(yield_label=True)):
    results.append(speaker)
num_speakers = len(set(results))
print(num_speakers)

自定义分段聚类方案代码 #

from sklearn.cluster import SpectralClustering, KMeans, AgglomerativeClustering
from sklearn.metrics import silhouette_score
from spectralcluster import SpectralClusterer
from resemblyzer import VoiceEncoder, preprocess_wav
from pyannote.audio.pipelines.speaker_verification import PretrainedSpeakerEmbedding
from pyannote.audio import Model
from pyannote.audio import Audio
from pyannote.core import Segment
from pyannote.audio.pipelines import VoiceActivityDetection
import numpy as np


audio_file = "my_audio.wav"
MY_TOKEN = ""  # huggingface_token
embedding_model = PretrainedSpeakerEmbedding("speechbrain/spkrec-ecapa-voxceleb")
encoder = VoiceEncoder()
model = Model.from_pretrained("pyannote/segmentation", 
                              use_auth_token=MY_TOKEN)
pipeline = VoiceActivityDetection(segmentation=model)
HYPER_PARAMETERS = {
  # onset/offset activation thresholds
  "onset": 0.5, "offset": 0.5,
  # remove speech regions shorter than that many seconds.
  "min_duration_on": 0.0,
  # fill non-speech regions shorter than that many seconds.
  "min_duration_off": 0.0
}
pipeline.instantiate(HYPER_PARAMETERS)
vad = pipeline(audio_file)
audio_model = Audio()

segments = list(vad.itertracks(yield_label=True))
embeddings = np.zeros(shape=(len(segments), 192))
#embeddings = np.zeros(shape=(len(segments), 256))

for i, diaz in enumerate(segments):
    print(i, diaz)
    waveform, sample_rate = audio_model.crop(audio_file, diaz[0])
    embed = embedding_model(waveform[None])
    #wav = preprocess_wav(waveform[None].flatten().numpy())
    #embed = encoder.embed_utterance(wav)
    embeddings[i] = embed
embeddings = np.nan_to_num(embeddings)

max_clusters = 10
silhouette_scores = []
# clustering = SpectralClusterer(min_clusters=2, max_clusters=max_clusters, custom_dist="cosine")
# labels = clustering.predict(embeddings)
# print(labels)

for n_clusters in range(2, max_clusters+1):
    # clustering = SpectralClustering(n_clusters=n_clusters, affinity='nearest_neighbors').fit(embeddings)
    # clustering = KMeans(n_clusters=n_clusters).fit(embeddings)
    clustering = AgglomerativeClustering(n_clusters).fit(embeddings)
    labels = clustering.labels_
    score = silhouette_score(embeddings, labels)
    print(n_clusters, score)
    silhouette_scores.append(score)

# Choose the number of clusters that maximizes the silhouette score
number_of_speakers = np.argmax(silhouette_scores) + 2  # add 2 to account for starting at n_clusters=2
print(number_of_speakers)

优化方案 #

1. 对齐pyannote Diarization的分段逻辑 #

pyannote的Diarization并非单纯依赖VAD，而是使用说话人分割模型生成细粒度的说话人候选片段（包含说话人切换的边界）。直接使用VAD会丢失说话人切换的关键信息，导致聚类错误。优化方式：

• 使用pyannote.audio.pipelines.SpeakerSegmentation替代VAD，该模块会输出每个时间步的说话人概率，提取具有说话活动且可能包含说话人切换的片段。
• 过滤掉时长小于0.5秒的片段，这类片段的嵌入特征噪声大，会干扰聚类结果。

2. 使用匹配的嵌入模型 #

pyannote Diarization默认使用自身的嵌入模型，与自定义方案中使用的speechbrain/spkrec-ecapa-voxceleb特征分布不一致。建议替换为pyannote官方嵌入模型：

embedding_model = PretrainedSpeakerEmbedding(
    "pyannote/embedding",
    use_auth_token=MY_TOKEN
)

3. 采用带时间约束的谱聚类 #

pyannote的Diarization使用带时间连续性约束的谱聚类，避免同一说话人片段被错误划分。可以使用spectralcluster库的约束功能，或者直接调用pyannote的SpeakerClustering模块：

from pyannote.audio.pipelines import SpeakerClustering

# 初始化聚类器，支持时间约束
clustering_pipeline = SpeakerClustering(
    embedding=embedding_model,
    min_clusters=2,
    max_clusters=10,
)
# 传入分段和嵌入特征
clusters = clustering_pipeline(segments, embeddings=embeddings)
num_speakers = len(set(clusters.labels()))

4. 优化说话人数量估计方法 #

仅依赖轮廓系数（silhouette score）在高维嵌入下准确性不足，pyannote使用**贝叶斯信息准则（BIC）**结合聚类结果选择最优簇数。可以替换为BIC估计：

from sklearn.mixture import GaussianMixture

bic_scores = []
for n_clusters in range(2, max_clusters+1):
    gmm = GaussianMixture(n_components=n_clusters, covariance_type='diag')
    gmm.fit(embeddings)
    bic_scores.append(gmm.bic(embeddings))

# BIC值越小越好
number_of_speakers = np.argmin(bic_scores) + 2

5. 端到端优化：复用pyannote的中间模块 #

直接复用pyannote Diarization的分段和嵌入提取逻辑，跳过冗余步骤，只替换聚类和说话人数量估计部分，既能保证准确性又能提升速度：

from pyannote.audio import Pipeline
from pyannote.audio.pipelines.speaker_verification import PretrainedSpeakerEmbedding
from pyannote.core import Segment
import numpy as np
from spectralcluster import SpectralClusterer

MY_TOKEN = ""
audio_file = "my_audio.wav"

# 加载pyannote的分割模型（复用diarization的核心模块）
diarization_pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization@2.1", use_auth_token=MY_TOKEN)
# 获取分割结果中的片段（包含说话人候选）
segments = []
for turn, _, _ in diarization_pipeline(audio_file, min_speakers=2, max_speakers=10).itertracks():
    segments.append(turn)

# 提取嵌入
embedding_model = PretrainedSpeakerEmbedding("pyannote/embedding", use_auth_token=MY_TOKEN)
audio = Audio()
embeddings = []
for seg in segments:
    waveform, _ = audio.crop(audio_file, seg)
    embed = embedding_model(waveform[None])
    embeddings.append(embed.numpy().flatten())
embeddings = np.array(embeddings)

# 使用带时间约束的谱聚类
clusterer = SpectralClusterer(
    min_clusters=2,
    max_clusters=10,
    custom_dist="cosine",
    enforce_cluster_size=False,
)
labels = clusterer.predict(embeddings)
num_speakers = len(set(labels))
print(num_speakers)

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