Keras 2 : examples : Siamese ネットワークを triplet 損失で使用した画像類似性推定 (翻訳/解説)

翻訳 : (株)クラスキャット セールスインフォメーション
作成日時 : 12/15/2021 (keras 2.7.0)

* 本ページは、Keras の以下のドキュメントを翻訳した上で適宜、補足説明したものです：

• Code examples : Computer Vision : Image similarity estimation using a Siamese Network with a triplet loss (Author: Hazem Essam and Santiago L. Valdarrama)

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

 

 

Keras 2 : examples : Siamese ネットワークを triplet 損失で使用した画像類似性推定

Description: triplet 損失関数を使用して画像の類似性を比較するために Siamese ネットワークを訓練する。

 

イントロダクション

Siamese ネットワーク は、各入力に対して特徴ベクトルを生成して比較するために使用される 2 つまたはそれ以上の同一のサブネットワークを持つタイプのネットワーク・アーキテクチャです。

Siamese ネットワークは重複の検出、異常の発見そして顔認識のような様々なユースケースに対して適用できます。

このサンプルは 3 つの同一のサブネットワークを持つ Siamese ネットワークを使用します。モデルに 3 つの画像を提供します、そこではそれらの 2 つは類似していて (アンカー と ポジティブサンプル)、3 番目は無関係です (ネガティブ・サンプル)。モデルのための目標は画像間の類似度を推定することを学習することです。

ネットワークが学習するために、triplet 損失関数を使用します。triplet 損失へのイントロダクションは FaceNet 論文 by Schroff et al,. 2015 で見つけられます。このサンプルでは、次のように triplet 損失関数を定義します :

L(A, P, N) = max(‖f(A) – f(P)‖² – ‖f(A) – f(N)‖² + margin, 0)

このサンプルは Totally Looks Like データセット by Rosenfeld et al., 2018 を使用しています。

 

セットアップ

import matplotlib.pyplot as plt
import numpy as np
import os
import random
import tensorflow as tf
from pathlib import Path
from tensorflow.keras import applications
from tensorflow.keras import layers
from tensorflow.keras import losses
from tensorflow.keras import optimizers
from tensorflow.keras import metrics
from tensorflow.keras import Model
from tensorflow.keras.applications import resnet


target_shape = (200, 200)

 

データセットのロード

Totally Looks Like データセットをロードしてローカル環境の ~/.keras ディレクトリ内で unzip します。

データセットは 2 つの分離したファイルから成ります :

• left.zip はアンカーとして使用する画像を含みます。
• right.zip はポジティブサンプル (アンカーのように見える画像) として使用する画像を含みます。

cache_dir = Path(Path.home()) / ".keras"
anchor_images_path = cache_dir / "left"
positive_images_path = cache_dir / "right"

!gdown --id 1jvkbTr_giSP3Ru8OwGNCg6B4PvVbcO34
!gdown --id 1EzBZUb_mh_Dp_FKD0P4XiYYSd0QBH5zW
!unzip -oq left.zip -d $cache_dir
!unzip -oq right.zip -d $cache_dir

(訳注: 以下は原文ママ)

zsh:1: command not found: gdown
zsh:1: command not found: gdown
unzip:  cannot find or open left.zip, left.zip.zip or left.zip.ZIP.
unzip:  cannot find or open right.zip, right.zip.zip or right.zip.ZIP.

 

データの準備

データをロードするために tf.data パイプラインを使用して Siamese ネットワークを訓練するために必要な triplet を生成していきます。

アンカー、ポジティブとネガティブファイル名を持つ zip 形式のリストをソースとして使用してパイプラインをセットアップします。パイプラインは対応する画像をロードして前処理します。

def preprocess_image(filename):
    """
    Load the specified file as a JPEG image, preprocess it and
    resize it to the target shape.
    """

    image_string = tf.io.read_file(filename)
    image = tf.image.decode_jpeg(image_string, channels=3)
    image = tf.image.convert_image_dtype(image, tf.float32)
    image = tf.image.resize(image, target_shape)
    return image


def preprocess_triplets(anchor, positive, negative):
    """
    Given the filenames corresponding to the three images, load and
    preprocess them.
    """

    return (
        preprocess_image(anchor),
        preprocess_image(positive),
        preprocess_image(negative),
    )

アンカー、ポジティブとネガティブファイル名を持つ zip 形式のリストをソースとして使用するデータパイプラインをセットアップしましょう。パイプラインの出力はロードされて前処理された総ての画像を持ち同じトリプレットを含みます。

# We need to make sure both the anchor and positive images are loaded in
# sorted order so we can match them together.
anchor_images = sorted(
    [str(anchor_images_path / f) for f in os.listdir(anchor_images_path)]
)

positive_images = sorted(
    [str(positive_images_path / f) for f in os.listdir(positive_images_path)]
)

image_count = len(anchor_images)

anchor_dataset = tf.data.Dataset.from_tensor_slices(anchor_images)
positive_dataset = tf.data.Dataset.from_tensor_slices(positive_images)

# To generate the list of negative images, let's randomize the list of
# available images and concatenate them together.
rng = np.random.RandomState(seed=42)
rng.shuffle(anchor_images)
rng.shuffle(positive_images)

negative_images = anchor_images + positive_images
np.random.RandomState(seed=32).shuffle(negative_images)

negative_dataset = tf.data.Dataset.from_tensor_slices(negative_images)
negative_dataset = negative_dataset.shuffle(buffer_size=4096)

dataset = tf.data.Dataset.zip((anchor_dataset, positive_dataset, negative_dataset))
dataset = dataset.shuffle(buffer_size=1024)
dataset = dataset.map(preprocess_triplets)

# Let's now split our dataset in train and validation.
train_dataset = dataset.take(round(image_count * 0.8))
val_dataset = dataset.skip(round(image_count * 0.8))

train_dataset = train_dataset.batch(32, drop_remainder=False)
train_dataset = train_dataset.prefetch(8)

val_dataset = val_dataset.batch(32, drop_remainder=False)
val_dataset = val_dataset.prefetch(8)

トリプレットの幾つかのサンプルを見てみましょう。最初の 2 つの画像がどのように良く似ているか、一方で 3 番目のものは常に異なることに注意してください。

def visualize(anchor, positive, negative):
    """Visualize a few triplets from the supplied batches."""

    def show(ax, image):
        ax.imshow(image)
        ax.get_xaxis().set_visible(False)
        ax.get_yaxis().set_visible(False)

    fig = plt.figure(figsize=(9, 9))

    axs = fig.subplots(3, 3)
    for i in range(3):
        show(axs[i, 0], anchor[i])
        show(axs[i, 1], positive[i])
        show(axs[i, 2], negative[i])


visualize(*list(train_dataset.take(1).as_numpy_iterator())[0])

 

埋め込み生成 (= generator) モデルのセットアップ

Siamese ネットワークはトリプレットの画像の各々に対して埋め込みを生成します。そのために、ImageNet 上で事前訓練された ResNet50 モデルを使用し、そして幾つかの Dense 層をそれに接続しますので、これらの埋め込みを分離することを学習できます。

層 conv5_block1_out までモデルの総ての層の重みを凍結します。モデルが既に学習した重みに影響を与えることを避けるためにこれは重要です。ボトムの幾つかの層は訓練可能なままにしておきますので、訓練の間にそれらの重みを再調整できます。

base_cnn = resnet.ResNet50(
    weights="imagenet", input_shape=target_shape + (3,), include_top=False
)

flatten = layers.Flatten()(base_cnn.output)
dense1 = layers.Dense(512, activation="relu")(flatten)
dense1 = layers.BatchNormalization()(dense1)
dense2 = layers.Dense(256, activation="relu")(dense1)
dense2 = layers.BatchNormalization()(dense2)
output = layers.Dense(256)(dense2)

embedding = Model(base_cnn.input, output, name="Embedding")

trainable = False
for layer in base_cnn.layers:
    if layer.name == "conv5_block1_out":
        trainable = True
    layer.trainable = trainable

 

Siamese ネットワーク・モデルのセットアップ

Siamese ネットワーク は入力としてトリプレット画像の各々を受け取り、埋め込みを生成し、そしてアンカーとポジティブ埋め込み間の距離、及びアンカーとネガティブ埋め込みの間の距離を出力します。

距離を計算するため、タプルとして両方の値を返すカスタム層 DistanceLayer を使用できます。

class DistanceLayer(layers.Layer):
    """
    This layer is responsible for computing the distance between the anchor
    embedding and the positive embedding, and the anchor embedding and the
    negative embedding.
    """

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    def call(self, anchor, positive, negative):
        ap_distance = tf.reduce_sum(tf.square(anchor - positive), -1)
        an_distance = tf.reduce_sum(tf.square(anchor - negative), -1)
        return (ap_distance, an_distance)


anchor_input = layers.Input(name="anchor", shape=target_shape + (3,))
positive_input = layers.Input(name="positive", shape=target_shape + (3,))
negative_input = layers.Input(name="negative", shape=target_shape + (3,))

distances = DistanceLayer()(
    embedding(resnet.preprocess_input(anchor_input)),
    embedding(resnet.preprocess_input(positive_input)),
    embedding(resnet.preprocess_input(negative_input)),
)

siamese_network = Model(
    inputs=[anchor_input, positive_input, negative_input], outputs=distances
)

 

総てをまとめる

Siamese ネットワークにより生成された 3 つの埋め込みを使用してトリプレット損失を計算できるように、モデルをカスタム訓練ループで実装する必要があります。

訓練プロセスの損失を追跡するために Mean メトリックインスタンスを作成しましょう。

class SiameseModel(Model):
    """The Siamese Network model with a custom training and testing loops.

    Computes the triplet loss using the three embeddings produced by the
    Siamese Network.

    The triplet loss is defined as:
       L(A, P, N) = max(‖f(A) - f(P)‖² - ‖f(A) - f(N)‖² + margin, 0)
    """

    def __init__(self, siamese_network, margin=0.5):
        super(SiameseModel, self).__init__()
        self.siamese_network = siamese_network
        self.margin = margin
        self.loss_tracker = metrics.Mean(name="loss")

    def call(self, inputs):
        return self.siamese_network(inputs)

    def train_step(self, data):
        # GradientTape is a context manager that records every operation that
        # you do inside. We are using it here to compute the loss so we can get
        # the gradients and apply them using the optimizer specified in
        # `compile()`.
        with tf.GradientTape() as tape:
            loss = self._compute_loss(data)

        # Storing the gradients of the loss function with respect to the
        # weights/parameters.
        gradients = tape.gradient(loss, self.siamese_network.trainable_weights)

        # Applying the gradients on the model using the specified optimizer
        self.optimizer.apply_gradients(
            zip(gradients, self.siamese_network.trainable_weights)
        )

        # Let's update and return the training loss metric.
        self.loss_tracker.update_state(loss)
        return {"loss": self.loss_tracker.result()}

    def test_step(self, data):
        loss = self._compute_loss(data)

        # Let's update and return the loss metric.
        self.loss_tracker.update_state(loss)
        return {"loss": self.loss_tracker.result()}

    def _compute_loss(self, data):
        # The output of the network is a tuple containing the distances
        # between the anchor and the positive example, and the anchor and
        # the negative example.
        ap_distance, an_distance = self.siamese_network(data)

        # Computing the Triplet Loss by subtracting both distances and
        # making sure we don't get a negative value.
        loss = ap_distance - an_distance
        loss = tf.maximum(loss + self.margin, 0.0)
        return loss

    @property
    def metrics(self):
        # We need to list our metrics here so the `reset_states()` can be
        # called automatically.
        return [self.loss_tracker]

 

訓練

これでモデルを訓練をする準備ができました。

siamese_model = SiameseModel(siamese_network)
siamese_model.compile(optimizer=optimizers.Adam(0.0001))
siamese_model.fit(train_dataset, epochs=10, validation_data=val_dataset)

Epoch 1/10
151/151 [==============================] - 277s 2s/step - loss: 0.5014 - val_loss: 0.3719
Epoch 2/10
151/151 [==============================] - 276s 2s/step - loss: 0.3884 - val_loss: 0.3632
Epoch 3/10
151/151 [==============================] - 287s 2s/step - loss: 0.3711 - val_loss: 0.3509
Epoch 4/10
151/151 [==============================] - 295s 2s/step - loss: 0.3585 - val_loss: 0.3287
Epoch 5/10
151/151 [==============================] - 299s 2s/step - loss: 0.3420 - val_loss: 0.3301
Epoch 6/10
151/151 [==============================] - 297s 2s/step - loss: 0.3181 - val_loss: 0.3419
Epoch 7/10
151/151 [==============================] - 290s 2s/step - loss: 0.3131 - val_loss: 0.3201
Epoch 8/10
151/151 [==============================] - 295s 2s/step - loss: 0.3102 - val_loss: 0.3152
Epoch 9/10
151/151 [==============================] - 286s 2s/step - loss: 0.2905 - val_loss: 0.2937
Epoch 10/10
151/151 [==============================] - 270s 2s/step - loss: 0.2921 - val_loss: 0.2952

<tensorflow.python.keras.callbacks.History at 0x7fc69064bd10>

(訳者注: 実験結果)

Epoch 1/10
151/151 [==============================] - 63s 252ms/step - loss: 0.5017 - val_loss: 0.3874
Epoch 2/10
151/151 [==============================] - 35s 234ms/step - loss: 0.3913 - val_loss: 0.3463
Epoch 3/10
151/151 [==============================] - 37s 245ms/step - loss: 0.3666 - val_loss: 0.3435
Epoch 4/10
151/151 [==============================] - 35s 235ms/step - loss: 0.3540 - val_loss: 0.3296
Epoch 5/10
151/151 [==============================] - 35s 234ms/step - loss: 0.3336 - val_loss: 0.3178
Epoch 6/10
151/151 [==============================] - 35s 234ms/step - loss: 0.3251 - val_loss: 0.3203
Epoch 7/10
151/151 [==============================] - 35s 234ms/step - loss: 0.3260 - val_loss: 0.3224
Epoch 8/10
151/151 [==============================] - 35s 235ms/step - loss: 0.3035 - val_loss: 0.2917
Epoch 9/10
151/151 [==============================] - 35s 234ms/step - loss: 0.2916 - val_loss: 0.3001
Epoch 10/10
151/151 [==============================] - 35s 235ms/step - loss: 0.2941 - val_loss: 0.2972
CPU times: user 10min 32s, sys: 47.8 s, total: 11min 20s
Wall time: 6min 29s

 

ネットワークが何を学習したかを調べる

この時点で、ネットワークが埋め込みを (それらが類似画像に属するか否かに依存して) 分離することをどのように学習したか確認できます。

埋め込み間の類似度を測定するために コサイン類似度 を使用できます。

各画像のために生成された埋め込み間の類似度をチェックするためにデータセットからサンプルをピックアップしましょう。

sample = next(iter(train_dataset))
visualize(*sample)

anchor, positive, negative = sample
anchor_embedding, positive_embedding, negative_embedding = (
    embedding(resnet.preprocess_input(anchor)),
    embedding(resnet.preprocess_input(positive)),
    embedding(resnet.preprocess_input(negative)),
)

最後に、アンカーとポジティブ画像間のコサイン類似度を計算してそれをアンカーとネガティブ画像の間の類似度と比較することができます。

アンカーとポジティブ画像間の類似度がアンカーとネガティブ画像間の類似度よりも大きいことを想定できるはずです。

cosine_similarity = metrics.CosineSimilarity()

positive_similarity = cosine_similarity(anchor_embedding, positive_embedding)
print("Positive similarity:", positive_similarity.numpy())

negative_similarity = cosine_similarity(anchor_embedding, negative_embedding)
print("Negative similarity", negative_similarity.numpy())

Positive similarity: 0.9940324
Negative similarity 0.9918252

Positive similarity: 0.9954051
Negative similarity 0.99360335

 

要約

1. tf.data API はモデルのために効率的な入力パイプラインを構築することを可能にします。それは大規模なデータセットを持つ場合に特に有用です。tf.data パイプラインについては tf.data: Build TensorFlow input pipelines で更に学習できます。
   

2. このサンプルでは、特徴埋め込みを生成するサブネットワークの一部として事前訓練済みの ResNet50 を使用しています。転移学習 を使用することで、訓練時間とデータセットのサイズを著しく削減できます。
   

3. ResNet50 ネットワークの最後の層の重みを 再調整 していますが残りの層はそのままにしていることに注目してください。各層に割当てられた名前を使用して、あるポイントに重みを凍結して最後の幾つかの層だけをオープンにすることができます。
   

4. DistanceLayer で行なったように、tf.keras.layers.Layer から継承したクラスを作成してカスタム層を作成できます。
   

5. 2 つの出力埋め込みが互いに類似しているかを測定するためにコサイン類似度メトリックを使用しました。
   

6. train_step() メソッドを override することでカスタム訓練ループを実装できます。train_step() は tf.GradientTape を使用していて、これはそれの内部で実行する総ての演算を記録します。このサンプルでは、ステップ毎にモデル重みを更新するため optimizer に渡された勾配にアクセスするためにそれを使用します。詳細は、Intro to Keras for researchers と 訓練ループをスクラッチから書く を確認してください。

以上