Skip to content

Models

BaseModel 

BaseModel(device_name)

Bases: ABC

Interface of a probabilistic model.

Initializes a model with the specified device name.

Parameters:

Name Type Description Default
device_name str

The name of the PyTorch device to be used for computations.

 required
Source code in pypolo/models/base_model.py 
12
13
14
15
16
17
18
19
20
21
22
@abstractmethod
def __init__(self, device_name: str) -> None:
    r"""Initializes a model with the specified device name.

    Args:
        device_name (str): The name of the PyTorch device to be used for
            computations.

    """
    super().__init__()
    self.dtype, self.device = torch_utils.get_dtype_and_device(device_name)

learn abstractmethod 

learn(x_new, y_new, num_iter, verbose=True)

Optimizes the model parameters.

Parameters:

Name Type Description Default
x_new ndarray

New training inputs of shape (num_inputs, dim_inputs).

 required
y_new ndarray

New training outputs of shape (num_outputs, dim_outputs).

 required
num_iter int

Number of optimization/training iterations.

 required
verbose bool

Print the optimization information or not?

 True
writer Union[SummaryWriter, None]

Tensorboard writer.

 required

Raises:

Type Description
NotImplementedError

This is an abstract method and must be implemented by derived classes.
Source code in pypolo/models/base_model.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@abstractmethod
def learn(self,
          x_new: np.ndarray,
          y_new: np.ndarray,
          num_iter: int,
          verbose: bool = True) -> None:
    r"""Optimizes the model parameters.

    Args:
        x_new (np.ndarray): New training inputs of shape
            (num_inputs, dim_inputs).
        y_new (np.ndarray): New training outputs of shape
            (num_outputs, dim_outputs).
        num_iter (int): Number of optimization/training iterations.
        verbose (bool): Print the optimization information or not?
        writer (Union[SummaryWriter, None]): Tensorboard writer.

    Raises:
        NotImplementedError: This is an abstract method and must be
            implemented by derived classes.
    """
    raise NotImplementedError

predict abstractmethod 

predict(x_test)

Makes predictions.

Parameters:

Name Type Description Default
x_test ndarray

Test inputs of shape (num_inputs, dim_inputs).

 required

Returns:

Type Description
Tuple[ndarray, ndarray]

Tuple[np.ndarray, np.ndarray]: A tuple containing predictive mean and predictive standard deviation of shape (num_inputs, 1).

Raises:

Type Description
NotImplementedError

This is an abstract method and must be implemented by derived classes.
Source code in pypolo/models/base_model.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@abstractmethod
def predict(self, x_test: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
    """Makes predictions.

    Args:
        x_test (np.ndarray): Test inputs of shape (num_inputs, dim_inputs).

    Returns:
        Tuple[np.ndarray, np.ndarray]: A tuple containing predictive mean
            and predictive standard deviation of shape (num_inputs, 1).

    Raises:
        NotImplementedError: This is an abstract method and must be
            implemented by derived classes.
    """
    raise NotImplementedError

GPRModel 

GPRModel(device_name, kernel, noise, lr_hyper=0.01, jitter=1e-06)

Bases: BaseModel, Module

Gaussian Process Regression.

Parameters:

Name Type Description Default
device_name str

The name of the device to run the model.

 required
kernel BaseKernel

The kernel function.

 required
noise float

The noise variance of the Gaussian likelihood.

 required
lr_hyper float

Learning rate of hyper-parameters.

 0.01
jitter float

The jitter to add to the diagonal of the covariance matrix. Defaults to 1e-6.

 1e-06
What is jitter and why is it necessary?

Jitter is a small positive number added to the diagonal of the covariance matrix to ensure that it is positive definite. This is necessary because the covariance matrix is not always positive definite due to numerical errors.

Source code in pypolo/models/gpr_model.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def __init__(self,
             device_name,
             kernel: BaseKernel,
             noise: float,
             lr_hyper: float = 0.01,
             jitter: float = 1e-6) -> None:
    r"""Gaussian Process Regression.

    Args:
        device_name (str): The name of the device to run the model.
        kernel (BaseKernel): The kernel function.
        noise (float): The noise variance of the Gaussian likelihood.
        lr_hyper (float, optional): Learning rate of hyper-parameters.
        jitter (float, optional): The jitter to add to the diagonal of the
            covariance matrix. Defaults to 1e-6.

    ??? note "What is jitter and why is it necessary?"

        Jitter is a small positive number added to the diagonal of the
        covariance matrix to ensure that it is positive definite.
        This is necessary because the covariance matrix is not always
        positive definite due to numerical errors.

    """
    BaseModel.__init__(self, device_name)
    nn.Module.__init__(self)
    self.kernel = kernel
    self._free_noise = Parameter(
        torch_utils.inv_softplus(
            torch.as_tensor(
                noise,
                dtype=self.dtype,
                device=self.device,
            )))
    self._init_optimizers(lr_hyper)
    self.jitter = jitter

noise property writable 

noise

The noise variance hyper-parameter.

Returns:

Type Description
Tensor

torch.Tensor: The noise variance of the Gaussian likelihood.
Positivity

The positivity of noise variance is ensured by a softplus function.

forward 

forward(x_test, noise_free=False)

Make prediction.

Parameters:

Name Type Description Default
x_test Tensor

Test inputs of shape (num_inputs, dim_inputs).

 required
noise_free bool

If True, predict the latent function values. Otherwise, predict the noisy targets.

 False

Returns:

Name Type Description
mean Tensor

Predictive mean of shape (num_inputs, 1)
std Tensor

Predictive standard deviation of shape (num_inputs, 1).
Difference between forward and predict

The forward method uses PyTorch tensors as inputs and outputs. The predict method uses NumPy arrays as inputs and outputs. Users should use predict method for model prediction.

Source code in pypolo/models/gpr_model.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def forward(
    self,
    x_test: torch.Tensor,
    noise_free: bool = False,
) -> Tuple[torch.Tensor, torch.Tensor]:
    r"""Make prediction.

    Args:
        x_test (torch.Tensor): Test inputs of shape
            (num_inputs, dim_inputs).
        noise_free (bool, optional): If True, predict the latent function
            values. Otherwise, predict the noisy targets.

    Returns:
        mean (torch.Tensor): Predictive mean of shape (num_inputs, 1)
        std (torch.Tensor): Predictive standard deviation of shape
            (num_inputs, 1).

    ??? note "Difference between `forward` and `predict`"

        The `forward` method uses PyTorch tensors as inputs and outputs.
        The `predict` method uses NumPy arrays as inputs and outputs.
        Users should use `predict` method for model prediction.

    """
    with torch.no_grad():
        L, iK_y = self._compute_common()
        Ksn = self.kernel(x_test, self.x_train)
        Kss_diag = self.kernel.diag(x_test)
        iL_Kns = torch.linalg.solve_triangular(L, Ksn.t(), upper=False)
        mean = Ksn @ iK_y
        var = Kss_diag - iL_Kns.square().sum(0).view(-1, 1)
        # Variance might be zero when lengthscale is too large.
        if torch.any(var <= 0.0):
            print(var.ravel().numpy())
            raise ValueError("Predictive variance <= 0.0!")
        var.clamp_(min=self.jitter)
        if not noise_free:
            var += self.noise
        std = var.sqrt()
    return mean, std

learn 

learn(x_new, y_new, num_iter, verbose=True)

Optimizes the model parameters.

Parameters:

Name Type Description Default
x_new ndarray

New training inputs of shape (num_inputs, dim_inputs).

 required
y_new ndarray

New training outputs of shape (num_outputs, dim_outputs).

 required
num_iter int

Number of optimization/training iterations.

 required
verbose bool

Print the optimization information or not?

 True
writer Union[SummaryWriter, None]

Tensorboard writer.

 required
Source code in pypolo/models/gpr_model.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def learn(self,
          x_new: np.ndarray,
          y_new: np.ndarray,
          num_iter: int,
          verbose: bool = True) -> None:
    r"""Optimizes the model parameters.

    Args:
        x_new (np.ndarray): New training inputs of shape
            (num_inputs, dim_inputs).
        y_new (np.ndarray): New training outputs of shape
            (num_outputs, dim_outputs).
        num_iter (int): Number of optimization/training iterations.
        verbose (bool): Print the optimization information or not?
        writer (Union[SummaryWriter, None]): Tensorboard writer.

    """
    self._add_data(x_new, y_new)
    self.train()
    progress_bar = tqdm(range(num_iter), disable=not verbose)
    for i in progress_bar:
        self.opt_hyper.zero_grad()
        loss = self._compute_loss()
        loss.backward()
        self.opt_hyper.step()
        progress_bar.set_description(
            f"Iter: {i:02d} loss: {loss.item(): .2f}")
    self.eval()

predict 

predict(x_test)

Makes predictions.

Parameters:

Name Type Description Default
x_test ndarray

Test inputs of shape (num_inputs, dim_inputs).

 required

Returns:

Type Description
Tuple[ndarray, ndarray]

Tuple[np.ndarray, np.ndarray]: A tuple containing predictive mean and predictive standard deviation of shape (num_inputs, 1).
Source code in pypolo/models/gpr_model.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def predict(self, x_test: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
    """Makes predictions.

    Args:
        x_test (np.ndarray): Test inputs of shape (num_inputs, dim_inputs).

    Returns:
        Tuple[np.ndarray, np.ndarray]: A tuple containing predictive mean
            and predictive standard deviation of shape (num_inputs, 1).

    """
    x_test_tensor = torch.as_tensor(x_test,
                                    dtype=self.dtype,
                                    device=self.device)
    mean_tensor, std_tensor = self.forward(x_test_tensor)
    return mean_tensor.cpu().numpy(), std_tensor.cpu().numpy()