앙상블 기법
앙상블(Ensemble)이란 일련의 예측 모델들로부터 예측 결과를 수집하여 더 좋은 예측을 도출하기 위한 방법입니다. 예를 들면, 훈련 데이터셋에서 무작위 다른 부분 데이터셋을 만들어 일련의 Decision Tree를 만들고 개별 트리의 예측을 구하여 종합해 가장 많은 선택을 받은 클래스를 최종 예측을 삼는 방법이 있을 수 있습니다.
여기선 기본이 되는 여러 앙상블 기법들을 살펴보고, 딥러닝에서 어떻게 활용할 수 있을지 간단히 코드를 통해 알아보려고 합니다.
1. 투표 기반
여러 분류기들의 예측을 모아서 가장 많이 선택된(투표 결과 높은) 클래스를 예측하는 방법을 직접 투표(hard voting)라고 합니다. 즉, 다수결 투표로 정해지는 것입니다. 머신러닝에서 많이 활용하는 사이킷런에서는 sklearn.ensemble에서 VotingClassifer 클래스를 불러오고 voting 매개변수를 'hard'로 불러올 수 있습니다.
from sklearn.ensemble import VotingClassifier
clf1 = LogisticRegression(random_state=1)
clf2 = RandomForestClassifier(n_estimators=50, random_state=1)
clf3 = GaussianNB()
eclf = VotingClassifier(
estimators=[('lr', clf1), ('rf', clf2), ('gnb', clf3)],
voting='hard'
)
사이킷런을 활용하지 않을 경우 hard voting의 원리에 따라 각 결과들을 저장하고 예측 결과(클래스)들에 대해 하나씩 세주는 Counter를 활용해 세주는 방식으로 구현할 수 있습니다.
import numpy as np
from collections import Counter
def hard_voting_ensemble(classifiers, X):
predictions = []
for classifier in classifiers:
predictions.append(classifier.predict(X))
final_predictions = []
for sample_predictions in zip(*predictions):
vote_count = Counter(sample_predictions)
final_predictions.append(vote_count.most_common(1)[0][0])
return np.array(final_predictions)
하지만, 이런 투표방식은 각 클래스의 예측 확률을 반영하지 못해 비교적 낮은 확률이라도 개수가 조금이라도 많으면 채택될 수 있습니다. 확률을 결과값으로 가져올 수 있다면 이들을 평균 내어서 확률이 가장 높은 클래스를 선택할 경우 조금 더 세밀한 앙상블 기법이 될 수 있습니다.
이를 간접 투표(soft voting)이라고 하며 위에서 사이킷런을 통해서 구현할 때는 voting만 'soft'로만 바꾸면 됩니다. 하지만, 사이킷런을 사용하지 않을 경우 아래 코드와 같이 사용할 수 있습니다.
import torch
def soft_voting_ensemble(models, inputs):
predictions = []
for model in models:
model.eval()
with torch.no_grad():
output = model(inputs) # 여기서 모델은 확률을 반환
predictions.append(output)
ensemble_pred = torch.mean(torch.stack(predictions), dim=0)
return torch.argmax(ensemble_pred, dim=1)
# Ensemble 예측
ensemble_models = [model1, model2, model3]
final_prediction = soft_voting_ensemble(ensemble_models, test_inputs)
2. 배깅과 페이스팅
앞서 다른 알고리즘으로 학습된 결과를 합치는 방법도 있지만, 같은 알고리즘을 통해 훈련할 때 훈련 데이터셋을 무작위로 서브셋을 구성해 모델을 각기 다르게 학습시키고 합치는 방법도 있습니다. 배깅(Bootstrap aggregating; Bagging)은 훈련 데이터셋을 중복 허용해 샘플링하는 방식이고, 중복을 허용하지 않고 샘플링하는 것을 페이스팅(pasting)이라고 합니다. 만약에 한 모델을 위해 적용한다면 배깅이 적합한 방식이라고 합니다. [2]
사이킷런으로는 아래와 같이 구현할 수 있습니다.
from sklearn.ensemble import BaggingClassifier
from sklearn.neighbors import KNeighborsClassifier
bagging = BaggingClassifier(KNeighborsClassifier(),
max_samples=0.5, max_features=0.5)
또는 딥러닝에서 활용할 수 있도록 pytorch로 구현할 수 있습니다.
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset, SubsetRandomSampler
import numpy as np
class BaggingEnsemble(nn.Module):
def __init__(self, base_model, n_estimators, input_size, hidden_size, output_size):
super(BaggingEnsemble, self).__init__()
self.base_model = base_model
self.n_estimators = n_estimators
self.models = nn.ModuleList([base_model(input_size, hidden_size, output_size) for _ in range(n_estimators)])
def forward(self, x):
outputs = [model(x) for model in self.models]
return torch.stack(outputs).mean(dim=0)
def train_bagging(model, train_loader, criterion, optimizer, n_epochs):
model.train()
for epoch in range(n_epochs):
for batch_idx, (data, target) in enumerate(train_loader):
optimizer.zero_grad()
output = model(data)
loss = criterion(output, target)
loss.backward()
optimizer.step()
def predict_bagging(model, test_loader):
model.eval()
predictions = []
with torch.no_grad():
for data, _ in test_loader:
output = model(data)
pred = output.argmax(dim=1, keepdim=True)
predictions.extend(pred.numpy())
return np.array(predictions)
# 사용 예시
if __name__ == "__main__":
X = torch.randn(1000, 10)
y = torch.randint(0, 2, (1000,))
dataset = TensorDataset(X, y)
train_loader = DataLoader(dataset, batch_size=32, shuffle=True)
test_loader = DataLoader(dataset, batch_size=32, shuffle=False)
input_size, hidden_size, output_size = 10, 50, 2
bagging_model = BaggingEnsemble(BaseModel, n_estimators=10, input_size=input_size, hidden_size=hidden_size, output_size=output_size)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(bagging_model.parameters())
train_bagging(bagging_model, train_loader, criterion, optimizer, n_epochs=5)
predictions = predict_bagging(bagging_model, test_loader)
print("Predictions shape:", predictions.shape)
랜덤 포레스트는 배깅이나 페이스팅을 이용해 하나의 베이스모델인 Decision Tree를 적용한 것들의 앙상블입니다. [3] 다만, 딥러닝 보다는 전통적인 머신러닝 기법이니만큼 굳이 pytorch의 필요성이 없기 때문에 참고자료의 내용으로 대신합니다.
3. 스태킹
모든 모델들의 예측을 모으는 과정을 간단함 함수(예를 들어, 투표 방식) 대신에 모델을 통해 앙상블을 수행하는 방식입니다. 개별 모델의 예측은 함께 모여서 마지막 모델에 예측을 위한 입력으로 사용되며, 마지막 모델은 교차 검증(cross-validation)을 통해 학습됩니다. 여기서 마지막 모델은 블렌더(Blender) 또는 메타 학습기(Meta learner)라고 불립니다.
사이킷런을 사용한다면 StackingClassifier 또는 StackingRegressor를 사용하면 되지만, 딥러닝은 조금 더 구현이 필요합니다. 메타 학습기(메타 모델)을 구현하고, 기존 모델 예측값을 합친 후에 다시 메타 모델에 넣어서 최종 예측값을 도출하는 과정이 추가되면 됩니다. 다소 길긴 하지만, 아래처럼 구현할 수 있습니다.
import torch
import torch.nn as nn
import torch.optim as optim
class BaseModel(nn.Module):
def __init__(self, input_size, hidden_size, output_size):
super(BaseModel, self).__init__()
self.fc1 = nn.Linear(input_size, hidden_size)
self.relu = nn.ReLU()
self.fc2 = nn.Linear(hidden_size, output_size)
def forward(self, x):
x = self.fc1(x)
x = self.relu(x)
x = self.fc2(x)
return x
class MetaModel(nn.Module):
def __init__(self, input_size, output_size):
super(MetaModel, self).__init__()
self.fc = nn.Linear(input_size, output_size)
def forward(self, x):
return self.fc(x)
def train_model(model, X, y, epochs=100):
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters())
for epoch in range(epochs):
optimizer.zero_grad()
outputs = model(X)
loss = criterion(outputs, y)
loss.backward()
optimizer.step()
def stacking_ensemble(X_train, y_train, X_test):
model1 = BaseModel(20, 64, 2)
model2 = BaseModel(20, 32, 2)
model3 = BaseModel(20, 16, 2)
train_model(model1, X_train, y_train)
train_model(model2, X_train, y_train)
train_model(model3, X_train, y_train)
with torch.no_grad():
pred1 = model1(X_train)
pred2 = model2(X_train)
pred3 = model3(X_train)
test_pred1 = model1(X_test)
test_pred2 = model2(X_test)
test_pred3 = model3(X_test)
meta_input = torch.cat((pred1, pred2, pred3), dim=1)
meta_test_input = torch.cat((test_pred1, test_pred2, test_pred3), dim=1)
meta_model = MetaModel(6, 2) # 6 = 2 (클래스 수) * 3 (기본 모델 수)
train_model(meta_model, meta_input, y_train)
with torch.no_grad():
final_pred = meta_model(meta_test_input)
return final_pred
ensemble_pred = stacking_ensemble(X_train, y_train, X_test)
ensemble_pred = torch.argmax(ensemble_pred, dim=1).numpy()
참고자료
[1] 오렐리앙 제롱. "핸즈온 머신러닝 2판"
[2] https://scikit-learn.org/stable/modules/ensemble.html#bagging
[3] https://scikit-learn.org/stable/modules/ensemble.html#forest
[4] https://scikit-learn.org/stable/modules/ensemble.html#stacked-generalization
