В первой статье речь пойдет о реализации модуля Sequential и слоёв Dense и Input через библиотеку Numpy для многослойной NN.
Данная статья нацелена на практическую реализацию слоев с минимумом теории, и предполагается что читатель знаком с базовой теорией обучения нейронных сетей.
Начнём с импорта библиотек:
import numpy as np
Реализация Dense слоя
class DenseLayer(): def __init__(self, units=1, activation='relu', weights=np.array([]), b=np.array([])): self.units = units self.fl_init = True self.activation = activation self.weights = weights self.b_new = b self.w, self.b = np.array([]), np.array([])
Поясню параметры:
units - количество нейронов
activation - функция активации
weights и b_new - скорректированные веса и смещения, которые мы в дальнейшем будем передавать модели
w, b - начальные веса и смещения
fl_init - флаг, показывающий созданы ли начальные веса и смещения
Далее воспользуемся магическим методом call, чтобы превратить наш класс в функтор:
def __call__(self, x): if (self.fl_init == True) and (self.weights.shape[0] == 0): self.w = np.random.normal(loc=0.0, scale=1.0, size=(x.shape[-1], self.units))/np.sqrt(2.0/x.shape[-1]) + 0.001 self.b = np.ones(shape=(self.units, ), dtype=np.float32) self.fl_init = False # print(self.w.shape, self.weights) elif self.weights.shape[0] != 0: self.weights = self.weights.reshape((x.shape[-1], self.units)) self.w = self.weights self.fl_init = False self.b_new = self.b_new.reshape((self.units, )) self.b = self.b_new self.fl_init = False y = x.dot(self.w) + self.b if self.activation == 'relu': return np.maximum(np.zeros(shape=y.shape), y), self.w, self.b, 1, self.units, self.activation if self.activation == 'Leaky_relu': return np.maximum(0.01*y, y), self.w, self.b, 1, self.units, self.activation if self.activation == 'softmax': return np.exp(y)/np.sum(np.exp(y), axis=0), self.w, self.b, 1, self.units, self.activation if self.activation == 'sigmoid': return 1 / (1 + np.exp(-y)), self.w, self.b, 1, self.units, self.activation if self.activation == 'tanh': return (np.exp(2*y) - 1)/(np.exp(2*y) + 1), self.w, self.b, 1, self.units, self.activation if self.activation == 'linear': return y, self.w, self.b, 1, self.units, self.activation
Принцип работы такой:
Сначала идет проверка, что начальные веса еще не созданы и нам не передавали уже скорректированные.
Если условие выполняется, то начальные веса создаются с нормальным распределением (математическое ожидание = 0, дисперсия = 1), а начальные смещения задаём как единицы -> Флаг переводим в значение False.
Если же скорректированные веса были переданы, то заменяем начальные веса на них.
Вычисляем y
Пропускаем её через переданную функцию активации (если не передана - то пропускаем через 'relu')
Реализуем простенький класс Input:
class Input(): def __init__(self, shape=None): self.shape = shape def __call__(self, x): if self.shape is not None: if x.shape != self.shape: return x.reshape(shape=self.shape), 0 else: return x, 0 return x, 0
Перейдем к написанию модуля Sequential:
class Sequential(): def __init__(self, layers): self.layers = layers # слои в NN
Начнем с метода fit:
Сперва реализуем вспомогательную функцию predict, которая будет возвращать для каждого слоя выходы, функции активации, веса и смещения, используемые слой и количество нейронов. Она нам потребуется в дальнейшем для метода обратного распространения ошибка(BP).
def predict(x): activations = [] predict_for_layers = [] weights = [] b_coef = [] layer_2 = [] units = [] predict = self.layers[0](x) layer_2.append(predict[1]) predict_for_layers.append(predict[0]) for i in range(1, len(self.layers)): predict = self.layers[i](predict[0]) activations.append(predict[-1]) predict_for_layers.append(predict[0]) weights.append(predict[1]) b_coef.append(predict[2]) layer_2.append(predict[3]) units.append(predict[4]) #print(len(units)) return predict_for_layers, activations, weights, b_coef, layer_2, units
Далее реализуем функции подсчета градиента:
def sigmoid_gradient(output): return output * (1 - output) def tanh_gradient(out): return 1/((np.exp(out) + np.exp(-out)/2)**2) def relu_gradient(x): return (x > 0) * 1 def leaky_relu_gradient(x): return (x > 0) * 1 + (x <= 0) * 0.01 def linear_gradient(x): return 1
Перейдем к самой реализации Backpropagation:
list_back = self.layers[::-1] for elem in range(x_input.shape[0]): x, y = x_input[elem].reshape(1, -1), y_input[elem] for epoch in range(epochs): predict_layers = predict(x) # 1 - y, 2 - w, 3 - b, 4 - слой, 5 - кол. нейронов predict_for_layers, activations, weights, b_coef, layers = predict_layers[0][::-1], predict_layers[1][::-1], predict_layers[2][::-1], predict_layers[3][::-1], predict_layers[4] units = predict_layers[5] layer_error = predict_for_layers[0] - y if len(layer_error.shape) == 1: layer_error = layer_error.reshape(1, -1) for ind in range(len(list_back) - 1): delta_weights = 0 if activations[ind] == 'linear': delta_weights = layer_error * relu_gradient(predict_for_layers[ind]) if activations[ind] == 'Leaky_relu': delta_weights = layer_error * leaky_relu_gradient(predict_for_layers[ind]) if activations[ind] == 'relu': delta_weights = layer_error * relu_gradient(predict_for_layers[ind]) if activations[ind] == 'sigmoid': delta_weights = layer_error * sigmoid_gradient(predict_for_layers[ind]) if activations[ind] == 'tanh': delta_weights = layer_error * tanh_gradient(predict_for_layers[ind]) b_coef[ind] -= alpha * (np.full(b_coef[ind].shape, layer_error.sum())) layer_error = delta_weights.dot(np.transpose(weights[ind])) weights[ind] -= alpha * (np.transpose(predict_for_layers[ind + 1]).dot(delta_weights)) weights_inp = weights[::-1] b_inp = b_coef[::-1] activations_inp = activations[::-1] for indx in range(1, len(self.layers)): if layers[indx] == 1: self.layers[indx] = DenseLayer(units=units[indx - 1], weights=weights_inp[indx - 1], b=b_inp[indx - 1], activation=activations_inp[indx - 1])
Опишу принцип работы:
В цикле по количеству эпох:
берется пара элементов - label + input.
С помощью ранее написанной функции predict считается выход (списки переворачиваются, чтобы начинать с последнего слоя).
Считаем ошибку на последнем слое.
delta_weights - производная, взвешенная по ошибкам - (уменьшаем ошибки предсказаний, сделанных с высокой уверенностью. Если наклон касательной линии(значение производной) был небольшим, то в сети содержится либо очень большое, либо очень малое значение) - считаем локальный градиент.
Далее перезаписываем layer_error.
Обновляем веса по следующему правилу:
обновление весов в NN
Проходимся по слоям и перезаписываем их с новыми весами и смещениями.
Остается написать функцию predict без вспомогательных выходов:
def predict(self, x): predict = self.layers[0](x) for i in range(1, len(self.layers)): predict = self.layers[i](predict[0]) return predict
Код полностью:
class Sequential(): def __init__(self, layers): self.layers = layers def fit(self, x_input, y_input, epochs=50, alpha=0.01): def predict(x): activations = [] predict_for_layers = [] weights = [] b_coef = [] layer_2 = [] units = [] predict = self.layers[0](x) layer_2.append(predict[1]) predict_for_layers.append(predict[0]) for i in range(1, len(self.layers)): predict = self.layers[i](predict[0]) activations.append(predict[-1]) predict_for_layers.append(predict[0]) weights.append(predict[1]) b_coef.append(predict[2]) layer_2.append(predict[3]) units.append(predict[4]) return predict_for_layers, activations, weights, b_coef, layer_2, units def sigmoid_gradient(output): return output * (1 - output) def tanh_gradient(out): return 1/((np.exp(out) + np.exp(-out)/2)**2) def relu_gradient(x): return (x > 0) * 1 def leaky_relu_gradient(x): return (x > 0) * 1 + (x <= 0) * 0.01 def linear_gradient(x): return 1 list_back = self.layers[::-1] for epoch in range(epochs): for elem in range(x_input.shape[0]): x, y = x_input[elem].reshape(1, -1), y_input[elem] predict_layers = predict(x) # 1 - y, 2 - w, 3 - b, 4 - слой, 5 - кол. нейронов predict_for_layers, activations, weights, b_coef, layers = predict_layers[0][::-1], predict_layers[1][::-1], predict_layers[2][::-1], predict_layers[3][::-1], predict_layers[4] units = predict_layers[5] layer_error = predict_for_layers[0] - y if len(layer_error.shape) == 1: layer_error = layer_error.reshape(1, -1) for ind in range(len(list_back) - 1): delta_weights = 0 if activations[ind] == 'linear': delta_weights = layer_error * relu_gradient(predict_for_layers[ind]) if activations[ind] == 'Leaky_relu': delta_weights = layer_error * leaky_relu_gradient(predict_for_layers[ind]) if activations[ind] == 'relu': delta_weights = layer_error * relu_gradient(predict_for_layers[ind]) if activations[ind] == 'sigmoid': delta_weights = layer_error * sigmoid_gradient(predict_for_layers[ind]) if activations[ind] == 'tanh': delta_weights = layer_error * tanh_gradient(predict_for_layers[ind]) b_coef[ind] -= alpha * (np.full(b_coef[ind].shape, layer_error.sum())) layer_error = delta_weights.dot(np.transpose(weights[ind])) weights[ind] -= alpha * (np.transpose(predict_for_layers[ind + 1]).dot(delta_weights)) weights_inp = weights[::-1] b_inp = b_coef[::-1] activations_inp = activations[::-1] for indx in range(1, len(self.layers)): if layers[indx] == 1: self.layers[indx] = DenseLayer(units=units[indx - 1], weights=weights_inp[indx - 1], b=b_inp[indx - 1], activation=activations_inp[indx - 1]) # Предсказание значений def predict(self, x): predict = self.layers[0](x) for i in range(1, len(self.layers)): #print(predict[0].shape) predict = self.layers[i](predict[0]) return predict
Приведу простой пример использования:
model = Sequential([ Input(), DenseLayer(units=3, activation='relu'), DenseLayer(units=1, activation='relu') ]) x = np.array([[3., 2.], [2., 2.], [3., 3.], [4., 4.]]) y = [5, 4, 6, 8] #print(x.shape) model.fit(x, y, epochs=40) x_test = np.array([[6., 5.], [5., 5.]]) print(model.predict(x_test)[0])
Выход:
[[11.26779558]
[ 9.52089243]]
Выводы
Как можно заметить, нейронная сеть после обучения смогла выдать практически правильный ответ на пример, который не встречала при обучении.
В реализации есть много недочётов, так что если есть комментарии, что исправить - пишите.
