微调 SpeechT5

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

内务管理

如果你想重现这个例子，请确保你有 GPU。在 notebook 中，你可以使用以下命令进行检查：

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 是用英语训练的，它可能无法识别荷兰语数据集中的某些字符。如果保持原样，这些字符将被转换为 `` 标记。然而，在荷兰语中，某些字符如 `à` 被用来强调音节。为了保留文本的含义，我们可以将此字符替换为常规的 `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-vector 模型。这将确保模型能够更好地捕获荷兰语中存在的独特语音特征。如果你想训练自己的 X-vector 模型，可以使用此脚本作为示例。

处理数据集

最后，让我们将数据处理成模型期望的格式。创建一个 `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,
    eval_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-vectors 进行了预训练，因此在使用英语说话者嵌入时表现最佳。如果合成语音听起来很差，请尝试使用不同的说话者嵌入。

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

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

< > 在 GitHub 上更新

音频课程