はじめに

備忘録としてPoetryの使い方を残します。

インストール

$ curl -sSL https://raw.githubusercontent.com/python-poetry/poetry/master/get-poetry.py | python

コマンド一覧

個人的に使いそうなコマンドを載せます。

Poetryのアップデート

$ poetry self update

プロジェクトの作成

$ poetry new <project-name>

パッケージのインストール

$ poetry add <package-name>

パッケージのアンインストール

$ poetry remove <package-name>

パッケージ一覧を表示

$ poetry show

仮想環境内でのコマンドの実行

$ poetry run <commands...>

pythonでファイルを実行

$ poetry run python <file-name>

仮想環境内でシェルを立ち上げる

$ poetry shell

$SHELLに従ってシェルを立ち上げます

インストール後の始め方

プロジェクト内に、仮想環境が作られるようにします

$ poetry config virtualenvs.in-project true

このコマンドはインストール後に、1回すればOK

サンプルプロジェクトを作成します

$ poetry new poetry-sample

プロジェクトディレクトリに移動

$ cd poetry-sample

追加したいパッケージをインストール
ここでは、numpyをインストール

$ poetry add numpy

poetry-sampleディレクトリで、sample.pyを作成

sample.py

import numpy as np

x = np.array([1, 2, 3])
print(x)

仮想環境内でシェルを立ち上げます

$ poetry shell

スクリプトを実行

$ python sample.py

または仮想環境でシェルを立ち上げずに、直接コマンドを実行して実行できます

poetry run python sample.py

以上のような流れで、プロジェクトの作成、パッケージの追加、プログラムの実行を行えます

スクリプトはpoetry-sample内になくても実行はできます

Pythonのバージョンの管理

pyenvを併用して行う方法です

Python3.7.0を例とします

まずpyenvで3.7.0がインストールされていなければインストールします

$ pyenv install 3.7.0

プロジェクト内のローカルのバージョンを変更します

$ pyenv local 3.7.0

コマンドで仮想環境を構築します

$ poetry env use 3.7.0

これでプロジェクト内のPythonのバージョンは3.7.0になります

最後に

とても便利なのでこれから使っていきたいです。

内容

前回グラフを作れたので、クラスタリング等を行ってみる。
また、韻の捉え方を拡張する。「eい」→「ee」,「oう」→「oo」と変換し、変換後に母音が一致したものも「韻」と捉える。これはカタカナ英語や子供の間違えやすい日本語を参考にした。「ei」と表記しないのは「えき」が「ええ」にはならないということだ。「い」と「う」は母音単体ではないといけない。（「れいぞうこ」が「れえぞおこ」）

グラフの操作

import networkx as nx
import matplotlib.pyplot as plt
import community

G = nx.Graph()
G.add_weighted_edges_from(edge_list)
#クラスタリング
partition = community.best_partition(G, weight="weight")
#コミュニティごとにノードを分けてリストにする。[[コミュニティ０のノードリスト],[コミュニティ１のノードリスト]…]
part_sub = [[] for _ in set(list(partition.values()))]
for key in partition.keys():
    part_sub[partition[key]].append(key)
#各コミュニティで固有ベクトル中心性が最大となるノードをリストに格納
max_betw_cent_node = []
for part in part_sub:
    G_part = nx.Graph()
    for edge in edge_list:
        if edge[0] in part and edge[1] in part:
            G_part.add_weighted_edges_from([edge])
    max_betw_cent_node.append(max(G_part.nodes(), key=lambda val:
                                 nx.eigenvector_centrality_numpy(G_part, weight="weight")[val]))

print([dic[i] for i in max_betw_cent_node])
#モジュラリティ指標
print(community.modularity(partition,G))

クラスタリングし、コミュニティ毎に固有ベクトル中心性が最大となるものを求めた。良い分割が出来ていれば、それぞれをtarget_wordとした時に、良い結果が得られると思われる。エッジの重みに使っている部分に閾値を設けることで重みに差が付くようにしてみる等を検討する。

韻の捉え方の拡張

from pykakasi import kakasi
import re

with open("./gennama.txt","r", encoding="utf-8") as f:
    data = f.read()

kakasi = kakasi()
kakasi.setMode('J', 'K')
kakasi.setMode('H', 'K')
conv = kakasi.getConverter()
text_data = conv.do(data)

#eイ→ee,oウ→ooのように変換したテキストを得る
def expansion(text_data):
    #最後の文字次第だが、i,uの余分が出ることを長さを合わせることで解決
    text_data_len = len(text_data)
    #いい椅子や、そういう噂のような「い、う」の連続への対処。
    text_data = text_data.replace("イイ", "イi").replace("ウウ","ウu")
    text_data = text_data.split("イ")
    new_text_data = []
    kakasi.setMode('K', 'a')
    conv = kakasi.getConverter()
    for i in range(text_data_len):
        if len(text_data[i]) > 0:
            if ("e" in conv.do(text_data[i][-1])):
                new_text_data.append(text_data[i] + "e")
            else:
                new_text_data.append(text_data[i] + "i")

    text_data = "".join(new_text_data).split("ウ")
    new_text_data = []
    for i in range(text_data_len):
        if len(text_data[i]) > 0:
            if ("o" in conv.do(text_data[i][-1])):
                new_text_data.append(text_data[i] + "o")
            else:
                new_text_data.append(text_data[i] + "u")

    return "".join(new_text_data)[:text_data_len]

print(expansion(text_data))

まず、データをカタカナへ変換し、「イ、ウ」で分割、直前の文字の母音によって処理を変更する、という方針でやっていたがなかなか苦労した。データの最後が「イ、ウ」の場合や、それ以外の時は「iu」が残る。引数のデータと長さが変わらないようにすることで対処したが、printしてみると「i」が最後に残っていた。「イイ、ウウ」という連続出現を想定していなかった。やはり、やってみるとすんなりはいかないものだし、やらないと気付かないことが多い。

今後の方針

（カタカナ変換データ、母音だけ残したデータ、拡張したデータ）それぞれで一致部分をスコア化し、（子音の一致、母音の一致、響きの一致）を捉えようと思う。長音、撥音、促音の一致を見ることは不要と判断した。
つまり、これまでのことを一度まとめてみる。N-gramもスペース分割部分も考慮すべきと思うし、一致部分の見方にも問題有りと思っている。現状最良の方法をまとめ、入力データをなにかしら用意して検証していきたい。

概要

海外ではエンジニアの面接においてコーディングテストというものが行われるらしく、多くの場合、特定の関数やクラスをお題に沿って実装するという物がメインである。

その対策としてLeetCodeなるサイトで対策を行うようだ。

早い話が本場でも行われているようなコーディングテストに耐えうるようなアルゴリズム力を鍛えるサイト。

せっかくだし人並みのアルゴリズム力くらいは持っておいた方がいいだろうということで不定期に問題を解いてその時に考えたやり方をメモ的に書いていこうかと思います。

Leetcode

ゼロから始めるLeetCode 目次

前回
[ゼロから始めるLeetCode Day35「160. Intersection of Two Linked Lists」(https://qiita.com/KueharX/items/f9c01184085586fbe491)

基本的にeasyのacceptanceが高い順から解いていこうかと思います。

Twitterやってます。

問題

難易度はeasy。
Top 100 Liked Questionsのeasyはこれが最後の問題になります。

問題としては、push,pop,top,getMinという関数を持つMinStackクラスを実装してくださいというものです。

なお、それぞれの関数の仕様は以下の通り。

push(x) -- Push element x onto stack.
pop() -- Removes the element on top of the stack.
top() -- Get the top element.
getMin() -- Retrieve the minimum element in the stack.

Input
["MinStack","push","push","push","getMin","pop","top","getMin"]
[[],[-2],[0],[-3],[],[],[],[]]

Output
[null,null,null,null,-3,null,0,-2]

Explanation
MinStack minStack = new MinStack();
minStack.push(-2);
minStack.push(0);
minStack.push(-3);
minStack.getMin(); // return -3
minStack.pop();
minStack.top(); // return 0
minStack.getMin(); // return -2

使用例は以上です。
Stackについては多くのプログラマーが知っていると思うのでここでは触れません。

解法

class MinStack:

    def __init__(self):
        """
        initialize your data structure here.
        """
        self.stack = []


    def push(self, x: int) -> None:
        minElement = self.getMin()
        if minElement == None or x < minElement:
            minElement = x
        self.stack.append((x,minElement))

    def pop(self) -> None:
        self.stack.pop()


    def top(self) -> int:
        if len(self.stack) == 0:
            return None
        else:
            return self.stack[len(self.stack) - 1][0]

    def getMin(self) -> int:
        if len(self.stack) == 0:
            return None
        else:
            return self.stack[len(self.stack) - 1][1]



# Your MinStack object will be instantiated and called as such:
# obj = MinStack()
# obj.push(x)
# obj.pop()
# param_3 = obj.top()
# param_4 = obj.getMin()
# Runtime: 72 ms, faster than 38.62% of Python3 online submissions for Min Stack.
# Memory Usage: 17.9 MB, less than 5.36% of Python3 online submissions for Min Stack.

スライスが便利、かと思いきや普通に他の言語でも普通に似たような実装になりそうですね・・・
何にせよstackを実装してみるのは面白いので書いてみることをお勧めします。

今回はこんな感じで。お疲れ様でした。

概要

バブルソートでソートされる様子を matplotlib を使って描画してみました
※jupyter notebook上に貼り付けて実行してください

bubble_sort.py

import numpy as np
import matplotlib.pyplot as plt
%matplotlib inline

list_a = [5,7,4,5,1,2,3,2,9,1,4]

left = np.arange(1, len(list_a) + 1)
height = list_a

plt.bar(left, height)
plt.show()

for i in range(len(list_a)):
    for j in range(0, len(list_a) - i - 1):
        if list_a[j] > list_a[j + 1]:
            list_a[j], list_a[j + 1] = list_a[j + 1], list_a[j]

    height = np.array(list_a)
    plt.bar(left, height)
    plt.show()

初めに

• 本記事は、python・機械学習等々を独学している小生の備忘録的な記事になります。
• 「自身が気になったコードを写経しながら勉強していく」という、きわめてシンプルなものになります。
• 建設的なコメントを頂けますと幸いです（気に入ったらLGTM & ストックしてください）。

お題:IBM HR Analytics Employee Attrition & Performance

• 今回のお題は、IBM HR Analytics Employee Attrition & Performanceです。kaggleに記載のあった説明によると、「従業員の退職理由」を探る問題のようです。
• 今回、以下のyoutubeの動画を見つつ写経していきました。

Link：Predict Employee Attrition Using Machine Learning & Python

データはkaggleから取ってきました。

Link:IBM HR Analytics Employee Attrition & Performance

分析はyoutubeの動画にある通り、Google Colaboratryを使用しました（便利な時代になったものです）。

Step 1：データ読み込み～内容確認

それではやっていきたいと思います。

1.1: ライブラリのインポート

# ライブラリの読み込み
import numpy as np
import pandas as pd
import seaborn as sns

基本となるライブラリを読み込みます。必要なライブラリはこの後も適宜追加していく感じになります。

次にデータの読み込みですが、kaggleのサイトからダウンロードしたcsvファイルをgoogle colabで読み込みます。

1.2: Google Colabへのファイルのアップロード

# データのアップロード
from google.colab import files
uploaded = files.upload()

これをすることによって、ローカルに保存してあるファイルをgoogle colab上に取り込むことができます。
普段はGoogle Driveにファイルのアップロードしてから、Google Driveを連携させて読み込んでいたので、こっちのほうが楽でいいですね。

1.3: pandasでの読み込み

アップロードしたデータを読み込んでいきます。

# データの読み込み
df = pd.read_csv('WA_Fn-UseC_-HR-Employee-Attrition.csv')

# データの確認
df.head(7)

おなじみのコードですね。次からはデータの中身を確認していきます。

データの中身の確認

以下のコードは（実際には）それぞれ別々に実行していますが、ここでは纏めて書いておきます。

# データフレームの行・列の数確認
df.shape

# それぞれの列の中身のデータ型を確認
df.dtypes

# 欠損値の確認
df.isna().sum()
df.isnull().values.any()

#基礎統計量の確認
df.describe()

# 退職者と在籍者数の数の確認（被説明変数の数の把握）
df['Attrition'].values_counts() #図1

# 退職者と在籍者数の可視化
sns.countplot(df['Attrition'])

# 年齢別の退職者数と在籍者数の可視化
import matplotlib.pyplot as plt
plt.subplots(figsize=(12,4))
sns.countplot(x='Age', hue='Attrition', data=df, palette='colorblind') #図2

【図1】

【図2】

ここまではいつもやるようなデータの確認です。まずしっかりデータの中身を確認することはやはり必要だと思います。

1.4: object型のユニーク値の確認

次に、先ほど確認したデータ型のうちobject型の列のユニーク値を確認します。

for column in df.columns:
  if df[column].dtype == object:
    print(str(column) + ':' + str(df[column].unique()))
    print(df[column].value_counts())
    print('___________________________________________')

1. 1行目：forループでそれぞれの列を繰り返しとってくる
2. 2行目：取ってきた列がobject型か判定
3. 3行目：カラム名 + そのカラムのユニーク値を出力
4. 4行目：各ユニーク値の個数を出力

1.5: 不要行の削除

.drop()で予測するのに意味をなさない列を削除します。

df = df.drop('Over18', axis=1)
df = df.drop('EmployeeNumber', axis=1)
df = df.drop('StandardHours', axis=1)
df = df.drop('EmployeeCount', axis=1)

これは説明不要ですね。退職する理由にならないものをdfの中から外します。

1.6: 列間の相関の確認

これもお馴染みの処理だと思います。各列間の相関（correlation）を確認し、heatmap可視化します。

df.corr()

plt.figure(figsize=(14, 14))
sns.heatmap(df.corr(), annot=True, fmt='.0%')

今回、heatmapを作成する際に指定しているのは以下の2つです。

Item	Description
annot	True に設定すると、セルに値を出力します。
fmt	annot=True に設定した場合、またはデータセットを指定した場合の出力フォーマットを文字列で指定

参考：Seaborn でヒートマップを作成する

1.7: sklearnでカテゴリカル(non numerical)データにラベリング

from sklearn.preprocessing import LabelEncoder

for column in df.columns:
    if df[columen].dtype == np.number:
        continue
    df[column] = LabelEncoder().fit_transform(df[column])

ここでは、sklearnのLabelEncoderを使ってobject型だったデータに数値データに置き換える（「分類器にかける前に文字データを離散値(0, 1, ・・・)に変換」）。

置き換えが終わったら、dfの列の順番を入れ替え分析しやすい形にします。

# Ageを新しい列に複製
df['Age_Years'] = df['Age']

# Age列を落とす
df = df.drop('Age', axis=1)

Step2: sklearnで分析

ここからが本番ですね（前処理が重要なのは言うまでもないですが）。

# dfをデータを説明変数と被説明変数に分割
X = df.iloc[:, 1:df.shape[1]].values
Y = df.iloc[:, 0].values

# 訓練データ、教師データのテストデータサイズ（25%）で分割
from sklearn.model_selection import train_test_split
X_train, X_test, Y_train, Y_test = train_test_split(X, Y, test_size=0.25, ramdom_state = 0)

# ランダムフォレストによる分類
from sklearn.ensemble import RandomForestClassifier
forest = RandomForestClassifier(n_estimators = 10, criterion = 'entropy', random_state = 0)
forest.fit(X_train, Y_train)

上から見ていきましょう。

• iloc[]を使って説明変数と被説明変数を分離します。
• sklearnのtrain_test_splitを使って訓練データとテストデータの分割。 train_test_splitの引数は以下の通りです。

Item	Description
arrays	分割対象の同じ長さを持った複数のリスト、Numpy の array, matrix, Pandasのデータフレームを指定。
test_size	小数もしくは整数を指定。小数で指定した場合、テストデータの割合を 0.0 〜 1.0 の間で指定します。整数を指定した場合は、テストデータに必ず含めるレコード件数を整数で指定します。指定しなかった場合や None を設定した場合は、train_size のサイズを補うように設定します。train_size を設定していない場合、デフォルト値として 0.25 を用います。
train_size	小数もしくは整数を指定。小数で指定した場合、トレーニングデータの割合を 0.0 〜 1.0 の間で指定します。整数を指定した場合は、トレーニングデータに必ず含めるレコード件数を整数で指定します。指定しなかった場合や None を設定した場合は、データセット全体から test_size を引いた分のサイズとします。
random_state	乱数生成のシードとなる整数または、RandomState インスタンスを設定します。指定しなかった場合は、Numpy のnp.random を用いて乱数をセットします。

(参照: scikit-learn でトレーニングデータとテストデータを作成する)

• ランファムフォレストを使って分類をします。ここでの引数は、

n_estimators: 木の数の指定(デフォルトは100)
criterion: gini or entropyを指定（デフォルトはgini）

この後、forest.fit(...)でモデルを学習させます。

それでは精度を見ていきましょう。

forest.score(X_train, Y_train)

この後、confusion_matrix(混合行列)を使ってAccuracyを計算してます。

from sklearn.metrics import confusion_matrix

cm = confusion_matrix(Y_test, forest.predict(X_test)) #cm: confusion_matrix

TN = cm[0][0]
TP = cm[1][1]
FN = cm[1][0]
FP = cm[0][1]

print(cm)
print('Model Testing Accuracy = {}'.format( (TP + TN) / (TP + TN + FN + FP)))

以上で、簡単ではありますがskleanを使って二値分類の写経になります。

　最後に

内容的にはそこまで難しいものではないにせよ、まだまだ理解していない箇所があることが認識できましたので、引き続き勉強していきたいと思います。

以上。

コメント

xlwingsを使ったグラフ作成の参考にしてください。
こうする方がベターだよというご指摘あればうれしいです。
apiを使うとvbaのコードが使える→ わからないことはExcelにて「マクロの記録」で調べられる。
test.xlsxは白紙のワークブックです。

環境

python 3.7
xlwings 0.18.0

サンプルコード

import xlwings as xw
from xlwings.constants import AxisType

DATA_NUM = 10
# エクセルワークブックの読み込み
wb = xw.Book('test.xlsx')

# データの挿入
data_x = list(x for x in range(DATA_NUM))
data_y1 = list(x*2 for x in range(DATA_NUM))
data_y2 = list(x*10-50 for x in range(DATA_NUM))

xw.Range('A1').value = 'x'
xw.Range('B1').value = 'y1'
xw.Range('C1').value = 'y2'

cell_x = xw.Range('A2')
cell_y1 = xw.Range('B2')
cell_y2 = xw.Range('C2')
for cnt in range(DATA_NUM):
    cell_x.value = data_x[cnt]
    cell_y1.value = data_y1[cnt]
    cell_y2.value = data_y2[cnt]

    cell_x = cell_x.offset(1, 0)
    cell_y1 = cell_y1.offset(1, 0)
    cell_y2 = cell_y2.offset(1, 0)

# グラフの挿入
chart = xw.Chart()
# グラフ位置・サイズの調整
chart.left = 200
chart.top = 10
chart.width = 300
chart.height = 200
# グラフ種類の設定
chart.chart_type = 'xy_scatter_lines_no_markers'
# データ範囲の設定
chart.set_source_data(xw.Range('A1:C11'))
# x軸の位置を下端に変更(xlminimum=4　←VBAの定数を読み込む方法わかる方教えてください)
# api[1]の1の意味がわかる方教えてください。とりあえず1にしておけば動作するのですが・・・。
chart.api[1].Axes(AxisType.xlValue).Crosses = 4
# 軸目盛りを内向きに変更(xlInside=2)
chart.api[1].Axes(AxisType.xlCategory).MajorTickMark = 2
chart.api[1].Axes(AxisType.xlValue).MajorTickMark = 2

問題

Pythonのrequestsで日本語が含まれているJSON形式のデータを取得時の文字化け問題

解決策

r = requests.get(url)
r.headers = {"Content-Type": "application/json; charset=utf-8"}
datas = r.json()
print(datas)

自分はこれで日本語が正しく出力できました。(macOS)

問題

Pythonのrequestsで日本語が含まれているJSON形式のデータを取得時の文字化け問題

解決策

下記の行を追加したら日本語が正しく表示されました〜

r.headers = {"Content-Type": "application/json; charset=utf-8"}

こんな感じ↓

r = requests.get(url)
r.headers = {"Content-Type": "application/json; charset=utf-8"}
datas = r.json()
print(datas)

EC2にpython仮想環境構築する際のエラー（No such file or directory）の対処法

AWSの公式ページに従ってEC2インスタンスにpython3の仮想環境を構築する際に出たエラーの対応方法。(boto3インストールの事前準備)

仮想環境の構築はvenvを使用。

エラー内容

エラーコード

#仮想環境を起動する
[ec2@~]$ source ~/my_app/env/bin/activatesource

#エラー
-bash: /home/ec2-user/my_app/env/bin/activatesource: No such file or directory

指定ディレクトリに「activatesource」が存在しない。

対処法

「activatesource」→「activate」にする。

対処法

#activateに修正し仮想環境を有効化
[ec2@ip- ~]$ source ~/my_app/env/bin/activate

#仮想環境(env)に入れた
(env) [ec2~]$

要因

チュートリアルどおりに仮想環境用のフォルダを作成したが、作成されたのはactivatesourceではなくactivateだった。

▼仮想環境用の作成
[ec2~]$ python3 -m venv my_app/env

▼作成したフォルダ内のリスト（~/my_app/env/bin/配下）

[ec2~ bin]$ ls
activate      activate.fish  easy_install-3.7  pip3    python
activate.csh  easy_install   pip               pip3.7  python3

VARISTAって?？

クラウド上で機械学習を行えるSaaSサービスです。
https://www.varista.ai/

DataRobotやGCPのAutoML Tableのかっこいい版といった感じです。
しかも基本無料で使えるため、誰でもすぐに利用可能です。

さっそく⭐️

Kaggle、Signateなどのデータサイエンスコンペサイトでは通常、
カーネルやjupyter、自前のpythonコードなどを利用して参加していると思います。

そこで今回はそれらのコンペに参加する上でVARISTAをつかうことで、どんなことができるのかを試してみました。
データは住宅価格予測（回帰）とタイタニック（分類）を利用して確認します。

1. データの確認?

VARISTAでは、ドラッグアンドドロップでデータをアップロードするだけで
データの自動解析が行われ、ある程度の情報を一瞬で確認することができます。

アップロードが完了すると、データの情報が表示されます。

ここでチェックするのは、各列の欠損数や、列のタイプくらいですかね。
IDなどの学習に利用しない列はトグルでOFFにしておきます。

また、となりのタブにある「集計情報」からは、各列の統計情報を確認することができます。
pandas. describe()の結果をGUIで確認できる感じです。
ちょっとデータを確認したいだけなのに、わざわざ数行のスクリプト用意するのめんどくさいとき結構ありますよね。

2. データの可視化?

KaggleなどでVARISTAを利用する上で最も重宝するのが、このビジュアライズ機能です！
pythonライブラリでもmatplotlibやseabornなど、データを可視化するライブラリはいろいろあると思います。
それらと比較して、何がいいか。
それは勝手にクールなグラフを生成してくれることです！
これからVARISTAが自動生成してくれる、各グラフをご紹介します。

まずはヒストグラム

こんな感じで各列ごとのヒストグラムを確認することができます。
もう一度言いますが、全自動です。なにもしてません。

つぎに相関図

各列と予測する列との相関関係が一眼でわかります。
数値列の場合、近似一時式の直線が表示されていたり、相関度が顔文字で表現されていたりします。
カテゴリ列の場合、最大、最小などのボックスチャートとなり表示されています。
自動でこれらの表示を分けて生成してくれるのがすばらしいですね。

また、これらのグラフは回帰データのもので分類問題だと以下のように表示が変わります。

タイタニックのデータでは、各数値、カテゴリごとの生存率が可視化されています。
これにより、「年齢が若ければ若いほど生存率が高い」などが一目瞭然です。
未知のデータに対しても、この機能だけでわかるインサイトは結構あるとおもいます。

ヒートマップ

おなじみのヒートマップです。

正直あまりみないですが、とりあえず可視化しますよね（自分だけか）

3. オート学習機能?

VARISTAはAutoMLツールなので、もちろん学習も可能です。
ですが、Kagglerの皆さんはもちろん精度を求めて、自分なりのパイプラインで学習を行うと思います。
だからといってVARISTAの学習は無駄か？というとそういうわけでもありません。
学習結果から得られるインサイトが結構あるのです。

学習が完了すると、このような結果画面が確認できます。
右側にあるFeature Importanceをみれば重要な変数が確認できます。

また、詳細タブを開くとより細かい学習結果が確認可能です。
VARISTAでは欠損値の補完も自動的に探索するのですが、その結果が一覧で表示されています。
データの欠損値補完を悩んでる場合は、ここを参考にしてもいいと思います。

さらにVARISTAではアンサンブル学習に対応しているため、いくつかのアルゴリズムで学習させた結果も残っています。
これを確認することで、どのアルゴリズムで最もスコアが出たのかを確認することができます。

おわり?

VARISTAではこれらの機能が無料で利用可能です。
スクリプトを書いてもいいけど、とりあえずデータを確認したいってときに使えるツールであることがわかるかと思います。
みなさんも是非お試しください?

https://www.varista.ai/

はじめに

TensorFlow は Google が開発しているディープラーニングのフレームワークです。
2019年の10月1日のメジャーアップデートを経て 2 系となり、 2020年5月23日現在のバージョンは 2.2.0 です。

TensorFlow は OSS 開発の過程で異なるフレームワーク（Keras）との統合や開発者の趣向反映（Define and Run → Define by Run）を経ている背景もあり、モデルの構築や訓練を様々な書き方で実現できます。

これは便利である反面、初学者にとっては理解を妨げる要因になり得ます。

今回は 2 系の TensorFlow で推奨されている記法を網羅的に紹介し、画像認識の分野で著名なモデルである VGG16 と ResNet50 を実装することで以下の達成レベルを目指します。

• 2 系の TensorFlow で記述されたソースコードからモデルの形状を把握できる
• 2 系の TensorFlow を利用して VGG シリーズ、 ResNet シリーズを独力で実装できる

対象読者

• TensorFlow のチュートリアルを試したが、自力でモデルを構築できない人
• TensorFlow で書かれたソースコードを読むと、見慣れない書き方があると感じる人
• Chainer、PyTorch でモデルを書くことができるが TensorFlow でモデルを書けない人

対象でない読者

• TensorFlow の Subclass API / Functional API / Sequential API について理解している人
• TensorFlow の built-in 訓練とカスタム訓練について理解している人

流れ

まず、 TensorFlow の 4 つのモデル構築 API について説明します。
その後、 2 つの訓練手法について説明します。
最後にこれらの手法を利用して VGG16 と ResNet50 の実装を行います。

TensorFlow における 4 つのモデル構築 API

TensorFlow ではモデルを構築するために、大きく分けて 2 つ、細かく分けると 4 つの API が用意されています。

• シンボリック（宣言型）API
  • Sequential API
  • Functional API
  • Primitive API（1.X 系の書き方。現在は非推奨）
• 命令型（モデル サブクラス化）API
  • Subclassing API

まず大きな分類について簡単に紹介します。

シンボリック（宣言型） API

モデルの形状を学習実行前に宣言（定義）する書き方です。
宣言とはコンパイルのようなものだと思ってください。

この記法で書かれたモデルは学習中に形状を変更することはできません。
そのため、一部の動的に形が変わるモデル（Tree-RNN など）は実装することができません。
その代わりに、学習を実行しなくてもモデル形状の確認ができるようになります。

命令型（モデル サブクラス化）API

シンボリック API とは異なり、宣言をしない命令的（≒直感的）な書き方です。

日本（Preferred Networks）発祥のディープラーニングフレームワークである Chainer が最初に採用した記法で、PyTorch もこの記法を採用しています。
Python でクラスを書くようにモデルを実装することができるため、層の変更や拡張などカスタマイズが容易です。
その代わりに、一度モデルにデータを与えるまでモデルがどのような形状なのかをプログラム側からは認識することができません。

続いて、具体的な書き方について簡単な例を交えて紹介します。

Sequential API

その名の通り、Sequential（連続的）に層を追加してモデルを実装する API です。
Keras や TensorFlow のチュートリアルでもこの書き方が使われることが多いため、一度は見たことがあるのではないでしょうか。

以下のように、空の tensorflow.keras.Sequential クラスをインスタンス化した後に add メソッドで層を追加していく方法と、tensorflow.keras.Sequential クラスの引数にリストとして層を与えてインスタンス化する方法が一般的です。

import tensorflow as tf
from tensorflow.keras import layers

def sequential_vgg16_a(input_shape, output_size):
    model = tf.keras.Sequential()
    model.add(layers.Conv2D(64, 3, 1, padding="same", batch_input_shape=input_shape))
    model.add(layers.BatchNormalization())
    # ...(中略)...
    model.add(layers.Dense(output_size, activation="softmax"))    
    return model

def sequential_vgg16_b(input_shape, output_size):
    model = tf.keras.Sequential([
        layers.Conv2D(64, 3, 1, padding="same", batch_input_shape=input_shape),
        layers.BatchNormalization(),
        # ...(中略)...
        layers.Dense(output_size, activation="softmax")
    ]
    return model

レイヤーを追加するメソッドのみをサポートしているため、入力、中間特徴、出力が複数になる、あるいは条件分岐が存在するような複雑なネットワークを記述することはできません。
層を順番に通していくだけの（VGG のような）シンプルなネットワークを実装する際にこの記法を利用できます。

Functional API

Sequential API では記述できない複雑なモデルを実装する API です。

まず tensorflow.keras.layers.Input をインスタンス化し、最初の層に渡します。
その後、ある層の出力を次の層へと渡していくことでモデルのデータフローを定義していきます。
最後に、得られた出力と最初の入力を tensorflow.keras.Model の引数として与えることでモデルを構築できます。

from tensorflow.keras import layers, Model

def functional_vgg16(input_shape, output_size, batch_norm=False):
    inputs = layers.Input(batch_input_shape=input_shape)

    x = layers.Conv2D(64, 3, 1, padding="same")(x)
    if batch_norm:
        x = layers.BatchNormalization()(x)
    x = layers.ReLU()(x)
    # ...(中略)...
    outputs = layers.Dense(output_size, activation="softmax")(x)

    return Model(inputs=inputs, outputs=outputs)    

上述の例では変数 batch_norm の値によって Batch Normalization 層の有無を切り替えています。
このように条件によってモデルの形状を変えるような柔軟な定義が必要な場合は Sequential API ではなく Functional API が必要になります。

なお、カッコの後にカッコが続く一見奇妙な書き方が登場しますが、これは TensorFlow 特有ではなく Python で一般的に利用できる書き方で、以下の 2 つは同じ処理を表します。

# 書き方 1
x = layers.BatchNormalization()(x)

# 書き方 2
layer = layers.BatchNormalization()
x = layer(x)

Primitive API

TensorFlow 1.X 系で主に利用されていた記法です。
2.X 系の現在は非推奨となっています。

上述の Sequential API と Functional API はモデルを通るデータのフローを記述していくことでモデルを定義することができましたが、 Primitive API ではその他の計算処理を含む全体の処理フローを宣言的に記述します。

今からこの書き方を覚えるメリットはあまりないので説明は省きますが、tensorflow.Session を利用して訓練を行っている場合はこの書き方に該当します。

import tensorflow as tf
sess = tf.Session()

Subclassing API

TensorFlow 2 系へのアップデートと共に利用可能になった API です。
Chainer や PyTorch とほとんど同じ書き方であり、Python でクラスを書くようにモデルを実装することができるので直感的でカスタマイズが容易です。

まず tensorflow.keras.Model を継承してクラスを作ります。
その後、 __init__ メソッドと call メソッドを実装することでモデルを構築します。

クラス内の __init__ メソッドでは親クラスの__init__メソッド呼び出しと学習したいレイヤーの登録を行います。ここに記載していないレイヤーの重みはデフォルトでは学習対象になりません。

クラス内のcall メソッドではレイヤーの順伝播を記載します。（Chainer の __call__、PyTorch の forwardと同じようなものです。）

from tensorflow.keras import layers, Model

class VGG16(Model):
    def __init__(self, output_size=1000):
        super().__init__()
        self.layers_ = [
            layers.Conv2D(64, 3, 1, padding="same")
            layers.BatchNormalization(),
            # ...(中略)...
            layers.Dense(output_size, activation="softmax")
        ]
    def call(self, inputs):
        for layer in self.layers_:
            inputs = layer(inputs)
        return inputs

他の書き方と比べると少々冗長にも見えますが、普通にクラスを書くようにモデルを実装できることがわかります。

なお、親クラスの初期化を行う super メソッドには引数を与えるパターンもありますが、これは 2 系の Python を考慮した書き方であり、 3 系の Python では引数なしでも同じ処理になります。

from tensorflow.keras import Model

# Python 3 系の書き方
class VGG16(Model):
    def __init__(self, output_size=1000):
        super().__init__()

# Python 2 系の書き方
class VGG16(Model):
    def __init__(self, output_size=1000):
        super().__init__(VGG16, self)

モデル構築 API 振り返り

モデル構築方法の説明は以上になります。
まとめると、以下のような使い分けができるかと思います。

• 層を一方的に通るだけのモデルを簡単に書きたい場合は Sequential API
• 複雑なモデルを学習実行前にきちんと形状確認できるように書きたい場合は Functional API
• Chainer や PyTorch 流の書き方で書きたい、あるいは動的なモデルを書きたい場合は Subclassing API

TensorFlow における 2 つの訓練方法

訓練を行う方法としては以下の 2 つが存在します。

• built-in 訓練
• カスタム訓練

※ 正式な名称があるわけではなさそうなので、本記事では便宜的に上記の名称を利用しています。

built-in 訓練

tensorflow.keras.Model の built-in function を利用して訓練を行う方法です。

Keras、 TensorFlow のチュートリアルでも利用されているためご存知の方が多いかと思います。
また、異なるライブラリですが scikit-learn でもこの方法が採用されています。

まず、上述の API で実装したモデル（tensorflow.keras.Model、あるいはこれを継承したオブジェクト）をインスタンス化します。

このインスタンスは built-in function として compile メソッドと fit メソッドを持っています。

この compile メソッドを実行して損失関数、最適化関数、評価指標を登録します。
その後 fit メソッドを実行することで訓練を行います。

import tensorflow as tf

(train_images, train_labels), _ = tf.keras.datasets.cifar10.load_data()

# 例示のため学習済みのモデルを使っています
model = tf.keras.applications.VGG16()

model.compile(
    optimizer=tf.keras.optimizers.Adam(),
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"],
)

model.fit(train_images, train_labels)

これで訓練が実行できます。

バッチサイズの指定、エポック数、コールバック関数の登録、バリデーションデータでの評価などは、fitメソッドのキーワード引数として登録することができるため、ある程度のカスタマイズも可能です。

多くの場合はこれで十分対応できるかと思いますが、この枠にハマらないケース（例えば GAN など複数のモデルを同時に訓練するケース）は後述するカスタム訓練で記述する必要があります。

カスタム訓練

これは特別 API が用意されているというわけではなく、普通に Python の for ループで訓練する方法です。

まず、上述の API で実装したモデル（tensorflow.keras.Model、あるいはこれを継承したオブジェクト）をインスタンス化します。

次に、損失関数、最適化関数の定義に加えてデータセットのバッチ化を行います。
その後、for ループでエポック、バッチを回していきます。

for ループ内ではまず tf.GradientTape スコープの中に順伝搬の処理を記述します。
その後 gradient メソッドを呼び出して勾配を計算し、apply_gradients メソッドで最適化関数に従って重みを更新しています。

import tensorflow as tf

batch_size = 32
epochs = 10

(train_images, train_labels), _ = tf.keras.datasets.cifar10.load_data()
# 例示のため学習済みのモデルを使っています

model = tf.keras.applications.VGG16()

buffer_size = len(train_images)
train_ds = tf.data.Dataset.from_tensor_slices((train_images, train_labels))
train_ds = train_ds.shuffle(buffer_size=buffer_size).batch(batch_size)

criterion = tf.keras.losses.SparseCategoricalCrossentropy()
optimizer = tf.keras.optimizers.Adam()

for epoch in range(epochs):
    for x, y_true in train_ds:
        with tf.GradientTape() as tape:
            y_pred = model(x, training=True)
            loss = criterion(y_true=y_true, y_pred=y_pred)
        gradients = tape.gradient(loss, model.trainable_variables)
        optimizer.apply_gradients(zip(gradients, model.trainable_variables))

これで訓練が実行できます。

上記の例ではバリデーションデータでの評価や TensorBoard への出力等を一切行っていませんが、普通に for ループを回しているだけなので、好きなように処理を追加していくことができます。

一方で記述量はどうしても多くなるためソースコードの品質担保がやや大変になります。
なお Chainer、PyTorch も（細かい違いはありますが）ほぼ同じ書き方ができます。

訓練方法振り返り

訓練方法の説明は以上になります。
まとめると、以下のような使い分けができるかと思います。

• 訓練中に特殊な処理を実行する必要がない場合、一般的な訓練手法の場合は built-in 訓練
• built-in の枠にハマらない場合、訓練中にいろいろな処理を追加して試行錯誤したい場合、手なりで書きたい場合は カスタム訓練

VGG16／ResNet 50 の概要

説明だけではよく分からない部分もあると思うので、実装を通して理解を深めていきます。

まず簡単に 2 つのモデルについて紹介します。

VGG16 とは

3x3 Convolution を 13 層、全結合層を 3 層持つ非常にシンプルな構造でありながら高性能なモデルです。
様々な画像認識タスクで画像特徴の抽出に利用されます。原論文は 37,000 を超える被引用数を持っており、とても有名です。

Sequential API、 Functional API、 Subclassing API で実装が可能です。

なお、原論文はこちらです。
https://arxiv.org/abs/1409.1556

ResNet50 とは

Residual 機構を持つ多層（Convolution を 49 層、全結合層を 1 層）モデルです。2020 年現在でもこの ResNet の亜種が画像分類の精度においてはトップクラスとなっており、こちらも高性能なモデルです。
VGG16 と同様に様々な画像認識タスクで画像特徴の抽出に利用されます。原論文は 45,000 を超える被引用数（BERT の約 10 倍）を持っており、こちらも非常に有名です。

Sequential API 単体では実装できません。Functional API、 Subclassing API で実装が可能です。

なお、原論文はこちらです。
https://arxiv.org/abs/1512.03385

VGG16 の実装

ではそれぞれの書き方で実装してみます。

VGG16 Sequential API

特に考えることもないので普通に書きます。

from tensorflow.keras import layers, Sequential


def sequential_vgg16(input_shape, output_size):
    params = {
        "padding": "same",
        "use_bias": True,
        "kernel_initializer": "he_normal",
    }
    model = Sequential()
    model.add(layers.Conv2D(64, 3, 1, **params, batch_input_shape=input_shape))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.Conv2D(64, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.MaxPool2D(2, padding="same"))
    model.add(layers.Conv2D(128, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.Conv2D(128, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.MaxPool2D(2, padding="same"))
    model.add(layers.Conv2D(256, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.Conv2D(256, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.Conv2D(256, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.MaxPool2D(2, padding="same"))
    model.add(layers.Conv2D(512, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.Conv2D(512, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.Conv2D(512, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.MaxPool2D(2, padding="same"))
    model.add(layers.Conv2D(512, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.Conv2D(512, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.Conv2D(512, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.MaxPool2D(2, padding="same"))
    model.add(layers.Flatten())
    model.add(layers.Dense(4096))
    model.add(layers.Dense(4096))
    model.add(layers.Dense(output_size, activation="softmax"))
    return model

かなり単純に書くことができますが、層が多いので見るのが辛いことがわかります。
例えば、どこかで ReLU が抜けてても気付かなそうです。
また、例えば Batch Normalization を無くしたい、と思った場合は 1 行ずつコメントアウトしていく必要があり、再利用生やカスタマイズ性に乏しいです。

VGG16 Functional API

Sequential API と比べて柔軟に書くことができます。
今回は再利用される層のまとまり(Convolution - BatchNormalization - ReLU)を関数にしてみます。

from tensorflow.keras import layers, Model


def functional_cbr(x, filters, kernel_size, strides):
    params = {
        "filters": filters,
        "kernel_size": kernel_size,
        "strides": strides,
        "padding": "same",
        "use_bias": True,
        "kernel_initializer": "he_normal",
    }

    x = layers.Conv2D(**params)(x)
    x = layers.BatchNormalization()(x)
    x = layers.ReLU()(x)
    return x


def functional_vgg16(input_shape, output_size):
    inputs = layers.Input(batch_input_shape=input_shape)
    x = functional_cbr(inputs, 64, 3, 1)
    x = functional_cbr(x, 64, 3, 1)
    x = layers.MaxPool2D(2, padding="same")(x)
    x = functional_cbr(x, 128, 3, 1)
    x = functional_cbr(x, 128, 3, 1)
    x = layers.MaxPool2D(2, padding="same").__call__(x)  # こう書いても良い
    x = functional_cbr(x, 256, 3, 1)
    x = functional_cbr(x, 256, 3, 1)
    x = functional_cbr(x, 256, 3, 1)
    x = layers.MaxPool2D(2, padding="same").call(x)  # こう書いても良い
    x = functional_cbr(x, 512, 3, 1)
    x = functional_cbr(x, 512, 3, 1)
    x = functional_cbr(x, 512, 3, 1)
    x = layers.MaxPool2D(2, padding="same")(x)
    x = functional_cbr(x, 512, 3, 1)
    x = functional_cbr(x, 512, 3, 1)
    x = functional_cbr(x, 512, 3, 1)
    x = layers.MaxPool2D(2, padding="same")(x)
    x = layers.Flatten()(x)
    x = layers.Dense(4096)(x)
    x = layers.Dense(4096)(x)
    outputs = layers.Dense(output_size, activation="softmax")(x)
    return Model(inputs=inputs, outputs=outputs)

だいぶスッキリ書くことができました。
BatchNormalization を無くしたくなっても、 ReLU を LeaklyReLU に変えたくなっても数行の改修で済みます。

VGG16 Subclassing API

Functional API と同様に再利用される層のまとまり（Convolution - BatchNormalization - ReLU）をクラスにして書いてみます。

from tensorflow.keras import layers, Model


class CBR(Model):
    def __init__(self, filters, kernel_size, strides):
        super().__init__()

        params = {
            "filters": filters,
            "kernel_size": kernel_size,
            "strides": strides,
            "padding": "same",
            "use_bias": True,
            "kernel_initializer": "he_normal",
        }

        self.layers_ = [
            layers.Conv2D(**params),
            layers.BatchNormalization(),
            layers.ReLU()
        ]

    def call(self, inputs):
        for layer in self.layers_:
            inputs = layer(inputs)
        return inputs


class VGG16(Model):
    def __init__(self, output_size=1000):
        super().__init__()
        self.layers_ = [
            CBR(64, 3, 1),
            CBR(64, 3, 1),
            layers.MaxPool2D(2, padding="same"),
            CBR(128, 3, 1),
            CBR(128, 3, 1),
            layers.MaxPool2D(2, padding="same"),
            CBR(256, 3, 1),
            CBR(256, 3, 1),
            CBR(256, 3, 1),
            layers.MaxPool2D(2, padding="same"),
            CBR(512, 3, 1),
            CBR(512, 3, 1),
            CBR(512, 3, 1),
            layers.MaxPool2D(2, padding="same"),
            CBR(512, 3, 1),
            CBR(512, 3, 1),
            CBR(512, 3, 1),
            layers.MaxPool2D(2, padding="same"),
            layers.Flatten(),
            layers.Dense(4096),
            layers.Dense(4096),
            layers.Dense(output_size, activation="softmax"),
        ]

    def call(self, inputs):
        for layer in self.layers_:
            inputs = layer(inputs)
        return inputs

__init__ でモデルの定義、call でモデルの呼び出しを担当しているため、Functional API よりも直感的に理解しやすいですが、コードは長めになります。
また命令型である Subclassing API はモデル生成時に入力の形状が必要ない（引数に input_shape が必要ない）こともポイントです。

VGG16 実装 振り返り

なるべく比較しやすいように書いたつもりですがいかがだったでしょうか。

今回の実装は Batch Normalization を Convolution 層の間に挟んでいるのと重みの初期化に He の初期化を利用していますが、原論文が提出された時はこれらのテクニックは未だ発表されていなかったので Batch Normalization 層はなく、重みの初期化には Grolot の初期化が使われていました。そのため、原論文では勾配消失を回避するために 7 層のモデルを学習してから層を徐々に付け足していくといった転移学習ライクな学習方法が採用されています。

上記の実装をより深く理解するために Batch Normalization 層をなくすと何が起きるのか、重みの初期化手法を変えると何が起きるのか、などを試してみるのも面白いと思います。

ResNet50 の実装

続いて ResNet50 を実装します。
Sequential API 単体では書けないので Functional API、 Subclassing API で書きます。

ResNet50 Functional API

再利用される Residual 機構を関数化して実装します。

from tensorflow.keras import layers, Model


def functional_bottleneck_residual(x, in_ch, out_ch, strides=1):
    params = {
        "padding": "same",
        "kernel_initializer": "he_normal",
        "use_bias": True,
    }
    inter_ch = out_ch // 4
    h1 = layers.Conv2D(inter_ch, kernel_size=1, strides=strides, **params)(x)
    h1 = layers.BatchNormalization()(h1)
    h1 = layers.ReLU()(h1)
    h1 = layers.Conv2D(inter_ch, kernel_size=3, strides=1, **params)(h1)
    h1 = layers.BatchNormalization()(h1)
    h1 = layers.ReLU()(h1)
    h1 = layers.Conv2D(out_ch, kernel_size=1, strides=1, **params)(h1)
    h1 = layers.BatchNormalization()(h1)

    if in_ch != out_ch:
        h2 = layers.Conv2D(out_ch, kernel_size=1, strides=strides, **params)(x)
        h2 = layers.BatchNormalization()(h2)
    else:
        h2 = x

    h = layers.Add()([h1, h2])
    h = layers.ReLU()(h)
    return h


def functional_resnet50(input_shape, output_size):
    inputs = layers.Input(batch_input_shape=input_shape)
    x = layers.Conv2D(64, 7, 2, padding="same", kernel_initializer="he_normal")(inputs)
    x = layers.BatchNormalization()(x)
    x = layers.MaxPool2D(pool_size=3, strides=2, padding="same")(x)

    x = functional_bottleneck_residual(x, 64, 256)
    x = functional_bottleneck_residual(x, 256, 256)
    x = functional_bottleneck_residual(x, 256, 256)

    x = functional_bottleneck_residual(x, 256, 512, 2)
    x = functional_bottleneck_residual(x, 512, 512)
    x = functional_bottleneck_residual(x, 512, 512)
    x = functional_bottleneck_residual(x, 512, 512)

    x = functional_bottleneck_residual(x, 512, 1024, 2)
    x = functional_bottleneck_residual(x, 1024, 1024)
    x = functional_bottleneck_residual(x, 1024, 1024)
    x = functional_bottleneck_residual(x, 1024, 1024)
    x = functional_bottleneck_residual(x, 1024, 1024)
    x = functional_bottleneck_residual(x, 1024, 1024)

    x = functional_bottleneck_residual(x, 1024, 2048, 2)
    x = functional_bottleneck_residual(x, 2048, 2048)
    x = functional_bottleneck_residual(x, 2048, 2048)

    x = layers.GlobalAveragePooling2D()(x)
    outputs = layers.Dense(
        output_size, activation="softmax", kernel_initializer="he_normal"
    )(x)
    return Model(inputs=inputs, outputs=outputs)

functional_bottleneck_residual メソッド内では h1、h2、h が登場します。
このように、データのフローが途中で分岐するモデルは Sequential API では記述できません。

また、h2 は入出力のチャネル数が同じ場合は何もせず、異なる場合はチャネル数を調整する処理（Projection）を行います。このような条件分岐も Sequential API では記述できません。

このメソッドを作ってしまえば、あとは順々に記述していくだけです。

ResNet50 Subclassing API

Functional API と同様に再利用される Residual 機構をクラス化して実装します。

class BottleneckResidual(Model):
    """ResNet の Bottleneck Residual Module です.
    1 層目の 1x1 conv で ch 次元を削減することで
    2 層目の 3x3 conv の計算量を減らし
    3 層目の 1x1 conv で ch 出力の次元を復元します.
    計算量の多い 2 層目 3x3 conv の次元を小さくすることから bottleneck と呼ばれます.
    """

    def __init__(self, in_ch, out_ch, strides=1):
        super().__init__()

        self.projection = in_ch != out_ch
        inter_ch = out_ch // 4
        params = {
            "padding": "same",
            "kernel_initializer": "he_normal",
            "use_bias": True,
        }

        self.common_layers = [
            layers.Conv2D(inter_ch, kernel_size=1, strides=strides, **params),
            layers.BatchNormalization(),
            layers.ReLU(),
            layers.Conv2D(inter_ch, kernel_size=3, strides=1, **params),
            layers.BatchNormalization(),
            layers.ReLU(),
            layers.Conv2D(out_ch, kernel_size=1, strides=1, **params),
            layers.BatchNormalization(),
        ]

        if self.projection:
            self.projection_layers = [
                layers.Conv2D(out_ch, kernel_size=1, strides=strides, **params),
                layers.BatchNormalization(),
            ]

        self.concat_layers = [layers.Add(), layers.ReLU()]

    def call(self, inputs):
        h1 = inputs
        h2 = inputs

        for layer in self.common_layers:
            h1 = layer(h1)

        if self.projection:
            for layer in self.projection_layers:
                h2 = layer(h2)

        outputs = [h1, h2]
        for layer in self.concat_layers:
            outputs = layer(outputs)
        return outputs


class ResNet50(Model):
    """ResNet50 です.
    要素は
    conv * 1
    resblock(conv * 3) * 3
    resblock(conv * 3) * 4
    resblock(conv * 3) * 6
    resblock(conv * 3) * 3
    dense * 1
    から構成されていて, conv * 49 + dense * 1 の 50 層です.
    """

    def __init__(self, output_size=1000):
        super().__init__()

        self.layers_ = [
            layers.Conv2D(64, 7, 2, padding="same", kernel_initializer="he_normal"),
            layers.BatchNormalization(),
            layers.MaxPool2D(pool_size=3, strides=2, padding="same"),
            BottleneckResidual(64, 256),
            BottleneckResidual(256, 256),
            BottleneckResidual(256, 256),
            BottleneckResidual(256, 512, 2),
            BottleneckResidual(512, 512),
            BottleneckResidual(512, 512),
            BottleneckResidual(512, 512),
            BottleneckResidual(512, 1024, 2),
            BottleneckResidual(1024, 1024),
            BottleneckResidual(1024, 1024),
            BottleneckResidual(1024, 1024),
            BottleneckResidual(1024, 1024),
            BottleneckResidual(1024, 1024),
            BottleneckResidual(1024, 2048, 2),
            BottleneckResidual(2048, 2048),
            BottleneckResidual(2048, 2048),
            layers.GlobalAveragePooling2D(),
            layers.Dense(
                output_size, activation="softmax", kernel_initializer="he_normal"
            ),
        ]

    def call(self, inputs):
        for layer in self.layers_:
            inputs = layer(inputs)
        return inputs

Functional API とそこまで大きな違いはありません。
__init__ レイヤーでリストに層をまとめるような書き方をしていますが、クラスの変数に登録されていれば良いのでこの辺は自由に書くことができます。

ResNet50 実装 振り返り

Sequential API 単体では実装できないモデルとして ResNet50 を紹介しました。
正直大きな違いはないので Functional API、 Subclassing API は好みで使い分けても問題ないかと思います。

訓練の実装

最後に訓練ループの実装を比較してみます。

全てのソースコードを載せるとかなり長くなるため部分的にメソッドを src.utils に切り出しています。
そこまで複雑なことをしているわけではないので補完しながら読んでいただければ助かります。

一応全てのソースは以下のリポジトリにあるので気になる方はご覧ください。
https://github.com/Anieca/deep-learning-models

built-in 訓練 実装

テストデータの精度算出、TensorBoard 用のログ出力などいくつかオプションを指定してみます。

import os
import tensorflow as tf

from src.utils import load_dataset, load_model, get_args, get_current_time


def builtin_train(args):
    # 1. load dataset and model
    (train_images, train_labels), (test_images, test_labels) = load_dataset(args.data)
    input_shape = train_images[: args.batch_size, :, :, :].shape
    output_size = max(train_labels) + 1
    model = load_model(args.arch, input_shape=input_shape, output_size=output_size)
    model.summary()

    # 2. set tensorboard cofigs
    logdir = os.path.join(args.logdir, get_current_time())
    tensorboard_callback = tf.keras.callbacks.TensorBoard(log_dir=logdir)

    # 3. loss, optimizer, metrics setting
    model.compile(
        optimizer=tf.keras.optimizers.Adam(),
        loss="sparse_categorical_crossentropy",
        metrics=["accuracy"],
    )

    # 4. dataset config (and validation, callback config)
    fit_params = {}
    fit_params["batch_size"] = args.batch_size
    fit_params["epochs"] = args.max_epoch
    if args.steps_per_epoch:
        fit_params["steps_per_epoch"] = args.steps_per_epoch
    fit_params["verbose"] = 1
    fit_params["callbacks"] = [tensorboard_callback]
    fit_params["validation_data"] = (test_images, test_labels)

    # 5. start train and test
    model.fit(train_images, train_labels, **fit_params)

かなりシンプルに書けます。

コールバック関数は他にも色々あるので興味がある方はドキュメントを読んでみてください。
https://www.tensorflow.org/api_docs/python/tf/keras/callbacks

カスタム訓練 実装

上記の built-in 訓練と同じ処理を自分で実装してみます。

import os
import tensorflow as tf

from src.utils import load_dataset, load_model, get_args, get_current_time


def custom_train(args):
    # 1. load dataset and model
    (train_images, train_labels), (test_images, test_labels) = load_dataset(args.data)
    input_shape = train_images[: args.batch_size, :, :, :].shape
    output_size = max(train_labels) + 1
    model = load_model(args.arch, input_shape=input_shape, output_size=output_size)
    model.summary()

    # 2. set tensorboard configs
    logdir = os.path.join(args.logdir, get_current_time())
    train_writer = tf.summary.create_file_writer(os.path.join(logdir, "train"))
    test_writer = tf.summary.create_file_writer(os.path.join(logdir, "test"))

    # 3. loss, optimizer, metrics setting
    criterion = tf.keras.losses.SparseCategoricalCrossentropy()
    optimizer = tf.keras.optimizers.Adam()
    train_loss_avg = tf.keras.metrics.Mean()
    train_accuracy = tf.keras.metrics.SparseCategoricalAccuracy()
    test_loss_avg = tf.keras.metrics.Mean()
    test_accuracy = tf.keras.metrics.SparseCategoricalAccuracy()

    # 4. dataset config
    buffer_size = len(train_images)
    train_ds = tf.data.Dataset.from_tensor_slices((train_images, train_labels))
    train_ds = train_ds.shuffle(buffer_size=buffer_size).batch(args.batch_size)
    test_ds = tf.data.Dataset.from_tensor_slices((test_images, test_labels))
    test_ds = test_ds.batch(args.batch_size)

    # 5. start train and test
    for epoch in range(args.max_epoch):
        # 5.1. initialize metrics
        train_loss_avg.reset_states()
        train_accuracy.reset_states()
        test_loss_avg.reset_states()
        test_loss_avg.reset_states()

        # 5.2. initialize progress bar
        train_pbar = tf.keras.utils.Progbar(args.steps_per_epoch)
        test_pbar = tf.keras.utils.Progbar(args.steps_per_epoch)

        # 5.3. start train
        for i, (x, y_true) in enumerate(train_ds):
            if args.steps_per_epoch and i >= args.steps_per_epoch:
                break
            # 5.3.1. forward
            with tf.GradientTape() as tape:
                y_pred = model(x, training=True)
                loss = criterion(y_true=y_true, y_pred=y_pred)
            # 5.3.2. calculate gradients from `tape` and backward
            gradients = tape.gradient(loss, model.trainable_variables)
            optimizer.apply_gradients(zip(gradients, model.trainable_variables))

            # 5.3.3. update metrics and progress bar
            train_loss_avg(loss)
            train_accuracy(y_true, y_pred)
            train_pbar.update(
                i + 1,
                [
                    ("avg_loss", train_loss_avg.result()),
                    ("accuracy", train_accuracy.result()),
                ],
            )

        # 5.4. start test
        for i, (x, y_true) in enumerate(test_ds):
            if args.steps_per_epoch and i >= args.steps_per_epoch:
                break
            # 5.4.1. forward
            y_pred = model(x)
            loss = criterion(y_true, y_pred)

            # 5.4.2. update metrics and progress bar
            test_loss_avg(loss)
            test_accuracy(y_true, y_pred)
            test_pbar.update(
                i + 1,
                [
                    ("avg_test_loss", test_loss_avg.result()),
                    ("test_accuracy", test_accuracy.result()),
                ],
            )

        # 5.5. write metrics to tensorboard
        with train_writer.as_default():
            tf.summary.scalar("Loss", train_loss_avg.result(), step=epoch)
            tf.summary.scalar("Acc", train_accuracy.result(), step=epoch)
        with test_writer.as_default():
            tf.summary.scalar("Loss", test_loss_avg.result(), step=epoch)
            tf.summary.scalar("Acc", test_accuracy.result(), step=epoch)

訓練開始まではそこまで変わらないですが、訓練ループ内（コメントの 5. ）の記述量がかなり多くなります。

訓練実装 振り返り

自分で TensorBoard 出力の管理やプログレスバーの作成と言ったユーティリティを管理するのはそれなりにコストがかかりますが、 built-in はかなり楽に使うことができます。

built-in に用意されていない処理を記述したい場合はカスタム訓練で書く必要がありますが、そうでなければ built-in を使った方が良さそうです。

終わりに

以上です。お疲れ様でした。

TensorFlow 2 系の色々な書き方について実装を交えて紹介しました。

あまりそれぞれの書き方について優劣はつけずにフラットに書いたつもりです。

自分で書くときは状況や好みに合わせて書けば良いと思いますが、ソースコードを探していると色々な書き方に出会うので、なんとなく全ての書き方を理解していると良いかと思います。

皆様のお役に立てれば幸いです。

はじめに

Google が開発している 深層学習ライブラリである TensorFlow はモデルの構築や訓練ループを様々な書き方で実現できます。これは有識者にとっては便利ですが、初学者にとっては理解を妨げる要因になり得ます。

今回は TensorFlow 2.X で推奨されている書き方を網羅的に紹介し、画像認識の分野で著名なモデルである VGG16 と ResNet50 を実装しながら使い方を解説します。

※ TensorFlow 2.X とはメジャーバージョンが 2 以上の TensorFlow を指すものとします。

対象読者

• TensorFlow のチュートリアルを試したが、自力でモデルを構築できない人
• TensorFlow で書かれたソースコードを読むと、見慣れない書き方があると感じる人
• Chainer、PyTorch でモデルを書くことができるが TensorFlow でモデルを書けない人

流れ

まず、 TensorFlow の 4 つのモデル構築 API について説明します。
その後、 2 つの訓練手法について説明します。
最後にこれらの書き方を利用して VGG16 と ResNet50 の実装を行います。

検証環境

• macOS Catalina 10.15.3
• Python 3.7.7
• tensorflow 2.2.0

>>> import sys
>>> sys.version
'3.7.7 (default, Mar 10 2020, 15:43:33) \n[Clang 11.0.0 (clang-1100.0.33.17)]'

pip list | grep tensorflow
tensorflow               2.2.0
tensorflow-estimator     2.2.0

TensorFlow における 4 つのモデル構築 API

TensorFlow にはモデルを構築するために大きく分けて 2 つ、細かく分けると 4 つの API が用意されています。

• シンボリック（宣言型）API
  • Sequential API
  • Functional API
  • Primitive API（1.X 系の書き方。非推奨）
• 命令型（モデル サブクラス化）API
  • Subclassing API

まず大きな分類について簡単に紹介します。

シンボリック（宣言型） API

モデルの形状を学習実行前に宣言（≒コンパイル）する書き方です。

この API で書かれたモデルは学習中に形状を変更できません。
そのため、一部の動的に形が変わるモデル（Tree-RNN など）は実装できません。
その代わりに、データをモデルに与える前からモデル形状の確認ができます。

命令型（モデル サブクラス化）API

シンボリック API とは異なり、宣言をしない命令的（≒直感的）な書き方です。

日本（Preferred Networks）発祥の深層学習ライブラリである Chainer が最初に採用した書き方で、PyTorch もこの書き方を採用しています。
Python でクラスを書くようにモデルを実装することができるため、層の変更や拡張などのカスタマイズが容易です。
その代わりに、一度データを与えるまでモデルがどのような形状なのかをプログラム側からは認識することができません。

続いて、具体的な書き方について簡単な例を交えて紹介します。

Sequential API

その名の通り、Sequential（連続的）に層を追加してモデルを実装する API です。
Keras や TensorFlow のチュートリアルでもこの書き方が使われることが多いため、一度は見たことがあるのではないでしょうか。

以下のように、空の tensorflow.keras.Sequential クラスをインスタンス化した後に add メソッドで層を追加していく方法と、tensorflow.keras.Sequential クラスの引数にリストとして層を与えてインスタンス化する方法が一般的です。

import tensorflow as tf
from tensorflow.keras import layers

def sequential_vgg16_a(input_shape, output_size):
    model = tf.keras.Sequential()
    model.add(layers.Conv2D(64, 3, 1, padding="same", batch_input_shape=input_shape))
    model.add(layers.BatchNormalization())
    # ...(中略)...
    model.add(layers.Dense(output_size, activation="softmax"))    
    return model

def sequential_vgg16_b(input_shape, output_size):
    model = tf.keras.Sequential([
        layers.Conv2D(64, 3, 1, padding="same", batch_input_shape=input_shape),
        layers.BatchNormalization(),
        # ...(中略)...
        layers.Dense(output_size, activation="softmax")
    ]
    return model

レイヤーを追加するメソッドのみをサポートしているため、入力、中間特徴、出力が複数になる、あるいは条件分岐が存在するような複雑なネットワークを記述することはできません。
層を順番に通していくだけの（VGG のような）シンプルなネットワークを実装する際にこの API を利用できます。

Functional API

Sequential API では記述できない複雑なモデルを実装する API です。

まず tensorflow.keras.layers.Input をインスタンス化し、最初の層に渡します。
その後、ある層の出力を次の層へと渡していくことでモデルのデータフローを定義していきます。
最後に、得られた出力と最初の入力を tensorflow.keras.Model の引数として与えることでモデルを構築できます。

from tensorflow.keras import layers, Model

def functional_vgg16(input_shape, output_size, batch_norm=False):
    inputs = layers.Input(batch_input_shape=input_shape)

    x = layers.Conv2D(64, 3, 1, padding="same")(inputs)
    if batch_norm:
        x = layers.BatchNormalization()(x)
    x = layers.ReLU()(x)
    # ...(中略)...
    outputs = layers.Dense(output_size, activation="softmax")(x)

    return Model(inputs=inputs, outputs=outputs)    

上述の例では変数 batch_norm の値によって Batch Normalization 層の有無を切り替えています。
このように条件によってモデルの形状を変えるような柔軟な定義が必要な場合は Sequential API ではなく Functional API が必要になります。

なお、カッコの後にカッコが続く一見奇妙な書き方が登場しますが、これは TensorFlow 特有ではなく Python で一般的に利用できる書き方で、以下の 2 つは同じ処理を表します。

# 書き方 1
x = layers.BatchNormalization()(x)

# 書き方 2
layer = layers.BatchNormalization()
x = layer(x)

Primitive API

TensorFlow 1.X 系で主に利用されていた API です。
2.X 系の現在は非推奨となっています。

上述の Sequential API と Functional API はモデルを通るデータのフローを記述していくことでモデルを定義することができましたが、 Primitive API ではその他の計算処理を含む全体の処理フローを宣言的に記述します。

今からこの書き方を覚えるメリットはあまりないので説明は省きますが、tensorflow.Session を利用して訓練を行っている場合はこの書き方に該当します。

import tensorflow as tf
sess = tf.Session()

Subclassing API

TensorFlow 2.X へのアップデートと共に利用可能になった API です。
Chainer や PyTorch とほとんど同じ書き方であり、Python でクラスを書くようにモデルを実装することができるので直感的でカスタマイズが容易です。

まず tensorflow.keras.Model を継承してクラスを作ります。
その後、 __init__ メソッドと call メソッドを実装することでモデルを構築します。

クラス内の __init__ メソッドでは親クラスの__init__メソッド呼び出しと学習したいレイヤーの登録を行います。ここに記載していないレイヤーの重みはデフォルトでは学習対象になりません。

クラス内のcall メソッドではレイヤーの順伝播を記載します。（Chainer の __call__、PyTorch の forwardと同じようなものです。）

from tensorflow.keras import layers, Model


class VGG16(Model):
    def __init__(self, output_size=1000):
        super().__init__()
        self.layers_ = [
            layers.Conv2D(64, 3, 1, padding="same"),
            layers.BatchNormalization(),
            # ...(中略)...
            layers.Dense(output_size, activation="softmax"),
        ]
    def call(self, inputs):
        for layer in self.layers_:
            inputs = layer(inputs)
        return inputs

他の書き方と比べると少々冗長にも見えますが、普通にクラスを書くようにモデルを実装できることがわかります。

なお、親クラスの初期化を行う super メソッドには引数を与えるパターンもありますが、これは 2 系の Python を考慮した書き方であり、 3 系の Python では引数なしでも同じ処理になります。

from tensorflow.keras import Model


# Python 3 系の書き方
class VGG16_PY3(Model):
    def __init__(self, output_size=1000):
        super().__init__()

# Python 2 系の書き方
class VGG16_PY2(Model):
    def __init__(self, output_size=1000):
        super().__init__(VGG16_PY2, self)

モデル構築 API 振り返り

モデル構築方法の説明は以上になります。
まとめると、以下のような使い分けができるかと思います。

• 層を一方的に通るだけのモデルを簡単に書きたい場合は Sequential API
• 複雑なモデルを学習実行前にきちんと形状確認できるように書きたい場合は Functional API
• Chainer や PyTorch 流の書き方で書きたい、あるいは動的なモデルを書きたい場合は Subclassing API

TensorFlow における 2 つの訓練方法

訓練を行う方法としては以下の 2 つが存在します。

• built-in 訓練
• カスタム訓練

※ 正式な名称があるわけではなさそうなので、本記事では便宜的に上記の名称を利用しています。

built-in 訓練

tensorflow.keras.Model の built-in function を利用して訓練を行う方法です。

Keras、 TensorFlow のチュートリアルでも利用されているためご存知の方が多いかと思います。
また、異なるライブラリですが scikit-learn でもこの方法が採用されています。

まず、上述の API で実装したモデル（tensorflow.keras.Model、あるいはこれを継承したオブジェクト）をインスタンス化します。

このインスタンスは built-in function として compile メソッドと fit メソッドを持っています。

この compile メソッドを実行して損失関数、最適化関数、評価指標を登録します。
その後 fit メソッドを実行することで訓練を行います。

import tensorflow as tf

(train_images, train_labels), _ = tf.keras.datasets.cifar10.load_data()

# 例示のため学習済みのモデルを使っています
model = tf.keras.applications.VGG16()

model.compile(
    optimizer=tf.keras.optimizers.Adam(),
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"],
)

model.fit(train_images, train_labels)

これで訓練が実行できます。

バッチサイズの指定、エポック数、コールバック関数の登録、バリデーションデータでの評価などは、fitメソッドのキーワード引数として登録することができるため、ある程度のカスタマイズも可能です。

多くの場合はこれで十分対応できるかと思いますが、この枠にハマらないケース（例えば GAN など複数のモデルを同時に訓練するケース）は後述するカスタム訓練で記述する必要があります。

カスタム訓練

これは特別 API が用意されているというわけではなく、普通に Python の for ループで訓練する方法です。

まず、上述の API で実装したモデル（tensorflow.keras.Model、あるいはこれを継承したオブジェクト）をインスタンス化します。

次に、損失関数、最適化関数の定義に加えてデータセットのバッチ化を行います。
その後、for ループでエポック、バッチを回していきます。

for ループ内ではまず tf.GradientTape スコープの中に順伝搬の処理を記述します。
その後 gradient メソッドを呼び出して勾配を計算し、apply_gradients メソッドで最適化関数に従って重みを更新しています。

import tensorflow as tf

batch_size = 32
epochs = 10

(train_images, train_labels), _ = tf.keras.datasets.cifar10.load_data()
# 例示のため学習済みのモデルを使っています

model = tf.keras.applications.VGG16()

buffer_size = len(train_images)
train_ds = tf.data.Dataset.from_tensor_slices((train_images, train_labels))
train_ds = train_ds.shuffle(buffer_size=buffer_size).batch(batch_size)

criterion = tf.keras.losses.SparseCategoricalCrossentropy()
optimizer = tf.keras.optimizers.Adam()

for epoch in range(epochs):
    for x, y_true in train_ds:
        with tf.GradientTape() as tape:
            y_pred = model(x, training=True)
            loss = criterion(y_true=y_true, y_pred=y_pred)
        gradients = tape.gradient(loss, model.trainable_variables)
        optimizer.apply_gradients(zip(gradients, model.trainable_variables))

これで訓練が実行できます。

上記の例ではバリデーションデータでの評価や TensorBoard への出力等を一切行っていませんが、普通に for ループを回しているだけなので、好きなように処理を追加していくことができます。

一方で記述量はどうしても多くなるためソースコードの品質担保がやや大変になります。
なお Chainer、PyTorch も（細かい違いはありますが）ほぼ同じ書き方ができます。

訓練方法振り返り

訓練方法の説明は以上になります。
まとめると、以下のような使い分けができるかと思います。

• 訓練中に特殊な処理を実行する必要がない場合、一般的な訓練手法の場合は built-in 訓練
• built-in の枠にハマらない場合、訓練中にいろいろな処理を追加して試行錯誤したい場合、手なりで書きたい場合は カスタム訓練

VGG16／ResNet 50 の概要

説明だけではよく分からない部分もあると思うので、実装を通して理解を深めていきます。

まず簡単に 2 つのモデルについて紹介します。

VGG16 とは

3x3 Convolution を 13 層、全結合層を 3 層持つ非常にシンプルな構造でありながら高性能なモデルです。
様々な画像認識タスクで画像特徴の抽出に利用されます。原論文は 37,000 を超える被引用数を持っており、とても有名です。

Sequential API、 Functional API、 Subclassing API で実装が可能です。

なお、原論文はこちらです。
https://arxiv.org/abs/1409.1556

ResNet50 とは

Residual 機構を持つ多層（Convolution を 49 層、全結合層を 1 層）モデルです。2020 年現在でもこの ResNet の亜種が画像分類の精度においてはトップクラスとなっており、こちらも高性能なモデルです。
VGG16 と同様に様々な画像認識タスクで画像特徴の抽出に利用されます。原論文は 45,000 を超える被引用数（BERT の約 10 倍）を持っており、こちらも非常に有名です。

Sequential API 単体では実装できません。Functional API、 Subclassing API で実装が可能です。

なお、原論文はこちらです。
https://arxiv.org/abs/1512.03385

VGG16 の実装

ではそれぞれの書き方で実装してみます。

VGG16 Sequential API

特に考えることもないので普通に書きます。

from tensorflow.keras import layers, Sequential


def sequential_vgg16(input_shape, output_size):
    params = {
        "padding": "same",
        "use_bias": True,
        "kernel_initializer": "he_normal",
    }
    model = Sequential()
    model.add(layers.Conv2D(64, 3, 1, **params, batch_input_shape=input_shape))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.Conv2D(64, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.MaxPool2D(2, padding="same"))
    model.add(layers.Conv2D(128, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.Conv2D(128, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.MaxPool2D(2, padding="same"))
    model.add(layers.Conv2D(256, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.Conv2D(256, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.Conv2D(256, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.MaxPool2D(2, padding="same"))
    model.add(layers.Conv2D(512, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.Conv2D(512, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.Conv2D(512, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.MaxPool2D(2, padding="same"))
    model.add(layers.Conv2D(512, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.Conv2D(512, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.Conv2D(512, 3, 1, **params))
    model.add(layers.BatchNormalization())
    model.add(layers.ReLU())
    model.add(layers.MaxPool2D(2, padding="same"))
    model.add(layers.Flatten())
    model.add(layers.Dense(4096))
    model.add(layers.Dense(4096))
    model.add(layers.Dense(output_size, activation="softmax"))
    return model

かなり単純に書くことができますが、層が多いので見るのが辛いことがわかります。
例えば、どこかで ReLU が抜けてても気付かなそうです。
また、例えば Batch Normalization を無くしたい、と思った場合は 1 行ずつコメントアウトしていく必要があり、再利用生やカスタマイズ性に乏しいです。

VGG16 Functional API

Sequential API と比べて柔軟に書くことができます。
今回は再利用される層のまとまり(Convolution - BatchNormalization - ReLU)を関数にしてみます。

from tensorflow.keras import layers, Model


def functional_cbr(x, filters, kernel_size, strides):
    params = {
        "filters": filters,
        "kernel_size": kernel_size,
        "strides": strides,
        "padding": "same",
        "use_bias": True,
        "kernel_initializer": "he_normal",
    }

    x = layers.Conv2D(**params)(x)
    x = layers.BatchNormalization()(x)
    x = layers.ReLU()(x)
    return x


def functional_vgg16(input_shape, output_size):
    inputs = layers.Input(batch_input_shape=input_shape)
    x = functional_cbr(inputs, 64, 3, 1)
    x = functional_cbr(x, 64, 3, 1)
    x = layers.MaxPool2D(2, padding="same")(x)
    x = functional_cbr(x, 128, 3, 1)
    x = functional_cbr(x, 128, 3, 1)
    x = layers.MaxPool2D(2, padding="same").__call__(x)  # こう書いても良い
    x = functional_cbr(x, 256, 3, 1)
    x = functional_cbr(x, 256, 3, 1)
    x = functional_cbr(x, 256, 3, 1)
    x = layers.MaxPool2D(2, padding="same").call(x)  # こう書いても良い
    x = functional_cbr(x, 512, 3, 1)
    x = functional_cbr(x, 512, 3, 1)
    x = functional_cbr(x, 512, 3, 1)
    x = layers.MaxPool2D(2, padding="same")(x)
    x = functional_cbr(x, 512, 3, 1)
    x = functional_cbr(x, 512, 3, 1)
    x = functional_cbr(x, 512, 3, 1)
    x = layers.MaxPool2D(2, padding="same")(x)
    x = layers.Flatten()(x)
    x = layers.Dense(4096)(x)
    x = layers.Dense(4096)(x)
    outputs = layers.Dense(output_size, activation="softmax")(x)
    return Model(inputs=inputs, outputs=outputs)

だいぶスッキリ書くことができました。
BatchNormalization を無くしたくなっても、 ReLU を LeaklyReLU に変えたくなっても数行の改修で済みます。

VGG16 Subclassing API

Functional API と同様に再利用される層のまとまり（Convolution - BatchNormalization - ReLU）をクラスにして書いてみます。

from tensorflow.keras import layers, Model


class CBR(Model):
    def __init__(self, filters, kernel_size, strides):
        super().__init__()

        params = {
            "filters": filters,
            "kernel_size": kernel_size,
            "strides": strides,
            "padding": "same",
            "use_bias": True,
            "kernel_initializer": "he_normal",
        }

        self.layers_ = [
            layers.Conv2D(**params),
            layers.BatchNormalization(),
            layers.ReLU()
        ]

    def call(self, inputs):
        for layer in self.layers_:
            inputs = layer(inputs)
        return inputs


class VGG16(Model):
    def __init__(self, output_size=1000):
        super().__init__()
        self.layers_ = [
            CBR(64, 3, 1),
            CBR(64, 3, 1),
            layers.MaxPool2D(2, padding="same"),
            CBR(128, 3, 1),
            CBR(128, 3, 1),
            layers.MaxPool2D(2, padding="same"),
            CBR(256, 3, 1),
            CBR(256, 3, 1),
            CBR(256, 3, 1),
            layers.MaxPool2D(2, padding="same"),
            CBR(512, 3, 1),
            CBR(512, 3, 1),
            CBR(512, 3, 1),
            layers.MaxPool2D(2, padding="same"),
            CBR(512, 3, 1),
            CBR(512, 3, 1),
            CBR(512, 3, 1),
            layers.MaxPool2D(2, padding="same"),
            layers.Flatten(),
            layers.Dense(4096),
            layers.Dense(4096),
            layers.Dense(output_size, activation="softmax"),
        ]

    def call(self, inputs):
        for layer in self.layers_:
            inputs = layer(inputs)
        return inputs

__init__ でモデルの定義、call でモデルの呼び出しを担当しているため、Functional API よりも直感的に理解しやすいですが、コードは長めになります。
また命令型である Subclassing API はモデル生成時に入力の形状が必要ない（引数に input_shape が必要ない）こともポイントです。

VGG16 実装 振り返り

なるべく比較しやすいように書いたつもりですがいかがだったでしょうか。

今回の実装は Batch Normalization を Convolution 層の間に挟んでいるのと重みの初期化に He の初期化を利用していますが、原論文が提出された時はこれらのテクニックは未だ発表されていなかったので Batch Normalization 層はなく、重みの初期化には Grolot の初期化が使われていました。そのため、原論文では勾配消失を回避するために 7 層のモデルを学習してから層を徐々に付け足していくといった転移学習ライクな学習方法が採用されています。

上記の実装をより深く理解するために Batch Normalization 層をなくすと何が起きるのか、重みの初期化手法を変えると何が起きるのか、などを試してみるのも面白いと思います。

ResNet50 の実装

続いて ResNet50 を実装します。
Sequential API 単体では書けないので Functional API、 Subclassing API で書きます。

ResNet50 Functional API

再利用される Residual 機構を関数化して実装します。

from tensorflow.keras import layers, Model


def functional_bottleneck_residual(x, in_ch, out_ch, strides=1):
    params = {
        "padding": "same",
        "kernel_initializer": "he_normal",
        "use_bias": True,
    }
    inter_ch = out_ch // 4
    h1 = layers.Conv2D(inter_ch, kernel_size=1, strides=strides, **params)(x)
    h1 = layers.BatchNormalization()(h1)
    h1 = layers.ReLU()(h1)
    h1 = layers.Conv2D(inter_ch, kernel_size=3, strides=1, **params)(h1)
    h1 = layers.BatchNormalization()(h1)
    h1 = layers.ReLU()(h1)
    h1 = layers.Conv2D(out_ch, kernel_size=1, strides=1, **params)(h1)
    h1 = layers.BatchNormalization()(h1)

    if in_ch != out_ch:
        h2 = layers.Conv2D(out_ch, kernel_size=1, strides=strides, **params)(x)
        h2 = layers.BatchNormalization()(h2)
    else:
        h2 = x

    h = layers.Add()([h1, h2])
    h = layers.ReLU()(h)
    return h


def functional_resnet50(input_shape, output_size):
    inputs = layers.Input(batch_input_shape=input_shape)
    x = layers.Conv2D(64, 7, 2, padding="same", kernel_initializer="he_normal")(inputs)
    x = layers.BatchNormalization()(x)
    x = layers.MaxPool2D(pool_size=3, strides=2, padding="same")(x)

    x = functional_bottleneck_residual(x, 64, 256)
    x = functional_bottleneck_residual(x, 256, 256)
    x = functional_bottleneck_residual(x, 256, 256)

    x = functional_bottleneck_residual(x, 256, 512, 2)
    x = functional_bottleneck_residual(x, 512, 512)
    x = functional_bottleneck_residual(x, 512, 512)
    x = functional_bottleneck_residual(x, 512, 512)

    x = functional_bottleneck_residual(x, 512, 1024, 2)
    x = functional_bottleneck_residual(x, 1024, 1024)
    x = functional_bottleneck_residual(x, 1024, 1024)
    x = functional_bottleneck_residual(x, 1024, 1024)
    x = functional_bottleneck_residual(x, 1024, 1024)
    x = functional_bottleneck_residual(x, 1024, 1024)

    x = functional_bottleneck_residual(x, 1024, 2048, 2)
    x = functional_bottleneck_residual(x, 2048, 2048)
    x = functional_bottleneck_residual(x, 2048, 2048)

    x = layers.GlobalAveragePooling2D()(x)
    outputs = layers.Dense(
        output_size, activation="softmax", kernel_initializer="he_normal"
    )(x)
    return Model(inputs=inputs, outputs=outputs)

functional_bottleneck_residual メソッド内では h1、h2、h が登場します。
このように、データのフローが途中で分岐するモデルは Sequential API では記述できません。

また、h2 は入出力のチャネル数が同じ場合は何もせず、異なる場合はチャネル数を調整する処理（Projection）を行います。このような条件分岐も Sequential API では記述できません。

このメソッドを作ってしまえば、あとは順々に記述していくだけです。

ResNet50 Subclassing API

Functional API と同様に再利用される Residual 機構をクラス化して実装します。

from tensorflow import layers, Model


class BottleneckResidual(Model):
    """ResNet の Bottleneck Residual Module です.
    1 層目の 1x1 conv で ch 次元を削減することで
    2 層目の 3x3 conv の計算量を減らし
    3 層目の 1x1 conv で ch 出力の次元を復元します.
    計算量の多い 2 層目 3x3 conv の次元を小さくすることから bottleneck と呼ばれます.
    """

    def __init__(self, in_ch, out_ch, strides=1):
        super().__init__()

        self.projection = in_ch != out_ch
        inter_ch = out_ch // 4
        params = {
            "padding": "same",
            "kernel_initializer": "he_normal",
            "use_bias": True,
        }

        self.common_layers = [
            layers.Conv2D(inter_ch, kernel_size=1, strides=strides, **params),
            layers.BatchNormalization(),
            layers.ReLU(),
            layers.Conv2D(inter_ch, kernel_size=3, strides=1, **params),
            layers.BatchNormalization(),
            layers.ReLU(),
            layers.Conv2D(out_ch, kernel_size=1, strides=1, **params),
            layers.BatchNormalization(),
        ]

        if self.projection:
            self.projection_layers = [
                layers.Conv2D(out_ch, kernel_size=1, strides=strides, **params),
                layers.BatchNormalization(),
            ]

        self.concat_layers = [layers.Add(), layers.ReLU()]

    def call(self, inputs):
        h1 = inputs
        h2 = inputs

        for layer in self.common_layers:
            h1 = layer(h1)

        if self.projection:
            for layer in self.projection_layers:
                h2 = layer(h2)

        outputs = [h1, h2]
        for layer in self.concat_layers:
            outputs = layer(outputs)
        return outputs


class ResNet50(Model):
    """ResNet50 です.
    要素は
    conv * 1
    resblock(conv * 3) * 3
    resblock(conv * 3) * 4
    resblock(conv * 3) * 6
    resblock(conv * 3) * 3
    dense * 1
    から構成されていて, conv * 49 + dense * 1 の 50 層です.
    """

    def __init__(self, output_size=1000):
        super().__init__()

        self.layers_ = [
            layers.Conv2D(64, 7, 2, padding="same", kernel_initializer="he_normal"),
            layers.BatchNormalization(),
            layers.MaxPool2D(pool_size=3, strides=2, padding="same"),
            BottleneckResidual(64, 256),
            BottleneckResidual(256, 256),
            BottleneckResidual(256, 256),
            BottleneckResidual(256, 512, 2),
            BottleneckResidual(512, 512),
            BottleneckResidual(512, 512),
            BottleneckResidual(512, 512),
            BottleneckResidual(512, 1024, 2),
            BottleneckResidual(1024, 1024),
            BottleneckResidual(1024, 1024),
            BottleneckResidual(1024, 1024),
            BottleneckResidual(1024, 1024),
            BottleneckResidual(1024, 1024),
            BottleneckResidual(1024, 2048, 2),
            BottleneckResidual(2048, 2048),
            BottleneckResidual(2048, 2048),
            layers.GlobalAveragePooling2D(),
            layers.Dense(
                output_size, activation="softmax", kernel_initializer="he_normal"
            ),
        ]

    def call(self, inputs):
        for layer in self.layers_:
            inputs = layer(inputs)
        return inputs

Functional API とそこまで大きな違いはありません。
__init__ レイヤーでリストに層をまとめるような書き方をしていますが、クラスの変数に登録されていれば良いのでこの辺は自由に書くことができます。

ResNet50 実装 振り返り

Sequential API 単体では実装できないモデルとして ResNet50 を紹介しました。
正直大きな違いはないので Functional API、 Subclassing API は好みで使い分けても問題ないかと思います。

訓練の実装

最後に訓練ループの実装を比較してみます。

全てのソースコードを載せるとかなり長くなるため部分的にメソッドを src.utils に切り出しています。
そこまで複雑なことをしているわけではないので補完しながら読んでいただければ助かります。

一応全てのソースは以下のリポジトリにあるので気になる方はご覧ください。
https://github.com/Anieca/deep-learning-models

built-in 訓練 実装

テストデータの精度算出、TensorBoard 用のログ出力などいくつかオプションを指定してみます。

import os
import tensorflow as tf

from src.utils import load_dataset, load_model, get_args, get_current_time


def builtin_train(args):
    # 1. load dataset and model
    (train_images, train_labels), (test_images, test_labels) = load_dataset(args.data)
    input_shape = train_images[: args.batch_size, :, :, :].shape
    output_size = max(train_labels) + 1
    model = load_model(args.arch, input_shape=input_shape, output_size=output_size)
    model.summary()

    # 2. set tensorboard cofigs
    logdir = os.path.join(args.logdir, get_current_time())
    tensorboard_callback = tf.keras.callbacks.TensorBoard(log_dir=logdir)

    # 3. loss, optimizer, metrics setting
    model.compile(
        optimizer=tf.keras.optimizers.Adam(),
        loss="sparse_categorical_crossentropy",
        metrics=["accuracy"],
    )

    # 4. dataset config (and validation, callback config)
    fit_params = {}
    fit_params["batch_size"] = args.batch_size
    fit_params["epochs"] = args.max_epoch
    if args.steps_per_epoch:
        fit_params["steps_per_epoch"] = args.steps_per_epoch
    fit_params["verbose"] = 1
    fit_params["callbacks"] = [tensorboard_callback]
    fit_params["validation_data"] = (test_images, test_labels)

    # 5. start train and test
    model.fit(train_images, train_labels, **fit_params)

かなりシンプルに書けます。

コールバック関数は他にも色々あるので興味がある方はドキュメントを読んでみてください。
https://www.tensorflow.org/api_docs/python/tf/keras/callbacks

カスタム訓練 実装

上記の built-in 訓練と同じ処理を自分で実装してみます。

import os
import tensorflow as tf

from src.utils import load_dataset, load_model, get_args, get_current_time


def custom_train(args):
    # 1. load dataset and model
    (train_images, train_labels), (test_images, test_labels) = load_dataset(args.data)
    input_shape = train_images[: args.batch_size, :, :, :].shape
    output_size = max(train_labels) + 1
    model = load_model(args.arch, input_shape=input_shape, output_size=output_size)
    model.summary()

    # 2. set tensorboard configs
    logdir = os.path.join(args.logdir, get_current_time())
    train_writer = tf.summary.create_file_writer(os.path.join(logdir, "train"))
    test_writer = tf.summary.create_file_writer(os.path.join(logdir, "test"))

    # 3. loss, optimizer, metrics setting
    criterion = tf.keras.losses.SparseCategoricalCrossentropy()
    optimizer = tf.keras.optimizers.Adam()
    train_loss_avg = tf.keras.metrics.Mean()
    train_accuracy = tf.keras.metrics.SparseCategoricalAccuracy()
    test_loss_avg = tf.keras.metrics.Mean()
    test_accuracy = tf.keras.metrics.SparseCategoricalAccuracy()

    # 4. dataset config
    buffer_size = len(train_images)
    train_ds = tf.data.Dataset.from_tensor_slices((train_images, train_labels))
    train_ds = train_ds.shuffle(buffer_size=buffer_size).batch(args.batch_size)
    test_ds = tf.data.Dataset.from_tensor_slices((test_images, test_labels))
    test_ds = test_ds.batch(args.batch_size)

    # 5. start train and test
    for epoch in range(args.max_epoch):
        # 5.1. initialize metrics
        train_loss_avg.reset_states()
        train_accuracy.reset_states()
        test_loss_avg.reset_states()
        test_loss_avg.reset_states()

        # 5.2. initialize progress bar
        train_pbar = tf.keras.utils.Progbar(args.steps_per_epoch)
        test_pbar = tf.keras.utils.Progbar(args.steps_per_epoch)

        # 5.3. start train
        for i, (x, y_true) in enumerate(train_ds):
            if args.steps_per_epoch and i >= args.steps_per_epoch:
                break
            # 5.3.1. forward
            with tf.GradientTape() as tape:
                y_pred = model(x, training=True)
                loss = criterion(y_true=y_true, y_pred=y_pred)
            # 5.3.2. calculate gradients from `tape` and backward
            gradients = tape.gradient(loss, model.trainable_variables)
            optimizer.apply_gradients(zip(gradients, model.trainable_variables))

            # 5.3.3. update metrics and progress bar
            train_loss_avg(loss)
            train_accuracy(y_true, y_pred)
            train_pbar.update(
                i + 1,
                [
                    ("avg_loss", train_loss_avg.result()),
                    ("accuracy", train_accuracy.result()),
                ],
            )

        # 5.4. start test
        for i, (x, y_true) in enumerate(test_ds):
            if args.steps_per_epoch and i >= args.steps_per_epoch:
                break
            # 5.4.1. forward
            y_pred = model(x)
            loss = criterion(y_true, y_pred)

            # 5.4.2. update metrics and progress bar
            test_loss_avg(loss)
            test_accuracy(y_true, y_pred)
            test_pbar.update(
                i + 1,
                [
                    ("avg_test_loss", test_loss_avg.result()),
                    ("test_accuracy", test_accuracy.result()),
                ],
            )

        # 5.5. write metrics to tensorboard
        with train_writer.as_default():
            tf.summary.scalar("Loss", train_loss_avg.result(), step=epoch)
            tf.summary.scalar("Acc", train_accuracy.result(), step=epoch)
        with test_writer.as_default():
            tf.summary.scalar("Loss", test_loss_avg.result(), step=epoch)
            tf.summary.scalar("Acc", test_accuracy.result(), step=epoch)

訓練開始まではそこまで変わらないですが、訓練ループ内（コメントの 5. ）の記述量がかなり多くなります。

訓練実装 振り返り

自分で TensorBoard 出力の管理やプログレスバーの作成と言ったユーティリティを管理するのはそれなりにコストがかかりますが、 built-in はかなり楽に使うことができます。

built-in に用意されていない処理を記述したい場合はカスタム訓練で書く必要がありますが、そうでなければ built-in を使った方が良さそうです。

終わりに

以上です。お疲れ様でした。

TensorFlow 2 系の色々な書き方について実装を交えて紹介しました。

あまりそれぞれの書き方について優劣はつけずにフラットに書いたつもりです。

自分で書くときは状況や好みに合わせて書けば良いと思いますが、ソースコードを探していると色々な書き方に出会うので、なんとなく全ての書き方を理解していると良いかと思います。

皆様のお役に立てれば幸いです。