梯度下降算法之随机梯度下降

梯度下降是搜索目标函数的最优值（最小值/最大值）的迭代优化过程。它是机器学习项目中最常用的改变模型参数以降低成本函数的方法之一。

梯度下降的主要目标是识别在训练和测试数据集上提供最大精度的模型参数。在梯度下降中，梯度是指向函数在特定点最陡上升的一般方向的向量。通过在梯度的相反方向上移动，算法可以逐渐向函数的较低值下降，直到达到函数的最小值。

随机梯度下降概念

随机梯度下降（Stochastic Gradient Descent, SGD）是梯度下降算法的一种变体，用于优化机器学习模型。它解决了在机器学习项目中处理大型数据集时，传统梯度下降方法的计算效率低下的问题。

在SGD中，每次迭代不使用整个数据集，而是仅选择单个随机训练示例（或小批量）来计算梯度和更新模型参数。这种随机选择将随机性引入到优化过程中，因此在随机梯度下降中称为“随机”。

使用SGD的优势在于其计算效率，尤其是在处理大型数据集时。通过使用单个示例或小批量，与需要处理整个数据集的传统梯度下降方法相比，每次迭代的计算成本显著降低。

随机梯度下降算法

初始化：随机初始化模型的参数。

设置参数：确定更新参数的迭代次数和学习率。

随机梯度下降循环：重复以下步骤，直到模型收敛或达到最大迭代次数:

（1）打乱训练数据集以引入随机性。

（2）以打乱的顺序对每个训练示例（或小批量）进行迭代。

（3）使用所述当前训练来计算所述成本函数相对于所述模型参数的梯度示例（或批次）。

（4）通过在负梯度的方向上采取由学习率缩放的步骤来更新模型参数。

（5）评估收敛标准，例如梯度迭代之间的成本函数差异。

返回优化参数：满足收敛条件或达到最大迭代次数后，返回优化的模型参数。

在SGD中，由于每次迭代只从数据集中随机选择一个样本，因此算法达到最小值所采用的路径通常比典型的梯度下降算法噪声更大。但这并不重要，因为算法所采用的路径并不重要，只要我们达到最小值，并且训练时间明显更短。

自定义SGD类实现随机梯度下降算法

我们将创建一个SGD类，其中包含在更新参数、拟合训练数据集和预测新测试数据时使用的方法。具体步骤，如下:

步骤1： 定义SGD（即Stochastic Gradient Descent）类

类SGD封装了用于训练线性回归模型的随机梯度下降算法。

我们初始化SGD优化器参数，如学习率（learn rate）、历元数量（number of epochs）、批量大小（batch size）和容差（tolerance）。它还将“权重”和“偏移”初始化为“无”。

预测函数(predict function)：此函数使用当前权重（weights）和偏差（bias）计算输入数据X的预测。它在输入X和权重之间执行矩阵乘法，然后加上偏置项。

均方误差函数(mean squared error)：该函数计算真实目标值y_true和预测值y_pred之间的均方误差。

梯度函数(gradient function)：使用均方误差损失函数的梯度公式计算损失函数相对于权重和偏差的梯度。

拟合方法(fit method)：该方法使用随机梯度下降将模型拟合到训练数据。它通过指定数量的历元进行迭代，混洗数据并更新每个历元中的权重和偏差。它还会定期打印损耗，并根据容差检查收敛。

它使用小批量数据（X_batch，y_batch）计算梯度。这与SGD的随机性质一致。

在计算梯度之后，该方法使用学习率和所计算的梯度来更新模型参数（self.weight和self.bias）。这与SGD中的参数更新步骤一致。

fit方法迭代通过每个时期的整个数据集（X，y）。然而，它使用小批量更新参数，如在小批量中迭代通过数据集的嵌套循环所指示的（对于范围（0，n_samples，self.batch_size））。这反映了SGD的随机性质，其中使用数据的子集来更新参数。

步骤2：实例化SGD并训练模型

我们将创建一个具有100行和5列的随机数据集，并在该数据上拟合我们的随机梯度下降类。然后，使用SGD的预测方法。

# coding: utf-8import numpy as np

class Stochastic_Gradient_Descent: """ 实现随机梯度下降法的类。 参数: - lr (float): 学习率，默认为0.01。 - epochs (int): 训练迭代次数，默认为1000。 - batch_size (int): 批量大小，默认为32。 - tol (float): 收敛容差，默认为1e-3。 属性: - weights (np.ndarray): 模型权重。 - bias (float): 模型偏置。 """ def __init__(self, lr=0.01, epochs=1000, batch_size=32, tol=1e-3): self.learning_rate = lr self.epochs = epochs self.batch_size = batch_size self.tolerance = tol self.weights = None self.bias = None def predict(self, X): """ 使用当前的权重和偏置预测目标值。 参数: - X (np.ndarray): 输入数据矩阵。 返回: - np.ndarray: 预测的目标值。 """ return np.dot(X, self.weights) + self.bias def mean_squared_error(self, y_true, y_pred): """ 计算真实值与预测值之间的均方误差。 参数: - y_true (np.ndarray): 真实目标值。 - y_pred (np.ndarray): 预测目标值。 返回: - float: 均方误差。 """ return np.mean( np.power(y_true - y_pred, 2)) def gradient(self, X_batch, y_batch): """ 计算给定批量数据的梯度。 参数: - X_batch (np.ndarray): 批量输入数据。 - y_batch (np.ndarray): 批量目标值。 返回: - np.ndarray, float: 权重和偏置的梯度。 """ y_pred = self.predict(X_batch) error = y_pred - y_batch gradient_weights = np.dot(X_batch.T, error) / X_batch.shape[0] gradient_bias = np.mean(error) return gradient_weights, gradient_bias def fit(self, X, y): """ 训练模型。 参数: - X (np.ndarray): 输入数据矩阵，其中每行代表一个样本的特征。 - y (np.ndarray): 目标值，一维数组，长度与X的行数相同。 返回: - np.ndarray, float: 最终的权重和偏置。权重为特征对应的影响力，偏置为截距。 """ # 初始化权重和偏置，使用随机值 n_samples, n_features = X.shape self.weights = np.random.randn(n_features) self.bias = np.random.randn() # 迭代训练过程 for epoch in range(self.epochs): # 每轮训练重新打乱数据顺序 indices = np.random.permutation(n_samples) X_shuffled = X[indices] y_shuffled = y[indices] # 使用批量梯度下降更新权重和偏置 for i in range(0, n_samples, self.batch_size): X_batch = X_shuffled[i:i + self.batch_size] y_batch = y_shuffled[i:i + self.batch_size] # 计算当前批次的梯度 gradient_weights, gradient_bias = self.gradient(X_batch, y_batch) # 根据梯度更新权重和偏置 self.weights -= self.learning_rate * gradient_weights self.bias -= self.learning_rate * gradient_bias # 每隔一定轮数打印当前损失 if epoch % 100 == 0: y_pred = self.predict(X) loss = self.mean_squared_error(y, y_pred) print(f'Epoch: {epoch}, Loss: {loss}') # 如果梯度的范数小于预设容忍值，则认为模型收敛 if np.linalg.norm(gradient_weights) < self.tolerance: print(f'Converged at epoch: {epoch}') break # 返回最终的权重和偏置 return self.weights, self.bias # 生成随机训练数据集X = np.random.rand(100, 5)# 生成对应的噪声目标值y = np.dot(X, np.array([1, 2, 3, 4, 5])) + np.random.randn(100) * 0.1# 创建随机梯度下降法的实例model = Stochastic_Gradient_Descent(lr=0.1, epochs=1000, batch_size=32, tol=1e-3)w, b = model.fit(X, y)print(f'Weights: {w}, Bias: {b}')# 使用模型进行预测y_pred = model.predict(X)print('y_pred by model predicting: \n', y_pred)y_pred = np.dot(X, w) + bprint('y_pred by formula：\n', y_pred)

输出：（略）动手实践（直接拷贝上面代码查看运行结果）

为了减少损失函数，这种取值并根据不同的参数调整这些值的循环称为反向传播。

使用TensorFlow实现随机梯度下降算法

# coding: utf-8import tensorflow as tfimport numpy as np

class Stochastic_Gradient_Descent: """ 使用随机梯度下降法进行模型训练的类。 参数: - lr: 学习率，默认为0.001。 - epochs: 训练轮数，默认为2000。 - batch_size: 批量大小，默认为32。 - tol: 允许的梯度范数最小值，用于判断是否收敛，默认为1e-3。 属性: - weights: 模型权重。 - bias: 模型偏置。 """ def __init__(self, lr=0.001, epochs=2000, batch_size=32, tol=1e-3): self.learning_rate = lr self.epochs = epochs self.batch_size = batch_size self.tolerance = tol self.weights = None self.bias = None

def predict(self, X): """ 基于输入X进行预测。 参数: - X: 输入数据，二维数组。 返回: - 预测结果，基于X和当前权重、偏置计算得到。 """ return tf.matmul(X, self.weights) + self.bias

def mean_squared_error(self, y_true, y_pred): """ 计算真实标签与预测标签之间的均方误差。 参数: - y_true: 真实标签。 - y_pred: 预测标签。 返回: - 均方误差。 """ return tf.reduce_mean(tf.square(y_true - y_pred))

def gradient(self, X_batch, y_batch): """ 计算给定批次数据的梯度。 参数: - X_batch: 批量输入数据。 - y_batch: 批量标签数据。 返回: - 权重和偏置的梯度元组。 """ # 使用梯度带记录操作和梯度 with tf.GradientTape() as tape: # 使用当前模型预测目标 y_pred = self.predict(X_batch) # 计算真实值和预测值之间的均方误差损失 loss = self.mean_squared_error(y_batch, y_pred) # 计算损失相对于权重和偏置的梯度 gradient_weights, gradient_bias = tape.gradient( loss, [self.weights, self.bias]) return gradient_weights, gradient_bias

def fit(self, X, y): """ 训练模型。 参数: - X: 输入数据，二维数组。 - y: 标签数据，一维数组。 返回: - 训练完成的权重和偏置。 """ n_samples, n_features = X.shape self.weights = tf.Variable(tf.random.normal((n_features, 1))) self.bias = tf.Variable(tf.random.normal(()))

for epoch in range(self.epochs): # 每轮训练开始时打乱数据顺序 indices = tf.random.shuffle(tf.range(n_samples)) X_shuffled = tf.gather(X, indices) y_shuffled = tf.gather(y, indices)

# 使用批量数据进行训练 for i in range(0, n_samples, self.batch_size): X_batch = X_shuffled[i:i+self.batch_size] y_batch = y_shuffled[i:i+self.batch_size]

# 计算梯度，并进行梯度裁剪以防止梯度爆炸 gradient_weights, gradient_bias = self.gradient( X_batch, y_batch) gradient_weights = tf.clip_by_value(gradient_weights, -1, 1) gradient_bias = tf.clip_by_value(gradient_bias, -1, 1) # 更新权重和偏差 self.weights.assign_sub(self.learning_rate * gradient_weights) self.bias.assign_sub(self.learning_rate * gradient_bias)

# 每隔100轮打印一次损失值 if epoch % 100 == 0: y_pred = self.predict(X) loss = self.mean_squared_error(y, y_pred) print(f"Epoch {epoch:3d}: loss = {loss:.4f}")

# 检查梯度范数，如果小于容忍度则认为模型已收敛 if tf.norm(gradient_weights) < self.tolerance: print(f"Converged after {epoch} epochs.") break return self.weights.numpy(), self.bias.numpy()

# 创建一个包含100行和5列的随机数据集X = np.random.randn(100, 5).astype(np.float32)# 根据数据集创建相应的目标值，添加随机噪声y = np.dot(X, np.array([1, 2, 3, 4, 5], dtype=np.float32) ) + np.random.randn(100).astype(np.float32) * 0.1

# 创建SGD模型实例model = Stochastic_Gradient_Descent( lr=0.005, epochs=1000, batch_size=12, tol=1e-5)w, b = model.fit(X, y)print(f'Weights: {w}, Bias: {b}')# 使用模型进行预测y_pred = model.predict(X)print('y_pred by model predicting: \n', y_pred)y_pred = np.dot(X, w) + bprint('y_pred by formula：\n', y_pred)

输出：（略）动手实践（直接拷贝上面代码查看运行结果）

随机梯度下降的优点

速度：SGD比梯度下降的其他变体（如批量梯度下降和小批量梯度下降）更快，因为它只使用一个示例来更新参数。

内存效率：由于SGD一次更新一个训练示例的参数，因此它是内存高效的，并且可以处理无法放入内存的大型数据集。

避免局部最小值：由于SGD中的噪声更新，它具有逃离局部最小值并收敛到全局最小值的能力。

随机梯度下降的缺点

噪声更新：SGD中的更新是有噪声的，并且具有很高的方差，这会使优化过程不太稳定，并导致在最小值附近振荡。

缓慢收敛：SGD可能需要更多的迭代才能收敛到最小值，因为它一次更新一个训练示例的参数。

对学习率的敏感性：在SGD中，学习率的选择至关重要，因为使用高学习率会导致算法超过最小值，而低学习率会使算法收敛缓慢。

不太准确：由于噪声更新，SGD可能不会收敛到精确的全局最小值，并可能导致次优解。这可以通过使用诸如学习速率调度和基于动量的更新之类的技术来缓解。