[실습] numpy로 만드는 단층 신경망 - 회귀 문제

Tensorflow 없이 numpy를 이용해서 단층 신경망을 만들어 보자.

 

우라는 Abalone 데이터를 가지고 전복의 나이를 예측하는 신경망을 구축해 볼 예정이다.

데이터 셋은 Kaggle에 있는 Abalone Dataset 에서 받아서 사용 할 수 있다.

 

 

✔︎  데이터 살펴보기

import pandas as pd

df = pd.read_csv('data/abalone.csv')
df.head()

 

(4177, 9) shape로 구성된 Abalone data

 

 

◼ 변수 설명

 

독립변수는 Sex ~ Shell weight 까지 8개로 구성되어 있고, 종속변수는 Rings 이다.

데이터는 총 4177개로 준비되어 있다.

 

 

📌   단층 신경망을 만들기 위한 설계도

 

우리는 위의 설계도 순서대로 함수를 하나하나 만들어 볼 것이다.

 

1. 모듈 불러오기

import numpy as np
import csv

# 실험 결과를 재현하기 위해 난수 발생패턴을 고정시키는 np.random.seed()함수값을 설정
np.random.seed(333)

 

 

2. 하이퍼 파라미터 설정

• learning rate 학습률
• 정규분포 난숫값 [ 평균 / 표준편차 ]

RND_MEAN = 0  # 평균
RND_STD = 0.0030  # 표준편차

LEARNING_RATE = 0.001  # 학습률

 

처음에는 이값을 무작위로 지정할 수 밖에 없다.

그렇다면 무작정 무작위값을 주기 보다, 랜덤하지만 적어도 최소한의 규칙성이 있는 범위 내에서 난수를 지정하는 것 이 더 좋을 것 이다.

이러한 방법은 Xavier 초기화 와 He 초기화라 하며 현재 가장 많이 쓰이는 방법이다.

 

 

 

3. 메인함수 정의

메인 함수 main_exec() 함수를 실행하게 되면

#1 데이터를 불러들이는 load_dataset() 함수,

#2 가중치와 편향을 초기화 해주는 init_model() 함수,

#3 학습 및 신경망 성능 테스트를 위한 train_and_test()함수

 

이렇게 세가지 함수가 차례로 실행되며, 신경망 모델을 생성하고 학습 전체 과정을 일괄 처리한다.

 

def main_exec(epoch=10, mb_size=10, report=1, train_rate=0.8):
    load_dataset()  # 데이터를 불러오는 함수
    init_model()   # 가중치와 편향을 초기화 해주는 함수
    train_and_test(epoch, mb_size, report, train_rate)  # 학습 및 신경망 성능 테스트하는 함수

 

메인 함수에서는 학습 횟수, 미니배치 크기, 중간 보고 주기, 학습 데이터 비율 등 학습과정에 관련된 하이퍼파라미터값들을 epoch_count, mb_size, report, train_rate 의 변수로 지정받아 이 값들을 실제로 이용할 train_and_test()에 전달하게 된다.

 

 

 

4. 데이터를 불러오는 load_dataset() 정의

def load_dataset():
    with open('data/abalone.csv') as csvfile:
        csvreader = csv.reader(csvfile)
        next(csvreader, None)  # None : 첫 행을 건너뛴다 (변수명을 쓰지 않겠다.)
        rows = []
        for row in csvreader:  # 한 행씩 가져온다.
            rows.append(row)

    global data, input_cnt, output_cnt
    input_cnt, output_cnt = 10, 1  # 독립변수의 크기, 종속변수의 크기

    # 버퍼 (가상 저장 공간을 만들어줌)
    data = np.zeros([len(rows), input_cnt + output_cnt
                     ])  # np.zeros() 함수는 지정해준 크기만큼 0값의 행렬을 생성

    # 원 핫 벡터 처리
    for n, row in enumerate(rows):  # n = index , row = value
        if row[0] == 'I': data[n, 0] = 1  # Sex의 값에 따라 원핫인코딩
        if row[0] == 'M': data[n, 1] = 1
        if row[0] == 'F': data[n, 2] = 1
        data[n, 3:] = row[1:]  # 나머지 데이터를 넣어줌

 

 

# load_dataset() 함수의 주요기능

 

csv파일 내용을 메모리로 읽어들여 이용할 수 있게 준비해주며, 데이터에 포함되어 있는 비선형 정보를 원-핫 벡터로 표현하여 출력한다.

csv 모듈의 reader()기능을 활용하여 Kaggle에서 가져온 'abalone.csv' 내용을 메모리로 읽어들니다.

 

Abalone 데이터 셋의 첫 번째 행은 데이터에 대한 설명으로 프로그램을 구축하는데 있어 필요 없다는 것을 알 수 있다.그래서 next() 함수를 통해 파일의 첫 행을 읽지 않고 건너뛰게 만든다. (즉 데이터의 첫 번째 행을 무시하기 위해 사용)

 

그리고 for 반복문을 활용해 csvreader에 담긴 전복 개체별 정보를 rows빈 리스트에 append()함수로 넣어준다.

 

그 다음 input_cnt와 output_cnt 변수에 입출력 벡터 크기를 각각 10과 1 로 설정한다.

이 값은 기존 8이었던 입력 벡터 크기가 '원-핫 벡터 표현'을 통해 10으로 증가한 것이며, 이 값을 통해 이후 입출력 벡터 정보를 저장할 data 행렬 생성 및 data의 크기 지정에 활용된다. 또한 이후 전역변수로 선언하여 다른 함수에서 활용될 예정이다.

 

반복문을 활용하여 rows에 담긴 '범주형 성별 정보'를 '원-핫 벡터로 변환하는 처리', 그리고 나머지 데이터 항목들을 일괄적으로 복제하는 처리를 수행하게 된다.

 

 

 

5. 파라미터 초기화 함수 init_model() 정의

def init_model():
    global weight, bias, input_cnt, output_cnt
    weight = np.random.normal(RND_MEAN, RND_STD,[input_cnt, output_cnt])  
    # normal 메서드 :  (mean, sd, shape)
    
    bias = np.zeros([output_cnt])
    # weight 가중치 행렬은 [10,1] , bias 편향 벡터는 [1]형태

 

weight는 처음에 설정한 하이퍼파라미터값과 np.random.normal()를 이용해 정규분포를 갖는 난숫값으로 초기화를 진행한다.  그 이유는 𝑙𝑜𝑐𝑎𝑙 𝑚𝑖𝑛𝑖𝑚𝑢𝑚을 피할 가능성을 조금이라도 높이기 위해서 이다.

 

편향(bias)은 초기에 너무 큰 영향을 주어 학습에 역효과를 불러오지 않도록 0으로 초기화하여 생성하였다.

 

 

 

6. 학습 및 평가 함수 train_and_test() 정의

위의 설계도를 보면 train_and_test() 함수를 정의하기 위해 5개의 함수가 필요하다.

 

• arrange_data()
• get_train_data()
• get_test_data()
• run_train()
• run_test()

학습과 평가를 위해 데이터 셋을 학습 데이터와 테스트 데이터로 분리해줘야 한다. 여기서는 7:3으로 분리할 예정이다.

그 다음 학습 데이터에 대한 미니배치 처리를 수행하고, 학습 데이터와 테스트 데이터의 독립변수와 종속변수를 나눠줄 것이다.

 

def train_and_test(epoch_count, mb_size, report, train_rate):

    # 미니배치가 볓 스텝으로 쪼개지는지 값을 반환해줌
    step_count = arrange_data(mb_size, train_rate)

    # 테스트 데이터의 독립, 종속변수 분할
    test_x, test_y = get_test_data()

    for epoch in range(epoch_count):
        losses, accs = [], []  # 전체 미니배치의 결과인 손실과 정확도를 받는 빈 리스트 정의
        for n in range(step_count):
            train_x, train_y = get_train_data(
                mb_size, n)  # 미니배치 사이즈의 학습데이터의 독립, 종속변수 반환
            loss, acc = run_train(train_x, train_y)
            losses.append(loss)
            accs.append(acc)

        if report > 0 and (epoch + 1) % report == 0:
            acc = run_test(test_x, test_y)
            print(
                f"Epoch{epoch + 1}: Train - loss = {np.mean(losses):5.3f}, accuracy = {np.mean(accs):5.3f} / Test={acc:5.3f}"
            )

    final_acc = run_test(test_x, test_y)
    print(f'\n 최종 테스트 : final accuracy = {final_acc:5.3f}')

 

train_and_test 함수는 epoch_count, mb_size, report, train_rate를 인자로 받는다,

 

 

6.1 데이터셋을 섞어주고 step_count를 계산해주는 arrange_data()

• mb_size, train_rate 인수값을 받아 전체 데이터셋을 무작위로 섞어준다.
• 학습 데이터에 필요한 1 에폭당 미니배치 스텝 수를 계산하여 step_count에 저장한다.
  (step_count = 전체데이터 / 미니배치 사이즈)

def arrange_data(mb_size, train_rate):

    # data는 load_dataset()의 전역변수
    # shuffle_map, test_begin_index 은 다른 동간에서도 사용할 수 있게 전역변수화
    global data, shuffle_map, test_begin_index

    # 전체 데이터셋을 무작위로 섞어준다.
    shuffle_map = np.arange(data.shape[0])
    np.random.shuffle(shuffle_map)

    # 미니배치 단위
    step_count = int(
        data.shape[0] * train_rate) // mb_size  # train_rate는 학습데이터의 비율을 말함

    test_begin_index = step_count * mb_size  # 경계선 / Train_set의 데이터 수

    return step_count

 

 

 

6.2 데이터를 분할 해주는 get_test_data(),  get_train_data()

# get_test_data()

• 전체 데이터셋에서 테스트에 사용할 데이터 만큼 분리한다.
• 분리된 데이터에서 독립변수와 종속변수를 분할하여 test_x, test_y에 저장한다.

def get_test_data():
    global data, shuffle_map, test_begin_index, output_cnt # 전역변수

    test_data = data[shuffle_map[test_begin_index:]] # 테스트 데이터 분리

    return test_data[:, :-output_cnt], test_data[:, -output_cnt:] # 독립변수, 종속변수 나눠주기

 

 

# get_train_data()

• 전체 데이터셋에서 인수값으로 전달 받은 mb_size만큼 미니배치 처리를 수행한다.
• 처리된 미니배치 데이터에서 독립변수와 종속변수로 나눠준다.
• 새로운 epoch이 수행될 때마다의 무작위 표본 추출을 시행한다.
  ➞  에폭값이 갱신될 때 학습 데이터가 다시 섞이게 되며, 학습의 효율성이 높아짐을 기대할 수 있게 됩니다.

def get_train_data(mb_size, nth):
    global data, shuffle_map, test_begin_index, output_cnt

    if nth == 0:   # shuffle_map을 epoch이 시작할 때 마다 shuffle 해준다.
        np.random.shuffle(shuffle_map[:test_begin_index])

    train_data = data[shuffle_map[mb_size * nth : mb_size * (nth + 1)]]  
    # 미니배치 수 만큼 뽑아서 순차적으로 학습시킴

    return train_data[:, :-output_cnt], train_data[:, -output_cnt:]  # 독립변수, 종속변수로 나눠준다.

 

 

 

6.3 학습 시키는 함수 run_train()

• 학습 데이터 train_x, train_y를 인자로 전달 받는다.
• 또 다른 다섯가지의 함수가 등장한다.
  [ forward_neuralnet(), forward_postproc(), eval_accuracy(), backprop_postproc(), backprop_neuralnet() ]
• 순전파와 역전파가 수행되며 학습이 진행, 그에 따른 손실값(loss)과 정확도(acc)를 return 해준다.
• 전달받은 loss와 acc는 append()를 통해 빈 리스트 losses, accs에 보관된다.

def run_train(x, y):
    output, aux_nn = forward_neuralnet(x)  # axu_nn, aux_pp = 역전파에 필요한 부가정보 
    loss, aux_pp = forward_postproc(output, y)  # output과 실제값 y로 loss를 구해줌
    accuracy = eval_accuracy(output, y)  # 정확도

    G_loss = 1.0
    G_output = backprop_postproc(G_loss, aux_pp)  # 역전파로 기울기를 구하는 함수
    backprop_neuralnet(G_output, aux_nn)  # w, b 갱신하는 과정

    return loss, accuracy

 

 

6.4 테스트를 수행하는 run_test()

• test_x, test_y를 전달받는다. 
• 테스트 데이터에 대한 acc를 반환한다.

def run_test(x, y):
    output, _ = forward_neuralnet(x)  # _: 값이 필요하지 않을 때 사용
    accuracy = eval_accuracy(output, y)  
    return accuracy # 정확도를 반환

 

학습이 완료된 후에, 최종 모델 정확도를 다시한번 출력하기 위해 한번 더 run_test()를 수행하고,

그 결과를 출력하여 학습 및 평가가 모두 완료되었다고 알려주는 역할을 한다.

 

 

 

7. 순전파 및 역전파 함수 정의

7.1 forward_postproc() : 신경망의 순전파에 따른 mse 구하는 과정

• output과 실제 y값을 인자로 받아 mse를 구해준다.

def forward_postproc(output, y):
    diff = output - y  # 예측값 - 실제값 (편차)
    square = np.square(diff)  # 제곱해주는 연산
    loss = np.mean(square)  # 편차 제곱의 평균 (mse)
    return loss, diff  # mse와 편차를 return

 

 

7.2 backprop_postproc() : mse의 역전파 과정 - 미분을 통한 값의 갱신을 수행

$ \frac{\partial L}{\partial x} = \frac{\partial L}{\partial y} \times \frac{\partial y}{\partial x} $ 로 역전파가 수행된다.

 

$\frac{\partial y}{\partial x}$ 의 경우 입력된 파라미터 값에 따른 𝑦 값의 미분을 구하는 과정으로써 이후 가중치와 편향에 따른 손실함수의 기울기를 구하는 과정에서 다뤄볼 것이다.

 

$  \frac{\partial L}{\partial y} $ 는 출력값 y에 대한 역전파 방법으로 어떤 손실함수로 구할것이냐에 따라 연산과정이 달라진다.

 

우리가 이번에 다뤄볼 회귀에 맞는 손실함수인 mse의 기울기를 구해보면 다음과 같다.

 

 

mse를 구하는 '예측', '편차', '제곱', '평균(손실)' 과정을 각 단계별로 미분하여 진행하면 쉽게 구할 수가 있게 된다.

 

 

$\frac{\partial L}{\partial mse} $ 는 1이므로 다음과정인 $ \frac{\partial mse}{\partial square}$ 를 봐보자.

 

 

결국 $ \frac{\partial mse}{\partial square}$ 를 수행하면 $ \frac{1}{ MN}$ 이 된다.

 

 

 그 다음으로 제곱에 대한 $ \frac{\partial square}{\partial diff}$ 를 살펴보면, 

 

 

2 * diff 라는 간단한 결과가 나온다.

 

 

마지막으로 $ \frac{\partial diff}{\partial output}$ 을 보면,

 

 

1로 나오는것을 확인 할 수 있다.

 

def backprop_postproc(G_loss, diff):  
    shape = diff.shape  
    G_loss = 1  # delta(L) / delta(mse) 미분

    g_loss_square = np.ones(shape) / np.prod(shape)  # delta(mse) / delta(square) 미분
    g_square_diff = 2 * diff  # delta(square) / delta(diff) 미분
    g_diff_output = 1  # delta(diff) / delta(output) 미분

    G_square = g_loss_square * G_loss  # 순차적으로 값을 곱해준다.
    G_diff = g_square_diff * G_square
    G_output = g_diff_output * G_diff
    
    return G_output

 

평균제곱오차의 각 단계별 입출력 간 부분적인 기울기를 구해놓고, 손실에 대한 기울기의 연쇄적 계산에 활용하여 최종적으로 mse의 역전파 처리인  G_output을 계산하였다.

 

 

 

7.3 신경망의 순전파 연산과정 forward_neuralnet()

train_x는 x로 전달받아 가중치와 행렬곱을 수행하고, bias와 더해진다.

그리고 이러한 결과는 output으로 저장하고 return 해준다. 

또한 순전파 과정을 수행하기위해 전달받은 x 는 이후 역전파과정 수행을 위해 다시 output과 함께  return 해준다.

 

def forward_neuralnet(x):  
    global weight, bias
    output = np.matmul(x, weight) + bias  # 행렬 계산
    return output, x  # 입력값 x는 역전파를 수행하기 위해 return

 

 

7.4 역전파를 수행하는 backprop_neuralnet()

신경망 연산과정을  𝒀 = 𝑿𝑾 + 𝑩 로 하고, 가중치의 손실 기울기에 대하여 구하는 과정을 살펴보자.

 

 

가중치의 손실 기울기는 $X^{T}G$ 로 정리할 수 있다.

이때 $G_{ij}$는 $Y_{ij}$ 성분에 대한 손실 함수의 기울기로, 평균제곱오차의 역전파 처리 과정 에서 살펴보았으며, G_output이라는 변수로 저장되어있다.

 

편향에 관한 손실 기울기는 $G_{ij}$ 의 각 행의  합을 구하는 방법으로 구해진다.

 

def backprop_neuralnet(G_output, x):
    global weight, bias  
    g_output_w = x.transpose()

    G_w = np.matmul(g_output_w, G_output)
    G_b = np.sum(G_output, axis=0)

    weight -= LEARNING_RATE * G_w  # 가중치 갱신
    bias -= LEARNING_RATE * G_b  # 편향 갱신

 

 

8. 신경망 학습 결과 테스트 및 평가함수 정의 eval_accuracy()

정확도 평가의 경우 회귀 분석 문제에서 정확도를 정의하는 다양한 방식이 있지만 그 중, 간단한 방식인 예측값과 실제값과의 차를 실제값으로 나눠주는 방식을 채택하여 오류율을 산출하고 그 값을 1에서 빼줌으로 정확도를 정의한다.

 

def eval_accuracy(output, y):  # y = test_y
    mdiff = np.mean(np.abs((output - y) / y))  # 오류율
    return 1 - mdiff  # 정확도

 

 

실행하기

LEARNING_RATE = 0.001
main_exec(epoch=100, mb_size=20, report=20, train_rate=0.80)

new_x = [0, 1, 0, 0.68, 0.54, 0.19, 1.43, 0.67, 0.382, 0.5]
output = forward_neuralnet(new_x)

print(output[0] + 1.5)

 

 

새로운 x값을 넣어서 학습을 수행해보면 정확도 81%로 전복의 나이를 13.7세로 예측한다.

 

weight와 bias를 살펴보면,

 

 

암컷 여부와 키가 전복의 고리 수를 추정하는데 큰 영향을 미친 것 을 확인할 수 있다.

하지만 난수 초깃값을 이용했기 때문에 다시 학습을 진행하면 다른 결과가 나올 수 있다.