3 P1 실습 — MNIST 첫 신경망 프로젝트

import random
import numpy as np
import matplotlib.pyplot as plt

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers

import torch
from torch import nn
from torch.utils.data import TensorDataset, DataLoader

SEED = 42

random.seed(SEED)
np.random.seed(SEED)
tf.random.set_seed(SEED)
torch.manual_seed(SEED)

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print("PyTorch device:", device)

from tensorflow.keras.datasets import mnist

(train_images_raw, train_labels), (test_images_raw, test_labels) = mnist.load_data()

print("train_images:", train_images_raw.shape, train_images_raw.dtype)
print("train_labels:", train_labels.shape, train_labels.dtype)
print("test_images :", test_images_raw.shape, test_images_raw.dtype)

print("pixel range:", train_images_raw.min(), "~", train_images_raw.max())
print("labels:", np.unique(train_labels))

fig, axes = plt.subplots(2, 5, figsize=(10, 4))

indices = np.random.choice(len(train_images_raw), size=10, replace=False)

for ax, idx in zip(axes.ravel(), indices):
    ax.imshow(train_images_raw[idx], cmap="gray")
    ax.set_title(f"label={train_labels[idx]}")
    ax.axis("off")

plt.tight_layout()
plt.show()

train_images = train_images_raw.reshape((60000, 28 * 28)).astype("float32") / 255
test_images = test_images_raw.reshape((10000, 28 * 28)).astype("float32") / 255

print(train_images.shape)
print(test_images.shape)
print(train_images.min(), train_images.max())

keras_model = keras.Sequential([
    layers.Dense(512, activation="relu"),
    layers.Dense(10, activation="softmax")
])

keras_model.compile(
    optimizer="rmsprop",
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"]
)

keras_model.summary()

history = keras_model.fit(
    train_images,
    train_labels,
    epochs=5,
    batch_size=128,
    validation_split=0.1,
    verbose=1
)

history_dict = history.history

epochs = range(1, len(history_dict["loss"]) + 1)

plt.figure(figsize=(7, 4))
plt.plot(epochs, history_dict["loss"], marker="o", label="train loss")
plt.plot(epochs, history_dict["val_loss"], marker="o", label="validation loss")
plt.xlabel("epoch")
plt.ylabel("loss")
plt.legend()
plt.show()

plt.figure(figsize=(7, 4))
plt.plot(epochs, history_dict["accuracy"], marker="o", label="train accuracy")
plt.plot(epochs, history_dict["val_accuracy"], marker="o", label="validation accuracy")
plt.xlabel("epoch")
plt.ylabel("accuracy")
plt.legend()
plt.show()

x_train_t = torch.tensor(train_images, dtype=torch.float32)
y_train_t = torch.tensor(train_labels, dtype=torch.long)

x_test_t = torch.tensor(test_images, dtype=torch.float32)
y_test_t = torch.tensor(test_labels, dtype=torch.long)

# Keras의 validation_split=0.1과 비슷하게 뒤 6,000개를 검증셋으로 사용
x_val_t = x_train_t[-6000:]
y_val_t = y_train_t[-6000:]
x_train_sub_t = x_train_t[:-6000]
y_train_sub_t = y_train_t[:-6000]

train_ds = TensorDataset(x_train_sub_t, y_train_sub_t)
val_ds = TensorDataset(x_val_t, y_val_t)
test_ds = TensorDataset(x_test_t, y_test_t)

train_loader = DataLoader(train_ds, batch_size=128, shuffle=True)
val_loader = DataLoader(val_ds, batch_size=256, shuffle=False)
test_loader = DataLoader(test_ds, batch_size=256, shuffle=False)

xb, yb = next(iter(train_loader))
print("input batch :", xb.shape)
print("target batch:", yb.shape)

torch_model = nn.Sequential(
    nn.Linear(28 * 28, 512),
    nn.ReLU(),
    nn.Linear(512, 10)
).to(device)

loss_fn = nn.CrossEntropyLoss()
optimizer = torch.optim.RMSprop(torch_model.parameters())

print(torch_model)

def evaluate_torch(model, loader, loss_fn, device):
    model.eval()
    total_loss = 0.0
    total_correct = 0
    total_count = 0

    with torch.no_grad():
        for x_batch, y_batch in loader:
            x_batch = x_batch.to(device)
            y_batch = y_batch.to(device)

            logits = model(x_batch)
            loss = loss_fn(logits, y_batch)

            total_loss += loss.item() * len(x_batch)
            total_correct += (logits.argmax(dim=1) == y_batch).sum().item()
            total_count += len(x_batch)

    return total_loss / total_count, total_correct / total_count


torch_history = {
    "loss": [],
    "accuracy": [],
    "val_loss": [],
    "val_accuracy": []
}

for epoch in range(5):
    torch_model.train()

    running_loss = 0.0
    running_correct = 0
    running_count = 0

    for x_batch, y_batch in train_loader:
        x_batch = x_batch.to(device)
        y_batch = y_batch.to(device)

        # 1. 이전 그레이디언트 초기화
        optimizer.zero_grad()

        # 2. 순전파
        logits = torch_model(x_batch)

        # 3. 손실 계산
        loss = loss_fn(logits, y_batch)

        # 4. 역전파: 그레이디언트 계산
        loss.backward()

        # 5. 파라미터 갱신
        optimizer.step()

        running_loss += loss.item() * len(x_batch)
        running_correct += (logits.argmax(dim=1) == y_batch).sum().item()
        running_count += len(x_batch)

    train_loss = running_loss / running_count
    train_acc = running_correct / running_count
    val_loss, val_acc = evaluate_torch(torch_model, val_loader, loss_fn, device)

    torch_history["loss"].append(train_loss)
    torch_history["accuracy"].append(train_acc)
    torch_history["val_loss"].append(val_loss)
    torch_history["val_accuracy"].append(val_acc)

    print(
        f"epoch {epoch+1}: "
        f"loss={train_loss:.4f}, acc={train_acc:.4f}, "
        f"val_loss={val_loss:.4f}, val_acc={val_acc:.4f}"
    )

epochs = range(1, len(torch_history["loss"]) + 1)

plt.figure(figsize=(7, 4))
plt.plot(epochs, torch_history["loss"], marker="o", label="train loss")
plt.plot(epochs, torch_history["val_loss"], marker="o", label="validation loss")
plt.xlabel("epoch")
plt.ylabel("loss")
plt.legend()
plt.show()

plt.figure(figsize=(7, 4))
plt.plot(epochs, torch_history["accuracy"], marker="o", label="train accuracy")
plt.plot(epochs, torch_history["val_accuracy"], marker="o", label="validation accuracy")
plt.xlabel("epoch")
plt.ylabel("accuracy")
plt.legend()
plt.show()

keras_test_loss, keras_test_acc = keras_model.evaluate(
    test_images, test_labels, verbose=0
)

print("Keras test loss    :", keras_test_loss)
print("Keras test accuracy:", keras_test_acc)

torch_test_loss, torch_test_acc = evaluate_torch(
    torch_model, test_loader, loss_fn, device
)

print("PyTorch test loss    :", torch_test_loss)
print("PyTorch test accuracy:", torch_test_acc)

keras_probs = keras_model.predict(test_images, verbose=0)
keras_pred = keras_probs.argmax(axis=1)

wrong_idx = np.where(keras_pred != test_labels)[0]

print("오분류 수:", len(wrong_idx))
print("전체 테스트 이미지 수:", len(test_labels))

n_show = min(12, len(wrong_idx))
chosen = wrong_idx[:n_show]

fig, axes = plt.subplots(3, 4, figsize=(9, 7))

for ax, idx in zip(axes.ravel(), chosen):
    ax.imshow(test_images_raw[idx], cmap="gray")
    ax.set_title(
        f"true={test_labels[idx]}, pred={keras_pred[idx]}"
    )
    ax.axis("off")

plt.tight_layout()
plt.show()

# TODO:
# 1. 실제 라벨과 예측 라벨을 이용한다.
# 2. 같은 조합이 몇 번 나타나는지 센다.
# 3. 가장 자주 발생한 오분류 조합을 찾아본다.

# 여기에 코드를 작성하세요.

# TODO: units 값을 바꾸어 실험한다.
units = 128

experiment_model = keras.Sequential([
    layers.Dense(units, activation="relu"),
    layers.Dense(10, activation="softmax")
])

experiment_model.compile(
    optimizer="rmsprop",
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"]
)

experiment_model.summary()

experiment_history = experiment_model.fit(
    train_images,
    train_labels,
    epochs=5,
    batch_size=128,
    validation_split=0.1,
    verbose=1
)