# Copyright 2016-2020 The GPflow Contributors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import Optional
import tensorflow as tf
import gpflow
from .. import posteriors
from ..base import InputData, MeanAndVariance, RegressionData
from ..kernels import Kernel
from ..logdensities import multivariate_normal
from ..mean_functions import MeanFunction
from ..utilities.model_utils import add_noise_cov
from .model import GPModel
from .training_mixins import InternalDataTrainingLossMixin
from .util import data_input_to_tensor
[docs]class GPR_deprecated(GPModel, InternalDataTrainingLossMixin):
r"""
Gaussian Process Regression.
This is a vanilla implementation of GP regression with a Gaussian
likelihood. Multiple columns of Y are treated independently.
The log likelihood of this model is given by
.. math::
\log p(Y \,|\, \mathbf f) =
\mathcal N(Y \,|\, 0, \sigma_n^2 \mathbf{I})
To train the model, we maximise the log _marginal_ likelihood
w.r.t. the likelihood variance and kernel hyperparameters theta.
The marginal likelihood is found by integrating the likelihood
over the prior, and has the form
.. math::
\log p(Y \,|\, \sigma_n, \theta) =
\mathcal N(Y \,|\, 0, \mathbf{K} + \sigma_n^2 \mathbf{I})
"""
def __init__(
self,
data: RegressionData,
kernel: Kernel,
mean_function: Optional[MeanFunction] = None,
noise_variance: float = 1.0,
):
likelihood = gpflow.likelihoods.Gaussian(noise_variance)
_, Y_data = data
super().__init__(kernel, likelihood, mean_function, num_latent_gps=Y_data.shape[-1])
self.data = data_input_to_tensor(data)
# type-ignore is because of changed method signature:
[docs] def maximum_log_likelihood_objective(self) -> tf.Tensor: # type: ignore
return self.log_marginal_likelihood()
def _add_noise_cov(self, K: tf.Tensor) -> tf.Tensor:
"""
Returns K + σ² I, where σ² is the likelihood noise variance (scalar),
and I is the corresponding identity matrix.
"""
return add_noise_cov(K, self.likelihood.variance)
[docs] def log_marginal_likelihood(self) -> tf.Tensor:
r"""
Computes the log marginal likelihood.
.. math::
\log p(Y | \theta).
"""
X, Y = self.data
K = self.kernel(X)
ks = self._add_noise_cov(K)
L = tf.linalg.cholesky(ks)
m = self.mean_function(X)
# [R,] log-likelihoods for each independent dimension of Y
log_prob = multivariate_normal(Y, m, L)
return tf.reduce_sum(log_prob)
[docs] def predict_f(
self, Xnew: InputData, full_cov: bool = False, full_output_cov: bool = False
) -> MeanAndVariance:
r"""
This method computes predictions at X \in R^{N \x D} input points
.. math::
p(F* | Y)
where F* are points on the GP at new data points, Y are noisy observations at training data
points.
"""
X, Y = self.data
err = Y - self.mean_function(X)
kmm = self.kernel(X)
knn = self.kernel(Xnew, full_cov=full_cov)
kmn = self.kernel(X, Xnew)
kmm_plus_s = self._add_noise_cov(kmm)
conditional = gpflow.conditionals.base_conditional
f_mean_zero, f_var = conditional(
kmn, kmm_plus_s, knn, err, full_cov=full_cov, white=False
) # [N, P], [N, P] or [P, N, N]
f_mean = f_mean_zero + self.mean_function(Xnew)
return f_mean, f_var
[docs]class GPR_with_posterior(GPR_deprecated):
"""
This is an implementation of GPR that provides a posterior() method that
enables caching for faster subsequent predictions.
"""
[docs] def posterior(
self,
precompute_cache: posteriors.PrecomputeCacheType = posteriors.PrecomputeCacheType.TENSOR,
) -> posteriors.GPRPosterior:
"""
Create the Posterior object which contains precomputed matrices for
faster prediction.
precompute_cache has three settings:
- `PrecomputeCacheType.TENSOR` (or `"tensor"`): Precomputes the cached
quantities and stores them as tensors (which allows differentiating
through the prediction). This is the default.
- `PrecomputeCacheType.VARIABLE` (or `"variable"`): Precomputes the cached
quantities and stores them as variables, which allows for updating
their values without changing the compute graph (relevant for AOT
compilation).
- `PrecomputeCacheType.NOCACHE` (or `"nocache"` or `None`): Avoids
immediate cache computation. This is useful for avoiding extraneous
computations when you only want to call the posterior's
`fused_predict_f` method.
"""
return posteriors.GPRPosterior(
kernel=self.kernel,
data=self.data,
likelihood_variance=self.likelihood.variance,
mean_function=self.mean_function,
precompute_cache=precompute_cache,
)
[docs] def predict_f(
self, Xnew: InputData, full_cov: bool = False, full_output_cov: bool = False
) -> MeanAndVariance:
"""
For backwards compatibility, GPR's predict_f uses the fused (no-cache)
computation, which is more efficient during training.
For faster (cached) prediction, predict directly from the posterior object, i.e.,:
model.posterior().predict_f(Xnew, ...)
"""
return self.posterior(posteriors.PrecomputeCacheType.NOCACHE).fused_predict_f(
Xnew, full_cov=full_cov, full_output_cov=full_output_cov
)
[docs]class GPR(GPR_with_posterior):
# subclassed to ensure __class__ == "GPR"
pass