TensorFlow深度学习实战——基于循环神经网络的词性标注模型

0. 前言

‌词性标注 (Part-Of-Speech tagging, POS tagging) 也被称为语法标注 (grammatical tagging) 或词类消疑 (word-category disambiguation)，是将语料库内单词的词性按其含义和上下文内容进行标记的文本数据处理技术‌，涉及识别文本中每个单词的语法类别，如名词、动词、形容词等。词性标注对于理解句子结构和语义至关重要，广泛应用于各类大规模语料库的自然语言处理和文本挖掘。

1. 词性标注

在本节中，我们将使用门控循环单元 (Gated Recurrent Unit, GRU) 层构建一个用于进行词性 (Part of Speech, POS) 标注的网络。POS 是指在多个句子中以相同方式使用的单词的语法类别，词性包括名词、动词、形容词等。例如，名词通常用于指代事物，动词通常用于描述动作，形容词用于描述事物的属性。传统上，词性标注通常是手动完成的，但现在基本上已解决了此问题，最初通过统计模型实现，而近年来则主要是通过端到端的深度学习模型完成。

2. Penn Treebank 数据集

为了获取训练数据，我们需要带有词性标记的句子。Penn Treebank 数据集是一个人工标注语料库，包含约 450 万个美式英语单词。然而，它是一个收费资源，但 Penn Treebank 中 10% 的样本作为 NLTK 的一部分可以免费获得，我们将使用该样本训练网络。
模型以句子中的单词序列作为输入，然后输出每个单词对应的词性标记。因此，对于输入单词序列 [The, cat, sat. on, the, mat, .]，输出序列应该是词性符号 [DT, NN, VB, IN, DT, NN, .]。
为了获取数据，需要首先安装 NLTK 库，然后下载 treebank 数据集：

>>> import nltk
>>> nltk.download("treebank")

2. 数据处理

(1) 首先导入所需库：

import numpy as np
import os
import shutil
import tensorflow as tf
import nltk

(2) 将 NLTK treebank 数据集惰性导入为一对文件，一个文件包含句子，另一个文件包含对应的词性序列：

def clean_logs(data_dir):logs_dir = os.path.join(data_dir, "logs")shutil.rmtree(logs_dir, ignore_errors=True)return logs_dirdef download_and_read(dataset_dir, num_pairs=None):sent_filename = os.path.join(dataset_dir, "treebank-sents.txt")poss_filename = os.path.join(dataset_dir, "treebank-poss.txt")if not(os.path.exists(sent_filename) and os.path.exists(poss_filename)):if not os.path.exists(dataset_dir):os.makedirs(dataset_dir)fsents = open(sent_filename, "w")fposs = open(poss_filename, "w")sentences = nltk.corpus.treebank.tagged_sents()for sent in sentences:fsents.write(" ".join([w for w, p in sent]) + "\n")fposs.write(" ".join([p for w, p in sent]) + "\n")fsents.close()fposs.close()sents, poss = [], []with open(sent_filename, "r") as fsent:for idx, line in enumerate(fsent):sents.append(line.strip())if num_pairs is not None and idx >= num_pairs:breakwith open(poss_filename, "r") as fposs:for idx, line in enumerate(fposs):poss.append(line.strip())if num_pairs is not None and idx >= num_pairs:breakreturn sents, poss# clean up log area
data_dir = "./data"
logs_dir = clean_logs(data_dir)# download and read source and target data into data structure
sents, poss = download_and_read("./datasets", num_pairs=NUM_PAIRS)
assert(len(sents) == len(poss))
print("# of records: {:d}".format(len(sents)))

数据集中包含 3,194 个句子，以上代码将句子和对应的词性标签写入到文件中，例如 treebank-sents.txt 文件的第一行包含第一个句子，treebank-poss.txt 文件的第一行包含第一个句子中每个单词的对应词性标签。下表展示了该数据集中的两个句子及其对应的词性标签：

句子	词性
Pierre Vinken, 61 years old, will join the board as a nonexecutive director Nov. 29.	NNP NNP , CD NNS JJ , MD VB DT NN IN DT JJ NN NNP CD.
Mr. Vinken is chairman of Elsevier N.V., the Dutch publishing group.	NNP NNP VBZ NN IN NNP NNP , DT NNP VBG NN.

(3) 然后，使用 TensorFlow 的词元分析器 Tokenizer 对句子进行词元化，并创建一个句子词元 (token) 列表。网络的每个输入记录当前是一个文本词元序列，但它们需要是整数序列。在标记化过程中，Tokenizer 还维护着词汇表中的词元，可以从中构建词元与整数之间的映射。需要考虑两个词汇表，一个是句子集合中的单词词元的词汇表，另一个是词性集合中的词性标记的词汇表。对这两个集合进行词元化，并生成所需的映射字典：

def tokenize_and_build_vocab(texts, vocab_size=None, lower=True):if vocab_size is None:tokenizer = tf.keras.preprocessing.text.Tokenizer(lower=lower)else:tokenizer = tf.keras.preprocessing.text.Tokenizer(num_words=vocab_size+1, oov_token="UNK", lower=lower)tokenizer.fit_on_texts(texts)if vocab_size is not None:# additional workaround, see issue 8092# https://github.com/keras-team/keras/issues/8092tokenizer.word_index = {e:i for e, i in tokenizer.word_index.items() if i <= vocab_size+1 }word2idx = tokenizer.word_indexidx2word = {v:k for k, v in word2idx.items()}return word2idx, idx2word, tokenizer# vocabulary sizes
word2idx_s, idx2word_s, tokenizer_s = tokenize_and_build_vocab(sents, vocab_size=9000)
word2idx_t, idx2word_t, tokenizer_t = tokenize_and_build_vocab(poss, vocab_size=38, lower=False)
source_vocab_size = len(word2idx_s)
target_vocab_size = len(word2idx_t)
print("vocab sizes (source): {:d}, (target): {:d}".format(source_vocab_size, target_vocab_size))

(4) 尽管句子中的词元数和对应的词性标记序列相同，但句子的长度有所不同。网络期望输入具有相同的长度，因此我们需要决定句子的长度。计算不同的百分位数，并打印这些百分位数的句子长度：

sequence_lengths = np.array([len(s.split()) for s in sents])
print([(p, np.percentile(sequence_lengths, p)) for p in [75, 80, 90, 95, 99, 100]])
# [(75, 33.0), (80, 35.0), (90, 41.0), (95, 47.0), (99, 58.0), (100, 271.0)]

可以将句子长度设置为 100，这样可能会导致一些句子被截断。长度短于指定长度的句子将在末尾进行填充。由于本节所用数据集很小，我们更倾向于尽可能多地使用它，因此最终选择了最大长度。

(5) 接下来，根据输入创建数据集。首先，将输入和输出序列中的词元和词性标签转换为整数序列。其次，将较短的序列填充到最大长度 271。需要注意的是，在填充后，我们对词性标记序列执行了一个额外的操作，使用 to_categorical() 函数将其转换为一个独热编码序列。TensorFlow 提供了处理整数序列输出的损失函数，但我们希望尽可能保持代码简单，因此选择自行进行转换。最后，使用 from_tensor_slices() 函数创建数据集，对其进行打乱，并将其拆分为训练、验证和测试集：

NUM_PAIRS = None
EMBEDDING_DIM = 128
RNN_OUTPUT_DIM = 256
BATCH_SIZE = 128max_seqlen = 271# create dataset
sents_as_ints = tokenizer_s.texts_to_sequences(sents)
sents_as_ints = tf.keras.preprocessing.sequence.pad_sequences(sents_as_ints, maxlen=max_seqlen, padding="post")
poss_as_ints = tokenizer_t.texts_to_sequences(poss)
poss_as_ints = tf.keras.preprocessing.sequence.pad_sequences(poss_as_ints, maxlen=max_seqlen, padding="post")
dataset = tf.data.Dataset.from_tensor_slices((sents_as_ints, poss_as_ints))
idx2word_s[0], idx2word_t[0] = "PAD", "PAD"
poss_as_catints = []
for p in poss_as_ints:poss_as_catints.append(tf.keras.utils.to_categorical(p, num_classes=target_vocab_size, dtype="int32"))
poss_as_catints = tf.keras.preprocessing.sequence.pad_sequences(poss_as_catints, maxlen=max_seqlen)
dataset = tf.data.Dataset.from_tensor_slices((sents_as_ints, poss_as_catints))# split into training, validation, and test datasets
dataset = dataset.shuffle(10000)
test_size = len(sents) // 3
val_size = (len(sents) - test_size) // 10
test_dataset = dataset.take(test_size)
val_dataset = dataset.skip(test_size).take(val_size)
train_dataset = dataset.skip(test_size + val_size)# create batches
batch_size = BATCH_SIZE
train_dataset = train_dataset.batch(batch_size)
val_dataset = val_dataset.batch(batch_size)
test_dataset = test_dataset.batch(batch_size)

3. 模型构建与训练

(1) 定义模型。
采用顺序模型，包括嵌入层、Dropout 层、双向 GRU 层、全连接层和 Softmax 激活。输入是形状为 (batch_size, max_seqlen) 的整数序列，通过嵌入层时，序列中的每个整数被转换为大小为 (embedding_dim) 的向量，因此张量的形状变为 (batch_size, max_seqlen, embedding_dim)。这些向量中的每个元素传递给具有 256 输出维度的双向 GRU 的相应时间步。
由于使用双向 GRU，这相当于在一个 GRU 上堆叠另一个 GRU，因此从双向 GRU 输出的张量的维度是 (batch_size, max_seqlen, 2rnn_output_dimension)。每个时间步张量的形状为 (batch_size, 1, 2rnn_output_dimension)，输入到一个全连接层，该层将每个时间步转换为与目标词汇表大小相同的向量，即 (batch_size, number_of_timesteps, output_vocab_size)。每个时间步代表一个输出词元的概率分布，因此最终的 softmax 层应用于每个时间步，以返回输出的词性标签序列。

NUM_EPOCHS = 50class POSTaggingModel(tf.keras.Model):def __init__(self, source_vocab_size, target_vocab_size,embedding_dim, max_seqlen, rnn_output_dim, **kwargs):super(POSTaggingModel, self).__init__(**kwargs)self.embed = tf.keras.layers.Embedding(source_vocab_size, embedding_dim, input_length=max_seqlen)self.dropout = tf.keras.layers.SpatialDropout1D(0.2)self.rnn = tf.keras.layers.Bidirectional(tf.keras.layers.GRU(rnn_output_dim, return_sequences=True))self.dense = tf.keras.layers.TimeDistributed(tf.keras.layers.Dense(target_vocab_size))self.activation = tf.keras.layers.Activation("softmax")def call(self, x):x = self.embed(x)x = self.dropout(x)x = self.rnn(x)x = self.dense(x)x = self.activation(x)return x# define model
embedding_dim = EMBEDDING_DIM
rnn_output_dim = RNN_OUTPUT_DIM

(2) 实例化模型并设置参数，使用 Adam 优化器、分类交叉熵损失函数和准确率指标编译模型：

model = POSTaggingModel(source_vocab_size, target_vocab_size,embedding_dim, max_seqlen, rnn_output_dim)
model.build(input_shape=(batch_size, max_seqlen))
model.summary()model.compile(loss="categorical_crossentropy",optimizer="adam", metrics=["accuracy", masked_accuracy()])

在以上代码中，除了准确率度量之外还使用了 masked_accuracy() 度量。由于填充的存在，标签和预测中有许多零，这导致准确率指标相对乐观。观察训练过程，在第一个 epoch 结束时模型的验证准确率为 0.9116，但实际上，生成的词性标签质量非常低。
最好的方法是用一个忽略两个数都是 0 的匹配项的损失函数来替换当前的损失函数；但是，更简单的方法是构建一个更严格的度量，并使用它来判断训练何时停止。因此，构建了一个新的准确率度量函数 masked_accuracy()：

def masked_accuracy():def masked_accuracy_fn(ytrue, ypred):ytrue = tf.keras.backend.argmax(ytrue, axis=-1)ypred = tf.keras.backend.argmax(ypred, axis=-1)mask = tf.keras.backend.cast(tf.keras.backend.not_equal(ypred, 0), tf.int32)matches = tf.keras.backend.cast(tf.keras.backend.equal(ytrue, ypred), tf.int32) * masknumer = tf.keras.backend.sum(matches)denom = tf.keras.backend.maximum(tf.keras.backend.sum(mask), 1)accuracy =  numer / denomreturn accuracyreturn masked_accuracy_fn

(3) 训练模型。设置模型检查点和 TensorBoard 回调，然后在模型上调用 fit() 方法，批大小为 128，进行 50 个 epoch 的模型训练：

num_epochs = NUM_EPOCHSbest_model_file = os.path.join(data_dir, "best_model.h5")
checkpoint = tf.keras.callbacks.ModelCheckpoint(best_model_file, save_weights_only=True,save_best_only=True)
tensorboard = tf.keras.callbacks.TensorBoard(log_dir=logs_dir)
history = model.fit(train_dataset, epochs=num_epochs,validation_data=val_dataset,callbacks=[checkpoint, tensorboard])# evaluate with test set
best_model = POSTaggingModel(source_vocab_size, target_vocab_size,embedding_dim, max_seqlen, rnn_output_dim)
best_model.build(input_shape=(batch_size, max_seqlen))
best_model.load_weights(best_model_file)
best_model.compile(loss="categorical_crossentropy",optimizer="adam", metrics=["accuracy", masked_accuracy()])test_loss, test_acc, test_masked_acc = best_model.evaluate(test_dataset)
print("test loss: {:.3f}, test accuracy: {:.3f}, masked test accuracy: {:.3f}".format(test_loss, test_acc, test_masked_acc))# predict on batches
labels, predictions = [], []
is_first_batch = True
accuracies = []for test_batch in test_dataset:inputs_b, outputs_b = test_batchpreds_b = best_model.predict(inputs_b)# convert from categorical to list of intspreds_b = np.argmax(preds_b, axis=-1)outputs_b = np.argmax(outputs_b.numpy(), axis=-1)for i, (pred_l, output_l) in enumerate(zip(preds_b, outputs_b)):assert(len(pred_l) == len(output_l))pad_len = np.nonzero(output_l)[0][0]acc = np.count_nonzero(np.equal(output_l[pad_len:], pred_l[pad_len:])) / len(output_l[pad_len:])accuracies.append(acc)if is_first_batch:words = [idx2word_s[x] for x in inputs_b.numpy()[i][pad_len:]]postags_l = [idx2word_t[x] for x in output_l[pad_len:] if x > 0]postags_p = [idx2word_t[x] for x in pred_l[pad_len:] if x > 0]print("labeled  : {:s}".format(" ".join(["{:s}/{:s}".format(w, p) for (w, p) in zip(words, postags_l)])))print("predicted: {:s}".format(" ".join(["{:s}/{:s}".format(w, p) for (w, p) in zip(words, postags_p)])))print(" ")is_first_batch = Falseaccuracy_score = np.mean(np.array(accuracies))
print("pos tagging accuracy: {:.3f}".format(accuracy_score))

训练过程如下所示。可以看到，masked_accuracy 和 val_masked_accuracy 数字比 accuracy 和 val_accuracy 数字更为保守，这是因为 masked_accuracy 不考虑输入为 PAD 字符的序列位置：

使用测试集中一些随机句子生成的词性标签，并对比相应的真实词性标签。可以看出，虽然指标值并不十分优秀，但模型已经学会了进行相当不错的词性标注：

相关链接

TensorFlow深度学习实战（1）——神经网络与模型训练过程详解
TensorFlow深度学习实战（2）——使用TensorFlow构建神经网络
TensorFlow深度学习实战（3）——深度学习中常用激活函数详解
TensorFlow深度学习实战（4）——正则化技术详解
TensorFlow深度学习实战（5）——神经网络性能优化技术详解
TensorFlow深度学习实战（7）——分类任务详解
TensorFlow深度学习实战（8）——卷积神经网络
TensorFlow深度学习实战（12）——词嵌入技术详解
TensorFlow深度学习实战（13）——神经嵌入详解
TensorFlow深度学习实战（14）——循环神经网络详解
TensorFlow深度学习实战——基于循环神经网络的文本生成模型
TensorFlow深度学习实战——基于循环神经网络的情感分析模型