神经网络进阶-用python实现一个完整的神经网络框架并在CIFAR10数据集上调参

Posted 2021-01-11 Super_JJboom

　　上一个博客中讲解了用python实现一个简单的两层神经网络，我们是把所有的网络层都直接写在了类中。但是作为一个神经网络框架，网络的结构应该是可以由使用者自定义的，这样一来也就不用为每个网络结构都重写所有代码，我们把每一层模块化，在神经网络的类中定义结构时使用这些模块化的层堆叠形成一个完整的神经网络。每一种层，分别实现forward和password两个函数，用来正向计算和反向传播。

这里我们实现的网络层和功能有：全连层、激活层、计算loss、自动训练

1、全连层

全连层的正向计算很简单，f(W,x)=xW+b。

反向传播分别求x,W,b的梯度，dx=f(W,x)·W.T,dW=x·f(W,x),db=f(W,x)。

 1 def affine_forward(x, w, b):
 2     #######################################
 3     #  x: input      shape (N, d_1, ..., d_k)
 4     #  w: Weights    shape (D, M)
 5     #  b: bias       shape (M)
 6     #
 7     #  Returns :
 8     #  out: output,shape (N, M)
 9     #  cache: (x, w, b)
10     ########################################
11     x1=x.reshape(x.shape[0],-1)
12     out=np.dot(x1,w)+b
13     cache=(x,w)
14     return out,cache
15 
16 def affine_backward(dout, cache):
17     ##################################################
18     # dout: Upstream derivative, of shape (N, M)
19     # cache: Tuple of:
20     #  x: Input   shape (N, d_1, ... d_k)
21     #  w: Weights shape (D, M)
22     #
23     # Returns a tuple of:
24     #  dx: shape (N, d1, ..., d_k)
25     #  dw: shape (D, M)
26     #  db: shape (M,)
27     ##################################################
28     x,w=cache
29     N=x.shape[0]
30     x1=x.reshape(N,-1)
31     dx=np.dot(dout,w.T).reshape(*x.shape)
32     dw=np.dot(x1.T,dout)
33     db=np.sum(dout,axis=0)
34     return dx,dw,db

View Code

2、激活层

激活层只实现常用的relu激活函数。f(x)=max(0,x)。

反向传播也很简单，被激活的继承上一级的梯度，没有激活的梯度为0。这里不涉及到参数的梯度计算，只涉及到梯度的传播。

 1 def relu_forward(x):
 2     out = np.maximum(0,x)
 3     cache = x
 4     return out, cache
 5 
 6 
 7 def relu_backward(dout, cache):
 8     x = cache
 9     dx = dout * (x>0)
10     return dx

View Code

实际使用过程中，一般全连接层后面都会接激活层，为了方便起见，可以合并两层。这里的合并跟之后的整个神经网络的堆叠思想一致。

 1 def affine_relu_forward(x, w, b):
 2     """
 3     全连接层和激活层的合并
 4 
 5     Inputs:
 6     - x: 全连接层的输入
 7     - w, b: 全连接层的权重参数
 8 
 9     Returns a tuple of:
10     - out: 被激活的全连接层的输出
11     - cache: 用于反向传播
12     """
13     a,fc_cache=affine_forward(x,w,b)
14     out,relu_cache=relu_forward(a)
15     return out, (fc_cache,relu_cache)
16 
17 
18 def affine_relu_backward(dout, cache):
19     fc_cache, relu_cache = cache
20     da=relu_backward(dout,relu_cache)
21     dx,dw,db=affine_backward(da,fc_cache)
22     return dx, dw, db

View Code

3、loss层

严格上讲这不算神经网络的一个层，只是为了训练而必须进行的一个计算，但是这里我们就把它也当作一个层好了。

loss函数在上一个博客中已经详细介绍过了。https://www.cnblogs.com/super-JJboom/p/9748468.html

 1 def svm_loss(x, y):
 2     
 3     N,C=x.shape
 4     correct_class_scores=x[range(N),y].reshape(-1,1)
 5     margins=np.maximum(0,x-correct_class_scores+1)
 6     loss=np.sum(margins)/N
 7     dx = np.zeros_like(x)
 8     dx[margins>0]=1
 9     num_pos = np.sum(margins > 0, axis=1)
10     dx[range(N),y]-=num_pos
11     dx/=N
12     
13     return loss, dx
14 
15 
16 def softmax_loss(x, y):
17     
18     N,C=x.shape
19     shift_x=x-np.max(x,axis=1,keepdims=True).reshape(-1,1)
20     Z=np.sum(np.exp(shift_x),axis=1,keepdims=True)
21     log_pro=-shift_x+np.log(Z)
22     loss=np.sum(log_pro[range(N),y])/N
23     probs=np.exp(-log_pro)
24     probs[range(N),y]-=1
25     dx=probs
26     dx/=N

View Code

到此为止，之前我们实现两层神经网络需要的层已经都实现了。先重构一下之前的实现吧。

 1 #2层神经网络
 2 class TwoLayerNet(object):
 3     #The architecure : affine - relu - affine - softmax.
 4     def __init__(self, input_dim=3*32*32,hidden_dim=100,num_classes=10,weight_scale=1e-3,reg=0.0):
 5         ################################
 6         # input_dim      输入维度
 7         # hidden_dims    隐藏层神经元个数
 8         # num_classes    输出个数
 9         # weight_scale   初始化权重
10         # reg            正则项系数
11         ################################
12         self.params={}
13         self.reg=reg
14         self.params[\'W1\']=weight_scale*np.random.randn(input_dim,hidden_dim)
15         self.params[\'b1\']=np.zeros(hidden_dim)
16         self.params[\'W2\']=weight_scale*np.random.randn(hidden_dim,num_classes)
17         self.params[\'b2\']=np.zeros(num_classes)
18 
19     def loss(self,X,y=None):
20         #返回loss和grad
21 
22         #前向计算
23         ar1_out,ar1_cache=affine_relu_forward(X,self.params[\'W1\'],self.params[\'b1\'])
24         a2_out,a2_cache=affine_forward(ar1_out,self.params[\'W2\'],self.params[\'b2\'])
25         scores=a2_out
26 
27         if y is None:
28             return scores
29 
30         loss,grads=0,{}
31         loss,dscores=softmax_loss(scores,y)
32         loss=loss+0.5*self.reg*(np.sum(self.params[\'W1\']**2)+np.sum(self.params[\'W2\']**2))
33         dx2,dw2,db2=affine_backward(dscores,a2_cache)
34         grads[\'W2\']=dw2+self.reg*self.params[\'W2\']
35         grads[\'b2\']=db2
36 
37         dx1,dw1,db1=affine_relu_backward(dx2,ar1_cache)
38         grads[\'W1\']=dw1+self.reg*self.params[\'W1\']
39         grads[\'b1\']=db1
40 
41         return loss,grads

View Code

看起来比之前的实现并没有简单多少。。。这是因为2层神经网络的结构过于简单，仅从代码量上来看并没有减少，但是对于后面要实现的更复杂的神经网络来说，就发挥了巨大的作用。

哦，对比之前的实现，发现少了自动化训练的实现。因为训练有很多参数可以选择和调节，之前没有实现，如果全部放入神经网络的类中的话会显得过于臃肿，所以把训练过程的实现单独拿出来作为一个类。

4、自动化训练

相比与之前的自动训练过程，这里增加了更多的可选项。可以选择优化方法，如：SGD，带动量的SGD，adam。每一轮数据迭代完之后显示数据。

  1 import numpy as np 
  2 from cs231n import optim
  3 
  4 class Solver(object):
  5     def __init__(self,model,data,**kwargs):
  6         \'\'\'
  7         初始化对象
  8         inputs:
  9         - model:网络结构对象
 10         - data:字典，包含带标签的训练集和验证集
 11         - kwargs:可选参数，详细见下面提取时候的注释
 12         \'\'\'
 13 
 14         self.model=model
 15         self.X_train=data[\'X_train\']
 16         self.y_train=data[\'y_train\']
 17         self.X_val=data[\'X_val\']
 18         self.y_val=data[\'y_val\']
 19 
 20         #解读kwargs
 21         self.update_rule=kwargs.pop(\'update_rule\',\'sgd\') #优化方法的选择，默认为随机梯度下降
 22         self.optim_config=kwargs.pop(\'optim_config\',{})  #优化的参数，学习率是必须有的选项。其他可以有动量因子之类的参数
 23         self.lr_decay=kwargs.pop(\'lr_decay\',1.0)         #学习率衰减因子，默认不衰减
 24         self.batch_size=kwargs.pop(\'batch_size\',128)     #批大小，默认128
 25         self.num_epochs=kwargs.pop(\'num_epochs\',10)      #训练全部数据的轮次，默认为10轮
 26         self.print_every=kwargs.pop(\'print_every\',10)    #多少轮显示一次进度
 27         self.verbose=kwargs.pop(\'verbose\',True)          #是否显示进度，为false的情况下上一个参数无效
 28 
 29         #含有不支持的参数
 30         if len(kwargs)>0:
 31             extra=\',\'.join(\'"%s"\' % k for k in kwargs.keys())
 32             raise ValueError(\'Unrecongnized arguments %s\' %extra)
 33 
 34         #检查优化方法是否支持
 35         if not hasattr(optim,self.update_rule):
 36             raise ValueError(\'invalid update_rule "%s"\' %self.update_rule)
 37 
 38         self.update_rule=getattr(optim,self.update_rule)
 39         
 40         self._reset() 
 41 
 42     def _reset(self):
 43         #初始化参数
 44         self.epoch=0
 45         self.best_val_acc=0
 46         self.best_params={}
 47         self.loss_history=[]
 48         self.train_acc_history=[]
 49         self.val_acc_history=[]
 50 
 51         #给给个参数矩阵复制一个优化参数，因为之后每个权重的参数不相同，需要自己保存
 52         self.optim_configs={}
 53         for p in self.model.params:
 54             d={k:v for k,v in self.optim_config.items()}
 55             self.optim_configs[p]=d
 56 
 57     def _step(self):
 58         #单步更新
 59 
 60         #随机取出batchsize个数据
 61         num_train=self.X_train.shape[0]
 62         batch_mask=np.random.choice(num_train,self.batch_size)
 63         X_batch=self.X_train[batch_mask]
 64         y_batch=self.y_train[batch_mask]
 65 
 66         #计算loss
 67         loss,grads=self.model.loss(X_batch,y_batch)
 68         self.loss_history.append(loss)
 69 
 70         #更新参数
 71         for p,w in self.model.params.items():
 72             dw=grads[p]
 73             config=self.optim_configs[p]
 74             next_w,next_config=self.update_rule(w,dw,config)
 75             self.model.params[p]=next_w
 76             self.optim_configs[p]=next_config
 77 
 78     #计算正确率
 79     def check_accuracy(self,X,y,num_samples=None,batch_size=128):
 80             N=X.shape[0]
 81 
 82             #如果num_sample不为空 则只从全部数据中选则num_sample个数据计算
 83             if num_samples is not None and N>num_samples:
 84                 mask=np.random.choice(N,num_samples)
 85                 N=num_samples
 86                 X=X[mask]
 87                 y=y[mask]
 88 
 89             num_batches=N//batch_size
 90             if N%batch_size!=0:
 91                 num_batches+=1
 92             y_pred=[]
 93             for i in range(num_batches):
 94                 start=i*batch_size
 95                 end=(i+1)*batch_size
 96                 scores=self.model.loss(X[start:end])
 97                 y_pred.append(np.argmax(scores,axis=1))
 98             y_pred=np.concatenate(y_pred,axis=0)
 99             acc=np.mean(y_pred==y)
100 
101             return acc
102 
103     def train(self):
104         num_train=self.X_train.shape[0]
105         iterations_per_epoch=max(num_train//self.batch_size,1)
106         num_iterations=self.num_epochs*iterations_per_epoch
107 
108         for t in range(num_iterations):
109             self._step()
110 
111             if self.verbose and t%self.print_every==0:
112                 print(\'Iteration %d /%d loss: %f\' %(t+1,num_iterations,self.loss_history[-1]) )
113 
114             #每个epoch执行相应操作
115             epoch_end=(t+1)%iterations_per_epoch==0
116             if epoch_end:
117                 self.epoch+=1
118                 for k in self.optim_configs:
119                     self.optim_configs[k][\'learning_rate\']*=self.lr_decay
120 
121             first_it=(t==0)
122             last_it=(t==num_iterations-1)
123             if first_it or last_it or epoch_end:
124                 train_acc=self.check_accuracy(self.X_train,self.y_train,num_samples=1280)
125                 val_acc=  self.check_accuracy(self.X_val  ,self.y_val)
126                 self.train_acc_history.append(train_acc)
127                 self.val_acc_history.append(val_acc)
128 
129                 #可视化进度
130                 if self.verbose:
131                     print (\'(Epoch %d / %d) train acc: %f; val_acc: %f\' % (
132                      self.epoch, self.num_epochs, train_acc, val_acc))
133 
134                 #检查、保存模型
135                 if val_acc>self.best_val_acc:
136                     self.best_val_acc=val_acc
137                     self.best_params={}
138                     for k,v in self.model.params.items():
139                         self.best_params[k]=v.copy()
140 
141             self.model.params=self.best_params

View Code

实现到这里，已经可以重新训练之前的两层神经网络了，训练代码全部整合带最后的测试代码里面了。

5、实现全连层神经网络框架

实现跟两层神经网络区别不大，只是网络层的堆叠使用了循环。这里还没有实现的批归一化和dropout操作后面会讲到。

  1 class FullyConnectedNet(object):
  2     #archtecture: {affine - [batch norm] - relu - [dropout]} x (L - 1) - affine - softmax
  3     def __init__(self,hidden_dims,input_dim=3*32*32,num_classes=10,dropout=0,
  4                  use_batchnorm=False,reg=0.0,weight_scale=1e-3,dtype=np.float32,seed=None):
  5 
  6         \'\'\'
  7         inputs:
  8         - hidden_dims:list,存储了有多少个中间层，每一层有多少个神经元
  9         - input_dim: 输入数据的维度大小
 10         - num_classes:类别的个数，也就是最后一层的神经元个数
 11         - dropout:失活率
 12         - use_batchnorm:是否在每一层之间使用批归一化操作
 13         - reg:正则权重
 14         - weight_scale:权重矩阵的初始数量级
 15         - seed:失活率随机
 16         \'\'\'
 17 
 18         self.use_batchnorm=use_batchnorm
 19         self.use_dropout=(dropout>0)
 20         self.reg=reg
 21         self.num_layers=1+len(hidden_dims)
 22         self.dtype=dtype
 23         self.params={}
 24 
 25         #初始化每层的参数w,b [gamma,beta,dropout]（如果有的话）
 26         layer_input=input_dim
 27         for i,hd in enumerate(hidden_dims):
 28             self.params[\'W%d\'%(i+1)]=weight_scale*np.random.randn(layer_input,hd)
 29             self.params[\'b%d\'%(i+1)]=weight_scale*np.zeros(hd)
 30             if self.use_batchnorm:
 31                 self.params[\'gamma%d\'%(i+1)]=np.ones(hd)
 32                 self.params[\'beta%d\'%(i+1)]=np.zeros(hd)
 33             layer_input=hd
 34         self.params[\'W%d\'%(self.num_layers)]=weight_scale*np.random.randn(layer_input,num_classes)
 35         self.params[\'b%d\'%(self.num_layers)]=weight_scale*np.zeros(num_classes)
 36         for k,v in self.params.items():
 37             self.params[k]=v.astype(dtype)
 38 
 39         self.dropout_param={}
 40         if self.use_dropout:
 41             self.dropout_param={\'mode\':\'train\',\'p\':dropout}
 42             if seed is not None:
 43                 self.dropout_param[\'seed\']=seed
 44 
 45         self.bn_params=[]
 46         if self.use_batchnorm:
 47             self.bn_params=[{\'mode\':\'train\'} for i in range(self.num_layers-1)]
 48 
 49     def loss(self,X,y=None):
 50         
 51         #跟之前一样，y=None时表示测试过程，直接返回最后一层的输出即可。否则表示训练过程，还要计算loss和gradient。
 52 
 53         X=X.astype(self.dtype)
 54         mode=\'test\' if y is None else \'train\'
 55         
 56         if self.dropout_param is not None:
 57             self.dropout_param[\'mode\'] = mode   
 58         if self.use_batchnorm:
 59             for bn_param in self.bn_params:
 60                 bn_param[\'mode\'] = mode
 61 
 62 
 63         #forward pass
 64         layer_input=X
 65         ar_cache={}
 66         dp_cache={}
 67 
 68         for lay in range(self.num_layers-1):
 69             if self.use_batchnorm:
 70                 layer_input, ar_cache[lay] = affine_bn_relu_forward(layer_input, 
 71                                         self.params[\'W%d\'%(lay+1)], self.params[\'b%d\'%(lay+1)], 
 72                                         self.params[\'gamma%d\'%(lay+1)], self.params[\'beta%d\'%(lay+1)], self.bn_params[lay])
 73             else:
 74                 layer_input,ar_cache[lay]=affine_relu_forward(layer_input,self.params[\'W%d\'%(lay+1)],self.params[\'b%d以上是关于神经网络进阶-用python实现一个完整的神经网络框架并在CIFAR10数据集上调参的主要内容，如果未能解决你的问题，请参考以下文章
 
 Python进阶三部曲网络编程
 Python进阶开发之网络编程,socket实现在线聊天机器人
 [8421论坛]2019 Socket网络编程入门到进阶与实战完整视频教程
 SOCKET编程进阶
 Python_DL_麦子学院（算法与应用_进阶）_14~20 _Cross entropy函数
 深度学习理论与实战PyTorch实现