import matplotlib.pyplot as plt
import torch
from IPython import display
from d2l import torch as d2l
batch_size = 256
train_iter, test_iter = d2l.load_data_fashion_mnist(batch_size) # Return the iterator of the training set and test set
# Will flatten each image, treating them as vectors of length 784. The image is 28*28, stretched to 784
# Because the data set has 10 categories, the grid output dimension is 10
num_inputs = 784
num_output = 10
w = torch.normal(0, 0.01, size=(num_inputs, num_output), requires_grad=True) # Weight size: shape --num_inputs: row, num_output: column
b = torch.zeros(num_output, requires_grad=True) # Deviation
'''
# Review: Given a matrix, we can sum all elements
x = torch.tensor([[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]])
print(x)
print(x.sum(0, keepdim=True), '\\
', x.sum(1, keepdim=True))
# tensor([[1., 2., 3.],
# [4., 5., 6.]])
# tensor([[5., 7., 9.]])
# tensor([[ 6.],
# [15.]])
'''
# Implement softmax. Using softmax on a matrix is to perform it on each row of the matrix.
"""
exponentiate each term (using exp);
Sum each row (each sample in the mini-batch is one row) to get the normalization constant for each sample;
Divide each row by its normalizing constant, making sure the results sum to 1.
"""
def softmax(x):
x_exp = torch.exp(x)
partition = x_exp.sum(1, keepdim=True)
return x_exp / partition # The broadcast mechanism is applied
# test
x = torch.normal(0, 1, (2, 5))
x_prob = softmax(x)
# print(x_prob, '\\
', x_prob.sum(1))
#Test result: Turn each element into a non-negative number. Furthermore, according to the principle of probability, the sum of each row is 1
# tensor([[0.1488, 0.0353, 0.1059, 0.5232, 0.1868],
# [0.1903, 0.0968, 0.3103, 0.1855, 0.2172]])
# tensor([1.0000, 1.0000])
# Implement softmax regression model
def net(x):
return softmax(torch.matmul(x.reshape((-1, w.shape[0])), w) + b) # x.reshape((-1, w.shape[0]):
# -1 means automatically calculating the dimension size and adjusting x to the same shape as w
# is wx + b
# Demonstrate how to take out my corresponding predicted value based on my label in my predicted value.
# Create a data y_hat, which contains the predicted probabilities of 2 samples in 3 categories, using y as the index of the probability in y_hat
y = torch.tensor([0, 2]) # represents two real labels
y_hat = torch.tensor([[0.1, 0.3, 0.6], [0.3, 0.2, 0.5]]) # Predicted value
# print(y_hat[[0, 1], y]) # Here [0, 1] represents the 0th and 1st rows of y_hat, y is [0, 2], which represents the 0th and 2nd columns, The combination is (0,0) and (1,2)
# Implement the cross entropy loss function -ln(y_hat[range(len(y_hat)), y])
def cross_entropy(y_hat, y):
return -torch.log(y_hat[range(len(y_hat)), y])
# print(cross_entropy(y_hat, y)) # tensor([2.3026, 0.6931])
# Compare predicted categories to true y elements
def accuracy(y_hat, y):
"""Calculate the number of correct predictions"""
if len(y_hat.shape) > 1 and y_hat.shape[1] > 1: # The number of rows and columns of y_hat is greater than one
y_hat = y_hat.argmax(axis=1) # The subscript with the largest element value in each row is stored in y_hat
cmp = y_hat.type(y.dtype) == y # Convert y_hat to the data type of y and then compare it to become a bool tensor
return float(cmp.type(y.dtype).sum()) # Then convert cmp into the same shape as y and calculate the sum
# print(accuracy(y_hat, y) / y) # tensor([ inf, 0.5000])
# We can evaluate the accuracy of any model net
def evaluate_accuracy(net, data_iter):
"""Calculate the accuracy of the model on the specified data set"""
if isinstance(net, torch.nn.Module):
net.eval() # Set the model to evaluation mode
metric = Accumulator(2) # Number of correct predictions, total number of predictions Accumulator object, which is used to accumulate two values: the number of correct predictions and the total number of predictions.
for x, y in data_iter:
metric.add(accuracy(net(x), y), y.numel())
return metric[0] / metric[1]
class Accumulator: # @save
"""Accumulate on n variables"""
def __init__(self, n):
self.data = [0.0] * n
def add(self, *args):
self.data = [a + float(b) for a, b in zip(self.data, args)]
def reset(self):
self.data = [0.0] * len(self.data)
def __getitem__(self, idx):
return self.data[idx]
def train_epoch_ch3(net, train_iter, loss, updater): # @save
"""Train the model for one iteration cycle (see Chapter 3 for definition)"""
# Set the model to training mode
if isinstance(net, torch.nn.Module):
net.train()
# Sum of training losses, sum of training accuracy, number of samples
metric = Accumulator(3)
for X, y in train_iter:
# Calculate gradient and update parameters
y_hat = net(X)
l = loss(y_hat, y)
if isinstance(updater, torch.optim.Optimizer):
# Use PyTorch’s built-in optimizer and loss function
updater.zero_grad()
l.mean().backward()
updater.step()
else:
# Use custom optimizer and loss function
l.sum().backward()
updater(X.shape[0])
metric.add(float(l.sum()), accuracy(y_hat, y), y.numel())
# Return training loss and training accuracy
return metric[0] / metric[2], metric[1] / metric[2]
class Animator: # @save
"""Draw data in animation"""
def __init__(self, xlabel=None, ylabel=None, legend=None, xlim=None,
ylim=None, xscale='linear', yscale='linear',
fmts=('-', 'm--', 'g-.', 'r:'), nrows=1, ncols=1,
figsize=(3.5, 2.5)):
# Draw multiple lines incrementally
if legend is None:
legend = []
d2l.use_svg_display()
self.fig, self.axes = d2l.plt.subplots(nrows, ncols, figsize=figsize)
if nrows * ncols == 1:
self.axes = [self.axes, ]
# Use lambda function to capture parameters
self.config_axes = lambda: d2l.set_axes(
self.axes[0], xlabel, ylabel, xlim, ylim, xscale, yscale, legend)
self.X, self.Y, self.fmts = None, None, fmts
def add(self, x, y):
# Add multiple data points to the chart
if not hasattr(y, "__len__"):
y = [y]
n = len(y)
if not hasattr(x, "__len__"):
x = [x] * n
if not self.X:
self.X = [[] for _ in range(n)]
if not self.Y:
self.Y = [[] for _ in range(n)]
for i, (a, b) in enumerate(zip(x, y)):
if a is not None and b is not None:
self.X[i].append(a)
self.Y[i].append(b)
self.axes[0].cla()
for x, y, fmt in zip(self.X, self.Y, self.fmts):
self.axes[0].plot(x, y, fmt)
self.config_axes()
display.display(self.fig)
plt.draw()
plt.pause(0.001)
display.clear_output(wait=True)
def train_ch3(net, train_iter, test_iter, loss, num_epochs, updater): # @save
"""Training model (see Chapter 3 for definition)"""
animator = Animator(xlabel='epoch', xlim=[1, num_epochs], ylim=[0.3, 0.9],
legend=['train loss', 'train acc', 'test acc'])
for epoch in range(num_epochs):
train_metrics = train_epoch_ch3(net, train_iter, loss, updater)
test_acc = evaluate_accuracy(net, test_iter)
animator.add(epoch + 1, train_metrics + (test_acc,))
train_loss, train_acc = train_metrics
assert train_loss < 0.5, train_loss
assert 1 >= train_acc > 0.7, train_acc
assert 1 >= test_acc > 0.7, test_acc
lr = 0.1
def updater(batch_size):
return d2l.sgd([w, b], lr, batch_size)
if __name__ == '__main__':
# print(evaluate_accuracy(net, test_iter))
num_epochs = 10
train_ch3(net, train_iter, test_iter, cross_entropy, num_epochs, updater)