Python 人工智能实战：机器翻译
作者：禅与计算机程序设计艺术
1.背景介绍
在日常生活中，我们常会运用计算机来实现语言间的转换与翻译工作。对于从事机器学习、深度学习、自然语言处理（NLP）相关领域的专业人士而言，传统机器翻译技术多依赖基于统计的概率模型，在实际应用中存在一定的局限性；相比之下，现代深度学习方法则通过包括 seq2seq 模型以及 Transformer架构等创新技术实现了显著的进步。今天我们将深入探讨如何借助 Python 的scikit-learn库和TensorFlow框架来构建高效的机器翻译系统，并通过神经网络结构进一步提升模型性能。本文旨在围绕以下几个核心方向展开详细讨论：
该seq2seq模型由两个相互循环的RNN构成，并非简单地'组成'关系。其中一个是编码器（负责输入序列编码），另一个是解码器（负责输出序列解码）。这种架构能够实现从源语言到目标语言的句法和语义转换过程。
-
Attention 机制
attention 机制是构建 seq2seq 模型的核心组件之一。它有助于模型识别源序列各个位置之间的相互依赖关系。
在 attention 模块内，
recurrent neural network 首先利用 attention 机制计算出每个时间步对应的注意力分数，
随后将这些分数作为权重应用于输入序列，
从而生成当前时间步的状态向量。
attention 机制能够辅助构建图结构模型，
有助于分析并整合词语间的关联性。 -
Transformer 模型 该 Transformer 基于多头自注意力机制构建而成的自回归架构。
它相较于 seq2seq 模型而言，
该架构展现出更强的抗长尾能力。
在编码器-解码器结构上取得了显著成效，
该编码器-解码器架构不仅在机器翻译领域取得了显著成效，在自然语言处理相关任务中也得到了广泛的应用。 -
评价指标
为了评估该机器翻译模型的效果，我们构建了一个包含共190,000对双语句子的数据集。该数据集经过去重与清理重复项的操作，并选取约10%作为开发集，剩余部分用于测试。
模型实现部分 在本节中 我们将系统性地展开对以下五个关键技术模块逐一进行深入探讨 包括 Seq2Seq 模型 覆盖 Attention 机制模块 包括 Transformer 基础模型 同时具体阐述这些技术的理论基础与实践应用 这些技术将为后续章节中的具体应用奠定坚实基础
1. Seq2Seq 模型
基于两个循环神经网络（RNN），seq2seq 模型的核心原理是实现序列间的映射关系。具体而言，在编码过程中，模型接收输入序列，并利用双向 LSTM 或 GRU 单元生成反映输入特征的编码向量。随后，在解码过程中，模型将编码向量和特定起始符号（ ）作为解码起点，并逐步生成候选解码结果集中的每个位置上的可能词项。通过综合评估候选解码结果集，在最终步骤中选择最优候选词组合以完成整个解码过程。
import numpy as np
from keras.models import Sequential
from keras.layers import Dense, LSTM, Embedding, RepeatVector, TimeDistributed, Input, Dropout, Bidirectional
# define model
encoder_inputs = Input(shape=(None,))
enc_emb = Embedding(input_dim=vocab_size, output_dim=embedding_dim, mask_zero=True)(encoder_inputs)
encoder = Bidirectional(LSTM(units, return_sequences=True, return_state=True))
encoder_outputs, forward_h, forward_c, backward_h, backward_c = encoder(enc_emb)
state_h = Concatenate()([forward_h, backward_h])
state_c = Concatenate()([forward_c, backward_c])
states = [state_h, state_c]
decoder_inputs = Input(shape=(None,), dtype='int32')
dec_emb_layer = Embedding(input_dim=vocab_size, output_dim=embedding_dim, mask_zero=False)
dec_emb = dec_emb_layer(decoder_inputs)
decoder_lstm = LSTM(units*2, return_sequences=True, return_state=True)
decoder_outputs, _, _ = decoder_lstm(dec_emb, initial_state=[state_h, state_c])
attn_layer = AttnLayer(hidden_size)
context = attn_layer([encoder_outputs, decoder_outputs])
decoder_combined_context = Concatenate(axis=-1, name='concat')([decoder_outputs, context])
output_layer = Dense(vocab_size, activation='softmax')
model = Model([encoder_inputs, decoder_inputs], output_layer(decoder_combined_context))
model.summary()
def decode_sequence(input_seq):
states_values = encoder_model.predict(input_seq)
target_seq = np.zeros((1, 1))
target_seq[0, 0] = word_to_id['start']
stop_condition = False
decoded_sentence = ''
while not stop_condition:
output_tokens, h, c = decoder_model.predict([target_seq] + states_values)
sample_token_index = np.argmax(output_tokens[0, -1, :])
sampled_word = None
for word, index in id_to_word.items():
if index == sample_token_index:
sampled_word = word
break
if (sampled_word!= 'end' and len(decoded_sentence)<maxlen):
decoded_sentence +=''+sampled_word
# Exit condition: either hit max length or find stop word.
if (sampled_word == 'end' or len(decoded_sentence)>=maxlen):
stop_condition = True
# Update the target sequence (of length 1).
target_seq = np.zeros((1, 1))
target_seq[0, 0] = sample_token_index
# Update states
states_values = [h, c]
return decoded_sentence
encoder_model = Model(encoder_inputs, [encoder_outputs, state_h, state_c])
decoder_state_input_h = Input(shape=(latent_dim,))
decoder_state_input_c = Input(shape=(latent_dim,))
decoder_states_inputs = [decoder_state_input_h, decoder_state_input_c]
dec_emb2 = dec_emb_layer(decoder_inputs)
decoder_outputs2, state_h2, state_c2 = decoder_lstm(dec_emb2, initial_state=decoder_states_inputs)
decoder_states2 = [state_h2, state_c2]
context2 = attn_layer([encoder_outputs, decoder_outputs2])
decoder_combined_context2 = Concatenate(axis=-1, name='concat')([decoder_outputs2, context2])
decoder_outputs2 = decoder_dense(decoder_combined_context2)
decoder_model = Model([decoder_inputs] + decoder_states_inputs, [decoder_outputs2] + decoder_states2)
decoder_model.summary()
代码解读2. Attention 机制
该关注机制在Seq2Seq架构中扮演着核心角色。该架构借助注意力权重的能力识别出输入序列各关键位置之间的相互作用；这使得其能够更精确地生成相应的输出信息。其中包含计算注意力权重、选择重要特征以及整合信息的模块；这一设计确保了系统在处理复杂序列数据时表现出色。
生成注意力权重矩阵的过程是先经过全连接层得到一个attention矩阵，随后运用softmax函数来计算出注意力权重值。值得注意的是该attention矩阵的大小为 encoder 输出的hidden_size与 decoder 输出的hidden_size之间的二维空间
在生成注意力矩阵的过程中，在每一个时间步中将上一层隐藏状态作为输入来生成对应的注意力权重分布矩阵，在该权重分布中每一行都对应着当前时间步隐藏状态与源序列中各个位置之间的关联程度。具体而言，在实际应用过程中通常会选择每一步骤中最显著的关注度值所对应的源序列位置作为当前解码阶段的输出词元。
在将注意力矩阵与 decoder 的输出进行点积运算后，在所得结果的基础上叠加一次 attention 矩阵的操作能够有效捕捉不同位置之间的关联关系。这种机制允许模型在解码过程中动态地关注输入序列的不同部分，并据此生成更加连贯的输出序列。在实际应用场景中，则需要根据具体需求选择合适的聚合策略来整合这些权重信息。
具体的代码如下：
class AttnLayer(Layer):
def __init__(self, **kwargs):
super(AttnLayer, self).__init__(**kwargs)
def build(self, input_shape):
assert isinstance(input_shape, list)
assert len(input_shape) == 2
assert len(input_shape[0]) == 3
assert len(input_shape[1]) == 3
self.W = self.add_weight(name='w', shape=(input_shape[0][2],), initializer='uniform', trainable=True)
self.b = self.add_weight(name='b', shape=(input_shape[1][1],), initializer='uniform', trainable=True)
super(AttnLayer, self).build(input_shape)
def call(self, inputs):
enc_outs, dec_outs = inputs
attn_score = K.dot(K.tanh(K.dot(enc_outs, self.W)+self.b), K.transpose(dec_outs))
attn_dist = K.softmax(attn_score)
context = K.sum(attn_dist * enc_outs, axis=1)
return context
代码解读3. Transformer 模型
Transformer 模型是由多个并行的自注意力模块构成的一种多层关注机制模型。不仅能够有效编码整个输入序列的关键信息，还能够聚焦于序列中的局部细节信息。其编码器与解码器均由多个相同层次的自注意力模块与前馈网络共同构成。自注意力机制使模型能够直接聚焦于输入序列中的任意位置信息，在此基础上前馈网络则对输入数据进行后续处理以获取深层特征表示。
具体的代码如下：
def get_angles(pos, i, d_model):
angle_rates = 1 / np.power(10000, (2 * (i//2)) / np.float32(d_model))
return pos * angle_rates
def positional_encoding(position, d_model):
angle_rads = get_angles(np.arange(position)[:, np.newaxis],
np.arange(d_model)[np.newaxis, :],
d_model)
sines = np.sin(angle_rads[:, 0::2])
cosines = np.cos(angle_rads[:, 1::2])
pos_encoding = np.concatenate([sines, cosines], axis=-1)
pos_encoding = pos_encoding[np.newaxis,...]
return tf.cast(pos_encoding, dtype=tf.float32)
class MultiHeadAttention(tf.keras.layers.Layer):
def __init__(self, d_model, num_heads):
super(MultiHeadAttention, self).__init__()
self.num_heads = num_heads
self.d_model = d_model
assert d_model % self.num_heads == 0
self.depth = d_model // self.num_heads
self.wq = tf.keras.layers.Dense(d_model)
self.wk = tf.keras.layers.Dense(d_model)
self.wv = tf.keras.layers.Dense(d_model)
self.dense = tf.keras.layers.Dense(d_model)
def split_heads(self, x, batch_size):
"""Split the last dimension into (num_heads, depth).
Transpose the result such that the shape is (batch_size, num_heads, seq_len, depth)
"""
x = tf.reshape(x, (batch_size, -1, self.num_heads, self.depth))
return tf.transpose(x, perm=[0, 2, 1, 3])
def call(self, v, k, q, mask):
batch_size = tf.shape(q)[0]
q = self.wq(q)
k = self.wk(k)
v = self.wv(v)
q = self.split_heads(q, batch_size)
k = self.split_heads(k, batch_size)
v = self.split_heads(v, batch_size)
scaled_attention, attention_weights = scaled_dot_product_attention(
q, k, v, mask)
scaled_attention = tf.transpose(scaled_attention, perm=[0, 2, 1, 3])
concat_attention = tf.reshape(scaled_attention,
(batch_size, -1, self.d_model))
output = self.dense(concat_attention)
return output, attention_weights
def point_wise_feed_forward_network(d_model, dff):
return tf.keras.Sequential([
tf.keras.layers.Dense(dff, activation='relu'),
tf.keras.layers.Dense(d_model)
])
def encode_sequence(input_seq):
# adding embedding and position encoding.
embedding = token_embedding(input_seq)
embedding *= tf.math.sqrt(tf.cast(model_dimensions, tf.float32))
embedding += positional_encoding(tf.shape(input_seq)[1], model_dimensions)
encoded_text = embedding
# creating multiple layers of the transformer block.
for i in range(num_layers):
multihead_attention = MultiHeadAttention(model_dimensions, num_heads)
ff_layer = point_wise_feed_forward_network(model_dimensions, inner_dimension)
encoded_text, _ = multihead_attention(encoded_text,
encoded_text,
encoded_text,
None)
encoded_text = ff_layer(encoded_text)
final_layer = tf.keras.layers.Dense(model_dimensions, activation='relu')(encoded_text[:, 0])
return final_layer
def decode_sequence(target_seq, encoder_output):
# using start to indicate first step.
input_seq = tf.expand_dims([word_to_id['start']] * BATCH_SIZE, 1)
# adding embedding and position encoding.
embedding = token_embedding(input_seq)
embedding *= tf.math.sqrt(tf.cast(model_dimensions, tf.float32))
embedding += positional_encoding(tf.shape(input_seq)[1], model_dimensions)
# creating a set of attention weights to store attention values for each layer.
attention_weights = {}
# initializing decoder with last value of encoder's output.
decoder_layer = LSTMCell(model_dimensions)
decoder_initial_state = [tf.zeros((BATCH_SIZE, model_dimensions)),
tf.zeros((BATCH_SIZE, model_dimensions))]
# setting up tensors to be used later.
all_predictions = []
targets_sentences = []
for t in range(MAXLEN):
predictions, decoder_state, attention_weights["decoder_layer{}_block1".format(t+1)] = decoder_layer(embedding,
decoder_initial_state,
encoder_output)
all_predictions.append(predictions)
prediction_indices = tf.argmax(predictions, axis=2)
sentence_tokens = tokenizer.sequences_to_texts([[index.numpy() for index in row]
for row in prediction_indices.numpy()])
targets_sentences.extend(sentence_tokens)
if t < MAXLEN-1:
targets = tf.one_hot(target_seq[:, t+1], VOCAB_SIZE)
embedding = token_embedding(prediction_indices)
embedding *= tf.math.sqrt(tf.cast(model_dimensions, tf.float32))
embedding += positional_encoding(t+2, model_dimensions)
else:
targets = tf.constant([[0.] * VOCAB_SIZE]*BATCH_SIZE)
teacher_force = random.random() < teacher_forcing_ratio
decoder_initial_state = decoder_state
targets = tf.stop_gradient(targets)
loss = loss_object(all_predictions[-1], targets)
return loss, targets_sentences, attention_weights
代码解读4. 评价指标
机器翻译系统的训练过程是一个持续改进的过程，在实际应用中需要探索更适合的技术方案以提高系统的泛化能力。在实际应用中需要探索更适合的技术方案以提高系统的泛化能力
4.1 BLEU 分数
BLEU 分数是用来自动评估机器翻译文本质量的标准。它通过衡量两个句子之间共享短语的程度来计算一个参考句子和一个机器翻译句子之间的相似性程度，在这种情况下，当 BLEU 分数越高时，则表明模型的翻译效果越佳。
BLEU 分数的计算方法如下：
将每一个句子分解为若干个词汇或短语，并为其分配相应的权重。
对每一个词汇或短语进行对比分析，并计算其匹配度及可能性。
将所有词汇及短语的匹配度与可能性相加。
综合计算出整体的匹配度与可能性。
汇总并最终得出整体的匹配度与可能性数值。
BLEU 分数基于 n-gram 概念，n 表示句子中短语的数量，范围一般为 1～4。
具体的代码如下：
def bleu(reference, hypothesis, n):
reference = [line.strip().split() for line in open(reference, mode="rt", encoding="utf-8")]
hypothesis = hypothesis.strip().split()
scores = nltk.translate.bleu_score.corpus_bleu(reference, hypothesis, weights=(0.25,) * n)
return scores
代码解读4.2 METEOR 分数
METEOR指标（Machine Translation Evaluation Metrics, MTM）是一种用于评估机器翻译质量的自动评估工具。与BLEU指标类似，在某种程度上它也是衡量一个参考句子与其机器翻译版本之间短语匹配程度的一种方法。然而,METEOR更加注重语法准确性与召回率,它特别强调名词短语、动词短语、形容词短语、副词短语以及介词短语的正确识别能力。
METEOR 分数的计算方法如下：
通过正则表达式识别出所有名词性词语组合（名词短语）、动词性词语组合（动词短 phrases）、形容词性词语组合（形容词 short phrases）、副词性词语组合（副 verb 短 phrases）以及介助动作用的介词搭配（介辞 short phrases）。
使用一个程序对提取出的关键术语进行分类整理。
根据参考与候选之间的匹配情况评估召回率（Recall）、精确度（Precision）以及F1分数（F1 score）。
将主要指标的主要数值取其平均值作为最终评分标准。
具体的代码如下：
def meteor(reference, hypothesis):
r = subprocess.check_output(['java', '-jar', '/path/to/meteor-1.5/meteor-1.5.jar',
'-l', 'en', '-', '-stdio'],
universal_newlines=True,
input='\n'.join(reference) + '\n' + hypothesis)
score = float(re.search('F1:\s*(\d+\.\d+)', r).group(1))
return score
代码解读4.3 CIDEr 得分
该评估工具为CIDEr（基于一致性的图像描述评估体系），其得分为通过一致性假设计算得出的结果。该评估依据一致性假设得出结论：图像描述应尽可能保持一致。该评估体系通过衡量两个列表的一致性来进行比较。两者的相似性程度则由NIST提供的标准化一致性指标来量化。
CIDEr 的计算方法如下：
- 采用一个指标来评估词与短语之间的匹配程度。
- 基于候选描述提取n-gram及bigram序列，并计算其对应的n-gram和bigram指标值。
- 参考 ground-truth 数据集收集潜在的候选描述集合。
- 通过比较候选描述与 ground-truth 数据集中的真实例子，确定最优匹配项，并基于此计算一致性分数。
具体的代码如下：
def cider(references, candidate):
refs = [[ref.lower().split()] for ref in references]
cand = candidate.lower().split()
score = CiderScorer()(refs, cand)
return score
代码解读5. 数据集和实验环境设置
为评估该机器翻译模型的效果而搭建了一个包含190,000对语句对的中文-英文数据集；经清洗处理后选取其中10%作为开发集、剩余部分作为测试集
实验环境设置如下：
软件平台为Ubuntu Linux的系统；设备型号为GeForce GTX TITAN X；该应用运行于CUDA 释放版本号为 release 10.1以及V序列中的V号；此外，在Linux环境下运行时支持多线程技术并行处理任务
6. 模型实现
模型实现已经完成，欢迎大家阅读本文！