1️⃣ DPO
⭐ DPO(Direct Preference Optimization) 란
: RLHF의 한계를 해결하기 위해 등장한 방법으로, 리워드 모델 없이 데이터를 직접 최적화하는 기법이다.
: RLHF보다 더 간단하고 효율적으로 모델을 미세 조정할 수 있다.
💁🏻♀️ PPO와 DPO 차이
: RLHF는 보상 모델 학습 후 PPO(강화학습)로 최적화
: DPO는 보상 모델을 생략하고, 직접 선호 데이터를 활용해 모델을 업데이트
2️⃣ RLHF와 DPO 비교
| 구분 | RLHF | DPO |
| 학습 방식 | 보상 모델을 학습한 후 강화 학습(PPO) 적용 | 선호 데이터만을 사용해 직접 최적화 |
| 보상 모델 필요 여부 | 필요 (human feedback -> reward model) | 불필요 (선호 데이터를 바로 최적화) |
| 학습 과정 복잡성 | PPO 알고리즘 사용 | 간단한 최적화 과정, 수식 기반 조정 |
| 장점 | 효과적인 강화 학습이 가능 | 학습이 더 간결하고 안정적 |
| 단점 | 학습 비용이 크고 튜닝이 어려움 | 아직 실험적 단계며 적용 사례가 부족 |
3️⃣ DPO 학습 과정
- 1단계: 선호 데이터 수집
- 2단계: 선호 데이터 기반 직접 모델 최적화
- 3단계: 보상 모델 없이 최적화된 모델 평가
4️⃣ DPO의 장점과 한계
- 장점
- 보상 모델 없이 선호 데이터만으로 최적화가 가능해 학습 과정이 단순화됨
- 안정적이고 빠른 최적화가 가능함
- 한계
- 선호 데이터만으로 최적화하기에 데이터 품질이 성능에 큰 영향을 미침
- 실험적 기법으로 RLHF 대비 검증된 사례가 적음
5️⃣ DPO를 활용한 모델 개선
- DPO가 적용되는 영역
- 대화형 AI에서 자연스러운 응답 개선
- 사용자 피드백을 반영해 모델 성능 조정
- RLHF 대비 학습 비용 절감이 필요한 경우
- 성능 평가 방법
- RLHF 기반 모델과 비교해 응답의 품질을 평가
- 사람 피드백을 기반으로 선호 모델 성능 분석
- DPO 적용 시 고려사항
- RLHF보다 간단하지만, 최적의 데이터셋이 필요
- RLHF 대비 보정 효과가 충분한지 검토 필요
6️⃣ RLHF에 DPO를 적용한 코드
🔩 학습 과정
- 모델 설정
- 4비트 양자화 설정하기 > `BitsAndBytesConfig`
- 모델에 양자화 설정을 적용하기 > `AutoModelForCausalLM`
- 토크나이저 로드 > `AutoTokenizer`
- tokenizer.pad_token is None -> pad_token = eos_token 설정하기 (설정하지 않으면 오류 발생 가능성이 있음)
- LoRA 설정 > `LoraConfig`
- 모델 준비 > `prepare_model_for_kbit_training` & `get_peft_model`
- 데이터셋 로드 > `load_dataset`
- 데이터 전처리 함수 만들기 > `tokenizer` 사용해 `prepare_text` 함수 만들기
- 토크나이저를 사용해 텐서화
- 데이터 토큰화 및 Pytorch 텐서로 형식 변환 > `prepare_text` 함수 사용 & `set_format`
- 데이터 배치 구성 함수 만들기 > `torch.stack` 사용해 데이터 배치 구성을 수행하는 `collate_fn` 함수 만들기
- 토크나이저 반환
- DPO 트레이너 클래스 구성 > `Trainer` 모듈의 `compute_loss 메서드` 오버라이딩 > 선호/비선호 응답에 대한 `DPO 손실 공식`을 이용해 loss 계산
- 훈련 설정: TrainingArguments 만들기 > `TrainingArguments`
- 학습에 필요한 설정 (batch size, epoch 수, 저장 디렉토리 등)
- 훈련 과정 정의: `DPOTrainer`
- 선호/비선호 응답 기반 손실 계산 및 모델 업데이트 로직 담당
- 학습: trainer.train()
- 학습 루프 시작 (args에 따라 배치 구성, 손실 계산, 옵티마이저 적용 수행)
🤖 학습된 모델 사용
- checkpoint path 불러오기 및 eval > `model.eval()` : 추론 모드 전환
- 응답 만들어주는 함수
- `tokenizer`: input 토큰화하기
- `torch.no_grad()`: 추론 시 학습하지 않으므로 grad 연산 제거하기
- `decode`: 숫자를 사람이 읽을 수 있는 문자로 변경해 리턴
- 응답 출력
1. 모델 설정
model_name = 'Bllossom/llama-3.2-Korean-Bllossom-3B'
2. 4비트 양자화 설정 - `BitsAndBytes`
bnb_config = BitsAndBytesConfig(
load_in_4bit=True, # 4비트 양자화 설정 on
bnb_4bit_quant_type='nf4', # 4비트 양자화 방식 설정
bnb_4bit_use_double_quant=True, # 두번 양자화하는거 사용 여부
bnb_4bit_compute_dtype=torch.float16 # 연산 시의 데이터 타입
)
3. 모델에 양자화 적용하기 - `AutoModelForCausalLM`
model = AutoModelForCausalLM.from_pretrained(
model_name,
quantization_config=bnb_config,
device_map='auto'
)
4. 토크나이저 로드 - `AutoTokenizer`
tokenizer = AutoTokenizer.from_pretrained(model_name)
if tokenizer.pad_token is None:
tokenizer.pad_token = tokenizer.eos_token
5. LoRA 설정 - `LoraConfig`
lora_config = LoraConfig(
r=8, # lora 차원 수
lora_alpha=32, # lora 스케일링 개수
target_modules=['q_proj', 'k_proj', 'v_proj', 'o_proj', 'gate_proj', 'down_proj', 'up_proj'], # 각각의 가중치 행렬들이 로라에 적용할 대상 모델이 됨 # q_proj: query: 입력 데이터를 쿼리 벡터로 변환, key: 입력 데이터를 키 벡터로 변환, value: 입력 데이터를 값 벡터로 변환, o_proj: output: 출력값의 원래 차원으로 변환하는 가중치 행렬로, gate: 입력 벡터가 선형 변환을 수행한 과정을 거친 가중치 행렬, down: mlp에서 고차원 벡터를 낮은 차원으로 변환, up: 낮은 차원의 벡터를 원래 벡터로 확장하는 가중치 행렬
bias='none', #
task_type='CAUSAL_LM' # causal_lm: gpt 계열의 모델로 task_type 설정
)
6. 모델 준비 - `prepare_model_for_kbit_training`, `get_peft_model`
# prepare_model_for_kbit_training
# 4비트로 압축한 모델을 안전하고 효율적으로 finetune할 수 있도록 “훈련 가능한 구조”로 바꿔주는 함수
# 함수가 하는 일: 양자화 레이어 유지, 필요한 레이어만 float16, float32로 변환
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() # gpu 메모리 사용량을 줄이기 위해 역전파 과정에서 중간 결과를 저장하지 않고 바로 다시 중간결과를 계산하는 방식
7. 데이터셋 로드
dataset = load_dataset('mncai/orca_dpo_pairs_ko')
8. 데이터 전처리 함수 만들기
# 텍스트 모델이 이해할 수 있도록 토큰화하고 숫자 인덱스로 변환
def preprocess_text(sample):
input_enc=tokenizer(sample['question'], padding='max_length', max_length=256, truncation=True),
preferred_enctokenizer(sample['chosen'], padding='max_length', max_length=256, truncation=True)
dispreferred_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'],
'dispreferred_ids': dispreferred_enc['input_ids']
}
9. 데이터셋 토큰화 & pytorch 텐서로 형식 변환
# 데이터셋 토큰화
tokenized_dataset = dataset['train'].map(
preprocess_text,
remove_columns=['id', 'system', 'question', 'chose', 'rejected'] # 데이터셋에서 불필요한 정보 삭제
)
# pytorch 텐서로 형식 변환
# 받아온 dataset은 허깅페이스 데이터셋이라 pytorch 모델에서 학습할 수 있도록 형태를 변환
tokenized_dataset.set_format(type='torch', columns=['input_ids', 'attention_mask', 'preferred_ids', 'dispreferred_ids'])
10. 데이터 배치 구성 함수
def collate_fn(batch):
input_ids = torch.stack(item['input_ids'].clone().detach() for item in batch)
attention_mask = torch.stack(item['attention_mask'].clone().detach() for item in batch)
# max(len(item['preferred_ids']): 선호 응답으로 되어 있는 아이 중 맥스 값 중
# max(max(len(item['preferred_ids']) for item in batch), 1): 잘못 들어온 케이스에 대비해 최대 길이를 1로 설정
max_length = max(max(len(item['preferred_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'].tolist() + [tokenizer.pad_token_id] * (max_length - len(item['preferred_ids'])),
dtype=torch.long
)
for item in batch
]).clone().detach() # 배치 구성될 수 있게 clone.detach
# 비선호 응답
# 패딩 처리 후 배치 처리
dispreferred_ids = torch.stack([
torch.tensor(
item['dispreferred_ids'].tolist() + [tokenizer.pad_token_id] * (max_length - len(item['dispreferred_ids'])),
dtype=torch.long
) if isinstance(item['dispreferred_ids'], torch.tensor)
else torch.tensor(
item['dispreferred_ids'].tolist() + [tokenizer.pad_token_id] * (max_length - len(item['dispreferred_ids'])),
dtype=torch.long
)
for item in batch
]).clone().detach()
return {
'input_ids': input_ids,
'attention_mask': attention_mask,
'preferred_ids': preferred_ids,
'dispreferred_ids': dispreferred_ids
}
11. DPOTrainer 클래스 구성 - Trainer 모듈의 compute_loss 메서드 오버라이딩
class DPOTrainer(Trainer):
# 선호, 비선호 응답에 대한 loss 구하기
def compute_loss(self, model, inputs, beta=0.1, *args, **kwargs):
# 연산을 할 수 있게 device(장치)를 맞춰줌
input_ids=inputs['input_ids'].to(model.device),
attention_mask=inputs['attention_mask'].to(model.device),
preferred_ids=inputs['preferred_ids'].to(model.device),
dispreferred_ids=inputs['dispreferred_ids'].to(model.device)
# 선호/비선호 응답에 대한 출력 구하기
preferred_outputs = model(input_ids=input_ids, attention_mask=attention_mask, labels=preferred_ids)
dispreferred_outputs = model(input_ids=input_ids, attention_mask=attention_mask, labels=dispreferred_ids)
# 출력에서 손실값 구하기
preferred_loss = preferred_outputs.loss
dispreferred_loss = dispreferred_outputs.loss
# DPO 손실의 공식을 적용하기
# import torch.nn.functional as F
loss = -F.logsigmoid(beta * (dispreferred_loss - preferred_loss)).mean() # 선호할 응답의 확률이 비선호 응답의 확률보다 높아지게 학습을 유도해서 DPO의 손실 공식을 적용
return loss
12. 훈련 설정 - 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
)
13. 훈련 과정 정의
trainer = DPOTrainer(
model=model,
args=training_args,
data_collator=collate_fn
)
14. 학습 - DPO 기법
trainer.train()
'이것저것' 카테고리의 다른 글
| 파인튜닝 (Fine-tuning) (0) | 2025.05.02 |
|---|---|
| 알고리즘 문제 풀기 튜토리얼 02 - github에 코드 업로드, pr 올리기 (0) | 2025.03.02 |
| [SK네트웍스 Family AI 캠프 11기] 1차 프로젝트 회고 (0) | 2025.02.28 |
| 알고리즘 문제 풀기 튜토리얼 01 (0) | 2025.02.19 |
| Windows 환경에서 oh-my-posh를 사용한 터미널 커스텀 (+VSCode 터미널, 글자 깨짐 해결) (0) | 2025.02.12 |
