@author:wepon
@blog:http://blog.csdn.net/u012162613/article/details/43221829
轉載:http://blog.csdn.net/u012162613/article/details/43221829
本文介紹多層感知機算法,特別是詳細解讀其代碼實現,基於python theano,代碼來自:Multilayer Perceptron,如果你想詳細了解多層感知機算法,可以參考:UFLDL教程,或者參考本文第一部分的算法簡介。
經詳細注釋的代碼:放在我的github地址上,可下載。
一、多層感知機(MLP)原理簡介
多層感知機(MLP,Multilayer Perceptron)也叫人工神經網絡(ANN,Artificial Neural Network),除了輸入輸出層,它中間可以有多個隱層,最簡單的MLP只含一個隱層,即三層的結構,如下圖:
從上圖可以看到,多層感知機層與層之間是全連接的(全連接的意思就是:上一層的任何一個神經元與下一層的所有神經元都有連接)。多層感知機最底層是輸入層,中間是隱藏層,最后是輸出層。
輸入層沒什么好說,你輸入什么就是什么,比如輸入是一個n維向量,就有n個神經元。
隱藏層的神經元怎么得來?首先它與輸入層是全連接的,假設輸入層用向量X表示,則隱藏層的輸出就是
f(W1X+b1),W1是權重(也叫連接系數),b1是偏置,函數f 可以是常用的sigmoid函數或者tanh函數:


最后就是輸出層,輸出層與隱藏層是什么關系?其實隱藏層到輸出層可以看成是一個多類別的邏輯回歸,也即softmax回歸,所以輸出層的輸出就是softmax(W2X1+b2),X1表示隱藏層的輸出f(W1X+b1)。
MLP整個模型就是這樣子的,上面說的這個三層的MLP用公式總結起來就是,函數G是softmax
因此,MLP所有的參數就是各個層之間的連接權重以及偏置,包括W1、b1、W2、b2。對於一個具體的問題,怎么確定這些參數?求解最佳的參數是一個最優化問題,解決最優化問題,最簡單的就是梯度下降法了(SGD):首先隨機初始化所有參數,然后迭代地訓練,不斷地計算梯度和更新參數,直到滿足某個條件為止(比如誤差足夠小、迭代次數足夠多時)。這個過程涉及到代價函數、規則化(Regularization)、學習速率(learning rate)、梯度計算等,本文不詳細討論,讀者可以參考本文頂部給出的兩個鏈接。
了解了MLP的基本模型,下面進入代碼實現部分。
二、多層感知機(MLP)代碼詳細解讀(基於python+theano)
(1)導入必要的python模塊
主要是numpy、theano,以及python自帶的os、sys、time模塊,這些模塊的使用在下面的程序中會看到。
- import os
- import sys
- import time
- import numpy
- import theano
- import theano.tensor as T
(2)定義MLP模型(HiddenLayer+LogisticRegression)
這一部分定義MLP的基本“構件”,即上文一直在提的HiddenLayer和LogisticRegression
- HiddenLayer
- class HiddenLayer(object):
- def __init__(self, rng, input, n_in, n_out, W=None, b=None,
- activation=T.tanh):
- """
- 注釋:
- 這是定義隱藏層的類,首先明確:隱藏層的輸入即input,輸出即隱藏層的神經元個數。輸入層與隱藏層是全連接的。
- 假設輸入是n_in維的向量(也可以說時n_in個神經元),隱藏層有n_out個神經元,則因為是全連接,
- 一共有n_in*n_out個權重,故W大小時(n_in,n_out),n_in行n_out列,每一列對應隱藏層的每一個神經元的連接權重。
- b是偏置,隱藏層有n_out個神經元,故b時n_out維向量。
- rng即隨機數生成器,numpy.random.RandomState,用於初始化W。
- input訓練模型所用到的所有輸入,並不是MLP的輸入層,MLP的輸入層的神經元個數時n_in,而這里的參數input大小是(n_example,n_in),每一行一個樣本,即每一行作為MLP的輸入層。
- activation:激活函數,這里定義為函數tanh
- """
- self.input = input #類HiddenLayer的input即所傳遞進來的input
- """
- 注釋:
- 代碼要兼容GPU,則W、b必須使用 dtype=theano.config.floatX,並且定義為theano.shared
- 另外,W的初始化有個規則:如果使用tanh函數,則在-sqrt(6./(n_in+n_hidden))到sqrt(6./(n_in+n_hidden))之間均勻
- 抽取數值來初始化W,若時sigmoid函數,則以上再乘4倍。
- """
- #如果W未初始化,則根據上述方法初始化。
- #加入這個判斷的原因是:有時候我們可以用訓練好的參數來初始化W,見我的上一篇文章。
- 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)
- #用上面定義的W、b來初始化類HiddenLayer的W、b
- 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)
- )
- #隱含層的參數
- self.params = [self.W, self.b]
- LogisticRegression
邏輯回歸(softmax回歸),代碼詳解如下。
(如果你想詳細了解softmax回歸,可以參考: DeepLearning tutorial(1)Softmax回歸原理簡介+代碼詳解)
- """
- 定義分類層,Softmax回歸
- 在deeplearning tutorial中,直接將LogisticRegression視為Softmax,
- 而我們所認識的二類別的邏輯回歸就是當n_out=2時的LogisticRegression
- """
- #參數說明:
- #input,大小就是(n_example,n_in),其中n_example是一個batch的大小,
- #因為我們訓練時用的是Minibatch SGD,因此input這樣定義
- #n_in,即上一層(隱含層)的輸出
- #n_out,輸出的類別數
- class LogisticRegression(object):
- def __init__(self, input, n_in, n_out):
- #W大小是n_in行n_out列,b為n_out維向量。即:每個輸出對應W的一列以及b的一個元素。
- 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
- )
- #input是(n_example,n_in),W是(n_in,n_out),點乘得到(n_example,n_out),加上偏置b,
- #再作為T.nnet.softmax的輸入,得到p_y_given_x
- #故p_y_given_x每一行代表每一個樣本被估計為各類別的概率
- #PS:b是n_out維向量,與(n_example,n_out)矩陣相加,內部其實是先復制n_example個b,
- #然后(n_example,n_out)矩陣的每一行都加b
- self.p_y_given_x = T.nnet.softmax(T.dot(input, self.W) + self.b)
- #argmax返回最大值下標,因為本例數據集是MNIST,下標剛好就是類別。axis=1表示按行操作。
- self.y_pred = T.argmax(self.p_y_given_x, axis=1)
- #params,LogisticRegression的參數
- self.params = [self.W, self.b]
ok!這兩個基本“構件”做好了,現在我們可以將它們“組裝”在一起。
我們要三層的MLP,則只需要HiddenLayer+LogisticRegression,
如果要四層的MLP,則為HiddenLayer+HiddenLayer+LogisticRegression........以此類推。
下面是三層的MLP:
- #3層的MLP
- class MLP(object):
- def __init__(self, rng, input, n_in, n_hidden, n_out):
- self.hiddenLayer = HiddenLayer(
- rng=rng,
- input=input,
- n_in=n_in,
- n_out=n_hidden,
- activation=T.tanh
- )
- #將隱含層hiddenLayer的輸出作為分類層logRegressionLayer的輸入,這樣就把它們連接了
- self.logRegressionLayer = LogisticRegression(
- input=self.hiddenLayer.output,
- n_in=n_hidden,
- n_out=n_out
- )
- #以上已經定義好MLP的基本結構,下面是MLP模型的其他參數或者函數
- #規則化項:常見的L1、L2_sqr
- self.L1 = (
- abs(self.hiddenLayer.W).sum()
- + abs(self.logRegressionLayer.W).sum()
- )
- self.L2_sqr = (
- (self.hiddenLayer.W ** 2).sum()
- + (self.logRegressionLayer.W ** 2).sum()
- )
- #損失函數Nll(也叫代價函數)
- self.negative_log_likelihood = (
- self.logRegressionLayer.negative_log_likelihood
- )
- #誤差
- self.errors = self.logRegressionLayer.errors
- #MLP的參數
- self.params = self.hiddenLayer.params + self.logRegressionLayer.params
- # end-snippet-3
MLP類里面除了隱含層和分類層,還定義了損失函數、規則化項,這是在求解優化算法時用到的。
(3)將MLP應用於MNIST(手寫數字識別)
- """
- 加載MNIST數據集
- """
- def load_data(dataset):
- # dataset是數據集的路徑,程序首先檢測該路徑下有沒有MNIST數據集,沒有的話就下載MNIST數據集
- #這一部分就不解釋了,與softmax回歸算法無關。
- data_dir, data_file = os.path.split(dataset)
- if data_dir == "" and not os.path.isfile(dataset):
- # Check if dataset is in the data directory.
- 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':
- import urllib
- origin = (
- 'http://www.iro.umontreal.ca/~lisa/deep/data/mnist/mnist.pkl.gz'
- )
- print 'Downloading data from %s' % origin
- urllib.urlretrieve(origin, dataset)
- print '... loading data'
- #以上是檢測並下載數據集mnist.pkl.gz,不是本文重點。下面才是load_data的開始
- #從"mnist.pkl.gz"里加載train_set, valid_set, test_set,它們都是包括label的
- #主要用到python里的gzip.open()函數,以及 cPickle.load()。
- #‘rb’表示以二進制可讀的方式打開文件
- f = gzip.open(dataset, 'rb')
- train_set, valid_set, test_set = cPickle.load(f)
- f.close()
- #將數據設置成shared variables,主要時為了GPU加速,只有shared variables才能存到GPU memory中
- #GPU里數據類型只能是float。而data_y是類別,所以最后又轉換為int返回
- 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
- #test_mlp是一個應用實例,用梯度下降來優化MLP,針對MNIST數據集
- def test_mlp(learning_rate=0.01, L1_reg=0.00, L2_reg=0.0001, n_epochs=10,
- dataset='mnist.pkl.gz', batch_size=20, n_hidden=500):
- """
- 注釋:
- learning_rate學習速率,梯度前的系數。
- L1_reg、L2_reg:正則化項前的系數,權衡正則化項與Nll項的比重
- 代價函數=Nll+L1_reg*L1或者L2_reg*L2_sqr
- n_epochs:迭代的最大次數(即訓練步數),用於結束優化過程
- dataset:訓練數據的路徑
- n_hidden:隱藏層神經元個數
- batch_size=20,即每訓練完20個樣本才計算梯度並更新參數
- """
- #加載數據集,並分為訓練集、驗證集、測試集。
- datasets = 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]
- #shape[0]獲得行數,一行代表一個樣本,故獲取的是樣本數,除以batch_size可以得到有多少個batch
- n_train_batches = train_set_x.get_value(borrow=True).shape[0] / batch_size
- n_valid_batches = valid_set_x.get_value(borrow=True).shape[0] / batch_size
- n_test_batches = test_set_x.get_value(borrow=True).shape[0] / batch_size
- ######################
- # BUILD ACTUAL MODEL #
- ######################
- print '... building the model'
- #index表示batch的下標,標量
- #x表示數據集
- #y表示類別,一維向量
- index = T.lscalar()
- x = T.matrix('x')
- y = T.ivector('y')
- rng = numpy.random.RandomState(1234)
- #生成一個MLP,命名為classifier
- classifier = MLP(
- rng=rng,
- input=x,
- n_in=28 * 28,
- n_hidden=n_hidden,
- n_out=10
- )
- #代價函數,有規則化項
- #用y來初始化,而其實還有一個隱含的參數x在classifier中
- cost = (
- classifier.negative_log_likelihood(y)
- + L1_reg * classifier.L1
- + L2_reg * classifier.L2_sqr
- )
- #這里必須說明一下theano的function函數,givens是字典,其中的x、y是key,冒號后面是它們的value。
- #在function被調用時,x、y將被具體地替換為它們的value,而value里的參數index就是inputs=[index]這里給出。
- #下面舉個例子:
- #比如test_model(1),首先根據index=1具體化x為test_set_x[1 * batch_size: (1 + 1) * batch_size],
- #具體化y為test_set_y[1 * batch_size: (1 + 1) * batch_size]。然后函數計算outputs=classifier.errors(y),
- #這里面有參數y和隱含的x,所以就將givens里面具體化的x、y傳遞進去。
- test_model = theano.function(
- inputs=[index],
- outputs=classifier.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(
- inputs=[index],
- outputs=classifier.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]
- }
- )
- #cost函數對各個參數的偏導數值,即梯度,存於gparams
- gparams = [T.grad(cost, param) for param in classifier.params]
- #參數更新規則
- #updates[(),(),()....],每個括號里面都是(param, param - learning_rate * gparam),即每個參數以及它的更新公式
- updates = [
- (param, param - learning_rate * gparam)
- for param, gparam in zip(classifier.params, gparams)
- ]
- train_model = theano.function(
- inputs=[index],
- outputs=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]
- }
- )
- ###############
- # 開始訓練模型 #
- ###############
- print '... training'
- patience = 10000
- patience_increase = 2
- #提高的閾值,在驗證誤差減小到之前的0.995倍時,會更新best_validation_loss
- improvement_threshold = 0.995
- #這樣設置validation_frequency可以保證每一次epoch都會在驗證集上測試。
- validation_frequency = min(n_train_batches, patience / 2)
- best_validation_loss = numpy.inf
- best_iter = 0
- test_score = 0.
- start_time = time.clock()
- #epoch即訓練步數,每個epoch都會遍歷所有訓練數據
- epoch = 0
- done_looping = False
- #下面就是訓練過程了,while循環控制的時步數epoch,一個epoch會遍歷所有的batch,即所有的圖片。
- #for循環是遍歷一個個batch,一次一個batch地訓練。for循環體里會用train_model(minibatch_index)去訓練模型,
- #train_model里面的updatas會更新各個參數。
- #for循環里面會累加訓練過的batch數iter,當iter是validation_frequency倍數時則會在驗證集上測試,
- #如果驗證集的損失this_validation_loss小於之前最佳的損失best_validation_loss,
- #則更新best_validation_loss和best_iter,同時在testset上測試。
- #如果驗證集的損失this_validation_loss小於best_validation_loss*improvement_threshold時則更新patience。
- #當達到最大步數n_epoch時,或者patience<iter時,結束訓練
- while (epoch < n_epochs) and (not done_looping):
- epoch = epoch + 1
- for minibatch_index in xrange(n_train_batches):#訓練時一個batch一個batch進行的
- minibatch_avg_cost = train_model(minibatch_index)
- # 已訓練過的minibatch數,即迭代次數iter
- iter = (epoch - 1) * n_train_batches + minibatch_index
- #訓練過的minibatch數是validation_frequency倍數,則進行交叉驗證
- if (iter + 1) % validation_frequency == 0:
- # compute zero-one loss on validation set
- validation_losses = [validate_model(i) for i
- in xrange(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.
- )
- )
- #當前驗證誤差比之前的都小,則更新best_validation_loss,以及對應的best_iter,並且在tsetdata上進行test
- 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 xrange(n_test_batches)]
- test_score = numpy.mean(test_losses)
- print((' epoch %i, minibatch %i/%i, test error of '
- 'best model %f %%') %
- (epoch, minibatch_index + 1, n_train_batches,
- test_score * 100.))
- #patience小於等於iter,則終止訓練
- if patience <= iter:
- done_looping = True
- break
- end_time = time.clock()
- print(('Optimization complete. Best validation score of %f %% '
- 'obtained at iteration %i, with test performance %f %%') %
- (best_validation_loss * 100., best_iter + 1, test_score * 100.))
- print >> sys.stderr, ('The code for file ' +
- os.path.split(__file__)[1] +
- ' ran for %.2fm' % ((end_time - start_time) / 60.))