在深度学习中，无论采用哪种梯度下降的方法，将整个训练集遍历一遍的过程称为一个轮次（Epoch）。
如果训练集有 个样本：
随机梯度下降：每轮进行 次迭代；
全批量梯度下降：每轮仅 1 次迭代；
小批量梯度下降（批大小 ）：每轮进行 次迭代。
通常，网络模型需要多个轮次的重复训练，才能使参数充分收敛到较优状态。
import numpy as np张量¶
class Tensor:
def __init__(self, data):
self.data = np.array(data)
self.grad = np.zeros_like(self.data)
self.gradient_fn = None
self.parents = set()
def backward(self):
if self.gradient_fn is not None:
self.gradient_fn()
for p in self.parents:
p.backward()
def __str__(self):
return f'Tensor({self.data})'数据集¶
class Dataset:
def __init__(self, batch_size=1):
self.batch_size = batch_size
self.load()
self.train()
def load(self):
self.train_data = ([[22.5, 72.0],
[31.4, 45.0],
[19.8, 85.0],
[27.6, 63.0]],
[[95],
[210],
[70],
[155]])
self.test_data = ([[28.1, 58.0]],
[[165]])
def train(self):
self.data = self.train_data
def eval(self):
self.data = self.test_data
def all(self):
x, y = self.data
return Tensor(x), Tensor(y)
def __len__(self):
x, *_ = self.data
return len(x) // self.batch_size
def __getitem__(self, index):
s = slice(index * self.batch_size, (index + 1) * self.batch_size)
x, y = self.data
return Tensor(x[s]), Tensor(y[s])模型¶
class Linear:
def __init__(self, in_size, out_size):
self.weight = Tensor(np.ones((out_size, in_size)) / in_size)
self.bias = Tensor(np.zeros(out_size))
def __call__(self, x: Tensor):
return self.forward(x)
def forward(self, x: Tensor):
p = Tensor(x.data @ self.weight.data.T + self.bias.data)
def gradient_fn():
self.weight.grad += p.grad.T @ x.data
self.bias.grad += np.sum(p.grad, axis=0)
p.gradient_fn = gradient_fn
return p
@property
def parameters(self):
return [self.weight, self.bias]损失函数（均方误差）¶
class MSELoss:
def __call__(self, p: Tensor, y: Tensor):
return self.loss(p, y)
def loss(self, p: Tensor, y: Tensor):
mse = Tensor(np.mean(np.square(y.data - p.data)))
def gradient_fn():
p.grad += -2 * (y.data - p.data) / y.data.size
mse.gradient_fn = gradient_fn
mse.parents = [p]
return mse优化器（随机梯度下降）¶
class SGDOptimizer:
def __init__(self, parameters, lr):
self.parameters = parameters
self.lr = lr
def zero_grad(self):
for p in self.parameters:
p.grad = np.zeros_like(p.data)
def step(self):
for p in self.parameters:
p.data -= p.grad * self.lr训练器¶
随着迭代、批处理和轮次等概念的引入，模型训练和评估的代码也逐渐变得复杂起来。我们将其封装为训练器（Trainer），就可以在不同的项目复用。
最基本的训练器主要提供两个函数：
train（训练函数）：训练函数需要两个参数：数据集和轮数。训练函数将使用数据集的训练集，完成指定轮数的迭代训练。test（测试函数）：测试函数只需要一个参数：数据集。测试函数将使用数据集的测试集，对训练好的模型进行一轮测试。
💡 提示：观察一下训练器的训练函数，可以发现它的核心正是之前讲过的“五步循环”（清零梯度、前向传播、计算损失、反向传播、更新参数），只是外面又套了一层轮次循环。
class Trainer:
def __init__(self, model, loss_fn, optimizer):
self.model = model
self.loss_fn = loss_fn
self.optimizer = optimizer
# 训练函数
def train(self, dataset, epochs):
# 使用训练集
dataset.train()
# 轮次
for epoch in range(epochs):
# 迭代
for i in range(len(dataset)):
feature, label = dataset[i]
self.optimizer.zero_grad()
prediction = self.model(feature)
loss = self.loss_fn(prediction, label)
loss.backward()
self.optimizer.step()
# 测试函数
def test(self, dataset):
# 使用测试集
dataset.eval()
feature, label = dataset.all()
prediction = self.model(feature)
loss = self.loss_fn(prediction, label)
return prediction, lossLEARNING_RATE = 0.00001批大小¶
BATCH_SIZE = 2轮数¶
这是我们的第三个超参数。轮数（Epochs）定义了模型训练中，训练数据的重复迭代次数。
轮数并非越多越好：
在训练初期，模型参数通常会随着轮次增加而快速收敛，误差大幅下降；
随后，收敛速度逐渐趋于平缓；
如果轮数过多，模型可能会由于过拟合，导致在实际应用中的效果反而变差。
因此，我们需要通过监测网络模型在测试数据上的表现，找到性能提升的拐点（Turning Point），从而确定最佳轮数：
轮数太少：欠拟合；
轮数太多：可能过拟合。
一个有效的实践技巧叫做早停（Early Stopping）：就是在每个轮次结束后，和上一轮对比。如果损失值的下降幅度已经很小（比如：小于 1%），就说明模型已经基本收敛，可以提前停止训练，避免过拟合。
而另一相关的实践技巧叫做重排（Shuffle）：就是在每个轮次里打乱训练数据的迭代次序，以期进一步优化训练，获得更好的泛化能力。
EPOCHS = 1000建模¶
我们将模型、损失函数和优化器的实例作为训练器的参数。
dataset = Dataset(BATCH_SIZE)
model = Linear(2, 1)
loss_fn = MSELoss()
optimizer = SGDOptimizer(model.parameters, lr=LEARNING_RATE)
trainer = Trainer(model, loss_fn, optimizer)训练¶
有了训练器后，模型训练只需要调用训练器的训练函数。
trainer.train(dataset, EPOCHS)评估¶
同样的，评估也只需要调用训练器的测试函数。
prediction, loss = trainer.test(dataset)
print(f'prediction:\t{prediction}\nloss:\t{loss}')prediction: Tensor([[163.52327795]])
loss: Tensor(2.1807080003283335)
经过 1000 个轮次，损失值降低至个位数。
至此，我们可以宣布：模型训练成功！
损失曲线¶
最后，让我们监控一下每轮训练损失值的变化，直观地感受一下模型训练的过程，观察一下哪里是拐点。
我们使用 Matplotlib 图形库来展示随着轮次增加，损失值的变化。
import matplotlib.pyplot as plt
# 建模
dataset = Dataset(BATCH_SIZE)
model = Linear(2, 1)
loss_fn = MSELoss()
optimizer = SGDOptimizer(model.parameters, lr=LEARNING_RATE)
# 训练
losses = []
for epoch in range(EPOCHS):
dataset.train()
for i in range(len(dataset)):
feature, label = dataset[i]
optimizer.zero_grad()
prediction = model(feature)
loss = loss_fn(prediction, label)
loss.backward()
optimizer.step()
dataset.eval()
feature, label = dataset.all()
prediction = model(feature)
loss = loss_fn(prediction, label)
losses.append(loss.data)
# 展示
plt.figure(figsize=(5, 3))
plt.plot(losses, linewidth=2)
plt.xlabel('Epochs', fontsize=12)
plt.ylabel('MSE Loss', fontsize=12)
plt.title('Loss Curve', fontsize=14, fontweight='bold')
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
上图是模型训练的损失曲线（Loss Curve），直观地向我们展示了随着轮次的增加，损失值快速降低，然后逐渐趋于平缓的过程。
课后练习¶
尝试在迭代过程中实现早停机制。调整早停的比例（比如：0.1%、0.5%、1%），观察对训练轮数有什么影响？
尝试在迭代过程中实现重排机制。观察对损失值有什么影响？
观察上面的损失曲线，你认为哪里是拐点？