【古诗生成AI实战】之三——任务加载器与预处理器-Toy模板网

这篇具有很好参考价值的文章主要介绍了【古诗生成AI实战】之三——任务加载器与预处理器。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

本章内容属于数据处理阶段，将分别介绍任务加载器task和预处理器processor。

加载与模型关联的处理器 (processor),# 文本生成,# 古诗生成AI实战,人工智能,文本生成,古诗创作

[1] 数据集

在深入探讨数据处理的具体步骤之前，让我们先了解一下我们将要使用的数据集的形式。

本项目采用的是七绝数据集，总计83072条古诗，其形式如下：

偈颂一百二十三首  其二○:布袋头开天地大，翠岩眉毛在不在。一声鸿鴈海门秋，行人已在青山外。
偈颂一百二十三首  其二八:释迦老子弄精魂，死了依前似不曾。一树落花春过后，远山无限碧层层。
偈颂一百二十三首  其三○:一夜江风搅玉尘，孤峰不白转精神。从空放下从空看，彻骨寒来有几人。
...

实际上我们只需要古诗的正文部分：

布袋头开天地大，翠岩眉毛在不在。一声鸿鴈海门秋，行人已在青山外。
释迦老子弄精魂，死了依前似不曾。一树落花春过后，远山无限碧层层。
一夜江风搅玉尘，孤峰不白转精神。从空放下从空看，彻骨寒来有几人。
...

除了七绝数据集之外，我们的项目还包括其他类型的古诗数据集，如七律、五绝和五律。虽然在本项目的实战应用中我们并未使用这些数据集，但它们对于对不同古诗形式感兴趣的学习者来说是极好的实践资源。如果你有兴趣探索更多古诗风格，不妨尝试一下这些数据集。

需要注意的一点是，这些数据集都是以繁体字呈现的。因此，在处理这些数据之前，需要将繁体字转换为简体字。为此，我们可以利用opencc库，这是一个专门用于中文繁简转换的工具。使用opencc，你可以轻松地完成繁简之间的转换，确保数据适合模型训练和文本生成。

更多关于opencc库的信息和使用文档，可以参考其在GitHub上的官方页面：https://github.com/BYVoid/OpenCC。

[2] 任务加载器task

在我们详细了解了数据集的格式之后，接下来我们将探讨项目中非常关键的组件之一：任务加载器task。

那么，什么是任务加载器task呢？简单来说，这是我自定义的一个数据加载类Task，它的主要职责是从各种格式的数据文件（如txt、csv、json）中加载数据。考虑到不同数据集的格式可能有所不同，Task类还负责提取文件中的指定部分数据（例如，仅从csv文件的某些列中提取数据），并进行后续的文本清洗工作。

你可能会问，为什么我们要在项目中专门编写任务加载器task呢？其主要原因是为了增强项目代码的可扩展性，使得我们能够在不同任务和不同数据集之间灵活切换。这种设计理念在实际应用中极为重要。

当然，当读者亲自查看实战项目的代码后，就会直观感受到规范和高效代码的优势。

下面，让我们来看一下实战项目中针对七绝古诗数据集定义的任务加载器QiJueTask的伪代码：

class QiJueTask(DataTask):
    """Processor for the Example data set."""

    def get_train_examples(self, data_path):
        return self._create_examples(data_path, "train")

    def get_dev_examples(self, data_path):
        return []

    def get_test_examples(self, data_path):
        return []

    def get_single_examples(self, text):
        # 定制id
        guid = "%s-%s" % ('test', 0)
        example = InputExample(guid=guid, text=text)
        return [example]

    def get_labels(self):
        return []

    def check_text(self, text):
        """检查预处理后文本"""
        if any(char in text for char in ['_', '(', '（', '《', '[', '□', 'Ｃ', '￣', 'ｗ', 'p', '{', '}']):
            # logging.info(f'文本【{text}】中包含特殊字符，请修改预处理逻辑')
            return False
        if len(text) != 32:
            logging.info(f'文本【{text}】长度不为32，请修改预处理逻辑')
            return False
        return True

    def preprocess_text(self, text):
        """文本预处理"""
        text = text.replace(' ', '')
        return text


    def _create_examples(self, path: str, set_type: str) -> List[InputExample]:
        logging.info(f"加载{set_type}数据集中...")
        examples = []

        with open(path, "r", encoding='utf-8', ) as f:
            for i, line in tqdm(enumerate(f.readlines())):
                try:
                    title, content = line.strip().split(':')
                    content = self.preprocess_text(content)
                    if self.check_text(content):
                        # 定制id
                        guid = "%s-%s" % (set_type, i)
                        example = InputExample(guid=guid, text=content)
                        examples.append(example)
                except ValueError as e:
                    pass
        return examples

在我们的代码中，我们定义了一个名为QiJueTask的类，这个类的核心作用是处理不同类型的数据集，比如训练集、验证集和测试集，以及对单个文本样本的特定处理。这个类包含总计8个方法，下面是对这些方法的简要介绍：

· get_xxx_examples(self, data_path): 这些方法负责加载不同类型的训练数据集。它们通过调用内部方法_create_examples来完成这一任务，并传递数据路径和特定的标记（如"train"、"dev"、"test"）来区分训练、验证和测试数据。

· get_single_examples(self, text): 这个方法用于为单个文本样本创建InputExample对象。这在进行单个古诗文本的预测时显得非常重要。

注意：InputExample类是一个用于存放文本数据的容器，我将在后面详细介绍。

· get_labels(self): 这个方法返回一个空列表，因为在我们的任务中，标签是从输入文本中直接得到，而不是预先定义的。

· check_text(self, text): 这个方法用于检查经过预处理的文本。虽然我们对数据进行了清洗，但有时仍可能存在一些遗漏，因此这个方法提供了一个机会来进行额外的过滤和检查。

· preprocess_text(self, text): 这是文本预处理的方法，当前的主要操作是去除文本中的空格。

· _create_examples(self, path: str, set_type: str): 这是一个内部方法，用于从数据文件中读取数据，并将其转换为InputExample对象的列表。它接受数据路径和数据集的类型（训练、验证或测试）作为参数。

总的来说，任务加载器task的主要作用就是从数据集文件中读取文本、清洗文本、处理文本，然后将处理后的文本存放到List[InputExample]中。下面，我们将详细探讨InputExample类的结构和作用。

class InputExample(object):
    """原始样本类"""

    def __init__(self, guid, text, word_label=None, index_label=None, encode_label=None):
        """
        Create a new InputExample.

        :param guid: a unique textual identifier
        :param text: the sequence of text
        :param word_label: 单词形式的标签 如['sport','tech']
        :param index_label: 索引形式的标签 从总标签集里面取出word_label对应的索引 如[2,0]
        :param encode_label: 独热编码形式的标签 根据总标签集 把index_label转化成独热标签 如[0,1,1,0]
        """
        self.guid = guid
        self.text = text
        self.word_label = word_label
        self.index_label = index_label
        self.encode_label = encode_label

为了确保我们的代码具有高度的拓展性，InputExample类被设计得相当灵活，其属性丰富多样。这种设计不仅适用于我们当前的古诗生成任务，还可以方便地应用于其他类型的文本任务，比如文本分类任务，或是适配transformers库中的模型（例如Bert、GPT）。

那么，我们的任务加载器task是如何实现在不同任务和不同数据集之间灵活切换的能力呢？

让我们通过一个例子来说明这一点。假设目前我们手头上有四个不同格式的数据集。为了处理这些不同的数据集，我们可以为每个数据集定义一个特定的任务加载器task。通过这种方式，当需要切换数据集时，我们只需简单地修改数据集的名称，就可以迅速切换到不同的数据集，如下所示：

TASKS = {
    "qijue": QiJueTask,
    "qilv": QiLvTask,
    "wujue": WuJueTask,
    "wulv": WuLvTask,
}

之后想加载不同数据集的训练集，你只需要使用如下代码：

task = TASK['your_dataset_name']
train_data = task.get_train_examples('your_dataset_path')

[3] 预处理器processor

让我们来探讨预处理器processor的作用。processor的主要职责是基于数据集构建词典，并向外提供分词器tokenizer的功能。

对于那些有使用transformers库经验的人来说，他们会知道，在使用预训练模型处理文本数据之前，需要通过transformers.PreTrainedTokenizer将文本转换为向量形式。

考虑到本项目代码可能需要对transformers库中的预训练模型进行拓展使用，我们决定将分词器集成到预处理器processor中。这样做的好处是，如果我们选择使用transformers库的预训练模型，就无需手动构建词典和分词器；但如果不使用这些预训练模型，我们则需要自己从数据集中读取数据并构建词典，以便将文本转化为向量。进一步地，我们可能还需要对嵌入层进行预训练的词向量初始化。

‘嵌入’、‘word2vec’是NLP领域非常基础的概念。以下两篇文章介绍的十分详细：
1、【文本分类】深入理解embedding层的模型、结构与文本表示：介绍了嵌入的相关理论基础；
2、一文了解Word2vec 阐述训练流程：介绍了嵌入的训练方式。

现在，我们将详细讲解预处理器processor中词典的构建方式：

    def _build_vocab(cls, data: List[InputExample]):
        """构建词表"""

        if os.path.exists(cls.vocab_path):
            logging.info('加载词表中...')
            with open(cls.vocab_path, 'rb') as fh:
                words = pickle.load(fh)
            logging.info(f'词表大小={len(words)}')
            return words

        logging.info('构建词表中...')
        texts = []
        for example in tqdm(data):
            texts.append(cls.start_token + example.text + cls.end_token)

        all_words = [word for text in texts for word in text]
        counter = collections.Counter(all_words)
        words = sorted(counter.keys(), key=lambda x: counter[x], reverse=True)
        # 空格作为生词字符
        words.append(cls.unk_token)
        with open(cls.vocab_path, 'wb') as fh:
            pickle.dump(words, fh)
        logging.info(f'词表大小={len(words)} 内容={words}')
        return words

在我们的项目中，预处理器processor在构建词汇表时会遵循一系列明确的步骤。首先，它会检查指定的词汇表文件路径是否存在。如果该路径上已有词汇表文件，预处理器将直接从中加载这个词汇表。这对于加速处理流程非常有帮助，尤其是在处理大型数据集时。

如果词汇表文件不存在，预处理器则会开始从头构建一个新的词汇表。这个过程涉及处理传入的InputExample对象列表，从中提取所有文本内容。在构建词汇表的过程中，预处理器会在每个文本样本的前后分别添加开始和结束标记，这有助于模型理解文本的边界。接着，它会统计并记录每个唯一词汇的出现频率，并根据这些频率以降序的方式对词汇进行排列。

完成这些步骤后，最终得到的词典就是一个包含所有不同词汇的字符串列表。这个词汇表是后续文本处理和模型训练的基础，它确保了每个文本样本都能以一致和标准的方式被模型理解和处理。如下：

加载与模型关联的处理器 (processor),# 文本生成,# 古诗生成AI实战,人工智能,文本生成,古诗创作

其次就是预处理器processor向外提供分词器：

    def tokenizer(cls, text, add_end=True):
        if add_end:
            # 加上前缀字符
            text = cls.start_token + text + cls.end_token

            if len(text) < cls.max_seq_len:
                text += cls.unk_token * (cls.max_seq_len - len(text))
            elif len(text) > cls.max_seq_len:
                text = text[:cls.max_seq_len - 1] + cls.end_token
        else:
            text = cls.start_token + text
            if len(text) < cls.max_seq_len:
                text += cls.unk_token * (cls.max_seq_len - len(text))
            elif len(text) > cls.max_seq_len:
                text = text[:cls.max_seq_len]
        # 转id
        ids = [cls.vocab.index(word) for word in text]

        return ids

在我们的预处理器processor中，有一个特定的函数负责将文本字符串转换为模型能够理解的格式。这个函数接收两个参数：一个文本字符串和一个布尔值add_end。add_end参数用于指示是否在文本的末尾添加一个结束标记。

在处理文本时，函数首先会在文本的开头添加一个开始标记。如果add_end被设为True，那么在文本的末尾也会添加一个结束标记。这一步是为了让模型能够更清楚地识别文本的开始和结束。随后，根据设定的最大序列长度max_seq_len，文本可能会被截断或者填充未知字符标记，以确保其长度与max_seq_len一致。最后，文本中的每个字符都会被转换成对应的索引值，这些索引值代表了字符在词汇表vocab列表中的位置。

设置add_end参数的目的在于，当进行文本生成时，我们可以选择不在文本后面添加结束字符。这是因为如果文本末尾添加了结束字符，模型在预测下一个输出时有较大可能会生成结束字符，这可能不利于生成连续和流畅的文本内容。

[4] 提取word2vec

在我们自定义的模型中，嵌入层（embedding layer）起着至关重要的作用。这个层的参数可以通过两种方式初始化：一种是随机初始化，另一种则是使用已有的官方预训练词向量，这后者通常被称作微调（Fine-tuning）。

self.embedding = nn.Embedding.from_pretrained(embedding_pretrained, freeze=False)

举个例子，假设我们的输入中有一个字‘天’。如果我们选择随机初始化（比如全为0），那么在嵌入层中，‘天’对应的表征向量就会是一个全零向量[0,0,...,0]。这种情况下，项目初期所有词的表征向量都将是相同的，这显然不利于模型的有效学习。

为了解决这个问题，有研究者基于大规模语料库进行无监督训练，以获取各种字、词的表征向量。这些表征向量最终形成了一组称为word2vec的向量库。

然而，在中文处理中，尽管常用字可能只有几千或几万个，但通常我们会在词级别而非字级别进行分词，这就意味着常用的词组合非常庞大。因此，第三方提供的word2vec通常包含几十万个不同的字和词向量，其大小通常达到1GB甚至更大。

在本项目中，由于我们只使用了大约7千个词，所以我们需要从这些庞大的word2vec库中提取出仅与我们词典中存在的字词相对应的向量。下面是实现这一目的的代码示例：

# ## 加载预训练词向量
def get_embed(file):
    print('加载词向量中...')
    model = KeyedVectors.load_word2vec_format(file, binary=False)

    print('加载加载文本构建的词汇表中...')
    vocabulary = pickle.load(open('./dataset/qijue/qijue_vocab.pkl', 'rb'))

    # 构建词汇-向量字典
    ebed = []
    for word in vocabulary:
        if word in model.key_to_index.keys():
            ebed.append(model[word])
        else:
            ebed.append(np.asarray([0 for i in range(0, 300)], dtype='float32'))

    # 储存词汇-向量字典
    np.savez(r'./dataset/qijue/sg.qijue.300d.npz',embeddings=np.asarray(ebed, dtype='float32'))