SK 네트웍스 AI 캠프 - 2 _데이터 분석과 머신러닝 / 딥러닝 - Day31_AI 모델 검증을 통한 고도화

AI 모델 성능 고도화: 더 정확하게 예측하고 빠르게 동작하며 새로운 데이터에도 잘 대응하도록 개선하느 ㄴ과정

AI모델은 처음 학습했다고 항상 좋은 성능이 나오지 않는다 .

과소적합 학습이 부족한 상태

과적합 훈련데이터만 외운상태

데이터 부족

데이터 품질 문제

모델 구조문제(너무 단순사거나 너무 복잡하거나

 

고도화 방법

1. 데이터 수집 확대

2. 데이터 전처리 개선(결측치 제거, 중복제거, 정규화)

3. 데이터 증강(회전, 좌우반전, 확대, 축소, 밝기변경)

4. 특성 선택(중요한 정보만 선택한다. 불필요한 특징을 제거하면 성능이 좋아질 수 있다.)

5. 더 좋은 모델의 사용 

 - 선형회귀라면 random forest, xgboost, lightgbn으로 개선

 - 이미지 분야라면 cnn, resnet, efficientnet, vision transformer

6. 하이퍼파라미터 튜닝 ( learning rate, epoch, batch size, hidden layer, dropout)

7. 오버피팅 방지 (dropout, early stopping, weight decay)

8. 앙상블 여러 모델을 결합한다.

 - bagging randomforest이후 모델1, 2, 3을 거쳐 투표해 최종결과.

 - boosting adaboost, xgboost, lightgbm, catboost 이전 모델의 실수를 다음 모델이 보완한다. 

9. 전이학습 (이미 학습된 모델을 활용한다 resnet, vgg16, vision transformer...)

10. 손실함수 변경 (분류 crossentropyloss, 회귀 mseloss, 객체탐지 focal loss)

11, 최적화 알고리즘 변경 (sgd, momentum, rmsprop, adam, adamw)

12. 차원축소 활용

 - Principal Component Analysis: PCA 분산이 큰 방향 주성분을 찾아 데이터를 투영하는 방법.

   비지도 학습으로 label이 필요없다.

   선형변환으로 비선형 구조 반영이 불가하고 클래스 분리를 고려하지않는다. 

   기본 baseline으로 많이 사용한다 .

from sklearn.datasets import load_iris
from sklearn.decomposition import PCA
import matplotlib.pyplot as plt

data = load_iris()
X = data.data
y = data.target

pca = PCA(n_components=2)
X_pca = pca.fit_transform(X)

plt.scatter(X_pca[:, 0], X_pca[:, 1], c=y)
plt.title("PCA Result")
plt.show()

 -  Linear Discriminant Analysis : LDA 클래스간 분리를 최소화하고 클래스 내부 분산을 최소화한다.

    지도학습으로 label이 필요하며 클래스 정보를 적극활용해 선형변환한다. 

   분류 성능 개선에 직접적인 도움을 주고 클래스 경계가 뚜렷한 경우 매우 효과적이다. 

   클래스가 정규분포라는 가정에 클래스 수보다 차원이 체한된다. C-1

from sklearn.datasets import load_iris
from sklearn.discriminant_analysis import LinearDiscriminantAnalysis as LDA
import matplotlib.pyplot as plt

data = load_iris()
X = data.data
y = data.target

lda = LDA(n_components=2)
X_lda = lda.fit_transform(X, y)

plt.scatter(X_lda[:, 0], X_lda[:, 1], c=y)
plt.title("LDA Result")
plt.show()

 -  t-Distributed Stochastic Neighbor Embedding :  t-SNE  고차원에서 가까운 점들이 저차원에서도 가까이되도록 확률분포를 맞춤

    비지도학습, 비선형 차원 축소, 지역구조를 보존한다 .

     클러스터 시각화에 매우 강력하며 복잡한 비선형 구조표현이 가능하다. 

     계산 비용이 크고 새로운 데이터 투영이 어렵다.

from sklearn.datasets import load_iris
from sklearn.manifold import TSNE
import matplotlib.pyplot as plt

data = load_iris()
X = data.data
y = data.target

tsne = TSNE(n_components=2, perplexity=30, random_state=42)
X_tsne = tsne.fit_transform(X)

plt.scatter(X_tsne[:, 0], X_tsne[:, 1], c=y)
plt.title("t-SNE Result")
plt.show()

 -  Uniform Manifold Approximation and Projection : UMAP 다양체 구조를 그래프 기반으로 모델링해 저차원으로 보존한다 .

    비선형 차원축소 비지도 학습 t-sne의 개선형으로 많이 사용된다 .

     k-nn 그래프 구성, t-sne보다 빠르고 대규모 데이터에 적합하다 .

     하이퍼파라미터 영향이 크고 해석이 제한적이다 .

from sklearn.datasets import load_iris
import umap
import matplotlib.pyplot as plt

data = load_iris()
X = data.data
y = data.target

umap_model = umap.UMAP(n_components=2, random_state=42)
X_umap = umap_model.fit_transform(X)

plt.scatter(X_umap[:, 0], X_umap[:, 1], c=y)
plt.title("UMAP Result")
plt.show()

 

 

 

 

 

샘플코드로 분석해보자. ai_model_performance_optimization_torch.ipynb

import, seed 고정

# ============================================================
# 1. 필요한 라이브러리 불러오기
# ============================================================

# numpy는 배열 계산을 쉽게 하기 위해 사용합니다.
import numpy as np

# pandas는 표 형태의 데이터를 다루기 위해 사용합니다.
import pandas as pd

# matplotlib은 학습 손실과 정확도 그래프를 그리기 위해 사용합니다.
import matplotlib.pyplot as plt

# sklearn의 breast cancer 데이터셋을 사용하기 위해 불러옵니다.
from sklearn.datasets import load_breast_cancer

# train_test_split은 데이터를 학습용, 검증용, 테스트용으로 나누기 위해 사용합니다.
from sklearn.model_selection import train_test_split

# StandardScaler는 입력 데이터의 평균을 0, 표준편차를 1로 맞추는 표준화 전처리에 사용합니다.
from sklearn.preprocessing import StandardScaler

# accuracy_score는 분류 모델의 정확도를 계산하기 위해 사용합니다.
from sklearn.metrics import accuracy_score, confusion_matrix, classification_report

# torch는 PyTorch의 핵심 라이브러리입니다.
import torch

# torch.nn은 신경망 계층, 손실함수 등을 만들 때 사용합니다.
import torch.nn as nn

# torch.optim은 최적화 알고리즘을 사용할 때 필요합니다.
import torch.optim as optim

# TensorDataset은 입력 데이터와 정답 데이터를 하나의 데이터셋으로 묶기 위해 사용합니다.
from torch.utils.data import TensorDataset

# DataLoader는 데이터를 미니배치 단위로 나누어 학습할 수 있게 해줍니다.
from torch.utils.data import DataLoader

# 실행할 장치를 선택합니다.
# GPU가 있으면 cuda를 사용하고, 없으면 CPU를 사용합니다.
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

# 결과 재현을 위해 난수 시드를 고정합니다.
# 같은 코드를 다시 실행해도 가능한 비슷한 결과가 나오도록 합니다.
SEED = 42
np.random.seed(SEED)
torch.manual_seed(SEED)

# GPU 사용 시에도 재현성을 조금 더 높이기 위해 CUDA 시드를 고정합니다.
if torch.cuda.is_available():
    torch.cuda.manual_seed_all(SEED)

print("사용 장치:", device)

 

 

 

 

 

 

데이터셋 불러오기

# ============================================================
# 2. 데이터셋 불러오기
# ============================================================

# load_breast_cancer()는 유방암 진단용 예제 데이터셋을 불러옵니다.
# 입력 특성은 세포핵의 반지름, 질감, 면적 등 수치형 변수입니다.
data = load_breast_cancer()

# X는 모델에 입력되는 특성 데이터입니다.
X = data.data

# y는 정답 라벨입니다.
# 이 데이터셋에서는 0과 1로 구성된 이진 분류 라벨입니다.
y = data.target

# 특성 이름을 확인하기 위해 pandas DataFrame으로 변환합니다.
df = pd.DataFrame(X, columns=data.feature_names)

# 정답 라벨 컬럼을 추가합니다.
df["target"] = y

# 데이터 크기와 일부 데이터를 출력합니다.
print("입력 데이터 크기:", X.shape)
print("정답 데이터 크기:", y.shape)
display(df.head())

 

 

 

 

 

학습, 검증, 테스트용 데이터 분리

# ============================================================
# 3. 학습용 / 검증용 / 테스트용 데이터 분리
# ============================================================

# 먼저 전체 데이터를 학습+검증 데이터와 테스트 데이터로 나눕니다.
# test_size=0.2는 전체 데이터의 20%를 테스트 데이터로 사용한다는 뜻입니다.
# stratify=y는 정답 라벨의 비율을 학습/테스트 데이터에 비슷하게 유지합니다.
X_train_val, X_test, y_train_val, y_test = train_test_split(
    X,
    y,
    test_size=0.2,
    random_state=SEED,
    stratify=y
)

# 학습+검증 데이터를 다시 학습 데이터와 검증 데이터로 나눕니다.
# 여기서는 전체 기준으로 약 60% 학습, 20% 검증, 20% 테스트 구조가 됩니다.
X_train, X_val, y_train, y_val = train_test_split(
    X_train_val,
    y_train_val,
    test_size=0.25,
    random_state=SEED,
    stratify=y_train_val
)

print("학습 데이터:", X_train.shape, y_train.shape)
print("검증 데이터:", X_val.shape, y_val.shape)
print("테스트 데이터:", X_test.shape, y_test.shape)

 

 

 

데이터 전처리 및 tensor변환

# ============================================================
# 4. 데이터 전처리 - StandardScaler 표준화
# ============================================================

# StandardScaler는 각 특성의 평균을 0, 표준편차를 1로 맞춥니다.
# 신경망 모델은 입력값의 범위가 너무 다르면 학습이 불안정해질 수 있으므로 표준화가 중요합니다.
scaler = StandardScaler()

# fit_transform은 학습 데이터의 평균과 표준편차를 계산한 뒤 변환까지 수행합니다.
# 반드시 학습 데이터에만 fit을 해야 테스트 데이터 정보가 학습 과정에 새어 들어가지 않습니다.
X_train_scaled = scaler.fit_transform(X_train)

# 검증 데이터와 테스트 데이터는 학습 데이터에서 계산된 평균과 표준편차로만 변환합니다.
X_val_scaled = scaler.transform(X_val)
X_test_scaled = scaler.transform(X_test)

# numpy 배열을 PyTorch Tensor로 변환합니다.
# 입력 데이터는 실수 계산이므로 float32 타입을 사용합니다.
X_train_tensor = torch.tensor(X_train_scaled, dtype=torch.float32)
X_val_tensor = torch.tensor(X_val_scaled, dtype=torch.float32)
X_test_tensor = torch.tensor(X_test_scaled, dtype=torch.float32)

# 정답 데이터는 0 또는 1의 클래스 번호이므로 long 타입을 사용합니다.
# CrossEntropyLoss는 정답 라벨이 정수형 클래스 번호여야 합니다.
y_train_tensor = torch.tensor(y_train, dtype=torch.long)
y_val_tensor = torch.tensor(y_val, dtype=torch.long)
y_test_tensor = torch.tensor(y_test, dtype=torch.long)

print("Tensor 변환 완료")
print(X_train_tensor.shape, y_train_tensor.shape)

 

 

 

 

 

 

tensordataset으로 입력과 정답 tensor를 하나로 묶고

batch를 적용해 학습, 검증/테스트 dataloader 생성 이때 검증/테스트는 평가목적임으로 굳이 shuffle하지않는다. 

# ============================================================
# 5. Dataset과 DataLoader 생성
# ============================================================

# TensorDataset은 입력 Tensor와 정답 Tensor를 하나로 묶습니다.
train_dataset = TensorDataset(X_train_tensor, y_train_tensor)
val_dataset = TensorDataset(X_val_tensor, y_val_tensor)
test_dataset = TensorDataset(X_test_tensor, y_test_tensor)

# batch_size는 한 번에 모델에 넣을 데이터 개수입니다.
# 너무 작으면 학습이 불안정하고, 너무 크면 메모리를 많이 사용합니다.
BATCH_SIZE = 32

# 학습 데이터는 shuffle=True로 섞어서 모델이 데이터 순서에 외우지 않도록 합니다.
train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True)

# 검증/테스트 데이터는 평가 목적이므로 섞을 필요가 없습니다.
val_loader = DataLoader(val_dataset, batch_size=BATCH_SIZE, shuffle=False)
test_loader = DataLoader(test_dataset, batch_size=BATCH_SIZE, shuffle=False)

print("DataLoader 생성 완료")

 

 

 

 

 

 

기본모델 정의

nn.sequential 여러계층을 순서대로 연결. 

linear, relu, linear

forward로 통화하도록.

# ============================================================
# 6. 기본 모델 정의
# ============================================================

class BasicMLP(nn.Module):
    # __init__은 모델 구조를 정의하는 부분입니다.
    def __init__(self, input_dim, output_dim):
        # 부모 클래스 nn.Module의 초기화 기능을 실행합니다.
        super().__init__()

        # Sequential은 여러 계층을 순서대로 연결할 때 사용합니다.
        self.net = nn.Sequential(
            # Linear는 완전연결층입니다.
            # input_dim개의 입력 특성을 16개의 은닉 노드로 변환합니다.
            nn.Linear(input_dim, 16),

            # ReLU는 음수는 0으로 만들고 양수는 그대로 통과시키는 활성화 함수입니다.
            nn.ReLU(),

            # 마지막 Linear 계층은 16개의 은닉값을 output_dim개의 클래스 점수로 변환합니다.
            nn.Linear(16, output_dim)
        )

    # forward는 입력 데이터가 모델 안에서 어떻게 계산되는지 정의합니다.
    def forward(self, x):
        # Sequential에 정의된 계층을 차례대로 통과시킵니다.
        return self.net(x)


# 입력 특성 수는 X_train_tensor의 두 번째 차원입니다.
input_dim = X_train_tensor.shape[1]

# 이진 분류이므로 출력 클래스 수는 2입니다.
output_dim = 2

# 기본 모델 객체를 생성하고 device로 이동합니다.
basic_model = BasicMLP(input_dim, output_dim).to(device)

print(basic_model)

 

 

 

 

 

고도화 모델

nn.sequential

linear, batchnorem, relu, dropout을 적용

# ============================================================
# 7. 성능 고도화 모델 정의
# ============================================================

class ImprovedMLP(nn.Module):
    # 개선 모델은 기본 모델보다 더 깊고 안정적인 구조를 사용합니다.
    def __init__(self, input_dim, output_dim):
        super().__init__()

        self.net = nn.Sequential(
            # 첫 번째 완전연결층입니다.
            # 입력 특성을 64개의 은닉 노드로 확장합니다.
            nn.Linear(input_dim, 64),

            # BatchNorm1d는 각 미니배치의 분포를 정규화하여 학습을 안정화합니다.
            nn.BatchNorm1d(64),

            # ReLU 활성화 함수입니다.
            nn.ReLU(),

            # Dropout은 학습 중 일부 뉴런을 무작위로 꺼서 과적합을 줄입니다.
            # p=0.3은 약 30%의 뉴런 출력을 임시로 0으로 만든다는 뜻입니다.
            nn.Dropout(p=0.3),

            # 두 번째 완전연결층입니다.
            nn.Linear(64, 32),

            # 두 번째 Batch Normalization입니다.
            nn.BatchNorm1d(32),

            # 두 번째 ReLU 활성화 함수입니다.
            nn.ReLU(),

            # 두 번째 Dropout입니다.
            nn.Dropout(p=0.2),

            # 최종 출력층입니다.
            # output_dim=2이므로 두 클래스에 대한 점수를 출력합니다.
            nn.Linear(32, output_dim)
        )

    def forward(self, x):
        return self.net(x)


# 개선 모델 객체를 생성하고 device로 이동합니다.
improved_model = ImprovedMLP(input_dim, output_dim).to(device)

print(improved_model)

 

 

 

 

 

 

학습함수와 평가함수정의

loader에서 미니배치 단위로 데이터를 꺼내 zero_grad, model(), criteriion을 손실계산, loss.backward(), optimizer.step()

평가함수때는 model.eval(), no_grad()로 기울기를 계싼하지않으며 메모리 사용량을 줄이고 손실률 계산. 

torch.argmx() preds 계산. 

correct, total구해 avg_loss와 accuracy 출력

# ============================================================
# 8. 학습 함수와 평가 함수 정의
# ============================================================

def train_one_epoch(model, loader, criterion, optimizer):
    # model.train()은 모델을 학습 모드로 전환합니다.
    # Dropout과 BatchNorm이 학습 방식으로 동작합니다.
    model.train()

    # 한 epoch 동안의 손실 합계를 저장합니다.
    total_loss = 0.0

    # 맞힌 개수와 전체 개수를 저장합니다.
    correct = 0
    total = 0

    # DataLoader에서 미니배치 단위로 데이터를 꺼냅니다.
    for batch_X, batch_y in loader:
        # 입력과 정답 데이터를 현재 장치로 이동합니다.
        batch_X = batch_X.to(device)
        batch_y = batch_y.to(device)

        # 이전 미니배치에서 계산된 기울기를 초기화합니다.
        optimizer.zero_grad()

        # 모델에 입력 데이터를 넣어 예측값을 계산합니다.
        outputs = model(batch_X)

        # 예측값과 정답을 비교하여 손실을 계산합니다.
        loss = criterion(outputs, batch_y)

        # 손실을 기준으로 역전파를 수행하여 각 가중치의 기울기를 계산합니다.
        loss.backward()

        # 계산된 기울기를 사용하여 모델 파라미터를 업데이트합니다.
        optimizer.step()

        # 현재 미니배치 손실에 데이터 개수를 곱해 누적합니다.
        total_loss += loss.item() * batch_X.size(0)

        # outputs에서 가장 큰 점수를 가진 클래스 번호를 예측 클래스로 선택합니다.
        preds = torch.argmax(outputs, dim=1)

        # 예측과 정답이 같은 개수를 누적합니다.
        correct += (preds == batch_y).sum().item()

        # 전체 데이터 개수를 누적합니다.
        total += batch_y.size(0)

    # 평균 손실과 정확도를 반환합니다.
    avg_loss = total_loss / total
    accuracy = correct / total

    return avg_loss, accuracy


def evaluate(model, loader, criterion):
    # model.eval()은 모델을 평가 모드로 전환합니다.
    # Dropout은 꺼지고 BatchNorm은 저장된 통계값을 사용합니다.
    model.eval()

    total_loss = 0.0
    correct = 0
    total = 0

    # 평가 시에는 기울기를 계산할 필요가 없으므로 no_grad를 사용합니다.
    # 메모리 사용량과 계산량을 줄일 수 있습니다.
    with torch.no_grad():
        for batch_X, batch_y in loader:
            batch_X = batch_X.to(device)
            batch_y = batch_y.to(device)

            outputs = model(batch_X)
            loss = criterion(outputs, batch_y)

            total_loss += loss.item() * batch_X.size(0)

            preds = torch.argmax(outputs, dim=1)
            correct += (preds == batch_y).sum().item()
            total += batch_y.size(0)

    avg_loss = total_loss / total
    accuracy = correct / total

    return avg_loss, accuracy

 

 

 

 

 

전체 학습, 실행 함수 정의

fit_model

epoch만큼 range로 돌린다. 

train_one_epoch(), evaluate() 기록을 저장해 이전 성능보다 좋아지면 모델 저장. 성능이 개선되었으면 patience_counter를 0, 개선이 없으면 patience_counter를 1로 증가시킨다. 

early stopping 조건을 확인

가장 손실이 낮은 모델을 불러온다 .

# ============================================================
# 9. 전체 학습 실행 함수 정의
# ============================================================

def fit_model(
    model,
    train_loader,
    val_loader,
    criterion,
    optimizer,
    epochs=100,
    scheduler=None,
    early_stopping_patience=None
):
    # 학습 과정을 기록하기 위한 딕셔너리입니다.
    history = {
        "train_loss": [],
        "train_acc": [],
        "val_loss": [],
        "val_acc": []
    }

    # 검증 손실이 가장 낮았던 값을 저장합니다.
    best_val_loss = float("inf")

    # 가장 성능이 좋았던 모델의 가중치를 저장합니다.
    best_state = None

    # Early Stopping에서 개선이 없었던 epoch 수를 저장합니다.
    patience_counter = 0

    # 지정한 epoch 수만큼 학습을 반복합니다.
    for epoch in range(1, epochs + 1):
        # 한 epoch 학습을 수행합니다.
        train_loss, train_acc = train_one_epoch(model, train_loader, criterion, optimizer)

        # 검증 데이터로 모델 성능을 확인합니다.
        val_loss, val_acc = evaluate(model, val_loader, criterion)

        # scheduler가 있으면 검증 손실을 기준으로 learning rate를 조정합니다.
        if scheduler is not None:
            scheduler.step(val_loss)

        # 기록을 저장합니다.
        history["train_loss"].append(train_loss)
        history["train_acc"].append(train_acc)
        history["val_loss"].append(val_loss)
        history["val_acc"].append(val_acc)

        # 현재 검증 손실이 이전 최고 성능보다 낮으면 모델을 저장합니다.
        if val_loss < best_val_loss:
            best_val_loss = val_loss

            # state_dict는 모델의 학습된 가중치 정보를 담고 있습니다.
            best_state = {
                key: value.cpu().clone()
                for key, value in model.state_dict().items()
            }

            # 성능이 개선되었으므로 patience_counter를 0으로 초기화합니다.
            patience_counter = 0
        else:
            # 성능 개선이 없으면 patience_counter를 1 증가시킵니다.
            patience_counter += 1

        # 10 epoch마다 학습 상태를 출력합니다.
        if epoch % 10 == 0 or epoch == 1:
            print(
                f"Epoch {epoch:03d} | "
                f"Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.4f} | "
                f"Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.4f}"
            )

        # Early Stopping 조건을 확인합니다.
        if early_stopping_patience is not None:
            if patience_counter >= early_stopping_patience:
                print(f"Early Stopping 발생: {epoch} epoch에서 학습 중단")
                break

    # 가장 검증 손실이 낮았던 모델 가중치를 다시 불러옵니다.
    if best_state is not None:
        model.load_state_dict(best_state)

    return history

 

 

 

 

 

모델 학습

nn.crossentropyloss() 기본 모델의 손실함수 객체 만들기..

optimadam 사용하기

fit_model()

# ============================================================
# 10. 기본 모델 학습
# ============================================================

# 기본 모델의 손실함수입니다.
# CrossEntropyLoss는 다중 분류와 이진 분류 클래스 번호 학습에 사용할 수 있습니다.
basic_criterion = nn.CrossEntropyLoss()

# 기본 모델은 Adam Optimizer를 사용합니다.
# lr은 learning rate로, 가중치를 얼마나 크게 업데이트할지 결정합니다.
basic_optimizer = optim.Adam(
    basic_model.parameters(),
    lr=0.001
)

# 기본 모델 학습을 실행합니다.
basic_history = fit_model(
    model=basic_model,
    train_loader=train_loader,
    val_loader=val_loader,
    criterion=basic_criterion,
    optimizer=basic_optimizer,
    epochs=100,
    scheduler=None,
    early_stopping_patience=None
)

Epoch 001 | Train Loss: 0.5986, Train Acc: 0.8299 | Val Loss: 0.5474, Val Acc: 0.9211
Epoch 010 | Train Loss: 0.1604, Train Acc: 0.9619 | Val Loss: 0.1596, Val Acc: 0.9561
Epoch 020 | Train Loss: 0.0838, Train Acc: 0.9853 | Val Loss: 0.0945, Val Acc: 0.9737
Epoch 030 | Train Loss: 0.0600, Train Acc: 0.9883 | Val Loss: 0.0777, Val Acc: 0.9737
Epoch 040 | Train Loss: 0.0483, Train Acc: 0.9912 | Val Loss: 0.0715, Val Acc: 0.9737
Epoch 050 | Train Loss: 0.0407, Train Acc: 0.9912 | Val Loss: 0.0693, Val Acc: 0.9737
Epoch 060 | Train Loss: 0.0348, Train Acc: 0.9912 | Val Loss: 0.0681, Val Acc: 0.9737
Epoch 070 | Train Loss: 0.0300, Train Acc: 0.9912 | Val Loss: 0.0667, Val Acc: 0.9737
Epoch 080 | Train Loss: 0.0255, Train Acc: 0.9912 | Val Loss: 0.0657, Val Acc: 0.9737
Epoch 090 | Train Loss: 0.0216, Train Acc: 0.9912 | Val Loss: 0.0644, Val Acc: 0.9825
Epoch 100 | Train Loss: 0.0185, Train Acc: 0.9912 | Val Loss: 0.0630, Val Acc: 0.9825

 

 

 

 

 

 

 

개선모델 학습 

corssentropyloss 사용

optim.adamw사용!!

improved_model = ImprovedMLP(input_dim, output_dim).to(device) 가장 좋았던 모델을 불러와 적용

improved_model.parameters()

lr와 weight decay 조절. 가중치가 너무 커지는것을 막는다. 

optim.lr_scheduler.ReduceLROnPlateau() 검증 손실이 좋아지지않으면 lr를 줄인다.  학습 후반부에 더 세밀하게 최적점을 찾도록 도와주는것. 

fit_model()

# ============================================================
# 11. 개선 모델 학습
# ============================================================

# 개선 모델도 같은 분류 문제이므로 CrossEntropyLoss를 사용합니다.
improved_criterion = nn.CrossEntropyLoss()

# AdamW는 Adam에 Weight Decay를 더 안정적으로 적용한 최적화 알고리즘입니다.
# weight_decay는 가중치가 너무 커지는 것을 막아 과적합을 줄이는 데 도움을 줍니다.
improved_optimizer = optim.AdamW(
    improved_model.parameters(),
    lr=0.001,
    weight_decay=0.01
)

# ReduceLROnPlateau는 검증 손실이 좋아지지 않으면 learning rate를 줄입니다.
# 학습 후반부에 더 세밀하게 최적점을 찾도록 도와줍니다.
improved_scheduler = optim.lr_scheduler.ReduceLROnPlateau(
    improved_optimizer,
    mode="min",
    factor=0.5,
    patience=5
)

# 개선 모델 학습을 실행합니다.
# Early Stopping을 사용하여 검증 성능이 오래 개선되지 않으면 학습을 중단합니다.
improved_history = fit_model(
    model=improved_model,
    train_loader=train_loader,
    val_loader=val_loader,
    criterion=improved_criterion,
    optimizer=improved_optimizer,
    epochs=200,
    scheduler=improved_scheduler,
    early_stopping_patience=20
)

Epoch 001 | Train Loss: 0.6908, Train Acc: 0.6364 | Val Loss: 0.5712, Val Acc: 0.6754
Epoch 010 | Train Loss: 0.1394, Train Acc: 0.9677 | Val Loss: 0.1402, Val Acc: 0.9737
Epoch 020 | Train Loss: 0.1207, Train Acc: 0.9677 | Val Loss: 0.0866, Val Acc: 0.9825
Epoch 030 | Train Loss: 0.0537, Train Acc: 0.9883 | Val Loss: 0.0693, Val Acc: 0.9737
Epoch 040 | Train Loss: 0.0729, Train Acc: 0.9677 | Val Loss: 0.0577, Val Acc: 0.9912
Epoch 050 | Train Loss: 0.0593, Train Acc: 0.9795 | Val Loss: 0.0592, Val Acc: 0.9912
Early Stopping 발생: 58 epoch에서 학습 중단

 

 

 

 

시각화

# ============================================================
# 12. 학습 곡선 시각화 함수
# ============================================================

def plot_history(history, title):
    # 손실 그래프를 그립니다.
    plt.figure(figsize=(8, 5))
    plt.plot(history["train_loss"], label="Train Loss")
    plt.plot(history["val_loss"], label="Validation Loss")
    plt.xlabel("Epoch")
    plt.ylabel("Loss")
    plt.title(title + " - Loss")
    plt.legend()
    plt.grid(True)
    plt.show()

    # 정확도 그래프를 그립니다.
    plt.figure(figsize=(8, 5))
    plt.plot(history["train_acc"], label="Train Accuracy")
    plt.plot(history["val_acc"], label="Validation Accuracy")
    plt.xlabel("Epoch")
    plt.ylabel("Accuracy")
    plt.title(title + " - Accuracy")
    plt.legend()
    plt.grid(True)
    plt.show()


# 기본 모델의 학습 과정 그래프를 출력합니다.
plot_history(basic_history, "Basic Model")

# 개선 모델의 학습 과정 그래프를 출력합니다.
plot_history(improved_history, "Improved Model")

 

별로 좋아진것같지않다. . 우선순위인 데이터 양을 늘리는게 좋겟다. 한계가 있는것. 

 

 

 

최종평가

# ============================================================
# 13. 테스트 데이터 최종 평가
# ============================================================

def predict_all(model, loader):
    # 모델을 평가 모드로 전환합니다.
    model.eval()

    # 전체 예측값과 전체 정답값을 저장할 리스트입니다.
    all_preds = []
    all_targets = []

    # 평가 시에는 기울기 계산을 하지 않습니다.
    with torch.no_grad():
        for batch_X, batch_y in loader:
            batch_X = batch_X.to(device)

            # 모델 예측 점수를 계산합니다.
            outputs = model(batch_X)

            # 가장 높은 점수를 가진 클래스를 최종 예측값으로 선택합니다.
            preds = torch.argmax(outputs, dim=1).cpu().numpy()

            # 예측값과 실제 정답을 리스트에 저장합니다.
            all_preds.extend(preds)
            all_targets.extend(batch_y.numpy())

    return np.array(all_targets), np.array(all_preds)


# 기본 모델의 테스트 예측 결과를 구합니다.
y_true_basic, y_pred_basic = predict_all(basic_model, test_loader)

# 개선 모델의 테스트 예측 결과를 구합니다.
y_true_improved, y_pred_improved = predict_all(improved_model, test_loader)

# 정확도를 계산합니다.
basic_test_acc = accuracy_score(y_true_basic, y_pred_basic)
improved_test_acc = accuracy_score(y_true_improved, y_pred_improved)

print("기본 모델 테스트 정확도:", round(basic_test_acc, 4))
print("개선 모델 테스트 정확도:", round(improved_test_acc, 4))

기본 모델 테스트 정확도: 0.9561
개선 모델 테스트 정확도: 0.9474

 

 

 

 

 

 

 

 

혼동행렬 및 결과비교표 만들기

# ============================================================
# 14. 혼동행렬과 분류 리포트 확인
# ============================================================

# 혼동행렬은 모델이 어떤 클래스를 맞히고 틀렸는지 표로 보여줍니다.
print("기본 모델 혼동행렬")
print(confusion_matrix(y_true_basic, y_pred_basic))

print("\n개선 모델 혼동행렬")
print(confusion_matrix(y_true_improved, y_pred_improved))

# classification_report는 precision, recall, f1-score를 한 번에 보여줍니다.
print("\n기본 모델 분류 리포트")
print(classification_report(y_true_basic, y_pred_basic, target_names=data.target_names))

print("\n개선 모델 분류 리포트")
print(classification_report(y_true_improved, y_pred_improved, target_names=data.target_names))

# ============================================================
# 15. 결과 비교표 만들기
# ============================================================

# 테스트 결과를 표 형태로 정리합니다.
result_df = pd.DataFrame({
    "Model": ["Basic MLP", "Improved MLP"],
    "Test Accuracy": [basic_test_acc, improved_test_acc],
    "Applied Methods": [
        "StandardScaler + Simple MLP + Adam",
        "StandardScaler + Deeper MLP + BatchNorm + Dropout + AdamW + Weight Decay + Scheduler + Early Stopping"
    ]
})

display(result_df)

 

 

 

 

 

 

어제 확인했던 face_churn_app을 확인해보자. 

mysql을 추가했다. 

 

 

 

 

import

비밀번호 해시용 임의 salt를 생성하기 위해 secrets를 사용한다. 

# 환경 변수에서 DB 접속 정보를 읽기 위해 os 모듈을 사용합니다.
import os

# 비밀번호를 평문으로 저장하지 않기 위해 안전한 해시 계산에 hashlib를 사용합니다.
import hashlib

# 비밀번호 해시용 임의 salt를 생성하기 위해 secrets를 사용합니다.
import secrets

# 타입 힌트를 위해 Optional, Tuple을 사용합니다.
from typing import Optional, Tuple

# NumPy 배열인 얼굴 임베딩을 bytes로 변환하고 복원하기 위해 numpy를 사용합니다.
import numpy as np

# MySQL 접속을 위해 mysql-connector-python 패키지를 사용합니다.
import mysql.connector

 

 

 

 

사용자 테이블 정의

접속 값 읽기

# 사용자 테이블 이름을 상수로 정의합니다.
USER_TABLE = "face_users"


# DB 접속 기본값을 환경 변수에서 읽습니다.
# 실제 운영 환경에서는 아래 환경 변수를 사용자의 MySQL 환경에 맞게 설정하면 됩니다.
DB_HOST = os.getenv("MYSQL_HOST", "localhost")
DB_PORT = int(os.getenv("MYSQL_PORT", "3306"))
DB_USER = os.getenv("MYSQL_USER", "root")
DB_PASSWORD = os.getenv("MYSQL_PASSWORD", "1234")
DB_NAME = os.getenv("MYSQL_DATABASE", "face_churn_db")

 

 

 

연결객체를 만들어 반환한다.

def get_connection(database: Optional[str] = DB_NAME):
    """MySQL 연결 객체를 생성하여 반환합니다."""

    # mysql.connector.connect는 MySQL 서버에 접속하는 함수입니다.
    # database=None이면 특정 DB 선택 없이 서버에만 접속합니다.
    return mysql.connector.connect(
        host=DB_HOST,
        port=DB_PORT,
        user=DB_USER,
        password=DB_PASSWORD,
        database=database,
        charset="utf8mb4",
        use_unicode=True,
    )

 

 

 

 

 

테이블이 없으면 자동으로 생성한다.

def init_db() -> None:
    """DB와 사용자 테이블이 없으면 자동으로 생성합니다."""

    # 먼저 database 없이 MySQL 서버에 접속합니다.
    conn = get_connection(database=None)

    # cursor는 SQL 문장을 실행하는 객체입니다.
    cur = conn.cursor()

    # 지정한 데이터베이스가 없으면 생성합니다.
    cur.execute(
        f"CREATE DATABASE IF NOT EXISTS {DB_NAME} "
        "DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
    )

    # DB 생성 명령을 실제 반영합니다.
    conn.commit()

    # cursor와 연결을 닫아 자원을 해제합니다.
    cur.close()
    conn.close()

    # 생성된 데이터베이스에 다시 접속합니다.
    conn = get_connection(database=DB_NAME)
    cur = conn.cursor()

    # 사용자 계정, 비밀번호 해시, 이름, 얼굴 임베딩을 저장할 테이블을 생성합니다.
    # face_embedding은 512차원 float32 벡터를 bytes로 변환한 값을 저장합니다.
    cur.execute(
        f"""
        CREATE TABLE IF NOT EXISTS {USER_TABLE} (
            user_id VARCHAR(100) PRIMARY KEY,
            user_name VARCHAR(100) NOT NULL,
            password_salt VARCHAR(64) NOT NULL,
            password_hash VARCHAR(128) NOT NULL,
            face_embedding LONGBLOB NOT NULL,
            face_image_path VARCHAR(500),
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
        """
    )

    conn.commit()
    cur.close()
    conn.close()

 

 

 

 

salt가 없으면 secrets.token_hex(16) 32자리 난수 문자열을 생성한다.

hashlib.ppkdf2_hmac() 반복해시를 수행한다. 

16진수문자열로 변환하여 db에 저장하기 쉽게 만든다.

def hash_password(password: str, salt: Optional[str] = None) -> Tuple[str, str]:
    """비밀번호를 PBKDF2 방식으로 해시 처리합니다."""

    # salt가 없으면 32자리 난수 문자열을 새로 생성합니다.
    if salt is None:
        salt = secrets.token_hex(16)

    # pbkdf2_hmac은 반복 해시를 수행하여 무차별 대입 공격을 어렵게 합니다.
    digest = hashlib.pbkdf2_hmac(
        "sha256",
        password.encode("utf-8"),
        salt.encode("utf-8"),
        100_000,
    )

    # bytes 결과를 16진수 문자열로 변환하여 DB에 저장하기 쉽게 만듭니다.
    return salt, digest.hex()

 

 

 

 

얼굴 numpy배열을 저장하기위해 mysql blob bytes로 저장한다.

astype float32로 해야 크기와 값이 안정적이다 .return.

def embedding_to_bytes(embedding: np.ndarray) -> bytes:
    """얼굴 임베딩 NumPy 배열을 MySQL BLOB 저장용 bytes로 변환합니다."""

    # float32 타입으로 고정해야 저장/복원 시 크기와 값이 안정적입니다.
    return embedding.astype(np.float32).tobytes()

 

 

 

 

mysql blob에서 읽은 bytes를 얼굴 임베딩 numpy배열로 복원한다. 

frombuffer 로 bytes데이터를 float32배열로 해석하고

np.linalg.norm코사인 유사도 계산 안정을 위해 다시 정규화한다.

def bytes_to_embedding(blob: bytes) -> np.ndarray:
    """MySQL BLOB에서 읽은 bytes를 얼굴 임베딩 NumPy 배열로 복원합니다."""

    # frombuffer는 bytes 데이터를 float32 배열로 해석합니다.
    embedding = np.frombuffer(blob, dtype=np.float32)

    # 코사인 유사도 계산 안정성을 위해 다시 정규화합니다.
    norm = np.linalg.norm(embedding)
    if norm == 0:
        return embedding
    return embedding / norm

 

 

 

 

사용자 id존재여부 확인

def user_exists(user_id: str) -> bool:
    """사용자 ID가 이미 등록되어 있는지 확인합니다."""

    init_db()
    conn = get_connection()
    cur = conn.cursor()

    # COUNT(*)로 해당 ID의 존재 여부를 확인합니다.
    cur.execute(f"SELECT COUNT(*) FROM {USER_TABLE} WHERE user_id = %s", (user_id,))
    count = cur.fetchone()[0]

    cur.close()
    conn.close()

    return count > 0

 

 

 

사용자 정보와 얼굴 임베딩 저장

def create_or_update_user(user_id: str, password: str, user_name: str, embedding: np.ndarray, image_path: str) -> None:
    """사용자 정보와 얼굴 임베딩을 MySQL에 저장합니다."""

    init_db()

    # 입력 비밀번호를 salt + hash 형태로 변환합니다.
    salt, password_hash = hash_password(password)

    # 얼굴 임베딩 배열을 BLOB 저장용 bytes로 변환합니다.
    embedding_blob = embedding_to_bytes(embedding)

    conn = get_connection()
    cur = conn.cursor()

    # 같은 user_id가 있으면 사용자 정보와 얼굴 정보를 갱신합니다.
    cur.execute(
        f"""
        INSERT INTO {USER_TABLE}
            (user_id, user_name, password_salt, password_hash, face_embedding, face_image_path)
        VALUES
            (%s, %s, %s, %s, %s, %s)
        ON DUPLICATE KEY UPDATE
            user_name = VALUES(user_name),
            password_salt = VALUES(password_salt),
            password_hash = VALUES(password_hash),
            face_embedding = VALUES(face_embedding),
            face_image_path = VALUES(face_image_path)
        """,
        (user_id, user_name, salt, password_hash, embedding_blob, image_path),
    )

    conn.commit()
    cur.close()
    conn.close()

 

 

 

 

id비번확인

def verify_user_password(user_id: str, password: str) -> Tuple[bool, Optional[str], str]:
    """아이디와 비밀번호가 MySQL에 저장된 값과 일치하는지 확인합니다."""

    init_db()
    conn = get_connection()
    cur = conn.cursor(dictionary=True)

    # 입력된 user_id에 해당하는 사용자 정보를 조회합니다.
    cur.execute(
        f"SELECT user_id, user_name, password_salt, password_hash FROM {USER_TABLE} WHERE user_id = %s",
        (user_id,),
    )
    row = cur.fetchone()

    cur.close()
    conn.close()

    # 사용자가 없으면 실패를 반환합니다.
    if row is None:
        return False, None, "등록되지 않은 아이디입니다."

    # DB에 저장된 salt로 입력 비밀번호를 다시 해시합니다.
    _, input_hash = hash_password(password, salt=row["password_salt"])

    # secrets.compare_digest는 문자열 비교 시 타이밍 공격 위험을 줄입니다.
    if not secrets.compare_digest(input_hash, row["password_hash"]):
        return False, None, "암호가 일치하지 않습니다."

    return True, row["user_name"], "아이디와 암호 확인이 완료되었습니다."

 

 

 

 

 

id에 해당하는 얼굴 임베딩을 조회해 결과가없으면 none을 반환, 

def get_user_face_embedding(user_id: str) -> Optional[np.ndarray]:
    """특정 사용자 ID의 등록 얼굴 임베딩을 MySQL에서 읽어옵니다."""

    init_db()
    conn = get_connection()
    cur = conn.cursor()

    # user_id에 해당하는 얼굴 임베딩 BLOB을 조회합니다.
    cur.execute(f"SELECT face_embedding FROM {USER_TABLE} WHERE user_id = %s", (user_id,))
    row = cur.fetchone()

    cur.close()
    conn.close()

    # 조회 결과가 없으면 None을 반환합니다.
    if row is None or row[0] is None:
        return None

    # BLOB 데이터를 NumPy 임베딩 배열로 복원합니다.
    return bytes_to_embedding(row[0])