GPflow2:具有部分共享内核的多输出内核(MOK)

Question

我有一个关于gpflow 2中的多输出内核的问题，对于我正在开发的应用程序，我想创建一个独立的多输出内核，它可以在某些输出维度上共享内核，但不是全部。GPflow中的两个相关类是SharedIndependent和SeparateIndependent多输出内核类。但是，对于所有P输出维度，它们都使用一个共享内核，对于P输出维度使用P单独的内核。

下面所附的代码是来自多输出内核笔记本(https://gpflow.readthedocs.io/en/master/notebooks/advanced/multioutput.html)的一个小的适应。这4个输出维度可分为2组(每个输出维度2个)：一组具有高灵活性，另一组具有低灵活性。我想在这个任务中使用2个内核来检测平方指数核的两个不同的长度尺度参数。为了完整起见，我从示例笔记本中为SharedIndependent类和SeparateIndependent类添加了实现。

目前，我已经尝试将它们合并如下：SeparateIndependent([SharedIndependent(SquaredExponential()+Linear(), output_dim=2) for _ in range(2)])。这将导致以下错误：(15是诱导点数)

ValueError:维度必须相等，但对于输入形状为2、15、2、15、2、1、15、15的“{节点add_8}} =AddV2T=DT_DOUBLE”，则为2和15。

为此，我还尝试创建一个新的多输出内核类，它在很大程度上模仿了SeparateIndependent类，并对K(X, X2)和K_diag(X)函数中的列表理解循环进行了一些修改。这没有成功，因为不知道如何为多输出内核类注册正确的Kuu()、Kuf()和conditional()函数。尝试创建与只有不同名称的SeparateIndependent类相同的类也失败。

我很高兴知道是否可以组合SharedIndependent和SeparateIndependent类，或者构建一个新的MOK类是解决这个问题的一个更好的解决方案。如果是的话，解决注册条件表达式(Kuu()、Kuf()和conditional())问题的最佳方法是什么？

import numpy as np
import matplotlib.pyplot as plt
import tensorflow as tf
import gpflow
from gpflow.kernels import SquaredExponential, Linear, SharedIndependent, SeparateIndependent
from gpflow.inducing_variables import SharedIndependentInducingVariables, InducingPoints
from gpflow.utilities import print_summary
from gpflow.ci_utils import ci_niter
gpflow.config.set_default_float(np.float64)
np.random.seed(0)
MAXITER = ci_niter(2000)

N = 100  # number of points
D = 1  # number of input dimensions
M = 15  # number of inducing points
L = P = 4  # number of latent GPs, number of observations = output dimension

def generate_data(N=100):
    X = np.random.rand(N)[:, None] * 10 - 5  # Inputs = N x D
    G = np.hstack((0.5 * np.sin(X/2) + X, 3.0 * np.cos(X/2) - X,0.5 * np.sin(4 * X) + X, 3.0 * np.cos(4*X) - X))  # G = N x L
    W = np.array([[0.5, -0.3, 0, 0], [0.5, -0.3, 0, 0], [0, 0, -0.4, 0.6],[0.0, 0.0, 0.6, -0.4]])  # L x P
    F = np.matmul(G, W)  # N x P
    Y = F + np.random.randn(*F.shape) * [0.2, 0.2, 0.2, 0.2]

    return X, Y
X, Y = data = generate_data(N)
print(X.shape, Y.shape)
Zinit = np.linspace(-5, 5, M)[:, None]

def plot_model(m, name, lower=-7.0, upper=7.0):
    pX = np.linspace(lower, upper, 100)[:, None]
    pY, pYv = m.predict_y(pX)
    if pY.ndim == 3:
        pY = pY[:, 0, :]
    plt.plot(X, Y, "x")
    plt.gca().set_prop_cycle(None)
    plt.plot(pX, pY)
    for i in range(pY.shape[1]):
        top = pY[:, i] + 2.0 * pYv[:, i] ** 0.5
        bot = pY[:, i] - 2.0 * pYv[:, i] ** 0.5
        plt.fill_between(pX[:, 0], top, bot, alpha=0.3)
    plt.xlabel("X")
    plt.ylabel("f")
    plt.title(f"{name} kernel.")
    plt.show()

# initialization of inducing input locations (M random points from the training inputs)
Z = Zinit.copy()
# create multi-output inducing variables from Z
iv = SharedIndependentInducingVariables(InducingPoints(Z))

def optimize_model_with_scipy(model):
    optimizer = gpflow.optimizers.Scipy()
    optimizer.minimize(
        model.training_loss_closure(data),
        variables=model.trainable_variables,
        method="l-bfgs-b",
        options={"disp": True, "maxiter": MAXITER},
    )


# create multi-output kernel
kernels = [
        (SeparateIndependent([SquaredExponential() + Linear() for _ in range(P)]),'Seperate Independent'),
        (SharedIndependent(SquaredExponential()+Linear(), output_dim=P), 'Shared Independent'),
        (SeparateIndependent([SharedIndependent(SquaredExponential()+Linear(), output_dim=2) for _ in range(2)]), 'Partially shared independent')
       ]
for (kernel, name) in kernels:
    m = gpflow.models.SVGP(kernel, gpflow.likelihoods.Gaussian(), inducing_variable=iv, num_latent_gps=P)
    print_summary(m)
    optimize_model_with_scipy(m)
    print_summary(m)
    plot_model(m, name)

Answer 1

更新:我已经通过创建一个单独的多输出内核类来解决这个问题，这个类包含了D共享独立的多输出内核。条件表达式遍历这些表达式，将它们应用于适当的输出维度。请注意，这仅为共享独立诱导变量实现。

class CustomMultiOutput(MultioutputKernel, Combination):    
    def __init__(self, kernels, name=None):
        kernels = [SharedIndependent(k, output_dim=nu) for k in kernels]
        super().__init__(kernels=kernels, name=name)

    @property
    def num_latent_gps(self):
        return len(self.kernels)

    @property
    def latent_kernels(self):
        """The underlying kernels in the multioutput kernel"""
        return tuple(self.kernels)

@conditional.register(object, SharedIndependentInducingVariables, CustomMultiOutput, object)
def custom_shared_independent_conditional(
    Xnew,inducing_variable,
    kernel, f,*,
    full_cov=False, full_output_cov=False, q_sqrt=None, white=False, ):
    N, P = Xnew.shape[0], q_sqrt.shape[0]
    nu = kernel.nu
    fmeans, fvars = [], []

    for i, k in enumerate(kernel.kernels):
        nu_idx_start, nu_idx_end = int(i * kernel.nu), int((i + 1) * kernel.nu)
        f_i = f[:, nu_idx_start:nu_idx_end]
        q_sqrt_i = q_sqrt[nu_idx_start:nu_idx_end, :]
        Kmm = covariances.Kuu(inducing_variable, k, jitter=default_jitter())  # [M, M]
        Kmn = covariances.Kuf(inducing_variable, k, Xnew)  # [M, N]
        Knn = k.kernel(Xnew, full_cov=full_cov)

        fmean, fvar = base_conditional(
            Kmn, Kmm, Knn, f_i, full_cov=full_cov, q_sqrt=q_sqrt_i, white=white
        )  # [N, P],  [P, N, N] or [N, P]
        fmeans.append(tf.transpose(fmean,perm=[1,0]))
        if full_cov:
            fvars.append(tf.transpose(fvar,perm=[0,2,1]))
        else:
            fvars.append(tf.transpose(fvar,perm=[1,0]))

    fmeans = tf.transpose(tf.reshape(fmeans,shape=(-1,N)))
    if full_cov:
        fvars = tf.reshape(fvars,shape=(-1,N,N))
    else:
        fvars = tf.transpose(tf.reshape(fvars, shape=(-1,N)))

    return fmeans, expand_independent_outputs(fvars, full_cov, full_output_cov)

Answer 2

通过重用内核对象(没有任何自定义类)，您应该能够绑定一些(但不是所有的)内核参数：

k_group1 = SquaredExponential() + Linear()
k_group2 = SquaredExponential() + Linear()
kernel_list = [k_group1, k_group1, k_group2, k_group2]
# or if you have many more classes: [k_group1 for _ in range(num_group1)] + [k_group2 for _ in range(num_group2)]
mo_kernel = SeparateIndependent(kernel_list)