[언어지능 딥러닝] 6차 미프 (2) - Gemma-2B finetuning

!pip install -q -U transformers==4.38.2
!pip install -q -U datasets==2.18.0
!pip install -q -U bitsandbytes==0.42.0
!pip install -q -U peft==0.9.0
!pip install -q -U trl==0.7.11
!pip install -q -U accelerate==0.27.2

 

 

import os
import pandas as pd
from tqdm.auto import tqdm

import torch
from datasets import Dataset
from transformers import (AutoTokenizer,
                          AutoModelForCausalLM,
                          BitsAndBytesConfig,
                          pipeline,
                          TrainingArguments)
from peft import (LoraConfig,
                  PeftModel)
from trl import SFTTrainer

 

hugging face login은 따로 해야 한다.

from huggingface_hub import notebook_login
notebook_login()

 

 

이 이후에 런타임 -> 세션 다시 시작을 한번 해줘야 한다.

 

hugginface transformer에서 불러올 gemma 모델의 id를 적어넣어준다.

WORKSPACE = '/content/drive/MyDrive/nlp-project'
MODEL_ID = 'google/gemma-1.1-2b-it'

 

4bit 양자화를 하면 2.2gb vram만 필요하다.

그냥 로딩하면 무려 9gb가 필요하다.

# declare 4 bits quantize
quantization_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.float16
)
# load 4 bits model
model = AutoModelForCausalLM.from_pretrained(MODEL_ID,
                                             device_map='auto', #gpu 모드로 읽어오기
                                             quantization_config=quantization_config,
                                             token=HF_TOKEN) #허깅페이스의 토큰 ID
# load tokenizer
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID,
                                          add_special_tokens=True,
                                          token=HF_TOKEN)
tokenizer.padding_side = 'right'

 

multi gpu로 돌리려면 다음 양식을 참조

# pip install accelerate
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch

tokenizer = AutoTokenizer.from_pretrained("google/gemma-1.1-2b-it")
model = AutoModelForCausalLM.from_pretrained(
    "google/gemma-1.1-2b-it",
    device_map="auto",
    torch_dtype=torch.bfloat16
)

input_text = "Write me a poem about Machine Learning."
input_ids = tokenizer(input_text, return_tensors="pt").to("cuda")

outputs = model.generate(**input_ids)
print(tokenizer.decode(outputs[0]))

 

 

piepeline 라이브러리로 pipe라인을 만든다.

pipe = pipeline("text-generation",
                model=model,
                tokenizer=tokenizer,
                max_new_tokens=512)
pipe

 

프롬프트는 다음과 같은 형식을 띄고 있어야 한다.

<bos><start_of_turn>user
Write a hello world program<end_of_turn>
<start_of_turn>model

 

예시로 하자면,

doc = """엄청나게 즐거운 시간이었습니다. 강추!!!"""
messages = [
    {
        "role": "user",
        "content": "다음 문장은 영화리뷰입니다. 긍정 또는 부정으로 분류해주세요:\n\n{}".format(doc)
    }
]
prompt = pipe.tokenizer.apply_chat_template(messages,
                                            tokenize=False,
                                            add_generation_prompt=True)
                                            
outputs = pipe(
    prompt,
    do_sample=True,
    temperature=0.2, #broad(범위)
    top_k=50, #
    top_p=0.95, #확률이 낮은 것들은 뺀다. 95% 범위 안에서
    add_special_tokens=True
)
outputs

 

이는 chatgpt의 api 사용법과 비슷하다.

 

print(outputs[0]["generated_text"])
><bos><start_of_turn>user
다음 문장은 영화리뷰입니다. 긍정 또는 부정으로 분류해주세요:

엄청나게 즐거운 시간이었습니다. 강추!!!<end_of_turn>
<start_of_turn>model
긍정으로 분류합니다.

영화는 긍정적인 감성과 장면을 제공합니다.

 

답만 보기 위해서 마지막 프롬프트의 길이만 빼준다.

print(outputs[0]["generated_text"][len(prompt):])
>긍정으로 분류합니다.
영화는 긍정적인 감성과 장면을 제공합니다.

 

Gemma with LoRA

gemma 모델을 불러오고, LoRA를 만든다.

# lora config
lora_config = LoraConfig(
    r=6,
    lora_alpha = 8,
    lora_dropout = 0.05,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"],
    task_type="CAUSAL_LM",
)

 

 

긍정 또는 부정으로만 내놓도록 프롬프트를 만든다.

# 학습을 위한 prompt를 생성합니다.
def gen_train_prompt(example):
    prompt_list = []
    for i in range(len(example['document'])):
        doc = example['document'][i]
        label = '긍정' if example['label'][i] == 1 else '부정'
        prompt_list.append(r"""<bos><start_of_turn>user
다음 문장은 영화리뷰입니다. 긍정 또는 부정으로 분류해주세요:

{}<end_of_turn>
<start_of_turn>model
{}<end_of_turn><eos>""".format(doc, label))
    return prompt_list

 

이 다음으로는 hugging face에서 사용하기 위한 dataset을 만든다.

# huggingface dataset을 생성하는 함수입니다.
def make_dataset(df, sample=-1):
    df = df[['document', 'label']]
    if sample > 0:
        df = df.sample(sample)
    dataset = Dataset.from_pandas(df)
    return dataset
    
data = {
    'document': [
        '영화 강추 합니다.',
        '시간이 너무 아깝습니다.'
    ],
    'label': [
        1,
        0
    ]
}
df_train = pd.DataFrame(data)
df_train

train_dataset = make_dataset(df_train)
train_dataset

 

이 다음에는 trainer를 정의한다.

# trainer 정의
trainer = SFTTrainer(
    model=model, # 학습할 모델
    train_dataset=train_dataset,  # 학습할 데이터 셋
    max_seq_length=256,  # 최대 토큰 갯수
    args=TrainingArguments(
        output_dir=os.path.join(WORKSPACE, "nsmc-tutorial"),
        # num_train_epochs = 1,  # epoc으로 할 경우 너무 많이 걸리 수 있음
        max_steps=10,  # 학습 step 수 원래는 1000~2000
        per_device_train_batch_size=2,  # gpu당 입력 batch_size
        gradient_accumulation_steps=4,  # gradient 누적 후 학습. 4번을 모아서 학습(많은 배치를 위해)
        optim="paged_adamw_8bit",  # optimizer (QLoRA) optimizer를 쓸 때, 필요하면 cpu로 넘긴다.
        warmup_steps=1000,  # learning rate warmup step 1000번째 step이 가장 높아진다.
        learning_rate=1e-4,  # learning rate
        # bf16=True,  # bf16 사용 여부 (3090 이상에서 가능)
        fp16=True,  # fp16 사용 여부 (예전 GPU에서 사용 가능, T4)
        logging_steps=10,  # 얼마만에 한번 씩 중간 결과를 확인할 것인가?
        push_to_hub=False,  # huggingface에 올릴 수 있음
        report_to=None,  # W&B에 학습결과 공유 가능
    ),
    peft_config=lora_config,  # QLoRA config
    formatting_func=gen_train_prompt,  # 프롬프트 생성 함수
)

 

이후에 trainer.train()을 한다.

train에서 loss는 각 모델마다 지정된 loss를 통해 계산하고, 기본적으로 GPT와 같이 다음 문장에 대한 예측을 바탕으로 loss를 줄여나간다.

 

Lora의 학습된 weight을 저장

학습된 양이 매우 작다.

 

type(trainer.model)

>peft.peft_model.PeftModelForCausalLM
def _wrapped_call_impl(*args, **kwargs)

# save lora (delta weight)
trainer.model.save_pretrained("lora_adapter")

# 저장된 결과 확인
!ls -lh ./lora_adapter

> total 29M

 

필요한 부분만 학습되어 있다. 델타 W이다.

 

Original 모델에 학습된 모델을 더해서 저장

# original model load (before finetuned)
model = AutoModelForCausalLM.from_pretrained(MODEL_ID,
                                             device_map='auto',
                                             torch_dtype=torch.float16,
                                             token=HF_TOKEN)
                                             
# merge : original + delta wieght
model = PeftModel.from_pretrained(model,
                                  "lora_adapter",
                                  device_map='auto',
                                  torch_dtype=torch.float16,
                                  token=HF_TOKEN)
model = model.merge_and_unload()


# save fine-tunned model
model.save_pretrained(os.path.join("nsmc-tutorial", "checkpoint-final"))

# 저장 결과 확인
!ls -lh nsmc-tutorial/checkpoint-final
>total 4.7G

 

학습 결과 평가

# 평가 데이터
data = {
    'document': [
        '영화 재밌어요',
        '졸려서 눈물이 났어요.'
    ],
    'label': [
        1,
        0
    ]
}
df_test = pd.DataFrame(data)
df_test

# 평가 데이터셋 생성
test_dataset = make_dataset(df_test)
test_dataset

# 평가를 위한 prompt
def gen_test_prompt(example):
    prompt_list = []
    for i in range(len(example['document'])):
        doc = example['document'][i]
        prompt_list.append(r"""<bos><start_of_turn>user
다음 문장은 영화리뷰입니다. 긍정 또는 부정으로 분류해주세요:

{}<end_of_turn>
<start_of_turn>model
""".format(doc)) #답변이 없게 만들었다.
    return prompt_list
    
# prompt 확인
prompt = gen_test_prompt(test_dataset[1:])[0]
print(prompt)

# pipeline 정의
pipe = pipeline("text-generation",
                model=model,
                tokenizer=tokenizer,
                max_new_tokens=10)
                
# infer
total_sample_cnt, total_correct_cnt = 0, 0
for example in test_dataset.iter(1):
    label = '긍정' if example['label'][0] == 1 else '부정'

    prompt = gen_test_prompt(example)
    outputs = pipe(
        prompt,
        do_sample=True,
        temperature=0.2,
        top_k=50,
        top_p=0.95,
        add_special_tokens=True
    )
    pred = outputs[0][0]['generated_text'][len(prompt[0]):]
    total_sample_cnt += 1
    total_correct_cnt += 1 if label == pred else 0

    print(example['document'][0], ":", pred)
    print('-' * 20)
print(f"Test Accuracy: {total_correct_cnt} / {total_sample_cnt} = {total_correct_cnt/total_sample_cnt:.4f}")

 

 

이제 본격적으로 train한다.

 

데이터를 다운로드

https://github.com/e9t/nsmc

# 데이터를 다운로드할 폴더를 생성합니다.
os.makedirs(os.path.join(WORKSPACE, "data", "nsmc"), exist_ok=True)

!wget https://github.com/e9t/nsmc/raw/master/ratings_train.txt \
    -O {WORKSPACE}/data/nsmc/train.tsv
    
!wget https://github.com/e9t/nsmc/raw/master/ratings_test.txt \
    -O {WORKSPACE}/data/nsmc/test.tsv
    
!ls {WORKSPACE}/data/nsmc

df_train = pd.read_csv(os.path.join(WORKSPACE, "data", "nsmc", "train.tsv"),
                       sep='\t')
df_train

df_test = pd.read_csv(os.path.join(WORKSPACE, "data", "nsmc", "test.tsv"),
                      sep='\t')
df_test

 

이렇게 df_train과 df_test를 정의했으면, 다음 순서를 따르면 된다.

 

1. gemma를 로딩

2. LoRA 적용을 위한 설정 정의

3. train dataset 생성

4. prompt 생성 함수 정의

5. trainer를 이용한 학습

6. 학습된 weight 저장

7. original 모델에 합해서 최종 모델 저장

8. 재시작 후 평가 및 추론

 

모든 코드를 모아보면 다음과 같다.

# declare 4 bits quantize
quantization_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.float16
)
# load 4 bits model
model = AutoModelForCausalLM.from_pretrained(MODEL_ID,
                                             device_map='auto',
                                             quantization_config=quantization_config,
                                             token=HF_TOKEN)
# load tokenizer
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID,
                                          add_special_tokens=True,
                                          token=HF_TOKEN)
tokenizer.padding_side = 'right'

# lora config
lora_config = LoraConfig(
    r=6,
    lora_alpha = 8,
    lora_dropout = 0.05,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"],
    task_type="CAUSAL_LM",
)

# huggingface dataset을 생성하는 함수입니다.
def make_dataset(df, sample=-1):
    df = df[['document', 'label']]
    if sample > 0:
        df = df.sample(sample)
    dataset = Dataset.from_pandas(df)
    return dataset
    
    
# 학습을 위한 prompt를 생성합니다.
def gen_train_prompt(example):
    prompt_list = []
    for i in range(len(example['document'])):
        doc = example['document'][i]
        label = '긍정' if example['label'][i] == 1 else '부정'
        prompt_list.append(r"""<bos><start_of_turn>user
다음 문장은 영화리뷰입니다. 긍정 또는 부정으로 분류해주세요:

{}<end_of_turn>
<start_of_turn>model
{}<end_of_turn><eos>""".format(doc, label))
    return prompt_list
    
# trainer 정의
trainer = SFTTrainer(
    model=model, # 학습할 모델
    train_dataset=train_dataset,  # 학습할 데이터 셋
    max_seq_length=256,  # 최대 토큰 갯수
    args=TrainingArguments(
        output_dir=os.path.join(WORKSPACE, "nsmc-tutorial"),
        # num_train_epochs = 1,  # epoc으로 할 경우 너무 많이 걸리 수 있음
        max_steps=1000,  # 학습 step 수
        per_device_train_batch_size=2,  # gpu당 입력 batch_size
        gradient_accumulation_steps=4,  # gradient 누적 후 학습
        optim="paged_adamw_8bit",  # optimizer (QLoRA)
        warmup_steps=1000,  # learning rate warmup step
        learning_rate=1e-4,  # learning rate
        # bf16=True,  # bf16 사용 여부 (3090 이상에서 가능)
        fp16=True,  # fp16 사용 여부 (예전 GPU에서 사용 가능, T4)
        logging_steps=100,  # 얼마만에 한번 씩 중간 결과를 확인할 것인가?
        push_to_hub=False,  # huggingface에 올릴 수 있음
        report_to=None,  # W&B에 학습결과 공유 가능
    ),
    peft_config=lora_config,  # QLoRA config
    formatting_func=gen_train_prompt,  # 프롬프트 생성 함수
)

# train
trainer.train()

type(trainer.model)

# save lora (delta weight)
trainer.model.save_pretrained("lora_adapter2")

# 저장된 결과 확인
!ls -lh ./lora_adapter2

 

아래부터는 조심.

평가 함수로 불러올 것이기 때문에, os.path를 잘 설정해 줘야 한다.

 

# original model load (before finetuned)
model = AutoModelForCausalLM.from_pretrained(MODEL_ID,
                                             device_map='auto',
                                             torch_dtype=torch.float16,
                                             token=HF_TOKEN)
                                             
# merge : original + delta wieght
model = PeftModel.from_pretrained(model,
                                  "lora_adapter2",
                                  device_map='auto',
                                  torch_dtype=torch.float16,
                                  token=HF_TOKEN)
model = model.merge_and_unload()

# save fine-tunned model
model.save_pretrained(os.path.join("nsmc-gemma", "checkpoint-final"))

 

 

평가 및 추론

모델을 불러올 때, AutoModelForCausalLM의 Model ID만 우리가 저장한 모델로 불러온다.

df_test = pd.read_csv(os.path.join(WORKSPACE, "data", "nsmc", "test.tsv"),
                      sep='\t')
df_test

# declare 4 bits quantize
quantization_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.float16
)
# load 4 bits model
model = AutoModelForCausalLM.from_pretrained(os.path.join("nsmc-gemma", "checkpoint-final"),
                                             device_map='auto',
                                             quantization_config=quantization_config,
                                             token=HF_TOKEN)
# load tokenizer
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID,
                                          add_special_tokens=True,
                                          token=HF_TOKEN)
tokenizer.padding_side = 'right'


# 평가 데이터셋 생성
test_dataset = make_dataset(df_test)
test_dataset

# 평가를 위한 prompt
def gen_test_prompt(example):
    prompt_list = []
    for i in range(len(example['document'])):
        doc = example['document'][i]
        prompt_list.append(r"""<bos><start_of_turn>user
다음 문장은 영화리뷰입니다. 긍정 또는 부정으로 분류해주세요:

{}<end_of_turn>
<start_of_turn>model
""".format(doc))
    return prompt_list
    
# prompt 확인
prompt = gen_test_prompt(test_dataset[1:])[0]
print(prompt)

# pipeline 정의
pipe = pipeline("text-generation",
                model=model,
                tokenizer=tokenizer,
                max_new_tokens=10)
                
                
# 각 token에 대한 평가
total_sample_cnt, total_correct_cnt = 0, 0
for example in test_dataset.iter(1):
    label = '긍정' if example['label'][0] == 1 else '부정'

    prompt = gen_test_prompt(example)
    outputs = pipe(
        prompt,
        do_sample=True,
        temperature=0.2,
        top_k=50,
        top_p=0.95,
        add_special_tokens=True
    )
    pred = outputs[0][0]['generated_text'][len(prompt[0]):]
    total_sample_cnt += 1
    total_correct_cnt += 1 if label == pred else 0

    print(example['document'][0], ":", pred)
    print('-' * 20)
print(f"Test Accuracy: {total_correct_cnt} / {total_sample_cnt} = {total_correct_cnt/total_sample_cnt:.4f}")



                
def gen_prompt(pipe, doc):
    messages = [
        {
            "role": "user",
            "content": "다음 문장은 영화리뷰입니다. 긍정 또는 부정으로 분류해주세요:\n\n{}".format(doc)
        }
    ]
    prompt = pipe.tokenizer.apply_chat_template(messages,
                                                tokenize=False,
                                                add_generation_prompt=True)
    return prompt
    
    
def gen_response(pipe, doc):
    prompt = gen_prompt(pipe, doc)

    outputs = pipe(
        prompt,
        do_sample=True,
        temperature=0.2,
        top_k=50,
        top_p=0.95,
        add_special_tokens=True
    )
    return outputs[0]["generated_text"][len(prompt):]
    
while True:
    doc = input('문장 > ')
    doc = doc.strip()
    if len(doc) == 0:
        break
    result = gen_response(pipe, doc)
    print(f'감정 > {result}\n\n')

 

최종적으로, 이러한 수순으로 huggingface library 및 LLM fine tuning을 진행할 수 있다.