TensorFlowアドベントカレンダー2020の4日目の記事です（参加したの12/5です）。アドベントカレンダー初参加です。
この記事では何番煎じだかわからない、TensorFlowから訓練経過をSlackなどに送信する方法を書いていきます。ただし、requestsなどの送信処理を自分で書かなくてよいのがポイントです。10行未満できます。
結論
事前準備としてSlackの「Incoming Webhooks」を用意します。
tf.keras.callbacks.RemoteMonitorを継承してこういうコールバックを作りましょう。
class SlackCallback(tf.keras.callbacks.RemoteMonitor):
def __init__(self):
self.SLACK_ROOT = "https://hooks.slack.com"
self.SLACK_PATH = "/services/XXXXXXXXX/XXXXXXXXXXX/xxxxxxxxxxxxxxxxxxxxxxxx"
super().__init__(root=self.SLACK_ROOT, path=self.SLACK_PATH, send_as_json=True)
def on_epoch_end(self, epoch, logs=None):
super().on_epoch_end(epoch, logs={"text":str(logs)})
これで終わり。Webhookのパスが漏れると誰でも投稿できてしまうので、ソースを公開する前提では、パスを外部ファイル化してコミットから外すか、環境変数を活用するなど工夫しましょう。使い方は、TensorBoardやCallbackと同じで、
model.fit(X_train, y_train, validation_data=(X_test, y_test), callbacks=[SlackCallback()])
とします。これでエポック終了後にKerasのログがSlackに流れます。ImageNetの訓練など、訓練時間が長くなるときの進捗確認に便利だと思います。POSTで投げられればなんでもいいので、Slackでなくても応用できます。
Slackへの送信処理をrequestレベルから書かなくて良いのが楽でいいですね。
Incoming Webhooks
Slackへの送信は、「何でもできるトークンを用意してポストする」という説明がされることがありますが、これはレガシーAPIで古い仕様です。最新の仕様だとIncoming Webhooksを使うのが手軽でしょう。
Incoming Webhooksについては、丁寧な記事がいっぱいあるのでここでは説明しません。
SlackのIncoming Webhooksを使い倒す
【2020年度版】Slack通知はSlack AppのIncoming Webhooksを使おう！やり方を解説
PythonでSlackのIncoming Webhookを試してみる
Incoming Webhooksが利用できると、SlackのAPI画面では次のような表示になります。
tf.keras.callbacks.RemoteMonitor
tf.keras（Keras）ではログを送信するためのクラスは既に用意されています。tf.keras.callbacks.RemoteMonitorというのがそれです。
tf.keras.callbacks.RemoteMonitor
Keras documentコールバックの使い方 （こっちのほうが説明が丁寧）
requestsライブラリが必要です。RemoteMonitorは、ロスや精度といったログを、エポックの終わりに指定URLへ送信するというものです。まさに今求めているものですね。エラーハンドリングもされているので、送信失敗したら訓練が止まるというのも気にしなくていいです。
内部の実装はこちら。かなり単純なものですがいちいち実装すると面倒くさそうな処理です。TensorFlow側で用意されているので活用しましょう。
送信時に「TypeError: Object of type float32 is not JSON serializable」というエラーが出ることがありますが（TF2.1環境）、2.3.1にアップデートしたらエラーが消えました。TensorFlowのソースの差分を見てもその部分がケアされています。
RemoteMonitorをfitに入れるだけではダメ
RemoteMonitorはコールバックですが、これをfitに入れるだけではSlackのAPIとの整合性が取れません。別途クラスを用意せず、こう書くほうが単純なように思えます。ただしこれはメッセージがSlackに反映されません。
# SLACK_ROOTとSLACK_PATHにURLがある前提で、
slack_monitor = tf.keras.callbacks.RemoteMonitor(
root=SLACK_ROOT, path=SLACK_PATH, send_as_json=True, field="text"
)
model.fit(X_train, y_train, validation_data=(X_test, y_test),
batch_size=128, epochs=10, callbacks=[slack_monitor])
Incoming Webhooksのサンプルを見ると、
curl -X POST -H 'Content-type: application/json' --data '{"text":"Hello, World!"}' https://hooks.slack.com/services/...
となっているので、これで正しそうに思えますが、実際に動かしてみるとSlack上でメッセージが表示されませんでした。
RemoteMonitorを拡張する
Kerasのログに対して、送信時に（Slack側で表示できるような）何らかの処理を加えたいことがあります。Slackで表示されない対策だけでなく、もう少し高度な処理をSlackでさせたいときに必要になります。これは単にRemoteMonitorを継承すればいいです。冒頭に示したコードがそれです。
class SlackCallback(tf.keras.callbacks.RemoteMonitor):
def __init__(self):
self.SLACK_ROOT = "https://hooks.slack.com"
self.SLACK_PATH = "/services/XXXXXXXXX/XXXXXXXXXXX/xxxxxxxxxxxxxxxxxxxxxxxx"
super().__init__(root=self.SLACK_ROOT, path=self.SLACK_PATH, send_as_json=True)
def on_epoch_end(self, epoch, logs=None):
super().on_epoch_end(epoch, logs={"text":str(logs)})
RemoteMonitorを継承して、コンストラクタで親クラスのコンストラクタを呼び出し、on_epoch_endでlogを加工して親クラスのon_epoch_endにわたすという単純なものです。本来Kerasのログについていないようなエポック数を送信することもできます。
class SlackCallback(tf.keras.callbacks.RemoteMonitor):
def __init__(self):
self.SLACK_ROOT = "https://hooks.slack.com"
self.SLACK_PATH = "/services/XXXXXXXXX/XXXXXXXXXXX/xxxxxxxxxxxxxxxxxxxxxxxx"
super().__init__(root=self.SLACK_ROOT, path=self.SLACK_PATH, send_as_json=True)
def on_epoch_end(self, epoch, logs=None):
logs["epoch"] = epoch+1 # エポック数を追加（0インデックスなので+1している）
super().on_epoch_end(epoch, logs={"text":str(logs)})
現在時刻をログに追加することもできますし、複数GPUや複数ランタイムで別々の訓練ジョブを走らせて、ランタイムのIDを付与してSlackで統合管理するとかやると便利ではないかと思います。Incoming Webhooksでは画像を扱うこともできるので、GANのような生成モデルの途中経過をSlackに投げるということも応用できそうですね。
なお、super().on_epoch_end(...)の、str(logs)のstrを外したらSlackでメッセージが表示されなくなったので、メッセージ全体の文字列キャストが必要なのかなと思いました。
MNISTサンプル
MNISTの訓練経過をエポック単位でSlackに送信するサンプルコードです。
import tensorflow as tf
import tensorflow.keras.layers as layers
import numpy as np
## logをstrするために継承する
class SlackCallback(tf.keras.callbacks.RemoteMonitor):
def __init__(self):
self.SLACK_ROOT = "https://hooks.slack.com"
self.SLACK_PATH = "/services/XXXXXXXXX/XXXXXXXXXXX/xxxxxxxxxxxxxxxxxxxxxxxx"
super().__init__(root=self.SLACK_ROOT, path=self.SLACK_PATH, send_as_json=True)
def on_epoch_end(self, epoch, logs=None):
logs["epoch"] = epoch+1
super().on_epoch_end(epoch, logs={"text":str(logs)})
def main():
(X_train, y_train), (X_test, y_test) = tf.keras.datasets.mnist.load_data()
X_train = X_train.astype(np.float32).reshape(-1, 784) / 255.0
X_test = X_test.astype(np.float32).reshape(-1, 784) / 255.0
# MLP
inputs = layers.Input((784,))
x = layers.Dense(128, activation="relu")(inputs)
x = layers.Dense(10, activation="softmax")(x)
model = tf.keras.models.Model(inputs, x)
model.compile("adam", "sparse_categorical_crossentropy", ["sparse_categorical_accuracy"])
# Slack callback
model.fit(X_train, y_train, validation_data=(X_test, y_test),
batch_size=128, epochs=10, callbacks=[SlackCallback()])
if __name__ == "__main__":
main()
ぜひ活用してみてください。
宣伝
技術書典10で新刊出します。NumPy関数だけで画像や動画を処理する（フォトショのような画像編集ソフトと同じ処理をする）本です。実践演習形式で221問収録予定です。こちらもお楽しみに。