皆さん、機械学習でモデルの学習時、何していますか？
ぼーっとコンソールに出力されるログを眺めていませんか？
「お、思ったより下がった！」「あれ、精度悪くなってる、頑張れ！！」
と思いながら、見るのは意外と楽しいですよね。
でも応援したところで、モデルがその分頑張ってくれるわけでもないし、ドラマチックな展開があるわけでもないので、見てても時間の無駄なんですよね。
そんなことを考えてるときに気付いてしまいました。
「あれ、この時間筋トレしたらよくね？」
そんなわけで、機械学習の合間に筋トレできるような環境を考えました。
NTTドコモの服部です。
こちらはNTTドコモサービスイノベーション部 AdventCalender2019 23日目の記事です。
この記事の対象者
- 筋トレ好き
- 運動不足な機械学習界隈の人
- kagglerなどLightGBMをよく使う人
なんで筋トレするの？
モデルの精度向上のためには、何度も仮説・検証を繰り返す必要があります。
しかし、全然精度が上がらない時には、やる気が低下することもあると思います。
そんな時には筋トレが有効です。
筋トレをすると、「テストステロン」というホルモンの分泌が活性化します。
テストステロンは脳のやる気をつかさどるホルモンであり、これを分泌させることで
精度が上がらなくて落ち込むときも
「よし！もう一回！」
と考えられるようになり、精度向上まったなしです。
また、世の中的にも筋トレが最近流行ってきているように感じます。
女子高生が筋トレするアニメが放映されたり、山手線で筋トレするアプリが出たり。やはり筋トレの需要は高そうです。
山手線の車内で筋トレ…JR東日本が導入した“電車専用アプリ”。 でも、周囲の目は気にならないの？
方向性
筋トレの強制まではさすがに出来ないので、
- 「筋トレしなきゃ！」と思わせる状況を作る
→筋トレのメニュー・始める合図、リズム等の音を再生 - 筋トレのモチベーションを上げる
→学習状況（精度）によって、筋トレの回数などを変化させてゲーム性を持たす
といったところにフォーカスします。
また、実装対象のモデルとしては、LightGBMを対象とします。
※「深層学習」系のモデルは学習時間が長い傾向があり、試すにも筋トレし続けるのがきついので、断念しました...orz
LightGBMとは
LightGBMはMicrosoftが開発した勾配ブースティングのライブラリです。
Kaggle1でもよく使われており、高精度かつ学習速度も早いのが魅力です。
XGBoostも有名ですが、最近はLightGBMのほうが使われている印象があります。
LightGBMについては、下記の記事がわかりやすいです。
LightGBM公式ドキュメント(English)
「初手LightGBM」をする7つの理由
LightGBM 徹底入門 – LightGBMの使い方や仕組み、XGBoostとの違いについて
LightGBMのCallback関数
今回はLightGBMのCallback関数を主に使って実装します。
LightGBMのCallback関数は、学習時に自分で定義したCallback関数を渡すことで、学習時に実行されます。
基本的に、多くの機械学習ライブラリ(NNフレームワークを除く)では、学習はライブラリ側で実行されるため、学習途中については使う側がカスタマイズしにくいのですが、callback関数により様々なカスタマイズが可能になります。
例えば
- 学習経過をloggerに送る（さらにはSlack/LINEに送る）
- 独自のEarlyStoppingの実装
- ハイパーパラメータを動的に変更する
などの使いみちがあり、ヘビーユーザーからは重宝されています。
Callback関数の仕組み
Callback関数は、namedtuple型の引数を受け取れるよう定義します（下のサンプルコードでいうenv）。
その中に学習状況を含んだ変数が含まれているので、それを使った処理を定義します。
受け取れるものは、
- ハイパーパラメータ
- 現在のiteration回数
- 現在のiterationでのtrain/validのスコア
- 学習中のモデル(Boosterクラス)
などです。
定義したcallback関数を、lgb.train()やlgb.fit()など学習実行するときに、callbacksという引数名でリスト形式で渡します。
以下、簡単なサンプルコードです。
import lightgbm as lgb
....
def callback_func(env):
"""
自分で定義するCallback関数
引数のenv(namedtuple)から学習の状況を取得できる
"""
if env.iteration + 1 % 100 == 0:
print(f"今、{env.iteration + 1}イテレーション終わったよ")
lgb.train(
lgb_param,
.....
callbacks=[callback_func] #callbacksに定義した関数をリスト形式で渡す
)
下記のリンクを見ると、より詳しい中身や他の使いみちが理解できるかと思います。
LightGBM/callback.py at master · Microsoft/LightGBM · GitHub
細かすぎて伝わらないLightGBM活用法 (callback関数)
LightGBMのcallbackを利用して学習履歴をロガー経由で出力する
実装した機能
学習時に筋トレガイダンスを出す
- 学習開始時: 筋トレメニュー開始の案内
- 例：「腹筋30回スタート」
- 一定間隔で、メトロノーム音（ポーン音）を出す
- 例：5秒毎にポーン
- 学習終了/最大回数終了時に、終了の案内
- 例：「お疲れさまでした。」「よく頑張りました。」
筋トレには大きく
- 時間が決まっているもの（例：プランク30秒）
- 回数が決まっているもの（例：腹筋30回）
の２つがあり、メトロノーム音についてはどちらの種類かで出すべきタイミングが変わります。
時間が決まっているもの
目標時間になるまで一定の間隔で音を出します。
ただし、普通の時間にするのではなく、「学習時くらいLightGBMと同じ時間軸で頑張ってほしい」という思いから、目標時間も音を鳴らす間隔も機械学習のiterationで設定することにしました。
「30秒プランク」よりも 「300iterationプランク」 のほうがモデルとも仲良くなれる気がします。
回数が決まっているもの
回数が決まっているものはペースが重要なため、iteration回数に合わせるのは難しいので、秒数で設定します。
最初の数イテレーションで、指定した秒数分のiteration回数を計算し、以降はその回数毎にメトロノーム音を出します。
筋トレメニューは指定 or ランダム
同じトレーニングばかりするのもよくないので、自分で指定もしくはランダム設定ができるようにしました。
オススメはランダム設定で、何のトレーニングをするかも含めて楽しむことです。
精度が前回より良かったら、メトロノーム音が変わる
辛い筋トレ中に、「お、精度上がった！」と分かれば、筋トレのモチベーションもあがりますよね。
また、学習途中に精度が気になって筋トレに集中できないみたいなことを防げます。
これを実装するために、毎回学習ログをログファイルに出力し、
前回の学習ログを読み込み比較するよう実装しました。
精度が前回より悪かったら、ペナルティ（筋トレ追加）
ゲーム性をもたせる意味でも、モデルの精度は筋トレにも影響します。
こちらも前回学習ログとの比較で対応できます。
途中で学習を打ち切ってもペナルティ（筋トレ追加）
途中で学習を打ち切る≒精度が上がらなかった
みたいなものなので、その場合はペナルティの筋トレです。
Keyboard Interrupt ExceptionをCatchすることで対応します。
「Ctrl + C」で抜け駆けなんて許しません。
実装
実行環境/用意
- MacOS 10.14.5
- Python 3.6.8
- lightgbm 2.3.0
- VLC media player 3.0.8
インストール
# VLCのインストール
brew cask install vlc
# Python-VLCのインストール
pip install python-vlc
# LightGBMのインストール
pip install lightgbm
音源準備
音源の用意が必要です。
音声についてはMacOS標準の音声読み上げ機能2で作成し、
その他効果音についてはフリーの音源サイト3から用意しました。
音声は、自分好みのものを用意することで、よりやる気が上がるかもしれません。
ソースコード
事前に定義しておくconfig
筋トレのメニューや種類、実行回数、音源のパスなどを設定します。
本質的でないところなので、折りたたんでおきます。
筋トレのメニュー用Configコード
train_config = {
"planc":{
"train_menu": "planc",
"train_type": "duration",
"total_iteration": 500,
},
"abs":{
"train_menu": "abs",
"train_type": "iter",
"total_times": 50,
"seconds_1time": 3
},
"pushup":{
"train_menu": "pushup",
"train_type": "iter",
"total_times": 50,
"seconds_1time": 2
},
"squat":{
"train_menu": "squat",
"train_type": "iter",
"total_times": 50,
"seconds_1time": 2
},
}
def make_sound_dict(config):
sound_dict = {
"iteration_10":[
'sound/iter_sound_1.mp3'
],
"iteration_100":[
'sound/iter_sound_2.mp3'
],
"iteration_100_better":[
'sound/iter_sound_3.mp3'
],
"train_finish":[
'sound/finish.mp3'
]
}
if config["train_type"] == "duration":
sound_dict["train_start"] = [
f"sound/{config['total_iteration']}iter.mp3", # Nイテレーション
f"sound/{config['train_menu']}_train.mp3",
"sound/start.mp3"
]
elif config["train_type"] == "iter":
sound_dict["train_start"] = [
f"sound/{config['train_menu']}_train.mp3", # 筋トレ名（ex: 腕立て伏せ、腹筋、。。。）
f"sound/{config['total_times']}times.mp3", # Nイテレーション
"sound/start.mp3" # スタート
]
return sound_dict
Callback関数を含んだ筋トレ用クラス
少し長いですが、メインのところなのでそのまま載せます。
class MuscleSound():
"""
LightGBMで筋トレするためのCallback
"""
def __init__(self, train_config, train_menu="planc"):
if train_menu == "random":
# randomの場合はメニューからランダムで設定
train_menu = random.choice(train_config.keys())
assert(train_menu in train_config.keys())
self.train_menu = train_menu
self.config = train_config[train_menu]
self.sound_dict = make_sound_dict(self.config)
self.log_dir = "./muscle"
self.start_time = None
self.n_iter_1time = None
# setup
os.makedirs(self.log_dir, exist_ok=True)
self._setup_prev_log()
self._load_prev_log()
def media_play(self, media_list):
"""
指定されたmedia_listの音声ファイルを順番に再生
"""
p = vlc.MediaListPlayer()
vlc_media_list = vlc.MediaList(media_list)
p.set_media_list(vlc_media_list)
p.play()
def _setup_prev_log(self):
"""
前回の学習ログがcurr.logになっているのを
prev.logにrename
"""
log_filepath = os.path.join(self.log_dir, "curr.log")
if os.path.exists(log_filepath):
os.rename(
log_filepath,
os.path.join(self.log_dir, "prev.log")
)
def _load_prev_log(self, log_filepath="muscle/prev.log"):
"""
前回学習時のログを読み込む
"""
if os.path.exists(log_filepath):
self.prev_log = pd.read_csv(
log_filepath, names=["iter","score"]
).set_index("iter")["score"]
else:
self.prev_log = None
def _check_score(self, env):
"""
スコアの比較及び、ログの保存
"""
n_iter = env.iteration + 1
is_better_score = False
# Validationスコアを抽出
# valid_setsの最後のdatasetのスコアを使う
curr_score = env.evaluation_result_list[-1][2]
# 数値が上がるほど良いスコアになるかどうか
is_upper_metric = env.evaluation_result_list[-1][3]
# 前回のログに同じiterationのスコアがあれば比較
if self.prev_log is not None and n_iter in self.prev_log.index:
prev_score = self.prev_log.loc[n_iter]
is_better_score = curr_score > prev_score \
if is_upper_metric else curr_score < prev_score
# ログを保存
with open(os.path.join(self.log_dir, "curr.log"), "a") as f:
f.write(f"{n_iter},{curr_score}\n")
return is_better_score
def play_train_start(self, train_menu):
"""
学習スタート時の音再生
"""
self.play_media_list(self.sound_dict["train_start"])
# 読み終わる前に学習（筋トレ）が始まらないよう少しsleep
time.sleep(5)
def duration_sound(self, env):
"""
時間が決まっている筋トレ向け
一定のイテレーション回数毎のときに音を出す
"""
if (env.iteration + 1) > self.config["total_iteration"]:
# 筋トレする最大イテレーション数を超えたら何もしない
return
elif env.iteration + 1 == self.config["total_iteration"]:
# 終了回数に達したため、合図
self.media_play(self.sound_dict["train_finish"])
elif (env.iteration + 1) % 100 == 0:
# 100回毎の音
is_better_score = self._check_score(env)
if is_better_score:
self.media_play(self.sound_dict["iteration_100_better"])
else:
self.media_play(self.sound_dict["iteration_100"])
elif (env.iteration + 1) % 10 == 0:
# 10回毎の音
self.media_play(self.sound_dict["iteration_10"])
def iter_sound(self, env):
"""
時間に合わせた音再生（回数が決まっている筋トレ向け）
一定の秒数毎に音を出す
"""
if self.n_iter_1time is None:
return
if (env.iteration + 1) > self.config["total_times"]*self.n_iter_1time:
# 筋トレする最大回数数を超えたら何もしない
return
if (env.iteration + 1) == self.config["total_times"]*self.n_iter_1time:
# 最大回数に達したら終了の案内を出す
self.media_play(self.sound_dict["train_finish"])
if (env.iteration + 1)%self.n_iter_1time != 0:
# 1回あたりのイテレーション数で割り切れない場合は何もしない
return
if ((env.iteration + 1)//self.n_iter_1time) % 10 == 0:
# 100回毎の音
self.media_play(self.sound_dict["iteration_100"])
else:
# １0回毎の音
self.media_play(self.sound_dict["iteration_10"])
def __call__(self, env):
if env.iteration == 0:
# 学習開始時
self.media_play(self.sound_dict["train_start"])
if self.config["train_type"] == "times":
# １回あたりの適切なiteration数を設定
if env.iteration == 1:
self.start_time = time.time()
elif env.iteration == 11:
time_10iter = time.time() - self.start_time
self.n_iter_1time = int(self.config["seconds_1time"] / time_10iter * 10)
print("1回あたりのiteration数", self.n_iter_1time)
if not env.evaluation_result_list:
return
# 筋トレタイプに合わせたメトロノーム音再生
if self.config["train_type"] == "iter":
self.iter_sound(env)
elif self.config["train_type"] == "duration":
self.duration_sound(env)
中断時のペナルティ処理
学習を途中で止めてしまった場合にペナルティを課すための処理です。
こちらはCallback関数ではなく、KeyboardInterruptのExceptionをCatchして、処理をしています。
また学習時に簡単に書けるようデコレータとして使えるようにしてます。
def penalty_muscle(func):
def play_media_list(media_list):
"""
指定されたmedia_listの音声ファイルを順番に再生
"""
p = vlc.MediaListPlayer()
vlc_media_list = vlc.MediaList(media_list)
p.set_media_list(vlc_media_list)
p.play()
def wrapper_func(*args, **kwargs):
try:
func(*args, **kwargs)
except KeyboardInterrupt:
interrupt_list = [
'sound/keyboard_interrupt.mp3',
'sound/1000iter.mp3',
'sound/planc_train.mp3',
'sound/add.mp3'
]
print("Ctrl+Cしたので筋トレ追加!!!")
play_media_list(interrupt_list)
time.sleep(5)
for i in range(100):
if i % 10 == 0 and i > 0:
play_media_list([ 'sound/iter_sound_2.mp3'])
else:
play_media_list([ 'sound/iter_sound_1.mp3'])
time.sleep(1)
raise Exception(KeyboardInterrupt)
return wrapper_func
Callback関数を使った学習コード例
通常のLightGBMの学習に対して、
- 学習させる関数にpenalty_muscleのデコレータをつける
- MuscleSoundクラスのインスタンスを生成して、callbacksに渡す
だけです。
デコレータはKeyboardInterruptのCatchをしているだけなので、関数内でLightGBMの学習をしていれば、どんな関数でも良いです。
これでいつでも簡単に筋トレ版LightGBMが作れちゃいます。
@penalty_muscle # 学習を行う関数にデコレータをつける
def train_muscle_lgb(train_df, target_col, use_cols):
folds = KFold(n_splits=2, shuffle=True, random_state=2019)
for i, (trn_, val_) in enumerate(folds.split(train_df, train_df[target_col])):
print(f"============fold{i}============")
trn_data = lgb.Dataset(
train_df.loc[trn_, use_cols],
label=train_df.loc[trn_, target_col]
)
val_data = lgb.Dataset(
train_df.loc[val_, use_cols],
label=train_df.loc[val_, target_col]
)
lgb_param = {
"objective": "binary",
"metric": "auc",
"learning_rate": 0.01,
"verbosity": -1,
}
# MuscleSoundクラスのインスタンス生成(実質Callback関数)
callback_func = MuscleSound(train_config, train_menu="random")
model = lgb.train(
lgb_param,
trn_data,
num_boost_round=10000,
valid_sets=[trn_data, val_data],
verbose_eval=100,
early_stopping_rounds=500,
callbacks=[callback_func] #callback関数を指定
)
実際にやってみた
諸事情によりアニメーションGIFです。
本当は字幕に書いてある声・音がするので、脳内再生してください。
実際にやってみて感じた課題
実際に自分で筋トレしてみた結果、まだまだ課題はありました。
サボってしまう
いくら音が聞こえたって、何回もやるとサボりたくなります。にんげんだもの。
- あまりに何回もやると飽きるので、１日の筋トレ上限回数を決めておく
- IoT機器で筋トレやっていることを感知する。その結果をtwitterなどにあげる
などの対処が必要そうです。
後者はハードルが高そうですが、ここまでやれば皆使うかも？
学習時間が短すぎて筋トレにならない
学習データが少ないと筋トレする間もなかったです。数十万件くらいは欲しいですね。
そしてLightGBMやっぱり速いです。
対策は難しいですが、学習率を下げるか、特徴量作りまくるしかないですね。
(あれ？筋トレするためにLightGBM使う状況になってる？)
そもそも筋トレはペナルティなのか？
私はそんなに筋トレ好きではないのですが、筋トレ好きにとっては、筋トレ回数増やすことはペナルティではないのでは。。？あえて精度を上げなかったり、Ctrl+Cする人が出てくるかも。。？
筋トレとは、好きでやるものなのかペナルティでやるものなのか、奥が深いです。
まとめ
みなさんもモデルの学習中は筋トレして有意義な時間を過ごしましょう！！
-
世界で最も有名なデータ分析プラットフォーム(https://www.kaggle.com/) ↩