my_rnn_block.py

my_rnn_block.py

class RNN:
    def __init__(self, units, input_dim, activation="tanh", 
                 kernel_initializer="glorot_normal", 
                 bias_initializer='zeros', 
                 return_sequences=False):
        self.units = units
        self.input_dim = input_dim
        self.activation = activation
        self.kernel_initializer = kernel_initializer
        self.bias_initializer = bias_initializer
        self.return_sequences = return_sequences
        self.W = None
        self.V = None
        self.b = None
        self.act_func = Activation(self.activation)
        self.dW = None
        self.dV = None
        self.db = None
        self.Phi = None
        self.Z = None
        self.Phi_next = None        
        self.initialize_weights()
        
        
    def sigma(self, initializer, fan_in, fan_out):
        """重みと閾値の初期化用にガウス分布の標準偏差値を返す
           参照: https://keras.io/ja/initializers
        """
        if initializer=="glorot_normal": # for sigmoid, tanh.
            return np.sqrt(2. / (fan_in + fan_out))
        elif initializer=="he_normal":  # for relu
            return np.sqrt(2. / fan_in)
        elif initializer=="lecun_normal":  # for traial
            return np.sqrt(1. / fan_in)
        elif initializer=="one_normal":  # for traial
            return 1.0
        elif initializer=="zeros":
            return 0.0
        
        
    def initialize_weights(self):
        """重み・閾値の初期化"""
        self.W = np.random.randn(self.input_dim, self.units)*self.sigma(self.kernel_initializer, self.input_dim, self.units)
        self.V = np.random.randn(self.units, self.units)*self.sigma(self.kernel_initializer, self.units, self.units)
        self.b = np.random.randn(1, self.units)*self.sigma(self.bias_initializer, self.input_dim, self.units)
        
        
    def forward_prop(self, Phi):
        """順伝播演算"""
        self.Phi = Phi
        n_samples, T = Phi.shape[0], Phi.shape[1]
        Z = np.zeros([n_samples, T, self.units])
        Phi_next = np.zeros(Z.shape)
        
        for t in range(T):
            if t==0:
                Z[:,t,:] = np.dot(Phi[:,t,:], self.W)  + self.b
            else:
                Z[:,t,:] = np.dot(Phi[:,t,:], self.W) + np.dot(Phi_next[:,t-1,:], self.V) + self.b
            Phi_next[:,t,:] = self.act_func.forward_prop(Z[:,t,:])
        self.Z = Z
        self.Phi_next = Phi_next
        
        if self.return_sequences:
            return Phi_next
        else:
            return Phi_next[:,-1,:]
        
        
    def back_prop(self, _dPhi_next):
        """逆伝播演算"""
        n_samples, n_units = _dPhi_next.shape[0], _dPhi_next.shape[-1]
        T = self.Phi.shape[1]
        
        if self.return_sequences:
            dPhi_next = _dPhi_next.copy()
        else:
            dPhi_next = np.zeros([n_samples, T, n_units])
            dPhi_next[:,-1,:] = _dPhi_next.copy()
        
        self.dW = np.zeros(self.W.shape)
        self.dV = np.zeros(self.V.shape)
        self.db = np.zeros(self.b.shape)
        Delta = np.zeros(dPhi_next.shape)
        dPhi = np.zeros(self.Phi.shape)
        
        for t in range(T-1, -1, -1): # Back Propagation Through Time
            if t==T-1:
                Delta[:,t,:] = self.act_func.back_prop(self.Z[:,t,:], dPhi_next[:,t,:])
            else:
                Delta[:,t,:] = self.act_func.back_prop(self.Z[:,t,:], dPhi_next[:,t,:] + np.dot(Delta[:,t+1,:], self.V))
            dPhi[:,t,:] = np.dot(Delta[:,t,:], self.W.T)

        for t in range(T):
            if t!=0:
                self.dV += np.dot(self.Phi_next[:,t-1,:].T, Delta[:,t,:])/n_samples
            self.dW += np.dot(self.Phi[:,t,:].T, Delta[:,t,:])/n_samples
            self.db += np.dot(np.ones([1, n_samples]), Delta[:,t,:])/n_samples
        return dPhi
    
    
    def update_weights(self, mu):
        """重み・閾値アップデート"""
        self.W -= mu * self.dW
        self.V -= mu * self.dV
        self.b -= mu * self.db