深度学习算法实践12---卷积神经网络(CNN)实现

在搞清楚卷积神经网络(CNN)的原理之后,在本篇博文中,我们将讨论基于Theano的算法实现技术。我们还将以MNIST手写数字识别为例,创建卷积神经网络(CNN),训练该网络,使识别误差达到1%以内。

我们首先需要读入MNIST手写数字识别的训练样本集,为此我们定义了一个工具类:

from __future__ import print_function

__docformat__ = ‘restructedtext en‘

import six.moves.cPickle as pickle
import gzip
import os
import sys
import timeit

import numpy

import theano
import theano.tensor as T

class MnistLoader(object):
    def load_data(self, dataset):
        data_dir, data_file = os.path.split(dataset)
        if data_dir == "" and not os.path.isfile(dataset):
            new_path = os.path.join(
                os.path.split(__file__)[0],
                "..",
                "data",
                dataset
            )
            if os.path.isfile(new_path) or data_file == ‘mnist.pkl.gz‘:
                dataset = new_path

        if (not os.path.isfile(dataset)) and data_file == ‘mnist.pkl.gz‘:
            from six.moves import urllib
            origin = (
                ‘http://www.iro.umontreal.ca/~lisa/deep/data/mnist/mnist.pkl.gz‘
            )
            print(‘Downloading data from %s‘ % origin)
            urllib.request.urlretrieve(origin, dataset)

        print(‘... loading data‘)
        # Load the dataset
        with gzip.open(dataset, ‘rb‘) as f:
            try:
                train_set, valid_set, test_set = pickle.load(f, encoding=‘latin1‘)
            except:
                train_set, valid_set, test_set = pickle.load(f)
        def shared_dataset(data_xy, borrow=True):
            data_x, data_y = data_xy
            shared_x = theano.shared(numpy.asarray(data_x,
                                               dtype=theano.config.floatX),
                                 borrow=borrow)
            shared_y = theano.shared(numpy.asarray(data_y,
                                               dtype=theano.config.floatX),
                                 borrow=borrow)
            return shared_x, T.cast(shared_y, ‘int32‘)

        test_set_x, test_set_y = shared_dataset(test_set)
        valid_set_x, valid_set_y = shared_dataset(valid_set)
        train_set_x, train_set_y = shared_dataset(train_set)

        rval = [(train_set_x, train_set_y), (valid_set_x, valid_set_y),
            (test_set_x, test_set_y)]
        return rval

这个类在之前我们已经用过,在这里就不详细讲解了。之所以单独定义这个类,是因为如果我们将问题换为其他类型时,我们只需要修改这一个类,就可以实现训练数据的载入了,这样简化了程序修改工作量。

我们所采用的方法是将图像先接入卷积神经网络,之后再接入BP网络的隐藏层,然后再接入逻辑回归的输出层,因此我们需要先定义多层前向网络的隐藏层和逻辑回归输出层。隐藏层的定义如下所示:

from __future__ import print_function

__docformat__ = ‘restructedtext en‘

import os
import sys
import timeit

import numpy

import theano
import theano.tensor as T

from logistic_regression import LogisticRegression

# start-snippet-1
class HiddenLayer(object):
    def __init__(self, rng, input, n_in, n_out, W=None, b=None,
                 activation=T.tanh):
        self.input = input
        if W is None:
            W_values = numpy.asarray(
                rng.uniform(
                    low=-numpy.sqrt(6. / (n_in + n_out)),
                    high=numpy.sqrt(6. / (n_in + n_out)),
                    size=(n_in, n_out)
                ),
                dtype=theano.config.floatX
            )
            if activation == theano.tensor.nnet.sigmoid:
                W_values *= 4

            W = theano.shared(value=W_values, name=‘W‘, borrow=True)

        if b is None:
            b_values = numpy.zeros((n_out,), dtype=theano.config.floatX)
            b = theano.shared(value=b_values, name=‘b‘, borrow=True)

        self.W = W
        self.b = b

        lin_output = T.dot(input, self.W) + self.b
        self.output = (
            lin_output if activation is None
            else activation(lin_output)
        )
        # parameters of the model
        self.params = [self.W, self.b]

接下来我们定义逻辑回归算法类:

from __future__ import print_function

__docformat__ = ‘restructedtext en‘

import six.moves.cPickle as pickle
import gzip
import os
import sys
import timeit

import numpy

import theano
import theano.tensor as T

class LogisticRegression(object):
    def __init__(self, input, n_in, n_out):
        self.W = theano.shared(
            value=numpy.zeros(
                (n_in, n_out),
                dtype=theano.config.floatX
            ),
            name=‘W‘,
            borrow=True
        )
        self.b = theano.shared(
            value=numpy.zeros(
                (n_out,),
                dtype=theano.config.floatX
            ),
            name=‘b‘,
            borrow=True
        )
        self.p_y_given_x = T.nnet.softmax(T.dot(input, self.W) + self.b)
        self.y_pred = T.argmax(self.p_y_given_x, axis=1)
        self.params = [self.W, self.b]
        self.input = input
        print("Yantao: ***********************************")

    def negative_log_likelihood(self, y):
        return -T.mean(T.log(self.p_y_given_x)[T.arange(y.shape[0]), y])  

    def errors(self, y):
        if y.ndim != self.y_pred.ndim:
            raise TypeError(
                ‘y should have the same shape as self.y_pred‘,
                (‘y‘, y.type, ‘y_pred‘, self.y_pred.type)
            )
        if y.dtype.startswith(‘int‘):
            return T.mean(T.neq(self.y_pred, y))
        else:
            raise NotImplementedError()

这段代码在逻辑回归博文中已经详细讨论过了,这里就不再重复了,有兴趣的读者可以查看这篇博文(逻辑回归算法实现)。

做完上述准备工作之后,我们就可以开始卷积神经网络(CNN)实现了。

我们先来定义基于简化版Lenet5的卷积神经网络(CNN)的定义,代码如下所示:

from __future__ import print_function

import os
import sys
import timeit

import numpy

import theano
import theano.tensor as T
from theano.tensor.signal import pool
from theano.tensor.nnet import conv2d

class LeNetConvPoolLayer(object):
    def __init__(self, rng, input, filter_shape, image_shape, poolsize=(2, 2)):
        assert image_shape[1] == filter_shape[1]
        self.input = input
        fan_in = numpy.prod(filter_shape[1:])
        fan_out = (filter_shape[0] * numpy.prod(filter_shape[2:]) //
                   numpy.prod(poolsize))
        W_bound = numpy.sqrt(6. / (fan_in + fan_out))
        self.W = theano.shared(
            numpy.asarray(
                rng.uniform(low=-W_bound, high=W_bound, size=filter_shape),
                dtype=theano.config.floatX
            ),
            borrow=True
        )
        b_values = numpy.zeros((filter_shape[0],), dtype=theano.config.floatX)
        self.b = theano.shared(value=b_values, borrow=True)
        conv_out = conv2d(
            input=input,
            filters=self.W,
            filter_shape=filter_shape,
            input_shape=image_shape
        )
        pooled_out = pool.pool_2d(
            input=conv_out,
            ds=poolsize,
            ignore_border=True
        )
        self.output = T.tanh(pooled_out + self.b.dimshuffle(‘x‘, 0, ‘x‘, ‘x‘))
        self.params = [self.W, self.b]
        self.input = input

上面代码实现了对输入信号的卷积操作,并对结果进行最大化池化。

下面我们来看怎样初始化Lenet层,怎样将Lenet层输出信号转为MLP网络隐藏层的输入信号,具体代码如下所示:

layer0 = LeNetConvPoolLayer(
        rng,
        input=layer0_input,
        image_shape=(batch_size, 1, 28, 28),
        filter_shape=(nkerns[0], 1, 5, 5),
        poolsize=(2, 2)
    )

如上所示,我们的输入信号是28*28的黑白图像,而且我们采用的批量学习,因此输入图像就定义为(batch_size, 1, 28, 28),我们对图像进行5*5卷积操作,根据卷积操作定义,最终得到的卷积输出层为(28-5+1,28-5+1)=(24,24)的“图像”,我们采用2*2的最大池化操作,即取2*2区域像素的最大值作为新的像素点的值,则最终输出层得到12*12的输出信号。

接下来,我们将输出信号继续输入一个Lenet卷积池化层,代码如下所示:

    layer1 = LeNetConvPoolLayer(
        rng,
        input=layer0.output,
        image_shape=(batch_size, nkerns[0], 12, 12),
        filter_shape=(nkerns[1], nkerns[0], 5, 5),
        poolsize=(2, 2)
    )

如上所示,这时输入信号变化为12*12的图像,我们还使用5*5的卷积核,可以得到(12-5+1, 12-5+1)=(8,8)的图像,采用2*2最大池化操作后,得到(4,4)图像。可以通过调用layer1.output.flatten(2)将其变为一维信号,从而输入MLP的隐藏层。

下面我们定义Lenet引擎来实现装入数据,定义网络模型,训练网络工作,代码如下所示:

from __future__ import print_function

import os
import sys
import timeit

import numpy

import theano
import theano.tensor as T
from theano.tensor.signal import pool
from theano.tensor.nnet import conv2d

from mnist_loader import MnistLoader
from logistic_regression import LogisticRegression
from hidden_layer import HiddenLayer
from lenet_conv_pool_layer import LeNetConvPoolLayer

class LenetMnistEngine(object):
    def __init__(self):
        print("create LenetMnistEngine")

    def train_model(self):
        learning_rate = 0.1
        n_epochs = 200
        dataset = ‘mnist.pkl.gz‘
        nkerns = [20, 50]
        batch_size = 500
        (n_train_batches, n_test_batches, n_valid_batches,                     train_model, test_model, validate_model) =                     self.build_model(learning_rate, n_epochs,                         dataset, nkerns, batch_size)
        self.train(n_epochs, n_train_batches, n_test_batches,                     n_valid_batches, train_model, test_model,                     validate_model)

    def run(self):
        print("run the model")
        classifier = pickle.load(open(‘best_model.pkl‘, ‘rb‘))
        predict_model = theano.function(
            inputs=[classifier.input],
            outputs=classifier.logRegressionLayer.y_pred
        )
        dataset=‘mnist.pkl.gz‘
        loader = MnistLoader()
        datasets = loader.load_data(dataset)
        test_set_x, test_set_y = datasets[2]
        test_set_x = test_set_x.get_value()
        predicted_values = predict_model(test_set_x[:10])
        print("Predicted values for the first 10 examples in test set:")
        print(predicted_values)

    def build_model(self, learning_rate=0.1, n_epochs=200,
                        dataset=‘mnist.pkl.gz‘,
                        nkerns=[20, 50], batch_size=500):
        rng = numpy.random.RandomState(23455)
        loader = MnistLoader()
        datasets = loader.load_data(dataset)
        train_set_x, train_set_y = datasets[0]
        valid_set_x, valid_set_y = datasets[1]
        test_set_x, test_set_y = datasets[2]
        n_train_batches = train_set_x.get_value(borrow=True).shape[0]
        n_valid_batches = valid_set_x.get_value(borrow=True).shape[0]
        n_test_batches = test_set_x.get_value(borrow=True).shape[0]
        n_train_batches //= batch_size
        n_valid_batches //= batch_size
        n_test_batches //= batch_size
        index = T.lscalar()
        x = T.matrix(‘x‘)
        y = T.ivector(‘y‘)
        print(‘... building the model‘)
        layer0_input = x.reshape((batch_size, 1, 28, 28))
        layer0 = LeNetConvPoolLayer(
            rng,
            input=layer0_input,
            image_shape=(batch_size, 1, 28, 28),
            filter_shape=(nkerns[0], 1, 5, 5),
            poolsize=(2, 2)
        )
        layer1 = LeNetConvPoolLayer(
            rng,
            input=layer0.output,
            image_shape=(batch_size, nkerns[0], 12, 12),
            filter_shape=(nkerns[1], nkerns[0], 5, 5),
            poolsize=(2, 2)
        )
        layer2_input = layer1.output.flatten(2)
        layer2 = HiddenLayer(
            rng,
            input=layer2_input,
            n_in=nkerns[1] * 4 * 4,
            n_out=500,
            activation=T.tanh
        )
        layer3 = LogisticRegression(input=layer2.output, n_in=500, n_out=10)
        cost = layer3.negative_log_likelihood(y)
        test_model = theano.function(
            [index],
            layer3.errors(y),
            givens={
                x: test_set_x[index * batch_size: (index + 1) * batch_size],
                y: test_set_y[index * batch_size: (index + 1) * batch_size]
            }
        )
        validate_model = theano.function(
            [index],
            layer3.errors(y),
            givens={
                x: valid_set_x[index * batch_size: (index + 1) * batch_size],
                y: valid_set_y[index * batch_size: (index + 1) * batch_size]
            }
        )
        params = layer3.params + layer2.params + layer1.params + layer0.params
        grads = T.grad(cost, params)
        updates = [
            (param_i, param_i - learning_rate * grad_i)
            for param_i, grad_i in zip(params, grads)
        ]
        train_model = theano.function(
            [index],
            cost,
            updates=updates,
            givens={
                x: train_set_x[index * batch_size: (index + 1) * batch_size],
                y: train_set_y[index * batch_size: (index + 1) * batch_size]
            }
        )
        return (n_train_batches, n_test_batches, n_valid_batches,                     train_model, test_model, validate_model)

    def train(self, n_epochs, n_train_batches, n_test_batches, n_valid_batches,
                train_model, test_model, validate_model):
        print(‘... training‘)
        patience = 10000
        patience_increase = 2
        improvement_threshold = 0.995
        validation_frequency = min(n_train_batches, patience // 2)
        best_validation_loss = numpy.inf
        best_iter = 0
        test_score = 0.
        start_time = timeit.default_timer()
        epoch = 0
        done_looping = False
        while (epoch < n_epochs) and (not done_looping):
            epoch = epoch + 1
            for minibatch_index in range(n_train_batches):
                iter = (epoch - 1) * n_train_batches + minibatch_index
                if iter % 100 == 0:
                    print(‘training @ iter = ‘, iter)
                cost_ij = train_model(minibatch_index)
                if (iter + 1) % validation_frequency == 0:
                    validation_losses = [validate_model(i) for i
                                         in range(n_valid_batches)]
                    this_validation_loss = numpy.mean(validation_losses)
                    print(‘epoch %i, minibatch %i/%i, validation error %f %%‘ %
                          (epoch, minibatch_index + 1, n_train_batches,
                           this_validation_loss * 100.))
                    if this_validation_loss < best_validation_loss:
                        if this_validation_loss < best_validation_loss *                             improvement_threshold:
                            patience = max(patience, iter * patience_increase)
                        best_validation_loss = this_validation_loss
                        best_iter = iter
                        test_losses = [
                            test_model(i)
                            for i in range(n_test_batches)
                        ]
                        test_score = numpy.mean(test_losses)
                        with open(‘best_model.pkl‘, ‘wb‘) as f:
                            pickle.dump(classifier, f)
                        print((‘     epoch %i, minibatch %i/%i, test error of ‘
                               ‘best model %f %%‘) %
                              (epoch, minibatch_index + 1, n_train_batches,
                               test_score * 100.))
                if patience <= iter:
                    done_looping = True
                    break
        end_time = timeit.default_timer()
        print(‘Optimization complete.‘)
        print(‘Best validation score of %f %% obtained at iteration %i, ‘
              ‘with test performance %f %%‘ %
              (best_validation_loss * 100., best_iter + 1, test_score * 100.))
        print((‘The code for file ‘ +
               os.path.split(__file__)[1] +
               ‘ ran for %.2fm‘ % ((end_time - start_time) / 60.)), file=sys.stderr)

上述代码与之前的MLP的训练代码类似,这里就不再讨论了。在我的Mac笔记本上,运行大约6个小时,会得到错误率小于1%的结果。

时间: 2024-10-12 04:25:06

深度学习算法实践12---卷积神经网络(CNN)实现的相关文章

深度学习算法实践10---卷积神经网络(CNN)原理

其实从本篇博文开始,我们才算真正进入深度学习领域.在深度学习领域,已经经过验证的成熟算法,目前主要有深度卷积网络(DNN)和递归网络(RNN),在图像识别.视频识别.语音识别领域取得了巨大的成功,正是由于这些成功,能促成了当前深度学习的大热.与此相对应的,在深度学习研究领域,最热门的是AutoEncoder.RBM.DBN等产生式网络架构,但是这些研究领域,虽然论文比较多,但是重量级应用还没有出现,是否能取得成功还具有不确定性.但是有一些比较初步的迹象表明,这些研究领域还是非常值得期待的.比如A

深度学习算法实践7---前向神经网络算法原理

总体上来讲,神经网络的算法是比较复杂的,后面有非常精妙的数学原理,而且对这些数学方法,还需要证明其具有收敛性,所以很多神经网络的书籍,绝大部分都是一些数学模型介绍.推导和证明,对于非数学专业的我们来说,感沉像看天书一样.其实神经网络的精髓是将现实中的问题进行抽象,建立适合神经网络表示的模型,然后应用神经网络进行处理,不断调整优化网络结构和参数,直到最终达到满意的效果.所以成功应用神经网络,不需要太过高深的数学基础.但是如果可以适当掌握神经网络背后的数学原理,对于我们理解参数的含义,选择合适的网络

吴恩达【深度学习工程师】 04.卷积神经网络 第三周目标检测 (1)基本的对象检测算法

该笔记介绍的是<卷积神经网络>系列第三周:目标检测(1)基本的对象检测算法 主要内容有: 1.目标定位 2.特征点检测 3.目标检测  目标定位 使用算法判断图片中是不是目标物体,如果是还要再图片中标出其位置并使用边框标记出来                          我们研究过的问题中,图片分类的思想可以帮助学习分类定位,而分类定位的思想又可以帮助学习对象检测 a.分类定位问题,通常有一个较大的对象位于图片中间位置. b.对象检测问题,图片可以含有多个对象,或一张图片中会有多个不同分

深度学习算法实践15---堆叠去噪自动编码机(SdA)原理及实现

我们讨论了去噪自动编码机(dA),并讨论了Theano框架实现的细节.在本节中,我们将讨论去噪自动编码机(dA)的主要应用,即组成堆叠自动编码机(SdA),我们将以MNIST手写字母识别为例,用堆叠自动编码机(SdA)来解决这一问题. 堆叠自动编码机(SdA)是由一系列去噪自动编码机堆叠而成,每个去噪自动编码机的中间层(即编码层)作为下一层的输入层,这样一层一层堆叠起来,构成一个深层网络,这些网络组成堆叠去噪自动编码机(SdA)的表示部分.这部分通过无监督学习,逐层进行培训,每一层均可以还原加入

SIGAI深度学习第七集 卷积神经网络1

讲授卷积神经网络核心思想.卷积层.池化层.全连接层.网络的训练.反向传播算法.随机梯度下降法.AdaGrad算法.RMSProp算法.AdaDelta算法.Adam算法.迁移学习和fine tune等. 大纲: 卷积神经网络简介 视觉神经网络的核心思想 卷积神经网络的核心思想 卷积运算 卷积层的原理 多通道卷积 池化层的原理 全连接层 卷积网络结构 训练算法简介 卷积层的反向传播 池化层的反向传播 全连接层的反向传播 完整的反向传播算法 随机梯度下降法 参数值初始化 学习率的设定 梯度下降法的改

吴恩达【深度学习工程师】 04.卷积神经网络 第四周特殊应用(2)神经风格转换

该笔记介绍的是<卷积神经网络>系列第四周:特殊应用(2)神经风格转换 主要内容有: 1.神经风格转换 2.卷积网络隐藏单元可视化显示 3.神经风格转换代价函数 4.内容代价函数 5.风格代价函数 *6.数据从一维到三维的推广 神经风格转换 把C(content)内容图片转化为风格类似S(style)风格图片的G(generated)生成图片. 卷积网络隐藏单元可视化显示 通过观察可视化显示的不同隐藏单元的输出结果真正了解卷积网络中深度较大的层真正在做些什么. 可视化图块输出: 从第一层隐藏层第

TensorFlow深度学习实战---图像识别与卷积神经网络

全连接层网络结构:神经网络每两层之间的所有结点都是有边相连的. 卷积神经网络:1.输入层 2.卷积层:将神经网络中的每一个小块进行更加深入地分析从而得到抽象程度更高的特征. 3 池化层:可以认为将一张分别率较高的图片转化为分别率较低的图片,通过池化层,可以进一步缩小最后全连接层中节点的个数,从而达到减少整个神经网络中参数的目的. 4全连接层:可以认为图像中的信息已经被抽象成了信息含量更高的特征. 5 softmax层 循环神经网络 原文地址:https://www.cnblogs.com/bju

深度学习算法实践8---BP算法详解

BP算法是关于误差的反向传播算法,就是从输出层开始,将结果与预期结果相比较,求出误差,然后按照梯度最大下降方向,调整神经元的联接权值,然后依次逐层调整各层之间的连接权值,对于批量学习方式而言,不断重复上述过程,直到误差达到足够小时为止. 对于输出层而言,我们可以直接使用在上一篇博文中关于感知器模型的算法,BP算法的难点在于,如何处理隐藏层,因为隐藏层没有正确的输出信息用来计算误差. 下面我们将从输出层开始,推导出BP算法.在rwy推导算法之前,我们先来定义一下表示方法,和原来一样,我们用i来表示

SIGAI深度学习第十集 卷积神经网络4

大纲 应用简介人脸检测人脸识别通用目标检测图像分割风格迁移总结 讲述CNN典型应用,主要是在机器视觉领域里边,这是它应用最广的一个领域,包含下边几个应用:人脸识别:人脸检测:通用目标检测:图像分割:风格迁移. 检测.分类.分割基本上已经涵盖了图像理解的整体上要完成的一个目标,我们对所有图像的理解无非就是这三类问题:图像分类(判断一个图像它是什么).目标检测(找出图像里边所有的目标,包括它的大小和位置).图像分割(确定每个像素它属于哪一个目标). 卷积神经网络应用简介 先简单介绍一下CNN的应用情