이 글은 누구를 위한 것인가
- LLM을 특정 도메인에 맞게 정렬(align)하려는 ML 엔지니어
- RLHF와 DPO의 차이를 이해하고 적용하려는 팀
- 선호도 데이터 수집 파이프라인을 설계하는 개발자
들어가며
사전학습 LLM은 텍스트를 생성하지만, 인간의 가치와 의도에 맞게 동작하지 않을 수 있다. RLHF는 인간 피드백으로 보상 모델을 학습하고, PPO로 LLM을 최적화해 유용하고 무해한 응답을 생성하도록 정렬한다.
이 글은 bluefoxdev.kr의 RLHF 완전 가이드 를 참고하여 작성했습니다.
1. RLHF 파이프라인 설계
[RLHF 3단계 파이프라인]
1단계: SFT (Supervised Fine-Tuning)
기반 LLM + 시연 데이터(demonstrations) → SFT 모델
고품질 응답 예시로 초기 정렬
2단계: 보상 모델(RM) 학습
SFT 모델 → 여러 응답 생성
인간 평가자 → 선호도 쌍 수집 (chosen > rejected)
Bradley-Terry 모델로 보상 스칼라 학습
3단계: RL 최적화 (PPO)
Actor: SFT 모델 (정책 π_θ)
Critic: 보상 모델 (가치 함수)
Reference: 원본 SFT (KL 페널티용)
목표: E[r_RM(y)] - β·KL(π_θ || π_SFT)
β: KL 페널티 계수 (너무 크면 정렬 실패, 너무 작으면 드리프트)
[DPO vs RLHF]
RLHF: RM 학습 → PPO 최적화 (2단계, 복잡)
DPO: RM 없이 선호도 데이터로 직접 최적화 (1단계, 간단)
DPO 손실함수:
-log σ(β · log(π_θ(y_w)/π_ref(y_w)) - β · log(π_θ(y_l)/π_ref(y_l)))
DPO 장점: 안정적, 빠름, RM 없음
DPO 단점: 온라인 탐색 없음 → 분포 외 응답 취약
[선호도 데이터 품질]
Golden rule: 애매한 쌍(tie)은 제거
레이블러 간 일치율(Cohen's κ) > 0.6 목표
다양성: 같은 프롬프트에 대한 응답 변이
2. RLHF/DPO 구현
# 보상 모델 학습 (TRL 라이브러리)
from datasets import Dataset
from trl import RewardTrainer, RewardConfig
from transformers import AutoModelForSequenceClassification, AutoTokenizer
def train_reward_model():
model_name = "meta-llama/Llama-3.2-1B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_name)
tokenizer.pad_token = tokenizer.eos_token
# 보상 모델: 분류 헤드 추가 (num_labels=1)
model = AutoModelForSequenceClassification.from_pretrained(
model_name,
num_labels=1,
torch_dtype="auto",
)
# 선호도 데이터셋 (chosen/rejected 쌍)
preference_data = [
{
"prompt": "한국 역사에서 가장 중요한 사건은?",
"chosen": "한국 역사에서 가장 중요한 사건 중 하나는 1948년 대한민국 정부 수립입니다. 이는 한반도 분단과 냉전의 맥락에서...",
"rejected": "잘 모르겠는데요. 역사는 복잡하니까요.",
},
# ... 수천 개의 선호도 쌍
]
dataset = Dataset.from_list(preference_data)
config = RewardConfig(
output_dir="./reward_model",
num_train_epochs=2,
per_device_train_batch_size=4,
gradient_accumulation_steps=4,
learning_rate=1e-5,
max_length=512,
remove_unused_columns=False,
)
trainer = RewardTrainer(
model=model,
tokenizer=tokenizer,
args=config,
train_dataset=dataset,
)
trainer.train()
trainer.save_model("./reward_model_final")
# DPO 파인튜닝 (RLHF보다 단순)
from trl import DPOTrainer, DPOConfig
from transformers import AutoModelForCausalLM
def train_with_dpo():
model_name = "meta-llama/Llama-3.2-1B-Instruct"
model = AutoModelForCausalLM.from_pretrained(model_name, torch_dtype="auto")
ref_model = AutoModelForCausalLM.from_pretrained(model_name, torch_dtype="auto")
tokenizer = AutoTokenizer.from_pretrained(model_name)
tokenizer.pad_token = tokenizer.eos_token
# DPO 데이터셋 형식
dpo_dataset = Dataset.from_list([
{
"prompt": "코드 리뷰 피드백을 작성해줘: def add(a, b): return a+b",
"chosen": [
{"role": "user", "content": "코드 리뷰 피드백을 작성해줘: def add(a, b): return a+b"},
{"role": "assistant", "content": "함수 자체는 올바르게 동작합니다. 개선 제안: 1) 타입 힌트 추가: `def add(a: int | float, b: int | float) -> int | float` 2) docstring 추가 3) 엣지케이스(문자열 입력) 처리 고려."},
],
"rejected": [
{"role": "user", "content": "코드 리뷰 피드백을 작성해줘: def add(a, b): return a+b"},
{"role": "assistant", "content": "코드 좋아요."},
],
},
])
config = DPOConfig(
output_dir="./dpo_model",
num_train_epochs=3,
per_device_train_batch_size=2,
gradient_accumulation_steps=8,
learning_rate=5e-7,
beta=0.1, # KL 페널티 계수
max_length=1024,
max_prompt_length=512,
loss_type="sigmoid", # 기본 DPO 손실
)
trainer = DPOTrainer(
model=model,
ref_model=ref_model,
tokenizer=tokenizer,
args=config,
train_dataset=dpo_dataset,
)
trainer.train()
# 선호도 데이터 수집 파이프라인
import anthropic
from dataclasses import dataclass
from typing import Literal
client = anthropic.Anthropic()
@dataclass
class PreferencePair:
prompt: str
chosen: str
rejected: str
annotator_confidence: float # 0-1
async def collect_preferences(prompts: list[str], n_responses: int = 4) -> list[PreferencePair]:
"""각 프롬프트에 대해 여러 응답 생성 후 자동 선호도 평가"""
pairs = []
for prompt in prompts:
# n개 응답 생성 (temperature 변화로 다양성 확보)
responses = []
for i in range(n_responses):
response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=512,
temperature=0.5 + i * 0.2, # 0.5, 0.7, 0.9, 1.1
messages=[{"role": "user", "content": prompt}]
)
responses.append(response.content[0].text)
# Claude로 자동 선호도 평가 (대규모 수집 시)
ranking_prompt = f"""다음 응답들을 품질 기준으로 순위를 매겨주세요.
프롬프트: {prompt}
응답들:
{chr(10).join(f"[{i+1}] {r}" for i, r in enumerate(responses))}
JSON으로 반환: {{"ranking": [1,3,2,4], "best_idx": 0, "worst_idx": 3, "confidence": 0.85}}
ranking은 가장 좋은 순서대로 응답 번호를 나열한 것입니다."""
eval_result = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=200,
messages=[{"role": "user", "content": ranking_prompt}]
)
import json
ranking_data = json.loads(eval_result.content[0].text)
best_idx = ranking_data["ranking"][0] - 1
worst_idx = ranking_data["ranking"][-1] - 1
pairs.append(PreferencePair(
prompt=prompt,
chosen=responses[best_idx],
rejected=responses[worst_idx],
annotator_confidence=ranking_data["confidence"],
))
# 낮은 신뢰도 필터링
return [p for p in pairs if p.annotator_confidence >= 0.7]
# 보상 모델 평가 지표
def evaluate_reward_model(rm_model, tokenizer, test_pairs: list[PreferencePair]) -> dict:
correct = 0
total = len(test_pairs)
for pair in test_pairs:
chosen_score = get_reward_score(rm_model, tokenizer, pair.prompt, pair.chosen)
rejected_score = get_reward_score(rm_model, tokenizer, pair.prompt, pair.rejected)
if chosen_score > rejected_score:
correct += 1
return {
"accuracy": correct / total, # 목표: > 0.7
"total_pairs": total,
}
def get_reward_score(model, tokenizer, prompt: str, response: str) -> float:
text = f"{prompt}\n\n{response}"
inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
with __import__("torch").no_grad():
outputs = model(**inputs)
return outputs.logits[0].item()
마무리
RLHF의 핵심은 인간 선호도 신호로 보상 모델을 학습하고, KL 페널티로 원본 모델에서 너무 멀리 벗어나지 않도록 균형을 맞추는 것이다. DPO는 보상 모델 없이 선호도 데이터를 직접 최적화해 구현이 간단하고 안정적이다. 소규모 팀이라면 TRL 라이브러리의 DPOTrainer로 시작해 선호도 데이터 품질(신뢰도 ≥ 0.7)에 집중하는 것이 효율적이다.