[ML] 순수 파이썬 코드로 된 딥러닝 모델 분석하기-1. Regression

딥러닝 모델을 퓨어 파이썬 코드로 만들어 보자!

  현재 진행중인 산학협력과제에서도 스트레스 예측 모델을 케라스 코드 단 몇줄로 커버가 가능했을 정도로 머신러닝에서 프레임워크와 라이브러리들의 힘은 매우 강력하다. 하지만 교수님 말씀도 그렇고, 회사 대표님 말씀도 들어보니 머신러닝을 공부하는 데 있어 프레임워크에 계속 의지하다 보면 원리를 깊이 이해하는 능력이 떨어져 결국에는 막히게 된하고 한다.  따라서 기본에 충실해 보고자 이번에는 학교에서 수업 보조교재로도 사용하고 있는 "파이썬 날코딩으로 알고 짜는 딥러닝" 이라는 책을 활용해서 책에서 소개하는 개념과 소스코드를 면밀히 분석하고 뜯어보기로 했다. 물론 기존에 짜여진 코드를 분석하는 정도에 불과하지만, 남이 잘 짜놓은 코드를 보며 공부하는 것도 중요하다고 생각한다!

< Regression >

github.com/KONANtechnology/Academy.ALZZA/blob/master/codes/chap01/abalone.ipynb

KONANtechnology/Academy.ALZZA

   바로 코드를 보면서 개념을 추가해 보도록 하겠다. 먼저 모델은 abalone_exec이라는 함수가 전부다. 이 함수는 epoch, batch size, 출력단위를 파라미터로 입력받아, 해당 조건에 맞는 학습을 시켜준다. 이때 epoch 횟수는 전체 sample-data를 모두 사용해서 학습을 시킨 횟수이며 mb_size는 신경망이 여러 데이터를 한꺼번에 처리하는 것을 의미한다. 이처럼 에폭,미니배치 크기 등은 학습과정에 반영되지는 않으나 학습결과에 영향을 미치는 하이퍼파라미터(hyper-parameter)들이다.

def abalone_exec(epoch_count=10, mb_size=10, report=1):
    load_abalone_dataset() # 데이터 인풋
    init_model() # 모델 생성
    train_and_test(epoch_count, mb_size, report) # 모델 학습

그럼 이 함수가 가진 세가지 파트에 대해 분석해 보도록 하자.

< 1. Data Input : load_abalone_dataset() >

load_abalone_dataset() 함수는 모델을 학습시키기 위한 데이터 셋을 불러오는 함수이다. 

def load_abalone_dataset():
    with open('../../data/chap01/abalone.csv') as csvfile:
        csvreader = csv.reader(csvfile)
        next(csvreader, 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])

    for n, row in enumerate(rows):
        if row[0] == 'I': data[n, 0] = 1 # I,M,F는 성별을 뜻하며 이를 0,1로 나타내기 위한 원핫인코딩이다
        if row[0] == 'M': data[n, 1] = 1
        if row[0] == 'F': data[n, 2] = 1
        data[n, 3:] = row[1:]

제일 먼저 abalone 데이터셋을 불러와서 한줄씩 row에 붙인다. 이때 next()는 데이터셋에서 처음으로 등장하는 필드명을 제외하기 위해서 호출한다. 나머지 값들은 사용할 데이터 이므로 rows에 한줄씩 추가한다. 그 다음은 np.zeros함수로 데이터 로우 개수 x 입출력 벡터크기 만큼의 영행렬을 만든다. 이때 I,M,F에 해당하는 성별은 비선형 정보이기 때문에 이런 비선형 정보를 항목별로 분할하여 하나는 1, 나머지는 0으로 나타내는 방식을 one-hot vector라고 한다. 

< 2. Model init : init_model() >

init_model()은 모델의 weight과 bias을 입력벡터와 출력벡터를 통해 생성하게 된다. 

def init_model():
    global weight, bias, input_cnt, output_cnt
    weight = np.random.normal(RND_MEAN, RND_STD,[input_cnt, output_cnt])
    bias = np.zeros([output_cnt])

이때 RND_MEAN과 RND_STD는 정규분포의 평균과 표준편차 값이다. 이 값은 가중치 파라미터를 랜덤으로 초기화 할 때 사용한다. 가중치가 같은 값을 가지게 되면 중복성 문제가 발생할 수 있기 때문이다. 이후에 numpy.random.normal함수는 평균, 표준편차 값에 이어 input x output크기의 랜덤벡터를 생성해 준다.  따라서 weight은 10 x 1 크기의 가중치 파라미터로 설정되고, bias는 1 x 1크기의 영행렬이 된다.

np,random.normal 함수의 예시

< 3. Model train/test : train_and_test() >

train_and_test() 함수는 모델을 학습시키고 학습한 결과를 확인하는 함수이다. 이 함수 안에는 arrange_data(), get_test_data(), get_train_data(), run_train(), run_test() 함수가 들어가 있다. 하나씩 뜯어보도록 하자. 

def train_and_test(epoch_count, mb_size, report):
    step_count = arrange_data(mb_size)  # 1.
    test_x, test_y = get_test_data()  # 2.
    
    for epoch in range(epoch_count):
        losses, accs = [], []
        
        for n in range(step_count):
            train_x, train_y = get_train_data(mb_size, n)  # 3.
            loss, acc = run_train(train_x, train_y)  # 4.
            losses.append(loss)
            accs.append(acc)
            
        if report > 0 and (epoch+1) % report == 0:
            acc = run_test(test_x, test_y)  # 5.
            print('Epoch {}: loss={:5.3f}, accuracy={:5.3f}/{:5.3f}'. \
                  format(epoch+1, np.mean(losses), np.mean(accs), acc))
            
    final_acc = run_test(test_x, test_y)  # 6.
    print('\nFinal Test: final accuracy = {:5.3f}'.format(final_acc))

1. 먼저 step_count을 계산한다. step_count는 arrange_data()에 batch_size를 전달해서 구한다. 관련 코드는 아래와 같다. 

def arrange_data(mb_size):
    global data, shuffle_map, test_begin_idx
    shuffle_map = np.arange(data.shape[0])
    np.random.shuffle(shuffle_map)
    step_count = int(data.shape[0] * 0.8) // mb_size
    test_begin_idx = step_count * mb_size
    return step_count

arrange_data에서는 미니배치 처리를 위한 스텝수를 계산할 뿐 아니라, 데이터의 개수만큼 무작위로 섞은 데이터셋을 만든다. 그리고 test-set과 training set의 경계점에 해당하는 인덱스 위치를 저장해 둔다.

2. 그 다음 테스트 데이터를 가지고 와서 test_x와 test_y에 각각 집어넣는다. 데이터 셋에서 training data와 test data를 가져오는 과정은 아래 코드와 같다. 

def get_test_data():
    global data, shuffle_map, test_begin_idx, output_cnt
    test_data = data[shuffle_map[test_begin_idx:]]
    return test_data[:, :-output_cnt], test_data[:, -output_cnt:]

def get_train_data(mb_size, nth):
    global data, shuffle_map, test_begin_idx, output_cnt
    if nth == 0:
        np.random.shuffle(shuffle_map[:test_begin_idx])
    train_data = data[shuffle_map[mb_size*nth:mb_size*(nth+1)]]
    return train_data[:, :-output_cnt], train_data[:, -output_cnt:]

3. 이어서 매 step_count마다 run_train()함수를 통해 loss값과 accuracy를 구하게 된다. 그리고 나머지 코드는 그 결과를 출력하는 부분이므로 생략하기로 하고, 다음 run_train()과 run_test코드를 살펴보기로 한다.

def run_train(x, y):
    output, aux_nn = forward_neuralnet(x)
    loss, aux_pp = forward_postproc(output, y)
    accuracy = eval_accuracy(output, y)
    
    G_loss = 1.0
    G_output = backprop_postproc(G_loss, aux_pp)
    backprop_neuralnet(G_output, aux_nn)
    
    return loss, accuracy

def run_test(x, y):
    output, _ = forward_neuralnet(x)
    accuracy = eval_accuracy(output, y)
    return accuracy

위에 보이듯이 run_train()함수에는 forward_neuralnet()함수가 나온다. 이 함수는 인풋벡터인 x (train_x데이터)를 넣어 기존의 가중치 행렬인 weight과 bias를 통해 output출력을 내게 된다. 그 다음에는 forward_postproc()함수 부분이다. 여기서는 mse를 코드로 구현해서 정답과의 loss-value와 aux_pp값을 구한다. 이 aux_pp값은 출력값과 y값의 차이를 뜻하는데, 이를 바로 다음과정인 backprop_postproc()함수에서 이용한다.

def forward_neuralnet(x):
    global weight, bias
    output = np.matmul(x, weight) + bias
    return output, x

def forward_postproc(output, y):
    diff = output - y
    square = np.square(diff)
    loss = np.mean(square)
    return loss, diff

backprop_postproc()함수는 역전파 함수이며 맨 마지막 g_loss를 의미하는 1값과 이전에 mse에서 사용한 정답과의 차이를 의미하는 aux_pp값을 사용한다. 이후로 나오는 코드는 mse의 역전파 처리과정을 나타낸 것이다. 예를 들어 square = diff^2이였을 것이므로, 이를 diff에 대해 편미분하면 2 * diff가 나오는 식이다. 

다음은 backprop_neuralnet()부분이다. 여기서는 손실 기울기를 계산해서 이를 적당한 학습률을 통해 weight과 bias를 조정해 나가게 된다. 

def backprop_postproc(G_loss, diff):
    shape = diff.shape
    
    g_loss_square = np.ones(shape) / np.prod(shape)
    g_square_diff = 2 * diff
    g_diff_output = 1

    G_square = g_loss_square * G_loss
    G_diff = g_square_diff * G_square
    G_output = g_diff_output * G_diff
    
    return G_output
    
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

마지막은 run_test()부분에서 , forward_neuralnet()을 통해 얻어낸 output값과, 정답과의 차이의 절댓값을 계산해 평균낸 뒤, 결과를 확인하기 위해 만은 evaluation function이다. 

def eval_accuracy(output, y):
    mdiff = np.mean(np.abs((output - y)/y))
    return 1 - mdiff

 

> 결과

코드를 실행시킨 결과는 다음과 같다.

> 후기

  전체적으로 코드가 굉장히 깔끔했고, 이론으로 배운 내용들이 파이썬 모듈 하나하나씩 구현되어 있어 코드를 배우면서 이론도 복습하는 재미가 쏠쏠했다. 지나치기 쉬운 구현코드들을 뜯어보면서 개념을 어떻게 코드로 구현하는 지에 대해 많이 알게 된 것 같다. 시간은 오래걸리지만 이런 시도를 많이 해야겠다는 생각이 든다.