1. 왜 파라미터 효율적 미세 조정(PEFT)이 필요한가
1.1 대규모 언어 모델 미세 조정의 어려움
거대 언어 모델(LLM)의 성능은 놀랍지만, 8B, 13B, 70B처럼 거대한 모델을 통째로 미세 조정(full fine-tuning)하는 것은 현실적으로 많은 문제를 안고 있습니다.
- 막대한 컴퓨팅 자원: 8B 파라미터 모델을 전체 학습하려면 엄청난 GPU 메모리와 시간이 필요합니다.
- 높은 저장 비용: 각 작업마다 모델 전체 파라미터를 저장해야 하므로 저장 공간이 많이 소모됩니다.
- 파괴적 망각(Catastrophic Forgetting): 특정 작업에 과도하게 특화되면 모델이 기존의 일반적인 능력을 잃을 수 있습니다.
1.2 LoRA 기술의 등장
LoRA(Low-Rank Adaptation)는 이러한 문제를 해결하기 위해 등장한 파라미터 효율적 미세 조정 기법입니다. 핵심 아이디어는 다음과 같습니다.
- 사전 학습 모델 고정: 원본 모델의 가중치는 그대로 둡니다.
- 작은 어댑터 삽입: 모델의 핵심 레이어에 저차원 분해가 가능한 작은 학습 가능 행렬을 추가합니다.
- 학습 파라미터 대폭 감소: 전체 파라미터의 0.01%~1% 정도만 학습해도 전체 미세 조정에 준하는 성능을 냅니다.
8B 모델의 경우, LoRA를 사용하면 학습 가능한 파라미터 수를 80억 개에서 수백만 개 수준으로 줄일 수 있습니다.
2. LoRA 기술의 핵심 원리
2.1 수학적 배경: 저차원 행렬 분해
LoRA는 Transformer 구조의 자기 주의(Self-Attention) 모듈에 적용됩니다. 가중치 행렬 \(W_0\)의 변화량 \(\Delta W\)가 낮은 '내재적 순위(Intrinsic Rank)'를 가질 것이라는 가정에서 출발합니다.
순전파(Forward Propagation) 과정은 다음과 같이 표현됩니다.
h = W₀x + ΔWx여기서 \(\Delta W\)를 두 개의 저차원 행렬의 곱으로 분해합니다.
ΔW = BA여기서 \(B \in \mathbb{R}^{d \times r}\), \(A \in \mathbb{R}^{r \times k}\)이며, 순위 \(r\)은 \(d\)와 \(k\)보다 훨씬 작습니다. 이 분해의 장점은 다음과 같습니다.
- 파라미터 효율성: 학습 가능한 파라미터 수가 \(d \times k\)에서 \(r \times (d + k)\)로 줄어듭니다.
- 추론 지연 없음: 학습 후에는 BA를 원래 가중치 \(W_0\)에 합쳐서 추가 연산 없이 사용할 수 있습니다.
2.2 Transformer 구조 내 LoRA 적용 위치
LoRA는 일반적으로 다음 모듈에 적용됩니다.
- Query, Key, Value (Q/K/V) 투영 행렬
- 출력 투영 행렬
- 순방향 신경망(FFN)의 업/다운 투영 행렬
| 적용 모듈 | 권장 순위(r) | Alpha 파라미터 | 사용 사례 |
|---|---|---|---|
| Q/K/V 투영 | 8-32 | 16-32 | 일반 작업 적응 |
| 출력 투영 | 16-64 | 32-64 | 출력 형식 조정 |
| 순방향 신경망 | 4-16 | 8-16 | 지식 주입 작업 |
| 모든 선형 레이어 | 8-32 | 16-32 | 전면 미세 조정 |
2.3 LoRA의 변형 및 발전
표준 LoRA 외에도 다음과 같은 개선된 버전들이 연구되고 있습니다.
- AdaLoRA: 모듈의 중요도에 따라 순위를 동적으로 할당합니다.
- LoRA+: A와 B 행렬에 서로 다른 학습률(일반적으로 lr_B > lr_A)을 적용합니다.
- VeRA: 학습 가능한 파라미터를 더욱 줄이기 위해 벡터화된 무작위 행렬을 사용합니다.
3. 개발 환경 설정
3.1 하드웨어 요구 사항
8B 모델 학습에 필요한 GPU 리소스는 다음과 같습니다.
| GPU 모델 | VRAM 요구량 | 학습 배치 크기 | 예상 학습 시간 |
|---|---|---|---|
| RTX 3090 (24GB) | 20-24GB | 4-8 | 중간 |
| RTX 4090 (24GB) | 20-24GB | 4-8 | 빠름 |
| A100 (40/80GB) | 32-40GB | 16-32 | 매우 빠름 |
| V100 (32GB) | 28-32GB | 8-16 | 중간 |
3.2 소프트웨어 환경 구성
Python 가상 환경을 만들고 필요한 라이브러리를 설치합니다.
# conda 환경 생성
conda create -n lora-training python=3.10
conda activate lora-training
# PyTorch 설치 (CUDA 버전에 맞게 선택)
pip install torch==2.1.0 torchvision==0.16.0 torchaudio==2.1.0 --index-url https://download.pytorch.org/whl/cu118
# Transformer 관련 라이브러리 설치
pip install transformers==4.35.0 datasets==2.14.0 accelerate==0.24.0 peft==0.7.0
# 학습 및 평가 도구 설치
pip install bitsandbytes==0.41.0 trl==0.7.0 evaluate==0.4.0 sentencepiece==0.1.99
# 시각화 도구 설치
pip install tensorboard==2.14.0 matplotlib==3.7.0
3.3 환경 검증
verify_environment.py 스크립트를 만들어 설치가 제대로 되었는지 확인합니다.
import torch
import transformers
import peft
import accelerate
import bitsandbytes as bnb
print(f"PyTorch 버전: {torch.__version__}")
print(f"CUDA 사용 가능: {torch.cuda.is_available()}")
print(f"CUDA 버전: {torch.version.cuda}")
print(f"GPU 개수: {torch.cuda.device_count()}")
print(f"현재 GPU: {torch.cuda.current_device()}")
print(f"GPU 이름: {torch.cuda.get_device_name()}")
if torch.cuda.is_available():
print(f"GPU 메모리: {torch.cuda.get_device_properties(0).total_memory / 1024**3:.1f} GB")
print(f"Transformers 버전: {transformers.__version__}")
print(f"PEFT 버전: {peft.__version__}")
print(f"Accelerate 버전: {accelerate.__version__}")
print(f"BitsAndBytes 버전: {bnb.__version__}")
# 기본 기능 테스트
from transformers import AutoModelForCausalLM, AutoTokenizer
try:
tokenizer = AutoTokenizer.from_pretrained("facebook/opt-125m")
model = AutoModelForCausalLM.from_pretrained("facebook/opt-125m")
print("✓ 기초 모델 로드 테스트 통과")
except Exception as e:
print(f"✗ 기초 모델 로드 실패: {e}")
4. 데이터 준비 및 전처리
4.1 데이터 형식 설계
LoRA 학습은 작업 유형에 따라 다양한 데이터 형식을 지원합니다.
명령어 수행(Instruction Following) 형식
{
"instruction": "Translate the following English to Korean",
"input": "Hello, how are you?",
"output": "안녕하세요, 어떻게 지내세요?"
}
대화(Conversational) 형식
{
"conversations": [
{"role": "user", "content": "인공지능이란 무엇인가요?"},
{"role": "assistant", "content": "인공지능은..."}
]
}
텍스트 완성(Text Completion) 형식
{
"text": "오늘 날씨가 좋아서 공원에 산책을 나왔다. 공원에는..."
}
4.2 데이터 전처리 파이프라인
data_preprocessing.py 스크립트를 생성합니다.
import json
import pandas as pd
from datasets import Dataset, DatasetDict
from transformers import AutoTokenizer
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class DataPreprocessor:
def __init__(self, model_name, max_length=1024):
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.max_length = max_length
if self.tokenizer.pad_token is None:
self.tokenizer.pad_token = self.tokenizer.eos_token
def load_data(self, file_path, format_type="instruction"):
logger.info(f"{file_path}에서 데이터 로드 중")
with open(file_path, 'r', encoding='utf-8') as f:
if file_path.endswith('.json'):
data = json.load(f)
elif file_path.endswith('.jsonl'):
data = [json.loads(line) for line in f]
else:
raise ValueError("지원하지 않는 파일 형식입니다.")
logger.info(f"{len(data)}개의 데이터 로드 완료")
return data
def format_instruction_data(self, examples):
formatted_texts = []
for example in examples:
if 'input' in example and example['input']:
text = f"### Instruction:\n{example['instruction']}\n\n### Input:\n{example['input']}\n\n### Response:\n{example['output']}"
else:
text = f"### Instruction:\n{example['instruction']}\n\n### Response:\n{example['output']}"
formatted_texts.append(text)
return formatted_texts
def tokenize_function(self, examples):
tokenized = self.tokenizer(
examples["text"],
truncation=True,
padding=False,
max_length=self.max_length,
return_overflowing_tokens=False,
)
tokenized["labels"] = tokenized["input_ids"].copy()
return tokenized
def prepare_dataset(self, data, format_type="instruction", train_ratio=0.9):
logger.info("데이터 포맷팅 중...")
if format_type == "instruction":
texts = self.format_instruction_data(data)
else:
texts = [item["text"] for item in data]
dataset = Dataset.from_dict({"text": texts})
dataset = dataset.train_test_split(train_size=train_ratio, seed=42)
tokenized_dataset = dataset.map(
self.tokenize_function,
batched=True,
remove_columns=dataset["train"].column_names,
)
logger.info(f"훈련 세트 크기: {len(tokenized_dataset['train'])}")
logger.info(f"검증 세트 크기: {len(tokenized_dataset['test'])}")
return tokenized_dataset
if __name__ == "__main__":
preprocessor = DataPreprocessor("facebook/opt-1.3b")
data = [
{"instruction": "Translate English to Korean", "input": "Hello", "output": "안녕하세요"},
{"instruction": "Summarize this", "input": "AI is...", "output": "AI는..."}
]
dataset = preprocessor.prepare_dataset(data, format_type="instruction")
print(dataset)
4.3 데이터 품질 검사
data_quality_check.py 도구를 만듭니다.
import matplotlib.pyplot as plt
import seaborn as sns
from collections import Counter
import numpy as np
class DataQualityChecker:
def __init__(self, tokenizer):
self.tokenizer = tokenizer
def analyze_text_length(self, texts):
lengths = [len(self.tokenizer.encode(text)) for text in texts]
plt.figure(figsize=(12, 4))
plt.subplot(1, 2, 1)
plt.hist(lengths, bins=50, alpha=0.7, color='skyblue')
plt.xlabel('텍스트 길이 (tokens)')
plt.ylabel('빈도')
plt.title('텍스트 길이 분포')
plt.subplot(1, 2, 2)
plt.boxplot(lengths)
plt.ylabel('텍스트 길이 (tokens)')
plt.title('텍스트 길이 상자 그림')
plt.tight_layout()
plt.show()
stats = {
'min': min(lengths), 'max': max(lengths), 'mean': np.mean(lengths),
'median': np.median(lengths), 'std': np.std(lengths)
}
return stats
def check_special_tokens(self, texts):
special_tokens = ['<|endoftext|>', '<|startoftext|>', '[PAD]', '[CLS]', '[SEP]', '[MASK]']
token_counts = {token: 0 for token in special_tokens}
for text in texts:
tokens = self.tokenizer.encode(text)
token_text = self.tokenizer.decode(tokens)
for special_token in special_tokens:
if special_token in token_text:
token_counts[special_token] += 1
return token_counts
def check_dataset_quality(dataset, tokenizer):
checker = DataQualityChecker(tokenizer)
texts = [example['text'] for example in dataset['train']]
stats = checker.analyze_text_length(texts)
print("텍스트 길이 통계:", stats)
special_counts = checker.check_special_tokens(texts)
print("특수 토큰 통계:", special_counts)
return stats, special_counts
5. 모델 선택 및 설정
5.1 적합한 8B 모델 고르기
| 모델 이름 | 개발사 | 주요 특징 | 추천 활용 분야 |
|---|---|---|---|
| LLaMA-2-7B | Meta | 오픈소스, 상업용 가능, 성능 우수 | 일반 대화, 추론 |
| ChatGLM2-6B | 칭화대/지푸 AI | 중영 이중 언어, 효율적 추론 | 중국어 작업, 대화 |
| Baichuan2-7B | 바이촨 AI | 중영 이중 언어, 코드 능력 | 중국어 애플리케이션, 프로그래밍 |
| Qwen-7B | 알리바나 통이 | 중영 이중 언어, 멀티모달 지원 | 일반 작업, 연구 |
| InternLM-7B | 상하이 AI 연구소 | 중영 이중 언어, 수학 추론 | 교육, 추론 작업 |
5.2 모델 로딩 및 LoRA 설정
model_loader.py를 생성합니다.
import torch
from transformers import (
AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig, TrainingArguments
)
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
import os
class ModelLoader:
def __init__(self, model_name, use_4bit=True, use_8bit=False):
self.model_name = model_name
self.use_4bit = use_4bit
self.use_8bit = use_8bit
def load_tokenizer(self):
tokenizer = AutoTokenizer.from_pretrained(self.model_name)
if tokenizer.pad_token is None:
tokenizer.pad_token = tokenizer.eos_token
return tokenizer
def load_model(self):
bnb_config = None
if self.use_4bit:
bnb_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_use_double_quant=True,
bnb_4bit_quant_type="nf4",
bnb_4bit_compute_dtype=torch.bfloat16
)
elif self.use_8bit:
bnb_config = BitsAndBytesConfig(load_in_8bit=True)
model = AutoModelForCausalLM.from_pretrained(
self.model_name,
quantization_config=bnb_config,
device_map="auto",
trust_remote_code=True,
torch_dtype=torch.bfloat16
)
model = prepare_model_for_kbit_training(model)
return model
def setup_lora(self, model, lora_config=None):
if lora_config is None:
lora_config = LoraConfig(
r=16,
lora_alpha=32,
target_modules=["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"],
lora_dropout=0.05,
bias="none",
task_type="CAUSAL_LM",
)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
return model
def load_and_setup_model(model_name="meta-llama/Llama-2-7b-hf"):
loader = ModelLoader(model_name, use_4bit=True)
tokenizer = loader.load_tokenizer()
model = loader.load_model()
model = loader.setup_lora(model)
return model, tokenizer
if __name__ == "__main__":
model, tokenizer = load_and_setup_model("facebook/opt-1.3b")
print("모델 로드 성공!")
input_text = "오늘 날씨가 좋아서,"
inputs = tokenizer(input_text, return_tensors="pt")
with torch.no_grad():
outputs = model.generate(
inputs.input_ids, max_length=50, temperature=0.7,
do_sample=True, pad_token_id=tokenizer.pad_token_id
)
generated_text = tokenizer.decode(outputs[0], skip_special_tokens=True)
print(f"생성된 텍스트: {generated_text}")
5.3 LoRA 파라미터 상세 설정
lora_config_optimizer.py를 생성하여 다양한 설정 프리셋을 관리합니다.
from peft import LoraConfig
from dataclasses import dataclass
from typing import List
@dataclass
class LoRAConfigPreset:
name: str
r: int
lora_alpha: int
lora_dropout: float
target_modules: List[str]
description: str
class LoRAConfigOptimizer:
PRESETS = {
"efficient": LoRAConfigPreset(
name="efficient", r=8, lora_alpha=16, lora_dropout=0.05,
target_modules=["q_proj", "v_proj"],
description="가장 적은 파라미터, 자원 제한 환경에 적합"
),
"balanced": LoRAConfigPreset(
name="balanced", r=16, lora_alpha=32, lora_dropout=0.05,
target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
description="성능과 효율성의 균형"
),
"performance": LoRAConfigPreset(
name="performance", r=32, lora_alpha=64, lora_dropout=0.1,
target_modules=["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"],
description="고성능, 전체 미세 조정에 근접"
),
"custom": LoRAConfigPreset(
name="custom", r=16, lora_alpha=32, lora_dropout=0.05,
target_modules=["q_proj", "v_proj"],
description="사용자 정의 설정"
)
}
@classmethod
def get_preset(cls, preset_name: str) -> LoraConfig:
if preset_name not in cls.PRESETS:
raise ValueError(f"알 수 없는 프리셋: {preset_name}")
preset = cls.PRESETS[preset_name]
return LoraConfig(
r=preset.r, lora_alpha=preset.lora_alpha, lora_dropout=preset.lora_dropout,
target_modules=preset.target_modules, bias="none", task_type="CAUSAL_LM",
)
@classmethod
def create_custom_config(cls, r=16, lora_alpha=32, lora_dropout=0.05, target_modules=None):
if target_modules is None:
target_modules = ["q_proj", "v_proj"]
return LoraConfig(
r=r, lora_alpha=lora_alpha, lora_dropout=lora_dropout,
target_modules=target_modules, bias="none", task_type="CAUSAL_LM",
)
@classmethod
def analyze_model_architecture(cls, model):
model_modules = []
for name, module in model.named_modules():
if any(layer in name for layer in ['q_proj', 'k_proj', 'v_proj', 'o_proj',
'gate_proj', 'up_proj', 'down_proj',
'query', 'key', 'value', 'dense']):
model_modules.append(name)
return model_modules
def demonstrate_lora_configs():
print("사용 가능한 LoRA 설정 프리셋:")
for preset_name, preset in LoRAConfigOptimizer.PRESETS.items():
print(f"\n{preset_name}: {preset.description}")
print(f" 순위(r): {preset.r}, Alpha: {preset.lora_alpha}, Dropout: {preset.lora_dropout}")
print(f" 대상 모듈: {preset.target_modules}")
if __name__ == "__main__":
demonstrate_lora_configs()
6. 학습 프로세스 구현
6.1 학습 파라미터 설정
training_config.py를 생성합니다.
from transformers import TrainingArguments
from dataclasses import dataclass
from typing import Optional
import os
import torch
@dataclass
class TrainingConfigPreset:
name: str
per_device_train_batch_size: int
gradient_accumulation_steps: int
warmup_steps: int
max_steps: int
learning_rate: float
logging_steps: int
save_steps: int
eval_steps: Optional[int]
description: str
class TrainingConfigManager:
PRESETS = {
"quick": TrainingConfigPreset(
name="quick", per_device_train_batch_size=4, gradient_accumulation_steps=4,
warmup_steps=100, max_steps=1000, learning_rate=2e-4,
logging_steps=10, save_steps=500, eval_steps=200,
description="빠른 학습, 디버깅 및 실험에 적합"
),
"standard": TrainingConfigPreset(
name="standard", per_device_train_batch_size=8, gradient_accumulation_steps=4,
warmup_steps=200, max_steps=5000, learning_rate=1e-4,
logging_steps=50, save_steps=1000, eval_steps=500,
description="표준 학습, 성능과 시간의 균형"
),
"thorough": TrainingConfigPreset(
name="thorough", per_device_train_batch_size=4, gradient_accumulation_steps=8,
warmup_steps=500, max_steps=20000, learning_rate=5e-5,
logging_steps=100, save_steps=2000, eval_steps=1000,
description="철저한 학습, 최고 성능 추구"
)
}
@classmethod
def create_training_args(cls, output_dir, preset_name="standard", **overrides) -> TrainingArguments:
if preset_name not in cls.PRESETS:
raise ValueError(f"알 수 없는 프리셋: {preset_name}")
preset = cls.PRESETS[preset_name]
config_dict = {
'per_device_train_batch_size': preset.per_device_train_batch_size,
'gradient_accumulation_steps': preset.gradient_accumulation_steps,
'warmup_steps': preset.warmup_steps,
'max_steps': preset.max_steps,
'learning_rate': preset.learning_rate,
'logging_steps': preset.logging_steps,
'save_steps': preset.save_steps,
'eval_steps': preset.eval_steps,
}
config_dict.update(overrides)
training_args = TrainingArguments(
output_dir=output_dir,
overwrite_output_dir=True,
per_device_train_batch_size=config_dict['per_device_train_batch_size'],
per_device_eval_batch_size=config_dict['per_device_train_batch_size'],
gradient_accumulation_steps=config_dict['gradient_accumulation_steps'],
warmup_steps=config_dict['warmup_steps'],
max_steps=config_dict['max_steps'],
learning_rate=config_dict['learning_rate'],
logging_steps=config_dict['logging_steps'],
save_steps=config_dict['save_steps'],
eval_steps=config_dict['eval_steps'],
evaluation_strategy="steps" if config_dict['eval_steps'] else "no",
save_strategy="steps",
load_best_model_at_end=True if config_dict['eval_steps'] else False,
report_to="tensorboard",
run_name=os.path.basename(output_dir),
fp16=False,
bf16=torch.cuda.is_bf16_supported(),
remove_unused_columns=False,
dataloader_pin_memory=False,
)
return training_args
def setup_training(output_dir="./lora-output", preset="standard"):
config_manager = TrainingConfigManager()
training_args = config_manager.create_training_args(
output_dir=output_dir, preset_name=preset, learning_rate=1.5e-4, max_steps=8000
)
return training_args
6.2 트레이너(Trainer) 구현
trainer.py를 생성하여 전체 학습 파이프라인을 구축합니다.
import torch
from transformers import Trainer, DataCollatorForLanguageModeling
from peft import LoraConfig, get_peft_model
import os, json
from datetime import datetime
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class LoRATrainer:
def __init__(self, model, tokenizer, training_args):
self.model = model
self.tokenizer = tokenizer
self.training_args = training_args
self.data_collator = DataCollatorForLanguageModeling(
tokenizer=tokenizer, mlm=False,
)
def setup_trainer(self, train_dataset, eval_dataset=None, callbacks=None):
trainer = Trainer(
model=self.model, args=self.training_args,
train_dataset=train_dataset, eval_dataset=eval_dataset,
data_collator=self.data_collator, tokenizer=self.tokenizer,
callbacks=callbacks,
)
return trainer
def save_training_config(self, output_dir, lora_config, dataset_info):
config = {
"training_date": datetime.now().isoformat(),
"model_name": self.model.config._name_or_path,
"lora_config": {
"r": lora_config.r, "lora_alpha": lora_config.lora_alpha,
"lora_dropout": lora_config.lora_dropout,
"target_modules": lora_config.target_modules,
},
"training_args": {
"learning_rate": self.training_args.learning_rate,
"per_device_train_batch_size": self.training_args.per_device_train_batch_size,
"gradient_accumulation_steps": self.training_args.gradient_accumulation_steps,
"max_steps": self.training_args.max_steps,
"warmup_steps": self.training_args.warmup_steps,
},
"dataset_info": dataset_info,
}
config_path = os.path.join(output_dir, "training_config.json")
with open(config_path, 'w', encoding='utf-8') as f:
json.dump(config, f, indent=2, ensure_ascii=False)
logger.info(f"학습 설정이 {config_path}에 저장되었습니다.")
def run_training(model_name, dataset, output_dir, lora_preset="balanced", training_preset="standard"):
os.makedirs(output_dir, exist_ok=True)
from model_loader import ModelLoader
loader = ModelLoader(model_name, use_4bit=True)
tokenizer = loader.load_tokenizer()
model = loader.load_model()
from lora_config_optimizer import LoRAConfigOptimizer
lora_config = LoRAConfigOptimizer.get_preset(lora_preset)
model = loader.setup_lora(model, lora_config)
from training_config import TrainingConfigManager
training_args = TrainingConfigManager.create_training_args(
output_dir=output_dir, preset_name=training_preset
)
trainer = LoRATrainer(model, tokenizer, training_args)
train_dataset = dataset["train"]
eval_dataset = dataset["test"] if "test" in dataset else None
from transformers import EarlyStoppingCallback
callbacks = [EarlyStoppingCallback(early_stopping_patience=3)]
trainer_instance = trainer.setup_trainer(
train_dataset=train_dataset, eval_dataset=eval_dataset, callbacks=callbacks
)
dataset_info = {"train_size": len(train_dataset), "eval_size": len(eval_dataset) if eval_dataset else 0}
trainer.save_training_config(output_dir, lora_config, dataset_info)
logger.info("학습 시작...")
train_result = trainer_instance.train()
trainer_instance.save_model()
trainer_instance.save_state()
logger.info(f"학습 완료! 모델이 {output_dir}에 저장되었습니다.")
return train_result
if __name__ == "__main__":
from data_preprocessing import DataPreprocessor
preprocessor = DataPreprocessor("facebook/opt-1.3b")
sample_data = [
{"instruction": "인사말 해줘", "output": "안녕하세요! 만나서 반갑습니다!"},
{"instruction": "인공지능 소개해줘", "output": "인공지능은 컴퓨터 과학의 한 분야입니다..."}
]
dataset = preprocessor.prepare_dataset(sample_data)
run_training(
model_name="facebook/opt-1.3b", dataset=dataset,
output_dir="./test-training", lora_preset="balanced", training_preset="quick"
)
6.3 학습 모니터링 및 시각화
training_monitor.py를 생성합니다.
import matplotlib.pyplot as plt
import pandas as pd
import json
import os
from tensorboard.backend.event_processing.event_accumulator import EventAccumulator
class TrainingMonitor:
def __init__(self, log_dir):
self.log_dir = log_dir
def parse_tensorboard_logs(self):
event_acc = EventAccumulator(self.log_dir)
event_acc.Reload()
scalars = {}
for tag in event_acc.Tags()['scalars']:
events = event_acc.Scalars(tag)
scalars[tag] = [(e.step, e.value) for e in events]
return scalars
def plot_training_metrics(self, output_path=None):
scalars = self.parse_tensorboard_logs()
if not scalars:
print("학습 로그를 찾을 수 없습니다.")
return
fig, axes = plt.subplots(2, 2, figsize=(15, 10))
axes = axes.flatten()
metrics_to_plot = ['train/loss', 'eval/loss', 'train/learning_rate', 'eval/accuracy']
for i, metric in enumerate(metrics_to_plot):
if metric in scalars:
steps, values = zip(*scalars[metric])
axes[i].plot(steps, values, label=metric)
axes[i].set_title(metric)
axes[i].set_xlabel('Step')
axes[i].set_ylabel('Value')
axes[i].grid(True)
axes[i].legend()
plt.tight_layout()
if output_path:
plt.savefig(output_path, dpi=300, bbox_inches='tight')
print(f"학습 차트가 {output_path}에 저장되었습니다.")
plt.show()
def generate_training_report(self, output_path):
scalars = self.parse_tensorboard_logs()
if not scalars:
print("학습 로그를 찾을 수 없습니다.")
return
report = {"final_metrics": {}, "training_summary": {}}
for metric, values in scalars.items():
if values:
report["final_metrics"][metric] = values[-1][1]
if 'train/loss' in scalars:
train_losses = [v for _, v in scalars['train/loss']]
report["training_summary"]["min_train_loss"] = min(train_losses)
report["training_summary"]["final_train_loss"] = train_losses[-1]
report["training_summary"]["total_steps"] = len(train_losses)
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(report, f, indent=2, ensure_ascii=False)
print(f"학습 보고서가 {output_path}에 저장되었습니다.")
return report
def monitor_training_run(log_dir):
monitor = TrainingMonitor(log_dir)
monitor.plot_training_metrics("./training_metrics.png")
report = monitor.generate_training_report("./training_report.json")
return report
7. 모델 평가 및 테스트
7.1 자동 평가 프로세스
model_evaluator.py를 생성합니다.
import torch
from transformers import pipeline
import evaluate
import pandas as pd
import numpy as np
from tqdm import tqdm
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ModelEvaluator:
def __init__(self, model, tokenizer, device="cuda"):
self.model = model
self.tokenizer = tokenizer
self.device = device
self.bleu_metric = evaluate.load("bleu")
self.rouge_metric = evaluate.load("rouge")
def evaluate_perplexity(self, texts):
logger.info("perplexity 계산 중...")
perplexities = []
for text in tqdm(texts):
try:
inputs = self.tokenizer(text, return_tensors="pt", truncation=True, max_length=1024)
with torch.no_grad():
outputs = self.model(**inputs, labels=inputs["input_ids"])
loss = outputs.loss
perplexity = torch.exp(loss).item()
perplexities.append(perplexity)
except Exception as e:
logger.warning(f"perplexity 계산 중 오류: {e}")
continue
return {
"mean_perplexity": np.mean(perplexities),
"std_perplexity": np.std(perplexities),
"min_perplexity": np.min(perplexities),
"max_perplexity": np.max(perplexities)
}
def evaluate_generation_quality(self, prompts, references):
logger.info("생성 품질 평가 중...")
generated_texts = []
for prompt in tqdm(prompts):
inputs = self.tokenizer(prompt, return_tensors="pt", truncation=True, max_length=512)
with torch.no_grad():
outputs = self.model.generate(
inputs.input_ids.to(self.device),
max_length=len(inputs.input_ids[0]) + 50,
temperature=0.7, do_sample=True,
pad_token_id=self.tokenizer.pad_token_id
)
generated = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
generated_texts.append(generated[len(prompt):])
bleu_results = self.bleu_metric.compute(predictions=generated_texts, references=[[ref] for ref in references])
rouge_results = self.rouge_metric.compute(predictions=generated_texts, references=references)
return {"bleu": bleu_results, "rouge": rouge_results, "generated_texts": generated_texts}
def run_comprehensive_evaluation(self, test_dataset, num_samples=100):
logger.info("종합 평가 시작...")
results = {}
if len(test_dataset) > num_samples:
indices = np.random.choice(len(test_dataset), num_samples, replace=False)
sample_texts = [test_dataset[i]["text"] for i in indices]
else:
sample_texts = [example["text"] for example in test_dataset]
perplexity_results = self.evaluate_perplexity(sample_texts)
results["perplexity"] = perplexity_results
if "instruction" in test_dataset.features:
prompts = []
references = []
for example in test_dataset[:min(50, len(test_dataset))]:
if "input" in example and example["input"]:
prompt = f"### Instruction:\n{example['instruction']}\n\n### Input:\n{example['input']}\n\n### Response:\n"
else:
prompt = f"### Instruction:\n{example['instruction']}\n\n### Response:\n"
prompts.append(prompt)
references.append(example["output"])
generation_results = self.evaluate_generation_quality(prompts, references)
results["generation_quality"] = generation_results
logger.info("평가 완료!")
return results
def compare_models(base_model, lora_model, tokenizer, test_data):
base_evaluator = ModelEvaluator(base_model, tokenizer)
lora_evaluator = ModelEvaluator(lora_model, tokenizer)
logger.info("기본 모델 평가 중...")
base_results = base_evaluator.run_comprehensive_evaluation(test_data)
logger.info("LoRA 모델 평가 중...")
lora_results = lora_evaluator.run_comprehensive_evaluation(test_data)
comparison = {"base_model": base_results, "lora_model": lora_results, "improvement": {}}
if "perplexity" in base_results and "perplexity" in lora_results:
base_ppl = base_results["perplexity"]["mean_perplexity"]
lora_ppl = lora_results["perplexity"]["mean_perplexity"]
comparison["improvement"]["perplexity"] = {
"absolute": base_ppl - lora_ppl,
"relative": (base_ppl - lora_ppl) / base_ppl * 100
}
return comparison
7.2 수동 평가 인터페이스
human_evaluation.py를 생성하여 Gradio 기반 평가 도구를 만듭니다.
import gradio as gr
import pandas as pd
import torch
from typing import List, Dict
class HumanEvaluationInterface:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
self.evaluation_results = []
def generate_response(self, instruction, input_text=""):
if input_text:
prompt = f"### Instruction:\n{instruction}\n\n### Input:\n{input_text}\n\n### Response:\n"
else:
prompt = f"### Instruction:\n{instruction}\n\n### Response:\n"
inputs = self.tokenizer(prompt, return_tensors="pt", truncation=True, max_length=512)
with torch.no_grad():
outputs = self.model.generate(
inputs.input_ids,
max_length=len(inputs.input_ids[0]) + 200,
temperature=0.7, do_sample=True,
pad_token_id=self.tokenizer.pad_token_id
)
response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
return response[len(prompt):]
def create_interface(self):
def evaluate_model(instruction, input_text, rating, comments):
response = self.generate_response(instruction, input_text)
evaluation = {
"instruction": instruction, "input": input_text,
"response": response, "rating": rating,
"comments": comments, "timestamp": pd.Timestamp.now().isoformat()
}
self.evaluation_results.append(evaluation)
return response, f"평가 저장 완료! 총 평가 수: {len(self.evaluation_results)}"
def export_results():
if not self.evaluation_results:
return "내보낼 평가 데이터가 없습니다."
df = pd.DataFrame(self.evaluation_results)
filename = f"human_evaluation_{pd.Timestamp.now().strftime('%Y%m%d_%H%M%S')}.csv"
df.to_csv(filename, index=False, encoding='utf-8-sig')
return f"평가 결과가 {filename}에 내보내졌습니다."
with gr.Blocks(title="LoRA Model Human Evaluation") as interface:
gr.Markdown("# LoRA Model Human Evaluation Interface")
with gr.Row():
with gr.Column():
instruction = gr.Textbox(label="Instruction", placeholder="Enter instruction...", lines=3)
input_text = gr.Textbox(label="Input (optional)", placeholder="Enter context...", lines=3)
generate_btn = gr.Button("Generate Response")
with gr.Column():
response = gr.Textbox(label="Model Response", lines=5, interactive=False)
rating = gr.Slider(minimum=1, maximum=5, value=3, label="Rating (1-5)")
comments = gr.Textbox(label="Comments", placeholder="Enter comments...", lines=3)
evaluate_btn = gr.Button("Submit Evaluation")
export_btn = gr.Button("Export Results")
status = gr.Textbox(label="Status", interactive=False)
generate_btn.click(fn=self.generate_response, inputs=[instruction, input_text], outputs=[response])
evaluate_btn.click(fn=evaluate_model, inputs=[instruction, input_text, rating, comments], outputs=[response, status])
export_btn.click(fn=export_results, outputs=[status])
return interface
def launch_human_evaluation(model, tokenizer, share=True):
interface = HumanEvaluationInterface(model, tokenizer)
gradio_interface = interface.create_interface()
gradio_interface.launch(share=share)
8. 모델 배포 및 추론
8.1 최적화된 추론 스크립트
optimized_inference.py를 생성합니다.
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import PeftModel
import time
from typing import List, Dict, Any
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class OptimizedLoRAInference:
def __init__(self, base_model_path, lora_adapter_path, device="cuda"):
self.device = device
self.base_model_path = base_model_path
self.lora_adapter_path = lora_adapter_path
self.load_config()
self.load_model()
def load_config(self):
self.generation_config = {
"max_new_tokens": 256, "temperature": 0.7, "top_p": 0.9,
"top_k": 50, "repetition_penalty": 1.1, "do_sample": True,
}
def load_model(self):
logger.info("토크나이저 로딩 중...")
self.tokenizer = AutoTokenizer.from_pretrained(self.base_model_path)
if self.tokenizer.pad_token is None:
self.tokenizer.pad_token = self.tokenizer.eos_token
logger.info("기본 모델 로딩 중...")
self.base_model = AutoModelForCausalLM.from_pretrained(
self.base_model_path, torch_dtype=torch.float16,
device_map="auto", trust_remote_code=True
)
logger.info("LoRA 어댑터 로딩 중...")
self.model = PeftModel.from_pretrained(
self.base_model, self.lora_adapter_path, torch_dtype=torch.float16
)
logger.info("LoRA 가중치 병합 중...")
self.model = self.model.merge_and_unload()
self.model.eval()
logger.info("모델 로딩 완료!")
def format_prompt(self, instruction, input_text=None, context=None):
if context:
prompt = f"### Context:\n{context}\n\n"
else:
prompt = ""
prompt += f"### Instruction:\n{instruction}\n\n"
if input_text:
prompt += f"### Input:\n{input_text}\n\n"
prompt += "### Response:\n"
return prompt
def generate(self, prompt, **generation_kwargs):
config = self.generation_config.copy()
config.update(generation_kwargs)
inputs = self.tokenizer(prompt, return_tensors="pt", truncation=True, max_length=1024).to(self.device)
start_time = time.time()
with torch.no_grad():
outputs = self.model.generate(
**inputs, **config,
pad_token_id=self.tokenizer.pad_token_id,
eos_token_id=self.tokenizer.eos_token_id
)
generation_time = time.time() - start_time
generated_tokens = outputs[0][inputs.input_ids.shape[1]:]
generated_text = self.tokenizer.decode(generated_tokens, skip_special_tokens=True)
stats = {
"generation_time": generation_time, "total_tokens": len(outputs[0]),
"new_tokens": len(generated_tokens),
"tokens_per_second": len(generated_tokens) / generation_time
}
return generated_text, stats
def batch_generate(self, prompts, **generation_kwargs):
results = []
for prompt in prompts:
response, stats = self.generate(prompt, **generation_kwargs)
results.append({"prompt": prompt, "response": response, "stats": stats})
return results
def demonstrate_inference():
inference = OptimizedLoRAInference(
base_model_path="facebook/opt-1.3b", lora_adapter_path="./lora-output"
)
prompts = ["Explain machine learning concepts", "Write Python function for fibonacci", "Summarize climate change impacts"]
for prompt in prompts:
formatted_prompt = inference.format_prompt(prompt)
response, stats = inference.generate(formatted_prompt)
print(f"Prompt: {prompt}\nResponse: {response}\nStats: {stats}\n" + "-"*50)
if __name__ == "__main__":
demonstrate_inference()
8.2 API 서비스 배포
api_deployment.py를 생성하여 FastAPI 기반 서버를 만듭니다.
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uvicorn
import logging
from optimized_inference import OptimizedLoRAInference
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class GenerationRequest(BaseModel):
instruction: str
input_text: Optional[str] = None
context: Optional[str] = None
max_new_tokens: Optional[int] = 256
temperature: Optional[float] = 0.7
top_p: Optional[float] = 0.9
class GenerationResponse(BaseModel):
instruction: str
input_text: Optional[str]
response: str
generation_stats: dict
class BatchGenerationRequest(BaseModel):
requests: List[GenerationRequest]
class BatchGenerationResponse(BaseModel):
responses: List[GenerationResponse]
app = FastAPI(title="LoRA Model API Service", description="LoRA fine-tuned LLM deployment API", version="1.0.0")
model_inference = None
@app.on_event("startup")
async def startup_event():
global model_inference
try:
logger.info("모델 로딩 중...")
model_inference = OptimizedLoRAInference(
base_model_path="facebook/opt-1.3b", lora_adapter_path="./lora-output"
)
logger.info("모델 로딩 완료!")
except Exception as e:
logger.error(f"모델 로딩 실패: {e}")
raise e
@app.get("/")
async def root():
return {"message": "LoRA Model API Service Running", "status": "healthy"}
@app.get("/health")
async def health_check():
return {"status": "healthy"}
@app.post("/generate", response_model=GenerationResponse)
async def generate_text(request: GenerationRequest):
try:
prompt = model_inference.format_prompt(
instruction=request.instruction, input_text=request.input_text, context=request.context
)
generation_kwargs = {"max_new_tokens": request.max_new_tokens, "temperature": request.temperature, "top_p": request.top_p}
response, stats = model_inference.generate(prompt, **generation_kwargs)
return GenerationResponse(
instruction=request.instruction, input_text=request.input_text,
response=response, generation_stats=stats
)
except Exception as e:
logger.error(f"생성 실패: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/batch_generate", response_model=BatchGenerationResponse)
async def batch_generate_text(batch_request: BatchGenerationRequest):
try:
responses = []
for request in batch_request.requests:
prompt = model_inference.format_prompt(
instruction=request.instruction, input_text=request.input_text, context=request.context
)
generation_kwargs = {"max_new_tokens": request.max_new_tokens, "temperature": request.temperature, "top_p": request.top_p}
response, stats = model_inference.generate(prompt, **generation_kwargs)
responses.append(GenerationResponse(
instruction=request.instruction, input_text=request.input_text,
response=response, generation_stats=stats
))
return BatchGenerationResponse(responses=responses)
except Exception as e:
logger.error(f"배치 생성 실패: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/model_info")
async def get_model_info():
return {"base_model": model_inference.base_model_path, "lora_adapter": model_inference.lora_adapter_path, "device": model_inference.device}
def run_api_server(host="0.0.0.0", port=8000, reload=False):
uvicorn.run(app, host=host, port=port, reload=reload, log_level="info")
if __name__ == "__main__":
run_api_server()
9. 실전 예제: 한국어 대화 모델 미세 조정
9.1 데이터 준비
korean_data_processor.py를 생성하여 한국어 데이터를 처리합니다.
import json
import re
from datasets import Dataset
from transformers import AutoTokenizer
class KoreanDataProcessor:
def __init__(self, model_name):
self.tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
if self.tokenizer.pad_token is None:
self.tokenizer.pad_token = self.tokenizer.eos_token
def clean_text(self, text):
text = re.sub(r'\s+', ' ', text)
text = re.sub(r'[^\w\s\uac00-\ud7af\u3130-\u318f\u3000-\u303f。，！？；：""''（）《》]', '', text)
return text.strip()
def load_alpaca_format_data(self, file_path):
with open(file_path, 'r', encoding='utf-8') as f:
data = json.load(f)
processed_data = []
for item in data:
instruction = self.clean_text(item.get('instruction', ''))
input_text = self.clean_text(item.get('input', ''))
output = self.clean_text(item.get('output', ''))
if input_text:
text = f"명령: {instruction}\n입력: {input_text}\n답변: {output}"
else:
text = f"명령: {instruction}\n답변: {output}"
processed_data.append({"text": text})
return processed_data
def load_conversation_data(self, file_path):
with open(file_path, 'r', encoding='utf-8') as f:
data = json.load(f)
processed_data = []
for conversation in data:
if 'conversations' in conversation:
dialog_text = ""
for turn in conversation['conversations']:
if turn['role'] == 'user':
dialog_text += f"사용자: {self.clean_text(turn['content'])}\n"
else:
dialog_text += f"어시스턴트: {self.clean_text(turn['content'])}\n"
processed_data.append({"text": dialog_text.strip()})
return processed_data
def create_dataset(self, data_files, data_type="alpaca"):
all_data = []
for file_path in data_files:
if data_type == "alpaca":
data = self.load_alpaca_format_data(file_path)
else:
data = self.load_conversation_data(file_path)
all_data.extend(data)
dataset = Dataset.from_list(all_data)
def tokenize_function(examples):
return self.tokenizer(
examples["text"], truncation=True, padding=False,
max_length=1024, return_overflowing_tokens=False,
)
tokenized_dataset = dataset.map(
tokenize_function, batched=True, remove_columns=dataset.column_names,
)
return tokenized_dataset.train_test_split(test_size=0.1, seed=42)
def prepare_korean_dataset():
processor = KoreanDataProcessor("facebook/opt-1.3b")
data_files = ["./korean_alpaca_data.json"]
dataset = processor.create_dataset(data_files, data_type="alpaca")
print(f"훈련 세트 크기: {len(dataset['train'])}")
print(f"테스트 세트 크기: {len(dataset['test'])}")
return dataset
9.2 모델 학습 및 평가
korean_lora_training.py를 생성합니다.
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments, BitsAndBytesConfig
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from transformers import Trainer, DataCollatorForLanguageModeling
import os
from korean_data_processor import KoreanDataProcessor
def setup_korean_model_training(model_name, dataset, output_dir):
tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
if tokenizer.pad_token is None:
tokenizer.pad_token = tokenizer.eos_token
bnb_config = None
if torch.cuda.is_available():
bnb_config = BitsAndBytesConfig(
load_in_4bit=True, bnb_4bit_use_double_quant=True,
bnb_4bit_quant_type="nf4", bnb_4bit_compute_dtype=torch.bfloat16
)
model = AutoModelForCausalLM.from_pretrained(
model_name, quantization_config=bnb_config,
device_map="auto", trust_remote_code=True, torch_dtype=torch.bfloat16
)
model = prepare_model_for_kbit_training(model)
lora_config = LoraConfig(
r=16, lora_alpha=32,
target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
lora_dropout=0.05, bias="none", task_type="CAUSAL_LM",
)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
training_args = TrainingArguments(
output_dir=output_dir, per_device_train_batch_size=4,
per_device_eval_batch_size=4, gradient_accumulation_steps=4,
warmup_steps=200, max_steps=5000, learning_rate=2e-4,
logging_steps=50, save_steps=1000, eval_steps=500,
evaluation_strategy="steps", save_strategy="steps",
load_best_model_at_end=True, report_to="tensorboard",
fp16=False, bf16=torch.cuda.is_bf16_supported(),
remove_unused_columns=False,
)
data_collator = DataCollatorForLanguageModeling(tokenizer=tokenizer, mlm=False)
trainer = Trainer(
model=model, args=training_args, train_dataset=dataset["train"],
eval_dataset=dataset["test"], data_collator=data_collator, tokenizer=tokenizer,
)
return trainer, model, tokenizer
def run_korean_training():
processor = KoreanDataProcessor("facebook/opt-1.3b")
dataset = processor.create_dataset(["./korean_data.json"], "alpaca")
trainer, model, tokenizer = setup_korean_model_training(
"facebook/opt-1.3b", dataset, "./korean-lora-output"
)
print("한국어 LoRA 모델 학습 시작...")
trainer.train()
trainer.save_model()
print("학습 완료! 모델이 저장되었습니다.")
return model, tokenizer
if __name__ == "__main__":
run_korean_training()
10. 핵심 요약 및 실용적인 조언
이 가이드에서는 8B 파라미터 LLM에 LoRA를 적용하는 전 과정을 다루었습니다. 핵심 내용을 요약하면 다음과 같습니다.
- LoRA의 강점: 학습 파라미터를 100~1000배 줄이면서도 성능을 유지하고, 추론 시 추가 지연 시간이 없습니다.
- 설정 최적화: 순위(r), alpha 파라미터, 적용 대상 모듈을 신중히 선택하는 것이 중요합니다.
- 학습 전략: 적절한 학습률 스케줄링, 그래디언트 누적, 평가 전략이 성공적인 학습의 핵심입니다.
- 자원 관리: 4-bit/8-bit 양자화 기술을 활용하면 소비자용 GPU에서도 대형 모델을 학습할 수 있습니다.
실무자를 위한 조언을 드리자면,
- 작게 시작하라: 처음에는 소규모 데이터와 간단한 설정으로 실험해 보십시오.
- 점진적으로 최적화하라: 평가 결과를 바탕으로 하이퍼파라미터를 조금씩 조정해 나가십시오.
- 학습을 모니터링하라: 학습 손실과 평가 지표를 주의 깊게 관찰하며 필요시 전략을 조정하십시오.
- 충분히 테스트하라: 배포 전에 기능 테스트와 성능 평가를 철저히 수행하십시오.