TPU를 이용한 BERT GLUE task Train, Evaluation 해보기

BERT는 자연어 처리의 많은 문제를 해결하는 데 사용할 수 있습니다.

 

다음은 언어 모델을 평가하는 GLUE 벤치 마크에 대해서 BERT를 미세 조정하는 방법입니다.

 

GLUE 벤치 마크에는 다음과 같은 Task들이 있습니다.

 

1. CoLA (Corpus of Linguistic Acceptability) : 문장이 문법적으로 정확한지에 대한 Task

metrics : Matthews correlation

 

2. SST-2 (Stanford Sentiment Treebank) : 주어진 문장의 감정을 예측하는 Task

metrics : acc.

 

3. MRPC (Microsoft Research Paraphrase Corpus) : 한 쌍의 문장이 의미상 동일한 지 확인하는 Task

metrics : acc. / F1

 

4. QQP (Quora Question Pairs2) : 한 쌍의 질문이 의미 상 동일한 지 여부를 결정하는 Task

metrics : acc. / F1

 

5. MNLI (Multi-Genre Natural Language Inference) : 전체 문장과 가설 문장이 주어지면 전제가 가설을 수반하는지, 가설과 모순되는지 또는 둘 다(중립적) 여부를 예측하는 Task

metrics : matched acc. / mismatched acc.

 

6. QNLI (Question-answering Natural Language Inference) : 문맥 문장에 질문에 대한 답이 포함되어 있는지 여부를 결정하는 Task

metrics : acc

 

7. RTE (Recognizing Textual Entailment) : 문장이 주어진 가설을 수반하는지 여부를 결정하는 Task

metrics : acc.

 

8. WNLI (Winograd Natural Language Inference) : 대명사가 대체 된 문장이 원래 문장에 포함되는지 예측하는 Task

metrics : acc.

 

9. STS-B (STS shared task) : 문장들의 유사도를 얼마나 잘 파악하는지 평가하는 Task

metrics : Pearson / Spearman corr.

설정 (사전 준비)

colab.research.google.com/drive/1M88IPMasYWuAJ_0BexAA2_8L95mE-NxI?usp=sharing

TPU를 이용한 BERT GLUE Tasks 예제.ipynb

먼저 BERT를 미세 조정하기 위해 텍스트를 사용하기 전에 별도의 모델을 사용하여 텍스트를 전처리합니다.

 

다음 과정들은 전부 Colab에서 진행됩니다.

 

우선 Colab을 실행하시고 런타임 유형을 TPU로 변경해줍니다.

다음으로 필요한 패키지들을 설치해줍니다.

pip install -q -U tensorflow-text

pip install -q -U tf-models-official

pip install -U tfds-nightly

import os
import tensorflow as tf
import tensorflow_hub as hub
import tensorflow_datasets as tfds
import tensorflow_text as text # A dependency of the preprocessing model
import tensorflow_addons as tfa
from official.nlp import optimization
import numpy as np

tf.get_logger().setLevel('ERROR')

os.environ["TFHUB_MODEL_LOAD_FORMAT"]="UNCOMPRESSED"

TF Hub의 Cloud Storage 버킷에서 직접 체크 포인트를 읽도록 TFHub를 구성합니다. TPU에서 TF Hub 모델을 실행할 때만 권장됩니다.

 

이 설정이 없으면 TF Hub가 압축 파일을 다운로드하고 로컬에서 체크 포인트를 추출합니다. 이러한 로컬 파일에서 로드하려는 시도는 오류가 발생할 수 있습니다.

 

이 오류는 TPU가 Cloud Storage 버킷에서만 직접 읽을 수 있기 때문에 생기는 문제입니다.

TPU와 Cloud Storage 생성에 관하여 저번 게시물에서 확인할 수 있습니다.

choice-life.tistory.com/72

GCP(Google Cloud Platform) Storage, 버킷 만들기

 

다음 코드는 TPU 배포 전략을 정의합니다.

import os

if os.environ['COLAB_TPU_ADDR']:
  cluster_resolver = tf.distribute.cluster_resolver.TPUClusterResolver(tpu='')
  tf.config.experimental_connect_to_cluster(cluster_resolver)
  tf.tpu.experimental.initialize_tpu_system(cluster_resolver)
  strategy = tf.distribute.TPUStrategy(cluster_resolver)
  print('Using TPU')
elif tf.test.is_gpu_available():
  strategy = tf.distribute.MirroredStrategy()
  print('Using GPU')
else:
  raise ValueError('Running on CPU is not recommended.')

TensorFlow Hub에서 모델 로드

 

TensorFlow Hub에서 로드 할 BERT 모델을 선택하고 Fine Tuning할 수 있습니다. 선택할 수 있는 모델이 많지만 지금은 기본 BERT 모델을 이용하여 진행하도록 하겠습니다.

bert_model_name = 'bert_en_uncased_L-12_H-768_A-12' 

map_name_to_handle = {
    'bert_en_uncased_L-12_H-768_A-12':
        'https://tfhub.dev/tensorflow/bert_en_uncased_L-12_H-768_A-12/3'
}

map_model_to_preprocess = {
    'bert_en_uncased_L-12_H-768_A-12':
        'https://tfhub.dev/tensorflow/bert_en_uncased_preprocess/3'
}

tfhub_handle_encoder = map_name_to_handle[bert_model_name]
tfhub_handle_preprocess = map_model_to_preprocess[bert_model_name]

print('BERT model selected           :', tfhub_handle_encoder)
print('Preprocessing model auto-selected:', tfhub_handle_preprocess)

텍스트 전처리

Dataset.map을 사용하여 학습을 위한 입력 파이프 라인의 일부로 전처리를 수행 한 다음 추론을 위해 내보내는 모델에 병합하는 방법입니다. 이렇게하면 TPU 자체에 숫자 입력이 필요하지만 학습 및 추론이 원시 텍스트 입력에서 작동할 수 있는 장점이 있습니다.

 

다중 입력 모델을 구축하는 방법과 BERT에 대한 입력의 시퀀스 길이를 조절하는 방법도 있습니다.

bert_preprocess = hub.load(tfhub_handle_preprocess)
tok = bert_preprocess.tokenize(tf.constant(['Hello TensorFlow!']))
print(tok)

각 전처리 모델은 토큰 목록(.bert_pack_inputs(tensors, seq_length), 위의 tok과 같이)과 시퀀스 길이 인수를 받는 방법을 제공합니다.

text_preprocessed = bert_preprocess.bert_pack_inputs([tok, tok], tf.constant(20))

print('Shape Word Ids : ', text_preprocessed['input_word_ids'].shape)
print('Word Ids       : ', text_preprocessed['input_word_ids'][0, :16])
print('Shape Mask     : ', text_preprocessed['input_mask'].shape)
print('Input Mask     : ', text_preprocessed['input_mask'][0, :16])
print('Shape Type Ids : ', text_preprocessed['input_type_ids'].shape)
print('Type Ids       : ', text_preprocessed['input_type_ids'][0, :16])

다음으로 주의 해야할 점 입니다.

 

• input_mask 마스크를 사용하면 모델이 Context와 Padding을 명확하게 구분할 수 있습니다. (1이 Context를 의미)
• input_type_ids와 모양이 input_mask 같지만 Padding되지 않은 영역 내부에는 토큰이 속한 문장을 나타내는 0또는 1이 포함됩니다. (첫 번째 0은 첫번째 문장, 두 번째 1은 두번째 문장, 세 번째 0은 Padding을 의미)

다음은 모든 로직들을 캡슐화하는 전처리 모델을 생성합니다.

 

모델은 문자열을 입력으로 받아 BERT로 전달할 수 있는 적절한 형식의 개체를 반환합니다.

def make_bert_preprocess_model(sentence_features, seq_length=128):
  """Returns Model mapping string features to BERT inputs.

  Args:
    sentence_features: a list with the names of string-valued features.
    seq_length: an integer that defines the sequence length of BERT inputs.

  Returns:
    A Keras Model that can be called on a list or dict of string Tensors
    (with the order or names, resp., given by sentence_features) and
    returns a dict of tensors for input to BERT.
  """

  input_segments = [
      tf.keras.layers.Input(shape=(), dtype=tf.string, name=ft)
      for ft in sentence_features]

  # Tokenize the text to word pieces.
  bert_preprocess = hub.load(tfhub_handle_preprocess)
  tokenizer = hub.KerasLayer(bert_preprocess.tokenize, name='tokenizer')
  segments = [tokenizer(s) for s in input_segments]

  # Optional: Trim segments in a smart way to fit seq_length.
  # Simple cases (like this example) can skip this step and let
  # the next step apply a default truncation to approximately equal lengths.
  truncated_segments = segments

  # Pack inputs. The details (start/end token ids, dict of output tensors)
  # are model-dependent, so this gets loaded from the SavedModel.
  packer = hub.KerasLayer(bert_preprocess.bert_pack_inputs,
                          arguments=dict(seq_length=seq_length),
                          name='packer')
  model_inputs = packer(truncated_segments)
  return tf.keras.Model(input_segments, model_inputs)

 

전처리 모델을 시연해보면, 두개의 문장 입력 (Input1 및 input2)으로 테스트를 생성합니다.

test_preprocess_model = make_bert_preprocess_model(['my_input1', 'my_input2'])
test_text = [np.array(['some random test sentence']),
             np.array(['another sentence'])]
text_preprocessed = test_preprocess_model(test_text)

print('Keys           : ', list(text_preprocessed.keys()))
print('Shape Word Ids : ', text_preprocessed['input_word_ids'].shape)
print('Word Ids       : ', text_preprocessed['input_word_ids'][0, :16])
print('Shape Mask     : ', text_preprocessed['input_mask'].shape)
print('Input Mask     : ', text_preprocessed['input_mask'][0, :16])
print('Shape Type Ids : ', text_preprocessed['input_type_ids'].shape)
print('Type Ids       : ', text_preprocessed['input_type_ids'][0, :16])

방금 정의한 두 입력에 주의하면서 모델의 구조를 살펴보면 다음과 같이 진행됩니다.

tf.keras.utils.plot_model(test_preprocess_model)

데이터 세트의 모든 입력에 전처리를 적용하려면 데이터 세트의 map 함수를 사용합니다.

AUTOTUNE = tf.data.AUTOTUNE


def load_dataset_from_tfds(in_memory_ds, info, split, batch_size,
                           bert_preprocess_model):
  is_training = split.startswith('train')
  dataset = tf.data.Dataset.from_tensor_slices(in_memory_ds[split])
  num_examples = info.splits[split].num_examples

  if is_training:
    dataset = dataset.shuffle(num_examples)
    dataset = dataset.repeat()
  dataset = dataset.batch(batch_size)
  dataset = dataset.map(lambda ex: (bert_preprocess_model(ex), ex['label']))
  dataset = dataset.cache().prefetch(buffer_size=AUTOTUNE)
  return dataset, num_examples

 

모델 정의

 

BERT Encoder를 통해 전처리 된 입력을 공급하고 linear Classifier를 맨 위에 배치하고 정규화를 위해 드롭 아웃을 사용, 문장 또는 문장 쌍 분류를 위한 모델을 정의합니다.

 

모델은 Keras 기능 API를 사용하여 정의됩니다.

def build_classifier_model(num_classes):
  inputs = dict(
      input_word_ids=tf.keras.layers.Input(shape=(None,), dtype=tf.int32),
      input_mask=tf.keras.layers.Input(shape=(None,), dtype=tf.int32),
      input_type_ids=tf.keras.layers.Input(shape=(None,), dtype=tf.int32),
  )

  encoder = hub.KerasLayer(tfhub_handle_encoder, trainable=True, name='encoder')
  net = encoder(inputs)['pooled_output']
  net = tf.keras.layers.Dropout(rate=0.1)(net)
  net = tf.keras.layers.Dense(num_classes, activation=None, name='classifier')(net)
  return tf.keras.Model(inputs, net, name='prediction')

일부 전처리 된 입력에서 모델을 실행해 보면 다음과 같습니다.

test_classifier_model = build_classifier_model(2)
bert_raw_result = test_classifier_model(text_preprocessed)
print(tf.sigmoid(bert_raw_result))

모델의 구조는 다음과 같습니다. 3개의 입력을 BERT 모델에 적용할 수 있습니다.

tf.keras.utils.plot_model(test_classifier_model)

 

 

GLUE 작업 선택

 

GLUE 벤치 마크의 Tensorflow dataset을 사용합니다.

 

Colab을 사용하면 이러한 작은 데이터 세트를 로컬 파일 시스템에 다운로드 할 수 있으며, 아래 코드는 별도의 TPU 작업자 호스트가 colab 런타임의 로컬 파일 시스템에 액세스 할 수 없기 때문에 이를 완전히 메모리로 읽어들입니다.

 

더 큰 데이터 세트의 경우에는 자체 GCP의 Storage에 버킷을 만들고 TPU가 거기에서 데이터를 읽도록 해야합니다.

 

CoLA 데이터 세트(단일 문장 용) MRPC(다중 문장 용)로 데이터 크기가 작고 Fine Tuning에 오래 걸리지 않기 때문에 시작하기 좋습니다.

 

tfds_name = 'glue/cola' 

tfds_info = tfds.builder(tfds_name).info

sentence_features = list(tfds_info.features.keys())
sentence_features.remove('idx')
sentence_features.remove('label')

available_splits = list(tfds_info.splits.keys())
train_split = 'train'
validation_split = 'validation'
test_split = 'test'
if tfds_name == 'glue/mnli':
  validation_split = 'validation_matched'
  test_split = 'test_matched'

num_classes = tfds_info.features['label'].num_classes
num_examples = tfds_info.splits.total_num_examples

print(f'Using {tfds_name} from TFDS')
print(f'This dataset has {num_examples} examples')
print(f'Number of classes: {num_classes}')
print(f'Features {sentence_features}')
print(f'Splits {available_splits}')

with tf.device('/job:localhost'):
  # batch_size=-1 is a way to load the dataset into memory
  in_memory_ds = tfds.load(tfds_name, batch_size=-1, shuffle_files=True)

# The code below is just to show some samples from the selected dataset
print(f'Here are some sample rows from {tfds_name} dataset')
sample_dataset = tf.data.Dataset.from_tensor_slices(in_memory_ds[train_split])

labels_names = tfds_info.features['label'].names
print(labels_names)
print()

sample_i = 1
for sample_row in sample_dataset.take(5):
  samples = [sample_row[feature] for feature in sentence_features]
  print(f'sample row {sample_i}')
  for sample in samples:
    print(sample.numpy())
  sample_label = sample_row['label']

  print(f'label: {sample_label} ({labels_names[sample_label]})')
  print()
  sample_i += 1

데이터 세트는 문제 유형과 훈련에 적합한 손실 함수를 결정합니다.

 

def get_configuration(glue_task):

  loss = tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True)

  if glue_task == 'glue/cola':
    metrics = tfa.metrics.MatthewsCorrelationCoefficient(num_classes=2)
  else:
    metrics = tf.keras.metrics.SparseCategoricalAccuracy(
        'accuracy', dtype=tf.float32)

  return metrics, loss

 

모델 훈련

 

마지막으로 선택한 데이터 세트에서 모델을 학습할 수 있습니다.

 

여러 TPU 기기가 있는 TPU 작업자에 Colab 런타임을 연결한 상단의 설정 코드를 이용해서 TPU 배포 전략 범위 내에서 기본 Keras 모델을 만들고 컴파일 합니다. 반면 전처리는 TPU가 아닌 작업자의 호스트의 CPU에서 실행되므로 전처리를 위한 Keras 모델과 함께 매핑 된 학습 및 유효성 검사 데이터 세트가 빌드됩니다. Model.fit()은 전달된 데이터 세트를 모델 복제본 전달합니다.

 

옵티 마이저

 

Fine Tuning은 BERT 사전 훈련에서 최적화 프로그램 설정을 따릅니다. 초기 학습 속도의 선형 감쇠와 함께 AdamW 최적화 프로그램을 사용하며, 첫 번째 단계에 대한 워밍업 단계 등등, BERT 논문에 따르면 Fine Tuning을 위한 초기 학습률은 더 작습니다. 

 

epochs = 3
batch_size = 32
init_lr = 2e-5

print(f'Fine tuning {tfhub_handle_encoder} model')
bert_preprocess_model = make_bert_preprocess_model(sentence_features)

with strategy.scope():

  # metric have to be created inside the strategy scope
  metrics, loss = get_configuration(tfds_name)

  train_dataset, train_data_size = load_dataset_from_tfds(
      in_memory_ds, tfds_info, train_split, batch_size, bert_preprocess_model)
  steps_per_epoch = train_data_size // batch_size
  num_train_steps = steps_per_epoch * epochs
  num_warmup_steps = num_train_steps // 10

  validation_dataset, validation_data_size = load_dataset_from_tfds(
      in_memory_ds, tfds_info, validation_split, batch_size,
      bert_preprocess_model)
  validation_steps = validation_data_size // batch_size

  classifier_model = build_classifier_model(num_classes)

  optimizer = optimization.create_optimizer(
      init_lr=init_lr,
      num_train_steps=num_train_steps,
      num_warmup_steps=num_warmup_steps,
      optimizer_type='adamw')

  classifier_model.compile(optimizer=optimizer, loss=loss, metrics=[metrics])

  classifier_model.fit(
      x=train_dataset,
      validation_data=validation_dataset,
      steps_per_epoch=steps_per_epoch,
      epochs=epochs,
      validation_steps=validation_steps)

 

Export

 

방금 생성한 전처리  과정과 Fine Tuning된 BERT가 포함된 최종 모델을 생성합니다.

 

모델을 Colab에 저장하고 나중에 다운로드하여 나중에 보관할 수 있습니다.

 

main_save_path = './my_models'
bert_type = tfhub_handle_encoder.split('/')[-2]
saved_model_name = f'{tfds_name.replace("/", "_")}_{bert_type}'

saved_model_path = os.path.join(main_save_path, saved_model_name)

preprocess_inputs = bert_preprocess_model.inputs
bert_encoder_inputs = bert_preprocess_model(preprocess_inputs)
bert_outputs = classifier_model(bert_encoder_inputs)
model_for_export = tf.keras.Model(preprocess_inputs, bert_outputs)

print('Saving', saved_model_path)

# Save everything on the Colab host (even the variables from TPU memory)
save_options = tf.saved_model.SaveOptions(experimental_io_device='/job:localhost')
model_for_export.save(saved_model_path, include_optimizer=False,
                      options=save_options)

모델 테스트

 

모델 결과를 테스트하고 비교를 하기 위해 모델을 다시 로드하고 데이터 세트에서 분할 된 테스트의 일부 입력을 사용해서 테스트를 해봅니다.

 

테스트는 연결된 TPU가 아닌 Colab 호스트에서 수행될 때는 다음과 같이 표시할 수 있습니다.

Saved_model을 다른 곳에서 로드 할 때 생략할 수 있습니다.

with tf.device('/job:localhost'):
  reloaded_model = tf.saved_model.load(saved_model_path)

다음은 데이터에 따른 적용방법입니다.

def prepare(record):
  model_inputs = [[record[ft]] for ft in sentence_features]
  return model_inputs


def prepare_serving(record):
  model_inputs = {ft: record[ft] for ft in sentence_features}
  return model_inputs


def print_bert_results(test, bert_result, dataset_name):

  bert_result_class = tf.argmax(bert_result, axis=1)[0]

  if dataset_name == 'glue/cola':
    print('sentence:', test[0].numpy())
    if bert_result_class == 1:
      print('This sentence is acceptable')
    else:
      print('This sentence is unacceptable')

  elif dataset_name == 'glue/sst2':
    print('sentence:', test[0])
    if bert_result_class == 1:
      print('This sentence has POSITIVE sentiment')
    else:
      print('This sentence has NEGATIVE sentiment')

  elif dataset_name == 'glue/mrpc':
    print('sentence1:', test[0])
    print('sentence2:', test[1])
    if bert_result_class == 1:
      print('Are a paraphrase')
    else:
      print('Are NOT a paraphrase')

  elif dataset_name == 'glue/qqp':
    print('question1:', test[0])
    print('question2:', test[1])
    if bert_result_class == 1:
      print('Questions are similar')
    else:
      print('Questions are NOT similar')

  elif dataset_name == 'glue/mnli':
    print('premise   :', test[0])
    print('hypothesis:', test[1])
    if bert_result_class == 1:
      print('This premise is NEUTRAL to the hypothesis')
    elif bert_result_class == 2:
      print('This premise CONTRADICTS the hypothesis')
    else:
      print('This premise ENTAILS the hypothesis')

  elif dataset_name == 'glue/qnli':
    print('question:', test[0])
    print('sentence:', test[1])
    if bert_result_class == 1:
      print('The question is NOT answerable by the sentence')
    else:
      print('The question is answerable by the sentence')

  elif dataset_name == 'glue/rte':
    print('sentence1:', test[0])
    print('sentence2:', test[1])
    if bert_result_class == 1:
      print('Sentence1 DOES NOT entails sentence2')
    else:
      print('Sentence1 entails sentence2')

  elif dataset_name == 'glue/wnli':
    print('sentence1:', test[0])
    print('sentence2:', test[1])
    if bert_result_class == 1:
      print('Sentence1 DOES NOT entails sentence2')
    else:
      print('Sentence1 entails sentence2')

  print('BERT raw results:', bert_result[0])
  print()

해당 Task에 대한 테스트 결과입니다.

with tf.device('/job:localhost'):
  test_dataset = tf.data.Dataset.from_tensor_slices(in_memory_ds[test_split])
  for test_row in test_dataset.shuffle(1000).map(prepare).take(5):
    if len(sentence_features) == 1:
      result = reloaded_model(test_row[0])
    else:
      result = reloaded_model(list(test_row))

    print_bert_results(test_row, result, tfds_name)

 

TF Serving에서 모델을 사용하려면 미리 선언된 것들 중 하나를 통해 저장된 모델을 호출합니다. 입력에 약간의 차이가 있으며 다음과 같이 테스트할 수 있습니다.

with tf.device('/job:localhost'):
  serving_model = reloaded_model.signatures['serving_default']
  for test_row in test_dataset.shuffle(1000).map(prepare_serving).take(5):
    result = serving_model(**test_row)
    # The 'prediction' key is the classifier's defined model name.
    print_bert_results(list(test_row.values()), result['prediction'], tfds_name)

 

참고.

inmoonlight.github.io/2019/12/22/GLUE-benchmark/

General Language Understanding Evaluation (GLUE) benchmark