Python实现seq2seq【tensorflow2.3】
本帖最后由 糖逗 于 2021-4-22 11:12 编辑一、导入库
import numpy as np
import time
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers, optimizers
import tensorflow_addons as tfa
import datetime
from sklearn.model_selection import train_test_split
二、日期数据生成
PAD_ID = 0
class DateData:
def __init__(self, n):
np.random.seed(1)
self.date_cn = []
self.date_en = []
for timestamp in np.random.randint(143835585, 2043835585, n):
date = datetime.datetime.fromtimestamp(timestamp)
self.date_cn.append(date.strftime("%y-%m-%d"))
self.date_en.append(date.strftime("%d/%b/%Y"))
self.vocab = set(
+ ["-", "/", "<GO>", "<EOS>"] + [
i.split("/") for i in self.date_en])
self.v2i = {v: i for i, v in enumerate(sorted(list(self.vocab)), start=1)}
self.v2i["<PAD>"] = PAD_ID
self.vocab.add("<PAD>")
self.i2v = {i: v for v, i in self.v2i.items()}
self.x, self.y = [], []
for cn, en in zip(self.date_cn, self.date_en):
self.x.append( for v in cn])
self.y.append(
, ] + for v in en[:3]] + [
self.v2i], ] + for v in en] + [
self.v2i["<EOS>"], ])
self.x, self.y = np.array(self.x), np.array(self.y)
self.start_token = self.v2i["<GO>"]
self.end_token = self.v2i["<EOS>"]
def sample(self, n=64):
bi = np.random.randint(0, len(self.x), size=n)
bx, by = self.x, self.y
decoder_len = np.full((len(bx),), by.shape - 1, dtype=np.int32)
return bx, by, decoder_len
def idx2str(self, idx):
x = []
for i in idx:
x.append(self.i2v)
if i == self.end_token:
break
return "".join(x)
@property
def num_word(self):
return len(self.vocab)
三、模型构建
class seq2seq(keras.Model):
def __init__(self, source_dict_total_words, source_embedding_size, encoder_num_layers, encoder_rnn_size,
target_dict_total_words, target_embedding_size, decoder_rnn_size, start_token, batch_size,
end_token, target_size):
super(seq2seq, self).__init__()
'''
encoder参数说明
--source_dict_total_words：source字典的总单词个数
--source_embedding_size：souce压缩的长度
--encoder_num_layers:encoder堆叠的rnn cell数量
--encoder_rnn_size:encoder中RNN单元的隐层结点数量
decoder参数说明
--target_dict_total_words:target字典的总单词个数
--target_embedding_size:target压缩的长度:
--decoder_num_layers:decoder堆叠的rnn cell数量
--decoder_rnn_size:decoder中RNN单元的隐层结点数量
--target_size:target中句子的长度
其他参数说明
--start_token：decoder输入的开始标志<GO>在target字典中的对应数字编号
--end_token：decoder输入的结束标志<EOS>在target字典中的对应数字编号
--batch_size：数据的batch_size
'''
self.source_dict_total_words = source_dict_total_words
self.source_embedding_size = source_embedding_size
self.encoder_num_layers = encoder_num_layers
self.encoder_rnn_size = encoder_rnn_size
self.target_dict_total_words = target_dict_total_words
self.target_embedding_size = target_embedding_size
self.decoder_rnn_size = decoder_rnn_size
self.start_token = start_token
self.batch_size = batch_size
self.end_token = end_token
self.target_size = target_size
self.cross_entropy = keras.losses.SparseCategoricalCrossentropy(from_logits = True)
self.optimzer = optimizers.Adam(lr = 1e-2)
self.seq_len = tf.fill(, self.target_size-1)
#######################Encoder##################################
#1.embedding
self.encoder_embedding = layers.Embedding(self.source_dict_total_words, self.source_embedding_size,
embeddings_initializer = tf.initializers.RandomNormal(0., 0.1))
#2.单层或多层rnn
self.encoder_rnn_cells =
self.encoder_stacked_lstm = layers.StackedRNNCells(self.encoder_rnn_cells)
self.encoder_rnn = layers.RNN(self.encoder_stacked_lstm, return_state = True, return_sequences = True)
#######################Decoder##################################
#1.embedding
self.decoder_embedding = layers.Embedding(self.target_dict_total_words, self.target_embedding_size,
embeddings_initializer = tf.initializers.RandomNormal(0., 0.1))
#2.构造Decoder中的rnn单元
self.decoder_rnn_cell = tf.keras.layers.LSTMCell(self.decoder_rnn_size)
#3.构造Decoder中的dense单元
self.decoder_dense_layer = layers.Dense(self.target_dict_total_words,
kernel_initializer = tf.compat.v1.truncated_normal_initializer(mean = 0.0, stddev = 0.1))
#3.train
self.decoder_sampler = tfa.seq2seq.TrainingSampler()
self.training_decoder = tfa.seq2seq.BasicDecoder(cell = self.decoder_rnn_cell, sampler = self.decoder_sampler,
output_layer = self.decoder_dense_layer)
#4.predict
self.sampler = tfa.seq2seq.GreedyEmbeddingSampler()
self.predicting_decoder = tfa.seq2seq.BasicDecoder(cell = self.decoder_rnn_cell, sampler = self.sampler,
output_layer = self.decoder_dense_layer)
def encode(self, source):
embedded = self.encoder_embedding(source)
#init_s = , self.encoder_rnn_size)), tf.zeros((source.shape, self.encoder_rnn_size))]
res_list = self.encoder_rnn(embedded)
encoder_hidden = res_list
encoder_state = res_list
#print(encoder_hidden.shape, encoder_state.shape)
return
def train(self, source, target):
state = self.encode(source)
decoder_input = target[:, :-1] #ignore <EOS>
decoder_embeding_input = self.decoder_embedding(decoder_input)
output, _, _ = self.training_decoder(decoder_embeding_input, initial_state = state, sequence_length = self.seq_len)
return output.rnn_output
def predict(self, source):
initial_state = self.encode(source)
done, inputs, state = self.predicting_decoder.initialize(
self.decoder_embedding.variables,
start_tokens = tf.fill(, self.start_token),
end_token = self.end_token,
initial_state = initial_state,
)
pred_id = np.zeros((self.batch_size, self.target_size), dtype = np.int32)
for time in range(self.target_size):
output, state, inputs, done = self.predicting_decoder.step(
time = time, inputs = inputs, state = state, training = False)
pred_id[:, time] = output.sample_id
return pred_id
def step(self, source, target):
with tf.GradientTape() as tape:
logits = self.train(source, target)
dec_out = target[:, 1:]# ignore <GO>
loss = self.cross_entropy(dec_out, logits)
grads = tape.gradient(loss, self.trainable_variables)
self.optimzer.apply_gradients(zip(grads, self.trainable_variables))
return loss.numpy()
四、数据实验
epochs = 200
batch_size = 248
data = DateData(4000)
print("1.Chinese time order: yy/mm/dd ", data.date_cn[:3], "\n2.English time order: dd/M/yyyy ", data.date_en[:3])
print("3.vocabularies: \n", data.vocab)
print("4.x index sample: \n{}\n{}".format(data.idx2str(data.x), data.x),
"\n5.y index sample: \n{}\n{}".format(data.idx2str(data.y), data.y))
train_db = tf.data.Dataset.from_tensor_slices((np.array(data.x), np.array(data.y)))
train_db = train_db.batch(batch_size, drop_remainder=True)
optimizer = optimizers.Adam(lr = 1e-2)
model = seq2seq(source_dict_total_words = data.num_word, source_embedding_size = 16, encoder_num_layers = 1, encoder_rnn_size = 32,
target_dict_total_words = data.num_word, target_embedding_size = 16, decoder_rnn_size = 32,
start_token = data.start_token, batch_size = batch_size,
end_token = data.end_token, target_size = 11)#target_size是target单个句子的长度，包括<GO>和<EOS>
for epoch in range(epochs):
for step, (source, target) in enumerate(train_db):
loss = model.step(source, target)
if step % 5 == 0:
target = data.idx2str(np.array(target))
pred = model.predict(source)
res = data.idx2str(pred)
src = data.idx2str(np.array(source))
print(
"epoch: ", epoch,
"step:", step,
"| loss: %.3f" % loss,
"| input: ", src,
"| target: ", target,
"| inference: ", res,
)
代码参考：https://mofanpy.com/tutorials/machine-learning/nlp/seq2seq/
代码参考自莫烦的seq2seq项目。在原有的基础上加上自己的理解。 {:10_327:}
页:
[1]