基于 pyannote.audio 实现 Speaker Identification（说话人识别）

一、概述

pyannote.audio 直接能够进行 Speaker Verification、Speaker Diarization 等任务，但尚未提供 Speaker Identification。 本文简述如何基于 pyannote.audio 进行 Speaker Identification。

二、实现步骤

SpeakerDiarization 内部会进行声纹提取，故最初本打算直接使用。翻阅源码发现尽管可以通过 hook 获取 embeddings，但 embeddings 是经过加窗处理的并且会被 SpeakerDiarization 修改，所以采用如下方式。

1、Create SpeakerDiarization pipeline

from pyannote.audio.pipelines import SpeakerDiarization
from pyannote.core.annotation import Annotation

AUTH_TOKEN = "your auth token"

pipeline: SpeakerDiarization = Pipeline.from_pretrained(
    "pyannote/speaker-diarization@2.1", use_auth_token=AUTH_TOKEN, cache_dir="./models/pyannote")
output: Annotation = pipeline(wav_file_path)

2、Apply SpeakerDiarization

wav_file_folder = "./audio"
wav_file_name = "audio-01.wav"
wav_file_path = wav_file_folder + "/" + wav_file_name

output: Annotation = pipeline(wav_file_path, hook=hook)

tracks = list(output.itertracks())

3、定义相关函数

加载音频

import torch

def load_audio(wav_file_path, frame_offset, num_frames):
    signal, sr = torchaudio.load(str(wav_file_path),
                                 frame_offset=frame_offset,
                                 num_frames=num_frames,
                                 channels_first=False)
    return signal, sr

提取声纹

from speechbrain.pretrained import EncoderClassifier

# 直接使用创建好的分类器
classifier: EncoderClassifier = pipeline._embedding.classifier_

def extract_embedding(classifier: EncoderClassifier, signal, sr):
    waveform = classifier.audio_normalizer(signal, sr)
    # Fake batches:
    batch = waveform.unsqueeze(0)
    # Encode
    emb = classifier.encode_batch(batch, normalize=True)
    return emb

生成说话人名称

speaker_count = 0

def generate_speaker_name():
    global speaker_count
    speaker_count += 1
    return f"发音人_{speaker_count:02d}"

Speaker Verification

similarity = torch.nn.CosineSimilarity(dim=-1, eps=1e-6)

def speaker_verification(emb1, em2):
    score = 0
    score = float(similarity(emb1, em2)[0])
    decision = score > EMBENDING_THRESHOLD
    return score, decision

Speaker Identification

def speaker_identification(emb: torch.Tensor, speakers: Dict, segment: Segment):
    result_decision = False
    result_score = 0.0
    result_name = None
    for name, embedding in speakers.items():
        score, decision = speaker_verification(emb, embedding)
        if decision:
            result_decision = True
            result_score = score
            result_name = name
            break

    if not result_decision:
        result_name = classes()

    # 如果不存在则新增，存在则更新。
    speakers[result_name] = emb
    return result_decision, result_score, result_name

4、分段识别

# 音频采样率
SAMPLE_RATE = 16000
# 提取声纹：使用的音频长度
EMBEDDING_DURATION = 1.0
# 提取声纹：只使用达到一定长度的音频
EMBEDDING_DURATION_THRESHOLD = EMBEDDING_DURATION + 0.2
# 提取声纹：读取音频帧数量(torchaaudio)
EMBEDDING_NUM_FRAMES = int(SAMPLE_RATE * EMBEDDING_DURATION)
# 提取声纹：验证声纹达标阈值
EMBENDING_THRESHOLD = 0.25

tracks = list(output.itertracks())
speakers = {}

for idx in range(0, len(tracks)):
    track, speaker = tracks[idx]
    start = max(track.start, tracks[idx - 1]
                [0].end) if idx != 0 else track.start
    end = min(track.end, tracks[idx + 1]
              [0].start) if idx != len(tracks) - 1 else track.end
    segment = Segment(start=start, end=end)

    print(
        f"{speaker} {track} ==>> start={start:.3f} stop={end:.3f} duration={(end - start):.3f}")

    if segment.duration < EMBEDDING_DURATION_THRESHOLD:
        #print(f"Ignore: start={start:.3f} stop={end:.3f}")
        pass
    else:
        # 加载音频
        frame_offset = int(
            (segment.middle - EMBEDDING_DURATION / 2) * SAMPLE_RATE)
        signal, sr = load_audio(wav_file_path, frame_offset,
                                EMBEDDING_NUM_FRAMES)

        # 提取声纹
        emb = extract_embedding(classifier, signal, sr)

        # 说话人辨认
        decision, score, name = speaker_identification(emb, speakers, segment)
        print(f"{segment} {decision}, {score:.3f}, {name}")

print(f"说话人数量：{len(speakers)}")
for name, _ in speakers.items():
    print(name)

三、Know issues

pyannote.audio 和 SpeechBrain 提供的预训练模型是基于英文数据集训练的，目前没有发现有开源的中文模型，故处理中文的 EER 欠佳。

通过加窗、为每个说话人保留多个声纹等对识别率有一定的提升但有限。

附录

pyannote.audio 简介

pyannote.audio 是一个基于 Python 的开源音频处理工具包，它提供了一组用于音频分析、处理、识别和分类的模块。它基于深度学习和机器学习技术，可以用于语音信号处理、说话人识别、语音转录、声音事件检测等领域的应用。

该工具包的核心是一个基于深度学习的音频分析框架，可以对音频数据进行处理和建模，并进行说话人识别、语音转录、声音事件检测等任务。此外，pyannote.audio 还提供了许多实用工具和函数，如音频读取、处理、可视化等，可以方便地进行音频数据的处理和分析。

该工具包的优点是易于使用，提供了丰富的文档和示例，并且是一个开源工具包，可以方便地进行定制和扩展。

pyannote.audio 比较基础的功能是 segmentation，能够将语音进行加窗处理。在此基础上，可以进行如下任务：

1. Voice activity detection(VAD), 语音活动检测。用于识别输入信号中是否存在语音(即人声)活动。pyannote.audio 能输出相应的时间戳。

2. Overlapped speech detection, 重叠语音检测。重叠语音是指两个或多个说话者同时讲话的情况。pyannote.audio 能输出相应的时间戳。

3. Speaker counting, 说话人计数。统计一段音频中有不同时刻有几人在说话。pyannote.audio 能输出相应的时间戳。

4. Speaker change detection, 说话人改变检测。检测一段音频中多人说话在哪个时间点发生了切换。pyannote.audio 能输出相应的时间戳。

5. Speaker Verification, 说话人验证。确定一个人是否是一个给定说话人。但是，pyannote.audio 只是获取声纹而已。需要配合其他代码进行验证。获取声纹可以使用 pyannote/embedding 或 speechbrain/spkrec-ecapa-voxceleb 。

6. Speaker Diarization, 说话人分离。pyannote.audio 能分离出一条音频中每个人说话的时间段。如果处理多条音频，比如多条音频里都有”张三”，Speaker Diarization 并不知道是同一个人——这需要 Speaker Identification。

SpeechBrain 简介

pyannote.audio 官方示例默认使用 SpeechBrain 提取声纹和验证。前者对后者进行了封装。下面是一段简单的、直接使用 SpeechBrain 的代码。

import torchaudio
from speechbrain.pretrained import SpeakerRecognition, EncoderClassifier

AUTH_TOKEN = "your auth token"

classifier = EncoderClassifier.from_hparams(
    source="speechbrain/spkrec-ecapa-voxceleb",
    savedir="./models/speechbrain/spkrec-ecapa-voxceleb",
    use_auth_token=AUTH_TOKEN)

signal, fs = torchaudio.load('./audio/audio-03.wav')
embeddings = classifier.encode_batch(signal, normalize=True)
print(embeddings.shape)
print(embeddings)


verification = SpeakerRecognition.from_hparams(
    source="speechbrain/spkrec-ecapa-voxceleb",
    savedir="./models/speechbrain/spkrec-ecapa-voxceleb",
    use_auth_token=AUTH_TOKEN)

score, prediction = verification.verify_files(
    './audio/audio-03.wav', './audio/audio-04.wav')

print(score, prediction)

Speaker Verification、Speaker Diarization 和 Speaker Identification 的区别

Speaker Verification、Speaker Identification 和 Speaker Diarization 都是与说话人相关的语音处理任务，但它们的目标和方法略有不同。

• Speaker Verification（说话人验证）：是确定一个人是否是一个给定说话人的任务。这个任务通常涉及到两个步骤：注册和验证。在注册阶段，系统会录制目标说话人的语音作为该人的声纹模板；在验证阶段，系统会接收一个声音样本，并尝试确定该声音样本是否与给定说话人的声纹模板匹配。Speaker Verification 的应用包括语音门禁系统和电话银行验证系统等。

• Speaker Identification（说话人识别）：是从一个包含多个说话人的语音信号中自动识别出每个说话人的任务。这个任务通常涉及到两个步骤：说话人特征提取和说话人分类。在特征提取阶段，系统会将每个说话人的声纹特征提取出来，通常使用的特征包括梅尔频率倒谱系数（MFCC）和高斯混合模型（GMM）。在分类阶段，系统会使用训练好的分类器来确定每个声音段所属的说话人。Speaker Identification 的应用包括犯罪调查和语音检索等。

• Speaker Diarization（说话人分割聚类、说话人分离/日志）：是从多个说话人混合的语音信号中自动分离出不同的说话人的任务。这个任务通常涉及到两个步骤：语音段切分和说话人聚类。在语音段切分阶段，系统会将语音信号切分成独立的语音段，每个语音段通常只包含一个说话人。在聚类阶段，系统会尝试将同一个说话人的所有语音段聚合到一起，以便进行后续的分析和处理。Speaker Diarization 的应用包括电话会议记录和语音翻译等。

参考资料

pyannote.audio
SpeechBrain
声纹技术：从核心算法到工程实践