《深度学习图解》搭建深度学习框架


深度学习框架正是为了缓解这种代码复杂性而诞生。尤其是,如果你想在CPU上训练神经网络(这种硬件会带来10-100倍加速),深度学习框架可以显著减少代码复杂度(减少错误并加速开发),同时提高运行性能。

框架如何简化你的代码呢?

它让你不必写你本来要重复多次的代码,具体而言,深度学习框架中最有用的部分是它对自动反向传播与自动优化的支持——这些特性让你只写出模型的前向传播代码,就会使框架自动处理反向传播与权重更新。大多数框架甚至提供了常见层与损失函数的高级接口,使得写前向传播代码更容易。

损失函数介绍

通过调整网络测量误差的方式、网络层的参数数量和类型,以及应用的正则化类型来改变网络的学习过程。在深度学习研究中,所有这些技术都属于构建所谓损失函数的范畴。

神经网络并不能真正从数据中学习,它们只是使损失函数最小化。

之前章节学习就是调整神经网络中的权重,使误差降到0。在本节中,将从另一个角度来解释这一现象,通过选择误差的测量方式,来让神经网络学习我们感兴趣的模式:

image-20231206184331841

尽管神经网络架构和数据集都很相似,但误差函数却有根本的不同,也就导致了神经网络学到了截然不同的模式。

交叉熵介绍

Softmax函数

回顾下Softmax函数:Softmax函数是一种特殊的激活函数,它可以将神经网络的输出转化为一个概率分布。这意味着,对于多分类问题,Softmax函数可以将模型的原始输出转化为每个类别的预测概率。

Softmax函数的公式如下:
$$
\operatorname{Softmax}\left(z_i\right)=\frac{\exp \left(z_i\right)}{\sum_j \exp \left(z_j\right)}
$$

其中, $z_i$ 是第 $\mathrm{i}$ 个输出节点的原始输出值, $z_j$ 是所有输出节点的原始输出值。公式使所有输出值的和为1,这样每个输出值就可以被解释为对应类别的预测概率。

交叉熵

交叉熵损失函数可以衡量模型的预测概率分布与真实概率分布之间的差异,我们通常会使用Softmax函数将神经网络的输出转化为概率分布,然后使用交叉熵损失函数来优化我们的模型,使得模型的预测概率分布尽可能接近真实的概率分布。

我们可以像线性回归那样使用平方损失函数 $\left|\hat{\boldsymbol{y}}^{(i)}-\boldsymbol{y}^{(i)}\right|^2 / 2$ 。然而,想要预测分类结果正确,我们其实并不需要预测概率完全等于标签概率。例如,在图像分类的例子里,如果 $y^{(i)}=3$ ,那么我们只需要 $\hat{y}_3^{(i)}$ 比其他两个预测值 $\hat{y}_1^{(i)}$ 和 $\hat{y}_2^{(i)}$ 大就行了。即使 $\hat{y}_3^{(i)}$ 值为 0.6 ,不管其他两个预测值为多少,类别预测均正确。

交叉熵函数的公式如下:
$$
H(P, Q)=-\sum_i P(i) \log Q(i)
$$

其中, $P(i)$ 是样本的期望输出, $Q(i)$ 是样本的实际输出。这个公式表明,交叉摘刻画的是实际输出 (概率) 与期望输出 (概率) 的距离,也就是交叉摘的值越小,两个概率分布就越接近。

当然,遇到一个样本有多个标签时,例如图像里含有不止一个物体时,我们并不能做这一步简化。但即便对于这种情况,交叉熵同样只关心对图像中出现的物体类别的预测概率。

假设训练数据集的样本数为 $n$ ,交叉熵损失函数定义为
$$
\ell(\boldsymbol{\Theta})=\frac{1}{n} \sum_{i=1}^n H\left(\boldsymbol{y}^{(i)}, \hat{\boldsymbol{y}}^{(i)}\right)
$$

其中 $\Theta$ 代表模型参数。同样地,如果每个样本只有一个标签,那么交叉摘损失可以简写成 $\ell(\Theta)=-(1 / n) \sum_{i=1}^n \log \hat{y}{y^{(i)}}^{(i)}$ 。从另一个角度来看,我们知道最小化 $\ell(\Theta)$ 等价于最大化 $\exp (-n \ell(\Theta))=$ $\prod{i=1}^n \hat{y}_{y^{(i)}}^{(i)}$ ,即最小化交叉摘损失函数等价于最大化训练数据集所有标签类别的联合预测概率。

例如Softmax函数的输出为 [0.66, 0.24, 0.10],这个输出可以被解释为模型对每个类别的预测概率。

然后,我们使用交叉熵损失函数来计算损失:

交叉熵损失函数的计算过程如下:

  1. 对每个类别计算 -(y_true * log(y_pred))cross_entropy = [-1 * log(0.66), -0 * log(0.24), -0 * log(0.10)] = [0.41, 0, 0]
  2. 计算所有类别的损失的和:loss = sum(cross_entropy) = 0.41 + 0 + 0 = 0.41

所以,交叉熵损失为 0.41。这个值衡量了模型的预测概率分布与真实概率分布之间的差异。我们的目标是通过优化神经网络的参数来最小化这个损失值,使得模型的预测概率分布尽可能接近真实的概率分布。

为什么选交叉熵作为分类问题的损失函数,而不是传统机器学习更熟悉的均方差或者别的损失函数?

详见:CaptainBlackboard/D#0012-为什么选交叉熵作为分类问题的损失函数/D#0012.md at master · Captain1986/CaptainBlackboard (github.com)

张量介绍

张量是向量与矩阵的抽象形式。

向量是一维张量,矩阵是二维张量,而更高维的被称为n维张量。因此,搭建深度学习框架的第一件事情是创建这种基本类型,我们称之为张量(Tensor):

import numpy as np

class Tensor (object):
    
    def __init__(self, data):
        self.data = np.array(data)
    
    def __add__(self, other):
        return Tensor(self.data + other.data)
    
    def __repr__(self):
        return str(self.data.__repr__())
    
    def __str__(self):
        return str(self.data.__str__())
    
x = Tensor([1,2,3,4,5])
print(x)

y = x + x
print(y)

注意:它把所有数值信息都存储在一个Numpy数组(self.data)中,而且它支持一种张量操作(加法 add)。增加更多操作是相对容易:只需要在张量类中创建具有适当功能的其他函数即可。

PyTorch中的张量操作

Tensor是这个包的核心类,如果将其属性.requires_grad设置为True,它将开始追踪(track)在其上的所有操作(这样就可以利用链式法则进行梯度传播了)。完成计算后,可以调用.backward()来完成所有梯度计算。此Tensor的梯度将累积到.grad属性中。

注意:y.backward()时,如果y标量,则不需要为backward()传入任何参数;否则,需要传入一个与y同形的Tensor

为什么在反向转播时要这样传参?

因为我们不允许张量对张量求导,只允许标量对张量求导,求导结果是和自变量同形的张量。

例如:

假设y由自变量x计算而来,w是和y同形的张量,则y.backward(w)的含义是:先计算l = torch.sum(y * w),则l是个标量,然后求l对自变量x的导数。 参考

detach() 函数

在PyTorch中,detach()函数返回一个新的张量,该张量从当前图中分离出来。这意味着新的张量将不会跟踪应用于当前张量的任何操作。这对于调试或创建与当前图无关的计算非常有用。

返回的张量与原始张量共享相同的存储。在它们中的任何一个上的就地修改都将被看到,并可能在正确性检查中触发错误。

with torch.no_grad()函数

此外,还可以用with torch.no_grad()将需要关闭梯度计算的代码块包裹起来,在这个代码块中,PyTorch将不会追踪Tensor的梯度信息,这样就避免了不必要的内存消耗和计算开销。

作用:

  • 在测试或评估模型时,可以加快运行速度,节省显存。
  • 在不需要梯度的情况下,可以防止计算图的构建,避免出现梯度累积的问题。
  • 在需要固定某些参数不更新的情况下,可以使用with torch.no_grad()来实现。

例如:

import torch
import torchvision.models as models

# 加载预训练模型
model = models.resnet50(pretrained=True)

# 将模型设置为评估模式
model.eval()

# 创建一个随机输入
input = torch.randn(1, 3, 224, 224)

# 使用torch.no_grad()来关闭梯度计算
with torch.no_grad():
    output = model(input)
Function

Function是另外一个很重要的类。TensorFunction互相结合就可以构建一个记录有整个计算过程的有向无环图(DAG)。每个Tensor都有一个.grad_fn属性,该属性即创建该TensorFunction, 就是说该Tensor是不是通过某些运算得到的,若是,则grad_fn返回一个与这些运算相关的对象,否则是None。

例如:

x = torch.ones(2, 2, requires_grad=True)
print(x)
print(x.grad_fn)

# 输出
# tensor([[1., 1.],
#        [1., 1.]], requires_grad=True)
# None

y = x + 2
print(y)
print(y.grad_fn)

# 输出
# tensor([[3., 3.],
#        [3., 3.]], grad_fn=<AddBackward>)
# <AddBackward object at 0x1100477b8>

注意:x是直接创建的,所以它没有grad_fn, 而y是通过一个加法操作创建的,所以它有一个为<AddBackward>grad_fn

像x这种直接创建的称为叶子节点,叶子节点对应的grad_fnNone

例如:通过.requires_grad_()来用in-place的方式改变requires_grad属性:

a = torch.randn(2, 2) # 缺失情况下默认 requires_grad = False
a = ((a * 3) / (a - 1))
print(a.requires_grad) # False
a.requires_grad_(True)
print(a.requires_grad) # True
b = (a * a).sum()
print(b.grad_fn)

# 输出
# False
# True
# <SumBackward0 object at 0x118f50cc0>
求梯度
# 再来反向传播一次,注意grad是累加的
out2 = x.sum()
out2.backward()
print(x.grad)

out3 = x.sum()
x.grad.data.zero_()
out3.backward()
print(x.grad)

# 输出
# tensor([[5.5000, 5.5000],
#        [5.5000, 5.5000]])
# tensor([[1., 1.],
#        [1., 1.]])

注意:grad在反向传播过程中是累加的(accumulated),这意味着每一次运行反向传播,梯度都会累加之前的梯度,所以一般在反向传播之前需把梯度清零。

此外,如果我们想要修改tensor的数值,但是又不希望被autograd记录(即不会影响反向传播),那么我么可以对tensor.data进行操作。

x = torch.ones(1,requires_grad=True)

print(x.data) # 还是一个tensor
print(x.data.requires_grad) # 但是已经是独立于计算图之外

y = 2 * x
x.data *= 100 # 只改变了值,不会记录在计算图,所以不会影响梯度传播

y.backward()
print(x) # 更改data的值也会影响tensor的值
print(x.grad)Copy to clipboardErrorCopied

输出:

tensor([1.])
False
tensor([100.], requires_grad=True)
tensor([2.])

自动梯度计算(autograd)介绍

之前,你手动做反向传播。现在,将它自动化。

计算梯度这件事情需要反向经过网络:首先要计算输出层的梯度,然后用它计算倒数第二层的梯度,以此类推,直到为所有的权重都求得正确的梯度:

import numpy as np

class Tensor (object):
    
    def __init__(self, data, creators=None, creation_op = None):
        self.data = np.array(data)
        self.creation_op = creation_op
        self.creators = creators
        self.grad = None
    
    def backward(self, grad):
        self.grad = grad
        
        if(self.creation_op == "add"):
            self.creators[0].backward(grad)
            self.creators[1].backward(grad)

    def __add__(self, other):
        return Tensor(self.data + other.data,  creators=[self,other], creation_op="add")
    
    def __repr__(self):
        return str(self.data.__repr__())
    
    def __str__(self):
        return str(self.data.__str__())
    
x = Tensor([1,2,3,4,5])
y = Tensor([2,2,2,2,2])

z = x + y
z.backward(Tensor(np.array([1,1,1,1,1])))
print(x.grad)
print(y.grad)
print(z.creators)
print(z.creation_op)
[1 1 1 1 1]
[1 1 1 1 1]
[array([1, 2, 3, 4, 5]), array([2, 2, 2, 2, 2])]
add
  • 每个张量有两个属性,creators是一个列表,包含创建当前张量(默认为None)用到的所有张量。因此当两个张量x和y加到一起时,z包含两个creators,即x和y。

  • creation_op是一个相关特性,存储了creators在创建过程中用到的指令。因此,执行z=x+y会创建一个计算图,有三个节点(x,y和z)与两条边(z->x和z->y)。每条边都被creation_op标记为add,下图可以让你可以递归地反向传播梯度。

    image-20231205212517079

  • 当你调用z.backward()时,它基于计算z给定函数(add)为x和y传送正确的梯度。经过加法的反向传播意味着也要在反向传播时应用加法。在这个例子中,因为只有一个梯度要加到x或y,将梯度从z复制到x和y。

这种形式的autograd最优美的部分可能是它能很好地以递归方式工作,因为每个张量都在它的所有self.creators上调用了.backward()。

快速检查

曾经我们往一个方向进行前向传播,然后经过由激活函数构成的虚拟图向后传播。

你只是没有显式地把节点和边编码在一个图数据结构中,作为替代,现在我们创建了一个层(字典)列表,并手动编码了正确的前向与反向传播操作。现在,基于此搭建一个良好的接口让你可以不必再写那么多代码,这个接口让你递归地反向传播,而不用硬要手写复杂的反向传播代码。

注意:在前向传播中创建的图称为动态计算图,因为它是在前向传播过程中即时创建的。这是在较新的深度学习框架(如DyNet和PyTorch)中出现的autograd类型,较老的框架(如Theano和TensorFlow)使用的是静态计算图,它是在前向传播开始之前就指定好的。

多次使用的张量

Tensor的当前版本只支持向后传播到一个变量一次。但有时,在向前传播的过程中,你会使用同一个张量(神经网络的权重)多次,因此计算图的多个部分把梯度传播到同一个张量。但是在反向传播到一个用了多次的变量(是多个孩子的父节点)时,当前的代码会计算出错误的梯度:

a = Tensor([1,2,3,4,5])
b = Tensor([2,2,2,2,2])
c = Tensor([5,4,3,2,1])

d = a + b
e = b + c
f = d + e
f.backward(Tensor(np.array([1,1,1,1,1])))

b.grad.data == np.array([2,2,2,2,2])
array([False, False, False, False, False])

在这个例子中,变量b在创建f的过程中使用了两次。因此,它的梯度应该是两个导数的和:[2,2,2,2,2]。下面展示的是这一系列操作创建的计算图。注意:现在有两个指针指向b,所以它应当是来自e和d的梯度的和。

image-20231205215528817

不过当前的Tensor实现只是用前面的导数覆盖了每个导数。首先,b会得到来自d的梯度,然后它被来自e的梯度覆盖,我们需要改变梯度写入的方式。

升级autograd以支持多次使用的张量

增加一个新函数,并且更新三个旧函数。

import numpy as np

class Tensor (object):
    
    def __init__(self,data,
                 autograd=False,
                 creators=None,
                 creation_op=None,
                 id=None):
        
        self.data = np.array(data)
        self.autograd = autograd
        self.grad = None
        if(id is None):
            self.id = np.random.randint(0,100000)
        else:
            self.id = id
        
        self.creators = creators
        self.creation_op = creation_op
        self.children = {}
        
        if(creators is not None):
            for c in creators:
                if(self.id not in c.children):
                    c.children[self.id] = 1
                else:
                    c.children[self.id] += 1

    def all_children_grads_accounted_for(self):
        for id,cnt in self.children.items():
            if(cnt != 0):
                return False
        return True        
        
    def backward(self,grad=None, grad_origin=None):
        if(self.autograd):
            if(grad is None):
                grad = FloatTensor(np.ones_like(self.data))
            
            if(grad_origin is not None):
                if(self.children[grad_origin.id] == 0):
                    raise Exception("cannot backprop more than once")
                else:
                    self.children[grad_origin.id] -= 1

            if(self.grad is None):
                self.grad = grad
            else:
                self.grad += grad
            
            # grads must not have grads of their own
            assert grad.autograd == False
            
            # only continue backpropping if there's something to
            # backprop into and if all gradients (from children)
            # are accounted for override waiting for children if
            # "backprop" was called on this variable directly
            if(self.creators is not None and 
               (self.all_children_grads_accounted_for() or 
                grad_origin is None)):

                if(self.creation_op == "add"):
                    self.creators[0].backward(self.grad, self)
                    self.creators[1].backward(self.grad, self)
                    
    def __add__(self, other):
        if(self.autograd and other.autograd):
            return Tensor(self.data + other.data,
                          autograd=True,
                          creators=[self,other],
                          creation_op="add")
        return Tensor(self.data + other.data)

    def __repr__(self):
        return str(self.data.__repr__())
    
    def __str__(self):
        return str(self.data.__str__())  
    
a = Tensor([1,2,3,4,5], autograd=True)
b = Tensor([2,2,2,2,2], autograd=True)
c = Tensor([5,4,3,2,1], autograd=True)

d = a + b
e = b + c
f = d + e

f.backward(Tensor(np.array([1,1,1,1,1])))

print(b.grad.data == np.array([2,2,2,2,2]))
  • 使用条件判断一个张量有多少个子节点,如果传入的creators参数不为None,则遍历creators列表中的每个元素。对于每个元素c,检查self.id是否已经存在于c的children字典中。如果不存在,则将self.id添加到c的children字典中,并将其值设置为1;如果存在,则将c的children字典中self.id对应的值加1。
  • all_children_grads_accounted_for 函数是检查一个张量是否从每个子节点接收了正确数量的梯度,对于每个子类的计数,如果不等于0,则说明该子类的梯度还未计算完,直接返回False;如果遍历完所有子类后都没有返回False,则说明所有子类的梯度都已经计算完,返回True。
  • 在更新后的反向传播代码中:
    • if(self.autograd):判断是否开启自动求导功能,如果开启自动求导,则执行以下代码。
    • if(grad is None):如果grad参数为None,则将grad初始化为一个与self.data形状相同的全1的张量。
    • if(grad_origin is not None):如果grad_origin参数不为None,则检查grad_origin的id是否在self.children中,并且对应的计数器是否为0。如果计数器为0,则抛出异常;否则将计数器减1。检查确保你可以向后传播还是在等待一个梯度;在后一种情况下,减少计数器。
    • if(self.grad is None):累加来自于若干个子节点的梯度,即张量的相加。
    • assert grad.autograd == False:使用断言grad不具有自动求导功能。
    • if(self.creators is not None and (self.all_children_grads_accounted_for() or grad_origin is None))
      判断是否进行反向传播的条件,self.creators不为None,并且所有子变量的梯度都已经计算完毕,或者grad_origin为None。

添加更多函数支持

import numpy as np

class Tensor (object):
    
    def __init__(self,data,
                 autograd=False,
                 creators=None,
                 creation_op=None,
                 id=None):
        
        self.data = np.array(data)
        self.autograd = autograd
        self.grad = None
        if(id is None):
            self.id = np.random.randint(0,100000)
        else:
            self.id = id
        
        self.creators = creators
        self.creation_op = creation_op
        self.children = {}
        
        if(creators is not None):
            for c in creators:
                if(self.id not in c.children):
                    c.children[self.id] = 1
                else:
                    c.children[self.id] += 1

    def all_children_grads_accounted_for(self):
        for id,cnt in self.children.items():
            if(cnt != 0):
                return False
        return True 
        
    def backward(self,grad=None, grad_origin=None):
        if(self.autograd):
 
            if(grad is None):
                grad = Tensor(np.ones_like(self.data))

            if(grad_origin is not None):
                if(self.children[grad_origin.id] == 0):
                    raise Exception("cannot backprop more than once")
                else:
                    self.children[grad_origin.id] -= 1

            if(self.grad is None):
                self.grad = grad
            else:
                self.grad += grad
            
            # grads must not have grads of their own
            assert grad.autograd == False
            
            # only continue backpropping if there's something to
            # backprop into and if all gradients (from children)
            # are accounted for override waiting for children if
            # "backprop" was called on this variable directly
            if(self.creators is not None and 
               (self.all_children_grads_accounted_for() or 
                grad_origin is None)):

                if(self.creation_op == "add"):
                    self.creators[0].backward(self.grad, self)
                    self.creators[1].backward(self.grad, self)
                    
                if(self.creation_op == "sub"):
                    self.creators[0].backward(Tensor(self.grad.data), self)
                    self.creators[1].backward(Tensor(self.grad.__neg__().data), self)

                if(self.creation_op == "mul"):
                    new = self.grad * self.creators[1]
                    self.creators[0].backward(new , self)
                    new = self.grad * self.creators[0]
                    self.creators[1].backward(new, self)                    
                    
                if(self.creation_op == "mm"):
                    c0 = self.creators[0]
                    c1 = self.creators[1]
                    new = self.grad.mm(c1.transpose())
                    c0.backward(new)
                    new = self.grad.transpose().mm(c0).transpose()
                    c1.backward(new)
                    
                if(self.creation_op == "transpose"):
                    self.creators[0].backward(self.grad.transpose())

                if("sum" in self.creation_op):
                    dim = int(self.creation_op.split("_")[1])
                    self.creators[0].backward(self.grad.expand(dim,
                                                               self.creators[0].data.shape[dim]))

                if("expand" in self.creation_op):
                    dim = int(self.creation_op.split("_")[1])
                    self.creators[0].backward(self.grad.sum(dim))
                    
                if(self.creation_op == "neg"):
                    self.creators[0].backward(self.grad.__neg__())
                    
    def __add__(self, other):
        if(self.autograd and other.autograd):
            return Tensor(self.data + other.data,
                          autograd=True,
                          creators=[self,other],
                          creation_op="add")
        return Tensor(self.data + other.data)

    def __neg__(self):
        if(self.autograd):
            return Tensor(self.data * -1,
                          autograd=True,
                          creators=[self],
                          creation_op="neg")
        return Tensor(self.data * -1)
    
    def __sub__(self, other):
        if(self.autograd and other.autograd):
            return Tensor(self.data - other.data,
                          autograd=True,
                          creators=[self,other],
                          creation_op="sub")
        return Tensor(self.data - other.data)
    
    def __mul__(self, other):
        if(self.autograd and other.autograd):
            return Tensor(self.data * other.data,
                          autograd=True,
                          creators=[self,other],
                          creation_op="mul")
        return Tensor(self.data * other.data)    

    def sum(self, dim):
        if(self.autograd):
            return Tensor(self.data.sum(dim),
                          autograd=True,
                          creators=[self],
                          creation_op="sum_"+str(dim))
        return Tensor(self.data.sum(dim))
    
    def expand(self, dim,copies):

        trans_cmd = list(range(0,len(self.data.shape)))
        trans_cmd.insert(dim,len(self.data.shape))
        new_data = self.data.repeat(copies).reshape(list(self.data.shape) + [copies]).transpose(trans_cmd)
        
        if(self.autograd):
            return Tensor(new_data,
                          autograd=True,
                          creators=[self],
                          creation_op="expand_"+str(dim))
        return Tensor(new_data)
    
    def transpose(self):
        if(self.autograd):
            return Tensor(self.data.transpose(),
                          autograd=True,
                          creators=[self],
                          creation_op="transpose")
        
        return Tensor(self.data.transpose())
    
    def mm(self, x):
        if(self.autograd):
            return Tensor(self.data.dot(x.data),
                          autograd=True,
                          creators=[self,x],
                          creation_op="mm")
        return Tensor(self.data.dot(x.data))
    
    def __repr__(self):
        return str(self.data.__repr__())
    
    def __str__(self):
        return str(self.data.__str__())  
    
a = Tensor([1,2,3,4,5], autograd=True)
b = Tensor([2,2,2,2,2], autograd=True)
c = Tensor([5,4,3,2,1], autograd=True)

d = a + b
e = b + c
f = d + e

f.backward(Tensor(np.array([1,1,1,1,1])))

print(b.grad.data == np.array([2,2,2,2,2]))

使用autograd训练神经网络

以下是一个手写的反向传播的神经网络:

import numpy
np.random.seed(0)

data = np.array([[0,0],[0,1],[1,0],[1,1]])
target = np.array([[0],[1],[0],[1]])

weights_0_1 = np.random.rand(2,3)
weights_1_2 = np.random.rand(3,1)

for i in range(10):
    
    # Predict
    layer_1 = data.dot(weights_0_1)
    layer_2 = layer_1.dot(weights_1_2)
    
    # Compare
    diff = (layer_2 - target)
    sqdiff = (diff * diff)
    loss = sqdiff.sum(0) # mean squared error loss

    # Learn: this is the backpropagation piece
    layer_1_grad = diff.dot(weights_1_2.transpose())
    weight_1_2_update = layer_1.transpose().dot(diff)
    weight_0_1_update = data.transpose().dot(layer_1_grad)
    
    weights_1_2 -= weight_1_2_update * 0.1
    weights_0_1 -= weight_0_1_update * 0.1
    print(loss[0])

输出:

5.066439994622395
0.4959907791902342
0.4180671892167177
0.35298133007809646
0.2972549636567377
0.2492326038163328
0.20785392075862477
0.17231260916265176
0.14193744536652986
0.11613979792168384

但是有了新的autograd系统,代码就简单多了。你不必维护任何临时变量(因为动态图会追踪它们),而且也不必实现任何反向传播逻辑(因为.backward()方法会处理)。这样不仅更方便,而且更不容易在反向传播代码中犯愚蠢的错误,从而减少了出现bug的可能性:

import numpy as np
np.random.seed(0)

data = Tensor(np.array([[0,0],[0,1],[1,0],[1,1]]), autograd=True)
target = Tensor(np.array([[0],[1],[0],[1]]), autograd=True)

w = list()
w.append(Tensor(np.random.rand(2,3), autograd=True))
w.append(Tensor(np.random.rand(3,1), autograd=True))

for i in range(10):

    # Predict
    pred = data.mm(w[0]).mm(w[1])
    
    # Compare
    loss = ((pred - target)*(pred - target)).sum(0)
    
    # Learn
    loss.backward(Tensor(np.ones_like(loss.data)))

    for w_ in w:
        w_.data -= w_.grad.data * 0.1
        w_.grad.data *= 0

    print(loss)

增加自动优化

创建一个随机梯度下降(SGD)优化器

class SGD(object):
    
    def __init__(self, parameters, alpha=0.1):
        self.parameters = parameters
        self.alpha = alpha
    
    def zero(self): # 用于将模型参数清零
        for p in self.parameters:
            p.grad.data *= 0
        
    def step(self, zero=True): # 用于更新模型参数,遍历每个参数p
        
        for p in self.parameters:
            
            p.data -= p.grad.data * self.alpha
            
            if(zero):
                p.grad.data *= 0

如下所示,可以对前一个神经网络进一步简化,运行结果与前面一样:

import numpy
np.random.seed(0)

data = Tensor(np.array([[0,0],[0,1],[1,0],[1,1]]), autograd=True)
target = Tensor(np.array([[0],[1],[0],[1]]), autograd=True)

w = list()
w.append(Tensor(np.random.rand(2,3), autograd=True))
w.append(Tensor(np.random.rand(3,1), autograd=True))

optim = SGD(parameters=w, alpha=0.1)

for i in range(10):

    # Predict
    pred = data.mm(w[0]).mm(w[1])
    
    # Compare
    loss = ((pred - target)*(pred - target)).sum(0)
    
    # Learn
    loss.backward(Tensor(np.ones_like(loss.data)))
    optim.step()

    print(loss)

添加神经元层类型的支持

进一步的工作是将新函数添加到张量并创建方便的高阶类和函数,在几乎所有的框架中,最常见的抽象可能是神经元层抽象。它把一组常用的前向传播技术打包成一个简单的API,后者可用某种.forward()方法调用前者。以下是一个简单线性层的例子:

class Layer(object):
    
    def __init__(self):
        self.parameters = list()
        
    def get_parameters(self):
        return self.parameters


class Linear(Layer):

    def __init__(self, n_inputs, n_outputs):
        super().__init__()
        W = np.random.randn(n_inputs, n_outputs) * np.sqrt(2.0/(n_inputs))
        self.weight = Tensor(W, autograd=True)
        self.bias = Tensor(np.zeros(n_outputs), autograd=True)
        
        self.parameters.append(self.weight)
        self.parameters.append(self.bias)

    def forward(self, input):
        return input.mm(self.weight)+self.bias.expand(0,len(input.data))
  • 权重被组织成类(而且我添加了偏差权重(bias),因为这是一个真正的线性层)。你可以整体初始化这个神经元层,使得权重和偏差初始化到正确的大小,并且它总是采用正确的前向传播逻辑。
  • 创建了一个抽象类Layer,它有一个单独的getter。它可以支持更复杂的神经元层类型(比如包含其他神经元层的神经元层)。你只需要重载get_parameters()函数来控制什么样的张量之后会传到优化器(例如之前创建的SGD类)。

包含神经元层的神经元层

神经元层也可以包含其他层。

最普遍的神经元层是一种前向传播一组神经元层的顺序层,它的每一层都把输出传给下一层作为输入:


class Sequential(Layer):
    
    def __init__(self, layers=list()):
        super().__init__()
        
        self.layers = layers
    
    def add(self, layer):
        self.layers.append(layer)
        
    def forward(self, input):
        for layer in self.layers:
            input = layer.forward(input)
        return input
    
    def get_parameters(self):
        params = list()
        for l in self.layers:
            params += l.get_parameters()
        return params
    
import numpy
np.random.seed(0)

data = Tensor(np.array([[0,0],[0,1],[1,0],[1,1]]), autograd=True)
target = Tensor(np.array([[0],[1],[0],[1]]), autograd=True)

model = Sequential([Linear(2,3), Linear(3,1)])

optim = SGD(parameters=model.get_parameters(), alpha=0.05)

for i in range(10):
    
    # Predict
    pred = model.forward(data)
    
    # Compare
    loss = ((pred - target)*(pred - target)).sum(0)
    
    # Learn
    loss.backward(Tensor(np.ones_like(loss.data)))
    optim.step()
    print(loss)

损失函数层

有些层没有权重。

你也可以把输入的函数作为神经元层。这种神经元层最普遍的版本很有可能是损失函数层,如均方误差:

class MSELoss(Layer):
    
    def __init__(self):
        super().__init__()
    
    def forward(self, pred, target):
        return ((pred - target)*(pred - target)).sum(0)
    
import numpy
np.random.seed(0)

data = Tensor(np.array([[0,0],[0,1],[1,0],[1,1]]), autograd=True)
target = Tensor(np.array([[0],[1],[0],[1]]), autograd=True)

model = Sequential([Linear(2,3), Linear(3,1)])
criterion = MSELoss()

optim = SGD(parameters=model.get_parameters(), alpha=0.05)

for i in range(10):
    
    # Predict
    pred = model.forward(data)
    
    # Compare
    loss = criterion.forward(pred, target)
    
    # Learn
    loss.backward(Tensor(np.ones_like(loss.data)))
    optim.step()
    print(loss)

输出:

[2.33428272]
[0.06743796]
[0.0521849]
[0.04079507]
[0.03184365]
[0.02479336]
[0.01925443]
[0.01491699]
[0.01153118]
[0.00889602]

前面若干代码例子底层做的都是相同的运算,正是autograd在执行所有的反向传播,而前向传播步骤被打包进一个个精巧的类,保证其功能以正确顺序执行。

如何学习一个框架

简单而言,框架就是autograd加上一组预先定义好的神经元层和优化器。

你现在已经用底下的autograd系统相当快地写出多种神经元层类型,这使得把任意的神经元层组合在一起变得相当容易。坦率地说,这是现代框架的主要特性,它消除了为前向和后向传播手写每一个操作的必要。使用框架可以极大地加快从想法向实验转化的速度,并减少代码中的bug数量。

只是把框架看作与一大堆神经元层和优化器结合在一起的autograd系统可以帮助你学习它们。

非线性层

下面将非线性层加入Tensor,创建一些神经元层类型。

即将.sigmoid()和.tanh()函数添加到Tensor类中:

def sigmoid(self):
        if(self.autograd):
            return Tensor(1 / (1 + np.exp(-self.data)),
                          autograd=True,
                          creators=[self],
                          creation_op="sigmoid")
        return Tensor(1 / (1 + np.exp(-self.data)))

    def tanh(self):
        if(self.autograd):
            return Tensor(np.tanh(self.data),
                          autograd=True,
                          creators=[self],
                          creation_op="tanh")
        return Tensor(np.tanh(self.data))

以下代码展示了添加到 Tensor.backword() 方法的反向传播逻辑:

if(self.creation_op == "sigmoid"):
                   ones = Tensor(np.ones_like(self.grad.data))
                   self.creators[0].backward(self.grad * (self * (ones - self)))
               
if(self.creation_op == "tanh"):
                   ones = Tensor(np.ones_like(self.grad.data))
                   self.creators[0].backward(self.grad * (ones - (self * self)))

添加相应的类:

class Tanh(Layer):
    def __init__(self):
        super().__init__()
    
    def forward(self, input):
        return input.tanh()
    
class Sigmoid(Layer):
    def __init__(self):
        super().__init__()
    
    def forward(self, input):
        return input.sigmoid()

Tanh()Sigmoid()层放到Sequential()的参数列表中,而神经网络刚好知道如何用它们:

import numpy
np.random.seed(0)

data = Tensor(np.array([[0,0],[0,1],[1,0],[1,1]]), autograd=True)
target = Tensor(np.array([[0],[1],[0],[1]]), autograd=True)

model = Sequential([Linear(2,3), Tanh(), Linear(3,1), Sigmoid()])
criterion = MSELoss()

optim = SGD(parameters=model.get_parameters(), alpha=1)

for i in range(10):
    
    # Predict
    pred = model.forward(data)
    
    # Compare
    loss = criterion.forward(pred, target)
    
    # Learn
    loss.backward(Tensor(np.ones_like(loss.data)))
    optim.step()
    print(loss)
[1.06372865]
[0.75148144]
[0.57384259]
[0.39574294]
[0.2482279]
[0.15515294]
[0.10423398]
[0.07571169]
[0.05837623]
[0.04700013]

嵌入层

嵌入层把下标转换成激励信号。

交叉熵层

代码示例:

添加到Tensor:

def cross_entropy(self, target_indices):

        temp = np.exp(self.data)
        softmax_output = temp / np.sum(temp,
                                       axis=len(self.data.shape)-1,
                                       keepdims=True)
        
        t = target_indices.data.flatten()
        p = softmax_output.reshape(len(t),-1)
        target_dist = np.eye(p.shape[1])[t]
        loss = -(np.log(p) * (target_dist)).sum(1).mean()
    
        if(self.autograd):
            out = Tensor(loss,
                         autograd=True,
                         creators=[self],
                         creation_op="cross_entropy")
            out.softmax_output = softmax_output
            out.target_dist = target_dist
            return out

        return Tensor(loss)
class CrossEntropyLoss(object):
    
    def __init__(self):
        super().__init__()
    
    def forward(self, input, target):
        return input.cross_entropy(target)

进行预测与输出:

import numpy
np.random.seed(0)

# data indices
data = Tensor(np.array([1,2,1,2]), autograd=True)

# target indices
target = Tensor(np.array([0,1,0,1]), autograd=True)

model = Sequential([Embedding(3,3), Tanh(), Linear(3,4)])
criterion = CrossEntropyLoss()

optim = SGD(parameters=model.get_parameters(), alpha=0.1)

for i in range(10):
    
    # Predict
    pred = model.forward(data)
    
    # Compare
    loss = criterion.forward(pred, target)
    
    # Learn
    loss.backward(Tensor(np.ones_like(loss.data)))
    optim.step()
    print(loss)

输出:

1.3885032434928422
0.9558181509266037
0.6823083585795604
0.5095259967493119
0.39574491472895856
0.31752527285348264
0.2617222861964216
0.22061283923954234
0.18946427334830068
0.16527389263866668

这个损失函数与其他函数有一个明显的不同之处:最终的softmax与交叉熵损失的计算都在同一个类中,这种做法在深度神经网络中极为常见,几乎所有的神经网络都用这种方式工作。

当你想要结束一个网络并用交叉熵训练时,可以在前向传播中忽略softmax,而调用一个会自动调用softmax的交叉熵类,前者是后者的一部分。

这种一致的方式组合它们的原因是性能。在交叉熵函数中一起计算softmax和负指数相似性的梯度比在不同模块中分别对它们做前向与反向传播要快得多,这点与梯度运算的快捷算法有关。

递归神经网络层

通过组合若干层,就可以在时间序列上学习。

这种神经元层叫作递归层,你将用三个线性层来创建它,并且.forward()方法会接收前一个隐藏状态的输出和当前训练数据的输入。

class RNNCell(Layer):
    
    def __init__(self, n_inputs, n_hidden, n_output, activation='sigmoid'):
        super().__init__()

        self.n_inputs = n_inputs
        self.n_hidden = n_hidden
        self.n_output = n_output
        
        if(activation == 'sigmoid'):
            self.activation = Sigmoid()
        elif(activation == 'tanh'):
            self.activation == Tanh()
        else:
            raise Exception("Non-linearity not found")

        self.w_ih = Linear(n_inputs, n_hidden)
        self.w_hh = Linear(n_hidden, n_hidden)
        self.w_ho = Linear(n_hidden, n_output)
        
        self.parameters += self.w_ih.get_parameters()
        self.parameters += self.w_hh.get_parameters()
        self.parameters += self.w_ho.get_parameters()        
    
    def forward(self, input, hidden):
        from_prev_hidden = self.w_hh.forward(hidden)
        combined = self.w_ih.forward(input) + from_prev_hidden
        new_hidden = self.activation.forward(combined)
        output = self.w_ho.forward(new_hidden)
        return output, new_hidden
    
    def init_hidden(self, batch_size=1):
        return Tensor(np.zeros((batch_size,self.n_hidden)), autograd=True)
  • RNN有一个状态向量,用于将信息从一个时间步骤传递到下一个时间步骤:在这个例子中,它是hidden变量,作为forward函数的输入输出变量。

  • RNN 还有若干不同的权重矩阵:一个把输入向量映射到隐藏向量(处理输入数据),一个把隐藏向量映射到隐藏向量(即根据前一个隐藏向量更新当前的),以及可能会有一个隐藏-输出层基于隐藏向量做出预测。这里的 RNNCell 实现包含了所有三种。

  • self.w_ih 层是输入-隐藏层, self.w_hh 是隐藏-隐藏层, 还有 self.w_ho 是隐藏 -输出层。

    注意观察每种层的维度。self.w_ih 的输入大小和 self.w_ho 的输出大小都是词汇表的大小。所有其他维度都是基于参数 n_hidden 确定的。


文章作者: QT-7274
版权声明: 本博客所有文章除特別声明外,均采用 CC BY 4.0 许可协议。转载请注明来源 QT-7274 !
评论
  目录