[Hugging Face] 벤치마크 클래스 만들기 ( Benchmark Class)

이번 포스팅에서는 허깅페이스 라이브러리를 이용해서 벤치마크 클래스를 만들어보도록 하겠습니다. 클래스에 특정한 벤치마크를 넣어서 한번에 출력하여 모델의 성능을 판단할 수 있습니다.

• 모델 성능 : Acc, F1 등으로 모델의 성능을 파악
• 메모리 : 모델의 파라미터의 양을 파악
• 레이턴시 : 모델이 얼마나 빠르게 예측을 하는지 파악

총 3가지의 벤치마크를 파악하는 클래스를 만들어보겠습니다. 

 

파이프라인은 텍스트 분류를 통해서 모델의 성능을 벤치마킹하겠습니다.

 

필요한 모듈 임포트

---

from transformers import pipeline
from datasets import load_dataset
from pathlib import Path
from time import perf_counter

import torch
import evaluate
import numpy as np

코드를 돌리기 위한 모듈을 모두 임포트 합니다.

 

모델 호출

---

# nsmc로 미세 조정된 ELECTRA 호출
model_ckpt = "monologg/koelectra-base-finetuned-nsmc"
pipe = pipeline("text-classification", model=model_ckpt, device="cuda" if torch.cuda.is_available() else "cpu")

nsmc 데이터로 Fine-tuning된 ELECTRA를 통해서 구현해보도록 하겠습니다.

 

예제 생성

---

input_data = "지금 매우 흥미로워"
pipe(input_data)

# Output : [{'label': 'positive', 'score': 0.9846755862236023}]

Text에 대해 긍정으로 분류한 것을 볼 수 있습니다.

 

평가 지표 임포트

---

accuracy_score = evaluate.load("accuracy")

정확도 지표 임포트 합니다.

 

라벨 매핑을 위한 클래스 생성

---

label_str = nsmc.features["label"]

라벨이 str로 되어있어 int로 바꾸기 위한 클래스를 생성합니다.

 

평가 함수 정의 및 실행

---

def compute_accuracy(pipeline, dataset):
    preds, labels = [], []
    for example in dataset:
        pred = pipeline(example["document"])[0]['label']
        label = example['label']
        preds.append(label_str.str2int(pred))
        labels.append(label)
    accuracy = accuracy_score.compute(predictions=preds, references=labels)
    print(f"테스트 세트 정확도 - {accuracy['accuracy']:.4f}")
    return accuracy
   
compute_accuracy(pipe, nsmc)

# output : 테스트 세트 정확도 - 0.9027

데이터에 대해 정확도를 평가하는 함수를 정의 및 실행

 

모델 크기 계산 함수 정의 및 실행

---

def compute_size(pipeline, dataset):
    state_dict = pipeline.model.state_dict()
    tmp_path = Path("model.pt")
    torch.save(state_dict, tmp_path)
    
    # calculate file size (MB)
    size_mb = Path(tmp_path).state_dict().st_size / (1024 ** 2)
    
    # delete tmp_file
    tmp_path.unlink()
    print(f"모델 사이즈 : {size_mb:.3f} (MB)")
    
    
compute_size(pipe, nsmc)

# output : 모델 사이즈 : 422.629 (MB)

 

평균 레이턴시 계산 함수 정의 및 실행

---

def time_pipeline(pipeline,  query="영화 리뷰는 즐겁다"):
    latencies = []
    # warming up
    for _ in range(10):
        _ = pipeline(query)

    # 실행 측정
    for _ in range(100):
        start_time = perf_counter()
        _ = pipeline(query)
        latency = perf_counter() - start_time
        latencies.append(latency)

    # 통계 계산
    time_avg_ms = 1000 * np.mean(latencies)
    time_std_ms = 1000 * np.std(latencies)

    print(f"평균 레이턴시 : {time_avg_ms:.2f} +\- {time_std_ms:.2f} (ms)")

time_pipeline(pipe)

# output : 평균 레이턴시 : 6.01 +\- 0.02 (ms)

 

실행 함수 정의

---

 def run_benchmark(self):
        metrics = {}
        metrics[self.optim_type] = self.compute_size()
        metrics[self.optim_type].update(self.time_pipeline())
        metrics[self.optim_type].update(self.compute_accuracy())
        return metrics

벤치마크를 한번에 실행하고 기록할 함수를 정의합니다.

 

 

각각의 함수 및 벤치마크  실행함수 클래스화

---

class PerformanceBenchmark:
    # 초기화 함수
    def __init__(self, pipeline, dataset, optim_type="ELECTRA"):
        self.pipeline = pipeline
        self.dataset = dataset
        self.optim_type = optim_type
        
    # 정확도 계산 함수
    def compute_accuracy(self):
        preds, labels = [], []
        for example in self.dataset:
            pred = self.pipeline(example["document"])[0]['label']
            label = example['label']
            preds.append(label_str.str2int(pred))
            labels.append(label)
        accuracy = accuracy_score.compute(predictions=preds, references=labels)
        print(f"테스트 세트 정확도 - {accuracy['accuracy']:.4f}")
        return accuracy
    
    # 모델 사이즈 계산 함수
    def compute_size(self):
        state_dict = self.pipeline.model.state_dict()
        tmp_path = Path("model.pt")
        torch.save(state_dict, tmp_path)

        # calculate file size (MB)
        size_mb = Path(tmp_path).stat().st_size / (1024 ** 2)

        # delete tmp_file
        tmp_path.unlink()
        print(f"모델 사이즈 : {size_mb:.3f} (MB)")
        return {"size_mb" : size_mb}
    
    def time_pipeline(self,  query="영화 리뷰는 즐겁다"):
        latencies = []
        # warming up
        for _ in range(10):
            _ = self.pipeline(query)

        # 실행 측정
        for _ in range(100):
            start_time = perf_counter()
            _ = self.pipeline(query)
            latency = perf_counter() - start_time
            latencies.append(latency)

        # 통계 계산
        time_avg_ms = 1000 * np.mean(latencies)
        time_std_ms = 1000 * np.std(latencies)
        print(f"평균 레이턴시 : {time_avg_ms:.2f} +\- {time_std_ms:.2f} (ms)")
        return {"time_avg_ms" : time_avg_ms, "time_std_ms": time_std_ms}
        
    def run_benchmark(self):
        metrics = {}
        metrics[self.optim_type] = self.compute_size()
        metrics[self.optim_type].update(self.time_pipeline())
        metrics[self.optim_type].update(self.compute_accuracy())
        return metrics

각각의 함수들을 클래스화 하고 계산한 값들을 return으로 반환합니다.

 

모델 임포트 및 베치마크 클래스 실행

---

model_ckpt = "monologg/koelectra-base-finetuned-nsmc"
pipe = pipeline("text-classification", model=model_ckpt,device="cuda")

pb = PerformanceBenchmark(pipe, nsmc)
perf_metrics = pb.run_benchmark()

 

결과

# 모델 사이즈 : 422.629 (MB)
# 평균 레이턴시 : 6.00 +\- 0.04 (ms)
# 테스트 세트 정확도 - 0.9027

 

봐주셔서 감사합니다. 오류 있으면 댓글로 부탁드립니다. :)