GPflow

sgptools.utils.gpflow

TraceInducingPts

Bases: MonitorTask

GPflow monitoring task, used to trace the inducing points states at every step during optimization.

Parameters:

Name	Type	Description	Default
model	sgpr	GPflow GP/SGP model	required

Source code in sgptools/utils/gpflow.py

class TraceInducingPts(gpflow.monitor.MonitorTask):
    '''
    GPflow monitoring task, used to trace the inducing points
    states at every step during optimization. 

    Args:
        model (gpflow.models.sgpr): GPflow GP/SGP model
    '''
    def __init__(self, model):
        super().__init__()
        self.trace = []
        self.model = model

    def run(self, **kwargs):
        '''
        Method used to extract the inducing points and 
        apply IPP fixed points transform if available
        '''
        Xu = self.model.inducing_variable.Z
        Xu_exp = self.model.transform.expand(Xu, 
                            expand_sensor_model=False).numpy()
        self.trace.append(Xu_exp)

    def get_trace(self):
        '''
        Returns the inducing points collected at each optimization step

        Returns:
            trace (ndarray): (n, m, d); Array with the inducing points.
                            `n` is the number of optimization steps;
                            `m` is the number of inducing points;
                            `d` is the dimension of the inducing points.
        '''
        return np.array(self.trace)

get_trace()

Returns the inducing points collected at each optimization step

Returns:

Name	Type	Description
trace	ndarray	(n, m, d); Array with the inducing points. n is the number of optimization steps; m is the number of inducing points; d is the dimension of the inducing points.

Source code in sgptools/utils/gpflow.py

def get_trace(self):
    '''
    Returns the inducing points collected at each optimization step

    Returns:
        trace (ndarray): (n, m, d); Array with the inducing points.
                        `n` is the number of optimization steps;
                        `m` is the number of inducing points;
                        `d` is the dimension of the inducing points.
    '''
    return np.array(self.trace)

run(**kwargs)

Method used to extract the inducing points and apply IPP fixed points transform if available

Source code in sgptools/utils/gpflow.py

def run(self, **kwargs):
    '''
    Method used to extract the inducing points and 
    apply IPP fixed points transform if available
    '''
    Xu = self.model.inducing_variable.Z
    Xu_exp = self.model.transform.expand(Xu, 
                        expand_sensor_model=False).numpy()
    self.trace.append(Xu_exp)

get_model_params(X_train, y_train, max_steps=1500, lr=0.01, print_params=True, lengthscales=1.0, variance=1.0, noise_variance=0.1, kernel=None, **kwargs)

Train a GP on the given training set

Parameters:

Name	Type	Description	Default
X_train	ndarray	(n, d); Training set inputs	required
y_train	ndarray	(n, 1); Training set labels	required
max_steps	int	Maximum number of optimization steps	1500
lr	float	Optimization learning rate	0.01
print_params	bool	If True, prints the optimized GP parameters	True
lengthscales	float or list	Kernel lengthscale(s), if passed as a list, each element corresponds to each data dimension	1.0
variance	float	Kernel variance	1.0
noise_variance	float	Data noise variance	0.1
kernel	Kernel	gpflow kernel function	None

Returns:

Name	Type	Description
loss	list	Loss values obtained during training
variance	float	Optimized data noise variance
kernel	Kernel	Optimized gpflow kernel function

Source code in sgptools/utils/gpflow.py

def get_model_params(X_train, y_train, 
                     max_steps=1500, 
                     lr=1e-2, 
                     print_params=True, 
                     lengthscales=1.0, 
                     variance=1.0, 
                     noise_variance=0.1,
                     kernel=None,
                     **kwargs):
    """Train a GP on the given training set

    Args:
        X_train (ndarray): (n, d); Training set inputs
        y_train (ndarray): (n, 1); Training set labels
        max_steps (int): Maximum number of optimization steps
        lr (float): Optimization learning rate
        print_params (bool): If True, prints the optimized GP parameters
        lengthscales (float or list): Kernel lengthscale(s), if passed as a list, 
                                each element corresponds to each data dimension
        variance (float): Kernel variance
        noise_variance (float): Data noise variance
        kernel (gpflow.kernels.Kernel): gpflow kernel function

    Returns:
        loss (list): Loss values obtained during training
        variance (float): Optimized data noise variance
        kernel (gpflow.kernels.Kernel): Optimized gpflow kernel function
    """
    if kernel is None:
        kernel = gpflow.kernels.SquaredExponential(lengthscales=lengthscales, 
                                                   variance=variance)

    gpr_gt = gpflow.models.GPR(data=(X_train, y_train), 
                               kernel=kernel,
                               noise_variance=noise_variance)

    if max_steps > 0:
        loss = optimize_model(gpr_gt, max_steps=max_steps, lr=lr, **kwargs)
    else:
        loss = 0

    if print_params:
        print_summary(gpr_gt)

    return loss, gpr_gt.likelihood.variance, kernel

optimize_model(model, max_steps=2000, kernel_grad=True, lr=0.01, optimizer='scipy', method=None, verbose=False, trace_fn=None, convergence_criterion=True, trainable_variables=None, tol=None)

Trains a GP/SGP model

Parameters:

Name	Type	Description	Default
model	models	GPflow GP/SGP model to train.	required
max_steps	int	Maximum number of training steps.	2000
kernel_grad	bool	If False, the kernel parameters will not be optimized. Ignored when trainable_variables are passed.	True
lr	float	Optimization learning rate.	0.01
optimizer	str	Optimizer to use for training (scipy or tf).	'scipy'
method	str	Optimization method refer to scipy minimize and tf optimizers for full list	None
verbose	bool	If True, the training progress will be printed when using Scipy.	False
trace_fn	str	Function to trace metrics during training. If None, the loss values are returned; If traceXu, it the inducing points states at each optimization step are returned (increases computation time).	None
convergence_criterion	bool	If True and using a tensorflow optimizer, it enables early stopping when the loss plateaus.	True
trainable_variables	list	List of model variables to train.	None
tol	float	Convergence tolerance to decide when to stop optimization.	None

Source code in sgptools/utils/gpflow.py

def optimize_model(model, 
                   max_steps=2000, 
                   kernel_grad=True, 
                   lr=1e-2, 
                   optimizer='scipy', 
                   method=None,
                   verbose=False, 
                   trace_fn=None,
                   convergence_criterion=True,
                   trainable_variables=None,
                   tol=None):
    """
    Trains a GP/SGP model

    Args:
        model (gpflow.models): GPflow GP/SGP model to train.
        max_steps (int): Maximum number of training steps.
        kernel_grad (bool): If `False`, the kernel parameters will not be optimized. 
                            Ignored when `trainable_variables` are passed.
        lr (float): Optimization learning rate.
        optimizer (str): Optimizer to use for training (`scipy` or `tf`).
        method (str): Optimization method refer to [scipy minimize](https://docs.scipy.org/doc/scipy/reference/generated/scipy.optimize.minimize.html#scipy.optimize.minimize) 
                      and [tf optimizers](https://www.tensorflow.org/api_docs/python/tf/keras/optimizers) for full list
        verbose (bool): If `True`, the training progress will be printed when using Scipy.
        trace_fn (str): Function to trace metrics during training. 
                        If `None`, the loss values are returned;
                        If `traceXu`, it the inducing points states at each optimization step are returned (increases computation time).
        convergence_criterion (bool): If `True` and using a tensorflow optimizer, it 
                                      enables early stopping when the loss plateaus.
        trainable_variables (list): List of model variables to train.
        tol (float): Convergence tolerance to decide when to stop optimization.
    """
    # Train all variables if trainable_variables are not provided
    # If kernel_gradient is False, disable the kernel parameter gradient updates
    if trainable_variables is None and kernel_grad:
        trainable_variables=model.trainable_variables
    elif trainable_variables is None and not kernel_grad:
        trainable_variables=model.trainable_variables[:1]

    if optimizer == 'scipy':
        if method is None:
            method = 'L-BFGS-B'

        if trace_fn == 'traceXu':
            execute_task = TraceInducingPts(model)
            task_group = gpflow.monitor.MonitorTaskGroup(execute_task, 
                                                         period=1)
            trace_fn = gpflow.monitor.Monitor(task_group)

        opt = gpflow.optimizers.Scipy()
        losses = opt.minimize(model.training_loss,
                              trainable_variables,
                              method=method,
                              options=dict(disp=verbose, maxiter=max_steps),
                              tol=tol,
                              step_callback=trace_fn)
        if trace_fn is None:
            losses = losses.fun
        else:
            losses = trace_fn.task_groups[0].tasks[0].get_trace()
    else:
        if trace_fn is None:
            trace_fn = lambda x: x.loss
        elif trace_fn == 'traceXu':
            def trace_fn(traceable_quantities):
                return model.inducing_variable.Z.numpy()

        if method is None:
            method = 'adam'
        opt = tf.keras.optimizers.get(method)
        opt.learning_rate = lr
        loss_fn = model.training_loss
        if convergence_criterion:
            convergence_criterion = tfp.optimizer.convergence_criteria.LossNotDecreasing(
                                            atol=1e-5, 
                                            window_size=50,
                                            min_num_steps=int(max_steps*0.1))
        else:
            convergence_criterion = None
        losses = tfp.math.minimize(loss_fn,
                                   trainable_variables=trainable_variables,
                                   num_steps=max_steps,
                                   optimizer=opt,
                                   convergence_criterion=convergence_criterion,
                                   trace_fn=trace_fn)
        losses = losses.numpy()

    return losses

plot_loss(losses, save_file=None)

Helper function to plot the training loss

Parameters:

Name	Type	Description	Default
losses	list	list of loss values	required
save_file	str	If passed, the loss plot will be saved to the save_file	None

Source code in sgptools/utils/gpflow.py

def plot_loss(losses, save_file=None):
    """Helper function to plot the training loss

    Args:
        losses (list): list of loss values
        save_file (str): If passed, the loss plot will be saved to the `save_file`
    """
    plt.plot(losses)
    plt.title('Log Likelihood')
    plt.xlabel('Iteration')
    plt.ylabel('Log Likelihood')
    ax = plt.gca()
    ax.ticklabel_format(useOffset=False)

    if save_file is not None:
        plt.savefig(save_file, bbox_inches='tight')
        plt.close()
    else:
        plt.show()