微调 SpeechT5

现在您已经熟悉了文本到语音任务以及在英语语言数据上预训练的 SpeechT5 模型的内部工作原理，让我们看看如何将其微调到另一种语言。

准备工作

如果您想重现此示例，请确保您拥有 GPU。在笔记本中，您可以使用以下命令进行检查

nvidia-smi

在我们的示例中，我们将使用大约 40 小时的训练数据。如果您想使用 Google Colab 免费层级 GPU 进行操作，您需要将训练数据量减少到大约 10-15 小时，并减少训练步骤的数量。

您还需要一些额外的依赖项

pip install transformers datasets soundfile speechbrain accelerate

最后，不要忘记登录您的 Hugging Face 帐户，以便您可以上传您的模型并与社区分享

from huggingface_hub import notebook_login

notebook_login()

数据集

对于此示例，我们将采用 VoxPopuli 数据集的荷兰语 (nl) 语言子集。VoxPopuli 是一个大规模多语言语音语料库，包含来自 2009-2020 年欧洲议会活动录音的数据。它包含 15 种欧洲语言的标记音频转录数据。虽然我们将使用荷兰语子集，但您可以随意选择另一个子集。

这是一个自动语音识别 (ASR) 数据集，因此，如前所述，它不是训练 TTS 模型的最合适选择。但是，对于本练习来说已经足够好了。

让我们加载数据

from datasets import load_dataset, Audio

dataset = load_dataset("facebook/voxpopuli", "nl", split="train")
len(dataset)

输出

20968 个示例应该足以进行微调。SpeechT5 期望音频数据的采样率为 16 kHz，因此请确保数据集中的示例满足此要求

dataset = dataset.cast_column("audio", Audio(sampling_rate=16000))

预处理数据

首先，定义要使用的模型检查点并加载适当的处理器，该处理器包含我们将需要为训练准备数据的分词器和特征提取器

from transformers import SpeechT5Processor

checkpoint = "microsoft/speecht5_tts"
processor = SpeechT5Processor.from_pretrained(checkpoint)

SpeechT5 分词的文本清理

首先，为了准备文本，我们需要处理器的分词器部分，所以让我们获取它

tokenizer = processor.tokenizer

让我们看一个例子

dataset[0]

输出

{'audio_id': '20100210-0900-PLENARY-3-nl_20100210-09:06:43_4',
 'language': 9,
 'audio': {'path': '/root/.cache/huggingface/datasets/downloads/extracted/02ec6a19d5b97c03e1379250378454dbf3fa2972943504a91c7da5045aa26a89/train_part_0/20100210-0900-PLENARY-3-nl_20100210-09:06:43_4.wav',
  'array': array([ 4.27246094e-04,  1.31225586e-03,  1.03759766e-03, ...,
         -9.15527344e-05,  7.62939453e-04, -2.44140625e-04]),
  'sampling_rate': 16000},
 'raw_text': 'Dat kan naar mijn gevoel alleen met een brede meerderheid die wij samen zoeken.',
 'normalized_text': 'dat kan naar mijn gevoel alleen met een brede meerderheid die wij samen zoeken.',
 'gender': 'female',
 'speaker_id': '1122',
 'is_gold_transcript': True,
 'accent': 'None'}

您可能会注意到，数据集示例包含 raw_text 和 normalized_text 特征。在决定使用哪个特征作为文本输入时，重要的是要知道 SpeechT5 分词器没有任何数字标记。在 normalized_text 中，数字以文本形式写出。因此，它更适合，我们应该使用 normalized_text 作为输入文本。

由于 SpeechT5 是在英语上训练的，它可能无法识别荷兰语数据集中的某些字符。如果保持原样，这些字符将被转换为 <unk> 标记。但是，在荷兰语中，某些字符（如 à）用于强调音节。为了保留文本的含义，我们可以将此字符替换为常规的 a。

为了识别不受支持的标记，请使用 SpeechT5Tokenizer 提取数据集中所有唯一的字符，该分词器将字符作为标记处理。为此，我们将编写 extract_all_chars 映射函数，该函数将所有示例中的转录连接成一个字符串，并将其转换为一组字符。确保在 dataset.map() 中设置 batched=True 和 batch_size=-1，以便所有转录可以一次性用于映射函数。

def extract_all_chars(batch):
    all_text = " ".join(batch["normalized_text"])
    vocab = list(set(all_text))
    return {"vocab": [vocab], "all_text": [all_text]}


vocabs = dataset.map(
    extract_all_chars,
    batched=True,
    batch_size=-1,
    keep_in_memory=True,
    remove_columns=dataset.column_names,
)

dataset_vocab = set(vocabs["vocab"][0])
tokenizer_vocab = {k for k, _ in tokenizer.get_vocab().items()}

现在您有两组字符：一组来自数据集的词汇表，另一组来自分词器的词汇表。为了识别数据集中任何不受支持的字符，您可以取这两组之间的差集。结果集将包含数据集中但分词器中没有的字符。

dataset_vocab - tokenizer_vocab

输出

{' ', 'à', 'ç', 'è', 'ë', 'í', 'ï', 'ö', 'ü'}

为了处理上一步中识别出的不受支持的字符，我们可以定义一个将这些字符映射到有效标记的函数。请注意，空格已经由分词器中的替换，无需单独处理。

replacements = [
    ("à", "a"),
    ("ç", "c"),
    ("è", "e"),
    ("ë", "e"),
    ("í", "i"),
    ("ï", "i"),
    ("ö", "o"),
    ("ü", "u"),
]


def cleanup_text(inputs):
    for src, dst in replacements:
        inputs["normalized_text"] = inputs["normalized_text"].replace(src, dst)
    return inputs


dataset = dataset.map(cleanup_text)

既然我们已经处理了文本中的特殊字符，现在是时候将重点转移到音频数据了。

说话人

VoxPopuli 数据集包含来自多个说话人的语音，但数据集中有多少说话人？为了确定这一点，我们可以计算唯一说话人的数量以及每个说话人为数据集贡献的示例数量。在数据集中总共有 20,968 个示例的情况下，此信息将使我们更好地了解数据中说话人和示例的分布情况。

from collections import defaultdict

speaker_counts = defaultdict(int)

for speaker_id in dataset["speaker_id"]:
    speaker_counts[speaker_id] += 1

通过绘制直方图，您可以了解每个说话人有多少数据。

import matplotlib.pyplot as plt

plt.figure()
plt.hist(speaker_counts.values(), bins=20)
plt.ylabel("Speakers")
plt.xlabel("Examples")
plt.show()

直方图显示，数据集中大约三分之一的说话人拥有的示例少于 100 个，而大约十个说话人拥有的示例超过 500 个。为了提高训练效率并平衡数据集，我们可以将数据限制为示例数量在 100 到 400 之间的说话人。

def select_speaker(speaker_id):
    return 100 <= speaker_counts[speaker_id] <= 400


dataset = dataset.filter(select_speaker, input_columns=["speaker_id"])

让我们检查一下还剩下多少说话人

len(set(dataset["speaker_id"]))

输出

让我们看看还剩下多少示例

len(dataset)

输出

您还剩下不到 10,000 个示例，来自大约 40 位独特的说话人，这应该足够了。

请注意，如果示例很长，一些示例很少的说话人实际上可能拥有更多音频可用。但是，确定每个说话人的音频总量需要扫描整个数据集，这是一个耗时的过程，涉及加载和解码每个音频文件。因此，我们选择在此处跳过此步骤。

说话人嵌入

为了使 TTS 模型能够区分多个说话人，您需要为每个示例创建一个说话人嵌入。说话人嵌入是模型的附加输入，用于捕获特定说话人的声音特征。要生成这些说话人嵌入，请使用 SpeechBrain 中预训练的 spkrec-xvect-voxceleb 模型。

创建一个函数 create_speaker_embedding()，该函数接受输入音频波形并输出一个包含相应说话人嵌入的 512 元素向量。

import os
import torch
from speechbrain.pretrained import EncoderClassifier

spk_model_name = "speechbrain/spkrec-xvect-voxceleb"

device = "cuda" if torch.cuda.is_available() else "cpu"
speaker_model = EncoderClassifier.from_hparams(
    source=spk_model_name,
    run_opts={"device": device},
    savedir=os.path.join("/tmp", spk_model_name),
)


def create_speaker_embedding(waveform):
    with torch.no_grad():
        speaker_embeddings = speaker_model.encode_batch(torch.tensor(waveform))
        speaker_embeddings = torch.nn.functional.normalize(speaker_embeddings, dim=2)
        speaker_embeddings = speaker_embeddings.squeeze().cpu().numpy()
    return speaker_embeddings

重要的是要注意，speechbrain/spkrec-xvect-voxceleb 模型是在 VoxCeleb 数据集的英语语音上训练的，而本指南中的训练示例是荷兰语。虽然我们相信此模型仍将为我们的荷兰语数据集生成合理的说话人嵌入，但此假设可能并非在所有情况下都成立。

为了获得最佳结果，我们需要首先在目标语音上训练 X 向量模型。这将确保模型能够更好地捕获荷兰语中存在的独特声音特征。如果您想训练自己的 X 向量模型，可以使用此脚本作为示例。

处理数据集

最后，让我们将数据处理成模型期望的格式。创建一个 prepare_dataset 函数，该函数接受单个示例，并使用 SpeechT5Processor 对象对输入文本进行分词，并将目标音频加载到对数梅尔频谱图中。它还应添加说话人嵌入作为附加输入。

def prepare_dataset(example):
    audio = example["audio"]

    example = processor(
        text=example["normalized_text"],
        audio_target=audio["array"],
        sampling_rate=audio["sampling_rate"],
        return_attention_mask=False,
    )

    # strip off the batch dimension
    example["labels"] = example["labels"][0]

    # use SpeechBrain to obtain x-vector
    example["speaker_embeddings"] = create_speaker_embedding(audio["array"])

    return example

通过查看单个示例来验证处理是否正确

processed_example = prepare_dataset(dataset[0])
list(processed_example.keys())

输出

['input_ids', 'labels', 'stop_labels', 'speaker_embeddings']

说话人嵌入应为 512 元素向量

processed_example["speaker_embeddings"].shape

输出

(512,)

标签应为具有 80 个梅尔频段的对数梅尔频谱图。

import matplotlib.pyplot as plt

plt.figure()
plt.imshow(processed_example["labels"].T)
plt.show()

旁注：如果您发现此频谱图令人困惑，可能是因为您熟悉将低频放在底部，高频放在图顶部的约定。但是，当使用 matplotlib 库将频谱图绘制为图像时，y 轴会翻转，并且频谱图会上下颠倒显示。

现在我们需要将处理函数应用于整个数据集。这将需要 5 到 10 分钟。

dataset = dataset.map(prepare_dataset, remove_columns=dataset.column_names)

您会看到一个警告，指出数据集中的某些示例长度超过了模型可以处理的最大输入长度（600 个标记）。从数据集中删除这些示例。在这里，我们更进一步，为了允许更大的批次大小，我们删除了超过 200 个标记的任何内容。

def is_not_too_long(input_ids):
    input_length = len(input_ids)
    return input_length < 200


dataset = dataset.filter(is_not_too_long, input_columns=["input_ids"])
len(dataset)

输出

接下来，创建一个基本的训练/测试拆分

dataset = dataset.train_test_split(test_size=0.1)

数据整理器

为了将多个示例组合成一个批次，您需要定义一个自定义数据整理器。此整理器将使用填充标记填充较短的序列，确保所有示例都具有相同的长度。对于频谱图标签，填充部分将替换为特殊值 -100。此特殊值指示模型在计算频谱图损失时忽略频谱图的该部分。

from dataclasses import dataclass
from typing import Any, Dict, List, Union


@dataclass
class TTSDataCollatorWithPadding:
    processor: Any

    def __call__(
        self, features: List[Dict[str, Union[List[int], torch.Tensor]]]
    ) -> Dict[str, torch.Tensor]:
        input_ids = [{"input_ids": feature["input_ids"]} for feature in features]
        label_features = [{"input_values": feature["labels"]} for feature in features]
        speaker_features = [feature["speaker_embeddings"] for feature in features]

        # collate the inputs and targets into a batch
        batch = processor.pad(
            input_ids=input_ids, labels=label_features, return_tensors="pt"
        )

        # replace padding with -100 to ignore loss correctly
        batch["labels"] = batch["labels"].masked_fill(
            batch.decoder_attention_mask.unsqueeze(-1).ne(1), -100
        )

        # not used during fine-tuning
        del batch["decoder_attention_mask"]

        # round down target lengths to multiple of reduction factor
        if model.config.reduction_factor > 1:
            target_lengths = torch.tensor(
                [len(feature["input_values"]) for feature in label_features]
            )
            target_lengths = target_lengths.new(
                [
                    length - length % model.config.reduction_factor
                    for length in target_lengths
                ]
            )
            max_length = max(target_lengths)
            batch["labels"] = batch["labels"][:, :max_length]

        # also add in the speaker embeddings
        batch["speaker_embeddings"] = torch.tensor(speaker_features)

        return batch

在 SpeechT5 中，模型解码器部分的输入减少了 2 倍。换句话说，它从目标序列中丢弃了每个其他时间步。然后，解码器预测的序列长度是原来的两倍。由于原始目标序列长度可能是奇数，因此数据整理器会确保将批次的最大长度向下舍入为 2 的倍数。

data_collator = TTSDataCollatorWithPadding(processor=processor)

训练模型

从与用于加载处理器的相同检查点加载预训练模型

from transformers import SpeechT5ForTextToSpeech

model = SpeechT5ForTextToSpeech.from_pretrained(checkpoint)

use_cache=True 选项与梯度检查点不兼容。禁用它以进行训练，并重新启用缓存以进行生成，以加快推理时间

from functools import partial

# disable cache during training since it's incompatible with gradient checkpointing
model.config.use_cache = False

# set language and task for generation and re-enable cache
model.generate = partial(model.generate, use_cache=True)

定义训练参数。在这里，我们没有在训练过程中计算任何评估指标，我们将在本章稍后讨论评估。相反，我们只关注损失

from transformers import Seq2SeqTrainingArguments

training_args = Seq2SeqTrainingArguments(
    output_dir="speecht5_finetuned_voxpopuli_nl",  # change to a repo name of your choice
    per_device_train_batch_size=4,
    gradient_accumulation_steps=8,
    learning_rate=1e-5,
    warmup_steps=500,
    max_steps=4000,
    gradient_checkpointing=True,
    fp16=True,
    evaluation_strategy="steps",
    per_device_eval_batch_size=2,
    save_steps=1000,
    eval_steps=1000,
    logging_steps=25,
    report_to=["tensorboard"],
    load_best_model_at_end=True,
    greater_is_better=False,
    label_names=["labels"],
    push_to_hub=True,
)

实例化 Trainer 对象并将模型、数据集和数据整理器传递给它。

from transformers import Seq2SeqTrainer

trainer = Seq2SeqTrainer(
    args=training_args,
    model=model,
    train_dataset=dataset["train"],
    eval_dataset=dataset["test"],
    data_collator=data_collator,
    tokenizer=processor,
)

这样，我们就准备好开始训练了！训练将需要几个小时。根据您的 GPU，您在开始训练时可能会遇到 CUDA “内存不足”错误。在这种情况下，您可以将 per_device_train_batch_size 逐步减少 2 倍，并将 gradient_accumulation_steps 增加 2 倍以进行补偿。

trainer.train()

将最终模型推送到 🤗 Hub

trainer.push_to_hub()

推理

微调模型后，您可以将其用于推理！从 🤗 Hub 加载模型（请确保在以下代码片段中使用您的帐户名）

model = SpeechT5ForTextToSpeech.from_pretrained(
    "YOUR_ACCOUNT/speecht5_finetuned_voxpopuli_nl"
)

选择一个示例，这里我们将从测试数据集中取一个。获取说话人嵌入。

example = dataset["test"][304]
speaker_embeddings = torch.tensor(example["speaker_embeddings"]).unsqueeze(0)

定义一些输入文本并对其进行分词。

text = "hallo allemaal, ik praat nederlands. groetjes aan iedereen!"

预处理输入文本

inputs = processor(text=text, return_tensors="pt")

实例化一个声码器并生成语音

from transformers import SpeechT5HifiGan

vocoder = SpeechT5HifiGan.from_pretrained("microsoft/speecht5_hifigan")
speech = model.generate_speech(inputs["input_ids"], speaker_embeddings, vocoder=vocoder)

准备好收听结果了吗？

from IPython.display import Audio

Audio(speech.numpy(), rate=16000)

从新语言的此模型获得令人满意的结果可能具有挑战性。说话人嵌入的质量可能是一个重要因素。由于 SpeechT5 是使用英语 x 向量进行预训练的，因此在使用英语说话人嵌入时效果最佳。如果合成的语音听起来很差，请尝试使用不同的说话人嵌入。

增加训练时长也可能提高结果的质量。即便如此，语音显然是荷兰语而不是英语，并且它确实捕获了说话人的声音特征（与示例中的原始音频进行比较）。另一个可以尝试的是模型的配置。例如，尝试使用 config.reduction_factor = 1 来查看这是否可以改善结果。

在下一节中，我们将讨论如何评估文本到语音模型。

< > 在 GitHub 上更新

音频课程