\n",
"\n",
"**Question**\n",
"\n",
"Compléter la classe `ContrastiveTransformations` ci-dessous, qui doit appliquer `n_views` fois la transformation `base_transforms` à une même image `x`, et renvoyer la liste des vues ainsi obtenues.\n",
"\n",
"
"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "91d3f9fe",
"metadata": {},
"outputs": [],
"source": [
"class ContrastiveTransformations(object):\n",
"\n",
" def __init__(self, base_transforms, n_views=2):\n",
" self.base_transforms = base_transforms\n",
" self.n_views = n_views\n",
"\n",
" def __call__(self, x):\n",
" # À compléter\n",
" ..."
]
},
{
"cell_type": "markdown",
"id": "3011b695",
"metadata": {},
"source": [
"Le cadre de l'apprentissage contrastif peut facilement être étendu à davantage d'exemples *positifs*, en échantillonnant plus de deux augmentations d'une même image. Cependant, l'entraînement le plus efficace est en général obtenu avec seulement deux vues.\n",
"\n",
"Intéressons-nous maintenant aux augmentations spécifiques que nous allons utiliser. Le choix de l'augmentation de données est l'hyperparamètre le plus déterminant de SimCLR, puisqu'il affecte directement la structuration de l'espace latent et les motifs que le modèle pourra apprendre des données. Voici quelques-unes des augmentations les plus utilisées (crédit de la figure : [Ting Chen et Geoffrey Hinton](https://ai.googleblog.com/2020/04/advancing-self-supervised-and-semi.html)) :\n",
"\n",
"\n",
"\n",
"Si la situation (a) demande au modèle d'apprendre une forme d'invariance d'échelle pour rapprocher A et B dans l'espace latent, la situation (b) est plus exigeante puisque le modèle doit reconnaître un objet au-delà de sa vue limitée. Cependant, sans distorsion des couleurs, le modèle dispose d'une échappatoire : différents recadrages d'une même image se ressemblent en général beaucoup dans l'espace des couleurs. Il suffirait alors de comparer les histogrammes de couleurs pour deviner que deux patchs proviennent de la même image, sans avoir besoin de reconnaître réellement son contenu — le modèle risquerait alors de se focaliser uniquement sur cette information peu généralisable. En revanche, si l'on distord aléatoirement et indépendamment les couleurs des deux patchs, le modèle ne peut plus s'appuyer sur ce raccourci : en combinant recadrage aléatoire et distorsion des couleurs, il ne peut faire correspondre deux patchs qu'en apprenant des représentations réellement généralisables.\n",
"\n",
"Pour nos expériences, nous appliquons un ensemble de 5 transformations, suivant la configuration originale de SimCLR : retournement horizontal aléatoire, recadrage-redimensionnement, distorsion des couleurs, mise en niveaux de gris aléatoire, et flou gaussien. Par rapport à l'[implémentation originale](https://github.com/google-research/simclr), nous réduisons légèrement l'effet de la distorsion des couleurs (0,5 au lieu de 0,8 pour la luminosité, le contraste et la saturation, et 0,1 au lieu de 0,2 pour la teinte), ce qui a donné de meilleurs résultats, plus rapides et plus stables à entraîner dans nos expériences."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "d3d1a57b",
"metadata": {},
"outputs": [],
"source": [
"contrast_transforms = transforms.Compose([transforms.RandomHorizontalFlip(),\n",
" transforms.RandomResizedCrop(size=96),\n",
" transforms.RandomApply([\n",
" transforms.ColorJitter(brightness=0.5,\n",
" contrast=0.5,\n",
" saturation=0.5,\n",
" hue=0.1)\n",
" ], p=0.8),\n",
" transforms.RandomGrayscale(p=0.2),\n",
" transforms.GaussianBlur(kernel_size=9),\n",
" transforms.ToTensor(),\n",
" transforms.Normalize((0.5,), (0.5,))\n",
" ])"
]
},
{
"cell_type": "markdown",
"id": "54441753",
"metadata": {},
"source": [
"Après avoir discuté des techniques d'augmentation de données, intéressons-nous au jeu de données. Pour ce TP, nous utilisons [STL10](https://cs.stanford.edu/~acoates/stl10/) qui, comme CIFAR10, contient des images de 10 classes : avion, oiseau, voiture, chat, cerf, chien, cheval, singe, bateau, camion. Les images ont cependant une résolution plus élevée ($96\\times 96$ pixels), et l'on ne dispose que de 500 images étiquetées par classe. En complément, un jeu bien plus grand de $100\\,000$ images **non-étiquetées** est fourni, similaires aux images d'entraînement mais échantillonnées parmi un éventail plus large d'animaux et de véhicules — ce qui rend ce jeu de données idéal pour illustrer les bénéfices de l'apprentissage auto-supervisé.\n",
"\n",
"STL10 est fourni nativement par `torchvision`. Attention : ce jeu de données est relativement volumineux et de résolution plus élevée que CIFAR10 (~3Go), et son téléchargement peut prendre du temps. Pour notre étude de SimCLR, nous créons deux jeux de données avec nos transformations contrastives ci-dessus : `unlabeled_data` servira à entraîner notre modèle par apprentissage contrastif, et `train_data_contrast` servira de jeu de validation."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "319f9628",
"metadata": {},
"outputs": [],
"source": [
"unlabeled_data = STL10(root=DATASET_PATH, split='unlabeled', download=True,\n",
" transform=ContrastiveTransformations(contrast_transforms, n_views=2))\n",
"train_data_contrast = STL10(root=DATASET_PATH, split='train', download=True,\n",
" transform=ContrastiveTransformations(contrast_transforms, n_views=2))"
]
},
{
"cell_type": "markdown",
"id": "79a4bbb3",
"metadata": {},
"source": [
"Avant de nous lancer dans l'implémentation de SimCLR, visualisons quelques exemples de paires d'images obtenues par nos augmentations :"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "fa08fa83",
"metadata": {},
"outputs": [],
"source": [
"# Visualisation de quelques exemples\n",
"pl.seed_everything(42)\n",
"NUM_IMAGES = 6\n",
"imgs = torch.stack([img for idx in range(NUM_IMAGES) for img in unlabeled_data[idx][0]], dim=0)\n",
"img_grid = torchvision.utils.make_grid(imgs, nrow=6, normalize=True, pad_value=0.9)\n",
"img_grid = img_grid.permute(1, 2, 0)\n",
"\n",
"plt.figure(figsize=(10,5))\n",
"plt.title('Exemples augmentés du jeu de données STL10')\n",
"plt.imshow(img_grid)\n",
"plt.axis('off')\n",
"plt.show()\n",
"plt.close()"
]
},
{
"cell_type": "markdown",
"id": "e6f9c471",
"metadata": {},
"source": [
"On observe la grande variété de nos augmentations de données : recadrage aléatoire, niveaux de gris, flou gaussien, distorsion des couleurs. Faire correspondre deux patchs augmentés indépendamment d'une même image reste donc une tâche exigeante pour le modèle."
]
},
{
"cell_type": "markdown",
"id": "98ea9c25",
"metadata": {},
"source": [
"### Implémentation de SimCLR\n",
"\n",
"À l'aide du pipeline de chargement de données ci-dessus, nous pouvons maintenant implémenter SimCLR. À chaque itération, pour chaque image $x$, nous obtenons deux versions différemment augmentées, notées $\\tilde x_i$ et $\\tilde x_j$. Ces deux images sont encodées en un vecteur de caractéristiques 1D, entre lesquels on cherche à maximiser la similarité, tout en la minimisant avec toutes les autres images du batch. Le réseau encodeur est décomposé en deux parties : un réseau encodeur de base $f(\\cdot)$, et une tête de projection $g(\\cdot)$. Le réseau de base est en général un CNN profond, chargé d'extraire un vecteur de représentation à partir des exemples augmentés — nous utiliserons l'architecture ResNet-18 usuelle pour $f(\\cdot)$, et noterons sa sortie $f(\\tilde x_i) = h_i$. La tête de projection $g(\\cdot)$ projette la représentation $h$ dans l'espace où la perte contrastive est appliquée (comparaison de similarités entre vecteurs) : on la choisit en général comme un petit MLP muni de non-linéarités ; pour simplifier, nous suivons la configuration de l'article original de SimCLR, avec un MLP à deux couches et une activation ReLU dans la couche cachée.\n",
"\n",
"Après l'entraînement par apprentissage contrastif, nous retirerons la tête de projection $g(\\cdot)$ et utiliserons uniquement $f(\\cdot)$ comme extracteur de caractéristiques pré-entraîné : les représentations $z$ en sortie de $g(\\cdot)$ se sont en effet révélées moins performantes que celles de $f(\\cdot)$ pour le réglage fin sur une nouvelle tâche, vraisemblablement parce que $z$ est entraîné à devenir invariant à des caractéristiques (comme la couleur) qui peuvent pourtant être importantes pour les tâches *downstream*. La tête $g(\\cdot)$ n'est donc utile que pour la phase d'apprentissage contrastif.\n",
"\n",
"Voyons maintenant comment entraîner le modèle. Comme indiqué précédemment, on cherche à maximiser la similarité entre les représentations des deux versions augmentées d'une même image ($z_i$ et $z_j$), tout en la minimisant avec tous les autres exemples du batch. SimCLR applique pour cela la perte **InfoNCE**, initialement proposée par [van den Oord et al.](https://arxiv.org/abs/1807.03748) (voir le :ref:`chapitre précédent `). En résumé, la perte InfoNCE compare la similarité de $z_i$ et $z_j$ à la similarité de $z_i$ avec tous les autres exemples du batch, par un softmax sur les valeurs de similarité :\n",
"\n",
"$$\n",
"\\ell_{i,j}=-\\log \\frac{\\exp(\\text{sim}(z_i,z_j)/\\tau)}{\\sum_{k=1}^{2N}\\mathbb{1}_{[k\\neq i]}\\exp(\\text{sim}(z_i,z_k)/\\tau)}\n",
"$$\n",
"\n",
"La fonction $\\text{sim}$ est une mesure de similarité, et l'hyperparamètre $\\tau$ (la *température*) détermine à quel point la distribution est « piquée ». La similarité utilisée dans SimCLR est le cosinus :\n",
"\n",
"$$\n",
"\\text{sim}(z_i,z_j) = \\frac{z_i^\\top z_j}{\\lVert z_i\\rVert \\cdot \\lVert z_j\\rVert}\n",
"$$\n",
"\n",
"Implémentons maintenant SimCLR sous la forme d'un module PyTorch Lightning :"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "cc47f0ee",
"metadata": {},
"outputs": [],
"source": [
"class SimCLR(pl.LightningModule):\n",
"\n",
" def __init__(self, hidden_dim, lr, temperature, weight_decay, max_epochs=500):\n",
" super().__init__()\n",
" self.save_hyperparameters()\n",
" assert self.hparams.temperature > 0.0, 'La température doit être un flottant positif !'\n",
" # Modèle de base f(.)\n",
" self.convnet = torchvision.models.resnet18(num_classes=4*hidden_dim) # Sortie de la dernière couche linéaire\n",
" # Le MLP pour g(.) : Linear -> ReLU -> Linear\n",
" self.convnet.fc = nn.Sequential(\n",
" self.convnet.fc, # Linear(sortie du ResNet, 4*hidden_dim)\n",
" nn.ReLU(inplace=True),\n",
" nn.Linear(4*hidden_dim, hidden_dim)\n",
" )\n",
"\n",
" def configure_optimizers(self):\n",
" optimizer = optim.AdamW(self.parameters(),\n",
" lr=self.hparams.lr,\n",
" weight_decay=self.hparams.weight_decay)\n",
" lr_scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer,\n",
" T_max=self.hparams.max_epochs,\n",
" eta_min=self.hparams.lr/50)\n",
" return [optimizer], [lr_scheduler]\n",
"\n",
" def info_nce_loss(self, batch, mode='train'):\n",
" imgs, _ = batch\n",
" imgs = torch.cat(imgs, dim=0)\n",
"\n",
" # Encode toutes les images\n",
" feats = self.convnet(imgs)\n",
" # Calcule la similarité cosinus\n",
" cos_sim = F.cosine_similarity(feats[:,None,:], feats[None,:,:], dim=-1)\n",
" # Masque la similarité cosinus d'un exemple avec lui-même\n",
" self_mask = torch.eye(cos_sim.shape[0], dtype=torch.bool, device=cos_sim.device)\n",
" cos_sim.masked_fill_(self_mask, -9e15)\n",
" # Trouve l'exemple positif -> décalé de batch_size//2 par rapport à l'exemple original\n",
" pos_mask = self_mask.roll(shifts=cos_sim.shape[0]//2, dims=0)\n",
" # Perte InfoNCE\n",
" cos_sim = cos_sim / self.hparams.temperature\n",
" nll = -cos_sim[pos_mask] + torch.logsumexp(cos_sim, dim=-1)\n",
" nll = nll.mean()\n",
"\n",
" # Enregistrement de la perte\n",
" self.log(mode+'_loss', nll)\n",
" # Position de l'exemple positif dans le classement des similarités\n",
" comb_sim = torch.cat([cos_sim[pos_mask][:,None], # L'exemple positif en première position\n",
" cos_sim.masked_fill(pos_mask, -9e15)],\n",
" dim=-1)\n",
" sim_argsort = comb_sim.argsort(dim=-1, descending=True).argmin(dim=-1)\n",
" # Enregistrement des métriques de classement\n",
" self.log(mode+'_acc_top1', (sim_argsort == 0).float().mean())\n",
" self.log(mode+'_acc_top5', (sim_argsort < 5).float().mean())\n",
" self.log(mode+'_acc_mean_pos', 1+sim_argsort.float().mean())\n",
"\n",
" return nll\n",
"\n",
" def training_step(self, batch, batch_idx):\n",
" return self.info_nce_loss(batch, mode='train')\n",
"\n",
" def validation_step(self, batch, batch_idx):\n",
" self.info_nce_loss(batch, mode='val')"
]
},
{
"cell_type": "markdown",
"id": "8cd1308b",
"metadata": {},
"source": [
"\n",
"\n",
"**Question**\n",
"\n",
"Dans la méthode `info_nce_loss` ci-dessus, à quoi correspond `pos_mask` ? Pourquoi le décalage (`roll`) utilisé pour le construire est-il de `cos_sim.shape[0]//2` ?\n",
"\n",
"
\n",
"\n",
"\n",
"\n",
"**Correction**\n",
"\n",
"Pour un batch de $N$ images, `imgs` contient $2N$ vues après concaténation (les deux augmentations de chaque image, dans le même ordre : d'abord toutes les premières vues, puis toutes les secondes vues). `pos_mask` repère, pour chaque vue, la position de son unique exemple positif (l'autre vue de la même image) parmi les $2N$ vues. Comme les vues sont organisées en deux blocs de taille $N$ dans le même ordre, la vue positive associée à la vue d'indice $i$ se trouve exactement $N = \\text{cos\\_sim.shape[0]//2}$ positions plus loin (modulo $2N$) : d'où le décalage circulaire (`roll`) de cette valeur appliqué au masque diagonal (identité).\n",
"\n",
"
\n",
"\n",
"Plutôt que de valider également sur la perte d'apprentissage contrastif, on pourrait suivre la performance du réseau de base $f(\\cdot)$ sur une petite tâche *downstream*. Nous nous limiterons cependant ici au jeu de données STL10 et à sa tâche de classification comme tâche de test."
]
},
{
"cell_type": "markdown",
"id": "f81f7ab4",
"metadata": {},
"source": [
"### Entraînement\n",
"\n",
"Maintenant que SimCLR et le pipeline de chargement des données sont implémentés, nous pouvons entraîner le modèle, à l'aide de PyTorch Lightning pour simplifier. Pour sauvegarder le meilleur *checkpoint*, nous suivons la métrique `val_acc_top5`, qui indique à quelle fréquence le bon patch se trouve parmi les 5 exemples les plus similaires du batch — une métrique généralement moins bruitée que le top-1, donc plus fiable pour sélectionner le meilleur modèle."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "986a4c9e",
"metadata": {},
"outputs": [],
"source": [
"def train_simclr(batch_size, max_epochs=500, **kwargs):\n",
" trainer = pl.Trainer(default_root_dir=os.path.join(CHECKPOINT_PATH, 'SimCLR'),\n",
" accelerator=\"gpu\" if str(device).startswith(\"cuda\") else \"cpu\",\n",
" devices=1,\n",
" max_epochs=max_epochs,\n",
" callbacks=[ModelCheckpoint(save_weights_only=True, mode='max', monitor='val_acc_top5'),\n",
" LearningRateMonitor('epoch')])\n",
" trainer.logger._default_hp_metric = None # Option de log dont nous n'avons pas besoin\n",
"\n",
" # Vérifie si un modèle pré-entraîné existe déjà. Si oui, le charger et sauter l'entraînement\n",
" pretrained_filename = os.path.join(CHECKPOINT_PATH, 'SimCLR.ckpt')\n",
" if os.path.isfile(pretrained_filename):\n",
" print(f'Modèle pré-entraîné trouvé à {pretrained_filename}, chargement...')\n",
" model = SimCLR.load_from_checkpoint(pretrained_filename) # Charge automatiquement le modèle avec ses hyperparamètres\n",
" else:\n",
" train_loader = data.DataLoader(unlabeled_data, batch_size=batch_size, shuffle=True,\n",
" drop_last=True, pin_memory=True, num_workers=NUM_WORKERS)\n",
" val_loader = data.DataLoader(train_data_contrast, batch_size=batch_size, shuffle=False,\n",
" drop_last=False, pin_memory=True, num_workers=NUM_WORKERS)\n",
" pl.seed_everything(42) # Pour la reproductibilité\n",
" model = SimCLR(max_epochs=max_epochs, **kwargs)\n",
" trainer.fit(model, train_loader, val_loader)\n",
" model = SimCLR.load_from_checkpoint(trainer.checkpoint_callback.best_model_path) # Charge le meilleur checkpoint\n",
"\n",
" return model"
]
},
{
"cell_type": "markdown",
"id": "5b70895f",
"metadata": {},
"source": [
"En apprentissage contrastif, on observe généralement que plus le batch est grand, meilleures sont les performances : un grand batch permet de comparer chaque image à davantage d'exemples négatifs, ce qui lisse les gradients de la perte. Dans notre cas cependant, un batch de taille 256 s'est révélé suffisant pour obtenir de bons résultats."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "022db477",
"metadata": {},
"outputs": [],
"source": [
"simclr_model = train_simclr(batch_size=256,\n",
" hidden_dim=128,\n",
" lr=5e-4,\n",
" temperature=0.07,\n",
" weight_decay=1e-4,\n",
" max_epochs=500)"
]
},
{
"cell_type": "markdown",
"id": "cd9833e9",
"metadata": {},
"source": [
"Pour avoir une intuition du déroulement de l'entraînement par apprentissage contrastif, on peut consulter TensorBoard :"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "2e3d4367",
"metadata": {},
"outputs": [],
"source": [
"%tensorboard --logdir ../saved_models/TP_SSL/tensorboards/SimCLR/"
]
},
{
"cell_type": "markdown",
"id": "3ce2ec06",
"metadata": {},
"source": [
"\n",
"\n",
"L'apprentissage contrastif bénéficie beaucoup d'un entraînement long : la courbe ci-dessus provient d'un entraînement d'environ une journée sur une NVIDIA TitanRTX. Entraîner le modèle plus longtemps encore pourrait réduire davantage sa perte, mais sans gain observé sur la tâche *downstream* de classification. De façon générale, l'apprentissage contrastif bénéficie également de modèles plus grands, si suffisamment de données non-étiquetées sont disponibles."
]
},
{
"cell_type": "markdown",
"id": "70f52713",
"metadata": {},
"source": [
"## Régression logistique\n",
"\n",
"Une fois notre modèle entraîné par apprentissage contrastif, nous pouvons l'utiliser sur des tâches *downstream* et évaluer sa performance avec peu de données. Une méthode courante, qui vérifie aussi si le modèle a appris des représentations généralisables, consiste à entraîner une **régression logistique** sur les caractéristiques extraites : autrement dit, on apprend une unique couche linéaire qui associe les représentations à une prédiction de classe. Le réseau de base $f(\\cdot)$ n'étant pas modifié pendant cet entraînement, le modèle ne peut bien fonctionner que si les représentations $h$ décrivent déjà toute l'information nécessaire à la tâche. De plus, le risque de sur-apprentissage est limité puisque très peu de paramètres sont entraînés : on peut donc s'attendre à de bonnes performances même avec très peu de données.\n",
"\n",
"Nous supposons ici que les images ont déjà été encodées sous forme de vecteurs de caractéristiques (plutôt que de les encoder à la volée pendant l'entraînement) : cette approche est beaucoup plus efficace et permet un entraînement en quelques secondes."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "ee69a464",
"metadata": {},
"outputs": [],
"source": [
"class LogisticRegression(pl.LightningModule):\n",
"\n",
" def __init__(self, feature_dim, num_classes, lr, weight_decay, max_epochs=100):\n",
" super().__init__()\n",
" self.save_hyperparameters()\n",
" # Projection de la représentation h vers les classes\n",
" self.model = nn.Linear(feature_dim, num_classes)\n",
"\n",
" def configure_optimizers(self):\n",
" optimizer = optim.AdamW(self.parameters(),\n",
" lr=self.hparams.lr,\n",
" weight_decay=self.hparams.weight_decay)\n",
" lr_scheduler = optim.lr_scheduler.MultiStepLR(optimizer,\n",
" milestones=[int(self.hparams.max_epochs*0.6),\n",
" int(self.hparams.max_epochs*0.8)],\n",
" gamma=0.1)\n",
" return [optimizer], [lr_scheduler]\n",
"\n",
" def _calculate_loss(self, batch, mode='train'):\n",
" feats, labels = batch\n",
" preds = self.model(feats)\n",
" loss = F.cross_entropy(preds, labels)\n",
" acc = (preds.argmax(dim=-1) == labels).float().mean()\n",
"\n",
" self.log(mode + '_loss', loss)\n",
" self.log(mode + '_acc', acc)\n",
" return loss\n",
"\n",
" def training_step(self, batch, batch_idx):\n",
" return self._calculate_loss(batch, mode='train')\n",
"\n",
" def validation_step(self, batch, batch_idx):\n",
" self._calculate_loss(batch, mode='val')\n",
"\n",
" def test_step(self, batch, batch_idx):\n",
" self._calculate_loss(batch, mode='test')"
]
},
{
"cell_type": "markdown",
"id": "50577fd6",
"metadata": {},
"source": [
"Nous utilisons les jeux d'entraînement et de test de STL10 : 500 images par classe pour l'entraînement, 800 par classe pour le test."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "5aa81400",
"metadata": {},
"outputs": [],
"source": [
"img_transforms = transforms.Compose([transforms.ToTensor(),\n",
" transforms.Normalize((0.5,), (0.5,))])\n",
"\n",
"train_img_data = STL10(root=DATASET_PATH, split='train', download=True,\n",
" transform=img_transforms)\n",
"test_img_data = STL10(root=DATASET_PATH, split='test', download=True,\n",
" transform=img_transforms)\n",
"\n",
"print(\"Nombre d'exemples d'entraînement :\", len(train_img_data))\n",
"print(\"Nombre d'exemples de test :\", len(test_img_data))"
]
},
{
"cell_type": "markdown",
"id": "c55d7e1b",
"metadata": {},
"source": [
"Implémentons maintenant une fonction qui encode toutes les images d'un jeu de données : les représentations obtenues serviront d'entrée au modèle de régression logistique."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "ab6fb01b",
"metadata": {},
"outputs": [],
"source": [
"@torch.no_grad()\n",
"def prepare_data_features(model, dataset):\n",
" # Préparation du modèle\n",
" network = deepcopy(model.convnet)\n",
" network.fc = nn.Identity() # Retire la tête de projection g(.)\n",
" network.eval()\n",
" network.to(device)\n",
"\n",
" # Encode toutes les images\n",
" data_loader = data.DataLoader(dataset, batch_size=64, num_workers=NUM_WORKERS, shuffle=False, drop_last=False)\n",
" feats, labels = [], []\n",
" for batch_imgs, batch_labels in tqdm(data_loader):\n",
" batch_imgs = batch_imgs.to(device)\n",
" batch_feats = network(batch_imgs)\n",
" feats.append(batch_feats.detach().cpu())\n",
" labels.append(batch_labels)\n",
"\n",
" feats = torch.cat(feats, dim=0)\n",
" labels = torch.cat(labels, dim=0)\n",
"\n",
" # Trie les images par étiquette\n",
" labels, idxs = labels.sort()\n",
" feats = feats[idxs]\n",
"\n",
" return data.TensorDataset(feats, labels)"
]
},
{
"cell_type": "markdown",
"id": "bcc15b66",
"metadata": {},
"source": [
"Appliquons cette fonction aux jeux d'entraînement et de test :"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "5ad1ab42",
"metadata": {},
"outputs": [],
"source": [
"train_feats_simclr = prepare_data_features(simclr_model, train_img_data)\n",
"test_feats_simclr = prepare_data_features(simclr_model, test_img_data)"
]
},
{
"cell_type": "markdown",
"id": "706e3aa9",
"metadata": {},
"source": [
"Écrivons maintenant une fonction d'entraînement classique. Nous évaluons le modèle sur le jeu de test toutes les 10 époques (pour permettre un arrêt anticipé), la faible fréquence de validation limitant le risque de sur-apprentissage sur le jeu de test."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "ec5f9a6a",
"metadata": {},
"outputs": [],
"source": [
"def train_logreg(batch_size, train_feats_data, test_feats_data, model_suffix, max_epochs=100, **kwargs):\n",
" trainer = pl.Trainer(default_root_dir=os.path.join(CHECKPOINT_PATH, \"LogisticRegression\"),\n",
" accelerator=\"gpu\" if str(device).startswith(\"cuda\") else \"cpu\",\n",
" devices=1,\n",
" max_epochs=max_epochs,\n",
" callbacks=[ModelCheckpoint(save_weights_only=True, mode='max', monitor='val_acc'),\n",
" LearningRateMonitor(\"epoch\")],\n",
" enable_progress_bar=False,\n",
" check_val_every_n_epoch=10)\n",
" trainer.logger._default_hp_metric = None\n",
"\n",
" # Chargeurs de données\n",
" train_loader = data.DataLoader(train_feats_data, batch_size=batch_size, shuffle=True,\n",
" drop_last=False, pin_memory=True, num_workers=0)\n",
" test_loader = data.DataLoader(test_feats_data, batch_size=batch_size, shuffle=False,\n",
" drop_last=False, pin_memory=True, num_workers=0)\n",
"\n",
" # Vérifie si un modèle pré-entraîné existe déjà. Si oui, le charger et sauter l'entraînement\n",
" pretrained_filename = os.path.join(CHECKPOINT_PATH, f\"LogisticRegression_{model_suffix}.ckpt\")\n",
" if os.path.isfile(pretrained_filename):\n",
" print(f\"Modèle pré-entraîné trouvé à {pretrained_filename}, chargement...\")\n",
" model = LogisticRegression.load_from_checkpoint(pretrained_filename)\n",
" else:\n",
" pl.seed_everything(42) # Pour la reproductibilité\n",
" model = LogisticRegression(**kwargs)\n",
" trainer.fit(model, train_loader, test_loader)\n",
" model = LogisticRegression.load_from_checkpoint(trainer.checkpoint_callback.best_model_path)\n",
"\n",
" # Évalue le meilleur modèle sur les jeux d'entraînement et de test\n",
" train_result = trainer.test(model, train_loader, verbose=False)\n",
" test_result = trainer.test(model, test_loader, verbose=False)\n",
" result = {\"train\": train_result[0][\"test_acc\"], \"test\": test_result[0][\"test_acc\"]}\n",
"\n",
" return model, result"
]
},
{
"cell_type": "markdown",
"id": "3c4261a1",
"metadata": {},
"source": [
"Bien que le jeu d'entraînement de STL10 ne compte déjà que 500 images étiquetées par classe, nous allons expérimenter avec des jeux encore plus petits. Nous entraînons ainsi une régression logistique avec seulement 10, 20, 50, 100, 200 puis les 500 exemples par classe, ce qui nous donnera une intuition de la capacité de transfert des représentations apprises par apprentissage contrastif vers une tâche de reconnaissance d'image. Définissons d'abord une fonction pour créer ces sous-jeux de données :"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "96087ce1",
"metadata": {},
"outputs": [],
"source": [
"def get_smaller_dataset(original_dataset, num_imgs_per_label):\n",
" new_dataset = data.TensorDataset(\n",
" *[t.unflatten(0, (10, -1))[:,:num_imgs_per_label].flatten(0, 1) for t in original_dataset.tensors]\n",
" )\n",
" return new_dataset"
]
},
{
"cell_type": "markdown",
"id": "06e8933a",
"metadata": {},
"source": [
"Exécutons maintenant tous les modèles. Bien que 6 modèles soient entraînés, cette cellule ne prend qu'une minute ou deux grâce aux modèles pré-entraînés fournis."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "2eac0bee",
"metadata": {},
"outputs": [],
"source": [
"results = {}\n",
"for num_imgs_per_label in [10, 20, 50, 100, 200, 500]:\n",
" sub_train_set = get_smaller_dataset(train_feats_simclr, num_imgs_per_label)\n",
" _, small_set_results = train_logreg(batch_size=64,\n",
" train_feats_data=sub_train_set,\n",
" test_feats_data=test_feats_simclr,\n",
" model_suffix=num_imgs_per_label,\n",
" feature_dim=train_feats_simclr.tensors[0].shape[1],\n",
" num_classes=10,\n",
" lr=1e-3,\n",
" weight_decay=1e-3)\n",
" results[num_imgs_per_label] = small_set_results"
]
},
{
"cell_type": "markdown",
"id": "075c3311",
"metadata": {},
"source": [
"Affichons enfin les résultats :"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "0ee4cff3",
"metadata": {},
"outputs": [],
"source": [
"dataset_sizes = sorted([k for k in results])\n",
"test_scores = [results[k][\"test\"] for k in dataset_sizes]\n",
"\n",
"fig = plt.figure(figsize=(6,4))\n",
"plt.plot(dataset_sizes, test_scores, '--', color=\"#000\", marker=\"*\", markeredgecolor=\"#000\", markerfacecolor=\"y\", markersize=16)\n",
"plt.xscale(\"log\")\n",
"plt.xticks(dataset_sizes, labels=dataset_sizes)\n",
"plt.title(\"Classification STL10 selon la taille du jeu de données\", fontsize=14)\n",
"plt.xlabel(\"Nombre d'images par classe\")\n",
"plt.ylabel(\"Accuracy de test\")\n",
"plt.minorticks_off()\n",
"plt.show()\n",
"\n",
"for k, score in zip(dataset_sizes, test_scores):\n",
" print(f'Accuracy de test pour {k:3d} images par classe : {100*score:4.2f}%')"
]
},
{
"cell_type": "markdown",
"id": "bd5ca676",
"metadata": {},
"source": [
"\n",
"\n",
"**Question**\n",
"\n",
"Que constatez-vous sur l'évolution de l'accuracy de test en fonction du nombre d'images étiquetées par classe ? Le résultat avec seulement 10 images par classe vous semble-t-il remarquable ?\n",
"\n",
"
\n",
"\n",
"\n",
"\n",
"**Correction**\n",
"\n",
"Comme attendu, la performance de classification s'améliore avec la quantité de données disponibles. Cependant, avec seulement 10 images par classe, on classifie déjà plus de 60% des images correctement — un résultat remarquable, d'autant que les images sont de dimension plus élevée que, par exemple, CIFAR10. Avec le jeu de données complet, on atteint 81% d'accuracy. La progression entre 50 et 500 images par classe suggère une croissance à peu près linéaire de la performance pour une croissance exponentielle de la taille du jeu de données. Avec encore plus de données, on pourrait également envisager de *fine-tuner* $f(\\cdot)$ lui-même, ce qui permettrait aux représentations de s'adapter davantage à la tâche de classification considérée.\n",
"\n",
"Pour mettre ces résultats en perspective, comparons-les à un ResNet-18 entraîné **from scratch**, en supervision complète, sur ce même jeu de données.\n",
"\n",
"
"
]
},
{
"cell_type": "markdown",
"id": "a08e3168",
"metadata": {},
"source": [
"## Référence (*baseline*)\n",
"\n",
"Comme référence à nos résultats précédents, nous entraînons un ResNet-18 standard, initialisé aléatoirement, sur le jeu d'entraînement étiqueté de STL10. Les résultats nous donneront une indication des avantages de l'apprentissage contrastif sur données non-étiquetées, par rapport à un entraînement purement supervisé. L'implémentation est directe, puisque l'architecture ResNet est fournie par `torchvision`."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "a9720d43",
"metadata": {},
"outputs": [],
"source": [
"class ResNet(pl.LightningModule):\n",
"\n",
" def __init__(self, num_classes, lr, weight_decay, max_epochs=100):\n",
" super().__init__()\n",
" self.save_hyperparameters()\n",
" self.model = torchvision.models.resnet18(num_classes=num_classes)\n",
"\n",
" def configure_optimizers(self):\n",
" optimizer = optim.AdamW(self.parameters(),\n",
" lr=self.hparams.lr,\n",
" weight_decay=self.hparams.weight_decay)\n",
" lr_scheduler = optim.lr_scheduler.MultiStepLR(optimizer,\n",
" milestones=[int(self.hparams.max_epochs*0.7),\n",
" int(self.hparams.max_epochs*0.9)],\n",
" gamma=0.1)\n",
" return [optimizer], [lr_scheduler]\n",
"\n",
" def _calculate_loss(self, batch, mode='train'):\n",
" imgs, labels = batch\n",
" preds = self.model(imgs)\n",
" loss = F.cross_entropy(preds, labels)\n",
" acc = (preds.argmax(dim=-1) == labels).float().mean()\n",
"\n",
" self.log(mode + '_loss', loss)\n",
" self.log(mode + '_acc', acc)\n",
" return loss\n",
"\n",
" def training_step(self, batch, batch_idx):\n",
" return self._calculate_loss(batch, mode='train')\n",
"\n",
" def validation_step(self, batch, batch_idx):\n",
" self._calculate_loss(batch, mode='val')\n",
"\n",
" def test_step(self, batch, batch_idx):\n",
" self._calculate_loss(batch, mode='test')"
]
},
{
"cell_type": "markdown",
"id": "ab00081e",
"metadata": {},
"source": [
"Le ResNet ainsi entraîné sur-apprend facilement, puisque son nombre de paramètres est plus de 1000 fois supérieur à la taille du jeu de données. Pour rendre la comparaison avec les modèles contrastifs équitable, nous appliquons des augmentations de données similaires à celles utilisées précédemment : retournement horizontal, recadrage-redimensionnement, niveaux de gris, flou gaussien. Nous n'utilisons cependant pas la distorsion des couleurs : la distribution des couleurs s'est révélée être une caractéristique importante pour la classification, et son ajout n'apportait aucun gain de performance notable. De même, nous limitons le redimensionnement avant recadrage à 125% de la résolution d'origine (contre 1250% pour SimCLR) : pour la classification, le modèle doit en effet reconnaître l'objet dans son ensemble, alors qu'en apprentissage contrastif, on cherche seulement à vérifier que deux patchs appartiennent à la même image."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "e5fc4b31",
"metadata": {},
"outputs": [],
"source": [
"train_transforms = transforms.Compose([transforms.RandomHorizontalFlip(),\n",
" transforms.RandomResizedCrop(size=96, scale=(0.8, 1.0)),\n",
" transforms.RandomGrayscale(p=0.2),\n",
" transforms.GaussianBlur(kernel_size=9, sigma=(0.1, 0.5)),\n",
" transforms.ToTensor(),\n",
" transforms.Normalize((0.5,), (0.5,))\n",
" ])\n",
"\n",
"train_img_aug_data = STL10(root=DATASET_PATH, split='train', download=True,\n",
" transform=train_transforms)"
]
},
{
"cell_type": "markdown",
"id": "07073547",
"metadata": {},
"source": [
"La fonction d'entraînement du ResNet est presque identique à celle de la régression logistique. Nous validons ici toutes les 2 époques, pour surveiller un éventuel sur-apprentissage rapide dès les premières itérations."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "a67bccd4",
"metadata": {},
"outputs": [],
"source": [
"def train_resnet(batch_size, max_epochs=100, **kwargs):\n",
" trainer = pl.Trainer(default_root_dir=os.path.join(CHECKPOINT_PATH, \"ResNet\"),\n",
" accelerator=\"gpu\" if str(device).startswith(\"cuda\") else \"cpu\",\n",
" devices=1,\n",
" max_epochs=max_epochs,\n",
" callbacks=[ModelCheckpoint(save_weights_only=True, mode=\"max\", monitor=\"val_acc\"),\n",
" LearningRateMonitor(\"epoch\")],\n",
" check_val_every_n_epoch=2)\n",
" trainer.logger._default_hp_metric = None\n",
"\n",
" # Chargeurs de données\n",
" train_loader = data.DataLoader(train_img_aug_data, batch_size=batch_size, shuffle=True,\n",
" drop_last=True, pin_memory=True, num_workers=NUM_WORKERS)\n",
" test_loader = data.DataLoader(test_img_data, batch_size=batch_size, shuffle=False,\n",
" drop_last=False, pin_memory=True, num_workers=NUM_WORKERS)\n",
"\n",
" # Vérifie si un modèle pré-entraîné existe déjà. Si oui, le charger et sauter l'entraînement\n",
" pretrained_filename = os.path.join(CHECKPOINT_PATH, \"ResNet.ckpt\")\n",
" if os.path.isfile(pretrained_filename):\n",
" print(\"Modèle pré-entraîné trouvé à %s, chargement...\" % pretrained_filename)\n",
" model = ResNet.load_from_checkpoint(pretrained_filename)\n",
" else:\n",
" pl.seed_everything(42) # Pour la reproductibilité\n",
" model = ResNet(**kwargs)\n",
" trainer.fit(model, train_loader, test_loader)\n",
" model = ResNet.load_from_checkpoint(trainer.checkpoint_callback.best_model_path)\n",
"\n",
" # Évalue le meilleur modèle sur le jeu de validation\n",
" train_result = trainer.test(model, train_loader, verbose=False)\n",
" val_result = trainer.test(model, test_loader, verbose=False)\n",
" result = {\"train\": train_result[0][\"test_acc\"], \"test\": val_result[0][\"test_acc\"]}\n",
"\n",
" return model, result"
]
},
{
"cell_type": "markdown",
"id": "15cd2fdd",
"metadata": {},
"source": [
"Entraînons enfin ce modèle et observons ses résultats :"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "a414311e",
"metadata": {},
"outputs": [],
"source": [
"resnet_model, resnet_result = train_resnet(batch_size=64,\n",
" num_classes=10,\n",
" lr=1e-3,\n",
" weight_decay=2e-4,\n",
" max_epochs=100)\n",
"print(f\"Accuracy sur le jeu d'entraînement : {100*resnet_result['train']:4.2f}%\")\n",
"print(f\"Accuracy sur le jeu de test : {100*resnet_result['test']:4.2f}%\")"
]
},
{
"cell_type": "markdown",
"id": "5b2fc3a5",
"metadata": {},
"source": [
"\n",
"\n",
"**Question**\n",
"\n",
"Comparer l'accuracy de test obtenue par ce ResNet entraîné *from scratch* à celle de SimCLR (régression logistique sur les caractéristiques, avec 500 images par classe). Que peut-on en conclure sur l'intérêt de l'apprentissage contrastif lorsque peu de données étiquetées sont disponibles ?\n",
"\n",
"
\n",
"\n",
"\n",
"\n",
"**Correction**\n",
"\n",
"Le ResNet entraîné *from scratch* atteint 73,31% sur le jeu de test — près de 8 points de moins que le modèle contrastif (81%), et même légèrement moins que SimCLR utilisant seulement un dixième des données étiquetées (200 images par classe). Cela montre que l'apprentissage contrastif auto-supervisé apporte un gain de performance considérable en exploitant de grandes quantités de données non-étiquetées, particulièrement lorsque peu de données étiquetées sont disponibles pour la tâche finale.\n",
"\n",
"
"
]
},
{
"cell_type": "markdown",
"id": "ee633bd5",
"metadata": {},
"source": [
"## Conclusion\n",
"\n",
"Dans ce TP, nous avons étudié l'apprentissage contrastif auto-supervisé et implémenté SimCLR à titre d'exemple. Appliqué au jeu de données STL10, il a permis d'apprendre des représentations généralisables, utilisables pour entraîner de simples modèles de classification. Avec 500 images par classe, il a obtenu une accuracy supérieure de 8 points à un modèle entraîné uniquement en supervision, et des performances comparables en n'utilisant qu'un dixième des données étiquetées. Ces résultats expérimentaux se limitent à un seul jeu de données, mais des travaux comme ceux de [Ting Chen et al.](https://arxiv.org/abs/2006.10029) montrent des tendances similaires sur des jeux plus grands comme ImageNet. Au-delà des hyperparamètres discutés ici, la taille du modèle semble également jouer un rôle important en apprentissage contrastif : avec suffisamment de données non-étiquetées, des modèles plus grands permettent d'obtenir des résultats nettement meilleurs, proches de leurs équivalents supervisés. Il existe par ailleurs des approches combinant apprentissage contrastif et supervisé, avec des gains de performance au-delà de la seule supervision (voir [Khosla et al.](https://arxiv.org/abs/2004.11362)). Enfin, l'apprentissage contrastif n'est qu'une des approches d'apprentissage auto-supervisé apparues récemment : d'autres méthodes reposent sur la distillation, comme BYOL (voir le :ref:`chapitre précédent `), ou sur la réduction de redondance, comme Barlow Twins.\n",
"\n",
"### Références\n",
"\n",
"[1] Chen, T., Kornblith, S., Norouzi, M., and Hinton, G. (2020). A simple framework for contrastive learning of visual representations. ICML 2020. ([lien](https://arxiv.org/abs/2002.05709))\n",
"\n",
"[2] Chen, T., Kornblith, S., Swersky, K., Norouzi, M., and Hinton, G. (2020). Big self-supervised models are strong semi-supervised learners. NeurIPS 2021. ([lien](https://arxiv.org/abs/2006.10029))\n",
"\n",
"[3] Oord, A. V. D., Li, Y., and Vinyals, O. (2018). Representation learning with contrastive predictive coding. arXiv:1807.03748. ([lien](https://arxiv.org/abs/1807.03748))\n",
"\n",
"[4] Grill, J.B. et al. (2020). Bootstrap your own latent: A new approach to self-supervised learning. NeurIPS 2020. ([lien](https://arxiv.org/abs/2006.07733))\n",
"\n",
"[5] Khosla, P. et al. (2020). Supervised contrastive learning. NeurIPS 2020. ([lien](https://arxiv.org/abs/2004.11362))\n",
"\n",
"[6] Zbontar, J., Jing, L., Misra, I., LeCun, Y. and Deny, S. (2021). Barlow twins: Self-supervised learning via redundancy reduction. ICML 2021. ([lien](https://arxiv.org/abs/2103.03230))"
]
},
{
"cell_type": "markdown",
"id": "834103be",
"metadata": {},
"source": [
"### Crédits\n",
"\n",
"Ce notebook est adapté des **UvA Deep Learning Notebooks**, tutoriel SimCLR par **Phillip Lippe**.\n",
"\n",
"Source originale :\n",
"https://uvadlc-notebooks.readthedocs.io/en/latest/tutorial_notebooks/tutorial17/SimCLR.html\n",
"\n",
"Distribué sous licence MIT."
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
}
},
"nbformat": 4,
"nbformat_minor": 5
}