Source code for ddmtolab.Algorithms.MTSO.MTBO

"""
Multi-Task Bayesian Optimization (MTBO)

This module implements MTBO for expensive multi-task optimization with knowledge transfer via multi-task Gaussian
processes.

References
----------
    [1] Swersky, Kevin, Jasper Snoek, and Ryan P. Adams. "Multi-task bayesian optimization." Advances in neural information processing systems 26 (2013).

Notes
-----
Author: Jiangtao Shen
Email: j.shen5@exeter.ac.uk
Date: 2025.11.12
Version: 1.0
"""
from tqdm import tqdm
import torch
import time
from ddmtolab.Methods.Algo_Methods.algo_utils import *
from ddmtolab.Methods.Algo_Methods.bo_utils import mtgp_build, mtbo_next_point
import warnings

warnings.filterwarnings("ignore")


[docs] class MTBO: """ Multi-Task Bayesian Optimization for expensive multi-task optimization problems. Attributes ---------- algorithm_information : dict Dictionary containing algorithm capabilities and requirements """ algorithm_information = { 'n_tasks': '[2, K]', 'dims': 'unequal', 'objs': 'equal', 'n_objs': '1', 'cons': 'equal', 'n_cons': '0', 'expensive': 'True', 'knowledge_transfer': 'True', 'n_initial': 'unequal', 'max_nfes': 'unequal' } @classmethod def get_algorithm_information(cls, print_info=True): return get_algorithm_information(cls, print_info)
[docs] def __init__(self, problem, n_initial=None, max_nfes=None, save_data=True, save_path='./Data', name='MTBO', disable_tqdm=True): """ Initialize Multi-Task Bayesian Optimization algorithm. Parameters ---------- problem : MTOP Multi-task optimization problem instance n_initial : int or List[int], optional Number of initial samples per task (default: 50) max_nfes : int or List[int], optional Maximum number of function evaluations per task (default: 100) save_data : bool, optional Whether to save optimization data (default: True) save_path : str, optional Path to save results (default: './TestData') name : str, optional Name for the experiment (default: 'MTBO_test') disable_tqdm : bool, optional Whether to disable progress bar (default: True) """ self.problem = problem self.n_initial = n_initial if n_initial is not None else 50 self.max_nfes = max_nfes if max_nfes is not None else 100 self.save_data = save_data self.save_path = save_path self.name = name self.disable_tqdm = disable_tqdm
[docs] def optimize(self): """ Execute the Multi-Task Bayesian Optimization algorithm. Returns ------- Results Optimization results containing decision variables, objectives, and runtime """ data_type = torch.double start_time = time.time() problem = self.problem nt = problem.n_tasks dims = problem.dims n_initial_per_task = par_list(self.n_initial, nt) max_nfes_per_task = par_list(self.max_nfes, nt) # Initialize samples using Latin Hypercube Sampling decs = initialization(problem, self.n_initial, method='lhs') objs, _ = evaluation(problem, decs) nfes_per_task = n_initial_per_task.copy() pbar = tqdm(total=sum(max_nfes_per_task), initial=sum(n_initial_per_task), desc=f"{self.name}", disable=self.disable_tqdm) while sum(nfes_per_task) < sum(max_nfes_per_task): # Skip tasks that have exhausted their evaluation budget active_tasks = [i for i in range(nt) if nfes_per_task[i] < max_nfes_per_task[i]] if not active_tasks: break # Build multi-task Gaussian process surrogate model with normalized objectives objs_normalized, _, _ = normalize(objs, axis=0, method='minmax') mtgp = mtgp_build(decs, objs_normalized, dims, data_type=data_type) for i in active_tasks: # Select next sample point via acquisition function optimization candidate_np = mtbo_next_point(mtgp=mtgp, task_id=i, objs=objs_normalized, dims=dims, nt=nt, data_type=data_type) obj, _ = evaluation_single(problem, candidate_np, i) decs[i], objs[i] = vstack_groups((decs[i], candidate_np), (objs[i], obj)) nfes_per_task[i] += 1 pbar.update(1) pbar.close() runtime = time.time() - start_time all_decs, all_objs = build_staircase_history(decs, objs, k=1) # Save results results = build_save_results(all_decs=all_decs, all_objs=all_objs, runtime=runtime, max_nfes=nfes_per_task, bounds=problem.bounds, save_path=self.save_path, filename=self.name, save_data=self.save_data) return results