首页 > 代码库 > 深度学习算法实践12---卷积神经网络(CNN)实现

深度学习算法实践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=http://www.mamicode.com/W_values, name=‘W‘, borrow=True)>接下来我们定义逻辑回归算法类:

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=http://www.mamicode.com/numpy.zeros(  >

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

做完上述准备工作之后,我们就可以开始卷积神经网络(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=http://www.mamicode.com/b_values, borrow=True)> 上面代码实现了对输入信号的卷积操作,并对结果进行最大化池化。

下面我们来看怎样初始化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%的结果。

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