대학원생이 쉽게 설명해보기

아래 블로그 코드에서 `22년 1월 11일 기준 에러없이 설치하여 사용했습니다.

블로그 참조이므로 코드는 아래 블로그를 직접 방문하셔서 확인하면 좋을 것 같네요 : )

https://sosomemo.tistory.com/72

텐서플로우 2.2 버전부터 Accuracy()/'acc'로 쓰는 정확도의 표현이 match → equal로 Binary Accuracy와 차이를 두었습니다.

(TF ~2.1v) Calculates how often predictions matches labels.
                                     ↓
(TF 2.2v~) Calculates how often predictions equals labels.

따라서,

• Accuracy: 정확히 일치
  (얼마나 같은가; Equal)
• Binary_Accuracy: 지정해둔 threshold에 따라 Accuracy를 계산
  (얼마나 Match 되는가)

공식 문서에 따르면 Binary Accuracy는 default threshold가 0.5로 지정되어 있습니다.

TF 2.2 이하 버전에서 짜여진 코드를 가지고 최신 버전으로 학습시킬 때,
정확도 점수가 다른 경우 ['acc', 'binary_accuracy']를 확인해보는 것이 좋을 것 같습니다.

acc와 binary_acc의 차이를 알아볼 수 있는 예시입니다.

import tensorflow as tf

y_true = [[1], [1], [0], [0]]
y_pred = [[0.51], [1], [0.49], [0]]

print(tf.keras.metrics.Accuracy()(y_true, y_pred))  # 0.5
print(tf.keras.metrics.BinaryAccuracy()(y_true, y_pred)) # 1.0

Learning Rate WarmUp은 많은 논문에서 사용하고 있는 유명한 기법입니다.

WarmUp 방법을 통해 학습률은 시간이 지남에 따라 아래 그림처럼 변화합니다.

구현은 아래 두 가지 코드(scheduler, callback 버전)을 참고하시고, decay_fn 등 하이퍼파라미터는 알맞게 변경해서 사용하면 됩니다.

Scheduler 버전

class LRSchedule(tf.keras.optimizers.schedules.LearningRateSchedule):
    def __init__(self, init_lr, warmup_epoch,
                 steps_per_epoch,
                 decay_fn, *,
                 continue_epoch = 0):
        self.init_lr = init_lr
        self.decay_fn = decay_fn
        self.warmup_epoch = warmup_epoch
        self.continue_epoch = continue_epoch
        self.steps_per_epoch = steps_per_epoch
        self.lr = 1e-4 # remove

    def on_epoch_begin(self, epoch):
        epoch = tf.cast(epoch, tf.float64)
        
        global_epoch = tf.cast(epoch + 1, tf.float64)
        warmup_epoch_float = tf.cast(self.warmup_epoch, tf.float64)
        
        lr = tf.cond(
            global_epoch < warmup_epoch_float,
            lambda: tf.cast(self.init_lr * (global_epoch / warmup_epoch_float), tf.float64),
            lambda: tf.cast(self.decay_fn(epoch - warmup_epoch_float), tf.float64)
        )
        self.lr = lr
    
    def __call__(self, step):
        def compute_epoch(step):
            return step // self.steps_per_epoch
        
        epoch = compute_epoch(step)
        epoch = epoch + self.continue_epoch
        
        self.on_epoch_begin(epoch)
        
        return self.lr

def get_steps(x_size, batch_size):
    if x_size / batch_size == 0:
        return x_size // batch_size
    else:
        return x_size // batch_size + 1

# data_size: train_set 크기
data_size = 100000
BATCH_SIZE = 512
EPOCHS = 100
warmup_epoch = int(EPOCHS * 0.1)
init_lr = 0.1
min_lr = 1e-6
power = 1.
    
lr_scheduler = tf.keras.optimizers.schedules.PolynomialDecay(
    initial_learning_rate = init_lr,
    decay_steps = EPOCHS - warmup_epoch,
    end_learning_rate = min_lr,
    power = power
)

# get_steps: epoch당 전체 step 수 계산
lr_schedule = LRSchedule(init_lr, warmup_epoch,
                         steps_per_epoch = get_steps(data_size, BATCH_SIZE),
                         decay_fn = lr_scheduler,
                         continue_epoch = 0)

# 사용 예시
optimizer = tf.keras.optimizers.Adam(learning_rate = lr_schedule)

Callback 버전

class LRSchedule(tf.keras.callbacks.Callback):
    def __init__(self, init_lr, warmup_epoch, decay_fn):
        self.init_lr = init_lr
        self.decay_fn = decay_fn
        self.warmup_epoch = warmup_epoch
        self.lrs = []

    def on_epoch_begin(self, epoch, logs = None):
        global_epoch = tf.cast(epoch + 1, tf.float64)
        warmup_epoch_float = tf.cast(self.warmup_epoch, tf.float64)

        lr = tf.cond(
                global_epoch < warmup_epoch_float,
                lambda: init_lr * (global_epoch / warmup_epoch_float),
                lambda: self.decay_fn(global_epoch - warmup_epoch_float),
                )

        tf.print('learning rate: ', lr)
        tf.keras.backend.set_value(self.model.optimizer.lr, lr)
        
        self.lrs.append(lr)
        
        
epochs = 1000
warmup_epoch = int(epochs * 0.1)
init_lr = 0.1
min_lr = 1e-6
power = 1.
    
lr_scheduler = tf.keras.optimizers.schedules.PolynomialDecay(
    initial_learning_rate = init_lr,
    decay_steps = epochs - warmup_epoch,
    end_learning_rate = min_lr,
    power = power
)

# lr_schedule = LRSchedule(init_lr = init_lr,
#                          warmup_epoch = warmup_epoch,
#                          decay_fn = lr_scheduler)

# for i in range(epochs):
#     lr_schedule.on_epoch_begin(i)

# 사용 예시
model.fit(..., callbacks = [LRSchedule(init_lr = init_lr,
                                      warmup_epoch = warmup_epoch,
                                      decay_fn = lr_scheduler)],
          initial_epoch = 0)

파이썬 문법 중 하나인 데코레이터(Decorator)는 wrapping 방식을 통해 함수를 유연하게 적용하여, 특정 기능을 편리하게 사용하게 해줍니다.

아래처럼 특수기호 @를 붙여서 사용하는 방식이 데코레이터입니다.

@tf.function
def train_step(args):
    pass

글에서는 데코레이터의 개념, 동작방식이 아닌 몇 가지 유용하게 사용되는 또는 사용될만한 데코레이터를 알아보도록 하겠습니다.

@retry

@retry 데코레이터는 특정 예외가 발생한 경우, 설정한 횟수만큼 재시도하는 데코레이터입니다.

일반적인 상황에서는 잘 사용되진 않지만, 네트워크, DB 통신에서 유용하게 사용되는 함수입니다.

retrying 라이브러리나 직접 구현하여 사용하는 방법이 있지만, tenacity 라이브러리가 유명하게 사용됩니다. 아래 코드는 tenacity 공식 문서에 나와있는 예제입니다.

  - tenacity 공식 문서: https://tenacity.readthedocs.io/en/latest/

import random
from tenacity import retry

@retry
def do_something_unreliable():
    if random.randint(0, 10) > 1:
        raise IOError("Broken sauce, everything is hosed!!!111one")
    else:
        return "Awesome sauce!"

print(do_something_unreliable())

성공할때까지의 시도, wait 횟수 등도 parameter를 통해 조절할 수 있게 제공해줍니다.

@classmethod

@classmethod는 아래 @staticmethod와 함께 보면 좋을 것 같습니다.

어느정도 자바/C 언어를 경험한 분들이라면 클래스 또는 정적 변수에 익숙할 것입니다. 그 개념 그대로 파이썬에서도 클래스 변수, 클래스 메소드를 사용할 수 있습니다.

단일 클래스나 상속받은 클래스의 변수를 조정할 때 유용하게 사용됩니다.

암묵적으로 @classmethod 데코레이터를 사용하면 첫 번째 인자로 cls 파라미터를 사용합니다. 인스턴스 메소드라 불리우는 self와 비슷한 개념입니다. 우리는 cls 파라미터를 통해 클래스 변수에 접근할 수 있게 됩니다.

위에서 언급하였듯이 @classmethod를 통해 클래스 변수를 조정할 수 있고, upgrade_os 메소드처럼 새로운 생성자를 만들어 줄 수도 있습니다.

class Computer:
    os = 'Linux' # 클래스 변수
    
class Personal_Computer(Computer):
    def __init__(self, c_id, pos):
        self.c_id = c_id
        self.partial_os = pos
        
    @classmethod
    def change_os(cls, this_os):
        if cls.os != this_os:
            cls.os = this_os
        else:
            print(f'{cls.os} Already up-to-date!')
            
    @classmethod
    def upgrade_os(cls, c_id, new_os):
        return cls(c_id, new_os)
            
computer_1 = Personal_Computer('2021', 'window11')
computer_2 = Personal_Computer('2020', 'window10')

# 변경 전, Linux -> Window
print(f'os change Before: {computer_1.os}, {computer_2.os}')
Personal_Computer.change_os(this_os = 'Window')

# 변경 후
print(f'os change After: {computer_1.os}, {computer_2.os}', end = '\n\n')

# upgrade os
print(f'os upgrade Before: {computer_2.partial_os}')
upgraded_computer_2 = computer_2.upgrade_os(computer_1.c_id, 'window11')
print(f'os change After: {upgraded_computer_2.partial_os}')

@staticmethod

정적 메소드입니다. 이 데코레이터를 사용하면 아래 코드처럼 객체 생성없이 바로 클래스를 통해 접근하여 사용이 가능합니다. 유용해보이지만, 많이 사용하지 않는 기능입니다.

class Computer:
    os = 'Linux' # 클래스 변수
    
class Personal_Computer(Computer):      
    ..생략..
    
    @staticmethod
    def print_os():
        print(Computer.os)

Personal_Computer.print_os() # Linux

@property

@property 데코레이터는 getter/setter를 떠올리면 쉽다. 아래 코드처럼 사용 가능합니다.

@property는 getter, @name.setter는 setter 역할을 합니다.

class ERP:
    def __init__(self):
        self._salary = 100
    
    @property
    def salary(self):
        return self._salary
    
    @salary.setter
    def salary(self, value):
        if self._salary < 500:
            self._salary = value
        else:
            raise ValueError('No!')
        
erp = ERP()
print(erp.salary) # 100
erp.salry = 200
print(erp.salry) # 200

물론 파이썬 특성상 이렇게 하지않아도 변수에 자유롭게 접근하거나 새로운 값을 할당할 수 있습니다.
다만 파이썬에서는 언더바('_')를 활용해서 private/protected 의미를 부여하는데요.

'_name' 은 private, '__name'은 protected를 암묵적으로 의미합니다.

언더바를 사용하면 우리가 흔히 사용하는 기능인 객체 메소드 참조(예를 들면 Tab키를 눌러서 어떤 함수가 확인하는)가 되지 않습니다. 참조하려면 언더바를 명시적으로 붙여주어야 합니다.

뿐만 아니라 @name.setter의 기능은 당연히 setter와 같지만, 위의 예제처럼 변수 할당에 조건을 편리하게 걸 수 있다는 장점이 있습니다.

@dataclass

Python 3.7이상 버전부터 사용할 수 있는 @dataclass 데코레이터는 __init__, __eq__, __repr__ 등 함수를 자동으로 등록해주는 기능입니다.

from dataclasses import dataclass

@dataclass
class Person:
    name: str
    age: int
        
person1 = Person('kim', 10)
person2 = Person('kim', 10)

print(person1.name, person1.age)
print(person1 == person2)

위처럼 별도의 함수를 정의해주지 않아도 바로 사용할 수 있습니다.

@lru_cache

@lru_cache는 메모라이제이션(memorization) 기능을 의미합니다.

이 데코레이터를 사용하면 함수의 반환값을 max_size 크기까지 값을 저장할 수 있습니다.

from functools import lru_cache
import urllib

@lru_cache(maxsize = 32)
def get_pep(num):
    'Retrieve text of a Python Enhancement Proposal'
    resource = 'https://www.python.org/dev/peps/pep-%04d/' % num
    try:
        with urllib.request.urlopen(resource) as s:
            return s.read()
    except urllib.error.HTTPError:
        return 'Not Found'
    
id_check = []
for i in [8, 290, 308, 320, 8, 218, 320, 279, 289, 320, 9991]:
    pep = get_pep(i)
    
    if i == 8:
        id_check.append(pep)
    
print(get_pep.cache_info()) # 총 hits 3 -> 320 2번, 8 1번 hit
assert id_check[0] is id_check[1] # id 비교, @lru_cache 사용안할 시 에러 발생!

@lru_cache 기능으로 인해 id_check에 들어가는 객체 id가 동일해서(캐시를 이용하기 때문) AssertionError가 발생하지 않습니다

결론부터 말하자면, Gradient Accumulation 방법은 GPU memory issue를 보완하기 위한 방법입니다.

배치 크기는 성능에 영향을 주는 중요한 하이퍼파라미터 중 하나인데요.
이 때문에 배치 크기(Batch Size)와 성능(Performance) 관계는 이전부터 많은 방법(다양한 조건에서의 실험, 수학적 풀이, 최적화 관점 등)으로 연구되어 왔습니다.

대표적인 논문만 간단히 보자면,

• FAIR(Facebook AI Research) 논문
      (1) Accurate, Large Minibatch SGD: Training ImageNet in 1 Hour
      (2) https://arxiv.org/pdf/2105.07576.pdf
• 배치 크기를 위주로 실험한 논문
      (1) The effect of batch size on the generalizability of the convolutional neural ...
• Small Batch 관련 논문
      (1) On large-batch training for deep learning: Generalization gap and sharp minima
      (2) Revisiting small batch training for deep neural networks

위 논문뿐만 아니라 관련 논문을 보면 아래와 같이 결론을 얻을 수 있습니다.

    ① LB(Large Batch)는 학습 속도도 빠르고, 성능도 좋을 수 있다
    ② LB 일수록 더 큰 학습률(Learning Rate)를 사용하면 좋다 (하지만 큰 학습률을 사용하면 수렴이 안되는 문제가?)
    ③ 하지만 LB보다 SB(Small Batch)가 안정적으로 학습할 수 있어 일반화(Generalization)에 강하다

Gradient Accumulation 방법이 배치 크기와 관련이 있다보니 간단하게 배치 크기 관련 이야기를 해보았는데, 이번 글은 어떤 이유에서든 "비록 가지고 있는 GPU memory가 제한적이지만 일단 LB를 사용해서 학습시키고 싶다."라는 생각입니다.

Gradient Accumulation?

큰 배치 크기를 활용할때 가장 큰 문제는 '성능이 좋아질 수 있다', '안좋아질 수 있다'가 아닐겁니다. 아마도 가장 중요하게 고려하는 것은 지금 보유하고 있는 GPU를 가지고 '1,024, 2,048, ... 처럼 큰 배치 크기를 돌려볼 수 있느냐'입니다.

누구나 다 겪는 문제인데요. 만약 이와 같은 문제에 직면했다면, 이를 해결하기 위한 방법 중 하나로 Gradient Accumulation 방법을 생각해볼 수 있습니다. 가장 좋은 방법은 돈을 들이는 것...

일반적인 학습 방법과 Gradient Accumulation 방법의 차이는 아래 그림에서 바로 확인할 수 있습니다.

일반적인 학습 방법이 미니 배치를 통해 gradient를 구한 후, Update를 즉시 진행하는 방법이라면,

Gradient Accumulation 방법을 포함한 학습은 미니 배치를 통해 구해진 gradient를 n-step동안 Global Gradients에 누적시킨 후, 한번에 업데이트하는 방법입니다.

매우 간단합니다. 즉, 핵심 아이디어는 256 mini-batch를 통해 gradient를 업데이트하는 것과 64 mini-batch를 4번(64 * 4 = 256) 누적하여 업데이트하는 것이 비슷한 결과를 가져다 줄 것이라는 생각입니다.

이 예와 같다면, 우리가 보유하고 있는 GPU memory 한계상 64 mini-batch 밖에 사용할 수 없다면, 이 학습 방법을 통해 4번 누적하여 업데이트할 경우 256 mini-batch를 사용하여 학습하는 것과 다름없게 되는 셈이죠.

하지만 이 방법은 memory issue를 중점적으로 해결하고자 하는 방법이지 속도, 성능과는 아직 explicit하게 증명된 바가 없습니다.

아래의 간단한 구현을 통해 위에서 글로서 살펴본 것보다 좀 더 쉽게 이해할 수 있을 거에요.

전체 코드는 깃허브_저장소에 있습니다.

Setup

from tensorflow.keras.datasets import mnist
from tensorflow.keras.layers import Input, Dense, Conv2D, MaxPooling2D, Flatten
from tensorflow.keras.models import Model
import tensorflow as tf

print(tf.__version__)

MNIST Dataset Laod

(x_train, y_train), (x_test, y_test) = mnist.load_data()

Make TF Dataset

def make_datasets(x, y):
    # (28, 28) -> (28, 28, 1)
    def _new_axis(x, y):
        y = tf.one_hot(y, depth = 10)
        
        return x[..., tf.newaxis], y
            
    ds = tf.data.Dataset.from_tensor_slices((x, y))
    ds = ds.map(_new_axis, num_parallel_calls = tf.data.experimental.AUTOTUNE)
    ds = ds.shuffle(100).batch(32) # 배치 크기 조절하세요
    ds = ds.prefetch(tf.data.experimental.AUTOTUNE)
    
    return ds
    
ds = make_datasets(x_train, y_train)

Make Models

# rescaling, 1 / 255
preprocessing_layer = tf.keras.models.Sequential([
        tf.keras.layers.experimental.preprocessing.Rescaling(1./255),
    ])

# simple CNN model
def get_model():
    inputs = Input(shape = (28, 28, 1))
    preprocessing_inputs = preprocessing_layer(inputs)
    
    x = Conv2D(filters = 32, kernel_size = (3, 3), activation='relu')(preprocessing_inputs)
    x = MaxPooling2D((2, 2))(x)
    x = Conv2D(filters = 64, kernel_size = (3, 3), activation='relu')(x)
    x = MaxPooling2D((2, 2))(x)
    x = Conv2D(filters = 64, kernel_size =(3, 3), activation='relu')(x)
    
    x = Flatten()(x)
    x = Dense(64, activation = 'relu')(x)
    outputs = Dense(10, activation = 'softmax')(x)
    
    model = Model(inputs = inputs, outputs = outputs)
    
    return model

model = get_model()
model.summary()

Training with Gradient Accumulation

epochs = 10
num_accum = 4 # 누적 횟수

loss_fn = tf.keras.losses.CategoricalCrossentropy(from_logits = True)
optimizer = tf.keras.optimizers.Adam()
train_acc_metric = tf.keras.metrics.CategoricalAccuracy()

@tf.function
def train_step(x, y):
    with tf.GradientTape() as tape:
        logits = model(x)
        loss_value = loss_fn(y, logits)
    gradients = tape.gradient(loss_value, model.trainable_weights)
    
    # update metrics    
    train_acc_metric.update_state(y, logits)
    
    return gradients, loss_value

def train():
    for epoch in range(epochs):
        print(f'################ Start of epoch: {epoch} ################')
        # 누적 gradient를 담기 위한 zeros_like 선언
        accumulation_gradients = [tf.zeros_like(ele) for ele in model.trainable_weights]
        
        for step, (batch_x_train, batch_y_train) in enumerate(ds):
            gradients, loss_value = train_step(batch_x_train, batch_y_train)
            
            if step % num_accum == 0:
                accumulation_gradients = [grad / num_accum for grad in accumulation_gradients]
                optimizer.apply_gradients(zip(gradients, model.trainable_weights))

                # zero-like init
                accumulation_gradients = [tf.zeros_like(ele) for ele in model.trainable_weights]
            else:
                accumulation_gradients = [(accum_grad + grad) for accum_grad, grad in zip(accumulation_gradients, gradients)]

            if step % 100 == 0:
                print(f"Loss at Step: {step} : {loss_value:.4f}")
            
        train_acc = train_acc_metric.result()
        print(f'Accuracy : {(train_acc * 100):.4f}%')
        train_acc_metric.reset_states()
        
# start training
train()