在前几篇文章中，我们已经探讨了批量梯度下降的基本概念及其应用。然而，尽管批量梯度下降在理论上是完美的，但它在实际应用中可能会遇到计算效率的问题，特别是当数据集非常大时。为了解决这一问题，我们可以采用随机梯度下降和小批量梯度下降的方法，这些方法不仅加快了训练速度，还能够帮助模型跳出局部最优解。
随机梯度下降
回顾一下批量梯度下降公式：
θ_j := θ_j - α * Σ(hθ(x(i)) - y(i)) * x_j(i) for every j
虽然批量梯度下降是通过使用所有样本来估计梯度的，但在实际应用中，这样的计算可能会非常缓慢。因此，我们可以只使用一个训练样本来更新模型参数，这就是随机梯度下降的基本思想。
在随机梯度下降中，我们可以有两种采样方式：有放回和无放回。
- 无放回：意味着在同一轮次中，任何样本都不会重复使用。
- 有放回：指在同一轮次中，某些样本可能会重复出现。
其中，轮次（epoch）表示训练集被完整遍历一次。在无放回的情况下，我们从第一个样本开始，遍历到第m个样本，在每次遍历中，仅使用一个样本来更新权重：
∂J/∂θ_j = (hθ(x(i)) - y(i)) * x_j
θ_j := θ_j - α * (hθ(x(i)) - y(i)) * x_j(i)
小批量梯度下降
由于随机梯度下降在某些情况下可能不够稳定，因此引入了小批量梯度下降（Mini-Batch Gradient Descent）。该方法通过使用一部分样本来计算损失和梯度，从而在批量梯度下降和随机梯度下降之间取得平衡。小批量梯度下降在深度学习中被广泛使用。
∂J/∂θ_j = Σ(hθ(x(i)) - y(i)) * x_j for every j in the batch
θ_j := θ_j - α * Σ(hθ(x(i)) - y(i)) * x_j(i)
同样地，我们可以选择有放回或无放回的方式。在无放回的情况下，我们将训练集平均分成若干批次，并在一个轮次内用尽所有训练样本。
多项式回归
线性回归的局限性在于当数据呈现非线性关系时，模型无法很好地拟合数据。在这种情况下，我们可以使用多项式回归，通过多项式变换对特征进行工程化处理。例如，一级多项式回归（直线）可以表示为：
y = ax + b
而三次多项式回归则拟合一个三次曲线：
y = ax^3 + bx^2 + cx + d
代码实现
在实现过程中，我们将演示如何使用MLFlow进行实验跟踪，这是一个广受欢迎的工具，能够帮助我们记录和管理机器学习实验。以下是代码的关键步骤和注意事项。
首先，在终端运行以下命令以启动MLFlow服务：
docker compose up
这将会构建一个MLFlow服务，你可以通过http://localhost:5000进行访问。
接下来，我们将使用Python代码来演示随机梯度下降和小批量梯度下降的实际应用。
import numpy as np
from sklearn.datasets import load_diabetes
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
from time import time
import mlflow
import os
# 加载糖尿病数据集
diabetes = load_diabetes()
X = diabetes.data
y = diabetes.target
# 进行训练集和测试集的划分
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size = 0.3)
# 标准化处理
scaler = StandardScaler()
X_train = scaler.fit_transform(X_train)
X_test = scaler.transform(X_test)
# 添加截距项
intercept = np.ones((X_train.shape[0], 1))
X_train = np.concatenate((intercept, X_train), axis=1)
intercept = np.ones((X_test.shape[0], 1))
X_test = np.concatenate((intercept, X_test), axis=1)
# 实现线性回归的类
class LinearRegression:
def __init__(self, lr=0.001, num_epochs=500, batch_size=50, method='batch'):
self.lr = lr
self.num_epochs = num_epochs
self.batch_size = batch_size
self.method = method
def mse(self, ytrue, ypred):
return ((ypred - ytrue) ** 2).sum() / ytrue.shape[0]
def fit(self, X_train, y_train):
self.theta = np.zeros(X_train.shape[1])
for epoch in range(self.num_epochs):
if self.method == 'sto':
for i in range(X_train.shape[0]):
self._train(X_train[i].reshape(1, -1), y_train[i])
elif self.method == 'mini':
for i in range(0, X_train.shape[0], self.batch_size):
X_batch = X_train[i:i+self.batch_size, :]
y_batch = y_train[i:i+self.batch_size]
self._train(X_batch, y_batch)
else:
self._train(X_train, y_train)
def _train(self, X, y):
yhat = self.predict(X)
grad = X.T @ (yhat - y) / X.shape[0]
self.theta = self.theta - self.lr * grad
def predict(self, X):
return X @ self.theta
def coef(self):
return self.theta[1:]
def intercept(self):
return self.theta[0]
实验
我们接下来将展示批量梯度下降和小批量梯度下降的实验过程。
# 批量梯度下降
model = LinearRegression(method='batch', lr=0.1)
model.fit(X_train, y_train)
yhat = model.predict(X_test)
mse = model.mse(y_test, yhat)
print("Test MSE: ", mse)
# 小批量梯度下降
model = LinearRegression(method='mini', lr=0.1)
model.fit(X_train, y_train)
yhat = model.predict(X_test)
mse = model.mse(y_test, yhat)
print("Test MSE: ", mse)
如果你觉得这篇博文对你有帮助，请点赞、收藏、关注我，并且可以打赏支持我！
欢迎关注我的后续博文，我将分享更多关于人工智能、自然语言处理和计算机视觉的精彩内容。
谢谢大家的支持！