머신 러닝 - 다층 퍼셉트론 손글씨 분류 실습

- 다층 퍼셉트론을 활용한 손글씨 분류 실습

0부터 9까지 손글씨를 가지고 다층 퍼셉트론을 활용해 0부터 9를 분류하는 AI를 만들어보겠습니다. 

 

학습 데이터는 MINIST 데이터를 사용할 건데, MINIST 데이터는 28 * 28 스케일을 가지고 있습니다. 따라서 입력값은 28*28 = 784개가 될 것이며, 출력값은 0부터9까지인 10개가 될 것입니다. 지금부터 만들 다층 퍼셉트론의 구조입니다.

 

첫번째 레이어에서 784개의 입력을 받아 256개의 출력을 내고 두번째 레이어에서는 256개의 입력을 받아 128개의 출력을 냅니다. 그리고 3번째 레이어에서 128개의 입력으로 10개의 출력값을 내게됩니다. 이후 adam optimizer를 통해 역전파로 최적화를 하게되고 이때 사용하는 손실함수는 크로스 엔트로피 함수가 됩니다. 또한 드롭아웃을 2번째 레이어에 걸게되고 조기 종료를 통해 과대적합을 피해보도록 하겠습니다.

 

먼저 MINIST 데이터를 불러옵니다.

(x_train, y_train), (x_test, y_test) = tf.keras.datasets.mnist.load_data()

x_val  = x_train[50000:60000]
x_train = x_train[0:50000]
y_val  = y_train[50000:60000]
y_train = y_train[0:50000]

 

10000개를 검증 데이터로 쓰고 50000개를 학습데이터로 사용하겠습니다. 그다음은 데이터 구조를 변경해주겠습니다. AI 학습용도에 맞게 변경합니다.

x_train = x_train.reshape(50000, 784)
x_val = x_val.reshape(10000, 784)
x_test = x_test.reshape(10000, 784)

 

이렇게 하면 784개의 데이터로 1차원 배열로 변경됩니다.

 

하나의 데이터는 그레이스 케일이라고 하는 0(검정)~255(흰색)을 띄고 있습니다. 모든 값을 0에서 1사이의 값으로 정규화하여 더 학습시간을 단축하고 더 나은 성능을 이끌어내겠습니다.

x_train = x_train.astype('float32')
x_val = x_val.astype('float32')
x_test = x_test.astype('float32')

gray_scale = 255
x_train /= gray_scale
x_val /= gray_scale
x_test /= gray_scale

 

손실함수에서 크로스 엔트로피를 계산하기위해 실제값을 one hot encoding 방식으로 변경해줍니다.

num_classes = 10
y_train = tf.keras.utils.to_categorical(y_train, num_classes)
y_val = tf.keras.utils.to_categorical(y_val, num_classes)
y_test = tf.keras.utils.to_categorical(y_test, num_classes)

 

이제 뉴런을 만들고 드랍아웃을 설정해줍니다.

x = tf.placeholder(tf.float32, [None, 784])
y = tf.placeholder(tf.float32, [None, 10])
keep_prob = tf.placeholder(tf.float32)

def mlp(x):
    # hidden layer1
    w1 = tf.Variable(tf.random_uniform([784,256]))
    b1 = tf.Variable(tf.zeros([256]))
    h1 = tf.nn.relu(tf.matmul(x, w1) + b1)
    # hidden layer2
    w2 = tf.Variable(tf.random_uniform([256,128]))
    b2 = tf.Variable(tf.zeros([128]))
    h2 = tf.nn.relu(tf.matmul(h1, w2) + b2)
    h2_drop = tf.nn.dropout(h2, keep_prob)
    # output layer
    w3 = tf.Variable(tf.random_uniform([128,10]))
    b3 = tf.Variable(tf.zeros([10]))
    logits= tf.matmul(h2_drop, w3) + b3
    
    return logits
    
# 모델 예측
logits = mlp(x)

# 손실 함수 정의
loss_op = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits_v2(
    logits=logits, labels=y))

# 최적화 함수 설정과 학습 연산 정의
train_op = tf.train.AdamOptimizer(learning_rate=0.01).minimize(loss_op)

 

이제 조기 종료를 설정해줍니다. 매 주기(Epoch)마다 검증 데이터로 검증 정확도를 측정합니다.
검증 정확도가 5번 연속으로 최고 검증 정확도보다 높지 않을 때 조기 종료를 수행합니다.

# TensorFlow 변수 초기화
init = tf.global_variables_initializer()

# 모든 변수를 저장하고 복원하기 위한 Saver 생성
saver = tf.train.Saver()

# 학습에 사용할 하이퍼파라미터 설정

# 학습 에포크 횟수
epoch_cnt = 300

# 배치 크기
batch_size = 1000

# 한 번의 에포크당 반복 횟수
iteration = len(x_train) // batch_size

# 조기 종료(early stopping)를 위한 임계값과 카운트 초기화
earlystop_threshold = 5
earlystop_cnt = 0

 

데이터를 모델에 입력시킬 때(feed), 드랍아웃이 있을 경우, 항상 keep_prob를 설정해주셔야합니다.
학습 시, 10%의 드랍아웃을 하기 위해, keep_prob를 0.9로 설정합니다.
테스트 시, 드랍 아웃을 사용하지 않을 것이므로, keep_prob를 1.0으로 설정합니다.

 

# TensorFlow 세션 시작
with tf.Session() as sess:
    # 변수 초기화
    sess.run(init)
    
    # 이전 훈련 정확도 및 최대 검증 정확도 초기화
    prev_train_acc = 0.0
    max_val_acc = 0.0
    
    # 주어진 에포크 횟수만큼 반복
    for epoch in range(epoch_cnt):
        avg_loss = 0.0  # 평균 손실 초기화
        start = 0
        end = batch_size
        
        # 배치 단위로 학습 수행
        for i in range(iteration):
            _, loss = sess.run([train_op, loss_op], 
                               feed_dict={x: x_train[start: end], 
                                          y: y_train[start: end], 
                                          keep_prob: 0.9})
            start += batch_size
            end += batch_size
            avg_loss += loss / iteration  # 훈련 평균 손실 누적
            
        # 모델 검증
        preds = tf.nn.softmax(logits)  # 로짓에 소프트맥스 적용
        correct_prediction = tf.equal(tf.argmax(preds, 1), tf.argmax(y, 1))
        accuracy = tf.reduce_mean(tf.cast(correct_prediction, "float"))  # 정확도 계산
        
        # 훈련 및 검증 정확도 계산
        cur_train_acc = accuracy.eval({x: x_train, y: y_train, keep_prob: 1.0})
        cur_val_acc = accuracy.eval({x: x_val, y: y_val, keep_prob: 1.0})
        
        # 현재 에포크의 결과 출력
        print("epoch: " + str(epoch) +
              ", train acc: " + str(cur_train_acc) +
              ", val acc: " + str(cur_val_acc))
              #', train loss: ' + str(avg_loss) +
              #', val loss: ' + str(cur_val_loss))
        
        # 조기 종료 조건 확인
        if cur_val_acc < max_val_acc:
            if cur_train_acc > prev_train_acc or cur_train_acc > 0.99:
                if earlystop_cnt == earlystop_threshold:
                    print("early stopped on " + str(epoch))
                    break
                else:
                    print("overfitting warning: " + str(earlystop_cnt))
                    earlystop_cnt += 1
            else:
                earlystop_cnt = 0
        else:
            earlystop_cnt = 0
            max_val_acc = cur_val_acc
            # 모델 변수를 파일에 저장
            save_path = saver.save(sess, "model/model.ckpt")
            
        # 현재 훈련 정확도를 저장
        prev_train_acc = cur_train_acc

 

이제 테스트해보겠습니다.

# Start testing
with tf.Session() as sess:
    # Restore variables from disk.
    saver.restore(sess, "model/model.ckpt")
    correct_prediction = tf.equal(tf.argmax(preds, 1), tf.argmax(y, 1))
    # Calculate accuracy
    accuracy = tf.reduce_mean(tf.cast(correct_prediction, "float"))
    print("[Test Accuracy] :", accuracy.eval({x: x_test, y: y_test, keep_prob: 1.0}))

 

정확도가 0.9548임을 알 수 있습니다.