政安晨:【Keras机器学习示例演绎】(四十四)—— 使用序列到序列转换器实现英语到西班牙语的翻译
目录
简介
设置
下载数据
解析数据
文本数据矢量化
建立模型
训练我们的模型
解码测试句子
政安晨的个人主页:政安晨
欢迎 👍点赞✍评论⭐收藏
收录专栏: TensorFlow与Keras机器学习实战
希望政安晨的博客能够对您有所裨益,如有不足之处,欢迎在评论区提出指正!
本文目标:实现序列到序列转换器,并在机器翻译任务中对其进行训练。
简介
在本例中,我们将建立一个序列到序列 Transformer 模型,并在英语到西班牙语的机器翻译任务中对其进行训练。
您将学习如何
—— 使用 Keras TextVectorization 层对文本进行矢量化。
—— 实现 TransformerEncoder 层、TransformerDecoder 层和 PositionalEmbedding 层。
—— 为训练序列到序列模型准备数据。
—— 使用训练好的模型生成从未见过的输入句子的译文(序列到序列推理)。
设置
# We set the backend to TensorFlow. The code works with # both `tensorflow` and `torch`. It does not work with JAX # due to the behavior of `jax.numpy.tile` in a jit scope # (used in `TransformerDecoder.get_causal_attention_mask()`: # `tile` in JAX does not support a dynamic `reps` argument. # You can make the code work in JAX by wrapping the # inside of the `get_causal_attention_mask` method in # a decorator to prevent jit compilation: # `with jax.ensure_compile_time_eval():`. import os os.environ["KERAS_BACKEND"] = "tensorflow" import pathlib import random import string import re import numpy as np import tensorflow.data as tf_data import tensorflow.strings as tf_strings import keras from keras import layers from keras import ops from keras.layers import TextVectorization
下载数据
我们将使用 Anki 提供的英西翻译数据集。让我们下载它:
text_file = keras.utils.get_file( fname="spa-eng.zip", origin="http://storage.googleapis.com/download.tensorflow.org/data/spa-eng.zip", extract=True, ) text_file = pathlib.Path(text_file).parent / "spa-eng" / "spa.txt"
解析数据
每行包含一个英语句子和相应的西班牙语句子。英语句子是源序列,西班牙语句子是目标序列。我们在西班牙语句子前加上标记"[start]",在西班牙语句子后加上标记"[end]"。
with open(text_file) as f: lines = f.read().split("\n")[:-1] text_pairs = [] for line in lines: eng, spa = line.split("\t") spa = "[start] " + spa + " [end]" text_pairs.append((eng, spa))
下面是我们的句子对:
for _ in range(5): print(random.choice(text_pairs))
("On Saturday nights, it's difficult to find parking around here.", '[start] Los sábados por la noche es difícil encontrar aparcamiento por aquí. [end]') ('I was the worst student in the class.', '[start] Fui el peor estudiante en la clase. [end]') ('There is nothing to do today.', '[start] No hay nada que hacer hoy. [end]') ('The twins do resemble each other.', '[start] Los gemelos se parecen mutuamente. [end]') ('They found Tom in the crowd.', '[start] Encontraron a Tom entre la multitud. [end]')
现在,让我们把句子对分成训练集、验证集和测试集。
random.shuffle(text_pairs) num_val_samples = int(0.15 * len(text_pairs)) num_train_samples = len(text_pairs) - 2 * num_val_samples train_pairs = text_pairs[:num_train_samples] val_pairs = text_pairs[num_train_samples : num_train_samples + num_val_samples] test_pairs = text_pairs[num_train_samples + num_val_samples :] print(f"{len(text_pairs)} total pairs") print(f"{len(train_pairs)} training pairs") print(f"{len(val_pairs)} validation pairs") print(f"{len(test_pairs)} test pairs")
118964 total pairs 83276 training pairs 17844 validation pairs 17844 test pairs
文本数据矢量化
我们将使用文本矢量化层的两个实例来矢量化文本数据(一个用于英语,一个用于西班牙语),也就是说,将原始字符串转化为整数序列,其中每个整数代表词汇表中一个单词的索引。
英语层将使用默认的字符串标准化(去掉标点符号)和分割方案(在空白处分割),而西班牙语层将使用自定义标准化,即在要去掉的标点符号集中添加字符 "¿"。
注意:在生产级机器翻译模型中,我不建议去除两种语言中的标点符号。相反,我建议将每个标点符号转化为自己的标记,这可以通过向 TextVectorization 层提供自定义分割函数来实现。
strip_chars = string.punctuation + "¿" strip_chars = strip_chars.replace("[", "") strip_chars = strip_chars.replace("]", "") vocab_size = 15000 sequence_length = 20 batch_size = 64 def custom_standardization(input_string): lowercase = tf_strings.lower(input_string) return tf_strings.regex_replace(lowercase, "[%s]" % re.escape(strip_chars), "") eng_vectorization = TextVectorization( max_tokens=vocab_size, output_mode="int", output_sequence_length=sequence_length, ) spa_vectorization = TextVectorization( max_tokens=vocab_size, output_mode="int", output_sequence_length=sequence_length + 1, standardize=custom_standardization, ) train_eng_texts = [pair[0] for pair in train_pairs] train_spa_texts = [pair[1] for pair in train_pairs] eng_vectorization.adapt(train_eng_texts) spa_vectorization.adapt(train_spa_texts)
接下来,我们将格式化数据集。
在每个训练步骤中,模型将使用源句子和 0 到 N 的目标词来预测目标词 N+1(甚至更多)。
因此,训练数据集将产生一个元组(输入、目标),其中
—— encoder_inputs 是矢量化的源句,encoder_inputs 是 "到目前为止 "的目标句,即用于预测目标句中单词 N+1(及以后)的单词 0 到 N。
—— target 是偏移一步的目标句:它提供了目标句中的下一个词--模型将尝试预测的词。
def format_dataset(eng, spa): eng = eng_vectorization(eng) spa = spa_vectorization(spa) return ( { "encoder_inputs": eng, "decoder_inputs": spa[:, :-1], }, spa[:, 1:], ) def make_dataset(pairs): eng_texts, spa_texts = zip(*pairs) eng_texts = list(eng_texts) spa_texts = list(spa_texts) dataset = tf_data.Dataset.from_tensor_slices((eng_texts, spa_texts)) dataset = dataset.batch(batch_size) dataset = dataset.map(format_dataset) return dataset.cache().shuffle(2048).prefetch(16) train_ds = make_dataset(train_pairs) val_ds = make_dataset(val_pairs)
让我们快速浏览一下序列形状(我们的批次为 64 对,所有序列长度为 20 步):
for inputs, targets in train_ds.take(1): print(f'inputs["encoder_inputs"].shape: {inputs["encoder_inputs"].shape}') print(f'inputs["decoder_inputs"].shape: {inputs["decoder_inputs"].shape}') print(f"targets.shape: {targets.shape}")
inputs["encoder_inputs"].shape: (64, 20) inputs["decoder_inputs"].shape: (64, 20) targets.shape: (64, 20)
建立模型
我们的序列到序列转换器由一个转换编码器(TransformerEncoder)和一个转换解码器(TransformerDecoder)串联而成。为了让模型意识到词序,我们还使用了位置嵌入层(PositionalEmbedding layer)。
源序列将被传递给变换编码器,变换编码器将产生一个新的表示。然后,这个新的表示法将连同迄今为止的目标序列(目标词 0 到 N)一起传递给 TransformerDecoder。然后,TransformerDecoder 将设法预测目标序列中的下一个词(N+1 及以后)。
实现这一点的一个关键细节是因果掩码(参见 TransformerDecoder 上的方法 get_causal_attention_mask())。TransformerDecoder 一次可以看到整个序列,因此我们必须确保它在预测标记 N+1 时只使用目标标记 0 到 N 的信息(否则,它可能会使用未来的信息,这将导致推理时无法使用模型)。
import keras.ops as ops class TransformerEncoder(layers.Layer): def __init__(self, embed_dim, dense_dim, num_heads, **kwargs): super().__init__(**kwargs) self.embed_dim = embed_dim self.dense_dim = dense_dim self.num_heads = num_heads self.attention = layers.MultiHeadAttention( num_heads=num_heads, key_dim=embed_dim ) self.dense_proj = keras.Sequential( [ layers.Dense(dense_dim, activation="relu"), layers.Dense(embed_dim), ] ) self.layernorm_1 = layers.LayerNormalization() self.layernorm_2 = layers.LayerNormalization() self.supports_masking = True def call(self, inputs, mask=None): if mask is not None: padding_mask = ops.cast(mask[:, None, :], dtype="int32") else: padding_mask = None attention_output = self.attention( query=inputs, value=inputs, key=inputs, attention_mask=padding_mask ) proj_input = self.layernorm_1(inputs + attention_output) proj_output = self.dense_proj(proj_input) return self.layernorm_2(proj_input + proj_output) def get_config(self): config = super().get_config() config.update( { "embed_dim": self.embed_dim, "dense_dim": self.dense_dim, "num_heads": self.num_heads, } ) return config class PositionalEmbedding(layers.Layer): def __init__(self, sequence_length, vocab_size, embed_dim, **kwargs): super().__init__(**kwargs) self.token_embeddings = layers.Embedding( input_dim=vocab_size, output_dim=embed_dim ) self.position_embeddings = layers.Embedding( input_dim=sequence_length, output_dim=embed_dim ) self.sequence_length = sequence_length self.vocab_size = vocab_size self.embed_dim = embed_dim def call(self, inputs): length = ops.shape(inputs)[-1] positions = ops.arange(0, length, 1) embedded_tokens = self.token_embeddings(inputs) embedded_positions = self.position_embeddings(positions) return embedded_tokens + embedded_positions def compute_mask(self, inputs, mask=None): if mask is None: return None else: return ops.not_equal(inputs, 0) def get_config(self): config = super().get_config() config.update( { "sequence_length": self.sequence_length, "vocab_size": self.vocab_size, "embed_dim": self.embed_dim, } ) return config class TransformerDecoder(layers.Layer): def __init__(self, embed_dim, latent_dim, num_heads, **kwargs): super().__init__(**kwargs) self.embed_dim = embed_dim self.latent_dim = latent_dim self.num_heads = num_heads self.attention_1 = layers.MultiHeadAttention( num_heads=num_heads, key_dim=embed_dim ) self.attention_2 = layers.MultiHeadAttention( num_heads=num_heads, key_dim=embed_dim ) self.dense_proj = keras.Sequential( [ layers.Dense(latent_dim, activation="relu"), layers.Dense(embed_dim), ] ) self.layernorm_1 = layers.LayerNormalization() self.layernorm_2 = layers.LayerNormalization() self.layernorm_3 = layers.LayerNormalization() self.supports_masking = True def call(self, inputs, encoder_outputs, mask=None): causal_mask = self.get_causal_attention_mask(inputs) if mask is not None: padding_mask = ops.cast(mask[:, None, :], dtype="int32") padding_mask = ops.minimum(padding_mask, causal_mask) else: padding_mask = None attention_output_1 = self.attention_1( query=inputs, value=inputs, key=inputs, attention_mask=causal_mask ) out_1 = self.layernorm_1(inputs + attention_output_1) attention_output_2 = self.attention_2( query=out_1, value=encoder_outputs, key=encoder_outputs, attention_mask=padding_mask, ) out_2 = self.layernorm_2(out_1 + attention_output_2) proj_output = self.dense_proj(out_2) return self.layernorm_3(out_2 + proj_output) def get_causal_attention_mask(self, inputs): input_shape = ops.shape(inputs) batch_size, sequence_length = input_shape[0], input_shape[1] i = ops.arange(sequence_length)[:, None] j = ops.arange(sequence_length) mask = ops.cast(i >= j, dtype="int32") mask = ops.reshape(mask, (1, input_shape[1], input_shape[1])) mult = ops.concatenate( [ops.expand_dims(batch_size, -1), ops.convert_to_tensor([1, 1])], axis=0, ) return ops.tile(mask, mult) def get_config(self): config = super().get_config() config.update( { "embed_dim": self.embed_dim, "latent_dim": self.latent_dim, "num_heads": self.num_heads, } ) return config
接下来,我们组装端到端模型。
embed_dim = 256 latent_dim = 2048 num_heads = 8 encoder_inputs = keras.Input(shape=(None,), dtype="int64", name="encoder_inputs") x = PositionalEmbedding(sequence_length, vocab_size, embed_dim)(encoder_inputs) encoder_outputs = TransformerEncoder(embed_dim, latent_dim, num_heads)(x) encoder = keras.Model(encoder_inputs, encoder_outputs) decoder_inputs = keras.Input(shape=(None,), dtype="int64", name="decoder_inputs") encoded_seq_inputs = keras.Input(shape=(None, embed_dim), name="decoder_state_inputs") x = PositionalEmbedding(sequence_length, vocab_size, embed_dim)(decoder_inputs) x = TransformerDecoder(embed_dim, latent_dim, num_heads)(x, encoded_seq_inputs) x = layers.Dropout(0.5)(x) decoder_outputs = layers.Dense(vocab_size, activation="softmax")(x) decoder = keras.Model([decoder_inputs, encoded_seq_inputs], decoder_outputs) decoder_outputs = decoder([decoder_inputs, encoder_outputs]) transformer = keras.Model( [encoder_inputs, decoder_inputs], decoder_outputs, name="transformer" )
训练我们的模型
我们将使用准确率来快速监控验证数据的训练进度。请注意,机器翻译通常使用 BLEU 分数以及其他指标,而不是准确率。
这里我们只训练了 1 个轮次,但要使模型真正收敛,至少要训练 30 个轮次。
epochs = 1 # This should be at least 30 for convergence transformer.summary() transformer.compile( "rmsprop", loss="sparse_categorical_crossentropy", metrics=["accuracy"] ) transformer.fit(train_ds, epochs=epochs, validation_data=val_ds)
Model: "transformer"
┏━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━┓ ┃ Layer (type) ┃ Output Shape ┃ Param # ┃ Connected to ┃ ┡━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━┩ │ encoder_inputs │ (None, None) │ 0 │ - │ │ (InputLayer) │ │ │ │ ├─────────────────────┼───────────────────┼─────────┼──────────────────────┤ │ positional_embeddi… │ (None, None, 256) │ 3,845,… │ encoder_inputs[0][0] │ │ (PositionalEmbeddi… │ │ │ │ ├─────────────────────┼───────────────────┼─────────┼──────────────────────┤ │ decoder_inputs │ (None, None) │ 0 │ - │ │ (InputLayer) │ │ │ │ ├─────────────────────┼───────────────────┼─────────┼──────────────────────┤ │ transformer_encoder │ (None, None, 256) │ 3,155,… │ positional_embeddin… │ │ (TransformerEncode… │ │ │ │ ├─────────────────────┼───────────────────┼─────────┼──────────────────────┤ │ functional_5 │ (None, None, │ 12,959… │ decoder_inputs[0][0… │ │ (Functional) │ 15000) │ │ transformer_encoder… │ └─────────────────────┴───────────────────┴─────────┴──────────────────────┘
Total params: 19,960,216 (76.14 MB)
Trainable params: 19,960,216 (76.14 MB)
Non-trainable params: 0 (0.00 B)
5/1302 [37m━━━━━━━━━━━━━━━━━━━━ 42s 33ms/step - accuracy: 0.3558 - loss: 8.3596 WARNING: All log messages before absl::InitializeLog() is called are written to STDERR I0000 00:00:1699484373.932513 76082 device_compiler.h:187] Compiled cluster using XLA! This line is logged at most once for the lifetime of the process. 1302/1302 ━━━━━━━━━━━━━━━━━━━━ 64s 39ms/step - accuracy: 0.7073 - loss: 2.2372 - val_accuracy: 0.7329 - val_loss: 1.6477
解码测试句子
最后,我们来演示如何翻译全新的英语句子。我们只需将矢量化的英语句子和目标标记"[start]"输入模型,然后重复生成下一个标记,直到遇到标记"[end]"。
spa_vocab = spa_vectorization.get_vocabulary() spa_index_lookup = dict(zip(range(len(spa_vocab)), spa_vocab)) max_decoded_sentence_length = 20 def decode_sequence(input_sentence): tokenized_input_sentence = eng_vectorization([input_sentence]) decoded_sentence = "[start]" for i in range(max_decoded_sentence_length): tokenized_target_sentence = spa_vectorization([decoded_sentence])[:, :-1] predictions = transformer([tokenized_input_sentence, tokenized_target_sentence]) # ops.argmax(predictions[0, i, :]) is not a concrete value for jax here sampled_token_index = ops.convert_to_numpy( ops.argmax(predictions[0, i, :]) ).item(0) sampled_token = spa_index_lookup[sampled_token_index] decoded_sentence += " " + sampled_token if sampled_token == "[end]": break return decoded_sentence test_eng_texts = [pair[0] for pair in test_pairs] for _ in range(30): input_sentence = random.choice(test_eng_texts) translated = decode_sequence(input_sentence)
经过 30 个纪元后,我们得到的结果是:
她把钱递给了他[开始] Ella le pasó el dinero [结束]
汤姆从没听玛丽唱过歌[Tom nunca ha oído cantar a Mary [end]
也许她明天会来。[start] tal vez ella vendrá mañana [end]
我喜欢写作。[开头] 我喜欢写作 [结尾]
他的法语正在一点一点地进步。[开始] su francés va a [UNK] sólo un poco [结束]
我住的酒店让我给你打电话。[开始] 我住的酒店让我给你打电话 [结束]
还没有评论,来说两句吧...