Rede Neural implementada com numpy

multilayer_perceptron.py

import numpy as np
from typing import List, Callable, Tuple
import math

def shuffle(a: np.ndarray, b: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
    assert len(a) == len(b)
    p = np.random.permutation(len(a))
    return a[p], b[p]


def sigmoid(x: float) -> float:
    return 1 / (1 + math.exp(-x))


def sigmoid_derivative(x: float) -> float:
    return sigmoid(x) * (1 - sigmoid(x))


class MultilayerPerceptron:
    def __init__(self, learning_rate: float, initial_weights: List[np.ndarray] = None, initial_bias: List[float] = None,
                 n_neurons_per_layer: List[float] = None,
                 activation_function: Callable[[float], float] = sigmoid,
                 derived_activation_function: Callable[[float], float] = sigmoid_derivative,
                 batch_size=None, n_epochs=100):
        if n_neurons_per_layer is not None:
            self._weights: List[np.ndarray] = []
            self._bias: List[float] = []
            for n, m in zip(n_neurons_per_layer[0:-1], n_neurons_per_layer[1:]):
                self._weights.append(np.random.randn(m, n).astype(np.float32) * np.sqrt(2.0/(m)))
                self._bias.append(0)
        else:
            if len(initial_bias) != len(initial_weights):
                raise ValueError(
                    "Deve haver um bias e uma lista de pesos por camada, sendo que ambos tenham o mesmo tamanho.")
            self._weights: List[np.ndarray] = initial_weights
            self._bias: List[float] = initial_bias

        self._learning_rate: float = learning_rate
        self._activation_function: Callable[[np.ndarray], np.ndarray] = np.vectorize(activation_function)
        self._derived_activation_function: Callable[[np.ndarray], np.ndarray] = np.vectorize(
            derived_activation_function)
        self._batch_size = batch_size
        self._n_epochs = n_epochs

    def _prepend_bias(self, x: np.ndarray) -> np.ndarray:
        return np.insert(x, 0, values=1, axis=0)

    def _feed_forward(self, x: np.ndarray) -> Tuple[List[np.ndarray], List[np.ndarray]]:
        layer_input = self._prepend_bias(x)
        layer_outputs = []
        layer_inner_outputs = []  # Antes da função de ativação
        for layer_weights, layer_bias in zip(self._weights, self._bias):
            # Adiciona o bias
            layer_weights_with_bias = np.insert(layer_weights, 0, values=layer_bias, axis=1)
            # Calcula saída dos neurônios antes da função de ativação
            layer_inner_output = (layer_input * layer_weights_with_bias).sum(axis=1)
            layer_inner_outputs.append(layer_inner_output)
            # Aplica a função de ativação
            layer_output = self._activation_function(layer_inner_output)
            layer_outputs.append(layer_output)
            # Passa a saída dessa camada como entrada para a próxima
            layer_input = self._prepend_bias(layer_output)

        return layer_outputs, layer_inner_outputs

    def predict(self, X: np.ndarray) -> np.ndarray:
        X = np.atleast_2d(X)
        output = []
        for x in X:
            output.append(self._feed_forward(x)[0][-1])
        return np.array(output)

    def _step(self, X: np.ndarray, Y: np.ndarray) -> float:
        updated_weights: List[np.ndarray] = [np.copy(neuron_weights) for neuron_weights in self._weights]
        updated_bias = self._bias[:]

        errors = []

        for x, y in zip(X, Y):
            layer_outputs, layer_inner_outputs = self._feed_forward(x)
            errors.append((0.5 * (y - layer_outputs[-1]) ** 2).sum())

            layer_inputs = [x] + layer_outputs

            delta = None
            for i in range(len(self._bias)):
                layer_inner_output = layer_inner_outputs[-i - 1]
                layer_output = layer_outputs[-i - 1]
                layer_input = layer_inputs[-i - 2]
                input_weights = np.array(self._weights[-i - 1])
                output_weights = np.array(self._weights[-i]) if i > 0 else None

                # Derivada do erro pela saída (após função de ativação)
                error_derived_by_layer_output = layer_output - y if i == 0 else (output_weights.T * delta).sum(axis=1)
                # Derivada da saída (após função de ativação) pela saída (antes da função de ativação)
                layer_output_derived_by_layer_inner_output = self._derived_activation_function(layer_inner_output)
                delta = error_derived_by_layer_output * layer_output_derived_by_layer_inner_output
                # Derivada da saída (antes da função de ativação) pelo peso
                layer_inner_output_derived_by_weight = (np.ones(input_weights.shape) * layer_input).T
                # Derivada da saída (antes da função de ativação) pelo bias
                layer_inner_output_derived_by_bias = 1

                error_derived_by_weights = (delta * layer_inner_output_derived_by_weight).T
                error_derived_by_bias = (delta * layer_inner_output_derived_by_bias).sum()

                updated_weights[-i - 1] = updated_weights[-i - 1] - self._learning_rate * error_derived_by_weights
                updated_bias[-i - 1] = updated_bias[-i - 1] - self._learning_rate * error_derived_by_bias

        self._weights = updated_weights
        self._bias = updated_bias

        return sum(errors) / float(len(errors))

    def fit(self, X: np.ndarray, Y: np.ndarray):
        X = np.atleast_2d(X)
        Y = np.atleast_2d(Y)

        errors_history = []
        for epoch in range(self._n_epochs):
            errors = []
            X_train, Y_train = shuffle(X, Y)
            minibatch_size = min(self._batch_size, len(X_train)) if self._batch_size else len(X_train)

            for i in range(0, X_train.shape[0], minibatch_size):
                # Get pair of (X, y) of the current minibatch/chunk
                X_train_mini = X_train[i:i + minibatch_size]
                Y_train_mini = Y_train[i:i + minibatch_size]
                errors.append(self._step(X_train_mini, Y_train_mini))

            epoch_error = sum(errors) / float(len(errors))
            errors_history.append(epoch_error)

        return errors_history

Implementação feita para a matéria de doutorado de Deep Learning em 2018.
Essa implementação suporta:

• Pesos aleatórios (he init) ou pesos pré definidos.
• Ajustar a quantidade de épocas de treinamento.
• Ajustar mini-batch.
• Utilizar qualquer função de ativação (desde que sua derivada também seja passada como parâmetro).