1.5 神经网络入门-神经元实现

410 阅读1分钟

1.5 神经元实现

  • 分拆数据集

    def load_data(filename):
        """read data from data file."""
        with open(filename, 'rb') as f:
            data = pickle.load(f, encoding='bytes')
            return data[b'data'], data[b'labels']
    
    # trensorflow.DataSet
    class CifarData:
        def __init__(self, filenames, need_shuffle):
            all_data = []
            all_labels = []
            for filename in filenames:
                data,labels = load_data(filename)
                for item,label in zip(data,labels):
                    if label in [0,1]:
                        all_data.append(item)
                        all_labels.append(label)
            self._data = np.vstack(all_data)
            # 归一化,将0-255的数归一成0-1直接的数
            self._data = self._data / 127.5 - 1 
            self._labels = np.hstack(all_labels)
            self._num_examples = self._data.shape[0]
            self._need_shuffle = need_shuffle
            self._indicator = 0
            if self._need_shuffle:
                self._shuffle_data()
            
        def _shuffle_data(self):
            # 混排 [0,1,2,3,4,5] -> [2,1,4,0,3,5]
            p = np.random.permutation(self._num_examples)
            self._data = self._data[p]
            self._labels = self._labels[p]
        
        def next_batch(self, batch_size):
            """return batch_size examples as a batch."""
            end_indicator = self._indicator + batch_size
            if end_indicator > self._num_examples:
                if self._need_shuffle:
                    self._shuffle_data()
                    self._indicator = 0
                    end_indicator = batch_size
                else:
                    raise Exception("have no more examples")
            if end_indicator > self._num_examples:
                raise Exception("batch size is lager then all examples")
            batch_data = self._data[self._indicator:end_indicator]
            batch_labels = self._labels[self._indicator:end_indicator]
            self._indicator = end_indicator
            return batch_data, batch_labels
            
    train_filename = [os.path.join(CIFAR_DIR,'data_batch_%d' % i) for i in range(1,6)]
    test_filenames = [os.path.join(CIFAR_DIR, 'test_batch')]
    
    train_data = CifarData(train_filename, True)
    test_data = CifarData(test_filenames, False)
    
    batch_data,batch_labels = train_data.next_batch(10)
    
  • 测试算法准确率

    init = tf.global_variables_initializer()
    batch_size = 20
    train_steps = 100000
    test_steps = 100
    
    with tf.Session() as sess:
        sess.run(init)
        for i in range(train_steps):
            batch_data, batch_labels = train_data.next_batch(batch_size)
            loss_val, acc_val, _ = sess.run(
                [loss, accuracy, train_op],
                feed_dict={
                    x: batch_data,
                    y: batch_labels})
            if (i+1) % 500 == 0:
                print ('[Train] Step: %d, loss: %4.5f, acc: %4.5f' \
                    % (i+1, loss_val, acc_val))
                    
            if (i+1) % 5000 == 0:
                test_data = CifarData(test_filenames, False)
                all_test_acc_val = []
                for j in range(test_steps):
                    test_batch_data, test_batch_labels \
                        = test_data.next_batch(batch_size)
                    test_acc_val = sess.run(
                        [accuracy],
                        feed_dict = {
                            x: test_batch_data, 
                            y: test_batch_labels
                        })
                    all_test_acc_val.append(test_acc_val)
                test_acc = np.mean(all_test_acc_val)
                print('[Test ] Step: %d, acc: %4.5f' % (i+1, test_acc))