import numpy as np
import time
from .layers.base import Layer
from .losses import get as get_loss
from .optimizers import get as get_optimizer
from .metrics import get as get_metric
[docs]
class Sequential:
"""
Sequential model — a linear stack of layers.
Usage
-----
>>> model = Sequential([
... Dense(64, activation='relu'),
... Dropout(0.3),
... Dense(10, activation='softmax'),
... ])
>>> model.compile(
... optimizer='adam',
... loss='categorical_crossentropy',
... metrics=['accuracy'],
... )
>>> history = model.fit(X_train, y_train, epochs=20, batch_size=32,
... validation_data=(X_val, y_val))
>>> y_pred = model.predict(X_test)
>>> results = model.evaluate(X_test, y_test)
"""
def __init__(self, layers=None):
self._layers = []
self._loss_fn = None
self._optimizer = None
self._metric_fns = []
self._metric_names = []
self._compiled = False
self.history = {}
if layers:
for layer in layers:
self.add(layer)
[docs]
def add(self, layer):
if not isinstance(layer, Layer):
raise TypeError(f"Expected a Layer, got {type(layer)}")
self._layers.append(layer)
return self
[docs]
def compile(self, optimizer="adam", loss="mse", metrics=None):
self._optimizer = get_optimizer(optimizer)
self._loss_fn = get_loss(loss)
self._metric_fns = []
self._metric_names = []
if metrics:
for m in metrics:
fn = get_metric(m)
self._metric_fns.append(fn)
name = m if isinstance(m, str) else getattr(m, "__name__", str(m))
self._metric_names.append(name)
self._compiled = True
return self
def _forward(self, x, training=False):
out = x
for layer in self._layers:
out = layer.forward(out, training=training)
return out
def _backward(self, grad):
for layer in reversed(self._layers):
grad = layer.backward(grad)
return grad
def _apply_optimizer(self):
for i, layer in enumerate(self._layers):
if not layer.trainable:
continue
p = layer.params
g = layer.grads
if not p or not g:
continue
# Prefix keys with layer index so optimizer state never collides
# across layers that share parameter names (e.g. every Dense has "W")
pfx = f"L{i}_"
prefixed_p = {pfx + k: v for k, v in p.items()}
prefixed_g = {pfx + k: v for k, v in g.items() if k in p}
updated = self._optimizer.apply_gradients(prefixed_p, prefixed_g)
for k in p:
new_val = updated[pfx + k]
if hasattr(layer, k):
setattr(layer, k, new_val)
[docs]
def fit(self, x, y, epochs=10, batch_size=32, validation_data=None,
shuffle=True, verbose=1, callbacks=None):
if not self._compiled:
raise RuntimeError("Model must be compiled before training. Call model.compile().")
n = x.shape[0]
self.history = {k: [] for k in ["loss"] + self._metric_names}
if validation_data is not None:
self.history.update({f"val_{k}": [] for k in ["loss"] + self._metric_names})
for epoch in range(1, epochs + 1):
t0 = time.time()
if shuffle:
idx = np.random.permutation(n)
x, y = x[idx], y[idx]
epoch_loss = 0.0
epoch_preds = []
epoch_targets = []
n_batches = 0
for start in range(0, n, batch_size):
xb = x[start:start + batch_size]
yb = y[start:start + batch_size]
y_pred = self._forward(xb, training=True)
loss = self._loss_fn.forward(y_pred, yb)
grad = self._loss_fn.backward(y_pred, yb)
self._backward(grad)
self._apply_optimizer()
epoch_loss += loss
epoch_preds.append(y_pred)
epoch_targets.append(yb)
n_batches += 1
epoch_loss /= n_batches
self.history["loss"].append(epoch_loss)
all_preds = np.concatenate(epoch_preds, axis=0)
all_targets = np.concatenate(epoch_targets, axis=0)
metric_vals = {}
for name, fn in zip(self._metric_names, self._metric_fns):
val = fn(all_preds, all_targets)
self.history[name].append(val)
metric_vals[name] = val
val_str = ""
if validation_data is not None:
xv, yv = validation_data
val_pred = self._forward(xv, training=False)
val_loss = self._loss_fn.forward(val_pred, yv)
self.history["val_loss"].append(val_loss)
val_str = f" — val_loss: {val_loss:.4f}"
for name, fn in zip(self._metric_names, self._metric_fns):
vval = fn(val_pred, yv)
self.history[f"val_{name}"].append(vval)
val_str += f" — val_{name}: {vval:.4f}"
if verbose:
elapsed = time.time() - t0
metric_str = "".join(
f" — {k}: {v:.4f}" for k, v in metric_vals.items()
)
print(f"Epoch {epoch}/{epochs} [{elapsed:.2f}s]"
f" — loss: {epoch_loss:.4f}{metric_str}{val_str}")
if callbacks:
for cb in callbacks:
cb(epoch, self.history)
return self.history
[docs]
def predict(self, x, batch_size=None):
if batch_size is None:
return self._forward(x, training=False)
results = []
for start in range(0, x.shape[0], batch_size):
xb = x[start:start + batch_size]
results.append(self._forward(xb, training=False))
return np.concatenate(results, axis=0)
[docs]
def evaluate(self, x, y, batch_size=None, verbose=1):
y_pred = self.predict(x, batch_size=batch_size)
loss = self._loss_fn.forward(y_pred, y)
results = {"loss": loss}
for name, fn in zip(self._metric_names, self._metric_fns):
results[name] = fn(y_pred, y)
if verbose:
parts = [f"loss: {loss:.4f}"] + [f"{k}: {v:.4f}" for k, v in results.items() if k != "loss"]
print(" — ".join(parts))
return results
[docs]
def summary(self):
print("=" * 65)
print(f"{'Layer':<25} {'Output Shape':<20} {'Params':>10}")
print("=" * 65)
total = 0
for i, layer in enumerate(self._layers):
cls = type(layer).__name__
name = layer.name or f"{cls.lower()}_{i}"
n_p = layer.count_params()
total += n_p
print(f"{name:<25} {'?':<20} {n_p:>10,}")
print("=" * 65)
print(f"Total params: {total:,}")
print("=" * 65)
[docs]
def get_weights(self):
return [
{k: v.copy() for k, v in layer.params.items()}
for layer in self._layers
]
[docs]
def set_weights(self, weights):
for layer, w_dict in zip(self._layers, weights):
for k, v in w_dict.items():
if hasattr(layer, k):
setattr(layer, k, v.copy())
[docs]
def save_weights(self, path):
weights = {}
for i, layer in enumerate(self._layers):
for k, v in layer.params.items():
weights[f"layer_{i}_{k}"] = v
np.savez(path, **weights)
print(f"Weights saved to {path}.npz")
[docs]
def load_weights(self, path):
data = np.load(path if path.endswith(".npz") else path + ".npz")
for i, layer in enumerate(self._layers):
for k in layer.params:
key = f"layer_{i}_{k}"
if key in data:
setattr(layer, k, data[key])
print(f"Weights loaded from {path}")
def __repr__(self):
lines = ["Sequential("]
for i, layer in enumerate(self._layers):
lines.append(f" ({i}) {type(layer).__name__}")
lines.append(")")
return "\n".join(lines)