[NLP] 자연어처리 pipline

Intro 

뉴스 기사(제목, 내용)을 입력 값으로 받아들이고 해당 뉴스의 제목이 낚시성인지 아닌지를 판별하는 모델을 만들어 볼 것이다. 이때 제목과 본문 사이에 관계가 아래 6가지에 포함되는 경우에 한하여 낚시성 기사로 판단한다. 

데이터 불러오기

: pd.dateframe()

 

- 옵션

1) fillpath or buffer : 위치한 경로

2) sep or delimiter : 구분자 지정 , default = ','

3) header : 헤더가 있는 행 지정

4) dtype : 각 컬럼의 데이터 타임 지정 

5) encoding : 문자 인코딩 지정

 

데이터셋 클래스

: 주워진 데이터셋을 가공하는 class

- 배치, shuffle, 멀티 프로세싱이 용이

- 구조화된 방식으로 데이터 관리 가능.

 

구조)

- init

- len : 전체 길이 반환

- getitem : index에 해당되는 데이터 반환

 

구현) 

class news_dataset(torch.utils.data.Dataset):
    """dataframe을 torch dataset class로 변환"""
    def __init__(self, news_dataset, labels):
        self.dataset = news_dataset
        self.labels = labels

    def __getitem__(self,idx):
        item = {
            key: val[idx].clone().detach() for key, val in self.dataset.items()
        }
        item['labels'] = torch.tensor(self.labels[idx])
        return item

    def __len__(self):
        return len(self.labels)

Tokenizing

: huggigface에 다양한 종류의 토크나이저를 가져온다. 

 

Model 

: huggingface에 다양한 모델을 가져온다. 이떄 토크나이저와 일치해야 한다. 

 

구현 - 데이터 불러오기)

def load_data(dataset_dir):
    """csv file을 dataframe으로 load"""
    dataset = pd.read_csv(dataset_dir)[:500]
    print("dataframe 의 형태")
    print(dataset.head())
    return dataset

 

구현 - 토크나이저)

def construct_tokenized_dataset(dataset,tokenizer, max_length):
    """[뉴스제목 + [SEP] + 뉴스본문]형태로 토크나이징"""
    concat_entity = []
    for title, body in zip(dataset["newsTitle"],dataset["newsContent"]):
        total = str(title) + "[SEP]" + str(body)
        concat_entity.append(total)
    print("tokenizer 에 들어가는 데이터 형태")
    print(concat_entity[:5])
    tokenized_senetences = tokenizer(
        concat_entity,
        return_tensors = "pt",
        padding = True,
        truncation = True,
        max_length = max_length,
        add_special_tokens = True,
        return_token_type_ids=False, # BERT 이후 모델(RoBERTa 등) 사용할때 On
    )
    print("tokenizing 된 데이터 형태")
    print(tokenized_senetences[:5])
    return tokenized_senetences

[SEP] 라는 문자열을 기사와 기사 내용 사이에 넣어 쌍을 만들고 concat_entity라는 리스트형 변수에 추가한다. 이후 tokenizer() 함수를 통해 토크나이저로 만들어준다.

 

Trainer

: 학습,평가,최적화를 위한 클래스

-> batch learning, learning schedular, earlystopping 사용 가능

 

옵션)

1. save_total_limit : 몇 개의 모델을 저장할 것인지

2. warmup_steps : warmup을 몇 step까지 진행할지.

- warmup : 모델의 학습률을 점진적으로 증가시키는 과정

3) compute : metrics (점수 계산 방식을 사전에 정의한 방식으로 바꿀 수 있음.)

4) call_back : traning 상태 별로 취할 action 설정

4-2) early_stopping (patience : 학습이 개선되지 않는 상황을 얼마나 참을지)

                                (threshold : 개선되지 않았음에 기준점)

5) optimizer - 옵티마이저 및 학습 스케줄러 설정

 

 

구현 - 전처리 과정)

def prepare_dataset(dataset_dir, tokenizer,max_len):
    """학습(train)과 평가(test)를 위한 데이터셋을 준비"""
    # load_data
    train_dataset = load_data(os.path.join(dataset_dir, "train.csv"))
    test_dataset = load_data(os.path.join(dataset_dir, "test.csv"))

    # split train / validation = 7.5 : 2.5
    train_dataset, val_dataset = train_test_split(train_dataset,test_size=0.25,random_state=42,stratify=train_dataset['label'])
    
    # split label
    train_label = train_dataset['label'].values
    val_label = val_dataset['label'].values
    test_label = test_dataset['label'].values

    # tokenizing dataset
    tokenized_train = construct_tokenized_dataset(train_dataset, tokenizer, max_len)
    tokenized_val = construct_tokenized_dataset(val_dataset, tokenizer, max_len)
    tokenized_test = construct_tokenized_dataset(test_dataset, tokenizer, max_len)
    print("--- tokenizing Done ---")

    # make dataset for pytorch.
    news_train_dataset = news_dataset(tokenized_train, train_label)
    news_val_dataset = news_dataset(tokenized_val, val_label)
    news_test_dataset = news_dataset(tokenized_test, test_label)
    print("--- dataset class Done ---")

    return news_train_dataset , news_val_dataset, news_test_dataset , test_dataset

 

구현 - 평가지표)

def compute_metrics(pred):
    """validation을 위한 metrics function"""
    labels = pred.label_ids
    #여러 개의 예측 중 확률이 가장 높은 값 추출.
    preds = pred.predictions.argmax(-1) 

    # calculate accuracy using sklearn's function
    acc = accuracy_score(labels, preds)

    # calculate f1 score using sklearn's function
    f1 = f1_score(labels, preds, average='micro')

    return {
        "accuracy": acc,
        "f1": f1,
    }

Accuracy와 F1 score를 사용.

 

구현 - 모델 설정)

def load_tokenizer_and_model_for_train():
    """학습(train)을 위한 사전학습(pretrained) 토크나이저와 모델을 huggingface에서 load"""
    # load model and tokenizer
    MODEL_NAME = args.model_name
    tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)

    # setting model hyperparameter
    model_config = AutoConfig.from_pretrained(MODEL_NAME)
    model_config.num_labels = 2
    print(model_config)

    model = AutoModelForSequenceClassification.from_pretrained(
        MODEL_NAME, config=model_config
    )
    print("--- Modeling Done ---")
    return tokenizer , model

 

구현 - trainer 모듈)

def load_trainer_for_train(model,news_train_dataset,news_val_dataset):
    """학습(train)을 위한 huggingface trainer 설정"""
    training_args = TrainingArguments(
        output_dir=args.save_path + "results",  # output directory
        save_total_limit=args.save_limit,  # number of total save model.
        save_steps=args.save_step,  # model saving step.
        num_train_epochs=args.epochs,  # total number of training epochs
        learning_rate=args.lr,  # learning_rate
        per_device_train_batch_size=args.batch_size,  # batch size per device during training
        per_device_eval_batch_size=2,  # batch size for evaluation
        warmup_steps=args.warmup_steps,  # number of warmup steps for learning rate scheduler
        weight_decay=args.weight_decay,  # strength of weight decay
        logging_dir=args.save_path + "logs",  # directory for storing logs
        logging_steps=args.logging_step,  # log saving step.
        evaluation_strategy="steps",  # evaluation strategy to adopt during training
            # `no`: No evaluation during training.
            # `steps`: Evaluate every `eval_steps`.
            # `epoch`: Evaluate every end of epoch.
        eval_steps=args.eval_step,  # evaluation step.
        load_best_model_at_end=True,
    )

    ## Add callback & optimizer & scheduler
    MyCallback = EarlyStoppingCallback(
        early_stopping_patience=3, early_stopping_threshold=0.001
    )

    optimizer = torch.optim.AdamW(
        model.parameters(),
        lr=args.lr,
        betas=(0.9, 0.999),
        eps=1e-08,
        weight_decay=args.weight_decay,
        amsgrad=False,
    )
    print("--- Set training arguments Done ---")

    trainer = Trainer(
        model=model,  # the instantiated 🤗 Transformers model to be trained
        args=training_args,  # training arguments, defined above
        train_dataset=news_train_dataset,  # training dataset
        eval_dataset=news_val_dataset,  # evaluation dataset
        compute_metrics=compute_metrics,  # define metrics function
        callbacks=[MyCallback],
        optimizers=(
            optimizer,
            get_cosine_with_hard_restarts_schedule_with_warmup(
                    optimizer,
                    num_warmup_steps=args.warmup_steps,
                    num_training_steps=len(news_train_dataset) * args.epochs,
            ),
        ),
    )
    print("--- Set Trainer Done ---")

    return trainer

 

구현 - train)

def train():
    """모델을 학습(train)하고 best model을 저장"""
    # fix a seed, 모델의 재현성 
    pl.seed_everything(seed=42, workers=False)

    # set device
    device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
    print("device:", device)

    # set model and tokenizer
    tokenizer , model = load_tokenizer_and_model_for_train()
    model.to(device)

    # set data
    news_train_dataset, news_val_dataset, news_test_dataset, test_dataset = prepare_dataset(args.dataset_dir,tokenizer,args.max_len)

    # set trainer
    trainer = load_trainer_for_train(model,news_train_dataset,news_val_dataset)

    # train model
    print("--- Start train ---")
    trainer.train()
    print("--- Finish train ---")
    model.save_pretrained("./best_model")

 

구현 - args 지정 및 학습 진행)

class args ():
  """학습(train)과 추론(infer)에 사용되는 arguments 관리하는 class"""
  dataset_dir = "./"
  model_type = "roberta" # 다른 모델 사용 가능 e.g) "bert" , "electra" ···
  model_name = "klue/roberta-large" # 다른 모델 사용 가능 e.g) "klue/bert-base" , "monologg/koelectra-base-finetuned-nsmc" ···
  save_path = "./"
  save_step = 200
  logging_step = 200
  eval_step = 100
  save_limit = 5
  seed = 42
  epochs = 1 # 10
  batch_size = 8 # 메모리 상황에 맞게 조절 e.g) 16 or 32
  max_len = 256
  lr = 3e-5
  weight_decay = 0.01
  warmup_steps = 300
  scheduler = "linear"
  model_dir = "./best_model" #추론 시, 저장된 모델 불러오는 경로 설정

train()

 

Inference(추론)

: 체크 포인트를 불러다 평가 데이터에 대한 추론 

model.eval()

: 모델을 평가모도로 전환 -> dropout, batch 등을 사용하지 않음.

 

torch.no_grad()

: 자동 미분 기능 비활성화 함수 -> 파라미터가 업데이트 되지 않음. 

 

구현 - 모델, 토크나이저 불러오기)

def load_model_for_inference():
    """추론(infer)에 필요한 모델과 토크나이저 load """
    # load tokenizer
    Tokenizer_NAME = args.model_name
    tokenizer = AutoTokenizer.from_pretrained(Tokenizer_NAME)

    ## load my model
    model = AutoModelForSequenceClassification.from_pretrained(args.model_dir)

    return tokenizer, model

 

구현 - 추론 함수)

def inference(model, tokenized_sent, device):
    """학습된(trained) 모델을 통해 결과를 추론하는 function"""
    dataloader = DataLoader(tokenized_sent, batch_size=args.batch_size, shuffle=False)
    model.eval()
    output_pred = []
    for i, data in enumerate(tqdm(dataloader)):
        with torch.no_grad():
            outputs = model(
                input_ids=data["input_ids"].to(device),
                attention_mask=data["attention_mask"].to(device),
            )
        logits = outputs[0]
        logits = logits.detach().cpu().numpy()
        result = np.argmax(logits, axis=-1)

        output_pred.append(result)
    return (np.concatenate(output_pred).tolist(),)

 

 

Evaluation

: 모델의 예측값과 평가 데이터의 라벨 값 사이의 평가 진행

i.e) f1_score 등

 

구현 - 추론 및 평가)

def infer_and_eval():
    """학습된 모델로 추론(infer)한 후에 예측한 결과(pred)를 평가(eval)"""
    # set device
    device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")

    # set model & tokenizer
    tokenizer, model = load_model_for_inference()
    model.to(device)

    # set dataㅁ
    news_train_dataset, news_val_dataset, news_test_dataset, test_dataset = prepare_dataset(args.dataset_dir,tokenizer,args.max_len)

    # predict answer
    pred_answer = inference(model, news_test_dataset, device)  # model에서 class 추론
    print("--- Prediction done ---")

    # evaluate between label and prediction
    labels = test_dataset['label'].values
    pred = pred_answer[0]

    acc = accuracy_score(labels, pred)
    f1 = f1_score(labels, pred, average='macro')
    print(f" ----- accuracy:{acc * 100:.1f}% -----")
    print(f"----- f1_score(macro): {f1 * 100:.1f}% ------")

    # make csv file with predicted answer
    output = pd.DataFrame(
        {
            "title": test_dataset["newsTitle"],
            "cleanBody": test_dataset["newsContent"],
            "clickbaitClass": pred,
        }
    )

    # 최종적으로 완성된 예측한 라벨 csv 파일 형태로 저장.
    result_path = "./prediction/"
    if not os.path.exists(result_path):
        os.makedirs(result_path)
    output.to_csv(
        os.path.join(result_path,"result.csv"), index=False
    )
    print("--- Save result ---")
    return output
    
output_df = infer_and_eval()
output_df.head(10)

 

 

result