深刻理解神经网络计算原理

class FC:
    
    def __init__(self,inputS:int=64,hidWidth:int=64,outputS:int=10,active:str="relu",preCacheSize:int = 1):
        '''
        只支持单隐藏层的全连接网络

        Parameters
        ----------
        inputS : int, optional
            输入维度. The default is 64.
        hidWidth : int, optional
            隐藏宽度. The default is 64.
        outputS : int, optional
            输出维度. The default is 10.
        active : string, optional
            激活函数. The default is "ReLU".

        Returns
        -------
        None.

        '''
        self.inputCache = np.zeros(shape=inputS)
        
        self.param = [  # 参数list，每个小list代表一个层的权重和偏置
                        [], # 空列表，用来统一化id序号
                        #'''第一层'''
                        [ np.random.normal(size=(hidWidth,inputS)),     #权重
                          np.zeros(shape=(hidWidth,1))], #偏置
                        #'''输出层/第二层'''
                        [ np.random.normal(size=(outputS,hidWidth)), 
                          np.zeros(shape=(outputS,1))],
            ]
        
        '''随batchSize动态变化，有待优化'''
        self.cache = [  # 缓存list，每个小list代表一个层的中间变量
                        #'''输入层'''
                        [np.zeros(shape=(inputS,preCacheSize)),    # 数据的X
                         np.zeros(shape=(outputS,preCacheSize))],  # 数据的Y
                        #'''第一层'''
                        [ np.zeros(shape=(hidWidth,preCacheSize)), # 线性计算结果
                         np.zeros(shape=(hidWidth,preCacheSize))], # 激活（非线性）结果
                        #'''输出层/第二层'''
                        [ np.zeros(shape=(outputS,preCacheSize)),
                         np.zeros(shape=(outputS,preCacheSize))]
            ]
        self.grad = copy.deepcopy(self.param)   # 梯度缓存的大小和可训练参数是一样多的
        
        # 统一激活函数
        self.activeFunc = active
        
        print(f"创建全连接网络成功，结构：{inputS}->{hidWidth}->{outputS}")
    
    def __ReLU(self,layerID:int,forward:bool=True):
        '''
        激活函数，支持正反向
        '''
        if forward:        
            self.cache[layerID][1] = np.maximum(0,self.cache[layerID][0]) # 维度会自动扩张
            return True
        else:   # 反向传播
            return (self.cache[layerID][1] > 0).astype(dtype=np.int0)
    
    def __Sigmoid(self,layerID:int,forward:bool=True):
        '''
        激活函数，支持正反向
        '''
        if forward:
            self.cache[layerID][1] = 1/(1+np.exp(np.subtract(0,self.cache[layerID][0])))
            return True
        else:
            return self.cache[layerID][1]*(1-self.cache[layerID][1])
    
    def __softmax(self,inputD:np.ndarray):
        '''
        激活函数，支持正向
        '''
        __DEBUG_SOFTMAX__ = False
        if __DEBUG_SOFTMAX__:
            print(f"inputD : {inputD.shape}")
        inputD = np.exp(inputD) # ~ [m_2,N]
        if __DEBUG_SOFTMAX__:
            print(f"inputD.exp : {inputD.shape}")
        sumD = np.sum(inputD,axis=0,keepdims=True)  # ~ [1,N]
        if __DEBUG_SOFTMAX__:
            print(f"sumD : {sumD.shape}")
        return inputD/sumD  # ~ [m_2,N]
    
    def forward(self,inputX:np.ndarray,inputY:np.ndarray,Debuging:bool = False):
        '''
        前向传播方法

        Parameters
        ----------
        inputX : np.ndarray
            输入样本数据，支持mini-Batch.（注意维度为：[dataSize,BatchSize]
        inputY : np.ndarray
            输入标签数据，支持mini-Batch.（注意维度为：[labelSize,BatchSize]
        Debuging : bool, optional
            调试打印使能，使用时不开启！. The default is False.

        Returns
        -------
        np.ndarray
            网络的输出层数据.

        '''
        self.cache[0][0] = inputX   # 存储数据，用于后续反向传播 ~ [m_0,N]
        self.cache[0][1] = inputY   # ~ [classes,N]
        if Debuging:
            print(f"输入数据大小：{self.cache[0][0].shape}，输入标签的大小：{self.cache[0][1].shape}")
        
        # 第一个线性层
        self.cache[1][0] = self.param[1][0] @ self.cache[0][0] + self.param[1][1]
        # ~ [m_1,N] = [m_1,m_0] @ [m_0,N] + [m_1,1]
        
        # 激活
        if self.activeFunc == "relu":
            self.__ReLU(1,forward=True) # ~ [m_1,N]
        elif self.activeFunc == "sigmoid":
            self.__Sigmoid(1,forward=True)  # ~ [m_1,N]
        else:
            print("activate error!")
            return    
        
        # 输出层
        self.cache[2][0] = self.param[2][0] @ self.cache[1][1] + self.param[2][1]
        # ~ [m_2,N] = [m_2,m_1] @ [m_1,N] + [m_2,1]
        # 激活
        self.cache[2][1] = self.__softmax(self.cache[2][0])
        # ~ [m_2,N]
        
        return self.cache[2][1]
    
    def backward(self):
        '''
        反向传播
        求各个参数层的梯度
        '''
        '--- 倒1层（softmax输出层） ---'
        # 损失函数对 softmax激活前&最后一个线性层后 的偏导
        dL_dz = (self.cache[2][1] - self.cache[0][1]) / (self.cache[0][1].shape[1]) # 预测值向量减去真实值的one-hot向量
        # [m2,N] = ( [m2,N] - [m2,N] ) /N
        
        # 损失函数对第二层（最后）可训练参数偏导
        self.grad[2][0] = dL_dz @ self.cache[1][1].T   # 损失对w偏导
        # ~ [m2,m1] = [m2,N] @ [m1,N]^T
        self.grad[2][1] = np.sum(dL_dz,axis=1,keepdims=True) # 损失对b偏导
        # ~ [m_2,1] = sum([m_2,N])
        
        '--- 倒2层 ---'
        # 损失函数对 第一个隐藏层的激活输出 的偏导     
        dL_dz = (dL_dz.T @ self.param[2][0]).T 
        # ~ [m1,N] = ([m_2,N]^T @ [m_2,m_1])^T
        # 损失函数对 第一个隐藏层的激活前&线性输出 的偏导
        if self.activeFunc == "relu":
            dL_dz = dL_dz * self.__ReLU(1,forward=False)  # ~ [m_1,N] * [m_1,N]
        elif self.activeFunc == "sigmoid":
            dL_dz = dL_dz * self.__Sigmoid(1,forward=False)
        else:
            print("active error in backforward")
            return False
    
        # 损失函数对第一层可训练参数偏导
        self.grad[1][0] =dL_dz @ self.cache[0][0].T   # 损失对w偏导
        self.grad[1][1] = np.sum(dL_dz,axis=1,keepdims=True) # 损失对b偏导
        
        '--- 结束 ---'
        return True
    
    def SGD(self,lr:int = 0.001):
        '''
        随机梯度下降法，固定学习率更新权重和偏置

        Parameters
        ----------
        lr : int, optional
            学习率. The default is 0.001.

        Returns
        -------
        None.

        '''
        for layer in range(2):
            for typ in range(2):
                self.param[layer+1][typ] -= lr * self.grad[layer+1][typ]
    
    def print(self):
        '''
        打印网络结构信息
        '''
        print("---层结构：")
        print(f"第一层：权重{self.param[1][0].shape}，偏置{self.param[1][1].shape}")
        print(f"第二层：权重{self.param[2][0].shape}，偏置{self.param[2][1].shape}")
        print("---梯度结构：")
        print(f"第一层：权重{self.grad[1][0].shape}，偏置{self.grad[1][1].shape}")
        print(f"第二层：权重{self.grad[2][0].shape}，偏置{self.grad[2][1].shape}")
        print("---缓存量：")
        print(f"输入层：{self.cache[0][0].shape}和{self.cache[0][1].shape}")
        print(f"第一层：权重{self.cache[1][0].shape}，偏置{self.cache[1][1].shape}")
        print(f"第二层：权重{self.cache[2][0].shape}，偏置{self.cache[2][1].shape}")
    

# 主程序
if __name__ == "__main__":
    # 参数设置
    EPOCH = 100
    BATCHSIZE = 32
    LEARNRATE = 0.001
    
    HIDLAYERWIDTH = 128
    ACTIVATEFUNC = "relu"
    
    # 数据记录
    writer = tensorboard.SummaryWriter('./log/MNIST-FC'+str(HIDLAYERWIDTH)+ACTIVATEFUNC)
    
    # 数据集加载
    transformer = torchvision.transforms.Compose([
                        torchvision.transforms.ToTensor(),
                        torchvision.transforms.Normalize(
                            (0.1307,), (0.3081,) )
                         ])
    
    trainDataset = torchvision.datasets.MNIST(root="./Data",    
                                           train = True,
                                           download= True,
                                           transform= transformer
                                           )
    trainLoader = torch.utils.data.DataLoader(trainDataset,
                                              batch_size= BATCHSIZE,    # Batch Size！
                                              shuffle= True,
                                              num_workers=2)
    
    testDataset = torchvision.datasets.MNIST(root="./Data",
                                           train = False,
                                           download= True,
                                           transform= transformer)
    
    testLoader = torch.utils.data.DataLoader(testDataset,
                                              batch_size= BATCHSIZE,    # Batch Size！
                                              shuffle= True,
                                              num_workers=2)
    testIter = iter(testLoader)
    
    trainD = trainDataset.data.numpy()
    trainD = trainD.reshape(trainD.shape[0],-1).T
    trainD = trainD / 256
    trainT = trainDataset.targets.numpy().T
    
    testD = testDataset.data.numpy()
    testD = testD.reshape(testD.shape[0],-1).T
    testD = testD / 256
    testT = testDataset.targets.numpy().T
    
    samp = next(iter(trainLoader))
    samp[0] = samp[0].numpy(); samp[1] = samp[1].numpy()
    for i in range(0, 4):
        plt.figure(i)
        plt.imshow(samp[0][i][0])
        plt.title(str(samp[1][i]))

    # 创建模型
    model = FC(inputS=28*28,
               hidWidth=HIDLAYERWIDTH,
               outputS=10,
               active=ACTIVATEFUNC)
    
    # 训练模型
    Log_id = 1
    for ep in range(1,EPOCH+1):
        # 一次训练
        print(f"-------- {ep}'th epoch begin --------")
        for idx,(sampX,sampY) in enumerate(trainLoader):
            # 数据维度预处理
            sampX = sampX.numpy()
            sampX = sampX.reshape(sampX.shape[0],-1).T
            sampX = sampX / 256
            
            sampY_vector = torch.nn.functional.one_hot(sampY,num_classes = 10).numpy().T
            # 前向传播
            pred =  model.forward(sampX, sampY_vector)
            # 反向传播
            model.backward()
            
            
            # SGD
            model.SGD(lr = LEARNRATE)
            
            # 数据打印
            if idx %100 == 0:
                
                # 损失计算
                loss = Loss_CrossEntropy(pred,sampY_vector)
                print(f"第{ep}的{idx}次Batch交叉熵损失:{loss}")
                
                # 准确度计算
                testX,testY = testIter.next()
                testX = testX.numpy()
                testX = testX.reshape(testX.shape[0],-1).T
                testX = testX / 256
                if Log_id % 100 == 0:
                    testIter = iter(testLoader)
                
                
                acc_train = model_accuracyTest(model, sampX, sampY.numpy())
                acc_test = model_accuracyTest(model, testX, testY.numpy())
                print(f"训练集准确度:{acc_train*100}%,测试集准确度:{acc_test*100}%")
                
                # 记录数据
                writer.add_scalar("测试集准确度", acc_test,Log_id)
                writer.add_scalar("训练集准确度", acc_train,Log_id)
                writer.add_scalar("交叉熵损失", loss,Log_id)
                Log_id +=1
        

def Loss_CrossEntropy(pred:np.ndarray, real:np.ndarray):
    # pred ~ [classes,N] , real ~ [classes,N]
    
    entropy = real * np.log(pred +1e-10)    # 1e-10防止无穷产生
    
    entropy = -np.sum(entropy) / real.shape[-1]
    return entropy

def model_accuracyTest(model:FC, inputD:np.ndarray, target:np.ndarray) -> float:
    
    res = model.forward(inputD, target) # 模型推导
    
    res = np.argmax(res,axis=0) # 选出概率最大的类型
    
    acc = np.mean(np.equal(res,target))

import numpy as np
import torch,torchvision

import matplotlib.pyplot as plt

import copy

import torch.utils.tensorboard as tensorboard