【机器学习】基于顺序到顺序Transformer机器翻译

引言

1.1 序列到序列模型详解

序列到序列(Seq2Seq)模型是深度学习中处理序列数据转换问题的关键架构。在自然语言处理(NLP)任务中，如机器翻译、文本摘要和聊天机器人等，Seq2Seq模型能够高效地将输入序列转换为期望的输出序列。

• 模型架构：

  • 编码器：作为模型的第一部分，负责读取并编码输入序列，生成一个包含输入信息的上下文向量。
  • 解码器：基于编码器提供的上下文向量，逐步生成输出序列，初始状态通常设置为编码器的最终状态。

• 工作流程：

  • 编码器通过时间步逐步处理输入序列，更新内部状态，最终产生上下文向量。
  • 解码器利用上下文向量和自身的状态，预测并生成输出序列的每个元素。

• 应用实例：

  • 在机器翻译中，Seq2Seq模型能够将一种语言的文本转换为另一种语言。
  • 文本摘要任务中，模型从长篇文章中提取关键信息，生成简短摘要。
  • 聊天机器人通过理解用户输入并生成流畅的对话回复。

• 特性优势：

  • 灵活性：能够处理不同长度的输入和输出序列。
  • 上下文依赖性：有效捕获输入序列中的长期依赖关系，对处理复杂语言结构至关重要。

• 技术扩展：

  • 注意力机制：提高模型性能，使解码器在生成时能够关注输入序列的相关部分。
  • 双向编码器：通过同时从两个方向读取输入序列，增强模型对信息的捕捉能力。

1.2 Transformer架构深度解析

Transformer模型，自2017年提出以来，已成为NLP领域的革命性技术，特别是在处理序列数据方面展现出卓越的性能。

• 架构组成：

  • 编码器：由多个层组成，每层包括自注意力层和前馈网络，专注于捕捉序列内部的依赖关系。
  • 解码器：同样由多层构成，增加了Encoder-Decoder注意力层，以关注输入序列中与当前输出最相关的部分。

• 自注意力机制：

  • 核心特性，允许模型在处理每个序列元素时，考虑整个序列，从而捕捉长距离依赖。

• 优点：

  • 高效性：并行处理能力，尤其在处理大规模数据集时，Transformer模型展现出显著的效率。
  • 上下文感知：通过自注意力机制，模型能够更好地理解和处理语言的上下文信息。
  • 预训练与微调：大规模预训练可以捕捉丰富的语言模式，微调则使模型适应特定任务。

• 局限性：

  • 资源需求：对数据量和计算资源的高需求可能限制了模型的普及和应用。
  • 解释性：模型的复杂性导致其决策过程难以解释。
  • 长序列处理：尽管自注意力有助于捕捉长距离依赖，但在极长序列上仍面临挑战。

1.3 机器翻译的现代视角

机器翻译作为NLP的重要组成部分，正随着技术的发展而不断进化。

1.3.1 工作原理深化

• 数据预处理：对平行语料进行清洗、分词、标注，为模型训练提供高质量的输入。
• 模型训练：利用大量双语数据训练模型，通过优化算法调整参数，最小化预测误差。
• 解码与生成：模型将源语言文本转换为目标语言文本，通过评分机制选择最佳翻译。

1.3.2 应用场景扩展

• 实时翻译服务：提供即时的跨语言交流能力，如旅行助手、国际会议等。
• 跨语言内容创作：帮助内容创作者触及更广泛的受众。
• 语言教育：辅助语言学习者理解不同语言的表达方式。

1.3.3 挑战与未来趋势

• 深层语义理解：提高模型对语言深层含义的把握，以生成更自然、准确的翻译。
• 领域适应性：开发能够快速适应特定领域术语和表达方式的模型。
• 实时翻译技术：优化模型以满足实时翻译的高速度和高准确性要求。

未来，随着深度学习技术的不断进步，机器翻译的准确性和效率将得到显著提升。同时，跨语言理解、多模态翻译等新兴领域将推动机器翻译向更深层次的智能化发展。

2.机器翻译实例

2.1.实例的主要内容

在这个例子中，我们将构建一个序列到序列（seq2seq）的Transformer模型，用于执行英语到西班牙语的机器翻译任务。

您将掌握以下技能：

1. 使用Keras的TextVectorization层将文本数据转换为向量表示。
2. 设计和实现TransformerEncoder层、TransformerDecoder层以及PositionalEmbedding层。
3. 准备用于训练seq2seq模型的数据集。
4. 应用训练好的模型进行序列到序列的推理，生成未见过的输入句子的翻译。

这里提供的代码基于《Python深度学习》第二版（第11章：文本深度学习）的示例，但做了适当的简化和调整。如果您想深入了解每个组件的工作原理以及Transformer的理论基础，我建议您阅读该书的相关章节。

2.2. 设置

# 设置后端为TensorFlow。此代码可以与'tensorflow'和'torch'一起工作。
# 它不适用于JAX，因为在`TransformerDecoder.get_causal_attention_mask()`中
# 使用的`jax.numpy.tile`在jit作用域下的行为问题：
# JAX中的`tile`不支持动态的`reps`参数。
# 您可以通过在`get_causal_attention_mask`方法内部使用装饰器来避免jit编译，
# 使代码在JAX上工作：`with jax.ensure_compile_time_eval():`。
import os

# 设置环境变量以使用TensorFlow作为Keras的后端
os.environ["KERAS_BACKEND"] = "tensorflow"

# 导入所需的库
import pathlib  # 用于路径操作
import random  # 用于生成随机数
import string  # 包含字符串常量
import re  # 正则表达式库
import numpy as np  # 科学计算库

# 导入TensorFlow的数据API
import tensorflow.data as tf_data
# 导入TensorFlow的字符串操作API
import tensorflow.strings as tf_strings

# 导入Keras库
import keras
# 从keras库中导入layers模块，用于构建神经网络层
from keras import layers
# 从keras库中导入ops模块，包含操作函数
from keras import ops
# 从keras.layers中导入TextVectorization，用于文本数据的向量化处理
from keras.layers import TextVectorization

2.3.数据预处理

2.3.1.下载数据

以下代码是使用Keras提供的get_file函数来下载并解压一个名为"spa-eng.zip"的文件，其中包含西班牙语到英语的机器翻译数据集。

from keras.utils import get_file
import pathlib

# 使用Keras的get_file函数下载文件
# fname: 要下载的文件名
# origin: 文件的原始URL
# extract: 是否解压文件
text_file_zip = keras.utils.get_file(
    fname="spa-eng.zip",
    origin="http://storage.googleapis.com/download.tensorflow.org/data/spa-eng.zip",
    extract=True,  # 设置为True以自动解压下载的文件
)

# 使用pathlib.Path转换下载文件的路径，并获取解压后的spa.txt文件路径
# parent: 获取父目录路径
# "spa-eng" / "spa.txt": 指定子目录和文件名
text_file = pathlib.Path(text_file_zip).parent / "spa-eng" / "spa.txt"

# 打印spa.txt文件的完整路径
print(f"spa.txt 文件路径: {text_file}")

2.3.2.数据解析

每行数据都包含一个英文句子（作为源序列）及其对应的西班牙语译文（作为目标序列）。为了标识句子的开始和结束，我们在西班牙语句子的开头添加"[start]“标记，并在结尾添加”[end]"标记。

# 打开spa.txt文件进行读取
with open(text_file, 'r', encoding='utf-8') as f:  # 指定编码为'utf-8'，防止编码错误
    # 读取所有行，以换行符分割，去除最后一个空元素
    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))

# 随机打印5个文本对样本
for _ in range(5):
    # 使用random.choice随机选择一个文本对
    print(random.choice(text_pairs))

这段代码的功能主要包括以下几个步骤：

1. 文件读取：使用open函数打开一个文本文件（text_file），这个文件是通过之前下载和解压得到的。

2. 数据分割：读取文件的全部内容，以换行符\n分割成行，存储到lines列表中。[:-1]用于去除最后一个空元素，这通常发生在文件最后一行后没有换行符的情况下。

3. 文本对准备：初始化一个空列表text_pairs，用于存储处理后的文本对。

4. 数据解析与处理：

   • 遍历lines列表中的每一行。
   • 使用split("\t")以制表符为分隔符将每行分割成英文和西班牙语文本。
   • 对西班牙语文本添加特定的标记"[start] "和" [end]"，这通常用于机器翻译任务中的序列开始和结束标记。

5. 文本对存储：将处理好的英文和西班牙语文本对添加到text_pairs列表中。

6. 随机样本打印：从text_pairs列表中随机选择并打印5个文本对样本。这是为了展示数据的一个随机抽样，以便于观察数据格式和内容。

总结来说，这段代码的目的是从一个文本文件中提取并格式化数据，然后展示这些数据的一个随机样本。这在机器学习和自然语言处理任务中是一个常见的数据预处理步骤，特别是在准备机器翻译模型的训练数据时。
打印的内容如下：

("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]')

2.3.3.划分数据集

我们将句子数据集分割为训练集、验证集和测试集。

import random

# 对文本对列表进行随机打乱，以确保数据的随机性
random.shuffle(text_pairs)

# 计算验证集样本数量，设置为总样本数量的15%
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)} 总文本对数")

# 打印训练集的文本对数量
print(f"{len(train_pairs)} 训练集文本对数")

# 打印验证集的文本对数量
print(f"{len(val_pairs)} 验证集文本对数")

# 打印测试集的文本对数量
print(f"{len(test_pairs)} 测试集文本对数")

这段代码的功能是对文本对列表进行随机打乱，并按照一定的比例划分为训练集、验证集和测试集，然后打印出各个数据集的大小。以下是详细的步骤解释：

1. 随机打乱：使用random.shuffle函数对text_pairs列表进行随机打乱，确保数据的随机性，这对于训练模型时避免偏差是很重要的。

2. 计算样本数量：

   • 使用len(text_pairs)获取文本对的总数。
   • 计算验证集样本数，通常是总样本数的15%，使用int(0.15 * len(text_pairs))进行计算。
   • 计算训练集样本数，为总样本数减去两倍的验证集样本数，即len(text_pairs) - 2 * num_val_samples。

3. 数据集划分：

   • 根据计算出的样本数量，使用切片操作将text_pairs列表划分为训练集train_pairs、验证集val_pairs和测试集test_pairs。

4. 打印数据集大小：

   • 打印总的文本对数量len(text_pairs)。
   • 打印训练集的大小len(train_pairs)。
   • 打印验证集的大小len(val_pairs)。
   • 打印测试集的大小len(test_pairs)。

代码的输出将提供对数据集划分情况的直观了解，这对于监督学习模型的训练和评估是非常关键的。通常，训练集用于模型训练，验证集用于模型调参，测试集用于最终评估模型性能。在这段代码中，测试集的大小与验证集相同，但实际应用中可以根据需要调整比例。

2.3.4.文本矢量化

我们将使用两个TextVectorization层的实例来将文本数据转换为数值表示（一个用于英语，一个用于西班牙语），也就是将原始字符串转换为整数序列，其中每个整数代表词汇表中对应单词的索引。

对于英语层，我们将使用默认的字符串标准化（去除标点符号）和分词策略（基于空白字符进行分词）。而对于西班牙语层，我们将采用自定义的标准化，即除了默认的标点符号外，还会额外去除字符"¿"。

然而，在实际生产环境的机器翻译模型中，并不推荐去除任何一种语言的标点符号。相反，更推荐的做法是将每个标点符号视为一个独立的标记，这可以通过为TextVectorization层提供一个自定义的分词函数来实现。

以下代码是用于文本数据的预处理和向量化的，主要包含以下步骤：

1. 定义去除字符：创建一个字符串strip_chars，包含所有要去除的标点符号和一个特殊的字符"¿“。然后从这个字符串中移除”[“和”]"，因为它们可能用于正则表达式中。

2. 设置词汇表大小和序列长度：设置词汇表的大小vocab_size为15000，序列长度sequence_length为20，批大小batch_size为64。

3. 自定义标准化函数：定义custom_standardization函数，使用TensorFlow的字符串操作API来将输入字符串转换为小写，并使用正则表达式替换掉所有strip_chars中定义的字符。

4. 文本向量化：创建两个TextVectorization实例，eng_vectorization和spa_vectorization，分别用于英文和西班牙语文本的向量化。它们将文本转换为整数序列，其中英文的输出序列长度固定，西班牙语的输出序列长度多一个以适应开始和结束标记。

5. 数据适配：使用训练集中的英文和西班牙语文本列表，调用adapt方法来适配向量化器，这样它们就可以根据这些文本数据构建词汇表。

import string
import re
import tensorflow as tf
from tensorflow.keras.layers import TextVectorization

# 定义要去除的字符，包括所有标点和特殊字符"¿"
strip_chars = string.punctuation + "¿"

# 从字符中移除方括号，因为它们可能用于正则表达式
strip_chars = strip_chars.replace("[", "").replace("]", "")

# 设置词汇表大小、序列长度和批大小
vocab_size = 15000
sequence_length = 20
batch_size = 64

# 自定义标准化函数，用于文本的小写转换和去除特定字符
def custom_standardization(input_string):
    lowercase = tf.strings.lower(input_string)  # 转换为小写
    # 使用正则表达式替换掉strip_chars中定义的字符
    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)

2.3.5.数据格式化

以下代码定义了两个函数，用于准备和格式化机器翻译任务的数据集，然后将格式化的数据集用于训练和验证：

# 导入必要的库
import tensorflow as tf
from tensorflow.keras.layers import TextVectorization
import tensorflow_datasets as tfds

# 假定全局变量已有定义：eng_vectorization 和 spa_vectorization 是两个 TextVectorization 实例

def format_dataset(eng, spa):
    # 使用英文向量化器处理英文数据
    eng = eng_vectorization(eng)
    # 使用西班牙语向量化器处理西班牙语数据
    spa = spa_vectorization(spa)
    
    # 格式化数据集，准备编码器输入和解码器输入
    # 解码器输入需要移除序列的第一个元素，因为我们使用teacher forcing技术
    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)
    
    # 从文本列表创建TensorFlow数据集
    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)

# 使用make_dataset函数创建训练和验证数据集
train_ds = make_dataset(train_pairs)
val_ds = make_dataset(val_pairs)

format_dataset 函数：

• 这个函数接收英文和西班牙语文本作为输入。
• 使用之前定义的向量化器将文本转换为整数序列。
• 准备编码器的输入（整个英文序列）和解码器的输入（除去最后一个元素的西班牙语序列），因为我们将在训练中使用teacher forcing。

make_dataset 函数：

• 这个函数接收文本对的列表。
• 使用zip函数将英文和西班牙语文本对解包，并转换为列表。
• 使用tf_data.Dataset.from_tensor_slices从文本列表创建TensorFlow数据集。
• 使用batch方法将数据集分批处理。
• 使用map方法应用format_dataset函数来格式化数据集。
• 使用cache、shuffle和prefetch方法来优化数据集的性能。

创建数据集：使用make_dataset函数分别创建训练数据集train_ds和验证数据集val_ds。

这些步骤是准备数据集以供深度学习模型训练的标准流程，特别是在自然语言处理任务中。通过这种方式，数据被有效地加载和预处理，以供模型训练使用。

2.4.建立模型

我们的序列到序列Transformer模型是基于TransformerEncoder和TransformerDecoder构建的。为了确保模型能够识别单词的顺序，我们引入了位置嵌入（Positional Embedding）层。

在模型的工作流程中，源序列首先通过TransformerEncoder进行处理，生成其新的表示形式。随后，这个新的表示形式与当前已知的目标序列（即目标单词0到N）一同输入到TransformerDecoder中。TransformerDecoder的任务是预测目标序列中的下一个单词（即N+1及之后的单词）。

为了实现这一目标，TransformerDecoder采用了因果掩码（Causal Masking）这一关键机制。由于TransformerDecoder能够一次性看到整个序列，我们必须确保在预测第N+1个单词时，它仅依赖于目标序列中前N个单词的信息（即不能使用未来的信息），这样才能确保模型在推理时能够正常工作。

import tensorflow as tf
from tensorflow.keras.layers import Layer, LayerNormalization, Embedding, MultiHeadAttention, Dense, Sequential

class TransformerEncoder(Layer):
    def __init__(self, embed_dim, dense_dim, num_heads, **kwargs):
        super(TransformerEncoder, self).__init__(**kwargs)
        # 嵌入维度，即模型中每个单词向量的维度
        self.embed_dim = embed_dim
        # 密集层的维度，通常大于embed_dim
        self.dense_dim = dense_dim
        # 多头注意力机制中的头数
        self.num_heads = num_heads
        # 初始化多头注意力层
        self.attention = MultiHeadAttention(num_heads=num_heads, key_dim=embed_dim)
        # 前馈网络，用于处理注意力机制的输出
        self.dense_proj = Sequential([
            Dense(dense_dim, activation="relu"),
            Dense(embed_dim)
        ])
        # 层归一化，用于稳定训练过程
        self.layernorm_1 = LayerNormalization()
        self.layernorm_2 = LayerNormalization()
        # 支持掩码
        self.supports_masking = True

    def call(self, inputs, mask=None):
        # 如果提供了掩码，将其转换为适合注意力机制的格式
        if mask is not None:
            padding_mask = tf.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(TransformerEncoder, self).get_config()
        config.update({
            "embed_dim": self.embed_dim,
            "dense_dim": self.dense_dim,
            "num_heads": self.num_heads,
        })
        return config

class PositionalEmbedding(Layer):
    def __init__(self, sequence_length, vocab_size, embed_dim, **kwargs):
        super(PositionalEmbedding, self).__init__(**kwargs)
        # 序列的最大长度
        self.sequence_length = sequence_length
        # 词汇表的大小
        self.vocab_size = vocab_size
        # 嵌入维度
        self.embed_dim = embed_dim
        # 词嵌入层，将词汇表中的每个词映射到嵌入空间
        self.token_embeddings = Embedding(input_dim=vocab_size, output_dim=embed_dim)
        # 位置嵌入层，为每个位置提供唯一的编码
        self.position_embeddings = Embedding(input_dim=sequence_length, output_dim=embed_dim)

    def call(self, inputs):
        # 获取输入序列的长度
        length = tf.shape(inputs)[-1]
        # 为每个位置生成一个位置向量
        positions = tf.range(start=0, limit=length, delta=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 tf.not_equal(inputs, 0)

    def get_config(self):
        config = super(PositionalEmbedding, self).get_config()
        config.update({
            "sequence_length": self.sequence_length,
            "vocab_size": self.vocab_size,
            "embed_dim": self.embed_dim,
        })
        return config

class TransformerDecoder(Layer):
    def __init__(self, embed_dim, latent_dim, num_heads, **kwargs):
        super(TransformerDecoder, self).__init__(**kwargs)
        # 嵌入维度
        self.embed_dim = embed_dim
        # 解码器中前馈网络的维度
        self.latent_dim = latent_dim
        # 多头注意力机制中的头数
        self.num_heads = num_heads
        # 自注意力层
        self.attention_1 = MultiHeadAttention(num_heads=num_heads, key_dim=embed_dim)
        # 编码器-解码器注意力层
        self.attention_2 = MultiHeadAttention(num_heads=num_heads, key_dim=embed_dim)
        # 解码器中的前馈网络
        self.dense_proj = Sequential([
            Dense(latent_dim, activation="relu"),
            Dense(embed_dim)
        ])
        # 层归一化
        self.layernorm_1 = LayerNormalization()
        self.layernorm_2 = LayerNormalization()
        self.layernorm_3 = 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 = tf.cast(mask[:, None, :], dtype="int32")
            padding_mask = tf.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 = tf.shape(inputs)
        batch_size, sequence_length = input_shape[0], input_shape[1]
        i = tf.range(start=sequence_length - 1, limit=-1, delta=-1)
        j = tf.range(start=0, limit=sequence_length, delta=1)
        mask = tf.cast(i >= j, dtype="int32")
        mask = tf.reshape(mask, (sequence_length, sequence_length))
        mult = tf.concat([
            tf.expand_dims(batch_size, -1),
            tf.ones([1], dtype=tf.int32)
        ], axis=0)
        return tf.tile(mask, mult)

    def get_config(self):
        config = super(TransformerDecoder, self).get_config()
        config.update({
            "embed_dim": self.embed_dim,
            "latent_dim": self.latent_dim,
            "num_heads": self.num_heads,
        })
        return config

这段代码定义了三个类，用于构建Transformer模型的编码器、位置编码和解码器。每个类都包含了初始化方法、前向传播方法、掩码计算方法（对于位置编码）和配置获取方法。这些类可以作为构建更复杂模型的组件，例如机器翻译或文本摘要任务中的Transformer模型。
接着，就可以实现类的实例化，建立端到端的训练模型：

# 定义嵌入维度、潜在空间维度和注意力头数  
embed_dim = 256  
latent_dim = 2048  
num_heads = 8  
  
# 假设 sequence_length 和 vocab_size 已经在前面定义  
# 序列长度（通常是最长句子的长度）  
# 词汇表大小（所有可能单词的数量）  
  
# 定义编码器输入  
encoder_inputs = keras.Input(shape=(None,), dtype="int64", name="encoder_inputs")  # 形状为(None,)表示任意长度的序列  
  
# 编码器位置嵌入和Transformer编码器  
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, name="encoder")  # 命名编码器模型  
  
# 定义解码器输入和解码器状态输入（即编码器的输出）  
decoder_inputs = keras.Input(shape=(None,), dtype="int64", name="decoder_inputs")  
encoded_seq_inputs = keras.Input(shape=(None, embed_dim), name="encoded_seq_inputs")  
  
# 解码器位置嵌入和Transformer解码器  
x = PositionalEmbedding(sequence_length, vocab_size, embed_dim)(decoder_inputs)  
x = TransformerDecoder(embed_dim, latent_dim, num_heads)([x, encoded_seq_inputs])  # TransformerDecoder需要两个输入  
x = layers.Dropout(0.5)(x)  # 添加Dropout层防止过拟合  
decoder_outputs = layers.Dense(vocab_size, activation="softmax")(x)  # 输出层，通过softmax得到概率分布  
  
# 构建解码器模型，它接受解码器输入和编码器输出作为输入  
decoder = keras.Model([decoder_inputs, encoded_seq_inputs], decoder_outputs, name="decoder")  
  
# 构建完整的Transformer模型，它接受编码器和解码器输入，输出解码器的预测结果  
transformer = keras.Model(  
    [encoder_inputs, decoder_inputs], decoder_outputs, name="transformer"  
)  
  
# 代码功能总结：  
# 该代码定义了一个基于Transformer的序列到序列模型，包括一个编码器和一个解码器。  
# 编码器接受源序列（encoder_inputs）作为输入，经过位置嵌入和Transformer编码器层后输出编码后的序列。  
# 解码器接受目标序列的当前部分（decoder_inputs）和编码器的输出（encoded_seq_inputs）作为输入，  
# 经过位置嵌入、Transformer解码器层、Dropout层和输出层后，输出对下一个目标单词的预测。  
# 最终，将编码器和解码器组合成一个完整的Transformer模型。

2.5.训练模型

在训练我们的Transformer模型时，我们会使用准确率作为验证数据上训练进度的快速监控指标。然而，值得注意的是，对于机器翻译任务，通常使用BLEU分数以及其他指标来评估模型的性能，而不仅仅是准确率。

为了简单起见，这里我们只进行了一个周期的训练，但为了达到模型的实际收敛效果，你应该至少进行30个周期的训练。

以下代码展示了如何使用Keras框架来训练一个Transformer模型。

import tensorflow as tf
from tensorflow.keras.models import Model

# 假设transformer是一个已经定义好的Transformer模型实例

# 设置训练的轮数，至少需要30轮才能达到收敛
epochs = 1  

# 打印模型的概览信息，包括层、参数数量等
transformer.summary()

# 编译模型，设置优化器、损失函数和评估指标
# "rmsprop" 是优化器的名字，用于调整网络权重
# "sparse_categorical_crossentropy" 是用于多分类问题的损失函数
# ["accuracy"] 是训练过程中要监控的指标
transformer.compile(
    optimizer="rmsprop",
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"]
)

# 训练模型
# train_ds 是训练数据集
# val_ds 是验证数据集
# epochs 是训练的轮数
# 在实际应用中，epochs 的值应该设置得更高以确保模型能够收敛
transformer.fit(
    train_ds,  # 训练数据集
    epochs=epochs,  # 训练的轮数
    validation_data=val_ds  # 验证数据集，用于评估模型在未见过的数据上的表现
)

代码主要实现以下功能：

1. 设置训练轮数 (epochs)：定义了模型需要训练的数据遍历次数。通常需要多次迭代以达到较好的性能。
2. 模型概览 (transformer.summary())：输出模型的详细信息，包括每层的名称、参数数量等，帮助了解模型结构。
3. 模型编译 (transformer.compile)：配置模型的训练参数，包括选择优化器(optimizer)、损失函数(loss)和评估指标(metrics)。
   • 优化器 (optimizer)：算法用于调整网络权重以最小化损失函数。
   • 损失函数 (loss)：衡量模型预测与实际值差异的函数，训练过程中尝试最小化此值。
   • 评估指标 (metrics)：除了损失函数外，还可以监控其他指标，如准确率。
4. 模型训练 (transformer.fit)：使用提供的训练数据和验证数据训练模型，并进行多轮迭代，直到达到设定的训练轮数或满足早停条件。

请注意，实际使用中，epochs 的值应该根据模型训练的收敛情况来调整，通常需要更多轮次以达到较好的性能。此外，train_ds 和 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)

2.6.机器翻译实践

以下代码演示了如何使用一个预训练的Transformer模型进行序列解码，即将英文句子翻译成西班牙语。

# 假设spa_vectorization是西班牙语的向量化处理对象
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]
        # 将输入和目标序列传入Transformer模型进行预测
        predictions = transformer([tokenized_input_sentence, tokenized_target_sentence])
        
        # 获取模型预测的最高概率的token索引
        sampled_token_index = tf.convert_to_numpy(
            tf.argmax(predictions[0, i, :])
        ).item(0)  # 使用.item(0)获取numpy数组的第一个元素
        # 根据索引获取对应的token
        sampled_token = spa_index_lookup[sampled_token_index]
        # 将预测的token添加到解码句子中
        decoded_sentence += " " + sampled_token

        # 如果预测的token是结束符，则结束解码
        if sampled_token == "[end]":
            break
    # 返回解码完成的句子
    return decoded_sentence

# 假设test_pairs是测试数据对的列表，test_eng_texts是提取的英文文本列表
test_eng_texts = [pair[0] for pair in test_pairs]
for _ in range(30):  # 进行30次解码测试
    # 随机选择一个英文句子
    input_sentence = random.choice(test_eng_texts)
    # 解码句子并打印翻译结果
    translated = decode_sequence(input_sentence)
    print(translated)

上述代码主要有以下功能：

1. 获取词汇表 (spa_vocab)：从西班牙语的向量化处理对象中获取词汇表。
2. 创建索引查找表 (spa_index_lookup)：创建一个字典，用于根据索引快速查找词汇。
3. 定义解码函数 (decode_sequence)：定义一个函数，用于将英文句子解码（翻译）成西班牙语句子。
   • 使用英文向量化处理输入句子。
   • 初始化解码句子，并在最大长度内循环解码。
   • 在每次循环中，使用西班牙语向量化处理当前解码的句子，并传入模型进行预测。
   • 根据模型预测选择概率最高的token，并添加到解码句子中。
   • 如果预测到结束符，则结束解码。
4. 测试解码功能：从测试数据集中随机选择句子，使用解码函数进行翻译，并打印翻译结果。

请注意，代码中用到的eng_vectorization、spa_vectorization和transformer应该是已经定义好的对象，分别用于英文的向量化处理、西班牙语的向量化处理和Transformer模型。此外，test_pairs应该是包含测试数据对的列表。
经过30个周期的训练后，我们得到了如下结果：

她把钱递给了他。
[start] ella le pasó el dinero [end]
（翻译正确）

汤姆从未听过玛丽唱歌。
[start] tom nunca ha oído cantar a mary [end]
（注意：在西班牙语中，人名通常不大写，所以"mary"应该小写为"mary"）

也许她明天会来。
[start] tal vez ella vendrá mañana [end]
（翻译正确）

我喜欢写作。
[start] me encanta escribir [end]
（翻译正确）

他的法语正在一点一点地提高。
[start] su francés va a [UNK] sólo un poco [end]
（翻译存在问题，“va a” 后面缺少了动词的将来时态形式，例如 “mejorar”，所以 “[UNK]” 应该替换为 “mejorar”）

我的酒店告诉我给你打电话。
[start] mi hotel me dijo que te [UNK] [end]
（翻译存在问题，“[UNK]” 应该替换为 “llame” 或 “llamar”，完整的句子应该是 “mi hotel me dijo que te llamara” 或 “mi hotel me dijo que te llamara a ti”，但后者在西班牙语中可能显得有些冗余）

3.总结和展望

3.1.总结

本文详细介绍了序列到序列(Seq2Seq)模型和Transformer架构在机器翻译任务中的应用。通过逐步解析和代码示例，我们对以下关键概念和技术有了深入的理解：

1. Seq2Seq模型：作为处理序列转换问题的核心架构，Seq2Seq模型包括编码器和解码器两个主要部分，能够有效处理不同长度的输入和输出序列。

2. Transformer模型：自2017年提出以来，Transformer模型已成为NLP领域的核心技术，其核心特性包括自注意力机制和并行处理能力，使其在处理长距离依赖和大规模数据集时表现出色。

3. 编码器和解码器实现：文中通过代码示例展示了如何使用Keras框架实现Transformer模型中的编码器和解码器，包括多头注意力、前馈网络和层归一化等关键组件。

4. 数据预处理和向量化：介绍了如何使用TextVectorization层将文本数据转换为数值表示，并使用自定义的标准化函数处理特殊字符。

5. 模型训练：展示了如何编译和训练Transformer模型，包括设置优化器、损失函数、评估指标，并使用训练和验证数据集进行模型训练。

6. 序列解码和翻译：提供了使用预训练Transformer模型进行序列解码的示例，即将英文句子翻译成西班牙语，并打印翻译结果。

3.2.展望

尽管Transformer模型在机器翻译等领域取得了显著的成果，但仍存在一些挑战和未来的发展方向：

1. 深层语义理解：提高模型对语言深层含义的把握，生成更自然、准确的翻译。

2. 领域适应性：开发能够快速适应特定领域术语和表达方式的模型。

3. 实时翻译技术：优化模型以满足实时翻译的高速度和高准确性要求。

4. 多模态翻译：探索将视觉、声音等多种模态的信息融合到机器翻译中，提高翻译的丰富性和准确性。

5. 模型解释性：提高模型的可解释性，帮助用户理解模型的决策过程。

6. 资源效率：研究如何降低模型对计算资源的需求，使Transformer模型更加普及。

随着深度学习技术的不断发展，我们有理由相信机器翻译的准确性和效率将得到进一步提升，同时，跨语言理解、多模态翻译等新兴领域将推动机器翻译向更深层次的智能化发展。