10장: Artificial Neural Network (p.327)
인공신경망 : 개념적인 설명은 Deep Learning from Scratch 참고
코드구현 중심
고수준API로 MNIST 훈련:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
(X_train, y_train), (X_test, y_test) = mnist.load_mnist() X_train = X_train.astype(np.float32).reshape(-1, 28*28) / 255.0 X_test = X_test.astype(np.float32).reshape(-1, 28*28) / 255.0 y_train = y_train.astype(np.int32) y_test = y_test.astype(np.int32) X_valid, X_train = X_train[:5000], X_train[5000:] y_valid, y_train = y_train[:5000], y_train[5000:] feature_cols = tf.contrib.learn.infer_real_valued_columns_from_input(X_train) dnn_clf = tf.contrib.learn.DNNClassifier(hidden_units=[300,100], n_classes=10, feature_columns=feature_cols) dnn_clf = tf.contrib.learn.SKCompat(dnn_clf) dnn_clf.fit(X_train,y_train, batch_size=50, steps=40000) from sklearn.metrics import accuracy_score y_pred = dnn_clf.predict(X_test) print(accuracy_score(y_test,y_pred['classes'])) |
한 5분 걸림.. 오래걸림
(사용한 함수에 대해 정확히 파악하는 중점으로, 내가 몰랐던 함수까지)
저수준 API로 심층 신경망 훈련 (전체 코드):
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 |
#----------------------------------------------- (X_train, y_train), (X_test, y_test) = mnist.load_mnist(normalize=True) y_train = y_train.astype(np.int32) y_test = y_test.astype(np.int32) X_valid, X_train = X_train[:5000], X_train[5000:] y_valid, y_train = y_train[:5000], y_train[5000:] #X_train(55000,784) y_train(55000,) 훈련시킬 데이터셋 #X_test(10000,784), y_test(10000,) 테스트데이터셋 #X_valid(5000,784), y_valid(5000,) 테스트전 검증할 데이터셋 n_inputs = 28 * 28 # X데이터 shape(28*28,) 이렇게 생김 n_hidden1 = 300 #1층의 layer n_hidden2 = 100 #2층의 layer n_outputs = 10 # softmax 사용할 것이고 숫자가 10개 reset_graph() #미니배치형식으로 담아돌릴 컨테이너 제작 X = tf.placeholder(tf.float32, shape=(None,n_inputs),name="X") y = tf.placeholder(tf.int32, shape=(None),name="y") def neuron_layer(X, n_neurons, name, activation=None): with tf.name_scope(name): n_inputs = int(X.get_shape()[1]) #예 (55000,784)X가 n_input=784 stddev = 2 / np.sqrt(n_inputs) #Xavier초기화를 사용한거같음 2는 layer인가? #가중치의 손실을 막기위해 절사평균사용 init = tf.truncated_normal((n_inputs, n_neurons), stddev=stddev) W = tf.Variable(init, name="kernel") #kernel이란 단어를 왜사용했나 맘에안듬 #(m,n) x (n,k) = (m,k) k개의 특성 수 식이 형성되므로 bias도 0으로 초기화하고 그 수만큼세팅 b = tf.Variable(tf.zeros([n_neurons]), name="bias") #(m,n) x (n,k) = m,k (m,k) + b = Z Z = (m,k) Z = tf.matmul(X,W) + b if activation is not None: return activation(Z) else: return Z #층을 이름범위로 묶기 with tf.name_scope("dnn"): #X = 미니배치로 받은 데이터, n_hidden1 = 300 hidden1 = neuron_layer(X, n_hidden1, name="hidden1", activation=tf.nn.relu) # 구현된 relu함수사용 #hidden1 = (55000,300) n_hidden2 = 100 hidden2 = neuron_layer(hidden1, n_hidden2, name="hidden2", activation=tf.nn.relu) #hidden2 = (55000,100) n_outputs = 10 logits = neuron_layer(hidden2, n_outputs, name="outputs") #logits = (55000,10) with tf.name_scope("loss"): #softmax함수를 사용한 loss 값 xentropy = tf.nn.sparse_softmax_cross_entropy_with_logits(labels=y, logits=logits) loss = tf.reduce_mean(xentropy, name="loss") learning_rate = 0.01 with tf.name_scope("train"): optimizer = tf.train.GradientDescentOptimizer(learning_rate) training_op = optimizer.minimize(loss) with tf.name_scope("eval"): correct = tf.nn.in_top_k(logits, y, 1) accuracy = tf.reduce_mean(tf.cast(correct, tf.float32)) init = tf.global_variables_initializer() saver = tf.train.Saver() n_epochs = 30 batch_size = 50 def shuffle_batch(X, y, batch_size): rnd_idx = np.random.permutation(len(X)) #예를들어 55000개 n_batches = len(X) // batch_size #len(X)는 X의 행데이터 1100 for batch_idx in np.array_split(rnd_idx, n_batches): X_batch, y_batch = X[batch_idx], y[batch_idx] yield X_batch, y_batch with tf.Session() as sess: init.run() start_time = time.time() for epoch in range(n_epochs):#30번 for X_batch, y_batch in shuffle_batch(X_train, y_train, batch_size): sess.run(training_op, feed_dict={X: X_batch, y: y_batch}) acc_batch = accuracy.eval(feed_dict={X: X_batch, y: y_batch}) acc_val = accuracy.eval(feed_dict={X: X_valid, y: y_valid}) print(epoch, "Batch accuracy:", acc_batch, "Val accuracy:", acc_val) end_time = time.time() total_time = end_time - start_time save_path = saver.save(sess, "C:\\Users\\고영민\\workspace\\parametersave\\dnn.ckpt") total_time = round(total_time) print("총 걸린시간 (초):",round(total_time)) |
이제 모르는 부분에 대해서 하나씩 따져보자
shuffle_batch 함수정의에서 :
|
1 2 |
print("len(X_train) :",len(X_train)) print(X_train.shape) |
len함수는 데이터의 행의 개수만 출력
|
1 2 3 4 |
a = (len(X_train) / 50) print("/가 하나만 들어간 연산 :",a) a = (len(X_train) // 50) print("/가 두개인 연산 :",a) |
즉 슬래쉬(/)가 두 개인 경우 소수점을 뗀 int형태가 된다.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
(X_train, y_train), (X_test, y_test) = mnist.load_mnist(normalize=True) y_train = y_train.astype(np.int32) y_test = y_test.astype(np.int32) X_valid, X_train = X_train[:5000], X_train[5000:] y_valid, y_train = y_train[:5000], y_train[5000:] X = X_train rnd_idx = np.random.permutation(len(X)) #예를들어 55000개 print("rnd_idx 개수:",len(rnd_idx)) n_batches = len(X) // 100 #len(X)는 X의 행데이터 print("n_batches 개수 :",n_batches) b = np.array_split(rnd_idx,n_batches) print("np.array_split 개수 :",np.shape(b)) |
rnd_idx 는 1~ 55000 숫자가 랜덤하게 배열되어있는 배열이다. 그 배열을 np.array_split 함수를 통해 n_batches (550)개로 나눈 것이다. 즉 rnd_idx : [22134 … 34850] 배열 구간을 550개로 나눈 100개가 된다. for batch_index in np.array_split(rnd_idx, n_batches) 는 위에 나눠진 100개 중에 첫 번째 100개가 들어가 있는 배열부터 batch_index에 전해주게 된다.
마지막으로 yield가 있는데 https://kkamikoon.tistory.com/90 잘 설명해 놓은 사이트 참조.
tf.nn.sparse_softmax_cross_entropy_with_logits(labels=y,logits=logits) 함수 :
labels와 logits의 softmax_cross_entropy 를 계산해준다.
매개변수 :
- labels : Tensor or shape [ d_0 … d_{r-1}] r 차원의 정답레이블. dtype=int32,64형을 사용
- logits : 스케일링 되지 않은 로그확률시킬 값. dtype=tf.float형
부수적인 설명 :
tf.nn.softmax(logits, axis=None, name=None, dim=None) 함수기초
식 : softmax = tf.exp(logits) / tf.reduce_sum(tf.exp(logits), axis)
logit이 커지면 부동소수점 반올림 오차로 소프트맥스 출력이 0 또는 1이될 수 있다. 이런 경우 크로스 엔트로피 함수 공식에 음의 무한대가 되는 log(0)이 포함된다. 이를 막기 위해 작은 양수 ε에 대한 log(ε)을 계산함으로써 해결한다. one_hot_vector 형태인 경우는 softmax_cross_entropy_with_logits()을 사용한다.
tf.nn.in_top_k(predictions, targets, k, name=None) :
책 내용 :
in_top_k(predictions, targets, k) 함수는 예측값(predictions)과 타깃 레이블(targets)를 입력받아 타깃 레이블의 예측값이 크기순으로 k번째 안에 들면 True 그렇지 않으면 False를 return 한다.
내가한 이해 :
predictions (예측값) 와 targets( 정답값) 을 인수로 받는다. 하지만 tf.nn.in_top_k(logits,y,1) 의 형태를 따라가보면 X_batch와 y_batch를 feed_dict로 받기 때문에 그 행렬의 형태는 logits = (1100,10) , y = (1100,)인 형태를 볼 수 있다. 위 함수는 (내가 잘못이해하는 것일 수 도 있다.) k 를 1로 설정하여 최고의 예측 값이 맞았을 때만 참으로 평가한다. 여기서 행렬의 형태가 다른데 어떻게 비교를 하지 라고 생각하다 아래의 코드를 보면서 y 를 자동으로 one_hot_encoding 시켜서 [ 0 0 0 0 0 1 0 0 0 0 ] 처럼 만들어 동작한 것이라 유추해보았다.(나중에 이해를 좀 확실하게 다시)
|
1 2 3 4 5 6 7 8 9 10 11 12 |
logit = np.random.normal((X_test)) print("logit shape :",logit.shape) print("y_test shape :",y_test.shape) a = tf.nn.in_top_k(logit, y_test, 1) init = tf.global_variables_initializer() with tf.Session() as sess: sess.run(init) result = sess.run(a) print(result) |
마지막으로 tf.cast 함수는 True, False 로 받을 값을 False = 0 True = 1로 바꿔준다.
최종 결과 :
위의 고수준 API 사용보다 걸린시간은 3분정도 줄었고 정확도 또한 7% 정도 증가했다.
훈련된 신경망 사용하기
이렇게 훈련시켜 놓은 신경망 가중치들을 저장해뒀으니 불러들여 사용하는 법을 알아본다.
|
1 2 3 4 5 6 7 |
with tf.Session() as sess: saver.restore(sess, "C:\\Users\\고영민\\workspace\\parametersave\\dnn.ckpt") X_new_scaled = X_test[:20] Z = logits.eval(feed_dict={X:X_new_scaled}) y_pred = np.argmax(Z, axis=1) print("저장된 가중치를 사용한 예측 답:",y_pred) print("실제 답 :",y_test[:20]) |
원래의 구성단계를 그대로 사용하고 실행단계만 바꾼 것( 맨끝에 전체 코드 개재 )
tf.train.Saver().save(sess, save_path, global_step=None, latest_filename=None, meta_graph_suffix=’meta’, write_meta_graph=True, write_state=True, strip_default_attrs=False)
중요한 매개변수 : (변수들 초기화 필요)
sess : 변수를 저장할 때 사용하는 세션
save_path : 변수를 저장할 파일위치, 새롭게 생길파일이름도
tf.train.Saver().restore(sess, save_path)
저장한 변수를 불러들이는 함수 (초기화가 필요없음)
sess : 매개변수를 복원하는데 사용한다.
save_path : 이전에 변수가 저장된 경로
saver.save()로 학습된 가중치 (variable)를 저장하고 불러들여 사용할 때는 saver.restore()로 저장되있는 변수를 불러와 variable에 매칭시킨다.
마무리로 훈련된 가중치들을 test dataset(10000개 데이터) 에 적용했을 때 결과다.
|
1 2 3 4 5 |
with tf.Session() as sess: saver.restore(sess,"C:\\Users\\고영민\\workspace\\parametersave\\dnn.ckpt") result = accuracy.eval(feed_dict={X:X_test, y:y_test}) print(result) |
신경망 하이퍼파라미터 튜닝하기 (과연 그런가? 밝혀내고 싶은 것들)
위에 본 예시로 짠 코드 조차도 파라미터 개수가 매우 많은 것을 볼 수 있다. 층 수, 층 마다의 뉴런 수, 각 층에서 활용하는 활성화 함수, 가중치 초기화 방법 등 엄청 많은 방법들로 이 신경망을 튜닝할 수 있다. 과연 이 많은 방법 중에서 최적의 조합인 파라미터를 찾아내는 방법이 있을까?
심층 신경망은 복잡한 함수를 모델링하는 데 얕은 신경망보다 훨씬 적은 수의 뉴련을 사용하기 때문에 더 빠르게 훈련된다. 아래 층에서 위층으로 올라갈수록 저수준 구조에서 고수준 구조를 모델링한다. 이는 base가 동일한 문제를 모델링한다면 저수준 구조에서의 가중치는 비슷하기 때문에 바로 고수준 구조의 학습을 할 수 있다. 비슷한 작업에서 가장 뛰어난 성능을 내는 미리 훈련된 네트워크 일부를 재사용하는 것이 일반적이다.
은닉층의 뉴런 수를 보자면 요즘엔 모든 은닉층에 같은 크기를 사용한다. 하이퍼파라미터가 층마다 따로 있지않아서 전체를 통틀어 하나만 조정하면 된다. 일반적으로 층의 뉴런 수보다 층 수를 늘리는 쪽이 이득이 많다. 완벽한 뉴런 수 찾기는 아직 밝혀지지 않았고 , 직관적인 방법으로 층의 개수와 마찬가지로 수를 늘려보다가 과대적합이 나타나면 그 전까지 점진적으로 늘리는 방법이 있다.
단순 접근 방식은 실제 필요한 것보다 더 많은 층과 뉴런을 가진 모델을 선택하고 과대적합이 되지 않도록 조기 종료 기법을 사용하는 것이다.(특히 dropout방법) 이를 stretch pants방식이라고 부른다. 맞는 사이즈 입어보는게 아니고 그냥 큰 사이즈를 입어서 알맞게 줄이는 방식이다.
전체코드 :
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 |
import numpy as np import tensorflow as tf import mnist import time import os def reset_graph(seed=42): tf.reset_default_graph() tf.set_random_seed(seed) np.random.seed(seed) #from tensorflow.examples.tutorials.mnist import input_data #mnist = input_data.read_data_sets("./samples/MNIST_data/", one_hot=True) #----------------------------------------------- (X_train, y_train), (X_test, y_test) = mnist.load_mnist(normalize=True) y_train = y_train.astype(np.int32) y_test = y_test.astype(np.int32) X_valid, X_train = X_train[:5000], X_train[5000:] y_valid, y_train = y_train[:5000], y_train[5000:] #X_train(55000,784) y_train(55000,) 훈련시킬 데이터셋 #X_test(10000,784), y_test(10000,) 테스트데이터셋 #X_valid(5000,784), y_valid(5000,) 테스트전 검증할 데이터셋 n_inputs = 28 * 28 # X데이터 shape(28*28,) 이렇게 생김 n_hidden1 = 300 #1층의 layer n_hidden2 = 100 #2층의 layer n_outputs = 10 # softmax 사용할 것이고 숫자가 10개 reset_graph() #미니배치형식으로 담아돌릴 컨테이너 제작 X = tf.placeholder(tf.float32, shape=(None,n_inputs),name="X") y = tf.placeholder(tf.int32, shape=(None),name="y") def neuron_layer(X, n_neurons, name, activation=None): with tf.name_scope(name): n_inputs = int(X.get_shape()[1]) #예 (55000,784)X가 n_input=784 stddev = 2 / np.sqrt(n_inputs) #Xavier초기화를 사용한거같음 2는 layer인가? #가중치의 손실을 막기위해 절사평균사용 init = tf.truncated_normal((n_inputs, n_neurons), stddev=stddev) W = tf.Variable(init, name="kernel") #kernel이란 단어를 왜사용했나 맘에안듬 #(m,n) x (n,k) = (m,k) k개의 특성 수 식이 형성되므로 bias도 0으로 초기화하고 그 수만큼세팅 b = tf.Variable(tf.zeros([n_neurons]), name="bias") #(m,n) x (n,k) = m,k (m,k) + b = Z Z = (m,k) Z = tf.matmul(X,W) + b if activation is not None: return activation(Z) else: return Z #층을 이름범위로 묶기 with tf.name_scope("dnn"): #X = 미니배치로 받은 데이터, n_hidden1 = 300 hidden1 = neuron_layer(X, n_hidden1, name="hidden1", activation=tf.nn.relu) # 구현된 relu함수사용 #hidden1 = (55000,300) n_hidden2 = 100 hidden2 = neuron_layer(hidden1, n_hidden2, name="hidden2", activation=tf.nn.relu) #hidden2 = (55000,100) n_outputs = 10 logits = neuron_layer(hidden2, n_outputs, name="outputs") #logits = (55000,10) with tf.name_scope("loss"): #softmax함수를 사용한 loss 값 xentropy = tf.nn.sparse_softmax_cross_entropy_with_logits(labels=y, logits=logits) loss = tf.reduce_mean(xentropy, name="loss") learning_rate = 0.01 with tf.name_scope("train"): optimizer = tf.train.GradientDescentOptimizer(learning_rate) training_op = optimizer.minimize(loss) with tf.name_scope("eval"): correct = tf.nn.in_top_k(logits, y, 1) accuracy = tf.reduce_mean(tf.cast(correct, tf.float32)) init = tf.global_variables_initializer() saver = tf.train.Saver() n_epochs = 30 batch_size = 50 def shuffle_batch(X, y, batch_size): rnd_idx = np.random.permutation(len(X)) #예를들어 55000개 n_batches = len(X) // batch_size #len(X)는 X의 행데이터 1100 for batch_idx in np.array_split(rnd_idx, n_batches): X_batch, y_batch = X[batch_idx], y[batch_idx] yield X_batch, y_batch """" with tf.Session() as sess: init.run() start_time = time.time() for epoch in range(n_epochs):#30번 for X_batch, y_batch in shuffle_batch(X_train, y_train, batch_size): sess.run(training_op, feed_dict={X: X_batch, y: y_batch}) acc_batch = accuracy.eval(feed_dict={X: X_batch, y: y_batch}) acc_val = accuracy.eval(feed_dict={X: X_valid, y: y_valid}) print(epoch, "Batch accuracy:", acc_batch, "Val accuracy:", acc_val) end_time = time.time() total_time = end_time - start_time save_path = saver.save(sess, "C:\\Users\\고영민\\workspace\\parametersave\\dnn.ckpt") total_time = round(total_time) print("총 걸린시간 (초):",round(total_time)) """"" """" with tf.Session() as sess: saver.restore(sess, "C:\\Users\\고영민\\workspace\\parametersave\\dnn.ckpt") X_new_scaled = X_test[:20] Z = logits.eval(feed_dict={X:X_new_scaled}) y_pred = np.argmax(Z, axis=1) print("저장된 가중치를 사용한 예측 답:",y_pred) print("실제 답 :",y_test[:20]) """ with tf.Session() as sess: saver.restore(sess,"C:\\Users\\고영민\\workspace\\parametersave\\dnn.ckpt") result = accuracy.eval(feed_dict={X:X_test, y:y_test}) print(result) |
References : Hands-On Machine Learning with Scikit Learn & TensorFlow
