Post

Pytorch Representation Learning: Part Two

Posted Updated
By
15 min read
Pytorch Representation Learning: Part Two

Project && Guide

Table of Contents

Section 4. Image Classification with Convnets and ResNets

4.1. Librraies Classifying MNIST & CIFAR-10 with Convnets & ResNets

This section demonstrates how to classify MNIST and CIFAR-10 datasets using convolutional neural networks (ConvNets) and Residual Networks (ResNets) in PyTorch.

MNIST (link) is a dataset of 70,000 handwritten digit images (0–9), each 28×28 pixels, widely used for training and testing image classification models.

CIFAR-10 (link) is a dataset of 60,000 color images (32×32 pixels) in 10 classes (airplane, car, bird, cat, deer, dog, frog, horse, ship, truck), used for benchmarking image classification algorithms.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# Import time and numpy for timing and numerical operations
import time
import numpy as np
from __future__ import print_function

# Import PyTorch core modules and neural network utilities
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.init as init
import torch.nn.functional as F

# Import torchvision for datasets and image transformations
import torchvision
import torchvision.transforms

# Import matplotlib for plotting images and results
import matplotlib.pyplot as plt

4.2. Define image transformations & Initialize datasets

Image transformations are used to preprocess and augment the data. Datasets are loaded with these transformations.

1
2
3
4
5

# Define transformation to convert images to tensors
mnist_transforms = torchvision.transforms.Compose([torchvision.transforms.ToTensor()])
# Load MNIST training and test datasets with the defined transform
mnist_train = torchvision.datasets.MNIST(root='./data', train=True, transform=mnist_transforms, download=True)
mnist_test = torchvision.datasets.MNIST(root='./data', train=False, transform=mnist_transforms, download=True)

4.3. Create multi-threaded DataLoaders

DataLoaders allow efficient batch processing and shuffling of the dataset.

1
2
3

# Create DataLoaders for efficient batch processing and shuffling
train_loader = torch.utils.data.DataLoader(mnist_train, batch_size=64, shuffle=True, num_workers=2)
test_loader = torch.utils.data.DataLoader(mnist_test, batch_size=64, shuffle=True, num_workers=2)

4.4. Main classifier that subclasses nn.Module

This section defines a convolutional neural network classifier for MNIST, using multiple convolutional layers, dropout, activation, and pooling.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

# Define a convolutional neural network classifier for MNIST
class Classifier(nn.Module):
    """Convnet Classifier"""
    def __init__(self):
        super(Classifier, self).__init__()
        # Sequential block of convolution, dropout, activation, and pooling layers
        self.conv = nn.Sequential(
            # Layer 1: Conv2d -> Dropout -> ReLU -> MaxPool
            nn.Conv2d(in_channels=1, out_channels=16, kernel_size=(3, 3), padding=1),
            nn.Dropout(p=0.5),
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=(2, 2), stride=2),
            # Layer 2: Conv2d -> Dropout -> ReLU -> MaxPool
            nn.Conv2d(in_channels=16, out_channels=32, kernel_size=(3, 3), padding=1),
            nn.Dropout(p=0.5),
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=(2, 2), stride=2),
            # Layer 3: Conv2d -> Dropout -> ReLU -> MaxPool
            nn.Conv2d(in_channels=32, out_channels=64, kernel_size=(3, 3), padding=1),
            nn.Dropout(p=0.5),
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=(2, 2), stride=2),
            # Layer 4: Conv2d -> Dropout -> ReLU -> MaxPool
            nn.Conv2d(in_channels=64, out_channels=128, kernel_size=(3, 3), padding=1),
            nn.Dropout(p=0.5),
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=(2, 2), stride=2)
        )
        # Final linear layer for classification
        self.clf = nn.Linear(128, 10)
    def forward(self, x):
        # Pass input through conv layers and flatten for linear classifier
        return self.clf(self.conv(x).squeeze())

1
2
3
4
5
6
7
8
9
10
11

# Check if CUDA (GPU) is available for faster training
cuda_available = torch.cuda.is_available()
print(cuda_available)

# Instantiate classifier, move to GPU if available, and set up optimizer and loss function
clf = Classifier()
if cuda_available:
    clf = clf.cuda()
optimizer = torch.optim.Adam(clf.parameters(), lr=1e-4)
criterion = nn.CrossEntropyLoss()
# CrossEntropyLoss combines LogSoftmax and NLLLoss for classification

4.5. CIFAR10

This section sets up data augmentation, normalization, and loading for the CIFAR-10 dataset.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# Define data augmentation and normalization for CIFAR-10 training and test sets
cifar_train_transform = torchvision.transforms.Compose([
    torchvision.transforms.RandomCrop(32, padding=4), # Randomly crop images with padding
    torchvision.transforms.RandomHorizontalFlip(),   # Randomly flip images horizontally
    torchvision.transforms.ToTensor(),               # Convert images to tensor
    torchvision.transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2023, 0.1994, 0.2010)), # Normalize
])

cifar_test_transform = torchvision.transforms.Compose([
    torchvision.transforms.ToTensor(),
    torchvision.transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2023, 0.1994, 0.2010)),
])

# Load CIFAR-10 datasets with transforms
trainset = torchvision.datasets.CIFAR10(root='./data', train=True, download=True, transform=cifar_train_transform)
trainloader = torch.utils.data.DataLoader(trainset, batch_size=128, shuffle=True, num_workers=2)
testset = torchvision.datasets.CIFAR10(root='./data', train=False, download=True, transform=cifar_test_transform)
testloader = torch.utils.data.DataLoader(testset, batch_size=128, shuffle=False, num_workers=2)

4.6. Create a single Residual Block

This section defines the Residual Block used in ResNet architectures, allowing shortcut connections for improved gradient flow.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# Define the ResNet block used in the network
class ResidualBlock(nn.Module):
    def __init__(self, in_channels, out_channels, stride=1):
        super(ResidualBlock, self).__init__()
        # First convolutional layer
        self.conv1 = nn.Conv2d(in_channels, out_channels, kernel_size=3, stride=stride, padding=1, bias=False)
        self.bn1 = nn.BatchNorm2d(out_channels)
        # Second convolutional layer
        self.conv2 = nn.Conv2d(out_channels, out_channels, kernel_size=3, stride=1, padding=1, bias=False)
        self.bn2 = nn.BatchNorm2d(out_channels)
        # Shortcut connection to match dimensions if needed
        self.shortcut = nn.Sequential()
        if stride != 1 or in_channels != out_channels:
            self.shortcut = nn.Sequential(
                nn.Conv2d(in_channels, out_channels, kernel_size=1, stride=stride, bias=False),
                nn.BatchNorm2d(out_channels)
            )
    def forward(self, x):
        out = F.relu(self.bn1(self.conv1(x)))
        out = self.bn2(self.conv2(out))
        out += self.shortcut(x) # Add shortcut connection
        out = F.relu(out)
        return out

4.7. ResNet Architecture and Training Setup

This section defines the ResNet architecture for CIFAR-10 classification and sets up the optimizer and loss function.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

# Define the ResNet architecture for CIFAR-10 classification
class ResNet(nn.Module):
    def __init__(self, block, num_blocks, num_classes=10):
        super(ResNet, self).__init__()
        self.in_channels = 16
        # Initial convolutional layer
        self.conv1 = nn.Conv2d(3, 16, kernel_size=3, stride=1, padding=1, bias=False)
        self.bn1 = nn.BatchNorm2d(16)
        # Residual blocks for each stage
        self.layer1 = self._make_layer(block, 16, num_blocks[0], stride=1)
        self.layer2 = self._make_layer(block, 32, num_blocks[1], stride=2)
        self.layer3 = self._make_layer(block, 64, num_blocks[2], stride=2)
        # Fully connected layer for classification
        self.linear = nn.Linear(64, num_classes)
    def _make_layer(self, block, out_channels, num_blocks, stride):
        strides = [stride] + [1]*(num_blocks-1)
        layers = []
        for stride in strides:
            layers.append(block(self.in_channels, out_channels, stride))
            self.in_channels = out_channels
        return nn.Sequential(*layers)
    def forward(self, x):
        out = F.relu(self.bn1(self.conv1(x)))
        out = self.layer1(out)
        out = self.layer2(out)
        out = self.layer3(out)
        out = F.avg_pool2d(out, 8)
        out = out.view(out.size(0), -1)
        out = self.linear(out)
        return out

# Instantiate the ResNet model with 3 blocks per stage
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = ResNet(ResidualBlock, [3, 3, 3]).to(device)

# Set up loss function and optimizer for training
criterion = nn.CrossEntropyLoss() # Cross-entropy loss for classification
optimizer = torch.optim.Adam(model.parameters(), lr=0.001) # Adam optimizer for model parameters
scheduler = torch.optim.lr_scheduler.StepLR(optimizer, step_size=20, gamma=0.5) # Learning rate scheduler

Section 5. Neural Machine Translation (Seq2Seq)

5.1. Important Libraries

These libraries are essential for building and training the NMT model, handling data, and performing numerical operations.

1
2
3
4
5
6
7
8
9
10
11
12

import time  # For measuring training duration
import numpy as np  # For numerical operations
from __future__ import print_function  # Compatibility between Python 2 and 3
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.init as init
import torch.nn.functional as F
from torch.autograd import Variable  # For automatic differentiation
from torch.nn.utils.rnn import pack_padded_sequence  # For variable-length sequences
import codecs  # For reading files with encoding
import nltk  # For natural language processing tasks

5.2. Read training, validation & test data

Parallel data files are read for training, validation, and testing. Each line contains a source and target sentence.

1
2
3

train_lines = [line.strip().split('\t') for line in codecs.open('data/jpn-train.txt', 'r', encoding='utf-8')] # Each line is [target, source]
dev_lines = [line.strip().split('\t') for line in codecs.open('data/jpn-dev.txt', 'r', encoding='utf-8')] # Dev set
test_lines = [line.strip().split('\t') for line in codecs.open('data/jpn-test.txt', 'r', encoding='utf-8')] # Test set

5.3. Compute source and target vocabularies

Vocabularies are built for both source (Japanese) and target (English) languages, including special tokens for start, end, unknown, and padding.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

src_vocab = set() # Japanese vocabulary
trg_vocab = set() # English vocabulary
for line in train_lines:
    for word in line[1]: # Japanese characters
        if word not in src_vocab:
            src_vocab.add(word)
    for word in line[0].split(): # English words
        if word not in trg_vocab:
            trg_vocab.add(word)

# Add special tokens
src_vocab.update(['<s>', '</s>', '<unk>', '<pad>'])
trg_vocab.update(['<s>', '</s>', '<unk>', '<pad>'])

src_word2id = {word: idx for idx, word in enumerate(src_vocab)}
src_id2word = {idx: word for idx, word in enumerate(src_vocab)}
trg_word2id = {word: idx for idx, word in enumerate(trg_vocab)}
trg_id2word = {idx: word for idx, word in enumerate(trg_vocab)}

print('Number of unique Japanese words : %d ' % (len(src_vocab)))
print('Number of unique English words : %d ' % (len(trg_vocab)))

5.4. Create Seq2Seq model with GRUs

The Seq2Seq model uses GRUs for both the encoder and decoder, with embedding layers for source and target languages.

5.4.1. Define the Seq2Seq model using GRUs for NMT

Steps:

  1. Create embedding layers for source and target vocabularies.
  2. Build encoder and decoder GRU layers for sequence modeling.
  3. Add a linear layer to project decoder outputs to vocabulary space.
  4. Implement forward and decode methods for training and inference.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

class Seq2Seq(nn.Module):
    """A Vanilla Sequence to Sequence (Seq2Seq) model with LSTMs."""
    def __init__(self, src_emb_dim, trg_emb_dim, src_vocab_size, trg_vocab_size, src_hidden_dim, trg_hidden_dim, pad_token_src, pad_token_trg, bidirectional=False, nlayers_src=1, nlayers_trg=1):
        super(Seq2Seq, self).__init__()
        self.src_embedding = nn.Embedding(src_vocab_size, src_emb_dim, pad_token_src)
        self.trg_embedding = nn.Embedding(trg_vocab_size, trg_emb_dim, pad_token_trg)
        self.encoder = nn.GRU(src_emb_dim // 2 if bidirectional else src_emb_dim, src_hidden_dim, nlayers_src, bidirectional=bidirectional, batch_first=True)
        self.decoder = nn.GRU(trg_emb_dim, trg_hidden_dim, nlayers_trg, batch_first=True)
        self.decoder2vocab = nn.Linear(trg_hidden_dim, trg_vocab_size)

    def forward(self, input_src, input_trg, src_lengths):
        src_emb = self.src_embedding(input_src)
        trg_emb = self.trg_embedding(input_trg)
        src_emb = pack_padded_sequence(src_emb, src_lengths, batch_first=True)
        _, src_h_t = self.encoder(src_emb)
        h_t = torch.cat((src_h_t[-1], src_h_t[-2]), 1) if self.bidirectional else src_h_t[-1]
        trg_h, _ = self.decoder(trg_emb, h_t.unsqueeze(0).expand(self.nlayers_trg, h_t.size(0), h_t.size(1)))
        trg_h_reshape = trg_h.contiguous().view(trg_h.size(0) * trg_h.size(1), trg_h.size(2))
        decoder2vocab = self.decoder2vocab(trg_h_reshape)
        decoder2vocab = decoder2vocab.view(trg_h.size(0), trg_h.size(1), decoder2vocab.size(1))
        return decoder2vocab

    def decode(self, decoder2vocab):
        decoder2vocab_reshape = decoder2vocab.view(-1, decoder2vocab.size(2))
        word_probs = F.softmax(decoder2vocab_reshape)
        word_probs = word_probs.view(decoder2vocab.size(0), decoder2vocab.size(1), decoder2vocab.size(2))
        return word_probs

5.4.2. Batch Preparation for Seq2Seq Training

This function prepares batches for training and evaluation. Steps:

  1. Extract and pad source and target sentences.
  2. Sort sentences by length for efficient masking.
  3. Convert sentences to indices and pad to max length.
  4. Create input and output tensors for teacher-forcing.
  5. Return a dictionary of tensors and lengths.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# Function to create parallel minibatches for training and evaluation
def get_parallel_minibatch(lines, src_word2id, trg_word2id, index, batch_size, volatile=False):
    # Get source sentences for this minibatch
    src_lines = [['<s>'] + list(line[1]) + ['</s>'] for line in lines[index: index + batch_size]]
    # Get target sentences for this minibatch
    trg_lines = [['<s>'] + line[0].split() + ['</s>'] for line in lines[index: index + batch_size]]
    # Sort source sentences by length
    src_lens = [len(line) for line in src_lines]
    sorted_indices = np.argsort(src_lens)[::-1]
    sorted_src_lines = [src_lines[idx] for idx in sorted_indices]
    sorted_trg_lines = [trg_lines[idx] for idx in sorted_indices]
    max_src_len = max([len(line) for line in sorted_src_lines])
    max_trg_len = max([len(line) for line in sorted_trg_lines])
    input_lines_src = [[src_word2id[w] if w in src_word2id else src_word2id['<unk>'] for w in line] + [src_word2id['<pad>']] * (max_src_len - len(line)) for line in sorted_src_lines]
    input_lines_trg = [[trg_word2id[w] if w in trg_word2id else trg_word2id['<unk>'] for w in line[:-1]] + [trg_word2id['<pad>']] * (max_trg_len - len(line)) for line in sorted_trg_lines]
    output_lines_trg = [[trg_word2id[w] if w in trg_word2id else trg_word2id['<unk>'] for w in line[1:]] + [trg_word2id['<pad>']] * (max_trg_len - len(line)) for line in sorted_trg_lines]
    input_lines_src = Variable(torch.LongTensor(input_lines_src), volatile=volatile)
    input_lines_trg = Variable(torch.LongTensor(input_lines_trg), volatile=volatile)
    output_lines_trg = Variable(torch.LongTensor(output_lines_trg), volatile=volatile)
    return {'input_src': input_lines_src, 'input_trg': input_lines_trg, 'output_trg': output_lines_trg, 'src_lens': [len(line) for line in sorted_src_lines]}

5.4.3. Instantiate the Seq2Seq model and move to GPU if available

The model is created and moved to GPU if CUDA is available for faster training.

1
2
3
4
5
6
7
8
9
10

cuda_available = torch.cuda.is_available()
seq2seq = Seq2Seq(
    src_emb_dim=128, trg_emb_dim=128,
    src_vocab_size=len(src_word2id), trg_vocab_size=len(trg_word2id),
    src_hidden_dim=512, trg_hidden_dim=512,
    pad_token_src=src_word2id['<pad>'],
    pad_token_trg=trg_word2id['<pad>'],
)
if cuda_available:
    seq2seq = seq2seq.cuda()

5.4.4. Set up optimizer, loss criterion, and batch size for training

Adam optimizer and cross-entropy loss are used, with padding ignored in the loss calculation. Steps:

  1. Initialize Adam optimizer for model parameters.
  2. Create a weight mask to ignore padding tokens in loss calculation.
  3. Set up cross-entropy loss with the mask.
  4. Define batch size for training.
1
2
3
4
5
6
7

optimizer = optim.Adam(seq2seq.parameters(), lr=4e-4) # Adam optimizer for model parameters
weight_mask = torch.ones(len(trg_word2id)) # Mask for loss function
if cuda_available:
    weight_mask = weight_mask.cuda()
weight_mask[trg_word2id['<pad>']] = 0 # Ignore padding in loss
loss_criterion = nn.CrossEntropyLoss(weight=weight_mask) # Cross-entropy loss
batch_size = 64 # Batch size for training

5.4.5. Training loop for Seq2Seq model with evaluation on dev and test sets

Steps:

  1. Loop over epochs and batches.
  2. Prepare minibatches and move to GPU if available.
  3. Forward pass, compute loss, backward pass, and optimizer step.
  4. Clip gradients to avoid exploding gradients.
  5. Evaluate on dev and test sets after each epoch.
  6. Print losses for monitoring.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

for epoch in range(15):
    losses = []
    for j in range(0, len(train_lines), batch_size):
        minibatch = get_parallel_minibatch(
            lines=train_lines, src_word2id=src_word2id,
            trg_word2id=trg_word2id, index=j, batch_size=batch_size
        )
        if cuda_available:
            minibatch['input_src'] = minibatch['input_src'].cuda()
            minibatch['input_trg'] = minibatch['input_trg'].cuda()
            minibatch['output_trg'] = minibatch['output_trg'].cuda()
        decoder_out = seq2seq(
            input_src=minibatch['input_src'], input_trg=minibatch['input_trg'], src_lengths=minibatch['src_lens']
        )
        loss = loss_criterion(
            decoder_out.contiguous().view(-1, decoder_out.size(2)),
            minibatch['output_trg'].contiguous().view(-1)
        )
        optimizer.zero_grad()
        loss.backward()
        torch.nn.utils.clip_grad_norm(seq2seq.parameters(), 5.)
        optimizer.step()
        losses.append(loss.data[0])
    # Evaluate on dev set
    dev_nll = []
    for j in range(0, len(dev_lines), batch_size):
        minibatch = get_parallel_minibatch(
            lines=dev_lines, src_word2id=src_word2id,
            trg_word2id=trg_word2id, index=j, batch_size=batch_size,
            volatile=True
        )
        if cuda_available:
            minibatch['input_src'] = minibatch['input_src'].cuda()
            minibatch['input_trg'] = minibatch['input_trg'].cuda()
            minibatch['output_trg'] = minibatch['output_trg'].cuda()
        decoder_out = seq2seq(
            input_src=minibatch['input_src'], input_trg=minibatch['input_trg'], src_lengths=minibatch['src_lens']
        )
        loss = loss_criterion(
            decoder_out.contiguous().view(-1, decoder_out.size(2)),
            minibatch['output_trg'].contiguous().view(-1)
        )
        dev_nll.append(loss.data[0])
    # Evaluate on test set
    test_nll = []
    for j in range(0, len(test_lines), batch_size):
        minibatch = get_parallel_minibatch(
            lines=test_lines, src_word2id=src_word2id,
            trg_word2id=trg_word2id, index=j, batch_size=batch_size,
            volatile=True
        )
        if cuda_available:
            minibatch['input_src'] = minibatch['input_src'].cuda()
            minibatch['input_trg'] = minibatch['input_trg'].cuda()
            minibatch['output_trg'] = minibatch['output_trg'].cuda()
        decoder_out = seq2seq(
            input_src=minibatch['input_src'], input_trg=minibatch['input_trg'], src_lengths=minibatch['src_lens']
        )
        loss = loss_criterion(
            decoder_out.contiguous().view(-1, decoder_out.size(2)),
            minibatch['output_trg'].contiguous().view(-1)
        )
        test_nll.append(loss.data[0])
    print('Epoch : %d Training Loss : %.3f' % (epoch, np.mean(losses)))
    print('Epoch : %d Dev Loss : %.3f' % (epoch, np.mean(dev_nll)))
    print('Epoch : %d Test Loss : %.3f' % (epoch, np.mean(test_nll)))
    print('-------------------------------------------------------------')

5.5. Evaluate Model Predictions on Dev Set

This section evaluates the trained model by translating a few sentences from the development set and comparing the outputs to the reference translations. Steps:

  1. Prepare a minibatch from the dev set.
  2. Move tensors to GPU if available.
  3. Run the minibatch through the model in teacher-forcing mode.
  4. Decode model outputs to predicted words.
  5. Compare predictions to gold references and print results.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

# Evaluate model predictions for a few sentences in the dev set
minibatch = get_parallel_minibatch(
    lines=dev_lines, src_word2id=src_word2id,
    trg_word2id=trg_word2id, index=0, batch_size=batch_size,
    volatile=True,
)
if cuda_available:
    minibatch['input_src'] = minibatch['input_src'].cuda()
    minibatch['input_trg'] = minibatch['input_trg'].cuda()
    minibatch['output_trg'] = minibatch['output_trg'].cuda()
res = seq2seq(
    input_src=minibatch['input_src'], input_trg=minibatch['input_trg'], src_lengths=minibatch['src_lens']
)
res = res.data.cpu().numpy().argmax(axis=-1)
gold = minibatch['output_trg'].data.cpu().numpy()
res = [[trg_id2word[x] for x in line] for line in res]
gold = [[trg_id2word[x] for x in line] for line in gold]
for r, g in zip(res, gold):
    if '</s>' in r:
        index = r.index('</s>')
    else:
        index = len(r)
    print('Prediction : %s ' % (' '.join(r[:index])))
    index = g.index('</s>')
    print('Gold : %s ' % (' '.join(g[:index])))
    print('---------------')

Resources

Project repository

GitHub Code: PyTorch Representation Learning Project

Machine Learning and Deep Learning, Deep Learning
PyTorch representation learning ConvNet ResNet image classification deep learning seq2seq neural machine translation
This post is licensed under CC BY 4.0 by the author.