Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Example of MLP architecture #93

Open
pplonski opened this issue Dec 15, 2023 · 1 comment
Open

Example of MLP architecture #93

pplonski opened this issue Dec 15, 2023 · 1 comment

Comments

@pplonski
Copy link

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
Copy link

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()

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

2 participants