keras 自然语言处理 lstm_seq2seq 案例分析
本例演示了如何实现一个基本的字符级使用 LSTM（长短期记忆神经网络）和 Seq2Seq（序列到序列模型）来进行自然语言处理的模型。我们将其应用于逐个字符，将短英语句子翻译为短法语句子。需要注意的是，在这个领域里，使用字符级机器翻译是相当不寻常的，因为使用单词级的模型更为常见。
算法摘要
- 我们从一个领域（例如英语句子）的输入序列开始，并从另一个领域（例如法语句子）中获得相应的目标序列。
- 一个编码器 LSTM 将输入序列转换为 2 个状态向量（我们保留最后一个 LSTM 状态并丢弃输出）。
- 一个解码器 LSTM 被训练，将目标序列转换为相同的序列，但是将时间步向前偏移一个步长，这个训练过程在这个上下文中称为“teacher forcing”。它使用来自编码器的状态向量作为初始状态。实际上，解码器在输入序列的条件下学习生成“targets[t+1…]”给定“targets[…t]”。
- 在推理模式下，当我们想要解码未知的输入序列时，我们要做以下操作：
- 编码输入序列为状态向量。
- 以大小为 1 的目标序列开始（只有开始字符）。
- 将状态向量和 1 个字符的目标序列提供给解码器，生成下一个字符的预测值。
- 使用这些预测值采样下一个字符 (我们简单地使用 argmax)。
- 将采样的字符添加到目标序列中。
- 重复操作，直到生成结束字符或达到字符限制。
导入依赖
import numpy as np
import tensorflow as tf
from tensorflow import keras
加载数据：fra-eng
这是一个包含超过64000个法语-英语单词和短语的Anki词汇卡组，它可以帮助学习者提高词汇量并加强其语言技能。Anki是一种基于记忆卡片的学习方法，被广泛用于学习基础词汇和语法规则。该词汇卡组可以在线或离线使用，并可以导入到Anki应用程序中，以便离线学习。该词汇卡组的单词和短语都包含音频发音，可以帮助学习者正确地发音字词。这里用于机器翻译。
!!curl -O http://www.manythings.org/anki/fra-eng.zip
!!unzip fra-eng.zip
配置项
这些配置决定着模型的参数和训练的行为。在配置模型时，我们需要用经过实验验证的数值来平衡训练时间和模型性能。
latent_dim：这是解码器中LSTM层的隐藏状态的维度大小。通俗地说，这决定了模型学习到的内部表示的复杂度。num_samples：这是用于训练模型的语料库中的句子数量。更多的语料库可以带来更好的性能，但需要更多的训练时间和计算资源。batch_size：这是一次传递给模型的句子数。更大的批量可以加速训练，但也可能导致内存不足等问题。
batch_size = 64 # 训练的 batch 大小。
epochs = 100 # 训练的 epoch 数。
latent_dim = 256 # 编码器空间的潜在维度。
num_samples = 10000 # 训练样本数量。
# 存储在磁盘上的数据文本文件路径。
data_path = "fra.txt"
数据预处理
该部分介绍了 LSTM Seq2Seq 模型中数据的预处理过程，包括读取数据、建立字典、将句子转换为序列、补全序列等操作。其中读取数据时，原始的文本形式数据被转换成形如 [input_text, target_text] 的列表，其中 input_text 和 target_text 分别表示源语言和目标语言的句子。接着建立了源语言和目标语言的字典，并将句子转换为单词序列，同时在序列前后添加特殊标记，使其具有一定的通用性。最后，需要对序列进行补全，以保证所有句子的长度一致。这些预处理操作可以提高模型的训练效果和泛化能力。
# 对数据进行向量化处理。
input_texts = [] # 编码器的输入
target_texts = [] # 解码器的目标输出
input_characters = set() # 所有的输入字符（去重）
target_characters = set() # 所有的输出字符（去重）
with open(data_path, "r", encoding="utf-8") as f: # 读取数据
lines = f.read().split("\n")
for line in lines[: min(num_samples, len(lines) - 1)]:
input_text, target_text, _ = line.split("\t")
# 将"\t"作为目标文本的"开始序列"字符，将"\n"作为"结束序列"字符。
target_text = "\t" + target_text + "\n"
input_texts.append(input_text)
target_texts.append(target_text)
for char in input_text:
if char not in input_characters: # 若当前字符不在输入字符集中，则添加进去
input_characters.add(char)
for char in target_text:
if char not in target_characters: # 若当前字符不在输出字符集中，则添加进去
target_characters.add(char)
# 将字符集按英文字母表顺序排序，并给每个字符赋一个数字作为索引，即编码
input_characters = sorted(list(input_characters))
target_characters = sorted(list(target_characters))
num_encoder_tokens = len(input_characters) # 输入字符数量
num_decoder_tokens = len(target_characters) # 输出字符数量
max_encoder_seq_length = max([len(txt) for txt in input_texts]) # 最大输入序列长度
max_decoder_seq_length = max([len(txt) for txt in target_texts]) # 最大输出序列长度
print("Number of samples:", len(input_texts))
print("Number of unique input tokens:", num_encoder_tokens)
print("Number of unique output tokens:", num_decoder_tokens)
print("Max sequence length for inputs:", max_encoder_seq_length)
print("Max sequence length for outputs:", max_decoder_seq_length)
# 编码器的输入要进行one-hot编码。
input_token_index = dict([(char, i) for i, char in enumerate(input_characters)])
# 解码器的输入和输出都要进行one-hot编码。
target_token_index = dict([(char, i) for i, char in enumerate(target_characters)])
encoder_input_data = np.zeros(
(len(input_texts), max_encoder_seq_length, num_encoder_tokens), dtype="float32"
)
decoder_input_data = np.zeros(
(len(input_texts), max_decoder_seq_length, num_decoder_tokens), dtype="float32"
)
decoder_target_data = np.zeros(
(len(input_texts), max_decoder_seq_length, num_decoder_tokens), dtype="float32"
)
# 构建编码器和解码器输入、输出数据
for i, (input_text, target_text) in enumerate(zip(input_texts, target_texts)):
for t, char in enumerate(input_text): # 对于输入文本中的每个字符
encoder_input_data[i, t, input_token_index[char]] = 1.0 # 将该字符的one-hot向量赋值给输入数据中相应的位置
encoder_input_data[i, t + 1 :, input_token_index[" "]] = 1.0 # 将剩余的位置赋值为空格（padding的作用）
for t, char in enumerate(target_text): # 对于目标文本中的每个字符
# 解码器的输入和目标输出相差一个时间步
decoder_input_data[i, t, target_token_index[char]] = 1.0 # 将该字符的one-hot向量赋值给解码器输入数据中相应的位置
if t > 0: # 从第二个字符开始，目标输出才能在当前时间步进行预测
# decoder_target_data的时间步比decoder_input_data多1，且不需要起始序列字符
# 因此需要将解码器目标输出的one-hot向量赋值给decoder_target_data中的上一个时间步
decoder_target_data[i, t - 1, target_token_index[char]] = 1.0
decoder_input_data[i, t + 1 :, target_token_index[" "]] = 1.0
decoder_target_data[i, t:, target_token_index[" "]] = 1.0
定义模型
这部分代码主要是使用Keras构建了一个LSTM Seq2Seq模型，将一个序列映射到另一个序列。简单的说，就是将输入语句翻译成另外一种语言。在该模型中主要分为编码器和解码器两个部分，建议在理解代码和注释后再进行深入学习。
# 定义输入序列并对其进行处理
encoder_inputs = keras.Input(shape=(None, num_encoder_tokens)) # encoder_inputs是输入序列，维度为 (None, num_encoder_tokens)
encoder = keras.layers.LSTM(latent_dim, return_state=True) # 用LSTM层对输入序列进行处理，返回输出和状态
encoder_outputs, state_h, state_c = encoder(encoder_inputs)
# 我们丢弃encoder_outputs，只保留状态
encoder_states = [state_h, state_c]
# 设置decoder，并使用encoder_states作为初始状态。
decoder_inputs = keras.Input(shape=(None, num_decoder_tokens))
#我们设置decoder返回完整的输出序列和内部状态。我们不在训练模型中使用返回状态，但是在推断时会用到。
decoder_lstm = keras.layers.LSTM(latent_dim, return_sequences=True, return_state=True) # 设置decoder，使用LSTM层，返回完整的输出序列和状态
decoder_outputs, _, _ = decoder_lstm(decoder_inputs, initial_state=encoder_states) # decoder_inputs为decoder的输入，initial_state为decoder的初始状态
decoder_dense = keras.layers.Dense(num_decoder_tokens, activation="softmax") # 设置全连接层，输出概率值
decoder_outputs = decoder_dense(decoder_outputs) #decoder_outputs为decoder的输出序列
# 定义模型，将encoder_input_data和decoder_input_data转换为decoder_target_data
model = keras.Model([encoder_inputs, decoder_inputs], decoder_outputs)
载入训练好的模型
这部分，介绍了如何使用已经训练好的序列到序列模型进行推理，即输入一段源语言文本，输出对应的翻译结果。
该部分首先加载了之前训练好的模型，包括编码器和解码器。然后，对于给定的源语言（英语）文本输入，通过构建Encoder模型，将其转化为一系列隐状态向量，并将这些状态向量作为输入，通过构建解码器模型，生成目标语言（法语）的翻译结果。其中，在解码器中，使用了beam search的方法来解码，以提高翻译的准确性。
在代码中，最后提供了一个函数，使得用户可以方便地输入自己的源语言文本，并输出对应的翻译结果。在实际应用中，可以将该函数集成到一个大型的自然语言处理系统中，以提供自然语言翻译的功能，例如在线翻译软件等。
- 对输入进行编码并检索初始解码器状态
- 在这个初始状态下运行解码器的一个步骤以及作为目标的“序列开始”令牌。输出将是下一个目标令牌。
- 对当前目标令牌和当前状态重复
# 定义采样模型
# 恢复模型并构建编码器和解码器
model = keras.models.load_model("s2s")
# 确认输入
encoder_inputs = model.input[0] # input_1
# 获取编码器的输出, 包括两个状态变量 h 和 c
encoder_outputs, state_h_enc, state_c_enc = model.layers[2].output # lstm_1
encoder_states = [state_h_enc, state_c_enc] # 将状态保持到列表中
encoder_model = keras.Model(encoder_inputs, encoder_states) # 定义模型的输入和输出
# 确认解码器的输入
decoder_inputs = model.input[1] # input_2
# 定义解码器的状态变量
decoder_state_input_h = keras.Input(shape=(latent_dim,))
decoder_state_input_c = keras.Input(shape=(latent_dim,))
decoder_states_inputs = [decoder_state_input_h, decoder_state_input_c] # 将状态变量放入列表
decoder_lstm = model.layers[3] # 导入lstm层
decoder_outputs, state_h_dec, state_c_dec = decoder_lstm(
decoder_inputs, initial_state=decoder_states_inputs # 使用传递进来的状态变量作为初始状态
)
decoder_states = [state_h_dec, state_c_dec] # 将新的状态值存储到列表中
decoder_dense = model.layers[4] #定义完全连接层
decoder_outputs = decoder_dense(decoder_outputs) # 初始化完成后，传递到完全连接层中
decoder_model = keras.Model(
[decoder_inputs] + decoder_states_inputs, [decoder_outputs] + decoder_states # 定义输入和状态的列表作为输入和输出
)
# Reverse-lookup token index to decode sequences back to
# something readable.
# 将编码后的数据反转回可读形式的字典
reverse_input_char_index = dict((i, char) for char, i in input_token_index.items())
reverse_target_char_index = dict((i, char) for char, i in target_token_index.items())
# 解码器解码序列函数
def decode_sequence(input_seq):
# Encode the input as state vectors.
# 对输入进行编码
states_value = encoder_model.predict(input_seq)
# Generate empty target sequence of length 1.
# 生成空的目标序列，并加入开始字符标记
target_seq = np.zeros((1, 1, num_decoder_tokens))
target_seq[0, 0, target_token_index["\t"]] = 1.0
# Sampling loop for a batch of sequences
# (to simplify, here we assume a batch of size 1).
# 采样循环
stop_condition = False
decoded_sentence = ""
while not stop_condition:
output_tokens, h, c = decoder_model.predict([target_seq] + states_value)
# Sample a token
# 采样一个标记
sampled_token_index = np.argmax(output_tokens[0, -1, :])
sampled_char = reverse_target_char_index[sampled_token_index]
decoded_sentence += sampled_char
# Exit condition: either hit max length
# or find stop character.
# 终止条件有两个，一个是采样的字符达到规定的长度，另一个是采样的字符为终止字符
if sampled_char == "\n" or len(decoded_sentence) > max_decoder_seq_length:
stop_condition = True
# Update the target sequence (of length 1).
# 更新目标序列，输入采样的字符，并使用新的标记为输入进入下一轮采样循环
target_seq = np.zeros((1, 1, num_decoder_tokens))
target_seq[0, 0, sampled_token_index] = 1.0
# Update states
# 更新状态变量
states_value = [h, c]
return decoded_sentence
# 定义采样模型
# 恢复模型并构建编码器和解码器
model = keras.models.load_model("s2s")
# 确认输入
encoder_inputs = model.input[0] # input_1
# 获取编码器的输出, 包括两个状态变量 h 和 c
encoder_outputs, state_h_enc, state_c_enc = model.layers[2].output # lstm_1
encoder_states = [state_h_enc, state_c_enc] # 将状态保持到列表中
encoder_model = keras.Model(encoder_inputs, encoder_states) # 定义模型的输入和输出
# 确认解码器的输入
decoder_inputs = model.input[1] # input_2
# 定义解码器的状态变量
decoder_state_input_h = keras.Input(shape=(latent_dim,))
decoder_state_input_c = keras.Input(shape=(latent_dim,))
decoder_states_inputs = [decoder_state_input_h, decoder_state_input_c] # 将状态变量放入列表
decoder_lstm = model.layers[3] # 导入lstm层
decoder_outputs, state_h_dec, state_c_dec = decoder_lstm(
decoder_inputs, initial_state=decoder_states_inputs # 使用传递进来的状态变量作为初始状态
)
decoder_states = [state_h_dec, state_c_dec] # 将新的状态值存储到列表中
decoder_dense = model.layers[4] #定义完全连接层
decoder_outputs = decoder_dense(decoder_outputs) # 初始化完成后，传递到完全连接层中
decoder_model = keras.Model(
[decoder_inputs] + decoder_states_inputs, [decoder_outputs] + decoder_states # 定义输入和状态的列表作为输入和输出
)
# Reverse-lookup token index to decode sequences back to
# something readable.
# 将编码后的数据反转回可读形式的字典
reverse_input_char_index = dict((i, char) for char, i in input_token_index.items())
reverse_target_char_index = dict((i, char) for char, i in target_token_index.items())
# 解码器解码序列函数
def decode_sequence(input_seq):
# Encode the input as state vectors.
# 对输入进行编码
states_value = encoder_model.predict(input_seq)
# Generate empty target sequence of length 1.
# 生成空的目标序列，并加入开始字符标记
target_seq = np.zeros((1, 1, num_decoder_tokens))
target_seq[0, 0, target_token_index["\t"]] = 1.0
# Sampling loop for a batch of sequences
# (to simplify, here we assume a batch of size 1).
# 采样循环
stop_condition = False
decoded_sentence = ""
while not stop_condition:
output_tokens, h, c = decoder_model.predict([target_seq] + states_value)
# Sample a token
# 采样一个标记
sampled_token_index = np.argmax(output_tokens[0, -1, :])
sampled_char = reverse_target_char_index[sampled_token_index]
decoded_sentence += sampled_char
# Exit condition: either hit max length
# or find stop character.
# 终止条件有两个，一个是采样的字符达到规定的长度，另一个是采样的字符为终止字符
if sampled_char == "\n" or len(decoded_sentence) > max_decoder_seq_length:
stop_condition = True
# Update the target sequence (of length 1).
# 更新目标序列，输入采样的字符，并使用新的标记为输入进入下一轮采样循环
target_seq = np.zeros((1, 1, num_decoder_tokens))
target_seq[0, 0, sampled_token_index] = 1.0
# Update states
# 更新状态变量
states_value = [h, c]
return decoded_sentence
调用函数翻译语句：
seq_index是当前选取的序列的索引。encoder_input_data是编码器输入的数据。input_seq = encoder_input_data[seq_index : seq_index + 1]选取一个序列，注意这里是选取一个序列而不是一组序列，所以需要添加一个维度。decoded_sentence = decode_sequence(input_seq)解码这个序列，得到解码后的句子。print("-")打印一个分隔行。print("输入的句子:", input_texts[seq_index])打印输入的句子。print("解码后的句子:", decoded_sentence)打印解码后的句子。
for seq_index in range(20):
# 选取一个序列（训练集中的一部分）
# 用于尝试解码。
input_seq = encoder_input_data[seq_index : seq_index + 1]
decoded_sentence = decode_sequence(input_seq)
print("-")
print("输入的句子:", input_texts[seq_index])
print("解码后的句子:", decoded_sentence)