自己动手实现深度学习框架-2 核心实现

目标

        完成框架设计文档中列出的基础类和需要在基础类中实现的接口。使用最简的单多层感知机(Multi-Layer Perceptron)模型对框架进行初步验证, 因此, 除了框架的核心部分外, 还要实现一个全连接层，一个激活函数，一个优化器和一个损失函数。

框架代码简介

        我把这个框架命名为cute-dl, 已经上传到github上: https://github.com/brandonlyg/cute-dl.
        目录结构为:
        -- cutedl: 框架实现代码
        -- example: 示例
        -- test: 单元测试

        MLP示例位于 example/mlp目录下。

层(Layer)和层参数(LayerParam)

        相关代码在model.py中.
        LayerParam只有属性的定义, 没什么逻辑在里面:

lass LayerParam(object):

    '''
    layer_name: 所属层的的名字
    name: 参数名
    value: 参数值
    '''
    def __init__(self, layer_name, name, value):
        self.__name = layer_name+"/"+name
        self.value = value

        #梯度
        self.gradient = None
        #更新次数
        self.udt = 0

    @property
    def name(self):
        return self.__name

    def reset(self):
        self.gradient = None
        self.udt = 0

        其中参数名字是使用树形结构, 例如: "1-MyLayer/W", 是"1-MyLayer"层的"W"参数的名字。其中"1"是层在模型中的唯一ID, "MyLayer"是层的标签(tag), "W"是参数在这个层中的唯一名字。

        Layer需要实现两个方法: 一个是__init__方法，一个是join方法. 其他方法不需要实现，只需按设计文档中的描述给出定义即可。
        先来看看__init__方法:

    '''
    outshape: 输出形状 2 或者 (2,3)
    kargs:
        activation: 激活函数的名字
        inshape: 输入形状
    '''
    def __init__(self, *outshape, **kargs)
        #输出形状
        if len(outshape) == 1 and type(outshape[0]) == type(()):
            self.__outshape = outshape[0]
        else:
            self.__outshape = outshape

        #输入形状
        self.__inshape = None

        #得到激活函数
        self.__activation = activations.get('linear')

        #层在模型中的id, 是层在模型中的索引
        self.__id = 0
        #层的名字
        self.__name = '/%d-%s'%(self.__id, self.tag)

        #得到可选参数
        #print("Layer kargs:", kargs)
        if 'inshape' in kargs:
            self.__inshape = kargs['inshape']
            if type(self.__inshape) != type(()):
                self.__inshape = (self.__inshape,)
            #print("------inshape:", self.__inshape)

        if 'activation' in kargs:
            self.__activation = activations.get(kargs['activation'])


        if self.__inshape is not None:
            self.init_params()

        实现的时主要处理这么几个问题:

1. 输入输出形状的处理. 保证输入输出形状不论是int还是tuple, 最后都要转换成tuple。
2. 必须要有激活函数。默认使用的线性激活函数。
3. 自动生成层的名字。

        join方法实现:

    '''
    加入到模型中
    pre_layer: 前一个层
    *inshape: 输入形状
    '''
    def join(self, pre_layer, *inshape):
        if self.__outshape == (-1,):
            self.__inshape = pre_layer.inshape
            self.__outshape = pre_layer.outshape
        else:
            self.__inshape = pre_layer.outshape
            if len(inshape) != 0:
                self.__inshape = inshape

        self.__id = pre_layer.layer_id + 1
        self.__name = '/%d-%s'%(self.__id, self.tag)

        self.init_params()

        这个方法主要功能是把当前层和另一层连接在一起, 让另一个层成为当前层的(在模型中的)前一层。这里的"连接"主要体现在: 把另一个层的输出作为输入。对层ID的处理上, 使用简单的累加保证层ID在模型中是唯一的, 同时还能通过ID的值知道层位于模型中的什么位置。 有了输入输出形状, 就可以调用子类实现的init_params方法初始化参数了。

激活函数(Activation)

        激活函数代码在activation.py中。
        接口定义:

'''
激活函数
'''
class Activation(object):
    name=''

    def __call__(self, in_batch):
        raise Exception("__call__ not implement")

    '''
    求梯度
    gradient: 该函数输出值的梯度
    '''
    def grad(self, gradient):
        raise Exception("gradient not implement")

        其中类属性name作为激活函数的名字。
        实现线性激活函数, 作为默认激活函数:

'''
线性激活函数, 没有激活
'''
class Linear(Activation):
    name='linear'

    def __call__(self, in_batch):
        return in_batch

    def grad(self, gradient):
        return gradient

        实现最常用的relu激活函数:

'''
relu 激活函数
'''
class Relu(Activation):
    name='relu'

    def __init__(self):
        self.__grad = None

    def __call__(self, in_batch):
        #得到 <= 0的数据的索引
        indices =  in_batch <= 0

        in_batch[indices] = 0
        self.__grad = indices

        return in_batch

    def grad(self, gradient):
        gradient[self.__grad] = 0
        self.__grad = None
        return gradient

        实现用名字(name)获取激活函数:

act_dict = {
    Linear.name: Linear,
    Relu.name: Relu
}

#创建激活函数
def get(name):
    #print(act_dict)
    #print('name:', name)
    ACT = act_dict[name]
    return ACT()

模型(Model)

        首先需要向模型中添加层

'''
    layers: Layer list
    '''
    def __init__(self, layers=None):
        self.__layers = layers

    '''
    添加层
    layer: Layer类型的对象
    '''
    def add(self, layer):
        if self.__layers is None:
            self.__layers = []

        self.__layers.append(layer)

        return self

        __init__和add方法都能实现这个功能。

        然后是层的的访问能力:

'''
    得到一个Layer对象
    idx: Layer对象的索引
    '''
    def get_layer(self, index):
        self.__check()
        if len(self.__layers) <= index:
            raise Exception("index out of range %d"%len(self.__layers))

        return self.__layers[index]

    @property
    def layer_count(self):
        return len(self.__layers)

    '''
    得到层的迭代器
    '''
    def layer_iterator(self):
        self.__check()

        for ly in self.__layers:
            yield ly

        接下来是组装模型:

    '''
    组装模型
    '''
    def assemble(self):
        self.__check()
        count = len(self.__layers)

        #输入层必须要有输入形状
        ly_0 = self.__layers[0]
        if ly_0.inshape is None or len(ly_0.inshape) == 0:
            raise Exception("input layer miss inshape")

        #把每一层的输入形状设置为上一层的输出形状,
        #设置输入形状的同时, 要求该层自动初始化参数(如果有参数的话)
        pre_ly = ly_0
        for ly in self.__layers[1:]:
            ly.join(pre_ly)
            pre_ly = ly

        向前传播:

    '''
    使用模型预测
    in_batch: 一批输入数据
    '''
    def predict(self, in_batch, training=False):
        self.__check()

        out = in_batch
        for ly in self.__layers:
            out = ly.forward(out, training)

        return out

        反向传播:

    '''
    反向传播梯度
    '''
    def backward(self, gradient):
        g = gradient
        #pdb.set_trace()
        count = len(self.__layers)
        for i in range(count-1, -1, -1):
            ly = self.__layers[i]
            g = ly.backward(g)

训练上下文会话(Session)

        Session代码在session.py中。
        初始化__init__:

    '''
    model: Model对象
    loss: Loss对象
    optimizer: Optimizer对象
    '''
    def __init__(self, model, loss, optimizer):
        self.__model = model
        self.__loss = loss
        self.__optimizer = optimizer

会话主要维护模型, 损失函数和优化器。这些对一个简单的MLP模型来说已经足够，至于genoptimizer以后再添加。

        训练模型:

    '''
    分批训练
    '''
    def batch_train(self, data, label):
        #使用模型预测
        out = self.__model.predict(data, training=True)
        #使用损失函数评估误差
        loss = self.__loss(out, label)
        grad = self.__loss.gradient
        #pdb.set_trace()
        #反向传播梯度
        self.__model.backward(self.__loss.gradient)

        #更新模型参数
        self.__optimizer(self.__model)

        return loss

        保存会话:

    '''
    保存session
    fpath: 保存的文件路径
        fpath+'.s.pkl' 是保存session的文件
        fpath+'.m.pkl' 是保存model的文件
    '''
    def save(self, fpath):
        model = self.__model
        self.__model = None

        model.save(fpath)

        realfp = fpath + ".s.pkl"
        with open(realfp, 'wb') as f:
            pickle.dump(self, f)

        这里把模型和会话分开保存, 是为了以后可以灵活地选择只加载模型或加载整个会话。下面是模型的保存方法, 在Model中实现:

    '''
    保存模型
    '''
    def save(self, fpath):
        dir = os.path.dirname(fpath)
        if not os.path.exists(dir):
            os.mkdir(dir)

        self.reset()
        realfp = fpath + ".m.pkl"
        with open(realfp, 'wb') as f:
            pickle.dump(self, f)

        加载会话:

    '''
    加载session
    '''
    @classmethod
    def load(cls, fpath):
        realfp = fpath + ".s.pkl"
        if not os.path.exists(realfp):
            return None

        sess = None
        with open(realfp, 'rb') as f:
            sess = pickle.load(f)

        model = Model.load(fpath)
        sess.set_model(model)

        return sess

损失函数(Loss)

        损失函数代码在loss.py中。首先定义接口:

'''
损失函数
'''
class Loss(object):

    '''
    梯度属性
    '''
    @property
    def gradient(self):
        raise Exception("gradient not impliment")


    '''
    计算误差和梯度
    y_true 数据的真实标签
    y_pred 模型预测的标签

    return 误差值
    '''
    def __call__(self, y_true, y_pred):
        raise Exception("__call__ not impliment")

        接下来给出均方误差损失函数实现:

'''
均方误差损失函数
'''
class Mse(Loss):

    def __init__(self):
        self.__grad = None

    def __call__(self, y_true, y_pred):
        err = y_true - y_pred
        loss =  (err**2).mean(axis=0)/2

        n = y_true.shape[0]
        self.__grad = err/n
        #pdb.set_trace()
        return loss.sum()

    @property
    def gradient(self):
        return self.__grad

学习率优化器(Optimizer)

        优化器代码在optimizer.py中。
        定义接口:

'''
学习率优化器
'''
class Optimizer(object):

    '''
    更新参数
    '''
    def __call__(self, model):
        raise Exception('not implement')

        实现一个固定学习率优化器, 没有用任何参数优化算法。

'''
固定学习率优化器
'''
class Fixed(Optimizer):

    '''
    lt: 学习率
    '''
    def __init__(self, lt=0.01):
        self.__lt = lt

    def __call__(self, model):
        #pdb.set_trace()
        for ly in model.layer_iterator():
            for p in ly.params:
                p.value -= self.__lt * p.gradient
                p.udt += 1

        到目前为止，一个能够支持最简单MLP模型的框架已经完成。接下来用一个MLP示例来验证一下。

MLPS示例

        使用MLP模型完成一个广义线性回归的任务, 代码在examples/mlp/linear-regression.py中。
        假设这个任务是拟合一个二次多项式函数:

'''
任务目标函数
'''
def target_func(x):
    ##加入服从参数(0, 0.25^2)正态分布噪声
    y = (x - 2)**2 + 0.25 * np.random.randn(len(x))
    return y

看一下这个函数的图像:

        从使用这个函数采样得到数据集:

'''
生成数据集
返回: train_x, train_y, test_x, test_y
train_x, train_y 训练数据集的数据和标签
test_x, test_y 验证数据解的数据和标签
'''
def generate_dataset():
    '''
    生成200条数据, 随机取出80%条作为训练数据集, 剩余数据为测试数据集
    '''
    fpath = "./ds.pkl"
    if os.path.exists(fpath):
        with open(fpath, 'rb') as f:
            ds = pickle.load(f)
            return ds

    count = 200
    x = np.linspace(-1, 5, count)
    y = target_func(x)

    #打乱顺序
    indices = np.arange(count)
    np.random.shuffle(indices)
    #训练数据集
    split = int(count*0.8)
    idxs = indices[:split]
    train_x = x[idxs].reshape((-1,1))
    train_y = y[idxs].reshape((-1,1))

    #测试数据集
    idxs = sorted(indices[split:])
    test_x = x[idxs].reshape((-1, 1))
    shape = test_x.shape
    test_y = y[idxs].reshape((-1, 1))

    ds = {
        'train_x': train_x,
        'train_y': train_y,
        'test_x': test_x,
        'test_y': test_y
    }
    with open(fpath, 'wb') as f:
        pickle.dump(ds, f)

    return ds

#得到数据集
ds_0 = generate_dataset()
print("train shape:", ds_0['train_x'].shape)
print("test shape:", ds_0['test_x'].shape)

#训练集只取一部分
count = 100
ds_1 = {
    'train_x': ds_0['train_x'][:16],
    'train_y': ds_0['train_y'][:16],
    'test_x': ds_0['test_x'],
    'test_y': ds_0['test_y']
}

        这里得到两个数据集, 一个数据集中有160条训练数据, 40条验证数据。另一个中有16条训练数据和40条验证数据。

        分批训练模型:

'''
训练模型
'''
def train(epochs, ds, model=None, batch_size=64, record_epochs=1):
    #加载/构建session
    sess = None
    if model is None:
        sess = Session.load(model_path)
    else:
        sess = Session(model,
                    loss=losses.Mse(),
                    optimizer = optimizers.Fixed()
                )

    train_x = ds['train_x']
    train_y = ds['train_y']
    test_x = ds['test_x']
    test_y = ds['test_y']

    batchs = int(train_x.shape[0]/batch_size)
    print("epochs:%d, batchs=%d"%(epochs, batchs))

    #记录训练历史
    history = {
        'loss': [],
        'val_loss': [],
        'epochs': [],
        'val_x': test_x,
        'val_y': test_y,
        'val_pred': None
    }

    print("start training ")
    t_start = time.time()
    steps = epochs * batchs

    epoch = 1
    #循环训练
    for step in range(steps):
        start = (step % batchs) * batch_size
        end = start + batch_size
        batch_x = train_x[start:end]
        batch_y = train_y[start:end]

        loss = sess.batch_train(batch_x, batch_y)

        cur_epoch = int(step/batchs) + 1

        #每轮打印一次
        if step > 0 and  step % batchs == 0:
            print((('epoch:%05d/%d loss=%f'%(cur_epoch, epochs, loss))+' '*50)[:50], end='
')

        #记录
        if step % batchs == 0 and (cur_epoch - epoch == record_epochs or cur_epoch == epochs):
            epoch = cur_epoch

            y_pred = sess.model.predict(test_x)
            val_loss = sess.loss(test_y, y_pred)

            history['loss'].append(loss)
            history['val_loss'].append(val_loss)
            history['epochs'].append(epoch)
            history['val_pred']  = y_pred

            print((('epoch:%05d/%d loss=%f, val_loss=%f'%(cur_epoch, epochs, loss, val_loss))+' '*50)[:50], end='
')
            print("")

    sess.save(model_path)
    print("training finished cost:%f" % (time.time() - t_start))

    return history

        通过这段代码可以看出，框架虽然看起来可用, 但训练模型是仍然需要不少代码，不够友好。不过没关系，目前先通过示例积累经验，以后在把分批训练的功能加入到Session中。要牢记现阶段的主要任务: 对框架进行初步验证。

验证1: 使用线性模型拟合目标函数

#欠拟合示例
def fit_1():
    model = Model([
        nnlys.Dense(32, inshape=1),
        nnlys.Dense(1)
    ])
    model.assemble()
    #这个模型是一个线性模型, 用来拟合非线性函数, 模型复杂度不够，一定会表现出欠拟合
    history = train(20000, ds_0, model, record_epochs=100)
    fit_report(history, report_path+'01.png')

        拟合报告:

可以看到不论是训练误差还是验证误差都很大, 下面的拟合图形更是惨不忍睹。模型呈欠拟合。

验证2：使用同样多参数的非线性模型拟合目标函数

#使用增加模型复杂度解决欠拟合问题
def fit_2():
    model = Model([
        nnlys.Dense(32, inshape=1, activation='relu'),
        nnlys.Dense(1)
    ])
    model.assemble()
    #使用了relu激活函数模型变成了非线性的, 增加了模型的复杂度
    history = train(30000, ds_0, model, record_epochs=300)
    history['loss'] = history['loss'][5:]
    history['val_loss'] = history['val_loss'][5:]
    history['epochs'] = history['epochs'][5:]
    fit_report(history, report_path+'02.png')

        拟合报告:

拟合情况比较理想。

验证3: 增加模型复杂度减少训练数据

#过拟合
def fit_3():
    model = Model([
        nnlys.Dense(512, inshape=1, activation='relu'),
        nnlys.Dense(128, activation='relu'),
        nnlys.Dense(1)
    ])
    model.assemble()
    #使用数据集ds_1, 只有16条训练数据
    history = train(30000, ds_1, model, batch_size=16, record_epochs=300)
    history['loss'] = history['loss'][20:]
    history['val_loss'] = history['val_loss'][20:]
    history['epochs'] = history['epochs'][20:]
    fit_report(history, report_path+'03.png')

        拟合报告:

可以看到训练误差持续降低, 而验证误差先低后高, 说明随着训练轮次的增加，模型过多地学习到了训练数据的模式, 导致泛化误差增大，呈现过过拟合。

验证3: 仍然使用较少的训练数据但降低模型的复杂度

#减少参数数量缓解过拟合
def fit_4():
    model = Model([
        nnlys.Dense(128, inshape=1, activation='relu'),
        nnlys.Dense(64, activation='relu'),
        nnlys.Dense(1)
    ])
    model.assemble()

    history = train(30000, ds_1, model, batch_size=16, record_epochs=300)
    history['loss'] = history['loss'][20:]
    history['val_loss'] = history['val_loss'][20:]
    history['epochs'] = history['epochs'][20:]
    fit_report(history, report_path+'04.png')

        拟合报告:

可以看到过拟合现象有所缓解，到25000左右才出现过拟合现象，拟合图形变得稍微好一点, 过拟合只是略有缓解。

总结

        目前已经实现了一个最简单可运行的深度学习框架。从验证情况看，它已经达到预期，能够支持简单的MLP模型, 但很直接地暴露出两个问题:

1. 训练模型需要比较多的代码，不够友好。
2. 框架本身没有提供处理过拟合现象的方案。
   下一步的主要目标就是解决这两个问题。