动手写深度学习框架（一）

class Tensor:
    def __init__(self,data,_parents=()):
        self.data = data                # 节点的值
        self.grad = 0                   # 梯度
        self._backward = lambda: None   # 初始化反向传播，等效为一个空函数
        self._parents = set(_parents)   # 节点的父节点
        
    def __repr__(self):  # 优化输出
        return f'Tensor(data={self.data})'
    
    def __add__(self, other):   # 实现加法运算
        out = Tensor(self.data + other.data, (self,other))        def _backward():
            self.grad += 1.0 * out.grad
            other.grad += 1.0 * out.grad
        out._backward = _backward   # 定义输出节点的反向传播函数
        return out    
    def __mul__(self, other):   # 实现乘法运算
        out = Tensor(self.data * other.data, (self,other))        def _backward():
            self.grad += other.data * out.grad
            other.grad += self.data * out.grad
        out._backward = _backward        return out    
    def relu(self):             # 实现 ReLu 运算, 激活函数
        out = Tensor(0 if self.data<0 else self.data, (self,))        def _backward():
            self.grad += (self.data>0)*out.grad 
        out._backward = _backward        return out    def backward(self):         # 实现所有节点的反向传播，计算梯度
        # 广搜 确定节点的拓扑排序
        topo = []
        vis = set()        def build_topo(v):
            if v not in vis:
                v.grad = 0.0
                vis.add(v)
                topo.append(v)                for parent in v._parents:
                    build_topo(parent)
        build_topo(self)        # 反向传播
        self.grad = 1.0
        for v in topo:
            v._backward()

# 测试梯度求解a = Tensor(2)
b = Tensor(3)
x = a*b
z = x + x
z.grad = 1      # 由于求解父节点的梯度依赖于子节点梯度，所以应该讲 z.grad 初始化为 1z._backward()   # 求 z 父节点 x 的梯度x._backward()   # 求 x 父节点 a,b 的梯度a.grad, b.grad, x.grad, z.grad

# 测试 backward()a = Tensor(2)
b = Tensor(3)
x = a*b
z = x + x
z.grad = 1      # z.grad 初始化为 1z.backward()
a.grad, b.grad, x.grad, z.grad

import randomclass Linear:
    def __init__(self, in_features, out_features):
        self.in_features = in_features
        self.out_features = out_features
        self.w = [[Tensor(random.random())] * out_features for _ in range(in_features)]
        self.b = [Tensor(random.random())] * out_features        
    def __call__(self, x):
        return self.forward(x)        
    def forward(self, x):
        # 手动实现矩阵乘法
        out = [Tensor(0.0)] * self.out_features        for i in range(self.out_features):
            out[i] = out[i]+self.b[i]            for j in range(self.in_features):
                out[i] = out[i] + x[j] * self.w[j][i]        return out    
    def parameters(self):   # 获得 Linear 的所有参数，可用于参数更新
        return [self.w, self.b]

# 测试 Linearnet = Linear(2,1)
x = [Tensor(1.0)] * 2net(x)

def squared_loss(y_hat, y):
    """均方损失"""
    loss = Tensor(0.0)    for i in range(len(y)):
        tmp = y_hat[i] + y[i]*Tensor(-1.0)
        tmp = tmp*tmp*Tensor(0.5)
        loss = loss + tmp    return loss

def sgd(params, lr):
    """随机梯度下降"""
    if isinstance(params, list):        for i in params:
            sgd(i, lr)    elif isinstance(params, Tensor):
        params.data -= params.grad*lr
        params.grad = 0.0

import numpy as np # 与框架关系不大，后面手写字数据集用def to_tensor(x):
    """将 list/numpy.ndarray 中的每个元素，转换成 Tensor"""
    ans = []    if isinstance(x, list) or isinstance(x, np.ndarray):        for i in range(len(x)):
            ans.append(to_tensor(x[i]))    else:
        ans = Tensor(x)    return ans# 准备数据X = [[1,2,3],[2,3,7],[5,3,1]]
Y = [[x_[0] + 2*x_[1] + 4*x_[2]] for x_ in X]  # y = [[17, 36, 15]]X = to_tensor(X)
Y = to_tensor(Y)for i in range(len(X)):    print(X[i],Y[i])

lr = 0.03               # 学习率num_epochs = 3          # 迭代次数net =  Linear(3,1)      # 构建神经网络loss = squared_loss     # 设置损失函数for epoch in range(num_epochs):    for i in range(len(X)):
        l = loss(net(X[i]),Y[i]) # X 和 y 的损失
        l.backward()
        params = net.parameters()
        sgd(params, lr)
    train_l = 0.0
    for i in range(len(X)): 
        train_l += loss(net(X[i]),Y[i]).data    print(f'epoch{epoch + 1}, loss {float(train_l/len(X)):f}')

net(X[0]), Y[0]

class MLP:
    def __init__(self, in_features, outs):
        self.linears = []
        self.num_linears = len(outs)        for i in range(len(outs)):   # 根据 outs 构建多个 Linear 层，上一层 Linear 的 out_features, 为下一层 Linear 的 in_features
            self.linears.append(Linear(in_features, outs[i]))
            in_features = outs[i]            
    def __call__(self, x):
        return self.forward(x)        
    def forward(self, x):
        for i in range(self.num_linears-1):
            x = self.linears[i](x)            for j in x:
                j.relu()    # 每一个 Linear 层后加一个 ReLU 激活函数
        x = self.linears[-1](x)        return x    
    def parameters(self):
        return [p for linear in self.linears for p in linear.parameters()]

# 准备数据X = [[1,2,3],[2,3,7],[5,3,1]]
Y = [[x_[0] + 2*x_[1] + 4*x_[2]] for x_ in X]  # y = [[17, 36, 15]]X = to_tensor(X)
Y = to_tensor(Y)for i in range(len(X)):    print(X[i],Y[i])

lr = 0.001num_epochs = 10net =  MLP(3,[2,1])
loss = squared_lossfor epoch in range(num_epochs):    for i in range(len(X)):
        l = loss(net(X[i]),Y[i]) # X 和 y 的损失
        l.backward()
        params = net.parameters()
        sgd(params, lr)
    train_l = 0.0
    for i in range(len(X)): 
        train_l += loss(net(X[i]),Y[i]).data    print(f'epoch{epoch + 1}, loss {float(train_l/len(X)):f}')

net(X[0]), Y[0]

!mkdir -p /home/aistudio/work/mnist
!unzip /home/aistudio/data/data33695/mnist.zip -d /home/aistudio/work/mnist/

import sys 
sys.path.append('/home/aistudio/work')import load_MNISTimport matplotlib.pyplot as pltdef load_datasets(show_examples=False):
    X_train = load_MNIST.load_train_images()
    y_train = load_MNIST.load_train_labels()
    X_test = load_MNIST.load_test_images()
    y_test = load_MNIST.load_test_labels()    if show_examples is True:
        sample = X_train[1, :, :]
        plt.imshow(sample)
        plt.show()        print('样例的矩阵形式为:\n {}'.format(sample))    return X_train, X_test, y_train, y_test


X_train_, X_test_, y_train_, y_test_ = load_datasets(show_examples=True)

X_train_.shape, y_train_.shape,type(X_train_),X_test_.shape, y_test_.shape

# 数据处理X_train = X_train_.reshape(60000,28*28)
y_train = y_train_.reshape(60000,1)
X_test = X_test_.reshape(10000,28*28)
y_test = y_test_.reshape(10000,1)
X_train[0],y_train[0]# 这个框架太粗糙，训练太慢，只使用部分数据train_data = 1000test_data =10X_train = to_tensor(X_train[0:train_data]) 
y_train = to_tensor(y_train[0:train_data])
X_test = to_tensor(X_test[0:test_data])
y_test = to_tensor(y_test[0:test_data])

lr = 0.00000001num_epochs = 2net = Linear(784,10)# net = MLP(28*28, [256,10])loss = squared_lossfor epoch in range(num_epochs):    for i in range(train_data):
        l = loss(net(X_train[i]),y_train[i]) # X 和 y 的损失
        l.backward()
        params = net.parameters()
        sgd(params, lr)        if i%100 == 0:            print(f'epoch{epoch + 1}, train loss {float(l.data):f}')
    test_l = 0.0
    for i in range(test_data): 
        test_l += loss(net(X_test[i]),y_test[i]).data    print(f'epoch{epoch + 1}, test loss {float(test_l/len(X_test)):f}')# net(X_train[0]), y_train[0]

import paddleimport paddle.nn as nn# 数据准备X_train_p = paddle.to_tensor(X_train_, stop_gradient=True).flatten(1).astype('float32')
X_test_p = paddle.to_tensor(X_test_, stop_gradient=True).flatten(1).astype('float32')
y_train_p = paddle.to_tensor(y_train_, stop_gradient=True).astype('float32')
y_test_p = paddle.to_tensor(y_test_, stop_gradient=True).astype('float32')

batch_size, lr, num_epochs = 256, 0.1, 10loss = paddle.nn.loss.MSELoss()
net = nn.Linear(28*28, 1) # 一个简单的全连接层sgd = paddle.optimizer.SGD(learning_rate=0.0000001, parameters=net.parameters())for epoch in range(num_epochs):    for i in range(60000):
        l = loss(net(X_train_p[i]),y_train_p[i]) # X 和 y 的损失
        l.backward()
        sgd.step()
        sgd.clear_grad()        if i%30000 == 0:            print(f'epoch{epoch + 1}, train loss {float(l.value()):f}')
    test_l = 0.0
    for i in range(1000): 
        test_l += loss(net(X_test_p[i]),y_test_p[i]).value()    print(f'epoch{epoch + 1}, test loss {float(test_l/10000):f}')# 粗略训练结果net(X_train_p[0]), y_train_p[0]