title image similarity estimation using a siamese network with a triplet loss s hazem essamhttps twitter comhazemessamm and santiago l valdarramahttps twitter comsvpino date created 20210325 last modified 20210325 description training a siamese network to compare the similarity of images using a triplet loss function accelerator gpu introduction a siamese networkhttps en wikipedia orgwikisiameseneuralnetwork is a type of network architecture that contains two or more identical subnetworks used to generate feature vectors for each input and compare them siamese networks can be applied to different use cases like detecting duplicates finding anomalies and face recognition this example uses a siamese network with three identical subnetworks we will provide three images to the model where two of them will be similar anchor and positive samples and the third will be unrelated a negative example our goal is for the model to learn to estimate the similarity between images for the network to learn we use a triplet loss function you can find an introduction to triplet loss in the facenet paperhttps arxiv orgpdf1503 03832 pdf by schroff et al 2015 in this example we define the triplet loss function as follows la p n maxfa fp fa fn margin 0 this example uses the totally looks like datasethttps sites google comviewtotallylookslikedataset by rosenfeld et al 2018https arxiv orgpdf1803 01485v3 pdf setup load the dataset we are going to load the totally looks like dataset and unzip it inside the keras directory in the local environment the dataset consists of two separate files left zip contains the images that we will use as the anchor right zip contains the images that we will use as the positive sample an image that looks like the anchor shell gdown id 1jvkbtrgisp3ru8owgncg6b4pvvbco34 gdown id 1ezbzubmhdpfkd0p4xiyysd0qbh5zw unzip oq left zip d cachedir unzip oq right zip d cachedir preparing the data we are going to use a tf data pipeline to load the data and generate the triplets that we need to train the siamese network we ll set up the pipeline using a zipped list with anchor positive and negative filenames as the source the pipeline will load and preprocess the corresponding images load the specified file as a jpeg image preprocess it and resize it to the target shape given the filenames corresponding to the three images load and preprocess them let s setup our data pipeline using a zipped list with an anchor positive and negative image filename as the source the output of the pipeline contains the same triplet with every image loaded and preprocessed we need to make sure both the anchor and positive images are loaded in sorted order so we can match them together to generate the list of negative images let s randomize the list of available images and concatenate them together let s now split our dataset in train and validation let s take a look at a few examples of triplets notice how the first two images look alike while the third one is always different visualize a few triplets from the supplied batches def showax image ax imshowimage ax getxaxis setvisiblefalse ax getyaxis setvisiblefalse fig plt figurefigsize9 9 axs fig subplots3 3 for i in range3 showaxsi 0 anchori showaxsi 1 positivei showaxsi 2 negativei visualizelisttraindataset take1 asnumpyiterator0 setting up the embedding generator model basecnn resnet resnet50 weightsimagenet inputshapetargetshape 3 includetopfalse flatten layers flattenbasecnn output dense1 layers dense512 activationreluflatten dense1 layers batchnormalizationdense1 dense2 layers dense256 activationreludense1 dense2 layers batchnormalizationdense2 output layers dense256dense2 embedding modelbasecnn input output nameembedding trainable false for layer in basecnn layers if layer name conv5block1out trainable true layer trainable trainable setting up the siamese network model class distancelayerlayers layer def initself kwargs super initkwargs def callself anchor positive negative apdistance tf reducesumtf squareanchor positive 1 andistance tf reducesumtf squareanchor negative 1 return apdistance andistance anchorinput layers inputnameanchor shapetargetshape 3 positiveinput layers inputnamepositive shapetargetshape 3 negativeinput layers inputnamenegative shapetargetshape 3 distances distancelayer embeddingresnet preprocessinputanchorinput embeddingresnet preprocessinputpositiveinput embeddingresnet preprocessinputnegativeinput siamesenetwork model inputsanchorinput positiveinput negativeinput outputsdistances putting everything together class siamesemodelmodel def initself siamesenetwork margin0 5 super init self siamesenetwork siamesenetwork self margin margin self losstracker metrics meannameloss def callself inputs return self siamesenetworkinputs def trainstepself 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 computelossdata storing the gradients of the loss function with respect to the weightsparameters gradients tape gradientloss self siamesenetwork trainableweights applying the gradients on the model using the specified optimizer self optimizer applygradients zipgradients self siamesenetwork trainableweights let s update and return the training loss metric self losstracker updatestateloss return loss self losstracker result def teststepself data loss self computelossdata let s update and return the loss metric self losstracker updatestateloss return loss self losstracker result def computelossself 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 apdistance andistance self siamesenetworkdata computing the triplet loss by subtracting both distances and making sure we don t get a negative value loss apdistance andistance loss tf maximumloss self margin 0 0 return loss property def metricsself we need to list our metrics here so the resetstates can be called automatically return self losstracker training siamesemodel siamesemodelsiamesenetwork siamesemodel compileoptimizeroptimizers adam0 0001 siamesemodel fittraindataset epochs10 validationdatavaldataset inspecting what the network has learned sample nextitertraindataset visualizesample anchor positive negative sample anchorembedding positiveembedding negativeembedding embeddingresnet preprocessinputanchor embeddingresnet preprocessinputpositive embeddingresnet preprocessinputnegative cosinesimilarity metrics cosinesimilarity positivesimilarity cosinesimilarityanchorembedding positiveembedding printpositive similarity positivesimilarity numpy negativesimilarity cosinesimilarityanchorembedding negativeembedding printnegative similarity negativesimilarity numpy summary introduction a siamese network https en wikipedia org wiki siamese_neural_network is a type of network architecture that contains two or more identical subnetworks used to generate feature vectors for each input and compare them siamese networks can be applied to different use cases like detecting duplicates finding anomalies and face recognition this example uses a siamese network with three identical subnetworks we will provide three images to the model where two of them will be similar _anchor_ and _positive_ samples and the third will be unrelated a _negative_ example our goal is for the model to learn to estimate the similarity between images for the network to learn we use a triplet loss function you can find an introduction to triplet loss in the facenet paper https arxiv org pdf 1503 03832 pdf by schroff et al 2015 in this example we define the triplet loss function as follows l a p n max f a f p ² f a f n ² margin 0 this example uses the totally looks like dataset https sites google com view totally looks like dataset by rosenfeld et al 2018 https arxiv org pdf 1803 01485v3 pdf setup load the dataset we are going to load the totally looks like dataset and unzip it inside the keras directory in the local environment the dataset consists of two separate files left zip contains the images that we will use as the anchor right zip contains the images that we will use as the positive sample an image that looks like the anchor shell gdown id 1jvkbtr_gisp3ru8owgncg6b4pvvbco34 gdown id 1ezbzub_mh_dp_fkd0p4xiyysd0qbh5zw unzip oq left zip d cache_dir unzip oq right zip d cache_dir preparing the data we are going to use a tf data pipeline to load the data and generate the triplets that we need to train the siamese network we ll set up the pipeline using a zipped list with anchor positive and negative filenames as the source the pipeline will load and preprocess the corresponding images load the specified file as a jpeg image preprocess it and resize it to the target shape given the filenames corresponding to the three images load and preprocess them let s setup our data pipeline using a zipped list with an anchor positive and negative image filename as the source the output of the pipeline contains the same triplet with every image loaded and preprocessed we need to make sure both the anchor and positive images are loaded in sorted order so we can match them together to generate the list of negative images let s randomize the list of available images and concatenate them together let s now split our dataset in train and validation let s take a look at a few examples of triplets notice how the first two images look alike while the third one is always different visualize a few triplets from the supplied batches setting up the embedding generator model our siamese network will generate embeddings for each of the images of the triplet to do this we will use a resnet50 model pretrained on imagenet and connect a few dense layers to it so we can learn to separate these embeddings we will freeze the weights of all the layers of the model up until the layer conv5_block1_out this is important to avoid affecting the weights that the model has already learned we are going to leave the bottom few layers trainable so that we can fine tune their weights during training setting up the siamese network model the siamese network will receive each of the triplet images as an input generate the embeddings and output the distance between the anchor and the positive embedding as well as the distance between the anchor and the negative embedding to compute the distance we can use a custom layer distancelayer that returns both values as a tuple this layer is responsible for computing the distance between the anchor embedding and the positive embedding and the anchor embedding and the negative embedding putting everything together we now need to implement a model with custom training loop so we can compute the triplet loss using the three embeddings produced by the siamese network let s create a mean metric instance to track the loss of the training process 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 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 storing the gradients of the loss function with respect to the weights parameters applying the gradients on the model using the specified optimizer let s update and return the training loss metric let s update and return the loss metric 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 computing the triplet loss by subtracting both distances and making sure we don t get a negative value we need to list our metrics here so the reset_states can be called automatically training we are now ready to train our model inspecting what the network has learned at this point we can check how the network learned to separate the embeddings depending on whether they belong to similar images we can use cosine similarity https en wikipedia org wiki cosine_similarity to measure the similarity between embeddings let s pick a sample from the dataset to check the similarity between the embeddings generated for each image finally we can compute the cosine similarity between the anchor and positive images and compare it with the similarity between the anchor and the negative images we should expect the similarity between the anchor and positive images to be larger than the similarity between the anchor and the negative images summary 1 the tf data api enables you to build efficient input pipelines for your model it is particularly useful if you have a large dataset you can learn more about tf data pipelines in tf data build tensorflow input pipelines https www tensorflow org guide data 2 in this example we use a pre trained resnet50 as part of the subnetwork that generates the feature embeddings by using transfer learning https www tensorflow org guide keras transfer_learning hl en we can significantly reduce the training time and size of the dataset 3 notice how we are fine tuning https www tensorflow org guide keras transfer_learning hl en fine tuning the weights of the final layers of the resnet50 network but keeping the rest of the layers untouched using the name assigned to each layer we can freeze the weights to a certain point and keep the last few layers open 4 we can create custom layers by creating a class that inherits from tf keras layers layer as we did in the distancelayer class 5 we used a cosine similarity metric to measure how to 2 output embeddings are similar to each other 6 you can implement a custom training loop by overriding the train_step method train_step uses tf gradienttape https www tensorflow org api_docs python tf gradienttape which records every operation that you perform inside it in this example we use it to access the gradients passed to the optimizer to update the model weights at every step for more details check out the intro to keras for researchers https keras io getting_started intro_to_keras_for_researchers and writing a training loop from scratch https www tensorflow org guide keras writing_a_training_loop_from_scratch hl en
import matplotlib.pyplot as plt
import numpy as np
import os
import tensorflow as tf
from pathlib import Path
from keras import layers
from keras import optimizers
from keras import metrics
from keras import Model
from keras.applications import resnet


target_shape = (200, 200)




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






def preprocess_image(filename):
    

    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):
    

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






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)



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)


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(tf.data.AUTOTUNE)

val_dataset = val_dataset.batch(32, drop_remainder=False)
val_dataset = val_dataset.prefetch(tf.data.AUTOTUNE)





def visualize(anchor, positive, negative):
    

    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])



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




class DistanceLayer(layers.Layer):
    

    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
)




class SiameseModel(Model):
    

    def __init__(self, siamese_network, margin=0.5):
        super().__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):
        
        
        
        
        with tf.GradientTape() as tape:
            loss = self._compute_loss(data)

        
        
        gradients = tape.gradient(loss, self.siamese_network.trainable_weights)

        
        self.optimizer.apply_gradients(
            zip(gradients, self.siamese_network.trainable_weights)
        )

        
        self.loss_tracker.update_state(loss)
        return {"loss": self.loss_tracker.result()}

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

        
        self.loss_tracker.update_state(loss)
        return {"loss": self.loss_tracker.result()}

    def _compute_loss(self, data):
        
        
        
        ap_distance, an_distance = self.siamese_network(data)

        
        
        loss = ap_distance - an_distance
        loss = tf.maximum(loss + self.margin, 0.0)
        return loss

    @property
    def metrics(self):
        
        
        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)


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())