贝叶斯优化中的探索与利用策略：采集函数详解
贝叶斯优化（Bayesian Optimization）是一种在黑盒函数上寻找全局最优解的有效方法，尤其适用于函数评估成本高昂的场景。其核心思想包含两个主要组成部分：
- 代理模型（Surrogate Model）：也被称为替代模型或高斯过程模型。它基于已观测到的函数值点，构建一个易于处理的数学模型来近似目标黑盒函数。这个模型不仅能预测给定输入处的函数值，还能提供预测的不确定性估计。
- 优化策略（Optimization Strategy）：用于决定下一个采样点的位置。这一策略通常通过采集函数（Acquisition Function）实现。采集函数是一个基于代理模型构建的辅助函数，它衡量了搜索空间中每个点被评估的"价值"。高价值的点通常同时满足以下两个条件：
- 该点有较大潜力成为新的全局最优点（利用）。
- 评估该点能显著降低代理模型在未知区域的不确定性（探索）。
本文将深入探讨贝叶斯优化中的采集函数，并以HEBO（Heteroscedastic Evolutionary Bayesian Optimization）框架为例，剖析其在实际应用中的具体实现。
HEBO框架概述
HEBO框架在SCOOT等项目中被广泛采用，其设计旨在克服传统贝叶斯优化中的局限性。以下是HEBO解决的一些关键问题：
背景挑战
-
建模假设与实际问题：传统高斯过程（GP）模型通常假设数据具有同方差性，但实际机器学习超参数优化任务往往表现出显著的异方差性（即不同区域的噪声水平不同）和非平稳性（即函数行为在不同区域差异很大）。标准的GP模型难以有效处理此类复杂情境。
-
采集函数与优化器假设：传统的采集函数优化器常假设黑盒函数只包含连续变量，并使用一阶或二阶优化方法。这忽略了超参数优化中常见的离散变量。此外，多数贝叶斯优化框架仅采用单一采集函数，未能充分利用多种策略组合的优势。
HEBO的核心观点
-
问题1：超参数优化任务是否平稳？
回答：即使是简单的机器学习任务也常表现出高度的非平稳性。
-
问题2：超参数优化任务是否同质？
回答：实践中，超参数优化任务常常具有显著的异方差性。
-
问题3：不同采集函数输出的候选点是否冲突？
回答：确实，没有"万能"的采集函数。不同的采集函数往往在探索与利用之间做出不同权衡，其建议的候选点可能相互矛盾。
HEBO解决方案总结
- 针对非平稳性，对输入数据进行转换处理。
- 针对异方差性，对输出数据进行转换处理。
- 为应对不同采集函数间的冲突，HEBO引入了：
- 具有鲁棒性的采集函数。
- 多目标采集函数。
HEBO的整体流程包括伪随机序列生成、输入/输出数据转换、高斯过程模型训练、构建并优化多目标采集函数，最终选出最优候选点。该方法在多项BBO（Black-Box Optimization）竞赛中表现卓越，其核心在于对数据处理、代理模型和采集函数各环节的深度优化。
采集函数的设计与实现
HEBO框架中针对单目标和多目标优化使用了不同的采集函数及其对应的优化器。本文将聚焦于两种主要的采集函数实现：AcquisitionFunctionMACE 和 UniversalAcquisitionFunction。
AcquisitionFunctionMACE (用于单目标带约束优化)
AcquisitionFunctionMACE 被设计用于处理具有约束条件的单目标优化问题。它通过同时优化多种采集准则（LCB、EI、PI）来平衡探索与利用，并将它们转化为一个多目标优化问题。
import torch
import numpy as np
from torch.distributions import Normal
# 辅助函数（在实际代码中会从其他模块导入或作为类方法）
# 这里仅为示例，实际的硬约束和隐藏约束检查会更复杂
def _get_hard_constraint_violation(x_cont, max_seq_len, num_hard_constr, pred_mean, search_space):
if num_hard_constr == 0 or search_space is None:
return torch.zeros([x_cont.size()[0], 0], device=x_cont.device)
# 模拟硬约束违反，例如根据某个参数范围
violations = torch.rand(x_cont.size()[0], num_hard_constr, device=x_cont.device) * 0.5
return violations
def _get_hidden_constraint_violation(x_cont, x_disc, rf_with_threshold, num_hidden_constr, pred_mean, search_space):
if num_hidden_constr == 0 or rf_with_threshold is None or search_space is None:
return torch.zeros([x_cont.size()[0], 0], device=x_cont.device)
rf_model, threshold = rf_with_threshold
# 模拟随机森林预测，真实场景需要将输入转换并调用rf_model
rf_predictions = torch.rand(x_cont.size()[0], num_hidden_constr, device=x_cont.device)
violations = (rf_predictions - threshold).clamp(min=0) # 大于阈值为违反
return violations
# 抽象基类，用于定义采集函数接口
class BaseAcquisition(torch.nn.Module):
def __init__(self, surrogate_model, **config):
super().__init__()
self.surrogate_model = surrogate_model
self.config = config
@property
def objective_count(self) -> int:
raise NotImplementedError
@property
def gp_constraint_count(self) -> int:
raise NotImplementedError
@property
def hidden_constraint_count(self) -> int:
raise NotImplementedError
@property
def hard_constraint_count(self) -> int:
raise NotImplementedError
@property
def total_constraint_count(self) -> int:
return self.gp_constraint_count + self.hard_constraint_count + self.hidden_constraint_count
def evaluate(self, continuous_x: torch.FloatTensor, discrete_x: torch.LongTensor) -> torch.FloatTensor:
raise NotImplementedError
class AcquisitionFunctionMACE(BaseAcquisition):
def __init__(self, surrogate_model, current_best_objective, gp_constraint_count, constraint_threshold_values, **config):
super().__init__(surrogate_model, **config)
self.exploration_weight = config.get('kappa', 2.0) # LCB探索系数
self.numerical_stability_eps = config.get('eps', 1e-4) # 避免除零
self._hard_constraint_count = config.get('num_hard_constr', 3)
self._hidden_constraint_count = config.get('num_hidden_constr', 1)
self.rf_classifier_and_threshold = config.get('rf_with_thres', None)
self.max_sequence_length = config.get('max_sequence_length', 12)
self.search_space = config.get('space', None)
self.best_objective_value = current_best_objective # 当前观测到的最佳目标值
self._gp_constraint_count = gp_constraint_count # GP模型建模的约束数量
self.constraint_thresholds = constraint_threshold_values # 约束阈值
if not self.search_space: # 如果没有搜索空间，则禁用硬约束和隐藏约束
self._hard_constraint_count = 0
self._hidden_constraint_count = 0
if self._hidden_constraint_count > 0:
assert isinstance(self.rf_classifier_and_threshold, tuple), \
f'隐藏约束计数为{self._hidden_constraint_count}，必须提供随机森林模型及对应阈值。'
@property
def objective_count(self) -> int:
return 3 # LCB, EI, PI 三个准则
@property
def gp_constraint_count(self) -> int:
return self._gp_constraint_count
@property
def hidden_constraint_count(self) -> int:
return self._hidden_constraint_count
@property
def hard_constraint_count(self) -> int:
return self._hard_constraint_count
def evaluate(self, continuous_x: torch.FloatTensor, discrete_x: torch.LongTensor) -> torch.FloatTensor:
"""
评估采集函数值。MACE旨在最小化 (-log(EI), -log(PI), LCB)。
"""
with torch.no_grad():
# 使用高斯过程模型预测均值和方差
predicted_mean, predicted_variance = self.surrogate_model.predict(continuous_x, discrete_x)
# --- 目标函数部分：计算三种采集准则 ---
# 提取目标函数的预测值和方差（第一列）
obj_mean_pred = predicted_mean[:, :1]
obj_variance_pred = predicted_variance[:, :1]
# 计算噪声标准差，用于EI/PI计算
# 假设 self.surrogate_model.noise_std 是噪声的标准差
noise_std_factor = np.sqrt(2.0) * self.surrogate_model.noise_std[0]
# 计算目标函数的标准差，并确保数值稳定性
obj_std_dev = obj_variance_pred.sqrt().clamp(min = torch.finfo(obj_variance_pred.dtype).eps)
# 1. 计算 LCB (Lower Confidence Bound)：均值 - kappa * 标准差
lcb_val = (obj_mean_pred + noise_std_factor * torch.randn(obj_mean_pred.shape, device=continuous_x.device)) - self.exploration_weight * obj_std_dev
# 2. 计算标准化改进量 (Normalized Improvement)
# (当前最佳值 - epsilon - 预测值) / 标准差
std_improvement = ((self.best_objective_value - self.numerical_stability_eps - obj_mean_pred - noise_std_factor * torch.randn(obj_mean_pred.shape, device=continuous_x.device)) / obj_std_dev)
# 创建标准正态分布对象
std_normal_dist = Normal(0., 1.)
# 3. 计算 PI (Probability of Improvement)
prob_improvement = std_normal_dist.cdf(std_improvement)
# 4. 计算 EI (Expected Improvement)
# EI = std_dev * (Phi(normed) * normed + phi(normed))
log_pdf_normed = std_normal_dist.log_prob(std_improvement)
expected_improvement = obj_std_dev * (prob_improvement * std_improvement + log_pdf_normed.exp())
# --- 处理数值稳定性：当EI和PI值极小时使用对数近似 ---
# 定义近似对数EI和对数PI
# 这些近似公式在 std_improvement 极小（非常负）时更稳定
log_ei_approx = obj_std_dev.log() - 0.5 * std_improvement**2 - (std_improvement**2 - 1).log()
log_pi_approx = -0.5 * std_improvement**2 - torch.log(-1 * std_improvement) - torch.log(torch.sqrt(torch.tensor(2 * np.pi, device=continuous_x.device)))
# 判断哪些点需要使用近似计算：当标准化改进量非常小且精确对数值不有限时
should_use_approx = ~((std_improvement > -6) & torch.isfinite(expected_improvement.log()) & torch.isfinite(prob_improvement.log())).reshape(-1)
# 初始化输出张量，3列对应3个采集准则
acquisition_outputs = torch.zeros(continuous_x.shape[0], self.objective_count, device=continuous_x.device)
# 第一列：LCB目标（探索性）
acquisition_outputs[:, 0] = lcb_val.reshape(-1)
# 第二列：EI准则（取负对数用于最小化）
acquisition_outputs[:, 1][should_use_approx] = -1 * log_ei_approx[should_use_approx].reshape(-1)
acquisition_outputs[:, 1][~should_use_approx] = -1 * expected_improvement[~should_use_approx].log().reshape(-1)
# 第三列：PI准则（取负对数用于最小化）
acquisition_outputs[:, 2][should_use_approx] = -1 * log_pi_approx[should_use_approx].reshape(-1)
acquisition_outputs[:, 2][~should_use_approx] = -1 * prob_improvement[~should_use_approx].log().reshape(-1)
# --- 模型约束部分 ---
gp_constraint_violations = torch.zeros([acquisition_outputs.size()[0], 0], device=continuous_x.device)
if self.constraint_thresholds is not None and self.gp_constraint_count > 0:
# 提取约束函数的预测值和方差（从第二列开始）
constr_mean_pred = predicted_mean[:, 1 : 1 + self.gp_constraint_count]
constr_variance_pred = predicted_variance[:, 1 : 1 + self.gp_constraint_count]
# 计算约束函数的标准差
constr_std_dev = constr_variance_pred.sqrt().clamp(min = torch.finfo(constr_variance_pred.dtype).eps)
# 计算约束违反程度：预测值 + kappa*标准差 - 阈值
# 目标是使约束值 < 阈值，因此此值越小越好（或为负）
gp_constraint_violations = (constr_mean_pred + noise_std_factor * torch.randn(constr_mean_pred.shape, device=continuous_x.device)) + self.exploration_weight * constr_std_dev - self.constraint_thresholds
# 获取硬约束违反程度（基于先验知识的约束）
hard_constraint_violations = _get_hard_constraint_violation(continuous_x, self.max_sequence_length,
self.hard_constraint_count, predicted_mean, self.search_space)
# 获取隐藏约束违反程度（基于随机森林模型的约束）
hidden_constraint_violations = _get_hidden_constraint_violation(continuous_x, discrete_x, self.rf_classifier_and_threshold,
self.hidden_constraint_count, predicted_mean, self.search_space)
# 合并所有输出：3个采集准则目标 + 模型约束 + 硬约束 + 隐藏约束
return torch.concat([acquisition_outputs, gp_constraint_violations, hard_constraint_violations, hidden_constraint_violations], dim=1)
UniversalAcquisitionFunction (用于多目标带约束优化)
UniversalAcquisitionFunction 旨在更通用地处理多目标优化问题，同时考虑模型约束、硬约束和隐藏约束。它为每个目标和模型约束都计算一个下置信界（LCB），将多目标优化转化为寻找帕累托前沿的问题。
import torch
import numpy as np
# 假设 BaseAcquisition 是已定义的基类，_get_hard_constraint_violation, _get_hidden_constraint_violation 也已定义
class UniversalAcquisitionFunction(BaseAcquisition):
def __init__(self, surrogate_model, objective_count, gp_constraint_count, **config):
super().__init__(surrogate_model, **config)
self._objective_count = objective_count
self._gp_constraint_count = gp_constraint_count
self._hard_constraint_count = config.get('num_hard_constr', 3)
self._hidden_constraint_count = config.get('num_hidden_constr', 1)
self.rf_classifier_and_threshold = config.get('rf_with_thres', None)
self.search_space = config.get('space', None)
self.objective_exploration_weight = config.get('kappa', 2.0) # 目标LCB探索系数
self.constraint_exploration_weight = config.get('c_kappa', 0.) # 约束LCB探索系数
self.use_noise_in_prediction = config.get('use_noise', True)
self.max_sequence_length = config.get('max_sequence_length', 12)
if not self.search_space:
self._hard_constraint_count = 0
self._hidden_constraint_count = 0
assert self.surrogate_model.output_dim == self.objective_count + self.gp_constraint_count, \
f"模型输出维度({self.surrogate_model.output_dim})与目标数({self.objective_count})加GP约束数({self.gp_constraint_count})不匹配。"
assert self.objective_count >= 1, "目标函数数量必须至少为1。"
if self._hidden_constraint_count > 0:
assert isinstance(self.rf_classifier_and_threshold, tuple), \
f'隐藏约束计数为{self._hidden_constraint_count}，必须提供随机森林模型及对应阈值。'
@property
def objective_count(self) -> int:
return self._objective_count
@property
def gp_constraint_count(self) -> int:
return self._gp_constraint_count
@property
def hidden_constraint_count(self) -> int:
return self._hidden_constraint_count
@property
def hard_constraint_count(self) -> int:
return self._hard_constraint_count
def evaluate(self, continuous_x: torch.FloatTensor, discrete_x: torch.LongTensor) -> torch.FloatTensor:
"""
通用采集函数评估方法，处理多目标和约束优化。
目标是最小化 (lcb_o1, ..., lcb_om)
并满足 lcb_c1 < 0, ..., lcb_cn < 0
"""
with torch.no_grad():
predicted_mean, predicted_variance = self.surrogate_model.predict(continuous_x, discrete_x)
# 计算标准差，并确保数值稳定性
std_dev = predicted_variance.sqrt().clamp(min = torch.finfo(predicted_variance.dtype).eps)
# 如果启用噪声，在预测均值上添加高斯噪声
if self.use_noise_in_prediction:
noise_std = self.surrogate_model.noise_std # 假设模型有 noise_std 属性
predicted_mean += noise_std * torch.randn(predicted_mean.shape, device=continuous_x.device)
# 初始化输出张量，形状与预测值相同
acquisition_values = torch.ones(predicted_mean.shape, device=continuous_x.device)
# 计算目标函数的下置信界（LCB）
# 对于最小化问题，LCB = 均值 - kappa * 标准差
acquisition_values[:, :self.objective_count] = predicted_mean[:, :self.objective_count] - \
self.objective_exploration_weight * std_dev[:, :self.objective_count]
# 计算约束函数的下置信界
# 对于约束c < 0，LCB = 均值 - c_kappa * 标准差
# 这里是为了处理约束，将其转换为一个需要最小化的值。如果LCB > 0，则违反约束。
acquisition_values[:, self.objective_count:] = predicted_mean[:, self.objective_count:] - \
self.constraint_exploration_weight * std_dev[:, self.objective_count:]
# 获取硬约束违反程度
hard_constraint_violations = _get_hard_constraint_violation(continuous_x, self.max_sequence_length,
self.hard_constraint_count, predicted_mean, self.search_space)
# 获取隐藏约束违反程度
hidden_constraint_violations = _get_hidden_constraint_violation(continuous_x, discrete_x, self.rf_classifier_and_threshold,
self.hidden_constraint_count, predicted_mean, self.search_space)
# 合并所有输出：目标LCB + 模型约束LCB + 硬约束 + 隐藏约束
return torch.concat([acquisition_values, hard_constraint_violations, hidden_constraint_violations], dim=1)
两种采集函数的对比
AcquisitionFunctionMACE 和 UniversalAcquisitionFunction 在设计理念和应用场景上存在显著差异：
AcquisitionFunctionMACE(MACEConstr)：- 主要用于单目标优化问题，但在内部将LCB、EI、PI三种采集准则转化为一个多目标优化问题来处理，以平衡探索与利用。
- 对目标函数输出通常会进行幂变换（如Box-Cox或Yeo-Johnson），以使数据更接近正态分布，从而提高高斯过程模型的拟合性能。
- 在选择最终的建议点时，会倾向于选择预测值最小和不确定性最大的点。
UniversalAcquisitionFunction(GeneralAcq)：- 主要用于真正的多目标优化问题，为每个目标和每个模型约束都计算其下置信界（LCB）。
- 通常不默认对目标函数输出进行变换，直接使用原始观测值。
- 在多目标场景下，如果提供了参考点，会使用期望超体积改进（EHVI）等指标进行选择；否则，基于不确定性等策略进行选择。
优化器：如何利用采集函数选择候选点
在HEBO框架中，优化器负责管理整个贝叶斯优化流程，包括历史数据管理、代理模型训练、采集函数评估以及最终候选点的选择。下面介绍两个主要的优化器：ConstrainedBayesianOptimizer 和 GeneralBayesianOptimizer。
import pandas as pd
import numpy as np
import torch
from typing import Optional
# 抽象的优化器基类，用于定义通用接口
class AbstractOptimizer:
support_parallel_optimization = False
support_combinatorial_optimization = False
support_contextual_optimization = False
def __init__(self, design_space):
self.design_space = design_space
self.X = pd.DataFrame(columns = self.design_space.parameter_names)
self.y = np.zeros((0, 1)) # 默认初始为单目标
def add_observation(self, X_new: pd.DataFrame, y_new: np.ndarray):
"""添加新的观测数据"""
if self.X.empty:
self.X = X_new
self.y = y_new
else:
self.X = pd.concat([self.X, X_new], ignore_index=True)
self.y = np.vstack([self.y, y_new])
def suggest(self, num_suggestions: int = 1, fixed_params: Optional[dict] = None, rf_model_and_threshold = None) -> pd.DataFrame:
raise NotImplementedError
def _check_uniqueness(self, candidate_df: pd.DataFrame) -> np.ndarray:
"""检查候选点是否已在历史观测中存在"""
if self.X.empty:
return np.ones(candidate_df.shape[0], dtype=bool)
# 将 DataFrame 转换为元组列表进行比较，效率更高
# 需确保转换时列的顺序与 design_space.parameter_names 一致
candidate_values = candidate_df[self.design_space.parameter_names].values
historical_values = self.X[self.design_space.parameter_names].values
historical_tuples = {tuple(row) for row in historical_values}
candidate_is_unique = [tuple(row) not in historical_tuples for row in candidate_values]
return np.array(candidate_is_unique)
def _get_best_historical_id(self, fixed_params: Optional[dict] = None) -> int:
"""获取历史数据中最佳观测点的索引（假设目标是最小化第一列）"""
target_y = self.y[:, 0]
if fixed_params:
filtered_indices = []
for i, (_, row) in enumerate(self.X.iterrows()):
match = True
for k, v in fixed_params.items():
if row[k] != v:
match = False
break
if match:
filtered_indices.append(i)
if filtered_indices:
return filtered_indices[target_y[filtered_indices].argmin()]
else: # 如果没有满足固定参数的历史点，则返回全局最佳
return target_y.argmin()
return target_y.argmin()
# 辅助函数（模拟实际功能）
# DesignSpace 类应该包含参数定义和 transform/inverse_transform 方法
class DesignSpace:
def __init__(self, params_config: list):
self.parameters = {p['name']: p for p in params_config}
self.parameter_names = [p['name'] for p in params_config]
self.numeric_parameter_names = [p['name'] for p in params_config if p['type'] == 'continuous']
self.categorical_parameter_names = [p['name'] for p in params_config if p['type'] == 'discrete']
self.numeric_param_count = len(self.numeric_parameter_names)
self.categorical_param_count = len(self.categorical_parameter_names)
self.total_parameter_count = len(self.parameter_names)
self.opt_lower_bound = torch.tensor([0.0] * self.total_parameter_count) # 示例
self.opt_upper_bound = torch.tensor([1.0] * self.total_parameter_count) # 示例
def transform(self, df: pd.DataFrame):
# 模拟将DataFrame转换为连续和离散张量
cont_vals = df[self.numeric_parameter_names].values
disc_vals = df[self.categorical_parameter_names].values
return torch.from_numpy(cont_vals).float(), torch.from_numpy(disc_vals).long()
def inverse_transform(self, x_cont: torch.FloatTensor, x_disc: torch.LongTensor):
# 模拟将张量转换回DataFrame
df_cont = pd.DataFrame(x_cont.numpy(), columns=self.numeric_parameter_names)
df_disc = pd.DataFrame(x_disc.numpy(), columns=self.categorical_parameter_names)
return pd.concat([df_cont, df_disc], axis=1)
# 模拟幂变换功能
def _enable_power_transform_helper(data_array: np.ndarray) -> torch.FloatTensor:
# 实际应调用scipy.stats.boxcox或sklearn.preprocessing.PowerTransformer
# 为简化，这里返回输入的 PyTorch 张量
return torch.FloatTensor(data_array)
# 模拟代理模型实例，用于占位
def _get_surrogate_model_instance(model_type: str, num_cont: int, num_disc: int, output_dim: int, num_unique_cats: Optional[list]=None, **config):
class MockSurrogateModel(torch.nn.Module):
def __init__(self, num_outputs: int):
super().__init__()
self.output_dim = num_outputs
self.noise_std = torch.tensor([0.1] * num_outputs) # 模拟噪声标准差
def predict(self, x_cont: torch.FloatTensor, x_disc: torch.LongTensor):
mean = torch.randn(x_cont.shape[0], self.output_dim, device=x_cont.device)
variance = torch.exp(torch.randn(x_cont.shape[0], self.output_dim, device=x_cont.device)) * 0.1 + 0.01
return mean, variance
def fit(self, X_cont: torch.FloatTensor, X_disc: torch.LongTensor, y_obs: torch.FloatTensor):
pass # 模拟训练
def sample_y(self, X_cont: torch.FloatTensor, X_disc: torch.LongTensor, n_samples: int):
mean, var = self.predict(X_cont, X_disc)
return mean.unsqueeze(0) + var.sqrt().unsqueeze(0) * torch.randn(n_samples, *mean.shape, device=x_cont.device)
return MockSurrogateModel(output_dim)
# 模拟确保硬约束满足
def _ensure_hard_constraints_satisfied(df_samples: pd.DataFrame, max_sequence_length_val: int) -> pd.DataFrame:
# 在实际场景中，这会根据 design_space 中的规则过滤或调整 df_samples
return df_samples
# 模拟进化优化器类，用于占位
class EvolutionarySearchEngine:
def __init__(self, design_space, acquisition_func, pop=100, iters=100, verbose=False, es='nsga2'):
self.design_space = design_space
self.acquisition_func = acquisition_func
self.pop = pop
self.iters = iters
self.es = es
def optimize(self, initial_suggest=None, fixed_input=None):
# 模拟进化优化器的输出，返回随机生成的DataFrame
num_to_generate = initial_suggest.shape[0] if initial_suggest is not None else self.pop
if fixed_input: # 如果有固定输入，只生成相应数量的补充
num_to_generate = max(1, num_to_generate)
# 模拟生成结果，确保返回DataFrame
mock_samples_cont = torch.rand(num_to_generate, self.design_space.numeric_param_count)
mock_samples_disc = torch.randint(0, 2, (num_to_generate, self.design_space.categorical_param_count)).long()
result_df = self.design_space.inverse_transform(mock_samples_cont, mock_samples_disc)
if initial_suggest is not None:
result_df = pd.concat([initial_suggest, result_df], ignore_index=True)
# 确保去重
result_df = result_df.drop_duplicates()
return result_df
class ConstrainedBayesianOptimizer(AbstractOptimizer):
support_parallel_optimization = True
support_combinatorial_optimization = True
support_contextual_optimization = True
def __init__(self,
design_space: DesignSpace,
gp_constraint_count: int,
constraint_threshold_values: Optional[np.ndarray] = None,
surrogate_model_type: str = 'multi_task',
gp_base_model_type: str = 'gpy',
initial_random_samples_count: Optional[int] = None,
acquisition_class = AcquisitionFunctionMACE,
evolutionary_strategy: str = 'nsga2',
model_configuration: Optional[dict] = None,
sobol_seed: Optional[int] = None,
num_explicit_hard_constraints: Optional[int] = None,
num_learned_hidden_constraints: Optional[int] = 1,
max_sequence_length: Optional[int] = 12):
"""
初始化约束贝叶斯优化器。
"""
super().__init__(design_space)
self.evolutionary_strategy = evolutionary_strategy
self.y = np.zeros((0, gp_constraint_count + 1)) # 目标值 + GP约束值
self.surrogate_model_type = surrogate_model_type
# 初始随机采样次数：至少为2，或设计空间维度+1
self.initial_random_samples = max(2, 1 + self.design_space.total_parameter_count) \
if initial_random_samples_count is None else initial_random_samples_count
self.sobol_seed = sobol_seed
# SobolEngine 假设存在，这里用torch.rand模拟准随机采样
self.sobol_sampler = lambda n: torch.rand(n, self.design_space.total_parameter_count)
self.acquisition_function_class = acquisition_class
self.gp_base_model_type = gp_base_model_type
self.model_configuration = model_configuration if model_configuration is not None else {}
# 约束相关参数
self.num_explicit_hard_constraints = num_explicit_hard_constraints
self.num_learned_hidden_constraints = num_learned_hidden_constraints
self.max_sequence_length = max_sequence_length
self._gp_constraint_count = gp_constraint_count
# 处理约束阈值的多种输入格式
if isinstance(constraint_threshold_values, list):
self.constraint_thresholds = np.array(constraint_threshold_values).reshape([1, -1])
assert self.constraint_thresholds.shape[1] == gp_constraint_count, \
'如果提供列表，请为每个输出维度提供一个阈值。'
elif isinstance(constraint_threshold_values, (int, float)):
self.constraint_thresholds = np.array([constraint_threshold_values]).reshape([1, -1])
assert gp_constraint_count == 1, '如果提供单个数值，只支持一个GP模型约束。'
elif constraint_threshold_values is None:
self.constraint_thresholds = None
def _generate_quasi_random_samples(self, n_samples: int, fixed_params: Optional[dict] = None) -> pd.DataFrame:
"""
生成准随机采样点。
"""
raw_samples = self.sobol_sampler(n_samples) # 从[0,1]区间采样
# 映射到实际参数空间
transformed_samples = raw_samples * \
(self.design_space.opt_upper_bound - self.design_space.opt_lower_bound) + \
self.design_space.opt_lower_bound
x_continuous = transformed_samples[:, :self.design_space.numeric_param_count]
x_discrete = transformed_samples[:, self.design_space.numeric_param_count:].long()
# 对离散化后的连续变量进行取整处理
for i, param_name in enumerate(self.design_space.numeric_parameter_names):
if self.design_space.parameters[param_name]['is_discrete_after_transformation']: # 假设参数字典有此键
x_continuous[:, i] = x_continuous[:, i].round()
# 转换回原始参数空间DataFrame
df_samples = self.design_space.inverse_transform(x_continuous, x_discrete)
if fixed_params is not None:
for k, v in fixed_params.items():
df_samples[k] = v
return df_samples
def suggest(self, num_suggestions: int = 1, fixed_params: Optional[dict] = None, rf_model_and_threshold = None) -> pd.DataFrame:
"""
生成下一个建议点的主方法。
"""
if self.acquisition_function_class != AcquisitionFunctionMACE and num_suggestions != 1:
raise RuntimeError('并行优化目前仅支持 AcquisitionFunctionMACE 采集函数。')
if self.X.shape[0] < self.initial_random_samples:
# 初始随机采样阶段
samples = self._generate_quasi_random_samples(num_suggestions, fixed_params)
if self.num_explicit_hard_constraints:
samples = _ensure_hard_constraints_satisfied(samples, self.max_sequence_length)
return samples
else:
# 贝叶斯优化阶段
X_transformed_cont, X_transformed_disc = self.design_space.transform(self.X)
# 对目标值进行幂变换
objective_y_raw = self.y[:, 0:1]
objective_y_transformed = _enable_power_transform_helper(objective_y_raw)
constraint_y_transformed = None
constraint_threshold_transformed = None
if self._gp_constraint_count > 0:
constraint_y_raw = self.y[:, 1:]
# 将约束阈值与历史约束数据拼接，一起进行幂变换
constraint_data_with_threshold = np.concatenate([constraint_y_raw, self.constraint_thresholds], axis = 0)
constraint_data_with_threshold_transformed = _enable_power_transform_helper(constraint_data_with_threshold)
# 合并目标和约束的变换后数据
y_for_model = torch.cat([objective_y_transformed, constraint_data_with_threshold_transformed[:-1, ]], dim=1)
constraint_threshold_transformed = constraint_data_with_threshold_transformed[-1:, ]
else:
y_for_model = objective_y_transformed
# 创建并训练代理模型
surrogate_model = _get_surrogate_model_instance(
self.surrogate_model_type,
self.design_space.numeric_param_count,
self.design_space.categorical_param_count,
y_for_model.shape[1],
**self.model_configuration
)
surrogate_model.fit(X_transformed_cont, X_transformed_disc, y_for_model)
# 获取当前最佳点的预测值
best_history_idx = self._get_best_historical_id(fixed_params)
best_x_df = self.X.iloc[[best_history_idx]]
best_x_cont, best_x_disc = self.design_space.transform(best_x_df)
best_pred_mean, _ = surrogate_model.predict(best_x_cont, best_x_disc)
current_best_pred_obj = best_pred_mean[:, 0].detach().numpy().squeeze()
# 计算自适应 kappa 参数 (探索系数)
current_iter = max(1, self.X.shape[0] // num_suggestions)
upsi_param = 0.5
delta_param = 0.01
adaptive_kappa = np.sqrt(upsi_param * 2 * ((2.0 + self.X.shape[1] / 2.0) * np.log(current_iter) + np.log(3 * np.pi**2 / (3 * delta_param))))
# 创建采集函数实例
acquisition_func = self.acquisition_function_class(
surrogate_model, # 训练好的代理模型
search_space = self.design_space, # 搜索空间
gp_constraint_count = self._gp_constraint_count, # 模型约束数量
num_hard_constr = self.num_explicit_hard_constraints, # 硬约束数量
num_hidden_constr = self.num_learned_hidden_constraints, # 隐藏约束数量
rf_with_thres = rf_model_and_threshold, # 随机森林约束模型
current_best_objective = current_best_pred_obj, # 当前最佳目标值预测
kappa = adaptive_kappa, # 探索系数
constraint_threshold_values = constraint_threshold_transformed, # 变换后的约束阈值
max_sequence_length = self.max_sequence_length
)
# 使用进化算法优化采集函数
evolutionary_optimizer = EvolutionarySearchEngine(self.design_space, acquisition_func,
pop = 100, iters = 100, verbose = False,
es=self.evolutionary_strategy)
candidate_solutions = evolutionary_optimizer.optimize(initial_suggest = best_x_df,
fixed_input = fixed_params).drop_duplicates()
# 过滤掉与历史观测重复的候选点
candidate_solutions = candidate_solutions[self._check_uniqueness(candidate_solutions)]
# 如果建议点数量不足，使用随机采样补充
retry_count = 0
while candidate_solutions.shape[0] < num_suggestions and retry_count < 3:
rand_samples = self._generate_quasi_random_samples(num_suggestions - candidate_solutions.shape[0], fixed_params)
if self.num_explicit_hard_constraints:
rand_samples = _ensure_hard_constraints_satisfied(rand_samples, self.max_sequence_length)
rand_samples = rand_samples[self._check_uniqueness(rand_samples)]
candidate_solutions = pd.concat([candidate_solutions, rand_samples], axis = 0, ignore_index = True)
retry_count += 1
if candidate_solutions.shape[0] < num_suggestions: # 确保数量足够
rand_samples = self._generate_quasi_random_samples(num_suggestions - candidate_solutions.shape[0], fixed_params)
if self.num_explicit_hard_constraints:
rand_samples = _ensure_hard_constraints_satisfied(rand_samples, self.max_sequence_length)
candidate_solutions = pd.concat([candidate_solutions, rand_samples], axis = 0, ignore_index = True)
# 从候选集中选择最终的建议点
selected_indices = np.random.choice(candidate_solutions.shape[0], num_suggestions, replace = False).tolist()
# 额外策略：确保选中预测均值最佳和不确定性最大的点（如果数量允许）
with torch.no_grad():
# 预测所有候选点的目标均值和方差
mean_predictions, variance_predictions = surrogate_model.predict(*self.design_space.transform(candidate_solutions))
predicted_obj_means = mean_predictions[:,0].squeeze().numpy()
predicted_obj_stds = variance_predictions[:,0].sqrt().squeeze().numpy() # 标准差
best_predicted_idx = np.argmin(predicted_obj_means) # 预测均值最小
highest_uncertainty_idx = np.argmax(predicted_obj_stds) # 不确定性最大
if highest_uncertainty_idx not in selected_indices and num_suggestions > 2:
selected_indices[0] = highest_uncertainty_idx
if best_predicted_idx not in selected_indices and num_suggestions > 2:
selected_indices[1] = best_predicted_idx
return candidate_solutions.iloc[selected_indices].copy()
GeneralBayesianOptimizer (用于通用多目标带约束优化)
import pandas as pd
import numpy as np
import torch
from typing import Optional
# 假设 DesignSpace, UniversalAcquisitionFunction, EvolutionarySearchEngine
# _get_hard_constraint_violation, _get_hidden_constraint_violation, _ensure_hard_constraints_satisfied
# 以及 HypervolumeCalculator 类（用于超体积计算）已定义
# 抽象的优化器基类 AbstractOptimizer 已在上面定义并省略重复
# DesignSpace, _get_surrogate_model_instance, _ensure_hard_constraints_satisfied 也已定义并省略重复
# 辅助函数：计算 Pareto 前沿
def _calculate_pareto_front(objectives_data: np.ndarray) -> np.ndarray:
"""
计算给定多目标数据点的Pareto前沿。
假设目标是最小化。
"""
if objectives_data.shape[0] == 0:
return np.array([])
is_pareto = np.ones(objectives_data.shape[0], dtype=bool)
for i in range(objectives_data.shape[0]):
for j in range(objectives_data.shape[0]):
if i == j:
continue
# 如果点j支配点i (j的所有目标都小于等于i，并且至少有一个严格小于)
if np.all(objectives_data[j] <= objectives_data[i]) and np.any(objectives_data[j] < objectives_data[i]):
is_pareto[i] = False
break
return objectives_data[is_pareto]
# 假设 HV 类用于超体积计算
class HypervolumeCalculator:
def __init__(self, ref_point: np.ndarray):
self.ref_point = ref_point
def calculate(self, pareto_front_points: np.ndarray) -> float:
# 模拟超体积计算，实际应使用更复杂的几何方法或库
# 这里简单地用一个虚拟值表示，确保其行为符合预期
if pareto_front_points.shape[0] == 0:
return 0.0
# 假设所有目标都是最小化，参考点是所有目标的上限
# 一个非常简化的超体积估计：产品 (ref_point_i - min(pf_i))
# 实际的超体积计算更为复杂，这里仅为占位符
valid_points = pareto_front_points[~np.isinf(pareto_front_points).any(axis=1)]
if valid_points.shape[0] == 0: return 0.0
# 确保 pareto_front_points 的维度与 ref_point 匹配
if valid_points.shape[1] != self.ref_point.shape[0]:
raise ValueError("Pareto front points dimensions mismatch reference point dimensions.")
# 对每个目标维度，计算其相对于参考点的贡献。
# 这里仅仅是一个简化示例，真实的超体积计算库如 pymoo 或 botorch 有专门实现
contribution = np.prod(self.ref_point - np.min(valid_points, axis=0))
return float(contribution)
class GeneralBayesianOptimizer(AbstractOptimizer):
"""
通用的贝叶斯优化器，支持多目标优化和约束优化。
"""
def __init__(self,
design_space: DesignSpace, # 设计空间
num_objectives: int = 1, # 目标函数数量
num_gp_constraints: int = 0, # 通过高斯过程建模的约束数量
num_explicit_hard_constraints: int = 0, # 硬约束数量
num_learned_hidden_constraints: int = 1, # 隐藏约束数量
max_sequence_len_for_hard_constr: int = 12, # 最大序列长度
initial_random_samples_count: Optional[int] = None, # 随机采样阶段数据量
surrogate_model_type: str = 'multi_task', # 代理模型名称
model_configuration: Optional[dict] = None, # 模型配置参数
objective_explore_weight: Optional[float] = 2., # 目标函数的探索系数
constraint_explore_weight: Optional[float] = 0., # 约束函数的探索系数
enable_prediction_noise: bool = False, # 是否在预测时添加噪声
evolutionary_pop_size: int = 100, # 进化算法种群大小
evolutionary_iterations: int = 200, # 进化算法迭代次数
hypervolume_ref_point: Optional[np.ndarray] = None, # 多目标优化的参考点
sobol_seed: Optional[int] = None # Sobol序列的随机种子
):
super().__init__(design_space)
self.num_objectives = num_objectives
self.num_gp_constraints = num_gp_constraints
self.num_explicit_hard_constraints = num_explicit_hard_constraints
self.num_learned_hidden_constraints = num_learned_hidden_constraints
self.max_sequence_len_for_hard_constr = max_sequence_len_for_hard_constr
self.initial_random_samples = max(2, 1 + self.design_space.total_parameter_count) \
if initial_random_samples_count is None else initial_random_samples_count
self.surrogate_model_type = surrogate_model_type
self.model_configuration = model_configuration if model_configuration is not None else {}
self.y = np.zeros((0, num_objectives + num_gp_constraints))
self.objective_explore_weight = objective_explore_weight
self.constraint_explore_weight = constraint_explore_weight
self.enable_prediction_noise = enable_prediction_noise
self.surrogate_model = None
self.evolutionary_pop_size = evolutionary_pop_size
self.evolutionary_iterations = evolutionary_iterations
self.iteration_counter = 0
self.hypervolume_ref_point = hypervolume_ref_point
self.sobol_sampler = lambda n: torch.rand(n, self.design_space.total_parameter_count) # 模拟Sobol采样
if num_objectives + num_gp_constraints > 1:
# 假设 _get_surrogate_model_class 能够返回模型类并检查其多输出支持
pass # 简化处理，跳过实际检查
def _generate_quasi_random_samples(self, n_samples: int, fixed_params: Optional[dict] = None) -> pd.DataFrame:
"""
生成准随机采样点。
"""
raw_samples = self.sobol_sampler(n_samples)
transformed_samples = raw_samples * \
(self.design_space.opt_upper_bound - self.design_space.opt_lower_bound) + \
self.design_space.opt_lower_bound
x_continuous = transformed_samples[:, :self.design_space.numeric_param_count]
x_discrete = transformed_samples[:, self.design_space.numeric_param_count:].long()
for i, param_name in enumerate(self.design_space.numeric_parameter_names):
if self.design_space.parameters[param_name]['is_discrete_after_transformation']: # 假设参数字典有此键
x_continuous[:, i] = x_continuous[:, i].round()
df_samples = self.design_space.inverse_transform(x_continuous, x_discrete)
if fixed_params is not None:
for k, v in fixed_params.items():
df_samples[k] = v
return df_samples
def suggest(self, num_suggestions: int = 1, fixed_params: Optional[dict] = None, rf_model_and_threshold = None) -> pd.DataFrame:
"""
生成建议点的主方法。
"""
self.iteration_counter += 1
if self.X.shape[0] < self.initial_random_samples:
samples = self._generate_quasi_random_samples(num_suggestions, fixed_params)
if self.num_explicit_hard_constraints:
samples = _ensure_hard_constraints_satisfied(samples, self.max_sequence_len_for_hard_constr)
return samples
else:
X_transformed_cont, X_transformed_disc = self.design_space.transform(self.X)
y_observations = torch.FloatTensor(self.y)
num_unique_categories = None
if X_transformed_disc.shape[1] > 0:
num_unique_categories = [len(self.design_space.parameters[name]['categories'])
for name in self.design_space.categorical_parameter_names]
self.surrogate_model = _get_surrogate_model_instance(
self.surrogate_model_type, X_transformed_cont.shape[1], X_transformed_disc.shape[1],
y_observations.shape[1], num_unique_cats=num_unique_categories, **self.model_configuration
)
self.surrogate_model.fit(X_transformed_cont, X_transformed_disc, y_observations)
# 设置探索系数：如果未提供则根据理论公式计算自适应值
adaptive_obj_explore_weight = self.objective_explore_weight
adaptive_constr_explore_weight = self.constraint_explore_weight
upsi_param = 0.1
delta_param = 0.01
if adaptive_obj_explore_weight is None:
adaptive_obj_explore_weight = np.sqrt(upsi_param * 2 * ((2.0 + self.X.shape[1] / 2.0) * np.log(self.iteration_counter) + np.log(3 * np.pi**2 / (3 * delta_param))))
if adaptive_constr_explore_weight is None:
adaptive_constr_explore_weight = np.sqrt(upsi_param * 2 * ((2.0 + self.X.shape[1] / 2.0) * np.log(self.iteration_counter) + np.log(3 * np.pi**2 / (3 * delta_param))))
acquisition_func = UniversalAcquisitionFunction(
self.surrogate_model, # 训练好的代理模型
self.num_objectives, # 目标函数数量
self.num_gp_constraints, # 模型约束数量
num_hard_constr = self.num_explicit_hard_constraints, # 硬约束数量
num_hidden_constr = self.num_learned_hidden_constraints, # 隐藏约束数量
max_sequence_length = self.max_sequence_len_for_hard_constr, # 序列长度限制
rf_with_thres = rf_model_and_threshold, # 随机森林约束模型
space = self.design_space, # 搜索空间
kappa = adaptive_obj_explore_weight, # 目标探索系数
c_kappa = adaptive_constr_explore_weight, # 约束探索系数
use_noise = self.enable_prediction_noise
)
evolutionary_optimizer = EvolutionarySearchEngine(self.design_space, acquisition_func,
pop=self.evolutionary_pop_size,
iters=self.evolutionary_iterations)
candidate_suggestions = evolutionary_optimizer.optimize()
if candidate_suggestions.shape[0] < num_suggestions:
rand_samples = self._generate_quasi_random_samples(num_suggestions - candidate_suggestions.shape[0], fixed_params)
if self.num_explicit_hard_constraints:
rand_samples = _ensure_hard_constraints_satisfied(rand_samples, self.max_sequence_len_for_hard_constr)
candidate_suggestions = pd.concat([candidate_suggestions, rand_samples], axis=0, ignore_index=True)
return candidate_suggestions
elif self.hypervolume_ref_point is None:
# 没有参考点时，随机选择，并确保包含不确定性最大的点
with torch.no_grad():
_, predicted_variances = self.surrogate_model.predict(*self.design_space.transform(candidate_suggestions))
# 找到总对数方差最大的点，代表不确定性最高
highest_uncertainty_idx = np.argmax(np.log(predicted_variances).sum(axis=1))
selected_indices = np.random.choice(candidate_suggestions.shape[0], num_suggestions, replace=False).tolist()
if highest_uncertainty_idx not in selected_indices:
selected_indices[0] = highest_uncertainty_idx
return candidate_suggestions.iloc[selected_indices]
else:
# 多目标优化且有参考点：使用EHVI（期望超体积改进）进行选择
assert self.num_objectives > 1
assert self.num_gp_constraints == 0 # EHVI在此简化版本中不支持GP约束
mc_samples_count = 10
hv_calculator = HypervolumeCalculator(ref_point=self.hypervolume_ref_point.reshape(-1))
with torch.no_grad():
predicted_mean, predicted_variance = self.surrogate_model.predict(*self.design_space.transform(candidate_suggestions))
# 从后验分布中采样
sampled_objectives = self.surrogate_model.sample_y(*self.design_space.transform(candidate_suggestions), mc_samples_count).numpy()
current_pareto_front = _calculate_pareto_front(self.y[:, :self.num_objectives]).copy()
selected_indices = []
for _ in range(num_suggestions):
ehvi_values = []
base_hv = hv_calculator.calculate(current_pareto_front)
for j in range(candidate_suggestions.shape[0]):
candidate_mc_samples = sampled_objectives[:, j, :self.num_objectives]
estimated_hvi = 0
for k in range(mc_samples_count):
temp_pareto_front = np.vstack([current_pareto_front, candidate_mc_samples[[k]]])
estimated_hvi += hv_calculator.calculate(temp_pareto_front) - base_hv
estimated_hvi /= mc_samples_count
ehvi_values.append(estimated_hvi)
best_candidate_idx = np.argmax(ehvi_values) if max(ehvi_values) > 0 else np.random.choice(candidate_suggestions.shape[0])
# 将选择点的最佳样本（此处的"最佳"可能需要更精细定义，此处简化为均值）
# 添加到当前Pareto前沿，以模拟下一轮的决策
current_pareto_front = np.vstack([current_pareto_front, predicted_mean[best_candidate_idx, :self.num_objectives].detach().numpy().reshape(1, -1)])
selected_indices.append(best_candidate_idx)
selected_indices = list(set(selected_indices))
if len(selected_indices) < num_suggestions:
remaining_candidates = [i for i in range(candidate_suggestions.shape[0]) if i not in selected_indices]
selected_indices.extend(np.random.choice(remaining_candidates, num_suggestions - len(selected_indices), replace=False).tolist())
return candidate_suggestions.iloc[selected_indices]
优化器对比
ConstrainedBayesianOptimizer(HEBOConstr)：- 专注于单目标带约束优化。
- 默认使用
AcquisitionFunctionMACE采集函数，结合了LCB、EI、PI三种准则。 - 对目标函数输出进行幂变换以改善高斯过程模型的拟合。
- 通过适应性 kappa 参数调整探索强度。
GeneralBayesianOptimizer(GeneralBO)：- 面向通用多目标带约束优化。
- 默认使用
UniversalAcquisitionFunction，为所有目标和GP约束计算LCB。 - 通常不进行目标函数输出的幂变换。
- 在多目标场景下，如果提供了参考点，会采用EHVI（Expected Hypervolume Improvement）策略来选择候选点，否则倾向于选择不确定性最高的点。
采集函数的优化：进化算法
采集函数通常是一个复杂且多模态的函数，直接求其最大值是困难的。HEBO框架通过进化算法来高效地优化采集函数，从而找到下一个最有前景的候选点。
import pandas as pd
import numpy as np
import torch
from typing import Optional
# 假设 DesignSpace, BaseAcquisition 是已定义的类
# 模拟 PyMoo库的 GA, NSGA2, MixedVariableGA, MixedVariableMating, MixedVariableDuplicateElimination, minimize 已导入
# 为了简化，这里直接定义或模拟 pymoo 相关的类和函数
class MockPymooProblem:
def __init__(self, n_var, n_obj, n_constr, xl, xu, vtypes):
self.n_var = n_var
self.n_obj = n_obj
self.n_constr = n_constr
self.xl = xl
self.xu = xu
self.vtypes = vtypes
def _evaluate(self, X_eval, out, *args, **kwargs):
raise NotImplementedError
class MockPymooAlgorithm:
def __init__(self, pop_size, sampling, crossover=None, mutation=None, mating=None, eliminate_duplicates=None):
self.pop_size = pop_size
self.sampling = sampling
self.crossover = crossover
self.mutation = mutation
self.mating = mating
self.eliminate_duplicates = eliminate_duplicates
self.has_been_initialized = False # 模拟状态
def setup(self, problem, seed=None, **kwargs):
# 模拟 pymoo 算法的 setup 方法
self.problem = problem
self.pop = self.sampling[:self.pop_size] # 初始种群
self.has_been_initialized = True
return self
def run(self):
# 模拟运行，返回一个带有 X 和 pop 属性的结果对象
class MockResult:
def __init__(self, X, pop):
self.X = X
self.pop = pop
# 模拟生成一个结果，可以是随机的或从初始种群中选择
if isinstance(self.sampling, np.ndarray) and self.sampling.shape[0] > 0:
best_X = self.sampling[0] # 简单选择第一个
# 模拟生成一个种群，这里就用初始种群
mock_pop_elements = []
for i in range(min(self.pop_size, self.sampling.shape[0])):
class MockIndividual:
def __init__(self, x_val):
self.X = x_val
mock_pop_elements.append(MockIndividual(self.sampling[i]))
return MockResult(best_X, mock_pop_elements)
else:
return MockResult(None, [])
# 模拟 pymoo 的 optimize.minimize 函数
def mock_minimize(problem, algorithm, termination, verbose=False):
# 模拟算法 setup 和 run
algorithm.setup(problem)
result = algorithm.run()
# 模拟 pymoo 的迭代次数限制
for _ in range(termination[1] - 1): # 运行 n_gen-1 世代
# 在实际 pymoo 中，这里会有更复杂的迭代逻辑
pass
return result
# 模拟 pymoo 的一些算子
class MockSBX:
def __init__(self, prob, eta): pass
class MockPM:
def __init__(self, prob, eta): pass
class MockMixedVariableMating:
def __init__(self, eliminate_duplicates): pass
class MockMixedVariableDuplicateElimination:
def __init__(self): pass
class MockNonpGA(MockPymooAlgorithm): # 假设NonpGA是pymoo中处理混合变量的GA
pass
class MockNSGA2(MockPymooAlgorithm): # 假设NSGA2是pymoo中处理混合变量的NSGA2
pass
# 采集函数问题包装器
class AcquisitionProblem(MockPymooProblem):
def __init__(self, acquisition_func: BaseAcquisition, design_space: DesignSpace, fixed_params: Optional[dict]):
self.acquisition_func = acquisition_func
self.design_space = design_space
self.fixed_params = fixed_params
# 获取参数空间的上下界
lower_bounds = design_space.opt_lower_bound.numpy()
upper_bounds = design_space.opt_upper_bound.numpy()
# 确定变量类型：连续变量为'real'，离散变量为'int'
variable_types = ['real'] * design_space.numeric_param_count + \
['int'] * design_space.categorical_param_count
super().__init__(
n_var=design_space.total_parameter_count,
n_obj=acquisition_func.objective_count,
n_constr=acquisition_func.total_constraint_count,
xl=lower_bounds,
xu=upper_bounds,
vtypes=variable_types # 假设pymoo支持 vtypes
)
def _evaluate(self, X_eval: np.ndarray, out: dict, *args, **kwargs):
# 将pymoo的numpy输入转换为torch tensor
X_eval_torch = torch.from_numpy(X_eval).float()
# 分离连续和离散部分
x_continuous = X_eval_torch[:, :self.design_space.numeric_param_count]
x_discrete = X_eval_torch[:, self.design_space.numeric_param_count:].long()
# 调用采集函数进行评估
results = self.acquisition_func.evaluate(x_continuous, x_discrete).detach().numpy()
# 赋值给pymoo的输出
out["F"] = results[:, :self.acquisition_func.objective_count] # 目标函数值
if self.acquisition_func.total_constraint_count > 0:
out["G"] = results[:, self.acquisition_func.objective_count:] # 约束函数值 (需要小于0)
# 辅助函数：初始化种群
def _get_initial_population(design_space: DesignSpace, pop_size: int, initial_suggest_df: Optional[pd.DataFrame], use_sobol: bool):
if initial_suggest_df is not None and not initial_suggest_df.empty:
# 如果提供了初始建议点，则将其转换为模型内部表示
initial_pop_numpy = initial_suggest_df[design_space.parameter_names].values
# 如果提供的少于pop_size，则补充
if initial_pop_numpy.shape[0] < pop_size:
remaining_samples = pop_size - initial_pop_numpy.shape[0]
if use_sobol:
sobol_sampler = lambda n: torch.rand(n, design_space.total_parameter_count)
random_part_raw = sobol_sampler(remaining_samples)
else:
random_part_raw = torch.rand(remaining_samples, design_space.total_parameter_count)
# 映射到参数空间
random_part_transformed = random_part_raw * (design_space.opt_upper_bound - design_space.opt_lower_bound) + design_space.opt_lower_bound
x_cont_rand = random_part_transformed[:, :design_space.numeric_param_count]
x_disc_rand = random_part_transformed[:, design_space.numeric_param_count:].long()
for i, name in enumerate(design_space.numeric_parameter_names):
if design_space.parameters[name]['is_discrete_after_transformation']:
x_cont_rand[:, i] = x_cont_rand[:, i].round()
df_rand = design_space.inverse_transform(x_cont_rand, x_disc_rand)
initial_pop_numpy = np.vstack([initial_pop_numpy, df_rand[design_space.parameter_names].values])
return initial_pop_numpy[:pop_size] # 确保返回pop_size个
else:
# 否则生成全新的初始种群
if use_sobol:
sobol_sampler = lambda n: torch.rand(n, design_space.total_parameter_count)
raw_samples = sobol_sampler(pop_size)
else:
raw_samples = torch.rand(pop_size, design_space.total_parameter_count)
transformed_samples = raw_samples * (design_space.opt_upper_bound - design_space.opt_lower_bound) + design_space.opt_lower_bound
x_cont = transformed_samples[:, :design_space.numeric_param_count]
x_disc = transformed_samples[:, design_space.numeric_param_count:].long()
for i, name in enumerate(design_space.numeric_parameter_names):
if design_space.parameters[name]['is_discrete_after_transformation']:
x_cont[:, i] = x_cont[:, i].round()
df_pop = design_space.inverse_transform(x_cont, x_disc)
return df_pop[design_space.parameter_names].values
class EvolutionarySearchEngine:
"""
进化算法优化器：使用进化算法优化采集函数，寻找最优候选点。
支持单目标和多目标优化，处理混合变量（连续和离散）优化问题。
"""
def __init__(self,
param_space: DesignSpace, # 设计空间
acquisition_func_instance: BaseAcquisition, # 采集函数实例
evolutionary_algo_type : str = None, # 进化算法类型，None时自动选择
**config): # 其他配置参数
self.param_space = param_space
self.evolutionary_algo_type = evolutionary_algo_type
self.acquisition_func_instance = acquisition_func_instance
self.population_size = config.get('pop', 100)
self.num_generations = config.get('iters',500)
self.verbose_output = config.get('verbose', False)
self.repair_operator = config.get('repair', None)
self.use_sobol_init = config.get('sobol_init', True)
assert(self.acquisition_func_instance.objective_count > 0), "采集函数必须至少定义一个目标。"
if self.evolutionary_algo_type is None:
# 单目标使用GA，多目标使用NSGA-II
self.evolutionary_algo_type = 'nsga2' if self.acquisition_func_instance.objective_count > 1 else 'ga'
def optimize(self, initial_candidate_df : Optional[pd.DataFrame] = None, fixed_input_params : Optional[dict] = None, return_full_population: bool = False) -> pd.DataFrame:
"""
执行进化算法优化过程。
"""
# 创建一个 pymoo Problem 实例来包装采集函数
optimization_problem = AcquisitionProblem(self.acquisition_func_instance, self.param_space, fixed_input_params)
# 初始化种群
initial_pop_as_numpy = _get_initial_population(self.param_space, self.population_size,
initial_candidate_df, self.use_sobol_init)
# 根据目标数量选择进化算法
if self.acquisition_func_instance.objective_count == 1:
# 单目标优化：使用混合变量遗传算法
algorithm = MockNonpGA(
pop_size = self.population_size,
sampling = initial_pop_as_numpy,
crossover = MockSBX(prob=0.9, eta=15), # 示例交叉算子
mutation = MockPM(prob=1.0/self.param_space.total_parameter_count, eta=20), # 示例变异算子
eliminate_duplicates = MockMixedVariableDuplicateElimination()
)
else:
# 多目标优化：使用NSGA-II算法
algorithm = MockNSGA2(
pop_size = self.population_size,
sampling = initial_pop_as_numpy,
mating = MockMixedVariableMating(eliminate_duplicates = MockMixedVariableDuplicateElimination()),
eliminate_duplicates = MockMixedVariableDuplicateElimination()
)
# 执行进化算法
optimization_result = mock_minimize(optimization_problem, algorithm,
('n_gen', self.num_generations), verbose = self.verbose_output)
# 处理优化结果
optimized_x_numpy = None
if optimization_result.X is not None and not return_full_population:
# 如果找到了最优解且不需要返回整个种群
# pymoo的X可能是字典或numpy数组
if isinstance(optimization_result.X, dict):
# 将字典转换为 DataFrame 统一处理
optimized_x_df_temp = pd.DataFrame([optimization_result.X])
else:
optimized_x_df_temp = pd.DataFrame(optimization_result.X)
# 确保列顺序正确，并转换为 numpy 浮点数组
optimized_x_numpy = optimized_x_df_temp[self.param_space.parameter_names].values.astype(float)
else:
# 返回整个种群或当没有明确最优解时
# 从种群中提取所有个体的参数
optimized_x_df_temp = pd.DataFrame([p.X for p in optimization_result.pop])
optimized_x_numpy = optimized_x_df_temp[self.param_space.parameter_names].values.astype(float)
if self.acquisition_func_instance.objective_count == 1 and not return_full_population:
# 单目标且不返回整个种群时，从中随机选择一个
optimized_x_numpy = optimized_x_numpy[[np.random.choice(optimized_x_numpy.shape[0])]]
# 保存优化结果（可选用于调试）
self.optimization_result = optimization_result
# 分离连续变量和离散变量
optimized_x_cont_torch = torch.from_numpy(optimized_x_numpy[:, :self.param_space.numeric_param_count]).float()
optimized_x_enum_torch = torch.from_numpy(optimized_x_numpy[:, self.param_space.numeric_param_count:]).long()
# 将优化结果转换回原始参数空间
final_suggested_df = self.param_space.inverse_transform(optimized_x_cont_torch, optimized_x_enum_torch)
# 如果有固定输入参数，覆盖优化结果
if fixed_input_params is not None:
for k, v in fixed_input_params.items():
final_suggested_df[k] = v
return final_suggested_df
工作原理：进化优化器将采集函数视为一个多目标或单目标优化问题。它使用诸如GA（遗传算法）或NSGA-II（非支配排序遗传算法II）等进化算法，在参数空间中迭代地生成、评估和选择候选解。每个候选解都由采集函数打分，高分的解被视为更有前景的采样点。进化算法的优点在于能够有效地搜索复杂、非凸的函数景观，并自然地处理连续和离散混合变量的优化问题。
通过这些精巧设计的采集函数和强大的优化器组合，HEBO框架实现了对复杂黑盒优化问题的鲁棒和高效求解。
