Keras 系列（三） Seq-Seq 与自编码器

MeteoAI

发布于 2019-07-22 18:07:11

1.4K0

发布于 2019-07-22 18:07:11

文章被收录于专栏：MeteoAIMeteoAI

最近铁柱一直在思考一个问题，如何在Keras中实现RNN序列到序列(sequence-to-sequence)的预测？网上大多数资料都是做的自然语言处理方向，时序方向的开源代码大多是基于TensorFlow,看得铁柱头昏脑胀。在查阅文献时，基于Seq-Seq的双向LSTM时序多步预测表现抢眼，也不知道是不是在往SCI灌水

，前几日做的单步预测实验，Lightgm的效果是要比单步LSTM效果好，这严重打击了我更新Keras系列的积极性，我感觉我对深度学习产生了盲目崇拜。

铁柱未来想验证多步预测上LSTM的效果，欢迎同行大佬赐教啊，此篇文章其实是一个“英雄招募帖”！铁柱私密邮箱：

deepwind@aliyun.com，你懂的。

下面言归正传：

什么是Seq-Seq

序列到序列（Seq2Seq）学习是关于训练模型以将来自一个领域（例如，英语的句子）的序列转换成另一个领域（例如翻译成中文的相同句子）的序列的模型。 "the cat sat on the mat" -> [Seq2Seq model] -> "那只猫坐在地毯上" 这可以用于时序数据的预测，比如以前提到的风功率预测。通常，只要您需要前后有顺序、有关联的数据，就可以使用它。有多种方式来处理这样的任务，或者使用RNN或者使用一维的卷积网络。在涉及到seq-seq的任务中，一般都会涉及到自编码器。

什么是自编码器

首先，自编码器(autoencoder) 是一种利用反向传播算法使得输出值等于输入值的神经网络（图二、图三），它先将输入压缩成潜在空间表征，然后通过这种表征来重构输出，输出是对输入的更加有效的表示（图三、图四）。该网络可以看作由两部分组成：一个编码器函数和一个生成重构的解码器。传统上，自动编码器被用于降维或特征学习（来自Siraj Rava小哥的自编码视频截图）。

Siraj Rava小哥的自编码视频截图一

图二

图三

图四

简单案例

当输入序列和输出序列长度相同时，您可以简单地用LSTM或GRU层（或其堆栈）来实现这些模型。以下的示范就是这种情况，它显示如何教导RNN学习如何对数字进行相加(加法)：

STEP 1. 定义解码器

from keras.models import Sequential
from keras import layers
from keras.utils import plot_model
import numpy as np
from six.moves import range
from IPython.display import Image

class CharacterTable(object):
    """
    给予一组的字符:
    + 将这些字符使用one-hot编码成数字表示
    + 译码one-hot编码数字表示成为原本的字符
    + 解码字符机率的向量以回复最有可能的字符
    """
    def __init__(self, chars):
        """初始化字符表

        # 参数:
            chars: 会出现在输入的可能字符集
        """
        self.chars = sorted(set(chars))
        self.char_indices = dict((c, i) for i, c in enumerate(self.chars))
        self.indices_char = dict((i, c) for i, c in enumerate(self.chars))

    def encode(self, C, num_rows):
        """对输入的字符串进行one-hot编码

        # 参数:
            C: 要被编码的字符
            num_rows: one-hot编码后要回传的最大行数。这是用来确保每一个输入都会得到
            相同行数的输出
        """
        x = np.zeros((num_rows, len(self.chars)))
        for i, c in enumerate(C):
            x[i, self.char_indices[c]] = 
        return x

    def decode(self, x, calc_argmax=True):
        """对输入的编码(向量)进行译码

        # 参数:
            x: 要被译码的字符向量或字符编码
            calc_argmax: 是否要用argmax算符找出机率最大的字符编码
        """
        if calc_argmax:
            x = x.argmax(axis=-1)
        return ''.join(self.indices_char[x] for x in x)

class colors:
    ok = '\033[92m'
    fail = '\033[91m'
close = '\033[0m'

STEP 2. 相关的参数与产生训练用的数据集

#模型与数据集的参数
TRAINING_SIZE = 50000 # 训练数据集的samples数
DIGITS = 3            # 加数或被加数的字符数
INVERT = True 

#输入的最大长度 'int + int' (比如, '345+678')
MAXLEN = DIGITS + 1 + DIGITS

#所有要用到的字符(包括数字、加号及空格)
chars = '0123456789+ '
ctable = CharacterTable(chars) # 创建CharacterTable的instance

questions = [] # 训练用的句子 "xxx+yyy"
expected = []  # 训练用的标签
seen = set()

print('Generating data...') # 产生训练数据

while len(questions) < TRAINING_SIZE:
    # 数字产生器 (3个字符)
    f = lambda: int(''.join(np.random.choice(list('0123456789'))
                           for i in range(np.random.randint(, DIGITS+))))
    a, b = f(), f()
    # 跳过己经看过的题目以及x+Y = Y+x这样的题目
    key = tuple(sorted((a, b)))
    if key in seen:
        continue    
    seen.add(key)

    # 当数字不足MAXLEN则填补空白
    q = '{}+{}'.format(a, b)
    query = q + ' ' * (MAXLEN - len(q))
    ans = str(a + b)

    # 答案的最大的字符长度为DIGITS + 1
    ans += ' ' * (DIGITS +  - len(ans))
    if INVERT:
        # 调转问题字符的方向, 比如. '12+345'变成'543+21'
        query = query[::-1]
    questions.append(query)
    expected.append(ans)

print('Total addition questions:', len(questions))

Generating data...
Total addition questions: 50000

STEP 3.数据的预处理

# 把数据做适当的转换, LSTM预期的数据结构 -> [samples, timesteps, features]
print('Vectorization...')
x = np.zeros((len(questions), MAXLEN, len(chars)), dtype=np.bool) # 初始一个维的numpy ndarray (特征数据)
y = np.zeros((len(questions), DIGITS + , len(chars)), dtype=np.bool) # 初始一个维的numpy ndarray (卷标数据)

# 将"特征数据"转换成LSTM预期的数据结构 -> [samples, timesteps, features]
for i, sentence in enumerate(questions):
    x[i] = ctable.encode(sentence, MAXLEN)      # <--- 要了解为什么要这样整理资料

print("Feature data: ", x.shape)

# 将"卷标数据"转换成LSTM预期的数据结构 -> [samples, timesteps, features]
for i, sentence in enumerate(expected):
    y[i] = ctable.encode(sentence, DIGITS + )  # <--- 要了解为什么要这样整理资料

print("Label data: ", y.shape)
# 打散 Shuffle(x, y)
indices = np.arange(len(y))
np.random.shuffle(indices)
x = x[indices]
y = y[indices]
# 保留%的数据来做为验证
split_at = len(x) - len(x) // 
(x_train, x_val) = x[:split_at], x[split_at:]
(y_train, y_val) = y[:split_at], y[split_at:]
print('Training Data:')
print(x_train.shape)
print(y_train.shape)
print('Validation Data:')
print(x_val.shape)
print(y_val.shape)

Vectorization...
Feature data:  (50000, 7, 12)
Label data:  (50000, 4, 12)
Training Data:
(45000, 7, 12)
(45000, 4, 12)
Validation Data:
(5000, 7, 12)
(5000, 4, 12)

STEP 4.构建网络架构

# 可以试着替代其它种的rnn units, 比如,GRU或SimpleRNN
RNN = layers.LSTM
HIDDEN_SIZE = 128
BATCH_SIZE = 128
LAYERS = 1

print('Build model...')
model = Sequential()

# ===== 编码 (encoder) ====

# 使用RNN“编码”输入序列，产生HIDDEN_SIZE的输出。
# 注意：在输入序列长度可变的情况下，使用input_shape =（None，num_features）
model.add(RNN(HIDDEN_SIZE, input_shape=(MAXLEN, len(chars)))) # MAXLEN代表是timesteps, 而len(chars)是one-hot编码的features

# 作为译码器RNN的输入，重复提供每个时间步的RNN的最后一个隐藏状态。
# 重复“DIGITS + 1”次，因为这是最大输出长度，例如当DIGITS = 3时，最大输出是999 + 999 = 1998（长度为4)。
model.add(layers.RepeatVector(DIGITS+1))

# ==== 解碼 (decoder) ====
# 译码器RNN可以是多层堆栈或单层。
for _ in range(LAYERS):
    # 通过将return_sequences设置为True，不仅返回最后一个输出，而且还以（num_samples，timesteps，output_dim）
    # 的形式返回所有输出。这是必要的，因为下面的TimeDistributed需要第一个维度是时间步长。
    model.add(RNN(HIDDEN_SIZE, return_sequences=True))

# 对输入的每个时间片推送到密集层来对于输出序列的每一时间步，决定选择哪个字符。
model.add(layers.TimeDistributed(layers.Dense(len(chars))))

model.add(layers.Activation('softmax'))
model.compile(loss='categorical_crossentropy',
             optimizer='adam',
             metrics=['accuracy'])
model.summary()

STEP 5.训练模型/验证评估

#我们将进行次的训练，并且在每次训练之后就进行检查。
for iteration in range(, ):
    print()
    print('-' * )
    print('Iteration', iteration)
    model.fit(x_train, y_train,
             batch_size=BATCH_SIZE,
             epochs=,
             validation_data=(x_val, y_val))

    for i in range():
        ind = np.random.randint(, len(x_val))
        rowx, rowy = x_val[np.array([ind])], y_val[np.array([ind])]
        preds = model.predict_classes(rowx, verbose=)

        q = ctable.decode(rowx[])
        correct = ctable.decode(rowy[])
        guess = ctable.decode(preds[], calc_argmax=False)
        print('Q', q[::-1] if INVERT else q, end=' ')
        print('T', correct, end=' ')
        if correct == guess:
            print(colors.ok + '☑' + colors.close, end=' ')
        else:
            print(colors.fail + '☒' + colors.close, end=' ')
        print(guess)