(26). 합성곱 신경망 구현

합성곱 신경망

28 x 28 크기의 흑백 이미지와 3 x 3 크기의 커널 10개로 합성곱을 수행한다.
그런 다음 2 x 2 크기의 최대 풀링을 수행하여 14 x 14 x 10로 특성맵의 크기를 줄인다.
이 특성 맵을 일렬로 펼쳐서 100개의 뉴런을 가진 완전 연결층과 연결 시킬 것이다.

 

정방향 계산

이번에 구현할 합성곱 신경망 클래스는 ConvolutionNetwork이다.
합성곱과 렐루 함수 그리고 풀링이 적용되는 부분을 주의깊게 보면 된다.
MultiClassNetwork 클래스의 forpass() 메서드에 있던 z1, a1, z2를 계산하는 식은 그대로 두고 그 앞에 합성곱과 풀링층을 추가하여 코드를 작성해 보자.

class ConvolutionNetwork(MultiClassNetwork):
  def forpass(slef, x):
    c_out1 = tf.nn.conv2d(x, self.conv_w, strides=1, padding='SAME') + self.conv_b

합성곱을 수행하는 conv2d()함수에 전달한 매개변수 값은 다음과 같다.

• self.conv_w
  합성곱에 사용할 가중치이다. 3 x 3 x 1 크기의 커널을 10개로 사용하므로 가중치의 전체 크기는 3 x 3 x 1 x 10이다.
• stride, padding
  특성 맵의 가로와 세로 크기를 일정하게 만들기 위하여 stride는 1, padding은 'SAME'으로 지정했다.

 

렐루 함수 적용

합성곱 계산을 수행한 다음 렐루 함수를 적용한다.

  def forpass(slef, x):
    z1 = np.dot(x, self.w1) + self.b1 #첫 번쨰 층 선형식 계산
    self.a1 = self.sigmoid(z1) #활성화 함수
    z2 = np.dot(self.a1, self.w2) + self.b2 #두번째 선형식 계산
    c_out1 = tf.nn.conv2d(x, self.conv_w, strides=1, padding='SAME') + self.conv_b
    r_out = tf.nn.relu(c_out)

    return z2

 

풀링 적용하고 완전 연결층 수정

max_pool2d()함수를 사용하여 2 x 2 크기의 풀링을 적용한다.
이 단계에서 만들어진 특성맵의 크기는 14 x 14 x 10이다. 풀링으로 특성 맵의 크기를 줄인 다음 tf.reshape() 함수를 사용해 일렬로 펼친다. 이때 배치 차원을 제외한 나머지 차원만 펼쳐야 한다. 그다음 코드는 완전 연결층에 해당한다.
np.dot() 함수를 텐서플로의 tf.matmul() 함수로 바꾸었다.
이는 conv2d()와 max_pool2d()등이 Tensor 객체를 반환하기 때문이다.

  def forpass(slef, x):
    c_out1 = tf.nn.conv2d(x, self.conv_w, strides=1, padding='SAME') + self.conv_b
    r_out = tf.nn.relu(c_out)
    #2 x 2 최대 풀링 적용
    p_out = tf.nn.max_pool2d(r_out, ksize=2, strides=2, padding='VALID')
    #첫 밴째 배치 차원 제외하고 일렬로
    f_out = tf.reshape(p_out, [x.shape[0], -1])
    z1 = tf.matmul(f_out, self.w1) + self.b1
    a1 = tf.nn.relu(z1)
    z2 = tf.matmul(a1, self.w2) + self.b2

    return z2

 

역방향 게산

합성곱의 역방향 계산을 직접 구현하는 것은 복잡하기도 하지만 학습에 유용하지도 않다.
그레이디언트를 구하기 위해 역방향 계산을 직접 구현하는 대신 텐서플로의 자동 미분(automatic differentiation) 기능을 사용하여 역방향 계산을 구해보자.

with 블럭으로 tf.GradientTape() 객체가 감시할 코드를 감싸야한다. tape 객체는 with 블럭 안에서 일어나는 모든 연산을 기록하고 텐서플로 변수인 tf.Variable 객체를 자동으로 추적한다. 그레이디언트를 계산하려면 미분 대상 객체와 변수를 tape 객체의 gradient() 메서드에 전달해야 한다.

training() 메소드에서 backprop() 메서드를 호출하여 가중치를 업데이트했다. 하지만 자동 미분 기능을 사용하면 ConvolutionNetwork의 backprop() 메서드를 구현할 필요가 없다.

  def training(self, x, y):
    m = len(x)
    with tf.GradientTape() as tape:
      z = self.forpass(x) #정방향 계산
      loss = tf.nn.softmax_cross_entropy_with_logits(y, z)
      loss = tf.reduce_mean(loss)

 정방향 계산을 수행한 다음 tf.nn.softmax_cross_entropy_with_logits()함수를 호출하여 정방향 계산의 결과(z)와 타깃(y)을 기반으로 손실값을 계산한다. 이렇게 하면 크로스 엔트로피 손실과 그레이디언트 계산을 올바르게 처리해 주므로 편리하다.

 

그레이디언트 계산

가중치와 절편을 업데이트해야 한다.
tape.gradient() 메소드를 사용하면 그레이디언트를 자동으로 계산할 수 있다. 합성곱층의 가중치와 절편인 con_w와 con_b를 포함하여 그레이디언트가 필요한 가중치를 리스트로 나열한다. 그다음에 optimizer.apply_gradients()를 이용하여 가중치를 업데이트한다. 

  def training(self, x, y):
    m = len(x)
    with tf.GradientTape() as tape:
      z = self.forpass(x) #정방향 계산
      loss = tf.nn.softmax_cross_entropy_with_logits(y, z)
      loss = tf.reduce_mean(loss)
    weights_list = [self.conv_w, self.conv_b, self.w1, self.b1, self.w2, self.b2]
    #가중치 그레이디언트 계산
    grads = tape.gradient(loss, weights_list)
    #가중치 업데이트
    self.optimizer.apply_gradients(zip(grads, weights_lsit))
  

 

옵티마이저 객체를 가중치 초기화

trainig() 메소드에 등장하는 self.optimizer를 fit() 메서드에서 만들어 보자.
여기서 확률적 경사 하강법(SGD)를 사용한다.

fit() 수정

  def fit(self, x, y, epochs=100, x_val=None, y_val=None):
    self.init_weights(x.shape, y.shape[1])  #은닉층과 출력층의 가중치 초기화
    self.optimizer = tf.optimizers.SGD(learning_rate=self.lr)

    for i in range(epochs):
      print('에포크', i, end=' ')
      batch_losses = []
      for x_batch, y_batch in self.gen_batch(x, y):
        print('.', end=' ')
        self.training(x_batch, y_batch)
        batch_losses.append(self.get_loss(x_batch, y_batch))
        print( )
        #배치 손실 평균 내어 훈련 손실값으로 지정
        self.losses.append(np.mean(batch_losses))
        #검증 세트에 대한 손실 계산
        self.val_losses.append(self.get_loss(x_val, y_val))

 

init_weights() 수정

    def init_weights(self, input_shape, n_classes):
      g = tf.initializers.glorot_uniform()
      self.conv_w = tf.Variable(g((3, 3, 1, self.n_kernels)))
      self.conv_b = tf.Variable(np.zeros(self.n_kernels), dtype=float)
      n_features = 14 * 14 * self.n_kernels
      self.w1 = tf.Variable(g((n_features, self.units)))  #특성 개수, 은닉층의 크기
      self.b1 = tf.Variable(np.zeros(self.units), dtype=float)  #은닉층 크기
      self.w2 = tf.Variable(g((self.units, n_classes))) #(은닉층 크기, 클래스 개수)
      self.b2 = tf.Variable(np.zeros(n_classes), dtype=float) #클래스 개수

가중치를 glorot_uniform() 함수로 초기화하고, 텐서플로의 자동 미분 기능을 사용하기 위해 가중치를 tf.Variable() 함수로 초기화한다.
합성곱의 가중치와 완전 연결층의 가중치를 tf.Variable() 함수로 선언할 때 입력값에 따라 자료형이 자동으로 결정된다.
np.zeros()함수는 기본적으로 64비트 실수를 만든다. 절편 변수를 가중치 변수와 동일하게 32비트 실수로 맞추기 위해 
dtype을 float을 지정했다.

glorot_uniform() 함수는 가중치를 초기화할 때 글로럿(Glorot) 초기화라는 방법을 사용할 수 있게 해 준다.
넘파이로 난수를 만들어 가중치를 초기화했는데 신경망 모델이 너무 커지면 손실 함수도 복잡해지기 때문에 출발점에 따라 결과가 달라질 수 있다.

경사 하강법은 출발점으로부터 기울기가 0이 되는 최점점을 찾아간다. 가중치를 적절하게 초기화하지 않으면 출발점이 적절하지 않은 곳에 설정되므로 위 그래프의 왼쪽 모습과 같이 엉뚱한 곳에서 최적점이라는 판단을 내릴 수 있다.
이렇게 찾은 지점을 지역 최적점(local minimum)이라고 부른다.
올바른 최적점은 전역 최적점(global minimum)이라고 부른다.

세이비어 글로럿(Xavier Glorot)이 제안하여 널리 사용되는 가중치 초기화 방식이다.

두 사이에서 균등하게 난수를 발생시켜 가중치를 초기화한다.
n_kernels크기와 채널 차원까지 고려하여 4차원 배열로 초기화한다.
만들어진 특성 맵의 개수는 n_kernel이다. 이 배열이 일렬로 펼쳐져서 완전 연결층에 주입된다.
이에 필요한 가중치 w1의 크기는 14 x 14 x n_kernel이 된다. 

 

전체 코드 

class ConvolutionNetwork(MultiClassNetwork):

  def __init__(self, n_kernels=10, units=10, batch_size=32, learning_rate=0.1):
    self.n_kernels = n_kernels  #합성곱 커널 개수
    self.kernel_size = 3  #커널 크기
    self.optimizer = None #옵티마이저
    self.conv_w = None  #합성곱층의 가중치
    self.conv_b = None #합성곱층의 절편
    self.units = units  #은닉층 뉴런 개수
    self.batch_size = batch_size  #배치 크기
    self.w1 = None  #은닉층 가중치
    self.b1 = None  #은닉층 절편
    self.w2 = None  #출력층 가중치
    self.b2 = None  #출력층 절편
    self.a1 = None  #은닉층 활성화 출력
    self.losses = []  #훈련 손실
    self.val_losses = []  #검승 손실
    self.lr = learning_rate #학습률

  def forpass(self, x):
    c_out = tf.nn.conv2d(x, self.conv_w, strides=1, padding='SAME') + self.conv_b
    r_out = tf.nn.relu(c_out)
    #2 x 2 최대 풀링 적용
    p_out = tf.nn.max_pool2d(r_out, ksize=2, strides=2, padding='VALID')
    #첫 밴째 배치 차원 제외하고 일렬로
    f_out = tf.reshape(p_out, [x.shape[0], -1])
    z1 = tf.matmul(f_out, self.w1) + self.b1
    a1 = tf.nn.relu(z1)
    z2 = tf.matmul(a1, self.w2) + self.b2
    return z2

  def init_weights(self, input_shape, n_classes):
    g = tf.initializers.glorot_uniform()
    self.conv_w = tf.Variable(g((3, 3, 1, self.n_kernels)))
    self.conv_b = tf.Variable(np.zeros(self.n_kernels), dtype=float)
    n_features = 14 * 14 * self.n_kernels
    self.w1 = tf.Variable(g((n_features, self.units)))  #특성 개수, 은닉층의 크기
    self.b1 = tf.Variable(np.zeros(self.units), dtype=float)  #은닉층 크기
    self.w2 = tf.Variable(g((self.units, n_classes))) #(은닉층 크기, 클래스 개수)
    self.b2 = tf.Variable(np.zeros(n_classes), dtype=float) #클래스 개수

  def training(self, x, y):
    m = len(x)
    with tf.GradientTape() as tape:
      z = self.forpass(x) #정방향 계산
      loss = tf.nn.softmax_cross_entropy_with_logits(y, z)
      loss = tf.reduce_mean(loss)
    weights_list = [self.conv_w, self.conv_b, self.w1, self.b1, self.w2, self.b2]
    #가중치 그레이디언트 계산
    grads = tape.gradient(loss, weights_list)
    #가중치 업데이트
    self.optimizer.apply_gradients(zip(grads, weights_list))


  def fit(self, x, y, epochs=100, x_val=None, y_val=None):
    self.init_weights(x.shape, y.shape[1])  #은닉층과 출력층의 가중치 초기화
    self.optimizer = tf.optimizers.SGD(learning_rate=self.lr)

    for i in range(epochs):
      print('에포크', i, end=' ')
      batch_losses = []
      for x_batch, y_batch in self.gen_batch(x, y):
        print('.', end=' ')
        self.training(x_batch, y_batch)
        batch_losses.append(self.get_loss(x_batch, y_batch))
      print()
      #배치 손실 평균 내어 훈련 손실값으로 지정
      self.losses.append(np.mean(batch_losses))
      #검증 세트에 대한 손실 계산
      self.val_losses.append(self.get_loss(x_val, y_val))


  def gen_batch(self, x, y):
    bins = len(x) // self.batch_size  #미니배치 횟수
    indexes = np.random.permutation(np.arange(len(x)))
    x = x[indexes]
    y = y[indexes]
    for i in range(bins):
      start = self.batch_size * i
      end = self.batch_size * (i + 1)
      yield x[start:end], y[start:end]


  def get_loss(self, x, y):
    z = self.forpass(x)
    #손실 계산
    loss = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits(y, z))
    return loss.numpy()  

 

데이터 불러오기 & 훈련

(x_train_all, y_train_all), (x_test, y_test) = tf.keras.datasets.fashion_mnist.load_data()

print(x_train_all.shape, y_train_all.shape)

from sklearn.model_selection import train_test_split

x_train, x_val, y_train, y_val = train_test_split(x_train_all, y_train_all, stratify=y_train_all, test_size=0.2, random_state=42)
print(np.bincount(y_train))
print(np.bincount(y_val))

원-핫 인코딩

y_train_encoded = tf.keras.utils.to_categorical(y_train)
y_val_encoded = tf.keras.utils.to_categorical(y_val)

print(y_train[0], y_train_encoded[0])

데이터 전처리

x_train = x_train.reshape(-1, 28, 28, 1)
x_val = x_val.reshape(-1, 28, 28, 1)
x_train = x_train / 255
x_val = x_val /255

print(x_train.shape, x_val.shape)

 

훈련

cn = ConvolutionNetwork(n_kernels=10, units=100, batch_size=128, learning_rate=0.01)
cn.fit(x_train, y_train_encoded, x_val=x_val, y_val=y_val_encoded, epochs=20)

plt.plot(cn.losses)
plt.plot(cn.val_losses)
plt.ylabel('loss')
plt.xlabel('iteration')
plt.legend(['train_loss', 'val_loss'])
plt.show()

정확도가 88%에 가깝다.

cn.score(x_val, y_val_encoded)