3.3.18 [LLM] 파인튜닝 기법: RLHF

RLHF (Reinforcement Learning from Human Feedback)

 

RLHF는 사람의 피드백을 활용하여(사람의 선호 데이터를 기반으로) 모델의 성능을 개선하는 강화 학습 방법이다. 특히 사용자 경험을 최적화하거나 모델의 응답 품질을 높이는 데 유용하다. (모델이 응답 품질을 개선할 수 있도록 보상(reward)을 설계한다.)

 

- 단순한 정답 학습이 아니라, "어떤 응답이 더 좋은가?"에 대한 선호 순위/피드백을 학습

- 주로 대화형 LLM(ChatGPT, Claude) 등에 사용

- 응답에 대한 보상 함수 R(α)를 사람이 설계하거나 평가하여 학습에 사용

 

• RLHF 학습 과정

👉🏻 주요 단계

- 1단계: LLM을 기본적으로 학습시킨다.

- 2단계: 사람이 모델 응답의 품질을 평가하여 보상 데이터를 생성한다.

- 3단계: 강화 학습을 통해 보상 신호로 모델을 조정한다.

단계	이름	설명
1단계	지도학습 (SFT: Supervised Fine-Tuning)	instruction 데이터로 LLM 기본 fine-tuning
2단계	보상 모델 학습 (Reward Model)	동일 질문에 대한 여러 응답 중 사람의 선호 순위 학습
3단계	강화학습 (PPO, ORPO)	보상 모델의 출력을 기반으로 LLM 최적화

 

👉🏻 모델의 보상 R은 응답 α에 대해 다음과 같이 정의한다.

R(α) = human feedback score

- 정책을 최적화한다.

 

• RLHF와 InstructGPT/ChatGPT

1. InstructGPT

OpenAI에서 RLHF를 활용해 학습한 GPT 모델이다.

사용자의 지시(instruction)을 따르는 데 최적화되어 있다.

RLHF를 통해 모델의 응답이 사용자 친화적이고 유용하도록 개선한다.

 

👉🏻 주요 특징

- 사용자 피드백을 기반으로 답변의 관련성과 품질 향상

- ChatGPT의 기반 기술

 

2. ChatGPT와 RLHF

ChatGPT는 GPT 모델을 기반으로, RLHF를 통해 사용자 요구에 맞는 자연스러운 응답을 생성한다.

 

👉🏻 과정

1단계: GPT 모델을 사전 학습한다.

2단계: 사람이 응답을 평가해 보상 모델을 학습한다.

3단계: 보상 모델로 강화 학습 수행(PPO 알고리즘 사용)한다.

 

👉🏻 역할

- 사람의 피드백으로 모델을 최적화하여 더 나은 대화 경험을 제공한다.

 

• RLHF의 장점과 한계

👉🏻 장점

- 응답 품질 개선

- 사용자 맞춤형 학습

- 다양한 도메인 적용

 

👉🏻 단점

- 피드백 의존성

- 높은 학습 비용

- 편향 가능성

 

 

실습 (RLHF)

 

0. 환경 설정

!pip install transformers torch stable-baselines3

 

1. LLM 모델 로드 및 텍스트 생성

from transformers import pipeline

generator = pipeline('text-generation', model='gpt2')

def generate_text(prompt, max_length=150):
  response = generator(prompt, max_length=max_length, num_return_sequences=1)
  return response[0]['generated_text']
  
prompt = "This is sunny day, and"
print(generate_text(prompt))

 

2. 강화학습을 위한 Feedback 환경 생성

!pip install 'shimmy>=2.0'

import gymnasium as gym
import numpy as np
from stable_baselines3 import PPO

class ContentFeedbackEnv(gym.Env):
  def __init__(self):
    super(ContentFeedbackEnv, self).__init__()
    self.action_space = gym.spaces.Discrete(3) # 0: 싫어요, 1: 좋아요, 2: 유해 콘텐츠 신고
    self.observation_space = gym.spaces.Box(low=0, high=1, shape=(1,), dtype=np.float32)
    self.history = []

  def step(self, action):
    if action == 1:
      reward = 1
      feedback = "Like"
    elif action == 2:
      reward = -2
      feedback = "Danger"
    else:
      reward = -1
      feedback = "Hate"

    self.history.append(feedback)

    obs = np.array([0.5])
    terminated = False
    truncated = False
    info = {}

    return obs, reward, terminated, truncated, info

  def reset(self, seed=None, options=None):
    super().reset(seed=seed)
    return np.array([0.5]), {}

 

3. PPO 모델 생성 및 학습

env = ContentFeedbackEnv()
model = PPO("MlpPolicy", env, verbose=1)

past_feedback = [1, 0, 2, 1, 1, 0, 2, 1, 0, 1]

for action in past_feedback:
  env.step(action)
  
# PPO 모델 학습
model.learn(total_timesteps=10000)
model.save('rlhf_content_model')

# 저장된 모델 로드
model = PPO.load('rlhf_content_model')

env = ContentFeedbackEnv()
model.set_env(env)

prompt = "This is windy day, so"
response = generate_text(prompt)

print(response)

action = 1

env.step(action)
model.learn(total_timesteps=10)

 

 

실습 (사용자 선호에 맞는 시 창작 모델)

 

• 0. 환경 설정

!python -m pip install --upgrade pip
!pip install typing_extensions pydantic openai
!pip install datasets transformers peft trl bitsandbytes

import os
import torch

os.environ["WANDB_DISABLED"] = "true"
os.environ["TOKENIZERS_PARALLELISM"] = "false"

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

 

• 1. 지도학습 (기반모델 Q-LoRA 파인튜닝)

(1) 학습용 데이터 준비

import json
from datasets import Dataset

# 데이터 로드 및 Dataset 변환
dataset_path = "./korean_poetry_dataset.json"

with open(dataset_path, "r", encoding="utf-8") as f:
    poem_data = json.load(f)

processed_data = [
    {"topic": item["text"]["topic"], "poem": item["text"]["poem"]}
    for item in poem_data
]

train_dataset = Dataset.from_list(processed_data)

# tokenizer 로드
from transformers import AutoTokenizer

model_name = "NCSOFT/Llama-VARCO-8B-Instruct"

tokenizer = AutoTokenizer.from_pretrained(model_name)

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

# 전처리 함수: 토큰화 + 라벨링
def preprocess_text(sample):
    input_texts = [f"주제: {t}\n\n시: {p}" for t, p in zip(sample["topic"], sample["poem"])]
    model_inputs = tokenizer(
        input_texts,
        padding="max_length",
        max_length=512,
        truncation=True
    )

    model_inputs["labels"] = model_inputs["input_ids"].copy()
    pad_token_id = tokenizer.pad_token_id
    model_inputs["labels"] = [
        [(l if l != pad_token_id else -100) for l in label]
        for label in model_inputs["labels"]
    ]

    return model_inputs
    
train_dataset = train_dataset.map(
    preprocess_text,
    batched=True,
    remove_columns=["topic", "poem"]
)

from transformers import DataCollatorForSeq2Seq

data_collator = DataCollatorForSeq2Seq(tokenizer, model=None)

 

(2) 파인튜닝 학습 준비

- 양자화 설정 > 모델 로드

- 학습 모드로 전환

- LoRA 학습 설정

- TrainingArguments 설정

from transformers import BitsAndBytesConfig

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

from transformers import AutoModelForCausalLM

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

model.gradient_checkpointing_enable()
model.config.use_cache = False
model.config.attn_implementation = "flash_attention_2"

from peft import LoraConfig

lora_config = LoraConfig(
    r=16,
    lora_alpha=32,
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM"
)

from peft import prepare_model_for_kbit_training

model = prepare_model_for_kbit_training(model)

from peft import get_peft_model

model = get_peft_model(model, lora_config)
model.train()

from transformers import TrainingArguments, Trainer

training_args = TrainingArguments(
    output_dir="./q_lora_poem",
    save_strategy="epoch",
    per_device_train_batch_size=1,
    gradient_accumulation_steps=8,
    learning_rate=2e-4,
    num_train_epochs=3,
    logging_dir="./logs",
    logging_steps=100,
    save_total_limit=2,
    optim="adamw_bnb_8bit",
    report_to="none"
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    tokenizer=tokenizer,
    data_collator=data_collator
)

 

(3) 학습 진행

trainer.train()

 

• 2. 학습된 모델로 시(응답) 생성

(1) 모델 로드

from transformers import pipeline

qlora_checkpoint = "./q_lora_poem/checkpoint-xxx"

model = AutoModelForCausalLM.from_pretrained(qlora_checkpoint)
tokenizer = AutoTokenizer.from_pretrained(model_name)

generate_pipeline = pipeline(
    "text-generation",
    model=model,
    tokenizer=tokenizer,
    pad_token_id=tokenizer.eos_token_id,
    batch_size=2
)

topics = ["바람", "비", "노을", "달빛", "안개", "사랑", "이별", "운명", "기다림", "후회", "추억", "시간", "청춘", "변화", "마지막 순간", "군중", "밤거리", "버스", "인생", "빌딩", "사람들", "거짓말", "욕망", "돈", "권력", "비밀", "죽음", "희망", "동물", "자연", "도시", "바다", "산", "하늘", "별", "꽃", "나무", "강", "바위", "흙", "눈", "빗방울", "눈물", "웃음"]

eval_file = "rlhf_evaluation_data.json"

try:
    with open(eval_file, "r", encoding="utf-8") as f:
        eval_dataset = json.load(f)
except FileNotFoundError:
    eval_dataset = []
    
num_batches = 5
batch_size = 20
total_samples = num_batches * batch_size
generated_samples = len(eval_dataset)

 

(2) 시 생성

# 시 생성 함수 정의
import time
import random
from tqdm import tqdm

def generate_poem_batch():
    batch_data = []

    with tqdm(total=batch_size, desc="✍️시 생성 중...", leave=False) as t:
        for _ in range(batch_size):
            topic = random.choice(topics)
            input_text = f"주제: {topic}\n\n시:"

            start_time = time.time()
            poem = generate_pipeline(
                input_text,
                max_new_tokens=100,
                temperature=0.8,
                top_p=0.9
            )[0]["generated_text"]
            end_time = time.time()

            gen_time = end_time - start_time
            batch_data.append({
                "topic": topic,
                "poem": poem,
                "selected": None
            })

            t.update(1)

            global generated_samples
            generated_samples += 1
            complete_rate = (generated_samples / total_samples) * 100
            remaining_time = ((total_samples - generated_samples) * gen_time) / 60

            print(f"\n{generated_samples}/{total_samples}개 완료 ({complete_rate:.2f}%)")
            print(f" - 예상 남은 시간: {remaining_time:.1f}분")
            print("-" * 100)

    return batch_data

for _ in tqdm(range(num_batches), desc="<<< 전체 진행 상황 >>>", position=0):
    eval_dataset.extend(generate_poem_batch())

    with open(eval_file, "w", encoding="utf-8") as f:
        json.dump(eval_dataset, f, ensure_ascii=False, indent=4)

 

(3) 피드백

- 생성된 시에 대해 selected="True"로 수정해 피드백 반영

 

• 3. Reward Model 학습

(1) 데이터 로드 및 처리

with open(eval_file, "r", encoding="utf-8") as f:
    evaluation_data = json.load(f)

reward_data = [
    {"text_a": f"주제: {item['topic']},", "text_b": item["poem"]}
     for item in evaluation_data if item["selected"]
]

reward_dataset = Dataset.from_list(reward_data)

# 데이터 전처리
def preprocess_reward_data(sample):
    
    model_inputs = tokenizer(
        sample["text_a"],
        text_pair=sample["text_b"],
        max_length=512,
        truncation=True
    )

    model_inputs["labels"] = model_inputs["input_ids"].copy()
    pad_token_id = tokenizer.pad_token_id
    model_inputs["labels"] = [
        [(l if l != pad_token_id else -100) for l in label]
        for label in model_inputs["labels"]
    ]

    return model_inputs
    
reward_dataset = reward_dataset.map(
    preprocess_reward_data,
    batched=True,
    remove_columns=["text_a", "text_b"]
)

 

(2) 학습 준비

- 양자화 설정 > 모델 로드

- LoRA 학습 설정

- TrainingArguments 설정

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

reward_model = AutoModelForCausalLM.from_pretrained(
    model_name,
    quantization_config=bnb_config,
    device_map="auto"
)

lora_config = LoraConfig(
    r=16,
    lora_alpha=32,
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM"
)

reward_model = prepare_model_for_kbit_training(reward_model)

reward_model = get_peft_model(reward_model, lora_config)

reward_training_args = TrainingArguments(
    output_dir="./reward_model",
    save_strategy="epoch",
    per_device_train_batch_size=1,
    gradient_accumulation_steps=8,
    learning_rate=2e-4,
    num_train_epochs=3,
    logging_dir="./logs",
    logging_steps=100,
    save_total_limit=2,
    remove_unused_columns=False,
    fp16=True
)

reward_trainer = Trainer(
    model=reward_model,
    args=reward_training_args,
    train_dataset=reward_dataset,
    tokenizer=tokenizer
)

 

(3) 학습 진행

reward_trainer.train()

 

• 4. RLHF (ORPO)

(1) 모델 로드

model = AutoModelForCausalLM.from_pretrained(qlora_checkpoint)
tokenizer = AutoTokenizer.from_pretrained(model_name)

model.train()
model.cuda()

for param in model.parameters():
    param.requires_grad = True

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

# !export PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True

# torch.cuda.empty_cache()

 

(2) ORPO 데이터셋 준비

with open(eval_file, "r", encoding="utf-8") as f:
    evaluation_data = json.load(f)

orpo_data = []

for item in evaluation_data:
    if item["selected"]:
        prompt_text = f"주제: {item['topic']}\n\n이 주제에 맞는 시를 작성해 주세요."
        chosen_text = item["poem"]
        rejected_text = ""

        tokenized_prompt = tokenizer(prompt_text, truncation=True, padding="max_length", max_length=64, return_tensors="pt")
        tokenized_chosen = tokenizer(chosen_text, truncation=True, padding="max_length", max_length=64, return_tensors="pt")
        tokenized_rejected = tokenizer(rejected_text, truncation=True, padding="max_length", max_length=64, return_tensors="pt")

        orpo_data.append({
            "prompt": prompt_text,
            "chosen": chosen_text,
            "rejected": rejected_text,
            "prompt_input_ids": tokenized_prompt["input_ids"].squeeze(0).cuda(),
            "prompt_attention_mask": tokenized_prompt["attention_mask"].squeeze(0).cuda(),
            "chosen_input_ids": tokenized_chosen["input_ids"].squeeze(0).cuda(),
            "chosen_attention_mask": tokenized_chosen["attention_mask"].squeeze(0).cuda(),
            "rejected_input_ids": tokenized_rejected["input_ids"].squeeze(0).cuda(),
            "rejected_attention_mask": tokenized_rejected["attention_mask"].squeeze(0).cuda()
        })

        orpo_dataset = Dataset.from_list(orpo_data)

 

(3) ORPO 설정

from trl import ORPOConfig

orpo_config = ORPOConfig(
    output_dir='./orpo_output',
    per_device_train_batch_size=1,
    num_train_epochs=5,
    learning_rate=2e-6,
    gradient_accumulation_steps=4,
    logging_steps=50,
    fp16=False,
    bf16=True,
    remove_unused_columns=False,
    gradient_checkpointing=True,
    max_grad_norm=1.0,
    warmup_steps=100,
    save_steps=500,
    save_total_limit=2
)

from trl.trainer.utils import DPODataCollatorWithPadding

data_collator = DPODataCollatorWithPadding(
    pad_token_id=tokenizer.pad_token_id,
    label_pad_token_id=-100,
    is_encoder_decoder=False
)

from trl import ORPOTrainer

orpo_trainer = ORPOTrainer(
    model=model,
    args=orpo_config,
    train_dataset=orpo_dataset,
    data_collator=data_collator,
    processing_class=tokenizer
)

 

(4) ORPO 적용

orpo_trainer.train()

 

• 최종 시 생성

orpo_checkpoint = "./orpo_output/checkpoint-xxx"

model = AutoModelForCausalLM.from_pretrained(orpo_checkpoint)
tokenizer = AutoTokenizer.from_pretrained(model_name)

generate_pipeline = pipeline(
    "text-generation",
    model=model,
    tokenizer=tokenizer,
    pad_token_id=tokenizer.eos_token_id
)

def generate_poem_final(num_samples=5):
    result = []

    for _ in range(num_samples):
        topic = random.choice(topics)
        input_text = f"주제: {topic}\n\n시:"

        poem = generate_pipeline(
            input_text,
            max_new_tokens=100,
            temperature=0.8,
            top_p=0.9
        )[0]["generated_text"]

        result.append({"topic": topic, "poem": poem})

    return result
    
generated_poem = generate_poem_final()
generated_poem