[미니프로젝트] 1 feature, 1 sample linear regression model 구현

오늘은 그간 배웠던 개념들로 선형 회귀 모델(linear regression model)을 구현해 볼까 한다.

1개의 feature, 1개의 sample인 모델이라 아기자기하지만, 동작 과정은 가능한한 깊게 뜯어보겠다.

 

오늘은 Class 구현을 먼저 해봤다. 구현된 모델의 동작과정을 기준으로 설명해보겠다.

 

목차

- Linear Regression Model Class

1. DatasetGenerator Class

2. Model Class

- 업데이트시 왜 오차가 줄어드나요?
- learning_rate는 왜 필요한가요? (추후 포스팅 예정)

3. SqauredError Class

4. main

 

- Linear Regression Model Class

 

Linear_Regression_Model.py

import numpy as np
import matplotlib.pyplot as plt

# Make Dataset Class 
class DatasetGenerator:
    def __init__(self, w, b, noise_factor = 0, HAS_NOISE=False):
        self.w = w  # w_answer: 우리가 맞추고자 하는 임의의 가중치 
        self.b = b  # b_answer: 우리가 맞추고자 하는 임의의 상수
        self.noise_factor = noise_factor    # training을 위한 임의의 오류 y = wx + b + "noise"
        self.HAS_NOISE = HAS_NOISE  # 오류 필요 여부

        # dataset
        self.X = None
        self.Y = None

        # DatasetGenerator() = 1.__init__ 2.make_dataset()
        self._make_dataset()

    def _make_dataset(self):
        self.X = np.random.randn(100, 1)
        if self.HAS_NOISE == False:
            self.Y = self.w * self.X + self.b
        else:
            # HAS_NOISE == TRUE일시, Y에 noise 추가
            self.Y = self.w * self.X + self.b + self.noise_factor * np.random.randn(100, 1)

    # return dataset
    def get_dataset(self):
        
        return self.X, self.Y

    # return (x의 최대값, 최소값)   : 시각화를 위한 데이터
    def get_Mm(self):
        x_Max = np.max(self.X)
        x_min = np.min(self.X)

        return x_Max, x_min

# Operate Class : 하나의 뉴런, Dense, 함수의 역할을 함
class Model:
    def __init__(self, learning_rate):
        # w := w - dw 연산시, 목적과 다르게 움직이는 오류를 줄이기 위한 "아주 작은 상수"
        self.learning_rate = learning_rate

        self.w = None   # 임의의 가중치: learning될 대상
        self.b = None   # 임의의 상수: learning될 대상
        self.pred = None    # hat(x): self.w * x + b : 임의의 가중치, 상수로 연산된 예상 값
        self.x = None   # 1 sample
        
        # init w, b
        self._init_params()

    def _init_params(self):
        self.w = np.random.randn(1) # init w
        self.b = np.random.randn(1) # init b

    # 정방향으로 임의의 가중치, 상수와 연산
    # return pred: 예상값
    def forward(self, x):
        self.x = x
        self.pred = self.w * x + self.b
        return self.pred
    
    """
    loss = (pred - y)** 2
    
    d_loss/d_pred = {(pred - y)(pred - y)}' 
    = (pred -y)'(pred - y) + (pred -y)(pred - y)'
    = 1(pred - y) + (pred - y)*1
    = 2(pred - y)
    """
    def backward(self, dloss_dpred):
        # chain rule + parameter update
        """
        d_pred/d_w = (wx + b)' : w를 기준으로
        = x

        d_pred/d_b = (wx + b)' : b를 기준으로
        = 1

        d_loss/d_w = (d_loss/d_pred) * (d_pred/d_w)
                   = dloss_dpred * dpred_dw
        d_loss/d_b = (d_loss/d_pred) * (d_pred/d_b)
                   = dloss_dpred * dpred_db
        """
        dpred_dw , dpred_db = self.x, 1
        dloss_dw = dloss_dpred * dpred_dw
        dloss_db = dloss_dpred * dpred_db

        """
        w := w - a*(dw)
        b := w - a*(db)
        """
        self.w = self.w - self.learning_rate * dloss_dw
        self.b = self.b - self.learning_rate * dloss_db

    def get_w_b(self):
        return self.w, self.b

# loss of pred and y Class
class SquaredError:
    def __init__(self):
        self.pred = None
        self.y = None

    def forward(self, pred, y):
        self.pred = pred
        self.y = y

        "loss = (pred - y)** 2"
        return (pred - y) ** 2

    def backward(self):
        """
        loss = (pred - y)** 2
        
        d_loss/d_pred = {(pred - y)(pred - y)}' 
        = (pred -y)'(pred - y) + (pred -y)(pred - y)'
        = 1(pred - y) + (pred - y)*1
        = 2(pred - y)
        """

        dloss_dpred = 2*(self.pred - self.y)
        return dloss_dpred

# w:= w - learning_rate * dw / b:= b - learning_rate * db
# epochs: 훈련 횟수, 반복 횟수
learning_rate, epochs= 0.1, 100

# 우리가 맞추고자 하는 가중치 w, 상수 b 
w_answer, b_answer = 2, -1
# w_answer, b_answer = np.random.randint(100), np.random.randint(100) 

# 노이즈의 정도
noise_factor = 1

# make Dataset
dg = DatasetGenerator(w=w_answer, b=b_answer, noise_factor=noise_factor, HAS_NOISE=True)

# x, y dataset
X, Y = dg.get_dataset()

# visualize dataset
fig, ax = plt.subplots(figsize=(10, 20))
ax.scatter(X, Y)

# initialize traing object
model = Model(learning_rate=learning_rate)
sqaured_error = SquaredError()

# visualize before start traning 
X_Max, X_min = dg.get_Mm()

x_predictor = np.linspace(X_min, X_Max, 2)
w_init, b_init = model.get_w_b()
y_predictor = w_init*x_predictor + b_init
ax.plot(x_predictor, y_predictor, 'blue')

# training
for epoch in range(epochs):
    for x,y in zip(X, Y):
        pred = model.forward(x)
        loss = sqaured_error.forward(pred, y)
        dloss_dpred = sqaured_error.backward()
        model.backward(dloss_dpred)

# traning result
w_result, b_result = model.get_w_b()

# answer과 traning result 비교
print(w_answer, b_answer)   # 2 -1
print(w_result, b_result)   # [1.94532528] [-1.62554133]

# visualize result
y_predictor = w_result*x_predictor + b_result
ax.plot(x_predictor, y_predictor, 'red')

plt.show()

위 코드와 결과를 얻을 수 있었다.

 

관련한 코드를 하나하나 살펴보겠다.

 

1. DatasetGenerator Class

먼저, dataset을 생성하는 Class이다. 

여기서 생성하는 dataset은 우리가 맞추고자 하는 w, b를 기준으로 한 X, Y의 dataset이다.

 

f(x) = y = wx + b일때, 임의의 x값들과 x에 대응되는 f(x)인 y이다.

# Make Dataset Class 
class DatasetGenerator:
    def __init__(self, w, b, noise_factor = 0, HAS_NOISE=False):
        self.w = w  # w_answer: 우리가 맞추고자 하는 임의의 가중치 
        self.b = b  # b_answer: 우리가 맞추고자 하는 임의의 상수
        self.noise_factor = noise_factor    # training을 위한 임의의 오류 y = wx + b + "noise"
        self.HAS_NOISE = HAS_NOISE  # 오류 필요 여부

        # dataset
        self.X = None
        self.Y = None

        # DatasetGenerator() = 1.__init__ 2.make_dataset()
        self._make_dataset()

    def _make_dataset(self):
        self.X = np.random.randn(100, 1)
        if self.HAS_NOISE == False:
            self.Y = self.w * self.X + self.b
        else:
            # HAS_NOISE == TRUE일시, Y에 noise 추가
            self.Y = self.w * self.X + self.b + self.noise_factor * np.random.randn(100, 1)

    # return dataset
    def get_dataset(self):
        
        return self.X, self.Y

    # return (x의 최대값, 최소값)   : 시각화를 위한 데이터
    def get_Mm(self):
        x_Max = np.max(self.X)
        x_min = np.min(self.X)

        return x_Max, x_min

그리 특별한 것은 없으나, 새로 추가된 거라면 noise가 있다.

높은 훈련 결과를 위해 일부러 오차를 집어 넣었다.

이미지로 치면 일부러 화질을 낮추고, 의미 없는 낙서를 집어 넣는 것이고,

오디오로 치면 잡음을 끼어 넣는 것이다.

noise의 정도(noise_factor)와 noise의 유무(HAS_NOISE)를 객체 생성때 설정할 수 있다.

 

get_Mm()은 x의 최대값, 최소값을 return하는 function으로,

x의 최대값, 최소값은 "훈련전"과 "훈련후"의 변화를 시각화하기 위해 사용된다.

 

2. Model Class

# Operate Class : 하나의 뉴런, Dense, 함수의 역할을 함
class Model:
    def __init__(self, learning_rate):
        # w := w - dw 연산시, 목적과 다르게 움직이는 오류를 줄이기 위한 "아주 작은 상수"
        self.learning_rate = learning_rate

        self.w = None   # 임의의 가중치: learning될 대상
        self.b = None   # 임의의 상수: learning될 대상
        self.pred = None    # hat(x): self.w * x + b : 임의의 가중치, 상수로 연산된 예상 값
        self.x = None   # 1 sample
        
        # init w, b
        self._init_params()

    def _init_params(self):
        self.w = np.random.randn(1) # init w
        self.b = np.random.randn(1) # init b

    # 정방향으로 임의의 가중치, 상수와 연산
    # return pred: 예상값
    def forward(self, x):
        self.x = x
        self.pred = self.w * x + self.b
        return self.pred
    
    """
    loss = (pred - y)** 2
    
    d_loss/d_pred = {(pred - y)(pred - y)}' 
    = (pred -y)'(pred - y) + (pred -y)(pred - y)'
    = 1(pred - y) + (pred - y)*1
    = 2(pred - y)
    """
    def backward(self, dloss_dpred):
        # chain rule + parameter update
        """
        d_pred/d_w = (wx + b)' : w를 기준으로
        = x

        d_pred/d_b = (wx + b)' : b를 기준으로
        = 1

        d_loss/d_w = (d_loss/d_pred) * (d_pred/d_w)
                   = dloss_dpred * dpred_dw
        d_loss/d_b = (d_loss/d_pred) * (d_pred/d_b)
                   = dloss_dpred * dpred_db
        """
        dpred_dw , dpred_db = self.x, 1
        dloss_dw = dloss_dpred * dpred_dw
        dloss_db = dloss_dpred * dpred_db

        """
        w := w - a*(dw)
        b := w - a*(db)
        """
        self.w = self.w - self.learning_rate * dloss_dw
        self.b = self.b - self.learning_rate * dloss_db

    def get_w_b(self):
        return self.w, self.b

 

- __init__(), _init_params()

: class의 변수 초기화, learning_rate를 설정하고, 임의의 w, b (가중치, 상수)를 초기화합니다.

 

-  forward(x)

: input이 "1 feature의 x"이고, "pred" = "w*x + b"으로 연산 후, "pred"를 return합니다.

 

- backward(dloss_dpred)

$$ \frac{dloss(x)}{dw}, \frac{dloss(x)}{db} $$ 

: backward를 통해 임의의 w, b(가중치, 상수)를 업데이트를 한다.

: dw, db를 구하기 위한 미분 과정은 아래 그림과 같다

 

위의 그림과 같은 과정을 통해서

 

# dloss_dpred = 2(pred - y)
dpred_dw , dpred_db = self.x, 1
dloss_dw = dloss_dpred * dpred_dw
dloss_db = dloss_dpred * dpred_db

 

코드와 같이 dw, db (w에 대한 미분값, b에 대한 미분값)이 구해진다.

구해진 미분값으로 임의의 w, b (가중치, 상수)를 업데이트한다.

        """
        w := w - a*(dw)
        b := w - a*(db)
        """
        self.w = self.w - self.learning_rate * dloss_dw
        self.b = self.b - self.learning_rate * dloss_db

 

코드와 같이 w,b가 업데이트되면, 다음 연산때는 오차가 조금 더 줄어든 형태로 연산이 진행된다.

 

여기서 궁금증 두가지!

- 업데이트시 왜 오차가 줄어드나요?

왜 오차가 줄어드는지를 이해하기 위해선 딥러닝 학습의 전체적인 맥락을 파악할 필요가 있다.

 

위 그림처럼, w, b의 업데이트가 진행되면, loss function의 극소값을 찾아가게 된다.

 

- learning_rate는 왜 필요한가요?

: 추후 포스팅 예정

https://doyou-study.tistory.com/91

 

3. SqauredError Class

# loss of pred and y Class
class SquaredError:
    def __init__(self):
        self.pred = None
        self.y = None

    def forward(self, pred, y):
        self.pred = pred
        self.y = y

        "loss = (pred - y)** 2"
        return (pred - y) ** 2

    def backward(self):
        """
        loss = (pred - y)** 2
        
        d_loss/d_pred = {(pred - y)(pred - y)}' 
        = (pred -y)'(pred - y) + (pred -y)(pred - y)'
        = 1(pred - y) + (pred - y)*1
        = 2(pred - y)
        """

        dloss_dpred = 2*(self.pred - self.y)
        return dloss_dpred

pred(예상값)과 y의 loss, error를 구하는 클래스입니다.

 

- __init__()

: pred, y 변수 생성

 

- forward(pred, y)

: loss, error인 (pred - y) ** 2 연산을 하고, return 한다. 

 

- backward()

$$ \frac{dloss}{dpred} = \frac{d}{dpred}\left [ (pred-y)^{2} \right ] = \left [ (pred-y)^{2} \right ]' = (pred - y)'(pred - y) + (pred - y)(pred - y)' = 2(pred - y) $$

:  위 수식처럼 "dloss/dpred", pred에 대한 loss()의 미분을 연산하고, 이를 리턴한다.

: 관련한 내용은 위의 수식과 코드 주석을 확인할 것

 

4. main

# w:= w - learning_rate * dw / b:= b - learning_rate * db
# epochs: 훈련 횟수, 반복 횟수
learning_rate, epochs= 0.1, 100

# 우리가 맞추고자 하는 가중치 w, 상수 b 
w_answer, b_answer = 2, -1
# w_answer, b_answer = np.random.randint(100), np.random.randint(100) 

# 노이즈의 정도
noise_factor = 1

# make Dataset
dg = DatasetGenerator(w=w_answer, b=b_answer, noise_factor=noise_factor, HAS_NOISE=True)

# x, y dataset
X, Y = dg.get_dataset()

# visualize dataset
fig, ax = plt.subplots(figsize=(10, 20))
ax.scatter(X, Y)

# initialize traing object
model = Model(learning_rate=learning_rate)
sqaured_error = SquaredError()

# visualize before start traning 
X_Max, X_min = dg.get_Mm()

x_predictor = np.linspace(X_min, X_Max, 2)
w_init, b_init = model.get_w_b()
y_predictor = w_init*x_predictor + b_init
ax.plot(x_predictor, y_predictor, 'blue')

# training
for epoch in range(epochs):
    for x,y in zip(X, Y):
        pred = model.forward(x)
        loss = sqaured_error.forward(pred, y)
        dloss_dpred = sqaured_error.backward()
        model.backward(dloss_dpred)

# traning result
w_result, b_result = model.get_w_b()

# answer과 traning result 비교
print(w_answer, b_answer)   # 2 -1
print(w_result, b_result)   # [1.94532528] [-1.62554133]

# visualize result
y_predictor = w_result*x_predictor + b_result
ax.plot(x_predictor, y_predictor, 'red')

plt.show()

 

이처럼 위에서 구현한 class를 이용해 dataset을 만들어 실제 training을 진행해봤다. 

 

추후에 더 크기를 키우고, 방정식 이외에 이미지, 음성에 대해서도 훈련하는 프로젝트를 진행하도록 하겠다.