Open pplonski opened 8 months ago
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()
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!