sklearnのMLPRegressorで簡易な転移学習

初めに

転移学習など少し高度なモデルや手法を実装する際はpytorchを採用することが多いのですが、簡単なNNであればsklearnにもsklearn.neural_network.MLPRegressorで実装されています。
今回は、簡易な転移学習をMLPRegressor.partial_fit()を使用して実施した際に、学習率を変える箇所にハマったので備忘録がてら残します。

コード

import pickle
import itertools

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.model_selection import cross_val_predict
from sklearn.model_selection import KFold
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.neural_network import MLPRegressor
from sklearn.metrics import r2_score

# https://github.com/scikit-learn/scikit-learn/issues/3299
class OnlinePipeline(Pipeline):
    def partial_fit(self, X, y=None):
        for i, step in enumerate(self.steps):
            name, est = step
            
            # scalerもpartial_fit持っている
            # 普通の転移学習ではsourceの平均と偏差を使う事が多いが、
	    # 今回は徐々に変更する
            est.partial_fit(X, y)
            
            # 最後のmodel以前は変換も行う
            if i < len(self.steps) - 1:
               X = est.transform(X)
        return self
	
def learn_model(X, y, lr=0.0001):
    pipeline = OnlinePipeline(steps=[
                        ('scaler', StandardScaler()),
                        ('model', MLPRegressor(hidden_layer_sizes=(50, 50), learning_rate="constant", solver="sgd", learning_rate_init=lr, random_state=1))
                        ])
    pipeline.fit(X, y.values.ravel())
    return pipeline

def partial_learn(pipeline, X1, y1, X0, y0, n_iter=50):
    list_r2_1 = []
    list_r2_0 = []
    for _ in range(n_iter):
        pipeline.partial_fit(X1, y1.values.ravel())
        
        y1_pred = pipeline.predict(X1)
        list_r2_1.append(r2_score(y1, y1_pred))
        
        y0_pred = pipeline.predict(X0)
        list_r2_0.append(r2_score(y0, y0_pred))
    return list_r2_0, list_r2_1

つまりポイント

一旦fit()を行うと、中身でoptimizerが新しく作成されないので、途中から学習率を変更することができないです。https://github.com/scikit-learn/scikit-learn/blob/98cf537f5c538fdbc9d27b851cf03ce7611b8a48/sklearn/neural_network/_multilayer_perceptron.py#L540のif not incremental or not hasattr(self, "_optimizer"):の箇所です。そのため、学習率を途中で変更したい場合は、_optimizer属性を消す必要があります。（partial_fit()の場合はincremental = True）

これを行わない場合、学習率の変更が反映されないので注意です。

    def _fit_stochastic(
        self,
        X,
        y,
        activations,
        deltas,
        coef_grads,
        intercept_grads,
        layer_units,
        incremental,
    ):

        params = self.coefs_ + self.intercepts_
        if not incremental or not hasattr(self, "_optimizer"):
            if self.solver == "sgd":
                self._optimizer = SGDOptimizer(
                    params,
                    self.learning_rate_init,
                    self.learning_rate,
                    self.momentum,
                    self.nesterovs_momentum,
                    self.power_t,
                )
            elif self.solver == "adam":
                self._optimizer = AdamOptimizer(
                    params,
                    self.learning_rate_init,
                    self.beta_1,
                    self.beta_2,
                    self.epsilon,
                )

対応策

delattr(new_model, "_optimizer")により、_optimizer属性を削除することでpartial_fit()するときにoptimizerが新しく作成されます。

def create_newmodel(pipeline_old, lr_new=0.01):
    
    new_model = pipeline_old["model"]
    new_model.learning_rate_init = lr_new
    # _optimizer を消すことで、再度lrを反映したoptimizerを作成する
    delattr(new_model, "_optimizer")
    
    pipeline_new = OnlinePipeline(steps=[
                        ('scaler', pipeline_old["scaler"]),
                        ('model', new_model)
                        ])
    
    return pipeline_new

pipeline0_01 = learn_model(X0, y0, lr=0.1)

y0_pred = pipeline0_01.predict(X0)
y1_pred = pipeline0_01.predict(X1)
r2_0 = r2_score(y0, y0_pred)
r2_1 = r2_score(y1, y1_pred)

print(r2_0)
print(r2_1)


pipeline_new = create_newmodel(pipeline0_01, lr_new=0.01)

list_r2_0, list_r2_1 = partial_learn(pipeline_new, X1, y1, X0, y0)
print(list_r2_0[-10:], list_r2_1[-10:])

n_iter=50
plt.plot(range(n_iter), list_r2_0)
plt.plot(range(n_iter), list_r2_1)
plt.show()

最後に

以上でsklearnで簡易に転移学習を実装できました。お手軽に試したいレベルであれば十分かもしれません。もし間違っている箇所があればコメントで教えてください。

以上です。