Orchestrating Your Network: The Model Class | CodeSignal Learn

Orchestrating Your Network: The Model Class | CodeSignal Learn

Introduction

Welcome back to the third lesson of "Building and Applying Your Neural Network Library"! You've made tremendous progress in this course. In our first lesson, we successfully modularized our core neural network components — dense layers and activation functions. Then, in our previous lesson, we organized our training components by creating dedicated modules for loss functions and optimizers. We now have a well-structured foundation with a clean separation of concerns. However, as you may have noticed from our previous training examples, we're still writing quite a bit of boilerplate code for each training session. We manually create layers, set up optimizers, write training loops, handle forward and backward passes, and coordinate all these components ourselves. While this gives us complete control, it also means we're repeating the same orchestration logic every time we want to train a model. In this lesson, we're going to orchestrate all these components into a unified, high-level interface. We'll build a powerful Model abstract class that acts as the conductor of our neural network orchestra, coordinating layers, optimizers, and loss functions through clean, intuitive methodslike compile(), fit(), and predict(), as well as a SequentialModel concrete subclass that can be seen as a better and improved replacement for the MLP class we developed previously, providing a much more elegant and maintainable API for building and training neural networks. Let's get started!

The Need for Orchestration

Think of a symphony orchestra — while each musician is skilled at playing their individual instrument, the magic happens when a conductor coordinates all these talents toward a unified performance. Similarly, we've built excellent individual components (layers, optimizers, losses), but we need a conductor to orchestrate them into a seamless training experience. Currently, our training process requires us to manually coordinate several moving parts: Instantiate layers and build our network architecture Create an optimizer with specific parameters Define our loss function Implement the training loop with forward passes, loss calculations, backward passes, and weight updates This manual orchestration is error-prone and repetitive — exactly the kind of work that should be automated. What we need is a Model class that serves as this conductor, providing a high-level API that handles the complexities of training coordination while still giving us the flexibility to customize our network architecture, choose different optimizers and loss functions, and control training parameters. Python # Current goal: elegant and extensible training API model = SequentialModel() model.add(DenseLayer(...)) model.compile(optimizer_name='sgd', learning_rate=0.5, loss_name='mse') model.fit(X_train, y_train, epochs=1000)

Sign up

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Designing the Base Model Class

Now that we understand the role of abstract classes, let's implement our Model orchestrator: import numpy as np from abc import ABC, abstractmethod from neuralnets.optimizers import SGD from neuralnets.losses import mse_loss, mse_loss_derivative class Model(ABC): def __init__(self): self.layers = [] # Layer stack - populated by subclasses self.optimizer = None # Training optimizer self.loss_fn = None # Loss function self.loss_fn_derivative = None # Loss derivative function self.is_compiled = False # Compilation status flag def compile(self, optimizer_name='sgd', learning_rate=0.01, loss_name='mse'): """ Configures the model for training by setting up optimizer and loss function. This method is easily extensible to support new optimizers and loss functions. """ # Set up the optimizer - easily extensible for new optimizers if optimizer_name.lower() == 'sgd': self.optimizer = SGD(learning_rate=learning_rate) else: raise ValueError(f"Unsupported optimizer: {optimizer_name}") # Set up the loss function - easily extensible for new loss functions if loss_name.lower() == 'mse': self.loss_fn = mse_loss self.loss_fn_derivative = mse_loss_derivative else: raise ValueError(f"Unsupported loss function: {loss_name}") self.is_compiled = True print(f"Model compiled with optimizer: {optimizer_name} (lr: {learning_rate}), loss: {loss_name}") This foundation establishes all the essential components our model will need to coordinate: Layer stack management: The layers list holds our network architecture Optimizer coordination: Centralized optimizer configuration and management Loss function setup: Unified loss function and derivative handling Compilation validation: The is_compiled flag prevents training misconfiguration Future extensibility: The string-based configuration system makes adding new optimizers and loss functions straightforward

Implementing Training and Prediction Logic

Now let's add the core training and prediction functionality. The abstract method pattern we're using ensures scalability — any new model architecture can plug into our training orchestration system by simply implementing the required methods. class Model(ABC): # ... previous code ... @abstractmethod def _forward(self, inputs): """ Performs the forward pass logic specific to the model architecture. Should be implemented by subclasses (e.g., SequentialModel). """ pass @abstractmethod def _backward(self, d_loss_wrt_prediction): """ Performs the backward pass logic specific to the model architecture. Should be implemented by subclasses (e.g., SequentialModel). """ pass def predict(self, X): """Generates output predictions for the input samples.""" return self._forward(X) def fit(self, X_train, y_train, epochs, batch_size=None, verbose=1): """Trains the model for a fixed number of epochs with automatic orchestration.""" if not self.is_compiled: raise RuntimeError("Model must be compiled before training. Call model.compile().") num_samples = X_train.shape[0] if batch_size is None: batch_size = num_samples # Full batch gradient descent for epoch in range(epochs): epoch_loss = 0.0 # Shuffle data each epoch for better convergence indices = np.arange(num_samples) np.random.shuffle(indices) # Process data in batches for i in range(0, num_samples, batch_size): batch_indices = indices[i:i+batch_size] X_batch = X_train[batch_indices] y_batch = y_train[batch_indices] # Coordinated forward-backward-update cycle y_pred = self._forward(X_batch) # Forward pass loss = self.loss_fn(y_batch, y_pred) # Loss calculation epoch_loss += loss * X_batch.shape[0] # Accumulate loss d_loss_wrt_pred = self.loss_fn_derivative(y_batch, y_pred) # Loss gradient self._backward(d_loss_wrt_pred) # Backward pass # Update all trainable layers for layer in self.layers: if hasattr(layer, 'weights') and hasattr(layer, 'd_weights') and layer.d_weights is not None: self.optimizer.update(layer) # Progress reporting epoch_loss /= num_samples if verbose and ((epochs < 10) or ((epoch + 1) % (epochs // 10) == 0) or epoch == 0): print(f"Epoch {epoch+1:4d}/{epochs}, Loss: {epoch_loss:.6f}") if verbose: print("Training finished.") The fit() method orchestrates the entire training process with several key responsibilities: Validation: Ensures the model is properly compiled before training Batch processing: Handles flexible batch sizes for memory efficiency and training stability Data management: Shuffles data each epoch to prevent overfitting to data order Training coordination: Manages the forward-backward-update cycle automatically Progress tracking: Provides informative training progress feedback Architecture independence: Works with any model architecture through the abstract method pattern, with _forward and _backward methods This design ensures our training system can scale to handle different model types, training strategies, and dataset sizes without requiring code changes to the core training logic.

Building the Sequential Model

Now we can create our concrete SequentialModel class that implements the abstract methods from our base Model . This class represents the familiar stack-of-layers architecture we've been working with, but now with a much cleaner interface and full integration into our extensible framework. import numpy as np from neuralnets.models import Model class SequentialModel(Model): def __init__(self, layers_list=None): """Initialize a sequential model, optionally with a pre-built layer list.""" super().__init__() # Initialize base Model attributes if layers_list: for layer in layers_list: self.add(layer) def add(self, layer): """Add a layer instance to the sequential stack.""" self.layers.append(layer) def _forward(self, inputs): """Forward pass: data flows sequentially through all layers.""" current_input = inputs for layer in self.layers: current_input = layer.forward(current_input) return current_input def _backward(self, d_loss_wrt_prediction): """Backward pass: gradients flow in reverse through all layers.""" current_d_loss = d_loss_wrt_prediction for layer in reversed(self.layers): current_d_loss = layer.backward(current_d_loss) The SequentialModel demonstrates excellent separation of concerns : Initialization flexibility : Supports both empty initialization and pre-built layer lists Architecture building : The add() method provides the familiar interface for building networks layer by layer Sequential processing : Forward pass flows data through layers in order, backward pass propagates gradients in reverse Clean responsibility : Focuses solely on managing a linear sequence of layers while inheriting all training orchestration from the parent class This design makes our library highly extensible — we can easily add other model types like convolutional networks or attention mechanisms by implementing the same abstract interface, while all the training logic remains reusable.

Using Our Orchestrated Model

Now let's see our orchestration in action! Here's how we can solve the XOR problem using our new high-level API, which is much cleaner and more maintainable than our previous manual approach. Notice how this same interface will work seamlessly when we extend our library with new layers, optimizers, or loss functions. import numpy as np from neuralnets.layers import DenseLayer from neuralnets.models import SequentialModel # XOR dataset X_train = np.array([[0,0], [0,1], [1,0], [1,1]], dtype=float) y_train = np.array([[0], [1], [1], [0]], dtype=float) # Build the model architecture - easily extensible to new layer types model = SequentialModel() model.add(DenseLayer(n_inputs=2, n_neurons=4, activation_fn_name='relu', weight_init_scale=0.1)) model.add(DenseLayer(n_inputs=4, n_neurons=1, activation_fn_name='sigmoid', weight_init_scale=0.1)) # Configure for training - ready for new optimizers and loss functions model.compile(optimizer_name='sgd', learning_rate=0.5, loss_name='mse') # Train the model - same interface works for any model architecture print("Training model for XOR problem...") model.fit(X_train, y_train, epochs=1000, batch_size=X_train.shape[0], verbose=1) # Test the trained model print("\n--- After Training ---") y_pred = model.predict(X_train) print("Input | True Output | Predicted Output | Rounded Prediction") for i in range(len(X_train)): print(f"{X_train[i]} |{y_train[i][0]:10.0f} |{y_pred[i][0]:15.4f} |{np.round(y_pred[i][0]):18.0f}") Notice how much cleaner this is compared to our previous training scripts! The beauty of this approach is that it maintains the flexibility to customize every aspect of our network while eliminating the repetitive boilerplate code. The extensible design means we can easily experiment with different architectures, optimizers, or hyperparameters without rewriting the training logic each time.

Discussing the Output

When we run our orchestrated model, we can see how it successfully learns the XOR problem while providing clear feedback about the training process. The consistent interface and automated orchestration ensure reliable, reproducible results across different model configurations. Model compiled with optimizer: sgd (lr: 0.5), loss: mse Training model for XOR problem... Epoch 1/1000, Loss: 0.249842 Epoch 100/1000, Loss: 0.219619 Epoch 200/1000, Loss: 0.055226 Epoch 300/1000, Loss: 0.014176 Epoch 400/1000, Loss: 0.006978 Epoch 500/1000, Loss: 0.004460 Epoch 600/1000, Loss: 0.003217 Epoch 700/1000, Loss: 0.002481 Epoch 800/1000, Loss: 0.001689 Epoch 900/1000, Loss: 0.001444 Epoch 1000/1000, Loss: 0.001444 Training finished. --- After Training --- Input | True Output | Predicted Output | Rounded Prediction [0. 0.] | 0 | 0.0471 | 0 [0. 1.] | 1 | 0.9742 | 1 [1. 0.] | 1 | 0.9743 | 1 [1. 1.] | 0 | 0.0471 | 0 The results demonstrate excellent performance — our orchestrated model achieves the same learning quality as our previous manual implementations, but with much cleaner, more maintainable code. The loss decreases smoothly from 0.25 to 0.0014, and the final predictions correctly solve the XOR problem with high confidence. More importantly, the training process is now fully automated, consistently implemented, and ready to scale to more complex problems and architectures.

Python

import numpy as np
from abc import ABC, abstractmethod
from neuralnets.optimizers import SGD
from neuralnets.losses import mse_loss, mse_loss_derivative

class Model(ABC):
    def __init__(self):
        self.layers = []              # Layer stack - populated by subclasses
        self.optimizer = None         # Training optimizer
        self.loss_fn = None          # Loss function
        self.loss_fn_derivative = None # Loss derivative function
        self.is_compiled = False     # Compilation status flag

    def compile(self, optimizer_name='sgd', learning_rate=0.01, loss_name='mse'):
        """
        Configures the model for training by setting up optimizer and loss function.
        This method is easily extensible to support new optimizers and loss functions.
        """
        # Set up the optimizer - easily extensible for new optimizers
        if optimizer_name.lower() == 'sgd':
            self.optimizer = SGD(learning_rate=learning_rate)
        else:
            raise ValueError(f"Unsupported optimizer: {optimizer_name}")

        # Set up the loss function - easily extensible for new loss functions
        if loss_name.lower() == 'mse':
            self.loss_fn = mse_loss
            self.loss_fn_derivative = mse_loss_derivative
        else:
            raise ValueError(f"Unsupported loss function: {loss_name}")
        
        self.is_compiled = True
        print(f"Model compiled with optimizer: {optimizer_name} (lr: {learning_rate}), loss: {loss_name}")

Python

class Model(ABC):
    # ... previous code ...
    
    @abstractmethod
    def _forward(self, inputs):
        """
        Performs the forward pass logic specific to the model architecture.
        Should be implemented by subclasses (e.g., SequentialModel).
        """
        pass

    @abstractmethod
    def _backward(self, d_loss_wrt_prediction):
        """
        Performs the backward pass logic specific to the model architecture.
        Should be implemented by subclasses (e.g., SequentialModel).
        """
        pass

    def predict(self, X):
        """Generates output predictions for the input samples."""
        return self._forward(X)

    def fit(self, X_train, y_train, epochs, batch_size=None, verbose=1):
        """Trains the model for a fixed number of epochs with automatic orchestration."""
        if not self.is_compiled:
            raise RuntimeError("Model must be compiled before training. Call model.compile().")
        
        num_samples = X_train.shape[0]
        if batch_size is None:
            batch_size = num_samples  # Full batch gradient descent

        for epoch in range(epochs):
            epoch_loss = 0.0
            # Shuffle data each epoch for better convergence
            indices = np.arange(num_samples)
            np.random.shuffle(indices)
            
            # Process data in batches
            for i in range(0, num_samples, batch_size):
                batch_indices = indices[i:i+batch_size]
                X_batch = X_train[batch_indices]
                y_batch = y_train[batch_indices]

                # Coordinated forward-backward-update cycle
                y_pred = self._forward(X_batch)                              # Forward pass
                loss = self.loss_fn(y_batch, y_pred)                       # Loss calculation
                epoch_loss += loss * X_batch.shape[0]                      # Accumulate loss
                
                d_loss_wrt_pred = self.loss_fn_derivative(y_batch, y_pred)  # Loss gradient
                self._backward(d_loss_wrt_pred)                             # Backward pass

                # Update all trainable layers
                for layer in self.layers:
                    if hasattr(layer, 'weights') and hasattr(layer, 'd_weights') and layer.d_weights is not None:
                        self.optimizer.update(layer)
            
            # Progress reporting
            epoch_loss /= num_samples
            if verbose and ((epochs < 10) or ((epoch + 1) % (epochs // 10) == 0) or epoch == 0):
                print(f"Epoch {epoch+1:4d}/{epochs}, Loss: {epoch_loss:.6f}")
        
        if verbose:
            print("Training finished.")

Python

import numpy as np
from neuralnets.models import Model

class SequentialModel(Model):
    def __init__(self, layers_list=None):
        """Initialize a sequential model, optionally with a pre-built layer list."""
        super().__init__()  # Initialize base Model attributes
        if layers_list:
            for layer in layers_list:
                self.add(layer)

    def add(self, layer):
        """Add a layer instance to the sequential stack."""
        self.layers.append(layer)

    def _forward(self, inputs):
        """Forward pass: data flows sequentially through all layers."""
        current_input = inputs
        for layer in self.layers:
            current_input = layer.forward(current_input)
        return current_input

    def _backward(self, d_loss_wrt_prediction):
        """Backward pass: gradients flow in reverse through all layers."""
        current_d_loss = d_loss_wrt_prediction
        for layer in reversed(self.layers):
            current_d_loss = layer.backward(current_d_loss)

Python

import numpy as np
from neuralnets.layers import DenseLayer
from neuralnets.models import SequentialModel

# XOR dataset
X_train = np.array([[0,0], [0,1], [1,0], [1,1]], dtype=float)
y_train = np.array([[0], [1], [1], [0]], dtype=float)

# Build the model architecture - easily extensible to new layer types
model = SequentialModel()
model.add(DenseLayer(n_inputs=2, n_neurons=4, activation_fn_name='relu', weight_init_scale=0.1))
model.add(DenseLayer(n_inputs=4, n_neurons=1, activation_fn_name='sigmoid', weight_init_scale=0.1))

# Configure for training - ready for new optimizers and loss functions
model.compile(optimizer_name='sgd', learning_rate=0.5, loss_name='mse')

# Train the model - same interface works for any model architecture
print("Training model for XOR problem...")
model.fit(X_train, y_train, epochs=1000, batch_size=X_train.shape[0], verbose=1)

# Test the trained model
print("\n--- After Training ---")
y_pred = model.predict(X_train)
print("Input  | True Output | Predicted Output | Rounded Prediction")
for i in range(len(X_train)):
    print(f"{X_train[i]} |{y_train[i][0]:10.0f} |{y_pred[i][0]:15.4f} |{np.round(y_pred[i][0]):18.0f}")

text

Model compiled with optimizer: sgd (lr: 0.5), loss: mse

Training model for XOR problem...
Epoch    1/1000, Loss: 0.249842
Epoch  100/1000, Loss: 0.219619
Epoch  200/1000, Loss: 0.055226
Epoch  300/1000, Loss: 0.014176
Epoch  400/1000, Loss: 0.006978
Epoch  500/1000, Loss: 0.004460
Epoch  600/1000, Loss: 0.003217
Epoch  700/1000, Loss: 0.002481
Epoch  800/1000, Loss: 0.001689
Epoch  900/1000, Loss: 0.001444
Epoch 1000/1000, Loss: 0.001444
Training finished.

--- After Training ---
Input  | True Output | Predicted Output | Rounded Prediction
[0. 0.] |           0 |           0.0471 |                  0
[0. 1.] |           1 |           0.9742 |                  1
[1. 0.] |           1 |           0.9743 |                  1
[1. 1.] |           0 |           0.0471 |                  0