Enable unobserved task support in MultiTaskGP

Summary:
Permits an MTGP to predict on an unobserved task, addressing these issues:
meta-pytorch#2360
meta-pytorch#3085

To do this, we assume that the unobserved task is maximally correlated with the target tasks (equally with each, by averaging the elements). Exact heuristic on correlation is definitely up for discussion, but this seems like a decent default assumption.

Will come in handy for TL initialization.

Differential Revision:
D90769576

D90769576

Author: Carl Hvarfner

botorch/models/fully_bayesian_multitask.py

@@ -7,7 +7,7 @@
 r"""Multi-task Gaussian Process Regression models with fully Bayesian inference."""
 
 from collections.abc import Mapping
-from typing import Any, NoReturn, TypeVar
+from typing import Any, NoReturn, Self, TypeVar
 
 import pyro
 import torch
@@ -19,7 +19,10 @@
     reshape_and_detach,
     SaasPyroModel,
 )
-from botorch.models.gpytorch import BatchedMultiOutputGPyTorchModel
+from botorch.models.gpytorch import (
+    BatchedMultiOutputGPyTorchModel,
+    MultiTaskGPyTorchModel,
+)
 from botorch.models.multitask import MultiTaskGP
 from botorch.models.transforms.input import InputTransform
 from botorch.models.transforms.outcome import OutcomeTransform
@@ -55,6 +58,7 @@ def set_inputs(
         train_Yvar: Tensor | None,
         task_feature: int,
         task_rank: int | None = None,
+        all_tasks: list[int] | None = None,
     ) -> None:
         """Set the training data.
 
@@ -73,7 +77,11 @@ def set_inputs(
         task_feature = task_feature % train_X.shape[-1]
         super().set_inputs(train_X, train_Y, train_Yvar)
         # obtain a list of task indicies
-        all_tasks = train_X[:, task_feature].unique().to(dtype=torch.long).tolist()
+        all_tasks = (
+            train_X[:, task_feature].unique().to(dtype=torch.long).tolist()
+            if all_tasks is None
+            else all_tasks
+        )
         self.task_feature = task_feature
         self.num_tasks = len(all_tasks)
         self.task_rank = task_rank or self.num_tasks
@@ -242,7 +250,10 @@ def __init__(
                 outputs for. If omitted, return outputs for all task indices.
             rank: The num of learned task embeddings to be used in the task kernel.
                 If omitted, use a full rank (i.e. number of tasks) kernel.
-            all_tasks: NOT SUPPORTED!
+            all_tasks: A list of all task indices. If omitted, all tasks will be
+                inferred from the task feature column of the training data. Used to
+                inform the model about the total number of tasks, including any
+                unobserved tasks.
             outcome_transform: An outcome transform that is applied to the
                 training data during instantiation and to the posterior during
                 inference (that is, the ``Posterior`` obtained by calling
@@ -310,6 +321,7 @@ def __init__(
             train_Yvar=train_Yvar,
             task_feature=task_feature,
             task_rank=self._rank,
+            all_tasks=all_tasks,
         )
         self.pyro_model: MultitaskSaasPyroModel = pyro_model
         if outcome_transform is not None:
@@ -383,6 +395,20 @@ def load_mcmc_samples(self, mcmc_samples: dict[str, Tensor]) -> None:
             _,
         ) = self.pyro_model.load_mcmc_samples(mcmc_samples=mcmc_samples)
 
+    def eval(self) -> Self:
+        r"""Puts the model in eval mode.
+
+        Circumvents the need to call MultiTaskGP.eval(), which computes the
+        task_covar_matrix for non-observed tasks. This is not needed for fully
+        Bayesian models, since the non-observed tasks' covar factors are instead
+        sampled.
+
+        Returns:
+            The model itself.
+        """
+        self._check_if_fitted()
+        return MultiTaskGPyTorchModel.eval(self)
+
     def posterior(
         self,
         X: Tensor,

botorch/models/multitask.py

@@ -30,7 +30,7 @@
 from __future__ import annotations
 
 import math
-from typing import Any
+from typing import Any, Self
 
 import torch
 from botorch.acquisition.objective import PosteriorTransform
@@ -238,7 +238,11 @@ def __init__(
                 "This is not allowed as it will lead to errors during model training."
             )
         all_tasks = all_tasks or all_tasks_inferred
-        self.num_tasks = len(all_tasks_inferred)
+        sorted_all_tasks = sorted(all_tasks)
+        self.num_tasks = len(all_tasks)
+        # Store for later buffer registration (after super().__init__)
+        self._all_tasks_inferred = all_tasks_inferred
+        self._sorted_all_tasks = sorted_all_tasks
         if outcome_transform == DEFAULT:
             outcome_transform = Standardize(m=1, batch_shape=train_X.shape[:-2])
         if outcome_transform is not None:
@@ -321,14 +325,55 @@ def __init__(
             default_task_value=None if output_tasks is None else output_tasks[0],
         )
         self.register_buffer("_task_mapper", task_mapper)
-        self._expected_task_values = set(all_tasks_inferred)
+        self._expected_task_values = set(all_tasks)
         if input_transform is not None:
             self.input_transform = input_transform
         if outcome_transform is not None:
             self.outcome_transform = outcome_transform
         self._validate_task_values = validate_task_values
         self.to(train_X)
 
+        # Register observed/unobserved task indices as buffers (must be after
+        # super().__init__()). Compute observed and unobserved task indices when
+        # all_tasks includes unobserved tasks
+        if set(self._sorted_all_tasks) != set(self._all_tasks_inferred):
+            observed_set = set(self._all_tasks_inferred)
+            self.register_buffer(
+                "_observed_task_indices",
+                torch.tensor(
+                    [
+                        i
+                        for i, t in enumerate(self._sorted_all_tasks)
+                        if t in observed_set
+                    ],
+                    dtype=torch.long,
+                ),
+            )
+            self.register_buffer(
+                "_unobserved_task_indices",
+                torch.tensor(
+                    [
+                        i
+                        for i, t in enumerate(self._sorted_all_tasks)
+                        if t not in observed_set
+                    ],
+                    dtype=torch.long,
+                ),
+            )
+        else:
+            # All tasks are observed - set observed indices to all tasks
+            self.register_buffer(
+                "_observed_task_indices",
+                torch.arange(len(self._sorted_all_tasks), dtype=torch.long),
+            )
+            self.register_buffer(
+                "_unobserved_task_indices",
+                torch.tensor([], dtype=torch.long),
+            )
+        # Clean up temporary attributes
+        del self._all_tasks_inferred
+        del self._sorted_all_tasks
+
     def _map_tasks(self, task_values: Tensor) -> Tensor:
         """Map raw task values to the task indices used by the model.
 
@@ -407,6 +452,28 @@ def forward(self, x: Tensor) -> MultivariateNormal:
         covar_x = self.covar_module(x_covar)
         return MultivariateNormal(mean_x, covar_x)
 
+    def eval(self) -> Self:
+        r"""Puts the model in ``eval`` mode.
+
+        When unobserved tasks are present (i.e., ``all_tasks`` includes tasks not in
+        the training data), this method sets the covariance factor for unobserved tasks
+        to the mean of the observed tasks' covariance factors. This provides a
+        reasonable initialization for prediction on unobserved tasks.
+        """
+        if len(self._unobserved_task_indices) > 0:
+            task_covar_module = self.covar_module.kernels[1]
+            # Get the current covar_factor (transformed from raw_covar_factor)
+            covar_factor = task_covar_module.covar_factor
+            # Compute mean of observed tasks' covar_factor rows
+            observed_covar_factor = covar_factor[self._observed_task_indices]
+            mean_covar_factor = observed_covar_factor.mean(dim=0)
+            # Create new covar_factor with unobserved tasks set to mean
+            new_covar_factor = covar_factor.clone()
+            new_covar_factor[self._unobserved_task_indices] = mean_covar_factor
+            # Set the new covar_factor (this applies inverse_transform internally)
+            task_covar_module._set_covar_factor(new_covar_factor)
+        return super().eval()
+
     @classmethod
     def get_all_tasks(
         cls,

test/models/test_fully_bayesian_multitask.py

@@ -417,14 +417,19 @@ def test_fit_model(
 
         # Check the keys in the state dict
         true_keys = EXPECTED_KEYS_NOISE if infer_noise else EXPECTED_KEYS
-        extra_keys = []
+        extra_keys = [
+            "_observed_task_indices",
+            "_unobserved_task_indices",
+        ]
         if use_outcome_transform:
-            extra_keys = [
-                "outcome_transform.stdvs",
-                "outcome_transform._is_trained",
-                "outcome_transform._stdvs_sq",
-                "outcome_transform.means",
-            ]
+            extra_keys.extend(
+                [
+                    "outcome_transform.stdvs",
+                    "outcome_transform._is_trained",
+                    "outcome_transform._stdvs_sq",
+                    "outcome_transform.means",
+                ]
+            )
         if model._task_mapper is not None:
             extra_keys.append("_task_mapper")
         self.assertEqual(set(model.state_dict().keys()), {*true_keys, *extra_keys})
@@ -514,6 +519,8 @@ def test_fit_model_with_outcome_transform(self):
     def test_fit_model_with_task_mapper(self) -> None:
         dtype = torch.double
         tkwargs = {"device": self.device, "dtype": dtype}
+        # Test with contiguous all_tasks that includes an unobserved task
+        # all_tasks=[0, 1, 2] is contiguous from 0, so no mapper is needed
         all_tasks = [0, 1, 2]
         observed_task_values = [0, 2]
         output_tasks = [2]
@@ -526,16 +533,36 @@ def test_fit_model_with_task_mapper(self) -> None:
             validate_task_values=False,
             **tkwargs,
         )
-        self.assertTrue(
-            torch.equal(model._task_mapper, torch.tensor([0, 1, 1], **tkwargs))
-        )
-        self.test_fit_model(
+        # With contiguous all_tasks=[0, 1, 2], no task mapper is needed
+        # because task values are already contiguous integers starting from 0
+        self.assertIsNone(model._task_mapper)
+        # Verify the pyro_model has the correct number of tasks (3, not 2)
+        self.assertEqual(model.pyro_model.num_tasks, 3)
+
+        # Also test non-contiguous all_tasks to ensure mapper is created
+        all_tasks_noncontig = [0, 2, 5]
+        observed_task_values_noncontig = [0, 5]
+        output_tasks_noncontig = [5]
+        _, _, _, model_noncontig = self._get_data_and_model(
+            infer_noise=True,
             use_outcome_transform=True,
-            all_tasks=all_tasks,
-            observed_task_values=observed_task_values,
-            output_tasks=output_tasks,
+            output_tasks=output_tasks_noncontig,
+            observed_task_values=observed_task_values_noncontig,
+            all_tasks=all_tasks_noncontig,
             validate_task_values=False,
+            **tkwargs,
         )
+        # With non-contiguous all_tasks=[0, 2, 5], a mapper is required
+        # Mapper maps: 0→0, 2→1, 5→2; other indices map to NaN
+        self.assertIsNotNone(model_noncontig._task_mapper)
+        self.assertEqual(model_noncontig._task_mapper[0].item(), 0.0)
+        self.assertEqual(model_noncontig._task_mapper[2].item(), 1.0)
+        self.assertEqual(model_noncontig._task_mapper[5].item(), 2.0)
+        self.assertTrue(torch.isnan(model_noncontig._task_mapper[1]))
+        self.assertTrue(torch.isnan(model_noncontig._task_mapper[3]))
+        self.assertTrue(torch.isnan(model_noncontig._task_mapper[4]))
+        # Verify pyro_model has correct number of tasks
+        self.assertEqual(model_noncontig.pyro_model.num_tasks, 3)
 
     def test_transforms(self, infer_noise: bool = False):
         tkwargs = {"device": self.device, "dtype": torch.double}

test/models/test_multitask.py

@@ -445,9 +445,12 @@ def test_all_tasks_input(self) -> None:
         model = MultiTaskGP(
             train_X=train_X, train_Y=train_Y, task_feature=0, all_tasks=[0, 1, 2, 3]
         )
-        self.assertEqual(model.num_tasks, 2)
+        self.assertEqual(model.num_tasks, 4)
         # Check that PositiveIndexKernel knows of all tasks.
-        self.assertEqual(model.covar_module.kernels[1].raw_covar_factor.shape[0], 2)
+        self.assertEqual(model.covar_module.kernels[1].raw_covar_factor.shape[0], 4)
+        # Check that observed and unobserved task indices are computed correctly.
+        self.assertEqual(model._observed_task_indices.tolist(), [0, 1])
+        self.assertEqual(model._unobserved_task_indices.tolist(), [2, 3])
 
     def test_MultiTaskGP_construct_inputs(self) -> None:
         for dtype, fixed_noise, skip_task_features_in_datasets in zip(
@@ -540,13 +543,98 @@ def test_validatation_of_task_values(self) -> None:
             validate_task_values=True,
         )
 
+        # Task 2 is in all_tasks, so it should be valid even with validation enabled
+        self.assertTrue(
+            torch.equal(
+                torch.tensor([1], **tkwargs),
+                model._map_tasks(task_values=torch.tensor([2], **tkwargs)),
+            )
+        )
+
+        # Task 3 is NOT in all_tasks, so it should raise an error
         with self.assertRaisesRegex(
             ValueError,
             "Received invalid raw task values. Expected raw value to be in"
-            r" \{0, 1\}, but got unexpected task"
-            r" values: \{2\}.",
+            r" \{0, 1, 2\}, but got unexpected task"
+            r" values: \{3\}.",
         ):
-            model._map_tasks(task_values=torch.tensor([2], **tkwargs))
+            model._map_tasks(task_values=torch.tensor([3], **tkwargs))
+
+    def test_multitask_gp_unobserved_tasks(self) -> None:
+        """Test MultiTaskGP with unobserved tasks.
+
+        This test verifies that:
+        1. Creating a model with all_tasks including unobserved tasks works
+        2. In train mode, unobserved task covar_factor is at random initialization
+        3. In eval mode, unobserved task covar_factor is set to mean of observed
+        4. Predictions work for the unobserved task
+        """
+        tkwargs = {"device": self.device, "dtype": torch.double}
+
+        # Create data for tasks 0 and 2 only (task 1 is unobserved)
+        _, (train_X, train_Y, _) = gen_multi_task_dataset(task_values=[0, 2], **tkwargs)
+
+        # Create model with all_tasks=[0, 1, 2] including unobserved task 1
+        model = MultiTaskGP(
+            train_X=train_X,
+            train_Y=train_Y,
+            task_feature=0,
+            all_tasks=[0, 1, 2],
+        )
+        model.to(**tkwargs)
+
+        # Verify model.num_tasks == 3
+        self.assertEqual(model.num_tasks, 3)
+
+        # Verify observed and unobserved task indices are correctly set
+        self.assertEqual(model._observed_task_indices.tolist(), [0, 2])
+        self.assertEqual(model._unobserved_task_indices.tolist(), [1])
+
+        # Get the task covariance module
+        task_covar_module = model.covar_module.kernels[1]
+
+        # In train mode, get the covar_factor for unobserved task (index 1)
+        model.train()
+        train_covar_factor = task_covar_module.covar_factor.clone()
+        unobserved_train_covar = train_covar_factor[1]
+        observed_train_covar = train_covar_factor[[0, 2]]
+        mean_observed_train = observed_train_covar.mean(dim=0)
+
+        # Unobserved task covar_factor should be at random init in train mode
+        # (very unlikely to be exactly equal to mean of observed)
+        self.assertFalse(
+            torch.allclose(unobserved_train_covar, mean_observed_train, atol=1e-6)
+        )
+
+        # Switch to eval mode
+        model.eval()
+
+        # In eval mode, get the covar_factor for unobserved task
+        eval_covar_factor = task_covar_module.covar_factor.clone()
+        unobserved_eval_covar = eval_covar_factor[1]
+        observed_eval_covar = eval_covar_factor[[0, 2]]
+        mean_observed_eval = observed_eval_covar.mean(dim=0)
+
+        # Unobserved task covar_factor should equal mean of observed in eval mode
+        self.assertTrue(
+            torch.allclose(unobserved_eval_covar, mean_observed_eval, atol=1e-6)
+        )
+
+        # Verify predictions work for the unobserved task
+        # Create test input for unobserved task (task 1)
+        test_X = torch.rand(3, 2, **tkwargs)
+        test_X[:, 0] = 1.0  # Set task feature to 1 (unobserved task)
+
+        with torch.no_grad():
+            posterior = model.posterior(X=test_X)
+
+        # Verify posterior has expected shape
+        self.assertEqual(posterior.mean.shape, torch.Size([3, 1]))
+        self.assertEqual(posterior.variance.shape, torch.Size([3, 1]))
+
+        # Verify we can sample from the posterior
+        samples = posterior.rsample(sample_shape=torch.Size([2]))
+        self.assertEqual(samples.shape, torch.Size([2, 3, 1]))
 
 
 class TestKroneckerMultiTaskGP(BotorchTestCase):