隨著深度學習模型在行動裝置、嵌入式系統等邊緣裝置上的部署需求日益增加,如何在有限的運算資源與記憶體限制下,讓模型維持可接受的效能,成為重要的實務課題。知識蒸餾(Knowledge Distillation)與量化(Quantization)是兩種常見且可互補的模型壓縮技術:知識蒸餾透過訓練一個較大的 teacher 模型,並讓一個較小的 student 模型模仿其輸出分布,使 student 能以更少的參數逼近 teacher 的效能;量化則是在模型訓練過程中或訓練完成後,將原本以 32 位元浮點數表示的權重與運算,轉換為較少位元整數的表示方式,藉此縮小模型體積並加速推論。雖然兩者的切入點不同,但都是以犧牲些微準確率為代價,換取計算上的效率。本篇章將分別介紹知識蒸餾與量化的基本原理,並以 MNIST 分類任務為例,實際觀察壓縮前後在模型大小、推論速度與準確率上的變化。

以下是用 MNIST 資料集進行的模型蒸餾範例:

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import DataLoader, Subset
from torchvision import datasets, transforms

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

BATCH_SIZE = 128
EPOCHS_TEACHER = 5
EPOCHS_STUDENT = 5
LR = 1e-3
TEMPERATURE = 2
ALPHA = 0.9
N_TRAIN = 2000

transform = transforms.ToTensor()
train_set = datasets.MNIST(root='./data', train=True, download=True, transform=transform)
test_set = datasets.MNIST(root='./data', train=False, download=True, transform=transform)

subset_indices = torch.randperm(len(train_set))[:N_TRAIN]
train_subset = Subset(train_set, subset_indices)
train_loader = DataLoader(train_subset, batch_size=BATCH_SIZE, shuffle=True)
test_loader = DataLoader(test_set, batch_size=BATCH_SIZE, shuffle=False)


class TeacherCNN(nn.Module):
	def __init__(self):
		super().__init__()
		self.features = nn.Sequential(
			nn.Conv2d(1, 32, kernel_size=3, padding=1),
			nn.ReLU(),
			nn.MaxPool2d(2),
			nn.Conv2d(32, 64, kernel_size=3, padding=1),
			nn.ReLU(),
			nn.MaxPool2d(2),
		)
		self.classifier = nn.Sequential(
			nn.Flatten(),
			nn.Linear(64 * 7 * 7, 128),
			nn.ReLU(),
			nn.Linear(128, 10),
		)

	def forward(self, x):
		x = self.features(x)
		x = self.classifier(x)
		return x


class StudentMLP(nn.Module):
	def __init__(self):
		super().__init__()
		self.net = nn.Sequential(
			nn.Flatten(),
			nn.Linear(28 * 28, 128),
			nn.ReLU(),
			nn.Linear(128, 10),
		)

	def forward(self, x):
		return self.net(x)


def evaluate(model, loader):
	model.eval()
	correct = 0
	total = 0
	with torch.no_grad():
		for x, y in loader:
			x, y = x.to(device), y.to(device)
			pred = model(x).argmax(dim=1)
			correct += (pred == y).sum().item()
			total += y.size(0)
	return 100 * correct / total


def train_standard(model, loader, epochs):
	optimizer = torch.optim.Adam(model.parameters(), lr=LR)
	for epoch in range(epochs):
		print(f'Standard epoch {epoch+1}/{epochs}')
		model.train()
		for x, y in loader:
			x, y = x.to(device), y.to(device)
			optimizer.zero_grad()
			loss = F.cross_entropy(model(x), y)
			loss.backward()
			optimizer.step()


def train_student_distill(student, teacher, loader, epochs):
	optimizer = torch.optim.Adam(student.parameters(), lr=LR)
	teacher.eval()
	for epoch in range(epochs):
		print(f'Distill epoch {epoch+1}/{epochs}')
		student.train()
		for x, y in loader:
			x, y = x.to(device), y.to(device)
			with torch.no_grad():
				teacher_logits = teacher(x)
			student_logits = student(x)

			hard_loss = F.cross_entropy(student_logits, y)
			soft_loss = F.kl_div(
				F.log_softmax(student_logits / TEMPERATURE, dim=1),
				F.softmax(teacher_logits / TEMPERATURE, dim=1),
				reduction='batchmean',
			)

			loss = ALPHA * hard_loss + (1 - ALPHA) * soft_loss
			optimizer.zero_grad()
			loss.backward()
			optimizer.step()


teacher = TeacherCNN().to(device)
train_standard(teacher, train_loader, EPOCHS_TEACHER)
teacher_acc = evaluate(teacher, test_loader)

student_baseline = StudentMLP().to(device)
train_standard(student_baseline, train_loader, EPOCHS_STUDENT)
baseline_acc = evaluate(student_baseline, test_loader)

student_distill = StudentMLP().to(device)
train_student_distill(student_distill, teacher, train_loader, EPOCHS_STUDENT)
distill_acc = evaluate(student_distill, test_loader)

n_teacher_params = sum(p.numel() for p in teacher.parameters())
n_stu_base_params = sum(p.numel() for p in student_baseline.parameters())
n_stu_dis_params = sum(p.numel() for p in student_distill.parameters())

print(f'Teacher accuracy: {teacher_acc:.2f}%, #params: {n_teacher_params}')
print(f'Student (baseline) accuracy: {baseline_acc:.2f}%, #params: {n_stu_base_params}')
print(f'Student (distillation) accuracy: {distill_acc:.2f}%, #params: {n_stu_dis_params}')

在上述範例中:

若要進行模型量化,則可以透過 torch.ao.quantization.quantize_dynamic,在模型訓練完成後,把參數從 32 位元浮點數轉換為 8 位元整數,範例如下:

import os
import time
import warnings

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.ao.quantization import quantize_dynamic
from torch.utils.data import DataLoader
from torchvision import datasets, transforms

warnings.filterwarnings('ignore', category=DeprecationWarning, message='.*torch.ao.quantization.*')

train_device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
infer_device = torch.device('cpu')

BATCH_SIZE = 128
EPOCHS = 5
LR = 1e-3

transform = transforms.ToTensor()
train_set = datasets.MNIST(root='./data', train=True, download=True, transform=transform)
test_set = datasets.MNIST(root='./data', train=False, download=True, transform=transform)
train_loader = DataLoader(train_set, batch_size=BATCH_SIZE, shuffle=True)
test_loader = DataLoader(test_set, batch_size=BATCH_SIZE, shuffle=False)


class ClassifierMLP(nn.Module):
	def __init__(self):
		super().__init__()
		self.net = nn.Sequential(
			nn.Flatten(),
			nn.Linear(28 * 28, 128),
			nn.ReLU(),
			nn.Linear(128, 10),
		)

	def forward(self, x):
		return self.net(x)


def train(model, loader, epochs, device):
	optimizer = torch.optim.Adam(model.parameters(), lr=LR)
	for epoch in range(epochs):
		print(f'Epoch {epoch + 1}/{epochs}')
		model.train()
		for x, y in loader:
			x, y = x.to(device), y.to(device)
			optimizer.zero_grad()
			loss = F.cross_entropy(model(x), y)
			loss.backward()
			optimizer.step()


def evaluate(model, loader, device):
	model.eval()
	correct = 0
	total = 0
	with torch.no_grad():
		for x, y in loader:
			x, y = x.to(device), y.to(device)
			pred = model(x).argmax(dim=1)
			correct += (pred == y).sum().item()
			total += y.size(0)
	return 100 * correct / total


def measure_inference_time(model, loader, device):
	model.eval()
	start = time.time()
	with torch.no_grad():
		for x, _ in loader:
			x = x.to(device)
			model(x)
	return time.time() - start


def get_model_size(model, path='temp_model.p'):
	torch.save(model.state_dict(), path)
	size_kb = os.path.getsize(path) / 1024
	os.remove(path)
	return size_kb


model_fp32 = ClassifierMLP().to(train_device)
train(model_fp32, train_loader, EPOCHS, train_device)

model_fp32 = model_fp32.to(infer_device)

model_int8 = quantize_dynamic(model_fp32, {nn.Linear}, dtype=torch.qint8)

fp32_acc = evaluate(model_fp32, test_loader, infer_device)
int8_acc = evaluate(model_int8, test_loader, infer_device)

# Backend Not supported
# model_int8_gpu = model_int8.to(train_device)
# int8_acc_gpu = evaluate(model_int8_gpu, test_loader, train_device)

fp32_size = get_model_size(model_fp32)
int8_size = get_model_size(model_int8)

fp32_time = measure_inference_time(model_fp32, test_loader, infer_device)
int8_time = measure_inference_time(model_int8, test_loader, infer_device)

print(f'FP32 accuracy: {fp32_acc:.2f}%, size: {fp32_size:.1f} KB, time: {fp32_time:.3f} s')
print(f'INT8 accuracy: {int8_acc:.2f}%, size: {int8_size:.1f} KB, time: {int8_time:.3f} s')

在上述範例中:

需要注意的是,quantize_dynamic 並無法處理 nn.Conv2d 等類型的網路層,範例如下:

import os
import warnings

import torch
import torch.nn as nn
from torch.ao.quantization import quantize_dynamic

warnings.filterwarnings('ignore', category=DeprecationWarning, message='.*torch.ao.quantization.*')


class TeacherCNN(nn.Module):
	def __init__(self):
		super().__init__()
		self.features = nn.Sequential(
			nn.Conv2d(1, 32, kernel_size=3, padding=1),
			nn.ReLU(),
			nn.MaxPool2d(2),
			nn.Conv2d(32, 64, kernel_size=3, padding=1),
			nn.ReLU(),
			nn.MaxPool2d(2),
		)
		self.classifier = nn.Sequential(
			nn.Flatten(),
			nn.Linear(64 * 7 * 7, 128),
			nn.ReLU(),
			nn.Linear(128, 10),
		)

	def forward(self, x):
		x = self.features(x)
		x = self.classifier(x)
		return x


def get_model_size(model, path='temp_model.p'):
	torch.save(model.state_dict(), path)
	size_kb = os.path.getsize(path) / 1024
	os.remove(path)
	return size_kb


teacher = TeacherCNN()
teacher_int8 = quantize_dynamic(teacher, {nn.Linear, nn.Conv2d}, dtype=torch.qint8)
teacher_fp32_size = get_model_size(teacher)
teacher_int8_size = get_model_size(teacher_int8)

print(f'Teacher FP32 size: {teacher_fp32_size:.1f} KB')
print(f'Teacher INT8 (attempted) size: {teacher_int8_size:.1f} KB')

print('--- features ---')
print(teacher_int8.features)
print('--- classifier ---')
print(teacher_int8.classifier)

在上述範例中,你可以看到量化後的模型,比原始模型的四分之一還要大了一些,並且 nn.Linear 已經變成了 DynamicQuantizedLinear,但是 nn.Conv2d 依然未改變,這些都說明了 nn.Conv2d 無法被 quantize_dynamic 處理。

而量化實際上的計算方式,其實也就是先縮放並平移至 8 位元整數範圍內,計算完畢後再還原,如下:

import numpy as np
import torch
import torch.nn as nn
from torch.ao.quantization import quantize_dynamic

torch.manual_seed(0)


class SingleLinear(nn.Module):
	def __init__(self):
		super().__init__()
		self.layer = nn.Linear(4, 3)

	def forward(self, x):
		return self.layer(x)


model = SingleLinear()
x = torch.randn(5, 4)

# PyTorch built-in method
model_int8 = quantize_dynamic(model, {nn.Linear}, dtype=torch.qint8)
q_weight = model_int8.layer.weight()
out_pytorch = model_int8(x).detach().numpy()

W = model.layer.weight.detach().numpy()  # shape (3, 4)
b = model.layer.bias.detach().numpy()
X = x.numpy()  # shape (5, 4)

# 1. Quantize weight: done once, symmetric, zero_point is 0
w_scale = np.abs(W).max() / 127
W_int8 = np.round(W / w_scale).clip(-127, 127).astype(np.int8)

# 2. Quantize activation: done every forward, asymmetric
x_min, x_max = X.min(), X.max()
x_scale = (x_max - x_min) / 255
x_zero_point = np.round(-x_min / x_scale).clip(0, 255)
X_uint8 = np.round(X / x_scale + x_zero_point).clip(0, 255).astype(np.uint8)

# 3. Integer-only math: multiply int8/uint8, sum in int32, then scale back to float
X_int32 = X_uint8.astype(np.int32)
W_int32 = W_int8.astype(np.int32)
raw_acc = X_int32 @ W_int32.T
correction = x_zero_point * W_int32.sum(axis=1)
out_manual = x_scale * w_scale * (raw_acc - correction) + b

print('=== Weight Comparison ===')
print('PyTorch quantized weight:')
print(q_weight.int_repr().numpy())
print('Manual quantized weight:')
print(W_int8)

print()
print('=== Output Comparison ===')
print('PyTorch quantize_dynamic output:')
print(out_pytorch)
print('Manual output (integer-only math):')
print(out_manual)

在上述範例中:

關於蒸餾和量化的其他細節與延伸方向,簡單說明如下: