RLHF: 인간 피드백으로 LLM 정렬하기

RLHF (Reinforcement Learning from Human Feedback) 는 대규모 언어 모델(LLM)을 인간의 의도와 선호에 맞게 정렬하는 학습 방법입니다. ChatGPT, Claude 등 현대 AI 어시스턴트의 핵심 기술입니다.

---

1. LLM 정렬 문제

사전 학습된 LLM의 한계

대규모 텍스트로 학습한 LLM은:

• 유해하거나 편향된 콘텐츠 생성 가능
• 질문에 직접 답하기보다 텍스트 완성 경향
• 거짓 정보를 자신 있게 생성 (hallucination)
• 인간의 의도를 이해하지 못함

정렬(Alignment)의 목표

1. Helpful (유용성): 사용자의 요청에 도움이 되는 응답
2. Harmless (무해성): 해로운 콘텐츠 생성 거부
3. Honest (정직성): 모르는 것은 모른다고 인정

---

2. RLHF 3단계 과정

1단계: Supervised Fine-Tuning (SFT)
       사람이 작성한 예시로 미세 조정

2단계: Reward Model Training
       인간 선호도 데이터로 보상 모델 학습

3단계: RL Fine-Tuning
       보상 모델로 PPO 학습

---

3. 1단계: Supervised Fine-Tuning (SFT)

과정

사람이 직접 작성한 고품질 응답 데이터로 LLM을 미세 조정합니다.

# SFT 데이터 형식
sft_data = [
    {
        "prompt": "파이썬에서 리스트를 정렬하는 방법은?",
        "response": "파이썬에서 리스트를 정렬하는 방법은 두 가지가 있습니다:\n\n1. `sort()` 메서드: 리스트 자체를 정렬합니다.\n```python\nmy_list = [3, 1, 4, 1, 5]\nmy_list.sort()\nprint(my_list)  # [1, 1, 3, 4, 5]\n```\n\n2. `sorted()` 함수: 새로운 정렬된 리스트를 반환합니다.\n```python\nmy_list = [3, 1, 4, 1, 5]\nnew_list = sorted(my_list)\nprint(new_list)  # [1, 1, 3, 4, 5]\n```"
    },
    # ... 더 많은 예시
]

구현

from transformers import AutoModelForCausalLM, AutoTokenizer, Trainer, TrainingArguments

def sft_training(model_name, dataset):
    model = AutoModelForCausalLM.from_pretrained(model_name)
    tokenizer = AutoTokenizer.from_pretrained(model_name)

    def preprocess(examples):
        texts = [f"User: {p}\nAssistant: {r}"
                 for p, r in zip(examples['prompt'], examples['response'])]
        return tokenizer(texts, truncation=True, padding=True, max_length=512)

    tokenized_dataset = dataset.map(preprocess, batched=True)

    training_args = TrainingArguments(
        output_dir="./sft_model",
        num_train_epochs=3,
        per_device_train_batch_size=4,
        learning_rate=2e-5,
        warmup_steps=100,
    )

    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=tokenized_dataset,
    )

    trainer.train()
    return model

---

4. 2단계: Reward Model (보상 모델)

인간 선호도 데이터 수집

같은 프롬프트에 대해 여러 응답을 생성하고, 사람이 순위를 매깁니다.

# 선호도 데이터 형식
preference_data = [
    {
        "prompt": "인공지능이란 무엇인가요?",
        "chosen": "인공지능(AI)은 인간의 학습, 추론, 지각 능력을 컴퓨터 시스템으로 구현한 기술입니다...",
        "rejected": "AI는 그냥 컴퓨터 프로그램이에요. 별거 아닙니다..."
    },
    # ...
]

Reward Model 구조

import torch
import torch.nn as nn
from transformers import AutoModel

class RewardModel(nn.Module):
    def __init__(self, base_model_name):
        super().__init__()
        self.base_model = AutoModel.from_pretrained(base_model_name)
        self.reward_head = nn.Linear(self.base_model.config.hidden_size, 1)

    def forward(self, input_ids, attention_mask):
        outputs = self.base_model(input_ids=input_ids, attention_mask=attention_mask)
        # 마지막 토큰의 hidden state 사용
        last_hidden = outputs.last_hidden_state[:, -1, :]
        reward = self.reward_head(last_hidden)
        return reward

Bradley-Terry 모델로 학습

두 응답 중 선호되는 응답의 보상이 더 높도록 학습:

$P(\text{chosen} > \text{rejected}) = \sigma(r_\theta(\text{chosen}) - r_\theta(\text{rejected}))$

def reward_model_loss(reward_model, chosen_ids, rejected_ids, chosen_mask, rejected_mask):
    """
    Bradley-Terry pairwise loss
    """
    chosen_rewards = reward_model(chosen_ids, chosen_mask)
    rejected_rewards = reward_model(rejected_ids, rejected_mask)

    # 선호 응답의 보상이 더 높아야 함
    loss = -torch.log(torch.sigmoid(chosen_rewards - rejected_rewards)).mean()

    return loss


def train_reward_model(model, dataset, epochs=3, lr=1e-5):
    optimizer = torch.optim.Adam(model.parameters(), lr=lr)

    for epoch in range(epochs):
        total_loss = 0
        for batch in dataset:
            loss = reward_model_loss(
                model,
                batch['chosen_ids'],
                batch['rejected_ids'],
                batch['chosen_mask'],
                batch['rejected_mask']
            )

            optimizer.zero_grad()
            loss.backward()
            optimizer.step()

            total_loss += loss.item()

        print(f"Epoch {epoch}, Loss: {total_loss / len(dataset):.4f}")

---

5. 3단계: PPO를 이용한 RL 학습

목표

보상 모델의 점수를 최대화하면서, SFT 모델과 너무 다르지 않도록 학습합니다.

$\max_\theta \mathbb{E}_{x \sim D, y \sim \pi_\theta(y|x)} [r_\phi(x, y)] - \beta D_{KL}(\pi_\theta || \pi_{ref})$

• r_φ: 보상 모델
• π_θ: 학습 중인 정책 (LLM)
• π_ref: 참조 모델 (SFT 모델)
• β: KL 페널티 계수

KL 페널티가 필요한 이유

• 보상 모델만 최대화하면 보상 해킹 발생
• 이상한 문장이 높은 보상을 받을 수 있음
• KL 페널티로 SFT 모델과 유사하게 유지

구현 (TRL 라이브러리 사용)

from trl import PPOTrainer, PPOConfig, AutoModelForCausalLMWithValueHead
from transformers import AutoTokenizer

def train_with_ppo(sft_model_path, reward_model, tokenizer):
    # PPO 설정
    config = PPOConfig(
        model_name=sft_model_path,
        learning_rate=1.41e-5,
        batch_size=16,
        mini_batch_size=4,
        gradient_accumulation_steps=4,
        optimize_cuda_cache=True,
        ppo_epochs=4,
        init_kl_coef=0.2,  # KL 페널티 초기값
    )

    # Value head가 추가된 모델
    model = AutoModelForCausalLMWithValueHead.from_pretrained(sft_model_path)
    ref_model = AutoModelForCausalLMWithValueHead.from_pretrained(sft_model_path)

    ppo_trainer = PPOTrainer(
        config=config,
        model=model,
        ref_model=ref_model,
        tokenizer=tokenizer,
    )

    for epoch in range(10):
        for batch in dataloader:
            prompts = batch['prompt']

            # 응답 생성
            response_tensors = ppo_trainer.generate(
                prompts,
                max_new_tokens=256,
                do_sample=True,
                temperature=0.7,
            )

            # 보상 계산
            rewards = []
            for prompt, response in zip(prompts, response_tensors):
                full_text = prompt + tokenizer.decode(response)
                reward = reward_model(full_text)
                rewards.append(reward)

            # PPO 업데이트
            stats = ppo_trainer.step(prompts, response_tensors, rewards)
            print(f"Epoch {epoch}, Reward: {stats['ppo/mean_scores']:.3f}")

    return model

---

6. RLHF의 문제점과 개선

문제점

1. 비용: 인간 레이블링 비용이 높음
2. 보상 해킹: 보상 모델을 속이는 방법 학습
3. 불안정성: PPO 학습이 불안정할 수 있음
4. 주관성: 인간 선호도의 일관성 부족

대안: DPO (Direct Preference Optimization)

보상 모델 없이 직접 선호도 데이터로 학습:

$L_{DPO} = -\log \sigma \left( \beta \log \frac{\pi_\theta(y_w|x)}{\pi_{ref}(y_w|x)} - \beta \log \frac{\pi_\theta(y_l|x)}{\pi_{ref}(y_l|x)} \right)$

def dpo_loss(model, ref_model, chosen_ids, rejected_ids, beta=0.1):
    """
    Direct Preference Optimization 손실
    보상 모델 없이 직접 학습
    """
    # 현재 모델의 로그 확률
    chosen_logprobs = get_logprobs(model, chosen_ids)
    rejected_logprobs = get_logprobs(model, rejected_ids)

    # 참조 모델의 로그 확률
    with torch.no_grad():
        ref_chosen_logprobs = get_logprobs(ref_model, chosen_ids)
        ref_rejected_logprobs = get_logprobs(ref_model, rejected_ids)

    # DPO 손실
    chosen_ratio = chosen_logprobs - ref_chosen_logprobs
    rejected_ratio = rejected_logprobs - ref_rejected_logprobs

    loss = -F.logsigmoid(beta * (chosen_ratio - rejected_ratio)).mean()

    return loss

---

7. 2024-2025 RLHF 발전

RLVR (Reinforcement Learning with Verifiable Rewards)

수학, 코드 등 검증 가능한 보상으로 학습:

def verifiable_reward(response, correct_answer):
    """
    정답을 자동으로 검증하여 보상 계산
    """
    # 수학 문제: 정답 비교
    if extract_answer(response) == correct_answer:
        return 1.0
    return 0.0

    # 코드 문제: 테스트 케이스 통과 여부
    # try:
    #     exec(response)
    #     return run_tests(response)
    # except:
    #     return 0.0

DeepSeek-R1 등 2025년 모델들이 RLVR을 적극 활용합니다.

GRPO (Group Relative Policy Optimization)

그룹 내 상대 비교를 통한 최적화:

def grpo_loss(model, prompts, group_size=4):
    """
    GRPO: 그룹 내 상대 비교로 학습
    보상 모델 없이 그룹 내 순위 기반
    """
    losses = []

    for prompt in prompts:
        # 같은 프롬프트에 여러 응답 생성
        responses = [model.generate(prompt) for _ in range(group_size)]

        # 그룹 내 상대 보상 계산
        rewards = compute_group_rewards(responses)

        # 상대적으로 좋은 응답 강화
        for response, reward in zip(responses, rewards):
            log_prob = model.log_prob(response)
            losses.append(-log_prob * reward)

    return torch.stack(losses).mean()

Safe RLHF

유용성(helpfulness)과 무해성(harmlessness)을 분리:

class SafeRLHF:
    def __init__(self, reward_model, cost_model):
        self.reward_model = reward_model  # 유용성
        self.cost_model = cost_model      # 유해성

    def compute_safe_reward(self, response, lambda_param=0.5):
        reward = self.reward_model(response)
        cost = self.cost_model(response)

        # 제약 조건: 유해성은 임계값 이하로 유지
        if cost > SAFETY_THRESHOLD:
            return -float('inf')

        return reward - lambda_param * cost

Constitutional AI (CAI)

AI가 자체적으로 응답을 평가하고 개선:

def constitutional_ai_step(model, response, principles):
    """
    헌법적 AI: 원칙에 따라 자체 검토
    """
    critique_prompt = f"""
    다음 응답을 검토하세요:
    {response}

    다음 원칙에 맞는지 확인하세요:
    {principles}

    문제가 있다면 수정된 응답을 작성하세요.
    """

    revised = model.generate(critique_prompt)
    return revised

---

8. 실습: 간단한 RLHF 파이프라인

from transformers import AutoModelForCausalLM, AutoTokenizer
from trl import SFTTrainer, RewardTrainer, PPOTrainer
from datasets import load_dataset

def simple_rlhf_pipeline():
    # 1. 기본 모델 로드
    model_name = "gpt2"
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    tokenizer.pad_token = tokenizer.eos_token

    # 2. SFT
    sft_dataset = load_dataset("your_sft_dataset")
    sft_model = AutoModelForCausalLM.from_pretrained(model_name)

    sft_trainer = SFTTrainer(
        model=sft_model,
        train_dataset=sft_dataset,
        dataset_text_field="text",
        max_seq_length=512,
    )
    sft_trainer.train()

    # 3. Reward Model
    preference_dataset = load_dataset("your_preference_dataset")
    reward_model = RewardModel(model_name)

    reward_trainer = RewardTrainer(
        model=reward_model,
        train_dataset=preference_dataset,
    )
    reward_trainer.train()

    # 4. PPO
    ppo_model = AutoModelForCausalLMWithValueHead.from_pretrained(sft_model)

    ppo_config = PPOConfig(
        learning_rate=1e-5,
        batch_size=8,
        ppo_epochs=4,
    )

    ppo_trainer = PPOTrainer(
        config=ppo_config,
        model=ppo_model,
        ref_model=sft_model,
        tokenizer=tokenizer,
        reward_model=reward_model,
    )

    # PPO 학습 루프
    for batch in ppo_dataset:
        queries = batch['query']
        responses = ppo_trainer.generate(queries)
        rewards = [reward_model(r) for r in responses]
        ppo_trainer.step(queries, responses, rewards)

    return ppo_model

---

9. 핵심 정리

---