Seq2Seq模型及注意力机制介绍

import numpy as np
from keras.utils import to_categorical

with open('date.txt','r') as f:  
    dataset = eval(f.read())
### character dict for human date format    
human_vocab = {' ': 0, '.': 1, '/': 2, '0': 3, '1': 4, '2': 5, '3': 6, '4': 7, '5': 8, '6': 9, '7': 10, '8': 11, '9': 12,
               '<pad>': 36, '<unk>': 35, 'a': 13, 'b': 14, 'c': 15, 'd': 16, 'e': 17, 'f': 18, 'g': 19, 'h': 20, 'i': 21,
               'j': 22, 'l': 23, 'm': 24, 'n': 25, 'o': 26, 'p': 27, 'r': 28, 's': 29, 't': 30, 'u': 31, 'v': 32, 'w': 33, 'y': 34}
### character dict for machine date format YYYY-MM-DD
machine_vocab = {'-': 0, '0': 1, '1': 2, '2': 3, '3': 4, '4': 5, '5': 6, '6': 7, '7': 8, '8': 9, '9': 10}
inv_machine_vocab = {0: '-', 1: '0', 2: '1', 3: '2', 4: '3', 5: '4', 6: '5', 7: '6', 8: '7', 9: '8', 10: '9'}
### process data
def string_to_int(string, length, vocab, is_human):
    string = string.lower()
    string = string.replace(',','')
    if len(string) > length:
        string = string[:length] #truncate string if string is too long         
    # transform string to a list of integers. if character not in the vocab, replace it with <unk> 
    rep = list(map(lambda x: vocab.get(x, vocab['<unk>']), string)) if is_human else list(map(lambda x: vocab.get(x), string))
    # pad short string with <pad>
    if len(string) < length:
        rep += [vocab['<pad>']] * (length - len(string))
    return rep

def preprocess_data(dataset, human_vocab, machine_vocab, Tx, Ty):
    m = len(dataset)
    X, Y = zip(*dataset) #(m,), (m,)
    X = np.array([string_to_int(i, Tx, human_vocab, is_human=True) for i in X]) #(m, Tx)
    Y = np.array([string_to_int(t, Ty, machine_vocab, is_human=False) for t in Y]) #(m, Ty)
    Xoh = np.array(list(map(lambda x: to_categorical(x, num_classes=len(human_vocab)), X))) #(m, Tx, len(human_vocab))
    Yoh = np.array(list(map(lambda x: to_categorical(x, num_classes=len(machine_vocab)), Y))) #(m, Ty, len(machine_vocab))
    return X, Y, Xoh, Yoh

Tx, Ty = 30, 10
X, Y, Xoh, Yoh = preprocess_data(dataset, human_vocab, machine_vocab, Tx, Ty)

from keras.layers import Bidirectional, Input, LSTM, Dense
from keras.models import Model

### Defined shared layers as global variables
n_a = 32 # number of units for the pre-attention, bi-directional LSTM's hidden state 'a'
n_s = 64 # number of units for the post-attention LSTM's hidden state "s"
post_attention_LSTM_cell = LSTM(n_s, return_state = True) # post-attention LSTM 
output_layer = Dense(len(machine_vocab), activation="softmax")

### Model architecture
def model(Tx, Ty, n_a, n_s, human_vocab_size, machine_vocab_size):
    X = Input(shape=(Tx, human_vocab_size))
    s0 = Input(shape=(n_s,), name='s0') #define s0 (initial hidden state)
    c0 = Input(shape=(n_s,), name='c0') #define c0 (initial cell state)
    s = s0 
    c = c0
    outputs = [] # Initialize empty list of outputs, final shape: Ty np arrays with shape (m, len(machine_vocab))        
    # Step 1: Define pre-attention Bi-LSTM. 
    a = Bidirectional(LSTM(units=n_a, return_sequences=True))(X) #(m, Tx, 2*n_a)  
    # Step 2: Iterate for Ty steps
    for t in range(Ty):
        # Step 2.A: Perform one step of the attention mechanism to get the context vector at step t
        context = one_step_attention(a, s) #the attention layer is defined below, (m, 1, 2*n_a)
        # Step 2.B: Apply the post-attention LSTM cell to the context vector
        s, _, c = post_attention_LSTM_cell(context, initial_state=[s,c]) #(m, n_s), (m, n_s)
        # Step 2.C: Apply Dense layer to the hidden state output of the post-attention LSTM
        out = output_layer(s) #(m, len(machine_vocab))
        # Step 2.D: Append "out" to the "outputs" list 
        outputs.append(out)
    # Step 3: Create model instance
    model = Model(inputs=[X,s0,c0], outputs=outputs)    
    return model

from keras.layers import Concatenate, Dot, RepeatVector, Activation 
import keras.backend as K

### Define a custom softmax function for convenient use in attention layer
def softmax(x, axis=1):
    """
    Arguments:
    x -- Tensor.
    axis -- Integer, axis along which the softmax normalization is applied.
    """
    ndim = K.ndim(x)
    if ndim == 2:
        return K.softmax(x)
    elif ndim > 2:
        e = K.exp(x - K.max(x, axis=axis, keepdims=True))
        s = K.sum(e, axis=axis, keepdims=True)
        return e / s
    else:
        raise ValueError('Cannot apply softmax to a tensor that is 1D')
        
### Defined shared layers as global variables
repeator = RepeatVector(Tx)
concatenator = Concatenate(axis=-1)
# [densor1,densor2] is F
densor1 = Dense(10, activation = "tanh") 
densor2 = Dense(1, activation = "relu")
activator = Activation(softmax, name='attention_weights') #using a custom softmax(axis = 1)
dotor = Dot(axes = 1)

### Perform one step of attention
def one_step_attention(a, s_prev):
    """    
    Arguments:
    a -- hidden state output of the Bi-LSTM, numpy-array of shape (m, Tx, 2*n_a)
    s_prev -- previous hidden state of the (post-attention) LSTM, numpy-array of shape (m, n_s)
    """
    s_prev = repeator(s_prev) #(m, Tx, n_s)
    concat = concatenator([a, s_prev]) #(m, Tx, 2*n_a+n_s)
    e = densor1(concat) #(m, Tx, 10)
    energies = densor2(e) #(m, Tx, 1)
    alphas = activator(energies) # compute the attention weights, (m, Tx, 1)
    context = dotor([alphas, a]) #(m, 1, 2*n_a) 
    return context