3.3.19 [LLM] 파인튜닝 기법: DPO

DPO (Direct Preference Optimization)

 

RLHF의 한계를 해결하기 위해 등장한 방법으로, 보상 모델 없이 선호 데이터를 직접 최적화하는 기법이다.

RLHF보다 더 간단하고 효율적으로 모델을 미세 조정할 수 있다.

- RLHF는 보상 모델을 학습 후 정책을 강화 학습(PPO)으로 최적화하지만, DPO는 보상 모델을 생략하고 직접 선호 데이터를 활용하여 모델을 업데이트한다.

 

• RLHF와 DPO 비교

구분	RLHF	DPO
학습 방식	보상 모델을 학습한 후 강화 학습 적용	선호 데이터(Preference Data)만을 사용해 직접 최적화
보상 모델 필요 여부	필요 (Human Feedback → Reward Model)	불필요 (선호 데이터를 바로 최적화)
학습 과정 복잡성	PPO 알고리즘 사용, 보상 모델 평가 필요	간단한 최적화 과정, 수식 기반 조정
장점	효과적인 강화 학습 가능	학습이 더 간결하고 안정적
단점	학습 비용이 크고 튜닝이 어려움	아직 실험적 단계이며 적용 사례 부족

 

• DPO 학습 과정

1단계: 선호 데이터(Preference Data) 수집

2단계: 선호 데이터 기반으로 직접 모델 최적화

3단계: 보상 모델 없이 최적화된 모델 평가

 

• DPO를 활용한 모델 개선

DPO가 적용되는 영역

- 대화형 AI(ChatGPT 등)에서 자연스러운 응답 개선

- 사용자 피드백을 반영하여 모델 성능 조정

- RLHF 대비 학습 비용 절감이 필요한 경우

 

성능 평가 방법

- RLHF 기반 모델과 비교하여 응답 품질 평가

- 사람 피드백을 기반으로 선호 모델 성능 분석

 

DPO 적용 시 고려사항

- RLHF보다 간단하지만, 최적의 데이터셋이 필요

- RHLF 대비 보정 효과가 충분한지 검토 필요

 

 

실습 (DPO)

 

• 0. 환경 설정

!python -m pip install --upgrade pip
!pip install typing_extensions==4.7.1 --upgrade
!pip install transformers peft datasets bitsandbytes accelerate

from huggingface_hub import login
import os
import torch

login(token="hf_xxx")

os.environ['WANDB_DISABLED'] = 'true'       # 학습과정에 대한 로깅 남기지 않음
os.environ['PYTORCH_CUDA_ALLOC_CONF'] = 'expandable_segments:True'

device = torch.device("cuda" if torch.cuda.is_avaliable() else "cpu")

 

• 1. 모델 로드

model_name="Bllossom/llama-3.2-Korean-Bllossom-3B"

from transformers import BitsAndBytesConfig

bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_use_double_quant=True,
    bnb_4bit_compute_dtype=torch.bfloat16
)

from transformers import AutoTokenizer, AutoModelForCausalLM

model = AutoModelForCausalLM.from_pretrained(
    model_name,
    quantization_config=bnb_config,
    torch_dtype=torch.bfloat16,
    device_map="auto"
)

tokenizer = AutoTokenizer.from_pretrained(model_name)

if tokenizer.pad_token is None:
    tokenizer.pad_token = tokenizer.eos_token

 

• 2. 학습 준비

(1) 모델 준비

from peft import LoraConfig

lora_config = LoraConfig(
    r=8,    # 저랭크 행렬 크기
    lora_alpha=32,
    target_modules=['q_proj', 'k_proj', 'v_proj', 'o_proj', 'gate_proj', 'down_proj', 'up_proj'],
    bias='none',
    task_type='CAUSAL_LM'
)

from peft import prepare_model_for_kbit_training, get_peft_model

model = prepare_model_for_kbit_training(model)
model = get_peft_model(model, lora_config)	# LoRA 설정 붙여서 학습 가능한 모델로

model.print_trainable_parameters()
model.train()
model.gradient_checkpointing_enable()	# 메모리 절약하면서 학습

 

(2) 데이터 준비

from datasets import load_dataset

dataset = load_dataset('mncai/orca_dpo_pairs_ko')

# 데이터 전처리 함수 정의
def preprocess_text(sample):
    input_enc = tokenizer(sample['question'], padding='max_length', max_length=256, truncation=True)
    preferred_enc = tokenizer(sample['chosen'], padding='max_length', max_length=256, truncation=True)
    despreferred_enc = tokenizer(sample['rejected'], padding='max_length', max_length=256, truncation=True)

    return {
        "input_ids": input_enc['input_ids'],
        "attention_mask": input_enc['attention_mask'],
        "preferred_ids": preferred_enc['input_ids'],
        "despreferred_ids": despreferred_enc['input_ids']
    }
    
tokenized_dataset = dataset['train'].map(
    preprocess_text,
    remove_columns=['id', 'system', 'question', 'chosen', 'rejected']
)

tokenized_dataset.set_format(type='torch', 
                             columns=['input_ids', 'attention_mask', 'preferred_ids', 'despreferred_ids'])

def collate_fn(batch):
    input_ids = torch.stack([item['input_ids'].clone().detach() for item in batch])		# 안정적으로 사용하기 위해 clone(), detach() 사용
    attention_mask = torch.stach([item['attention_mask'].clone().detach() for item in batch])

    max_length = max(max(len(item['prefereed_ids']) for item in batch), 1)

    preferred_ids = torch.stack([       # 하나씩 꺼내서 패딩 처리
        torch.tensor(
            item['preferred_ids'].tolist() + 
                    [tokenizer.pad_token_id] * (max_length - len(item['preferred_ids'])),
            dtype=torch.long
        ) if isinstance(item['preferred_ids'], torch.Tensor) else
        torch.tensor(
            item['preferred_ids'] + 
                    [tokenizer.pad_token_id] * (max_length - len(item['preferred_ids'])),
            dtype=torch.long
        )
        for item in batch
    ]).clone().detach()

    despreferred_ids = torch.stack([
        torch.tensor(
            item['despreferred_ids'].tolist() + 
                    [tokenizer.pad_token_id] * (max_length - len(item['despreferred_ids'])),
            dtype=torch.long
        ) if isinstance(item['despreferred_ids'], torch.Tensor) else
        torch.tensor(
            item['despreferred_ids'] + 
                    [tokenizer.pad_token_id] * (max_length - len(item['despreferred_ids'])),
            dtype=torch.long
        )
        for item in batch
    ]).clone().detach()

    return {
        "input_ids": input_ids,
        "attention_mask": attention_mask,
        "preferred_ids": preferred_ids,
        "despreferred_ids": despreferred_ids
    }

 

(3) Trainer 준비 및 train()

from transformers import Trainer
import torch.nn.functional as F

class DTOTrainer(Trainer):
    def compute_loss(self, model, inputs, beta=0.1, *args, **kwargs):
        input_ids = inputs['input_ids'].to(model.device)
        attention_mask = inputs['attention_mask'].to(model.device)
        preferred_ids = inputs['preferred_ids'].to(model.device)
        despreferred_ids = inputs['despreferred_ids'].to(model.device)

		# 로드한 데이터를 모델에 통과
        preferred_ooutputs = model(input_ids=input_ids, attention_mask=attention_mask, labels=preferred_ids)
        despreferred_ooutputs = model(input_ids=input_ids, attention_mask=attention_mask, labels=despreferred_ids)

        preferred_loss = preferred_ooutputs.loss
        despreferred_loss = despreferred_ooutputs.loss

        loss = F.logsigmoid(beta * (despreferred_loss - preferred_loss)).mean()
        return loss

from transformers import TrainingArguments

training_args = TrainingArguments(
    output_dir='./dpo_llama3_korean',
    per_device_train_batch_size=1,
    gradient_accumulation_steps=16,
    learning_rate=1e-4,
    num_train_epochs=3,
    save_total_limit=2,
    save_strategy="steps",
    save_steps=200,
    logging_steps=50,
    remove_unused_columns=False,
    fp16=True,
    optim="adamw_bnb_8bit",
    max_grad_norm=0
)

trainer = DTOTrainer(
    model=model,
    args=training_args,
    train_dataset=tokenized_dataset,
    data_collator=collate_fn
)

 

• 3. 학습

trainer.train()

 

• 4. 학습된 모델 응답 생성

from peft import PeftModel

checkpoint_path = './dpo_llama3_korean/checkpoint-xxx'

model = PeftModel.from_pretrained(model, checkpoint_path)
model.eval()

sample_data = dataset['train'].select(range(5))

def generate_response(question):
    inputs = tokenizer(question, return_tensors='pt', padding=True,
                       truncation=True, max_length=256).to(model.device)
    
    with torch.no_grad():
        output_ids = model.generate(
            input_ids=inputs['input_ids'],
            attention_mask=inputs['attention_mask'],
            max_length=256,
            temperature=0.7,
            top_p=0.9,
            do_sample=True
        )
    
    return tokenizer.decode(output_ids[0], skip_special_tokens=True)	# 이해할 수 있는 한글로 반환

for i, example in enumerate(sample_data):

    question = example['question']
    preferred_answer = example['chosen']

    generatted_response = generate_response(question)

    print(f"{i}번째 질문: {question}")
    print(f"정답 (선호 응답): {preferred_answer}")
    print(f"실제 모델 응답: {generate_response}")
    print("=" * 100)