Example of MLP architecture

[pplonski]

Thank you for this package. I'm looking for some example on how to implement simple MLP (Multi Layer Perceptron) with this package. Any code snippets or tutorials are welcome.

Below is some code that I glue, but I have no idea on how to do backpropagation, I would like to have fit() method implemented.

Thank you!

from numpy_ml.neural_nets.losses import CrossEntropy, SquaredError
from numpy_ml.neural_nets.utils import minibatch
from numpy_ml.neural_nets.activations import ReLU, Sigmoid
from numpy_ml.neural_nets.layers import FullyConnected
from numpy_ml.neural_nets.optimizers.optimizers import SGD

optimizer = SGD()
loss = SquaredError()

class MLP:

    def __init__(self):
        self.nn = OrderedDict()
        self.nn["L1"] = FullyConnected(
            10, act_fn="ReLU", optimizer=optimizer
        )
        self.nn["L2"] = FullyConnected(
            1, act_fn="Sigmoid", optimizer=optimizer
        )

    def forward(self, X, retain_derived=True):
        Xs = {}
        out, rd = X, retain_derived
        for k, v in self.nn.items():
            Xs[k] = out
            out = v.forward(out, retain_derived=rd)
        return out, Xs
        
    def backward(self, grad, retain_grads=True):
        dXs = {}
        out, rg = grad, retain_grads
        for k, v in reversed(list(self.nn.items())):
            dXs[k] = out
            out = v.backward(out, retain_grads=rg)
        return out, dXs

[achal-khanna]

Hello,

Thank you for sharing your MLP implementation. I've taken a look at your code and added a fit() method to handle the training process, including backpropagation and weight updates.

Below is the updated version of your code. Also, I had to create a custom optimizer GradientDescentOptimizer as there was an issue of shape mismatch during backward propagation using inbuilt optimizers #78.

import numpy_ml.neural_nets as npml
import numpy as np
import time
import matplotlib.pyplot as plt
from collections import OrderedDict

# Creating a custom optimizer
class GradientDescentOptimizer(npml.optimizers.OptimizerBase):
    def __init__(self, lr=0.01, scheduler=None):
        super().__init__(lr, scheduler)
        self.hyperparameters.update({
            "learning_rate": lr
        })

    def update(self, param, param_grad, param_name, cur_loss=None):
        lr = self.lr_scheduler(self.cur_step, cur_loss)

        # Perform the gradient descent update
        update = lr * param_grad
        return param - update

class MLP:
    def __init__(self, init: npml.initializers = "he_uniform", optimizer: npml.optimizers = GradientDescentOptimizer(lr = 0.005)) -> None:
        self.init = init
        self.optimizer = optimizer

        self._derived_variables = {}
        self._gradients = {}
        self._build_model()

    def _build_model(self) -> None:
        self.model = OrderedDict()
        self.model["L1"] = npml.layers.FullyConnected(
            10, act_fn=npml.activations.ReLU(), optimizer=self.optimizer, init = self.init
        )
        self.model["L2"] = npml.layers.FullyConnected(
            1, act_fn=npml.activations.Identity(), optimizer=self.optimizer, init = self.init
        ) 

    def forward(self, X: np.ndarray, retain_derived: bool = True) -> {np.ndarray, dict}:
        Xs = {}
        out, rd = X, retain_derived
        for k, v in self.model.items():
            Xs[k] = out
            out = v.forward(out, retain_derived=rd)
        return out, Xs
    
    def backward(self, grad: np.ndarray, retain_grads: bool = True) -> {np.ndarray, dict}:
        dXs = {}
        out, rg = grad, retain_grads
        for k, v in reversed(list(self.model.items())):
            dXs[k] = out
            out = v.backward(out, retain_grads=rg)
        return out, dXs

    def update(self, loss = None):
        for k, v in reversed(list(self.model.items())):
            v.update(loss)

    def fit(self, X_train: np.ndarray, Y_train:np.ndarray, n_epochs: int = 100, batchsize: int = 128, verbose: bool = True) -> None:
        self.verbose = verbose
        self.n_epochs = n_epochs
        self.batchsize = batchsize

        prev_loss = np.inf
        for epoch in range(self.n_epochs):
            loss, estart = 0.0, time.time()
            batch_generator, n_batches = npml.utils.minibatch(X_train, self.batchsize, shuffle=False)

            # Iterating over batch
            for j, b_ix in enumerate(batch_generator):
                bsize, bstart = len(b_ix), time.time()

                X_batch = X_train[b_ix]
                Y_batch = Y_train[b_ix]

                Y_pred, Xs = self.forward(X_batch)
                batch_loss = npml.losses.SquaredError.loss(Y_batch, Y_pred)
                loss += batch_loss

                batch_grad = npml.losses.SquaredError.grad(Y_batch, Y_pred, Y_pred, npml.activations.Identity())
                _, dXs = self.backward(batch_grad)
                self.update(batch_loss)

                if self.verbose:
                    fstr = "\t[Batch {}/{}] Train loss: {:.3f} ({:.1f}s/batch)"
                    print(fstr.format(j + 1, n_batches, batch_loss, time.time() - bstart))

            loss /= n_batches
            fstr = "[Epoch {}] Avg. loss: {:.3f}  Delta: {:.3f} ({:.2f}m/epoch)"
            print(fstr.format(epoch + 1, loss, prev_loss - loss, (time.time() - estart) / 60.0))
            prev_loss = loss

num_samples = 100  
num_features = 1    

X = 2 * np.random.rand(num_samples, num_features)
y = X*X + np.random.randn(num_samples, num_features) * 0.1

model = MLP()
model.fit(X_train=X, Y_train=y, n_epochs=100, batchsize=10)
y_pred, Xs = model.forward(X)

plt.scatter(X, y, color="blue")
plt.scatter(X, y_pred, color = "red")
plt.xlabel("X")
plt.ylabel("y")
plt.title("Synthetic Dataset")
plt.show()