主要包含两个部分
一个代理模型（surrogate model），用于对目标函数进行建模。代理模型通常有确定的公式或者能计算梯度，又或者有已知的凹凸性、线性等特性，总之就是更容易用于优化。更泛化地讲，其实它就是一个学习模型，输入是所有观测到的函数值点，训练后可以在给定任意x的情况下给出对f(x)的估计。
一个优化策略（optimization strategy），决定下一个采样点的位置，即下一步应在哪个输入x
处观测函数值f(x)。通常它是通过采集函数（acquisition function） 来实现的：
采集函数通常是一个由代理模型推出的函数，它的输入是可行集（feasible set）A上的任意值，输出值衡量了每个输入x有多值得被观测。通常会从以下两方面考虑：
有多大的可能性在x处取得最优值
评估x是否能减少贝叶斯统计模型的不确定性
采集函数通常也是容易求最优值的函数（例如：有公式/能算梯度等），下一个采样点就是可行集上的最大值点，即使采集函数的取最大值的点。
本文主要学习代理模型，模型代码参考SCOOT(WWW2025 oral)中hebo/model库的代码实现
常用的贝叶斯代理模型
GP：标准高斯过程
class GP(BaseModel):
# support_grad = True 表示该模型支持梯度计算，这对于贝叶斯优化等需要梯度信息的场景非常重要
support_grad = True
def __init__(self, num_cont, num_enum, num_out, **conf):
"""
初始化高斯过程回归模型
参数:
num_cont (int): 连续变量的数量
num_enum (int): 离散/枚举变量的数量
num_out (int): 输出变量的数量（通常为1，表示单目标优化）
**conf: 配置参数字典，可包含以下键值对：
- lr (float): 学习率，默认为3e-2
- num_epochs (int): 训练轮数，默认为100
- verbose (bool): 是否打印训练过程信息，默认为False
- print_every (int): 打印训练信息的频率，默认为10
- pred_likeli (bool): 预测时是否考虑似然噪声，默认为True
- noise_lb (float): 噪声下界，默认为1e-5
- optimizer (str): 优化器类型，可选'lbfgs'、'psgld'或'adam'，默认为'psgld'
- noise_guess (float): 噪声初始猜测值，默认为0.01
- ard_kernel (bool): 是否使用ARD核（自动相关性判定），默认为True
"""
# 调用父类BaseModel的初始化方法
super().__init__(num_cont, num_enum, num_out, **conf)
# 从配置中提取参数，如果未提供则使用默认值
self.lr = conf.get('lr', 3e-2) # 学习率
self.num_epochs = conf.get('num_epochs', 100) # 训练轮数
self.verbose = conf.get('verbose', False) # 是否打印训练过程信息
self.print_every = conf.get('print_every', 10) # 打印训练信息的频率
self.pred_likeli = conf.get('pred_likeli', True) # 预测时是否考虑似然噪声
self.noise_lb = conf.get('noise_lb', 1e-5) # 噪声下界
self.optimizer = conf.get('optimizer', 'psgld') # 优化器类型
self.noise_guess = conf.get('noise_guess', 0.01) # 噪声初始猜测值
self.ard_kernel = conf.get('ard_kernel', True) # 是否使用ARD核
# 初始化数据标准化器
# xscaler: 用于连续变量的最小-最大标准化器，将数据缩放到[-1, 1]区间
self.xscaler = TorchMinMaxScaler((-1, 1))
# yscaler: 用于目标变量的标准标准化器，使其均值为0，标准差为1
self.yscaler = TorchStandardScaler()
def fit_scaler(self, Xc : Tensor, Xe : Tensor, y : Tensor):
"""
拟合数据标准化器
参数:
Xc: 连续特征张量
Xe: 离散特征张量
y: 目标变量张量
"""
# 如果存在连续变量且数量大于0，则拟合连续变量标准化器
if Xc is not None and Xc.shape[1] > 0:
self.xscaler.fit(Xc)
# 拟合目标变量标准化器
self.yscaler.fit(y)
def xtrans(self, Xc : Tensor, Xe : Tensor, y : Tensor = None):
"""
对输入数据进行转换（标准化和类型转换）
参数:
Xc: 连续特征张量
Xe: 离散特征张量
y: 可选的目标变量张量
返回:
转换后的特征和目标变量（如果提供了y）
"""
# 处理连续变量：如果存在则标准化，否则创建空张量
if Xc is not None and Xc.shape[1] > 0:
Xc_t = self.xscaler.transform(Xc) # 标准化连续变量
else:
Xc_t = torch.zeros(Xe.shape[0], 0) # 创建空的连续变量张量
# 处理离散变量：如果存在则转换为long类型，否则创建空张量
if Xe is None:
Xe_t = torch.zeros(Xc.shape[0], 0).long() # 创建空的离散变量张量
else:
Xe_t = Xe.long() # 转换为long类型，通常用于嵌入层
# 如果提供了目标变量，则也进行标准化
if y is not None:
y_t = self.yscaler.transform(y) # 标准化目标变量
return Xc_t, Xe_t, y_t
else:
return Xc_t, Xe_t
def fit(self, Xc : Tensor, Xe : Tensor, y : Tensor):
"""
训练高斯过程模型
参数:
Xc: 连续特征张量
Xe: 离散特征张量
y: 目标变量张量
"""
# 1. 数据预处理：过滤NaN值
Xc, Xe, y = filter_nan(Xc, Xe, y, 'all')
# 2. 拟合标准化器并转换数据
self.fit_scaler(Xc, Xe, y)
Xc, Xe, y = self.xtrans(Xc, Xe, y)
# 3. 验证数据维度
assert(Xc.shape[1] == self.num_cont)
assert(Xe.shape[1] == self.num_enum)
assert(y.shape[1] == self.num_out)
# 4. 保存训练数据
self.Xc = Xc
self.Xe = Xe
self.y = y
# 5. 设置似然函数（噪声模型）
n_constr = GreaterThan(self.noise_lb) # 噪声下界约束
n_prior = LogNormalPrior(np.log(self.noise_guess), 0.5) # 对数正态先验
# 创建高斯似然函数，包含噪声约束和先验
self.lik = GaussianLikelihood(noise_constraint = n_constr, noise_prior = n_prior)
# 6. 创建GPyTorch模型
self.gp = GPyTorchModel(self.Xc, self.Xe, self.y, self.lik, **self.conf)
# 7. 初始化噪声参数
self.gp.likelihood.noise = max(1e-2, self.noise_lb)
# 8. 设置模型为训练模式
self.gp.train()
self.lik.train()
# 9. 选择优化器
if self.optimizer.lower() == 'lbfgs':
# L-BFGS优化器，适用于小批量数据的准牛顿方法
opt = torch.optim.LBFGS(self.gp.parameters(), lr = self.lr, max_iter = 5, line_search_fn = 'strong_wolfe')
elif self.optimizer == 'psgld':
# 预处理随机梯度Langevin动力学，用于贝叶斯采样
opt = pSGLD(self.gp.parameters(), lr = self.lr, factor = 1. / y.shape[0], pretrain_step = self.num_epochs // 10)
else:
# Adam优化器，默认选择
opt = torch.optim.Adam(self.gp.parameters(), lr = self.lr)
# 10. 创建边际对数似然目标函数
mll = gpytorch.mlls.ExactMarginalLogLikelihood(self.lik, self.gp)
# 11. 训练循环
for epoch in range(self.num_epochs):
def closure():
"""
优化器的闭包函数，计算损失并梯度
"""
# 前向传播：获取GP的后验分布
dist = self.gp(self.Xc, self.Xe)
# 计算负边际对数似然损失
loss = -1 * mll(dist, self.y.squeeze())
# 清零梯度
opt.zero_grad()
# 反向传播
loss.backward()
return loss
# 优化器步进
opt.step(closure)
# 打印训练信息
if self.verbose and ((epoch + 1) % self.print_every == 0 or epoch == 0):
print('After %d epochs, loss = %g' % (epoch + 1, closure().item()), flush = True)
# 12. 设置模型为评估模式
self.gp.eval()
self.lik.eval()
def predict(self, Xc, Xe):
"""
使用训练好的模型进行预测
参数:
Xc: 连续特征张量
Xe: 离散特征张量
返回:
mu: 预测均值
var: 预测方差
"""
# 1. 转换输入数据
Xc, Xe = self.xtrans(Xc, Xe)
# 2. 使用快速预测设置进行预测
with gpytorch.settings.fast_pred_var(), gpytorch.settings.debug(False):
# 获取GP预测
pred = self.gp(Xc, Xe)
# 如果考虑似然噪声，通过似然函数进行预测
if self.pred_likeli:
pred = self.lik(pred)
# 提取均值和方差
mu_ = pred.mean.reshape(-1, self.num_out)
var_ = pred.variance.reshape(-1, self.num_out)
# 3. 将预测结果转换回原始尺度
mu = self.yscaler.inverse_transform(mu_) # 逆标准化均值
var = var_ * self.yscaler.std**2 # 调整方差到原始尺度
# 4. 确保方差不为零（避免数值问题）
return mu, var.clamp(min = torch.finfo(var.dtype).eps)
def sample_y(self, Xc, Xe, n_samples = 1) -> FloatTensor:
"""
从后验分布中采样目标变量
参数:
Xc: 连续特征张量
Xe: 离散特征张量
n_samples: 采样数量
返回:
采样结果，形状为(n_samples, 样本数, 输出维度)
"""
# 1. 转换输入数据
Xc, Xe = self.xtrans(Xc, Xe)
# 2. 进行采样
with gpytorch.settings.debug(False):
if self.pred_likeli:
# 考虑噪声的采样
pred = self.lik(self.gp(Xc, Xe))
else:
# 不考虑噪声的采样（函数值采样）
pred = self.gp(Xc, Xe)
# 从后验分布中采样
samp = pred.rsample(torch.Size((n_samples,))).view(n_samples, Xc.shape[0], self.num_out)
# 将采样结果转换回原始尺度
return self.yscaler.inverse_transform(samp)
def sample_f(self):
"""
采样函数值（不支持，使用sample_y代替）
"""
raise NotImplementedError('Thompson sampling is not supported for GP, use `sample_y` instead')
@property
def noise(self):
"""
获取估计的噪声水平（转换回原始尺度）
"""
return (self.gp.likelihood.noise * self.yscaler.std**2).view(self.num_out).detach()
class GPyTorchModel(gpytorch.models.ExactGP):
"""
GPyTorch模型实现类，继承自ExactGP（精确高斯过程）
"""
def __init__(self,
x : torch.Tensor, # 训练数据的连续特征
xe : torch.Tensor, # 训练数据的离散特征
y : torch.Tensor, # 训练目标值
lik : GaussianLikelihood, # 似然函数
**conf): # 配置参数
# 调用父类构造函数，传入训练数据和似然函数
super().__init__((x, xe), y.squeeze(), lik)
# 特征提取器：处理连续和离散特征的组合
self.fe = deepcopy(conf.get('fe', DummyFeatureExtractor(x.shape[1], xe.shape[1], conf.get('num_uniqs'), conf.get('emb_sizes'))))
# 均值函数：默认使用常数均值
self.mean = deepcopy(conf.get('mean', ConstantMean()))
# 核函数选择：根据配置选择默认核或随机分解核
if conf.get("rd", False):
# 使用随机分解核（Random Decomposition）
self.cov = deepcopy(conf.get('kern', default_kern_rd(x, xe, y, self.fe.total_dim, conf.get('ard_kernel', True), conf.get('fe'), E=conf.get("E", 0.2))))
else:
# 使用默认核函数
self.cov = deepcopy(conf.get('kern', default_kern(x, xe, y, self.fe.total_dim, conf.get('ard_kernel', True), conf.get('fe'))))
def forward(self, x, xe):
"""
前向传播，定义高斯过程的行为
参数:
x: 连续特征
xe: 离散特征
返回:
MultivariateNormal: 多元正态分布，表示高斯过程的预测
"""
# 1. 特征提取：将连续和离散特征组合成统一表示
x_all = self.fe(x, xe)
# 2. 计算均值函数
m = self.mean(x_all)
# 3. 计算协方差矩阵（核函数）
K = self.cov(x_all)
# 4. 返回多元正态分布
return MultivariateNormal(m, K)
GPyGP:输入扭曲的高斯过程模型
class GPyGP(BaseModel):
"""
Input warped GP model implemented using GPy instead of GPyTorch
使用GPy库实现的输入扭曲高斯过程模型（而非GPyTorch）
Why doing so: 为什么这样做：
- Input warped GP 支持输入扭曲的高斯过程
"""
def __init__(self, num_cont, num_enum, num_out, **conf):
super().__init__(num_cont, num_enum, num_out, **conf) # 调用父类初始化
total_dim = num_cont # 总维度初始化为连续变量数量
if num_enum > 0: # 如果存在离散变量
self.one_hot = OneHotTransform(self.conf['num_uniqs']) # 创建one-hot编码器
total_dim += self.one_hot.num_out # 增加离散变量编码后的维度
self.xscaler = TorchMinMaxScaler((-1, 1)) # 连续变量标准化器，范围[-1,1]
self.yscaler = TorchStandardScaler() # 目标变量标准化器
self.verbose = self.conf.get('verbose', False) # 是否显示训练信息
self.num_epochs = self.conf.get('num_epochs', 200) # 训练轮数
self.warp = self.conf.get('warp', True) # 是否使用输入扭曲
self.space = self.conf.get('space') # 设计空间定义
self.num_restarts = self.conf.get('num_restarts', 10) # 优化重启次数
self.rd = self.conf.get('rd', False) # 是否使用随机分解
self.E = self.conf.get('E', 0.2) # 随机分解的边概率
if self.space is None and self.warp: # 如果没有设计空间但启用了扭曲
warnings.warn('Space not provided, set warp to False') # 发出警告
self.warp = False # 禁用扭曲
if self.warp: # 如果启用扭曲
for i in range(total_dim): # 为每个维度禁用日志记录器
logging.getLogger(f'a{i}').disabled = True # 禁用参数a的日志
logging.getLogger(f'b{i}').disabled = True # 禁用参数b的日志
def fit_scaler(self, Xc : FloatTensor, y : FloatTensor):
if Xc is not None and Xc.shape[1] > 0: # 如果存在连续变量
if self.space is not None: # 如果有设计空间信息
cont_lb = self.space.opt_lb[:self.space.num_numeric].view(1, -1).float() # 获取连续变量下界
cont_ub = self.space.opt_ub[:self.space.num_numeric].view(1, -1).float() # 获取连续变量上界
self.xscaler.fit(torch.cat([Xc, cont_lb, cont_ub], dim = 0)) # 结合数据和边界来拟合标准化器
else:
self.xscaler.fit(Xc) # 仅使用数据拟合标准化器
self.yscaler.fit(y) # 拟合目标变量标准化器
def trans(self, Xc : Tensor, Xe : Tensor, y : Tensor = None):
if Xc is not None and Xc.shape[1] > 0: # 处理连续变量
Xc_t = self.xscaler.transform(Xc) # 标准化连续变量
else:
Xc_t = torch.zeros(Xe.shape[0], 0) # 创建空的连续变量张量
if Xe is None or Xe.shape[1] == 0: # 处理离散变量
Xe_t = torch.zeros(Xc.shape[0], 0) # 创建空的离散变量张量
else:
Xe_t = self.one_hot(Xe.long()) # one-hot编码离散变量
Xall = torch.cat([Xc_t, Xe_t], dim = 1) # 合并连续和离散特征
if y is not None: # 如果提供了目标变量
y_t = self.yscaler.transform(y) # 标准化目标变量
return Xall.numpy(), y_t.numpy() # 转换为numpy数组返回
return Xall.numpy() # 只返回特征数据
def fit(self, Xc : FloatTensor, Xe : LongTensor, y : LongTensor):
Xc, Xe, y = filter_nan(Xc, Xe, y, 'all') # 过滤NaN值
self.fit_scaler(Xc, y) # 拟合标准化器
X, y = self.trans(Xc, Xe, y) # 转换数据格式
if self.rd: # 如果使用随机分解
cliques = get_random_graph(X.shape[1], self.E) # 生成随机图团结构
# process first clique 处理第一个团
pair = cliques[0] # 获取第一个团
k1 = GPy.kern.Linear(len(pair), active_dims=pair, ARD = False) # 线性核
k2 = GPy.kern.Matern32(len(pair), active_dims=pair, ARD = True) # Matern32核，启用ARD
k2.lengthscale = np.std(X, axis = 0)[pair] # 设置长度尺度为特征标准差
k2.variance = 0.5 # 设置核方差
k2.variance.set_prior(GPy.priors.Gamma(0.5, 1)) # 设置方差先验为Gamma分布
kern = k1 + k2 # 线性核 + Matern核
# process remaining cliques 处理剩余团
for pair in cliques[1:]:
k1 = GPy.kern.Linear(len(pair), active_dims=pair, ARD = False) # 线性核
k2 = GPy.kern.Matern32(len(pair), active_dims=pair, ARD = True) # Matern32核
geo_mean = 1 # 初始化几何均值
for d in pair: # 计算团内特征的几何标准差
geo_mean *= np.std(X, axis = 0)[d]
k2.lengthscale = geo_mean**(1/len(pair)) # 设置长度尺度为几何均值
k2.variance = 0.5 # 设置核方差
k2.variance.set_prior(GPy.priors.Gamma(0.5, 1)) # 设置方差先验
kern += k1 + k2 # 累加到总核函数
else: # 如果不使用随机分解
k1 = GPy.kern.Linear(X.shape[1], ARD = False) # 全局线性核
k2 = GPy.kern.Matern32(X.shape[1], ARD = True) # 全局Matern32核，启用ARD
k2.lengthscale = np.std(X, axis = 0).clip(min = 0.02) # 设置长度尺度，最小0.02
k2.variance = 0.5 # 设置核方差
k2.variance.set_prior(GPy.priors.Gamma(0.5, 1), warning = False) # 设置方差先验
kern = k1 + k2 # 线性核 + Matern核
if not self.warp: # 如果不使用输入扭曲
self.gp = GPy.models.GPRegression(X, y, kern) # 创建标准高斯过程回归
else: # 如果使用输入扭曲
xmin = np.zeros(X.shape[1]) # 初始化最小边界
xmax = np.ones(X.shape[1]) # 初始化最大边界
xmin[:Xc.shape[1]] = -1 # 设置连续变量范围为[-1,1]
warp_f = GPy.util.input_warping_functions.KumarWarping(X, Xmin = xmin, Xmax = xmax) # 创建Kumar扭曲函数
self.gp = GPy.models.InputWarpedGP(X, y, kern, warping_function = warp_f) # 创建输入扭曲高斯过程
self.gp.likelihood.variance.set_prior(GPy.priors.LogGaussian(-4.63, 0.5), warning = False) # 设置噪声先验
# 多重启优化：最大迭代次数、是否显示信息、重启次数、鲁棒模式
self.gp.optimize_restarts(max_iters = self.num_epochs, verbose = self.verbose, num_restarts = self.num_restarts, robust = True)
return self # 返回自身用于链式调用
def predict(self, Xc : FloatTensor, Xe : LongTensor) -> (FloatTensor, FloatTensor):
Xall = self.trans(Xc, Xe) # 转换输入数据
py, ps2 = self.gp.predict(Xall) # 使用GPy模型预测（均值和方差）
mu = self.yscaler.inverse_transform(FloatTensor(py).view(-1, 1)) # 逆标准化均值
var = self.yscaler.std**2 * FloatTensor(ps2).view(-1, 1) # 调整方差到原始尺度
return mu, var.clamp(torch.finfo(var.dtype).eps) # 返回均值和确保非负的方差
def sample_f(self):
raise NotImplementedError('Thompson sampling is not supported for GP, use `sample_y` instead') # 不支持函数采样
@property
def noise(self):
var_normalized = self.gp.likelihood.variance[0] # 获取标准化后的噪声方差
return (var_normalized * self.yscaler.std**2).view(self.num_out) # 转换到原始尺度并调整形状
RF：随机森林回归
class RF(BaseModel):
"""
随机森林回归模型实现
"""
def __init__(self, num_cont, num_enum, num_out, **conf):
super().__init__(num_cont, num_enum, num_out, **conf) # 调用父类初始化
self.n_estimators = self.conf.get('n_estimators', 100) # 树的数量，默认100
self.rf = RandomForestRegressor(n_estimators = self.n_estimators) # 创建随机森林回归器
self.est_noise = torch.zeros(self.num_out) # 初始化估计噪声为零张量
if self.num_enum > 0: # 如果存在离散变量
self.one_hot = OneHotTransform(self.conf['num_uniqs']) # 创建one-hot编码器
def xtrans(self, Xc : FloatTensor, Xe: LongTensor) -> np.ndarray:
"""
转换输入数据为numpy数组格式
"""
if self.num_enum == 0: # 如果没有离散变量
return Xc.detach().numpy() # 直接返回连续变量的numpy数组
else: # 如果有离散变量
Xe_one_hot = self.one_hot(Xe) # 对离散变量进行one-hot编码
if Xc is None: # 如果没有连续变量
Xc = torch.zeros(Xe.shape[0], 0) # 创建空的连续变量张量
return torch.cat([Xc, Xe_one_hot], dim = 1).numpy() # 合并连续和离散特征并转为numpy
def fit(self, Xc : torch.Tensor, Xe : torch.Tensor, y : torch.Tensor):
"""
训练随机森林模型
"""
Xc, Xe, y = filter_nan(Xc, Xe, y, 'all') # 过滤包含NaN的数据点
Xtr = self.xtrans(Xc, Xe) # 转换输入特征为numpy格式
ytr = y.numpy().reshape(-1) # 转换目标变量为一维numpy数组
self.rf.fit(Xtr, ytr) # 训练随机森林模型
# 计算训练集上的MSE作为噪声估计
mse = np.mean((self.rf.predict(Xtr).reshape(-1) - ytr)**2).reshape(self.num_out)
self.est_noise = torch.FloatTensor(mse) # 保存估计的噪声水平
@property
def noise(self):
"""
返回估计的噪声水平
"""
return self.est_noise
def predict(self, Xc : torch.Tensor, Xe : torch.Tensor):
"""
使用训练好的模型进行预测
返回: (预测均值, 预测方差 + 估计噪声)
"""
X = self.xtrans(Xc, Xe) # 转换输入数据
mean = self.rf.predict(X).reshape(-1, 1) # 预测均值（所有树的平均）
preds = [] # 存储每棵树的独立预测
for estimator in self.rf.estimators_: # 遍历所有决策树
preds.append(estimator.predict(X).reshape([-1,1])) # 收集每棵树的预测
var = np.var(np.concatenate(preds, axis=1), axis=1) # 计算树间预测方差
# 返回均值和总方差（模型方差 + 估计噪声）
return torch.FloatTensor(mean.reshape([-1,1])), torch.FloatTensor(var.reshape([-1,1])) + self.noise
SVGP：稀疏变分高斯过程(Stochastic Variational Gaussian Process)
class SVGP(BaseModel):
"""
稀疏变分高斯过程 (Stochastic Variational Gaussian Process)
适用于大规模数据的近似高斯过程模型
"""
support_grad = True # 支持梯度计算
support_multi_output = True # 支持多输出任务
def __init__(self, num_cont, num_enum, num_out, **conf):
super().__init__(num_cont, num_enum, num_out, **conf) # 调用父类初始化
# 配置参数
self.use_ngd = conf.get('use_ngd', False) # 是否使用自然梯度下降
self.lr = conf.get('lr', 1e-2) # 基础学习率
self.lr_vp = conf.get('lr_vp', 1e-1) # 变分参数学习率
self.lr_fe = conf.get('lr_fe', 1e-3) # 特征提取器学习率
self.num_inducing = conf.get('num_inducing', 128) # 诱导点数量
self.ard_kernel = conf.get('ard_kernel', True) # 是否使用ARD核
self.pred_likeli = conf.get('pred_likeli', True) # 预测时是否考虑噪声
self.beta = conf.get('beta', 1.0) # ELBO的beta参数
# 训练参数
self.batch_size = conf.get('batch_size', 64) # 批大小
self.num_epochs = conf.get('num_epochs', 300) # 训练轮数
self.verbose = conf.get('verbose', False) # 是否显示训练信息
self.print_every = conf.get('print_every', 10) # 打印频率
self.noise_lb = conf.get('noise_lb', 1e-5) # 噪声下界
# 数据标准化器
self.xscaler = TorchMinMaxScaler((-1, 1)) # 输入标准化器
self.yscaler = TorchStandardScaler() # 输出标准化器
def fit_scaler(self, Xc : FloatTensor, Xe : LongTensor, y : FloatTensor):
"""拟合数据标准化器"""
if Xc is not None and Xc.shape[1] > 0: # 如果有连续变量
self.xscaler.fit(Xc) # 拟合输入标准化器
self.yscaler.fit(y) # 拟合输出标准化器
def xtrans(self, Xc : FloatTensor, Xe : LongTensor, y : FloatTensor = None):
"""转换输入数据格式"""
if Xc is not None and Xc.shape[1] > 0: # 处理连续变量
Xc_t = self.xscaler.transform(Xc) # 标准化连续变量
else:
Xc_t = torch.zeros(Xe.shape[0], 0) # 创建空的连续变量张量
if Xe is None: # 处理离散变量
Xe_t = torch.zeros(Xc.shape[0], 0).long()
else:
Xe_t = Xe.long() # 确保离散变量为long类型
if y is not None: # 如果提供了目标变量
y_t = self.yscaler.transform(y) # 标准化目标变量
return Xc_t, Xe_t, y_t
else:
return Xc_t, Xe_t
def fit(self, Xc : FloatTensor, Xe : LongTensor, y : FloatTensor):
"""训练SVGP模型"""
Xc, Xe, y = filter_nan(Xc, Xe, y, 'any') # 过滤包含NaN的数据点
self.fit_scaler(Xc, Xe, y) # 拟合标准化器
Xc, Xe, y = self.xtrans(Xc, Xe, y) # 转换数据格式
# 验证数据维度
assert(Xc.shape[1] == self.num_cont)
assert(Xe.shape[1] == self.num_enum)
assert(y.shape[1] == self.num_out)
# 设置噪声约束和创建模型
n_constr = GreaterThan(self.noise_lb) # 噪声下界约束
self.gp = SVGPModel(Xc, Xe, y, **self.conf) # 创建SVGP模型
# 为每个输出创建独立的高斯似然函数
self.lik = nn.ModuleList([GaussianLikelihood(noise_constraint = n_constr) for _ in range(self.num_out)])
# 设置模型为训练模式
self.gp.train()
self.lik.train()
# 创建数据加载器
ds = TensorDataset(Xc, Xe, y) # 创建Tensor数据集
dl = DataLoader(ds, batch_size = self.batch_size, shuffle = True, drop_last = y.shape[0] > self.batch_size) # 数据加载器
# 配置优化器
if self.use_ngd: # 如果使用自然梯度下降
opt = torch.optim.Adam([
{'params' : self.gp.fe.parameters(), 'lr' : self.lr_fe}, # 特征提取器参数
{'params' : self.gp.gp.hyperparameters()}, # GP超参数
{'params' : self.lik.parameters()}, # 似然函数参数
], lr = self.lr)
opt_ng = gpytorch.optim.NGD(self.gp.variational_parameters(), lr = self.lr_vp, num_data = y.shape[0]) # 自然梯度优化器
else: # 使用标准Adam优化器
opt = torch.optim.Adam([
{'params' : self.gp.fe.parameters(), 'lr' : self.lr_fe}, # 特征提取器参数
{'params' : self.gp.gp.hyperparameters()}, # GP超参数
{'params' : self.gp.gp.variational_parameters(), 'lr' : self.lr_vp}, # 变分参数
{'params' : self.lik.parameters()}, # 似然函数参数
], lr = self.lr)
# 为每个输出创建变分ELBO目标函数
mll = [gpytorch.mlls.VariationalELBO(self.lik[i], self.gp.gp[i], num_data = y.shape[0], beta = self.beta) for i in range(self.num_out)]
# 训练循环
for epoch in range(self.num_epochs):
epoch_loss = 0. # 累计损失
epoch_cnt = 1e-6 # 批次计数（避免除零）
for bxc, bxe, by in dl: # 遍历数据批次
dist_list = self.gp(bxc, bxe, by) # 前向传播，获取分布列表
loss = 0 # 初始化损失
valid = torch.isfinite(by) # 检查有效值
# 计算每个输出的损失
for i, dist in enumerate(dist_list):
loss += -1 * mll[i](dist, by[valid[:, i], i]) * valid[:, i].sum() # 加权ELBO损失
loss /= by.shape[0] # 平均损失
# 反向传播和优化
if self.use_ngd:
opt.zero_grad()
opt_ng.zero_grad()
loss.backward()
opt.step()
opt_ng.step()
else:
opt.zero_grad()
loss.backward()
opt.step()
# 累计损失和计数
epoch_loss += loss.item()
epoch_cnt += 1
epoch_loss /= epoch_cnt # 计算平均epoch损失
if self.verbose and ((epoch + 1) % self.print_every == 0 or epoch == 0):
print('After %d epochs, loss = %g' % (epoch + 1, epoch_loss), flush = True)
# 设置模型为评估模式
self.gp.eval()
self.lik.eval()
def predict(self, Xc, Xe):
"""模型预测"""
Xc, Xe = self.xtrans(Xc, Xe) # 转换输入数据
with gpytorch.settings.fast_pred_var(), gpytorch.settings.debug(False): # 快速预测设置
pred = self.gp(Xc, Xe) # 获取预测分布
if self.pred_likeli: # 如果考虑似然噪声
for i in range(self.num_out):
pred[i] = self.lik[i](pred[i]) # 通过似然函数转换
# 合并所有输出的均值和方差
mu_ = torch.cat([pred[i].mean.reshape(-1, 1) for i in range(self.num_out)], dim = 1)
var_ = torch.cat([pred[i].variance.reshape(-1, 1) for i in range(self.num_out)], dim = 1)
# 逆标准化到原始尺度
mu = self.yscaler.inverse_transform(mu_)
var = var_ * self.yscaler.std**2
return mu, var.clamp(min = torch.finfo(var.dtype).eps) # 确保方差非负
def sample_y(self, Xc, Xe, n_samples = 1) -> FloatTensor:
"""
从后验分布采样目标变量
返回: (n_samples, 样本数, 输出维度)
"""
Xc, Xe = self.xtrans(Xc, Xe) # 转换输入数据
with gpytorch.settings.debug(False):
pred = self.gp(Xc, Xe) # 获取预测分布
if self.pred_likeli: # 如果考虑似然噪声
for i in range(self.num_out):
pred[i] = self.lik[i](pred[i]) # 通过似然函数转换
# 从每个输出分布采样并合并
samp = [pred[i].rsample(torch.Size((n_samples, ))).reshape(n_samples, -1, 1) for i in range(self.num_out)]
samp = torch.cat(samp, dim = -1)
return self.yscaler.inverse_transform(samp) # 逆标准化采样结果
def sample_f(self):
"""不支持函数采样"""
raise NotImplementedError('Thompson sampling is not supported for GP, use `sample_y` instead')
@property
def noise(self):
"""获取估计的噪声水平"""
noise = torch.FloatTensor([lik.noise for lik in self.lik]).view(self.num_out).detach() # 各输出噪声
return noise * self.yscaler.std**2 # 转换到原始尺度
SVIDKL：稀疏变分深度核学习(Stochastic Variational Deep Kernel Learning)
class DKLFe(nn.Module):
"""
深度核学习特征提取器 (Deep Kernel Learning Feature Extractor)
使用神经网络自动学习特征表示，替代手动设计的核函数
"""
def __init__(self, num_cont, num_enum, num_out, **conf):
super().__init__()
# 神经网络配置参数
self.num_hiddens = conf.get('num_hiddens', 64) # 隐藏层维度，默认64
self.num_layers = conf.get('num_layers', 2) # 隐藏层数量，默认2层
self.act = conf.get('act', nn.LeakyReLU()) # 激活函数，默认LeakyReLU
self.sn_norm = conf.get('sn_norm') # 谱归一化，默认不使用
# 特征预处理：处理连续和离散变量的嵌入转换
self.emb_trans = DummyFeatureExtractor(num_cont, num_enum, conf.get('num_uniqs'), conf.get('emb_sizes'))
# 构建深度神经网络特征提取器
self.fe = construct_hidden(self.emb_trans.total_dim, self.num_layers, self.num_hiddens, self.act, self.sn_norm)
self.total_dim = self.num_hiddens # 输出特征维度
def forward(self, x, xe):
"""
前向传播：将原始输入转换为深度特征表示
"""
x_all = self.emb_trans(x, xe) # 首先进行嵌入转换，处理混合类型输入
return self.fe(x_all) # 通过深度网络提取高级特征
class SVIDKL(SVGP):
"""
稀疏变分深度核学习 (Stochastic Variational Deep Kernel Learning)
在SVGP基础上集成深度特征学习的扩展
"""
def __init__(self, num_cont, num_enum, num_out, **conf):
# 创建配置的深拷贝，避免修改原始配置
new_conf = deepcopy(conf)
# 关键变化1：禁用ARD核
new_conf.setdefault('ard_kernel', False)
# 解释：当有神经网络特征提取器时，不需要使用ARD核
# 因为神经网络已经自动学习了特征的重要性权重
# 关键变化2：使用深度特征提取器
new_conf.setdefault('fe', DKLFe(num_cont, num_enum, num_out, **new_conf))
# 解释：用深度神经网络替代简单的特征转换，自动学习数据表示
# 关键变化3：使用简化的核函数
new_conf.setdefault('kern', ScaleKernel(MaternKernel(nu = 2.5)))
# 解释：由于特征提取器已经处理了复杂性，可以使用更简单的核函数
# 调用父类SVGP的初始化，传入修改后的配置
super().__init__(num_cont, num_enum, num_out, **new_conf)
CatBoost：梯度提升树模型
class CatBoost(BaseModel):
"""
CatBoost梯度提升树模型实现
专门优化类别特征处理的梯度提升算法
"""
def __init__(self, num_cont, num_enum, num_out, **conf):
super().__init__(num_cont, num_enum, num_out, **conf) # 调用父类初始化
# CatBoost配置参数
self.num_epochs = self.conf.get('num_epochs', 100) # 最大树的数量（迭代次数）
self.lr = self.conf.get('lr', 0.2) # 学习率
self.depth = self.conf.get('depth', 10) # 树深度，推荐范围[1, 10]
self.loss_function = self.conf.get('loss_function', 'RMSEWithUncertainty') # 损失函数，支持不确定性估计
self.posterior_sampling = self.conf.get('posterior_sampling', True) # 是否使用后验采样
self.verbose = self.conf.get('verbose', False) # 是否显示训练过程
self.random_seed = self.conf.get('random_seed', 42) # 随机种子
self.num_ensembles = self.conf.get('num_ensembles', 10) # 虚拟集成数量
# 确保迭代次数足够用于集成
if self.num_epochs < 2 * self.num_ensembles:
self.num_epochs = self.num_ensembles * 2 # 至少是集成数量的2倍
# 创建CatBoost回归器实例
self.model = CatBoostRegressor(
iterations=self.num_epochs, # 迭代次数
learning_rate=self.lr, # 学习率
depth=self.depth, # 树深度
loss_function=self.loss_function, # 损失函数（支持不确定性）
posterior_sampling=self.posterior_sampling, # 后验采样
verbose=self.verbose, # 是否显示训练信息
random_seed=self.random_seed, # 随机种子
allow_writing_files=False) # 禁止写入文件
def xtrans(self, Xc: FloatTensor, Xe: LongTensor) -> FeaturesData:
"""
转换输入数据为CatBoost所需的格式
"""
# 处理连续变量：转换为numpy float32格式
num_feature_data = Xc.numpy().astype(np.float32) if self.num_cont != 0 else None
# 处理离散变量：转换为字符串格式（CatBoost要求）
cat_feature_data = Xe.numpy().astype(str).astype(object) if self.num_enum != 0 else None
# 返回CatBoost特征数据对象
return FeaturesData(
num_feature_data=num_feature_data, # 数值特征
cat_feature_data=cat_feature_data) # 类别特征
def fit(self, Xc: FloatTensor, Xe: LongTensor, y: FloatTensor):
"""
训练CatBoost模型
"""
Xc, Xe, y = filter_nan(Xc, Xe, y, 'all') # 过滤NaN值
# 创建训练数据池
train_data = Pool(
data=self.xtrans(Xc=Xc, Xe=Xe), # 转换特征数据
label=y.numpy().reshape(-1)) # 转换标签为一维数组
self.model.fit(train_data) # 训练模型
def predict(self, Xc: FloatTensor, Xe: LongTensor) -> (FloatTensor, FloatTensor):
"""
使用CatBoost进行预测，返回均值和方差
"""
test_data = Pool(data=self.xtrans(Xc=Xc, Xe=Xe)) # 创建测试数据池
# 使用虚拟集成进行预测，获取不确定性估计
preds = self.model.virtual_ensembles_predict(
data=test_data, # 测试数据
prediction_type='TotalUncertainty', # 预测类型：总不确定性
virtual_ensembles_count=self.num_ensembles) # 虚拟集成数量
# 解析预测结果：
# preds[:, 0] = 预测均值
# preds[:, 1] = 知识不确定性（模型不确定性）
# preds[:, 2] = 偶然不确定性（数据噪声）
mean = preds[:, 0] # 预测均值
var = preds[:, 1] + preds[:, 2] # 总方差 = 模型方差 + 数据噪声
# 返回PyTorch张量格式的结果
return torch.FloatTensor(mean.reshape([-1,1])), \
torch.FloatTensor(var.reshape([-1,1]))
DeepEnsemble:深度集成模型
class DeepEnsemble(BaseModel):
"""
深度集成模型 - 通过多个独立训练的神经网络集成来估计不确定性
基于Lakshminarayanan et al. (2017) "Simple and Scalable Predictive Uncertainty Estimation using Deep Ensembles"
"""
# 模型能力标识
support_ts = True # 支持Thompson采样
support_grad = True # 支持梯度计算
support_multi_output = True # 支持多输出任务
support_warm_start = True # 支持热启动（从已有模型继续训练）
def __init__(self, num_cont, num_enum, num_out, **conf):
"""
初始化深度集成模型
参数:
num_cont: 连续变量数量
num_enum: 离散变量数量
num_out: 输出维度
**conf: 配置参数
"""
super().__init__(num_cont, num_enum, num_out, **conf) # 调用父类初始化
# 集成策略配置
self.bootstrap = self.conf.setdefault('bootstrap', False) # 是否使用自助采样创建数据多样性
self.rand_prior = self.conf.setdefault('rand_prior', False) # 是否使用随机先验网络
self.output_noise = self.conf.setdefault('output_noise', True) # 是否输出噪声估计
self.num_ensembles = self.conf.setdefault('num_ensembles', 5) # 集成模型数量
self.num_process = self.conf.setdefault('num_processes', 1) # 并行训练进程数
self.num_epochs = self.conf.setdefault('num_epochs', 500) # 每个模型的训练轮数
self.print_every = self.conf.setdefault('print_every', 50) # 训练信息打印频率
# 网络架构配置
self.num_layers = self.conf.setdefault('num_layers', 1) # 隐藏层数量
self.num_hiddens = self.conf.setdefault('num_hiddens', 128) # 隐藏层维度
self.l1 = self.conf.setdefault('l1', 1e-3) # L1正则化系数
self.batch_size = self.conf.setdefault('batch_size', 32) # 批大小
self.lr = self.conf.setdefault('lr', 5e-3) # 学习率
self.adv_eps = self.conf.setdefault('adv_eps', 0.) # 对抗训练扰动大小（未使用）
self.verbose = self.conf.setdefault('verbose', False) # 是否显示训练详情
self.basenet_cls = self.conf.setdefault('basenet_cls', BaseNet) # 基础网络类
