线性回归到前馈神经网络

机器学习算法比较擅长做预测，在这里主要专注于机器学习预测任务，即使用观察数据来进行预测。特别关注以下两个方面：

• 回归：预测一个量（例如距离、时间）并理解两个变量之间的关系。

• 分类：为输入分配一个离散类（例如，交通标志）。

通过创建不同的ML模型以根据输入观察预测输出。为了估计模型的拟合度，将使用损失函数，并使用选定的指标量化所创建模型的性能。

• 模型：从输入中提取的算法模式。

• 损失函数/成本函数：将模型的输出映射到单个实数的函数。

• 指标：量化算法性能的函数。

1. 线性回归

线性回归模型是一种简单的ML算法，它假设输入变量和输出变量之间存在线性关系。这样的模型由两个参数描述，斜率 m 和交点 b 使得 y = mx +b。拟合或训练这样的模型需要调整 m 和 b 以最小化所选的损失函数。

线性回归损失函数

均方误差损失(MSE)或称为L2损失是线性回归算法中最常用的函数之一。它是通过对真实值和预测的平方差求和来计算的。由于平方函数的性质，此损失对异常值（分布外数据点）非常敏感。如果数据集包含许多异常值，则L1损失（真实值和预测之间的绝对差异）可能是更好的选择。

在实践中，我们使用简化的线性回归表达式：Y = XW ，其中 W 是包含斜率和偏差的 (2x1)权重矩阵， X 是包含观测值的(nx2)矩阵。

L2 损失函数$L (W) = (XW-Y) ^ T (XW-Y)$

2. 逻辑回归

对于分类问题，可以使用线性表达式对属于某个类别的概率进行建模，这样的模型称为逻辑回归模型。然而，考虑到要对概率建模，我们需要一种方法将$mx+b$约束到[0, 1]区间。为此，将使用sigmoid函数将任何实数映射到[0, 1]区间内。

sigmoid函数：

\[\sigma (x) = \frac{e^ x}{\mathrm{1} + e^ x }\]

softmax函数则可以将逻辑函数扩展到多个类，并以向量而不是实数作为输出。softmax函数输出一个离散概率分布：一个与输入维度相似的向量，但其所有分量之和为1。在机器学习领域，sigmoid函数和softmax函数均称为激活函数。

softmax函数：

\[\sigma(z_i) = \frac{e^{z_{i}}}{\sum_{j=1}^K e^{z_{j}}} \ \ \ for\ i=1,2,\dots,K\]

交叉熵损失函数

交叉熵 (CE) 损失是分类问题中最常见的损失。总损失等于真实值one-hot编码向量的点积与softmax概率向量的对数的所有观察值的总和。

当分类类别数量是2时，交叉熵损失如下：

\[-{(y\log(p) + (1 - y)\log(1 - p))}\]

当分类类别数量超过2时，交叉熵损失如下：

\[-\sum_{c=1}^My_{o,c}\log(p_{o,c})\]

对于多类分类问题，真实值标签需要编码为向量来计算。一种常见的方法是独热（one-hot）编码方法，其中数据集的每个标签都分配一个整数。该整数用作唯一独热向量的唯一非零元素的索引。

总结：对于分类问题，需要将标签编码为维度为C的向量，其中C是数据集中的类数。多亏了softmax函数，模型输出了一个离散概率分布向量，也是C维的。为了计算输入和输出之间的交叉熵损失，我们计算了一个热向量和输出对数的点积。

TensorFlow实现逻辑回归

import tensorflow as tf


def softmax(logits):
    """
    softmax implementation
    args:
    - logits [tensor]: 1xN logits tensor
    returns:
    - soft_logits [tensor]: softmax of logits
    """
    exp = tf.exp(logits)
    denom = tf.math.reduce_sum(exp, 1, keepdims=True)
    return exp / denom


def cross_entropy(scaled_logits, one_hot):
    """
    Cross entropy loss implementation
    args:
    - scaled_logits [tensor]: NxC tensor where N batch size / C number of classes
    - one_hot [tensor]: one hot tensor
    returns:
    - loss [tensor]: cross entropy 
    """
    masked_logits = tf.boolean_mask(scaled_logits, one_hot) 
    return -tf.math.log(masked_logits)


def model(X, W, b):
    """
    logistic regression model
    args:
    - X [tensor]: input HxWx3
    - W [tensor]: weights
    - b [tensor]: bias
    returns:
    - output [tensor]
    """
    flatten_X = tf.reshape(X, (-1, W.shape[0]))
    return softmax(tf.matmul(flatten_X, W) + b)


def accuracy(y_hat, Y):
    """
    calculate accuracy
    args:
    - y_hat [tensor]: NxC tensor of models predictions
    - y [tensor]: N tensor of ground truth classes
    returns:
    - acc [tensor]: accuracy
    """
    # calculate argmax
    argmax = tf.cast(tf.argmax(y_hat, axis=1), Y.dtype)

    # calculate acc
    acc = tf.math.reduce_sum(tf.cast(argmax == Y, tf.int32)) / Y.shape[0]
    return acc

3. 梯度下降

拟合或训练ML算法包括找到最小化损失函数的权重组合。在某些情况下，可以找到解析解（例如，具有L2损失的线性回归）。然而，对于大多数算法，损失最小化问题的解析解不存在。

梯度下降算法是一种使用迭代寻找损失函数最小值的方法。该算法通过一定的学习率来逐步向损失函数最小值逼近。梯度的实质是变化率，找到损失函数最小值也就是找到损失函数变化最小的地方，也就是求解损失函数的导数，梯度即导数。

使用梯度下降算法的挑战之一是局部最小值的存在。局部最小值是损失函数域的局部子集中的最小值。局部最小值是此函数在某个小子集中可以采用的最小值，而不是全局最小值，我们的目标是找出损失函数在全域内的全局最小值。梯度下降算法可能会陷入局部最小值并输出次优解。

Tensorflow变量是具有固定类型和形状的张量，但它们的值可以通过操作改变。我们需要使用变量在Tensorflow中通过 tf.GradientTape api 计算梯度。

梯度下降优化器算法

对于梯度下降有几种不同的优化算法，其中用得最多的是随机梯度下降（SGD）。

随机梯度下降（SGD）

由于内存限制，整个数据集几乎从不一次加载并通过模型馈送，就像批量梯度下降的情况一样。相反，会创建成批的输入。一次仅对一个输入的批次执行的梯度下降称为随机梯度下降(SGD)，而多于一个但不是全部一次的批次（例如，20个批次，每个批次200张图像）称为小批量梯度下降。

TensorFlow实现随机梯度下降（SGD）

import argparse
import logging

import tensorflow as tf

from dataset import get_datasets
from logistic import softmax, cross_entropy, accuracy


def sgd(params, grads, lr, bs):
    """
    stochastic gradient descent implementation
    args:
    - params [list[tensor]]: model params
    - grad [list[tensor]]: param gradient
    - lr [float]: learning rate
    - bs [int]: batch_size
    """
    for param, grad in zip(params, grads):
        param.assign_sub(lr * grad / bs)


def training_loop(lr):
    """
    training loop
    args:
    - lr [float]: learning rate
    returns:
    - mean_acc [tensor]: training accuracy
    - mean_loss [tensor]: training loss
    """
    accuracies = []
    losses = []
    for X, Y in train_dataset:
        with tf.GradientTape() as tape:
            # forward pass
            X = X / 255.0
            y_hat = model(X)
            # calculate loss
            one_hot = tf.one_hot(Y, 43)
            loss = cross_entropy(y_hat, one_hot)
            losses.append(tf.math.reduce_mean(loss))

            grads = tape.gradient(loss, [W, b])
            sgd([W, b], grads, lr, X.shape[0]) 

            acc = accuracy(y_hat, Y)
            accuracies.append(acc)
    mean_acc = tf.math.reduce_mean(tf.concat(accuracies, axis=0))
    mean_loss = tf.math.reduce_mean(losses)
    return mean_loss, mean_acc

def model(X):
    """
    logistic regression model
    """
    flatten_X = tf.reshape(X, (-1, W.shape[0]))
    return softmax(tf.matmul(flatten_X, W) + b)


def validation_loop():
    """
    loop through the validation dataset
    """
    accuracies = []
    for X, Y in val_dataset:
        X = X / 255.0
        y_hat = model(X)
        acc = accuracy(y_hat, Y)
        accuracies.append(acc)
    mean_acc = tf.math.reduce_mean(tf.concat(accuracies, axis=0))
    return mean_acc


def get_module_logger(mod_name):
    logger = logging.getLogger(mod_name)
    handler = logging.StreamHandler()
    formatter = logging.Formatter('%(asctime)s %(levelname)-8s %(message)s')
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.setLevel(logging.DEBUG)
    return logger


if __name__  == '__main__':
    logger = get_module_logger(__name__)
    parser = argparse.ArgumentParser(description='Download and process tf files')
    parser.add_argument('--imdir', required=True, type=str,
                        help='data directory')
    parser.add_argument('--epochs', default=10, type=int,
                        help='Number of epochs')
    args = parser.parse_args()    

    logger.info(f'Training for {args.epochs} epochs using {args.imdir} data')
    # get the datasets
    train_dataset, val_dataset = get_datasets(args.imdir)

    # set the variables
    num_inputs = 1024*3
    num_outputs = 43
    W = tf.Variable(tf.random.normal(shape=(num_inputs, num_outputs),
                                    mean=0, stddev=0.01))
    b = tf.Variable(tf.zeros(num_outputs))

    lr = 0.1
    # training! 
    for epoch in range(args.epochs):
        logger.info(f'Epoch {epoch}')
        loss, acc = training_loop(lr)
        logger.info(f'Mean training loss: {loss:1f}, mean training accuracy {acc:1f}')
        val_acc = validation_loop()
        logger.info(f'Mean validation accuracy {val_acc:1f}')

其他梯度下降优化算法

• Momentum
• NAG
• Adagrad
• Adadelta
• RMSprop
• Adam
• AdaMax
• Nadam

学习率和退火

学习率在梯度下降方法的成功中起着至关重要的作用。一种非常流行的提高梯度下降方法性能的方法是学习率退火，学习率退火实质是在训练期间使用不同策略降低学习率。存在不同的策略，例如逐步退火、余弦退火或指数退火。另一种方法是使用学习调度器。

4. 前馈神经网络

前馈神经网络(FFNN)是逻辑回归算法的扩展。我们可以将逻辑回归视为具有单层的前馈神经网络。前馈神经网络堆叠多个隐藏层（不是输入或输出层的任何层），然后是非线性激活，例如sigmoid或softmax激活。

FFNN仅由全连接层组成，其中一层中的每个神经元都连接到前一层中的所有神经元。

让我们考虑一个FFNN其中一层包含n个神经元，在他前一层有m个神经元。该神经元使用前一层的输出执行线性运算 wX+b，这意味着X是一个mx1的向量。我们可以使用矩阵乘法一次执行所有操作，而不是遍历n个神经元中的每一个。该层的输出将通过计算 WX+B 获得，其中 B 是mx1 向量形式的偏差 ，W 是 nxm 矩阵形式的权重。

为了进一步简化，我们可以将偏置向量 B 合并到矩阵 W 中，使得 W 现在的维度为 nx(m+1)。 我们只需要创建一个维度为 (m+1)x1 的新输入向量 X，也就是当前馈神经网络只有一层的时候，他就是逻辑回归。

激活函数

激活函数在神经网络中起着至关重要的作用，因为它们为系统增加了非线性。最常见的激活函数是ReLU激活函数，但其他激活函数如sigmoid或双曲正切也很受欢迎。

ReLU激活函数： \(Relu(z) = max(0, z)\)

反向传播

拟合或训练一个模型本质上是一个优化问题，需要计算损失函数的梯度，逻辑回归只是单层网络，只需要计算一个单一函数的梯度，直接求导就可以。但是神经网络是多层网络，由很多单一函数复合而成复杂的复合函数。训练多层神经网络模型的本质即找到使整个复杂的复合损失函数梯度最小的参数集。对于复合函数的梯度（导数）求解遵循链式法则，计算最后一层的梯度，逐步传递直到第一层来更新参数，这就是反向传播。

链式法则允许您分解复合函数的导数计算。因为我们可以将ANN视为一个巨大的复合函数，所以链式法则是反向传播算法的核心。反向传播是用于计算相对于神经网络每个权重的损失梯度的机制。

TensorFlow实现最简单（2层）的前馈神经网络用于图像分类

import argparse

import tensorflow as tf
from tensorflow import keras

from utils import get_datasets, get_module_logger, display_metrics


def create_network():
    """ output a keras model """
    model = tf.keras.Sequential([
        tf.keras.layers.Flatten(input_shape=(32, 32, 3)),
        tf.keras.layers.Dense(128, activation='relu'),
        tf.keras.layers.Dense(43)])
    return model


if __name__  == '__main__':
    logger = get_module_logger(__name__)
    parser = argparse.ArgumentParser(description='Download and process tf files')
    parser.add_argument('-d', '--imdir', required=True, type=str,
                        help='data directory')
    parser.add_argument('-e', '--epochs', default=10, type=int,
                        help='Number of epochs')
    args = parser.parse_args()    

    logger.info(f'Training for {args.epochs} epochs using {args.imdir} data')
    # get the datasets
    train_dataset, val_dataset = get_datasets(args.imdir)

    model = create_network()

    model.compile(optimizer='adam',
              loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
              metrics=['accuracy'])
    history = model.fit(x=train_dataset, 
                        epochs=args.epochs, 
                        validation_data=val_dataset)
    display_metrics(history)

回首页