神经网络5：循环神经网络

▶ 循环神经网络

● 代码，参考【https://zybuluo.com/hanbingtao/note/541458】。主要实现了一个单层循环神经网络类 RecurrentLayer

from functools import reduce
import numpy as np

global_epsilon = 10e-4

class ReluActivator(object):                                    # ReLU 激活函数
    def forward(self, x):        
        return max(0, x)

    def backward(self, x):
        return int(x > 0)

class IdentityActivator(object):                                # 激活函数
    def forward(self, x):
        return x         

    def backward(self, x):
        return 1

def myMap(array, op):                                           # 对numpy数组进行element wise操作
    for i in np.nditer(array, op_flags=['readwrite']):
        i[...] = op(i)

class RecurrentLayer(object):
    def __init__(self, sCol, dCol, activator, ita):
        self.sCol = sCol
        self.dCol = dCol
        self.activator = activator
        self.ita = ita
        self.time = 0       
        self.stateTable = []                                    # 状态表，每行表示依次迭代的所有神经元的状态值
        self.stateTable.append(np.zeros((dCol, 1)))           
        self.U = np.random.uniform(-1e-4, 1e-4,(dCol, sCol))    # 初始化 U
        self.W = np.random.uniform(-1e-4, 1e-4,(dCol, dCol))    # 初始化 W

    def print(self):
        print("inputSize = %d, outputSize = %d, ita = %f"%(self.sCol,self.dCol,self.ita))
        print("stateTable = 
", self.stateTable)
        print("U = 
", self.U)
        print("W = 
", self.W)

    def forward(self, sArray):                                  # 前向计算        
        self.time += 1
        state = (np.dot(self.U, sArray) + np.dot(self.W, self.stateTable[-1]))
        myMap(state, self.activator.forward)
        self.stateTable.append(state)

    def backward(self, deltaArrayNextLayer, activator):    
        self.bpDelta(deltaArrayNextLayer, activator)
        self.bpGrad()

    def update(self):
        self.W -= self.ita * self.grad

    def bpDelta(self, deltaArrayNextLayer, activator):
        self.deltaTable = []                                    # 用来保存各个时刻的误差项
        for i in range(self.time):
            self.deltaTable.append(np.zeros((self.dCol, 1)))
        self.deltaTable.append(deltaArrayNextLayer)        
        for k in range(self.time - 1, 0, -1):                   # 迭代计算每个时刻的误差项
            state = self.stateTable[k+1].copy()                 # 根据k+1时刻的delta计算k时刻的delta
            myMap(self.stateTable[k+1], activator.backward)
            self.deltaTable[k] = np.dot(np.dot(self.deltaTable[k+1].T, self.W), np.diag(state[:,0])).T

    def bpGrad(self):
        self.gradTable = []                                     # 保存各个时刻的权重梯度
        for t in range(self.time + 1):
            self.gradTable.append(np.zeros((self.dCol, self.dCol)))
        for t in range(self.time, 0, -1):            
            self.gradTable[t] = np.dot(self.deltaTable[t], self.stateTable[t-1].T)
        self.grad = reduce(lambda a, b: a + b, self.gradTable, self.gradTable[0]) # 各个时刻梯度之和 gradTable[0] 初始为零阵
            
    def reset_state(self):
        self.time = 0                                           # 当前时刻初始化为t0
        self.stateTable = []                                    # 状态表，每行表示依次迭代的所有神经元的状态值
        self.stateTable.append(np.zeros((self.dCol, 1)))      

def createTestData():
    sArrayTest = [np.array([[1], [2], [3]]), np.array([[2], [3], [4]])]
    dArrayTest = np.array([[1], [2]])
    return sArrayTest, dArrayTest

def test():
    recurrentLayer = RecurrentLayer(3, 2, ReluActivator(), 1e-3)
    sArrayTest, dArrayTest = createTestData()
    recurrentLayer.forward(sArrayTest[0])
    recurrentLayer.forward(sArrayTest[1])
    recurrentLayer.backward(dArrayTest, ReluActivator())
    recurrentLayer.print()

def gradCheck():                                                # 梯度检查
    rl = RecurrentLayer(3, 2, IdentityActivator(), 1e-3)
    sArrayTest, dArrayTest = createTestData()
    rl.forward(sArrayTest[0])
    rl.forward(sArrayTest[1])            
    rl.backward(np.ones(rl.stateTable[-1].shape, dtype=np.float64), IdentityActivator())    # 以全1阵为初始 deltaNextLayer 作反向计算           
    for i in range(rl.W.shape[0]):
        for j in range(rl.W.shape[1]):
            rl.W[i,j] += global_epsilon
            rl.reset_state()
            rl.forward(sArrayTest[0])
            rl.forward(sArrayTest[1])
            err1 = np.sum(rl.stateTable[-1])
            rl.W[i,j] -= 2*global_epsilon
            rl.reset_state()
            rl.forward(sArrayTest[0])
            rl.forward(sArrayTest[1])
            err2 = np.sum(rl.stateTable[-1])
            expect_grad = (err1 - err2) / (2 * global_epsilon)
            rl.W[i,j] += global_epsilon
            print('w(%d,%d): expected %f - actural %f' % (i, j, expect_grad, rl.grad[i,j]))

if __name__ == "__main__":
    test()
    gradCheck()

● 输出结果

inputSize = 3, outputSize = 2, ita = 0.001000
stateTable =
 [array([[0.],
       [0.]]), array([[8.47915158e-05],
       [0.00000000e+00]]), array([[1.],
       [0.]])]
U =
 [[ 8.40480937e-05 -6.87332018e-05  4.60699419e-05]
 [ 6.21850646e-06 -8.85631147e-05 -5.18937760e-05]]
W =
 [[ 5.08985344e-06 -2.33817326e-05]
 [ 8.48161836e-06  4.77010103e-05]]
w(0,0): expected -0.000272 - actural -0.000272
w(0,1): expected -0.000151 - actural -0.000151
w(1,0): expected -0.000272 - actural -0.000272
w(1,1): expected -0.000151 - actural -0.000151