当前位置：首页 > news >正文

PyTorch模型Ensemble实现

news 2025/7/21 8:42:15

PyTorch 模型 Ensemble 实现

以下是使用 PyTorch 实现各种 Ensemble 方法的代码示例：

1. 基础 Ensemble 方法

1.1 投票法 (Voting)

import torch
import torch.nn as nn
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score# 创建示例数据
X, y = make_classification(n_samples=1000, n_features=20, n_classes=2, random_state=42)
X = torch.tensor(X, dtype=torch.float32)
y = torch.tensor(y, dtype=torch.long)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)# 定义不同的模型
class Model1(nn.Module):def __init__(self):super().__init__()self.fc = nn.Linear(20, 2)def forward(self, x):return self.fc(x)class Model2(nn.Module):def __init__(self):super().__init__()self.fc1 = nn.Linear(20, 10)self.fc2 = nn.Linear(10, 2)def forward(self, x):x = torch.relu(self.fc1(x))return self.fc2(x)class Model3(nn.Module):def __init__(self):super().__init__()self.fc1 = nn.Linear(20, 5)self.fc2 = nn.Linear(5, 5)self.fc3 = nn.Linear(5, 2)def forward(self, x):x = torch.relu(self.fc1(x))x = torch.relu(self.fc2(x))return self.fc3(x)# 训练单个模型
def train_model(model, X_train, y_train, epochs=100):criterion = nn.CrossEntropyLoss()optimizer = torch.optim.Adam(model.parameters())for epoch in range(epochs):optimizer.zero_grad()outputs = model(X_train)loss = criterion(outputs, y_train)loss.backward()optimizer.step()# 创建并训练模型
models = [Model1(), Model2(), Model3()]
for model in models:train_model(model, X_train, y_train)# 投票集成预测
def voting_predict(models, X):predictions = []with torch.no_grad():for model in models:outputs = model(X)_, predicted = torch.max(outputs.data, 1)predictions.append(predicted)# 多数投票stacked_preds = torch.stack(predictions)final_pred, _ = torch.mode(stacked_preds, dim=0)return final_pred# 评估
y_pred = voting_predict(models, X_test)
print(f"Voting Ensemble Accuracy: {accuracy_score(y_test, y_pred)}")

1.2 平均法 (Averaging)

# 平均概率集成预测
def averaging_predict(models, X):probabilities = []with torch.no_grad():for model in models:outputs = model(X)prob = torch.softmax(outputs, dim=1)probabilities.append(prob)avg_prob = torch.mean(torch.stack(probabilities), dim=0)_, final_pred = torch.max(avg_prob, 1)return final_pred# 评估
y_pred = averaging_predict(models, X_test)
print(f"Averaging Ensemble Accuracy: {accuracy_score(y_test, y_pred)}")

2. 高级 Ensemble 方法

2.1 Bagging 实现

from torch.utils.data import Dataset, DataLoaderclass SimpleDataset(Dataset):def __init__(self, X, y):self.X = Xself.y = ydef __len__(self):return len(self.X)def __getitem__(self, idx):return self.X[idx], self.y[idx]def bagging_train(models, dataset, n_estimators=10, sample_ratio=0.8, epochs=50):for i in range(n_estimators):# 创建子采样数据集indices = torch.randperm(len(dataset))[:int(len(dataset)*sample_ratio)]subset = torch.utils.data.Subset(dataset, indices)loader = DataLoader(subset, batch_size=32, shuffle=True)model = Model2()  # 使用相同的模型架构criterion = nn.CrossEntropyLoss()optimizer = torch.optim.Adam(model.parameters())for epoch in range(epochs):for inputs, labels in loader:optimizer.zero_grad()outputs = model(inputs)loss = criterion(outputs, labels)loss.backward()optimizer.step()models.append(model)# 初始化并训练Bagging模型
bagging_models = []
dataset = SimpleDataset(X_train, y_train)
bagging_train(bagging_models, dataset)# 使用Bagging模型预测
y_pred = averaging_predict(bagging_models, X_test)
print(f"Bagging Ensemble Accuracy: {accuracy_score(y_test, y_pred)}")

2.2 Stacking 实现

# 首先训练基模型
base_models = [Model1(), Model2(), Model3()]
for model in base_models:train_model(model, X_train, y_train)# 创建元特征数据集
def create_meta_features(models, X):with torch.no_grad():features = []for model in models:outputs = model(X)prob = torch.softmax(outputs, dim=1)features.append(prob)return torch.cat(features, dim=1)X_meta_train = create_meta_features(base_models, X_train)
X_meta_test = create_meta_features(base_models, X_test)# 定义并训练元模型
meta_model = nn.Sequential(nn.Linear(6, 10),  # 3个模型 * 2个类别 = 6个特征nn.ReLU(),nn.Linear(10, 2)
)train_model(meta_model, X_meta_train, y_train)# 评估Stacking
with torch.no_grad():outputs = meta_model(X_meta_test)_, y_pred = torch.max(outputs, 1)print(f"Stacking Ensemble Accuracy: {accuracy_score(y_test, y_pred)}")

3. 深度学习模型 Ensemble

3.1 模型权重平均

# 假设我们已经训练了多个模型
trained_models = [Model1(), Model2(), Model3()]
for model in trained_models:train_model(model, X_train, y_train)# 创建权重平均模型
class WeightedAverageEnsemble(nn.Module):def __init__(self, models, weights=None):super().__init__()self.models = nn.ModuleList(models)if weights is None:weights = [1/len(models)] * len(models)self.weights = weightsdef forward(self, x):outputs = []for model, weight in zip(self.models, self.weights):outputs.append(model(x) * weight)return torch.sum(torch.stack(outputs), dim=0)# 创建并评估集成模型
ensemble = WeightedAverageEnsemble(trained_models)
with torch.no_grad():outputs = ensemble(X_test)_, y_pred = torch.max(outputs, 1)print(f"Weighted Average Ensemble Accuracy: {accuracy_score(y_test, y_pred)}")

3.2 预测结果平均

def predict_with_models(models, X):with torch.no_grad():outputs = []for model in models:outputs.append(model(X))avg_output = torch.mean(torch.stack(outputs), dim=0)_, preds = torch.max(avg_output, 1)return predsy_pred = predict_with_models(trained_models, X_test)
print(f"Average Prediction Ensemble Accuracy: {accuracy_score(y_test, y_pred)}")

4. 实现注意事项

模型多样性：确保集成中的模型具有不同的架构或训练方式
训练稳定性：使用不同的随机种子初始化模型以增加多样性
计算效率：考虑使用多GPU训练不同的模型
验证策略：使用交叉验证来评估集成效果
超参数调整：对集成中的每个模型单独调参

# 示例：使用不同随机种子增加多样性
def train_with_random_seed(model_class, X_train, y_train, seed, epochs=100):torch.manual_seed(seed)model = model_class()train_model(model, X_train, y_train, epochs)return model# 训练多个不同初始化的模型
diverse_models = [train_with_random_seed(Model2, X_train, y_train, seed) for seed in range(5)]