keras 自然语言处理 lstm_seq2seq 案例分析_keras 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

1
2
3
4

加载数据：fra-eng

这是一个包含超过64000个法语-英语单词和短语的Anki词汇卡组，它可以帮助学习者提高词汇量并加强其语言技能。Anki是一种基于记忆卡片的学习方法，被广泛用于学习基础词汇和语法规则。该词汇卡组可以在线或离线使用，并可以导入到Anki应用程序中，以便离线学习。该词汇卡组的单词和短语都包含音频发音，可以帮助学习者正确地发音字词。这里用于机器翻译。

!!curl -O http://www.manythings.org/anki/fra-eng.zip
!!unzip fra-eng.zip

1
2
3

配置项

这些配置决定着模型的参数和训练的行为。在配置模型时，我们需要用经过实验验证的数值来平衡训练时间和模型性能。

• latent_dim：这是解码器中LSTM层的隐藏状态的维度大小。通俗地说，这决定了模型学习到的内部表示的复杂度。
• num_samples：这是用于训练模型的语料库中的句子数量。更多的语料库可以带来更好的性能，但需要更多的训练时间和计算资源。
• batch_size：这是一次传递给模型的句子数。更大的批量可以加速训练，但也可能导致内存不足等问题。

batch_size = 64  # 训练的 batch 大小。
epochs = 100  # 训练的 epoch 数。
latent_dim = 256  # 编码器空间的潜在维度。
num_samples = 10000  # 训练样本数量。
# 存储在磁盘上的数据文本文件路径。
data_path = "fra.txt"

1
2
3
4
5
6
7

数据预处理

该部分介绍了 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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

定义模型

这部分代码主要是使用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)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

载入训练好的模型

这部分，介绍了如何使用已经训练好的序列到序列模型进行推理，即输入一段源语言文本，输出对应的翻译结果。

该部分首先加载了之前训练好的模型，包括编码器和解码器。然后，对于给定的源语言（英语）文本输入，通过构建Encoder模型，将其转化为一系列隐状态向量，并将这些状态向量作为输入，通过构建解码器模型，生成目标语言（法语）的翻译结果。其中，在解码器中，使用了beam search的方法来解码，以提高翻译的准确性。

在代码中，最后提供了一个函数，使得用户可以方便地输入自己的源语言文本，并输出对应的翻译结果。在实际应用中，可以将该函数集成到一个大型的自然语言处理系统中，以提供自然语言翻译的功能，例如在线翻译软件等。

1. 对输入进行编码并检索初始解码器状态
2. 在这个初始状态下运行解码器的一个步骤以及作为目标的“序列开始”令牌。输出将是下一个目标令牌。
3. 对当前目标令牌和当前状态重复

# 定义采样模型
# 恢复模型并构建编码器和解码器
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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155

调用函数翻译语句：

• 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)

1
2
3
4
5
6
7
8
9