Project: Image Classifier in PyTorch

কী বানাবে

আগের project এর একই digit পড়ার কাজ, এবার team গুলো আসলে যেভাবে করে সেভাবে: PyTorch এ একটা convolutional network, Dataset আর DataLoader দিয়ে data load, augmentation, train/validation/test ভাগ, learning-rate schedule, আর checkpoint যেটা সেরা model টা রেখে দেয়। Test set এ 98.8% পায় 20,490 টা parameter দিয়ে, হাতে লেখা network এর 101,770 এর পাঁচ ভাগের এক ভাগ।

Code ছোট কারণ backward pass PyTorch করে দেয়। Engineering টা model এর চারপাশের সবকিছুতে: loader তাকে কী খাওয়ায়, dropout কখন চালু, কোন checkpoint রাখবে, আর test set এ কখন তাকাবে।

যা আগে জানা লাগবে

• Convolutional Neural Networks: filter, pooling, feature map।
• Training Neural Networks: AdamW, schedule, init।
• Project: Build a Neural Network from Scratch: MNIST এর file আর mnist.py loader।
• PyTorch সহ Python (CPU ই যথেষ্ট; তিন epoch এ মোটামুটি এক মিনিট)।

Architecture

Play

File	কাজ
model.py	CNN: দুইটা conv block, dropout + linear head
data.py	normalisation আর shift augmentation সহ Digits dataset; train/val/test loader
train.py	loop: AdamW, one-cycle schedule, validation, checkpoint, শেষে test
test_cnn.py	shape, augmentation, eval-mode আর overfit এর test

তিন ভাগ, তিন কাজ: train (55,000) weight fit করতে, validation (5,000) checkpoint বাছতে, test (10,000) শেষে একবার জানাতে।

ধাপে ধাপে বানাও

Step 1: model। conv → ReLU → max-pool এর দুইটা block 28×28 কে 7×7 এ নামায়, আর channel 1 থেকে 32 এ নেয়। padding=1 প্রতিটা 3×3 conv এ সাইজ ধরে রাখে, তাই ছোট করে শুধু pooling। Dropout training এর সময় random ভাবে 25% feature শূন্য করে, একটা regulariser।

"""A small CNN for 28x28 digits: two conv blocks, then a classifier head."""
import torch.nn as nn

class DigitCNN(nn.Module):
    def __init__(self):
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv2d(1, 16, 3, padding=1), nn.ReLU(), nn.MaxPool2d(2),   # 28x28 -> 14x14, 16 maps
            nn.Conv2d(16, 32, 3, padding=1), nn.ReLU(), nn.MaxPool2d(2),  # 14x14 -> 7x7, 32 maps
        )
        self.head = nn.Sequential(nn.Flatten(), nn.Dropout(0.25), nn.Linear(32 * 7 * 7, 10))

    def forward(self, x):
        return self.head(self.features(x))

Step 2: data। একটা Dataset দুইটা প্রশ্নের answer দেয়, কয়টা item আর i নম্বর item কী; DataLoader সেগুলো batch করে আর এলোমেলো করে। Training set এর mean আর std দিয়ে normalise করো (0.1307, 0.3081)। Augmentation হয় __getitem__ এর ভেতরে, শুধু training set এ: প্রতিবার একটা digit তোলার সময় সেটা 2 pixel পর্যন্ত সরে যায়, তাই model কখনো হুবহু একই ছবি দুইবার দেখে না।

"""Datasets and loaders, with a random-shift augmentation for training only."""
import torch
from torch.utils.data import DataLoader, Dataset
from mnist import load

class Digits(Dataset):
    def __init__(self, X, y, augment=False, seed=0):
        self.X = torch.tensor(X, dtype=torch.float32).view(-1, 1, 28, 28)
        self.X = (self.X - 0.1307) / 0.3081                 # normalise with the training mean and std
        self.y = torch.tensor(y)
        self.augment = augment
        self.g = torch.Generator().manual_seed(seed)

    def __len__(self):
        return len(self.y)

    def __getitem__(self, i):
        x = self.X[i]
        if self.augment:                                   # shift up to 2 pixels in each direction
            dx, dy = torch.randint(-2, 3, (2,), generator=self.g).tolist()
            x = torch.roll(x, shifts=(dy, dx), dims=(1, 2))
        return x, self.y[i]

def loaders(batch=128):
    X, y, X_test, y_test = load()
    train = Digits(X[:55000], y[:55000], augment=True)
    val = Digits(X[55000:], y[55000:])                     # 5,000 images held out for choosing the checkpoint
    test = Digits(X_test, y_test)
    g = torch.Generator().manual_seed(0)
    return (DataLoader(train, batch_size=batch, shuffle=True, generator=g),
            DataLoader(val, batch_size=1000), DataLoader(test, batch_size=1000))

Step 3: loop। প্রতি batch এ: zero_grad, forward, loss, backward, step, scheduler step। প্রতি epoch এ: eval() mode আর no_grad() এর ভেতরে validation এ evaluate, আর validation ভালো হলে checkpoint save। শেষে সেরা checkpoint load করে একবার test এ evaluate।

import torch
import torch.nn.functional as F
from data import loaders
from model import DigitCNN

torch.manual_seed(0)
torch.set_num_threads(4)

def evaluate(model, loader):
    model.eval()                                           # turns dropout off
    correct = total = 0
    with torch.no_grad():                                  # no graph needed for evaluation
        for x, y in loader:
            correct += (model(x).argmax(1) == y).sum().item()
            total += len(y)
    return correct / total

train_dl, val_dl, test_dl = loaders()
model = DigitCNN()
print("parameters:", sum(p.numel() for p in model.parameters()))
opt = torch.optim.AdamW(model.parameters(), lr=2e-3, weight_decay=1e-4)
sched = torch.optim.lr_scheduler.OneCycleLR(opt, max_lr=2e-3, total_steps=3 * len(train_dl))

best = 0.0
for epoch in range(1, 4):
    model.train()
    total = 0.0
    for x, y in train_dl:
        opt.zero_grad()
        loss = F.cross_entropy(model(x), y)
        loss.backward()
        opt.step()
        sched.step()
        total += loss.item() * len(y)
    val_acc = evaluate(model, val_dl)
    saved = ""
    if val_acc > best:
        best = val_acc
        torch.save({"model": model.state_dict(), "epoch": epoch, "val_acc": val_acc}, "best.pt")
        saved = "  -> saved best.pt"
    print(f"epoch {epoch}  train loss {total / len(train_dl.dataset):.4f}  val accuracy {val_acc:.2%}{saved}")

ckpt = torch.load("best.pt", weights_only=True)  # never unpickle arbitrary objects
model.load_state_dict(ckpt["model"])
print(f"best checkpoint: epoch {ckpt['epoch']}, val {ckpt['val_acc']:.2%}")
print(f"test accuracy (touched once, at the end): {evaluate(model, test_dl):.2%}")

Output:

parameters: 20490
epoch 1  train loss 0.7897  val accuracy 97.86%  -> saved best.pt
epoch 2  train loss 0.1402  val accuracy 98.70%  -> saved best.pt
epoch 3  train loss 0.1008  val accuracy 98.94%  -> saved best.pt
best checkpoint: epoch 3, val 98.94%
test accuracy (touched once, at the end): 98.80%

কী দেখলাম:

• 20,490 টা parameter, dense network এর 101,770 এর জায়গায়, আর result ও ভালো। ভাগ করে নেওয়া filter ছবির সাথে মানায়।
• Validation ওঠে 97.86% → 98.70% → 98.94%, আর প্রতিটা epoch নতুন সেরা checkpoint save করে। Epoch 3 খারাপ হলে epoch 2 এর file টাই থেকে যেত: early stopping হিসেবে checkpoint এভাবেই কাজ করে।
• প্রথম epoch এর train loss বেশি (0.79) কারণ one-cycle schedule ছোট learning rate দিয়ে শুরু করে warm-up করে; তারপর নামে 0.10 এ।
• Test accuracy 98.80%, হাতে লেখা dense network এর 96.64% এর জায়গায়। 336 এর বদলে 120 টা ভুল digit: ভুলের প্রায় দুই তৃতীয়াংশ গায়েব।
• Test একবারই ছোঁয়া হয়েছে। প্রতিটা সিদ্ধান্ত (কোন epoch রাখবে) শুধু validation দিয়ে নেওয়া, তাই 98.80% একটা সৎ হিসাব।

Test করো

import torch
import torch.nn.functional as F
from data import Digits
from model import DigitCNN

torch.manual_seed(0)

def test_output_shape():
    assert DigitCNN()(torch.zeros(4, 1, 28, 28)).shape == (4, 10)

def test_augmentation_only_moves_pixels():
    x = torch.rand(3, 784).numpy()
    ds = Digits(x, [0, 1, 2], augment=True)
    plain = Digits(x, [0, 1, 2])
    for i in range(3):
        assert torch.isclose(ds[i][0].sum(), plain[i][0].sum(), atol=1e-4)   # a roll keeps every pixel

def test_eval_mode_is_deterministic():
    m = DigitCNN().eval()
    x = torch.rand(2, 1, 28, 28)
    with torch.no_grad():
        assert torch.equal(m(x), m(x))          # dropout must be off

def test_can_overfit_one_batch():
    m = DigitCNN()
    x, y = torch.rand(16, 1, 28, 28), torch.arange(16) % 10
    opt = torch.optim.Adam(m.parameters(), lr=3e-3)
    for _ in range(150):
        opt.zero_grad(); F.cross_entropy(m(x), y).backward(); opt.step()
    m.eval()
    assert (m(x).argmax(1) == y).all()

if __name__ == "__main__":
    for name, fn in list(globals().items()):
        if name.startswith("test_"):
            fn()
            print("pass", name)

pass test_output_shape
pass test_augmentation_only_moves_pixels
pass test_eval_mode_is_deterministic
pass test_can_overfit_one_batch

• Output shape (batch, 10): ভুল flatten size ধরে, ছবির সাইজ বদলালে বা layer যোগ করলে সবচেয়ে common error।
• Augmentation শুধু pixel সরায়: roll দিয়ে সরালে মোট উজ্জ্বলতা একই থাকে। ধরো ভুল augmentation ছবিটাই শূন্য করে দিচ্ছে, তবুও train হবে, শুধু খারাপভাবে, আর তুমি কখনো বুঝবে না কেন।
• Eval mode deterministic: eval() এ dropout বন্ধ, তাই একই input একই output দেয়। Validation বা serving এর আগে model.eval() ভুলে যাওয়া একটা ক্লাসিক bug: prediction random হয়ে যায়, accuracy পড়ে।
• এক batch এ overfit: model যদি 16 টা random ছবি মুখস্থ করতে না পারে, training loop ভাঙা। Model বদলালেই আগে এটা চালাও।

Trade-offs

Choice	ভালো দিক	খরচ
ছোট CNN (20k param)	CPU তে এক মিনিটে train, 98.8%	BatchNorm সহ গভীর net 99.5%+ পায়; compute বেশি লাগে
Shift augmentation	বিনা খরচে বৈচিত্র্য, ভালো generalisation	বাস্তবসম্মত থাকতে হবে: 6 উল্টালে 9 হয়, তাই digit এ flip না
AdamW + one-cycle	কম tune করেই দ্রুত, স্থির training	সাধারণ SGD এর চেয়ে বেশি knob; schedule এর দৈর্ঘ্য epoch সংখ্যার সাথে মিলতে হবে
Validation দেখে checkpoint	পরের epoch খারাপ হলেও সেরা model রাখে	training এর 5,000 টা ছবি কম; প্রতিটা checkpoint এ disk জায়গা
weights_only=True দিয়ে load	load করার সময় checkpoint কোনো code চালাতে পারে না	ভেতরে যেকোনো Python object রাখা যায় না; সাধারণ tensor আর সংখ্যা রাখো

Production এর জন্য weight এর পাশে preprocessing এর constant ও save করতে (mean, std, input size), serving এর জন্য model export করতে (TorchScript বা ONNX), আর run এর setting আর metric track করতে (Part 10), যাতে আবার হুবহু বানানো যায়।

আরো এগিয়ে নাও

1. প্রতিটা conv এর পরে nn.BatchNorm2d যোগ করে আবার train করো। Epoch 1 এর পরের validation accuracy কি বদলায়?
2. 10 epoch train করো আর প্রতি epoch এর validation accuracy আঁকো। সেরা checkpoint কখন বদলানো বন্ধ করে?
3. predict.py লেখো যেটা best.pt load করে, তোমার নিজের আঁকা 28×28 এর একটা PNG নেয়, একই normalisation দেয় আর probability সহ প্রথম 3 টা digit print করে।
4. MNIST এর বদলে Fashion-MNIST নাও (কাপড়চোপড়, একই file format)। Model কোন class গুলো গুলিয়ে ফেলে, আর shift augmentation কি এখনো কাজে দেয়?