작성완료
LoRA 기반 파인튜닝 방법
LoRA 기반 파인튜닝 방법에는 라이브러리 설치, 학습 데이터 준비, 기본 설정, 디바이스 설정, 모델 및 토크나이저 로딩, LoRA 설정 및 적용, 데이터 로딩 및 전처리, DataLoader 및 옵티마이저 설정, 수동 학습 루프, 학습 결과 저장 과정이 포함됩니다. 또한 FastAPI를 이용한 모델 배포 방법과 서버 권장 사양도 제시됩니다.
✅ **LoRA **파인튜닝
1. 사전 준비
1.1. 라이브러리 설치
우선 필요한 라이브러리들을 설치합니다:
pip install transformers datasets peft accelerate
1.2. 학습 데이터 준비
{"prompt":"운동놀이학교는 어떤 프로그램을 운영하나요?","completion":"운동놀이학교는 9가지 분석 프로그램과 5가지 밸런스 육성 프로그램을 운영합니다."}
{"prompt":"운동놀이학교에서는 어떤 분석을 하나요?","completion":"운동놀이학교는 기질, 감각, 신체발달 분석 등 총 9가지 분석 프로그램을 운영합니다."}
{"prompt":"운동놀이학교는 누가 지도하나요?","completion":"운동놀이학교는 검증된 체육교육 전문가가 지도하는 교육 시스템입니다."}
2. 기본 설정
BASE_MODEL = "EleutherAI/polyglot-ko-1.3b"
OUTPUT_DIR = "./model"
TRAIN_DATA = "data.jsonl"
NUM_EPOCHS = 3
BATCH_SIZE = 1
LEARNING_RATE = 2e-4
3. 디바이스 설정
device = torch.device("mps" if torch.backends.mps.is_available() else "cpu")
print(f"✅ Using device: {device}")
- Mac 유저는 MPS를, 나머지는 CPU나 GPU 환경을 자동으로 사용합니다.
4. 모델 및 토크나이저 로딩
tokenizer = AutoTokenizer.from_pretrained(BASE_MODEL)
tokenizer.pad_token = tokenizer.pad_token or tokenizer.eos_token
base_model = AutoModelForCausalLM.from_pretrained(BASE_MODEL)
base_model.gradient_checkpointing_enable()
base_model.config.use_cache = False
5. LoRA 설정 및 적용
from peft import get_peft_model, LoraConfig, TaskType
peft_config = LoraConfig(
task_type=TaskType.CAUSAL_LM,
inference_mode=False,
r=16,
lora_alpha=32,
lora_dropout=0.05,
target_modules=["query_key_value", "dense"]
)
model = get_peft_model(base_model, peft_config)
model.print_trainable_parameters()
model.to(device)
- LoRA는 전체 모델이 아닌 일부 레이어에만 학습 가능한 Adapter를 넣는 경량화 파인튜닝 기법입니다.
- target_modules로 LoRA를 적용할 레이어를 지정합니다.
6. 데이터 로딩 및 전처리
dataset = load_dataset("json", data_files=TRAIN_DATA, split="train")
dataset = dataset.filter(lambda x: x.get("prompt") and x.get("completion"))
6.1. 전처리 함수
def tokenize(example):
prompt = example["prompt"].strip()
completion = example["completion"].strip()
full_text = f"[질문] {prompt}\n[답변] {completion}"
tokenized = tokenizer(
full_text,
truncation=True,
padding=False,
max_length=768
)
tokenized["labels"] = tokenized["input_ids"]
return tokenized
tokenized_dataset = dataset.map(
tokenize, remove_columns=dataset.column_names)
7. DataLoader 및 Collator
from transformers import DataCollatorForLanguageModeling
data_collator = DataCollatorForLanguageModeling(
tokenizer=tokenizer, mlm=False, pad_to_multiple_of=8)
dataloader = DataLoader(
tokenized_dataset, batch_size=BATCH_SIZE, collate_fn=data_collator)
- mlm=False: GPT류 모델은 마스킹이 아닌 auto-regressive 방식
8. 옵티마이저 & 러닝레이트 스케줄러
optimizer = AdamW(model.parameters(), lr=LEARNING_RATE)
lr_scheduler = get_scheduler(
name="linear",
optimizer=optimizer,
num_warmup_steps=0,
num_training_steps=len(dataloader) * NUM_EPOCHS
)
9. 수동 학습 루프
model.train()
model.enable_input_require_grads()
total_steps = len(dataloader) * NUM_EPOCHS
for epoch in range(NUM_EPOCHS):
print(f"🔁 Epoch {epoch + 1}/{NUM_EPOCHS}")
total_loss = 0.0
for step, batch in enumerate(dataloader):
batch = {k: v.to(device) for k, v in batch.items()}
outputs = model(**batch)
loss = outputs.loss
loss.backward()
optimizer.step()
lr_scheduler.step()
optimizer.zero_grad()
total_loss += loss.item()
current_step = epoch * len(dataloader) + step + 1
progress = (current_step / total_steps) * 100
if step % 10 == 0:
print(
f"Step {current_step}/{total_steps} | "
f"Loss: {loss.item():.4f} | "
f"Progress: {progress:.2f}%"
)
avg_loss = total_loss / len(dataloader)
print(f"✅ Epoch {epoch + 1} complete | Avg Loss: {avg_loss:.4f}")
10. 학습 결과 저장
model.save_pretrained(OUTPUT_DIR)
tokenizer.save_pretrained(OUTPUT_DIR)
print("🎉 학습 완료 및 저장 완료")
11. 전체 코드
import torch
from torch.utils.data import DataLoader
from torch.optim import AdamW
from transformers import (
AutoTokenizer,
AutoModelForCausalLM,
DataCollatorForLanguageModeling,
get_scheduler
)
from datasets import load_dataset
from peft import get_peft_model, LoraConfig, TaskType
# ✅ 설정
BASE_MODEL = "EleutherAI/polyglot-ko-1.3b"
OUTPUT_DIR = "./model"
TRAIN_DATA = "data.jsonl"
NUM_EPOCHS = 3
BATCH_SIZE = 1
LEARNING_RATE = 2e-4
def main():
# ✅ 디바이스 설정 (MPS 우선, 없으면 CPU)
device = torch.device(
"mps" if torch.backends.mps.is_available() else "cpu")
print(f"✅ Using device: {device}")
# ✅ 토크나이저 및 모델 로드
tokenizer = AutoTokenizer.from_pretrained(BASE_MODEL)
tokenizer.pad_token = tokenizer.pad_token or tokenizer.eos_token
base_model = AutoModelForCausalLM.from_pretrained(BASE_MODEL)
base_model.gradient_checkpointing_enable()
base_model.config.use_cache = False
# ✅ LoRA 설정 및 적용
peft_config = LoraConfig(
task_type=TaskType.CAUSAL_LM,
inference_mode=False,
r=16,
lora_alpha=32,
lora_dropout=0.05,
target_modules=["query_key_value", "dense"]
)
model = get_peft_model(base_model, peft_config)
model.print_trainable_parameters()
model.to(device)
# ✅ 데이터 로드 및 필터링
dataset = load_dataset("json", data_files=TRAIN_DATA, split="train")
dataset = dataset.filter(lambda x: x.get("prompt") and x.get("completion"))
# ✅ 전처리 함수
def tokenize(example):
prompt = example["prompt"].strip()
completion = example["completion"].strip()
full_text = f"[질문] {prompt}\n[답변] {completion}"
tokenized = tokenizer(
full_text,
truncation=True,
padding=False,
max_length=768
)
tokenized["labels"] = tokenized["input_ids"]
return tokenized
# ✅ 전처리 적용
tokenized_dataset = dataset.map(
tokenize, remove_columns=dataset.column_names)
# ✅ 데이터 collator 및 DataLoader
data_collator = DataCollatorForLanguageModeling(
tokenizer=tokenizer, mlm=False, pad_to_multiple_of=8)
dataloader = DataLoader(
tokenized_dataset, batch_size=BATCH_SIZE, collate_fn=data_collator)
# ✅ 옵티마이저 & 스케줄러
optimizer = AdamW(model.parameters(), lr=LEARNING_RATE)
lr_scheduler = get_scheduler(
name="linear",
optimizer=optimizer,
num_warmup_steps=0,
num_training_steps=len(dataloader) * NUM_EPOCHS
)
total_steps = len(dataloader) * NUM_EPOCHS # 👈 전체 스텝 수 미리 계산
# ✅ 수동 학습 루프
model.train()
model.enable_input_require_grads()
for epoch in range(NUM_EPOCHS):
print(f"🔁 Epoch {epoch + 1}/{NUM_EPOCHS}")
total_loss = 0.0
for step, batch in enumerate(dataloader):
batch = {k: v.to(device) for k, v in batch.items()}
outputs = model(**batch)
loss = outputs.loss
loss.backward()
optimizer.step()
lr_scheduler.step()
optimizer.zero_grad()
total_loss += loss.item()
# ✅ 진행률 계산 및 출력
current_step = epoch * len(dataloader) + step + 1
progress = (current_step / total_steps) * 100
if step % 10 == 0:
print(
f"Step {current_step}/{total_steps} | "
f"Loss: {loss.item():.4f} | "
f"Progress: {progress:.2f}%"
)
avg_loss = total_loss / len(dataloader)
print(f"✅ Epoch {epoch + 1} complete | Avg Loss: {avg_loss:.4f}")
# ✅ 저장
model.save_pretrained(OUTPUT_DIR)
tokenizer.save_pretrained(OUTPUT_DIR)
print("🎉 학습 완료 및 저장 완료")
if __name__ == "__main__":
main()
✅ FastAPI로 LoRA 파인튜닝 모델 배포
1. 사전 준비
필요한 라이브러리를 먼저 설치하세요
pip install fastapi uvicorn torch transformers peft
전체 코드 구조 요약
📁 프로젝트 구조
.
├── server.py 👈 서버 코드
├── model/ 👈 LoRA 파인튜닝 모델 저장 폴더
└── history.log 👈 질문/답변 로그 기록 파일 (자동 생성됨)
2. 모델 불러오기
def load_model():
peft_config = PeftConfig.from_pretrained(ADAPTER_PATH)
base_model = AutoModelForCausalLM.from_pretrained(
peft_config.base_model_name_or_path)
tokenizer = AutoTokenizer.from_pretrained(ADAPTER_PATH)
model = PeftModel.from_pretrained(base_model, ADAPTER_PATH)
return tokenizer, model
- PeftConfig로 base 모델 경로를 추출하고,
- AutoModelForCausalLM과 PeftModel을 조합하여 최종 모델을 완성합니다.
- Tokenizer는 LoRA 학습 시 저장된 경로(ADAPTER_PATH)에서 불러옵니다.
3. 디바이스 설정
device = torch.device(
"mps" if torch.backends.mps.is_available()
else "cuda" if torch.cuda.is_available()
else "cpu"
)
- Mac 사용자는 M1/M2 칩 기반 MPS 지원
- GPU가 있다면 cuda, 없으면 cpu로 fallback
4. 요청 모델 정의
class PromptRequest(BaseModel):
prompt: str
- POST 요청에서 prompt 필드만 받습니다.
- 요청 예시 :
{ "prompt": "LoRA란 무엇인가요?" }
5. 텍스트 생성 흐름
5.1. 입력 포맷 구성
formatted_prompt = f"[질문] {prompt.strip()}\n[답변]"
- LoRA 학습 시 사용한 포맷과 동일하게 구성해야 자연스러운 답변을 유도할 수 있습니다.
5.2. 토크나이징
inputs = tokenizer(formatted_prompt, return_tensors="pt").to(device)
inputs.pop("token_type_ids", None)
- token_type_ids는 일부 모델에서는 필요 없으므로 제거해줍니다.
5.3. 텍스트 생성
outputs = model.generate(
**inputs,
do_sample=False,
repetition_penalty=1.1,
no_repeat_ngram_size=2,
max_new_tokens=MAX_OUTPUT_LEN,
eos_token_id=tokenizer.eos_token_id,
pad_token_id=tokenizer.pad_token_id,
)
- do_sample=False: 항상 가장 가능성 높은 답변 선택 (안정적인 응답)
- repetition_penalty=1.1: 동일 문장 반복 방지
- max_new_tokens: 생성되는 토큰 수 제한
5.4. 응답 정제 및 문장 자르기
decoded = tokenizer.decode(outputs[0], skip_special_tokens=True)
decoded = decoded.split("[답변]")[-1].strip()
decoded = re.sub(r'(\d+:)+', '', decoded).strip()
- [답변] 이후만 추출
- 숫자나 시각 표현을 제거 (예: 12:34 같은 표현)
result = re.split(r'(다\.|요\.|입니다\.|합니다\.|해요\.|예요\.|이예요\.)', decoded)
if len(result) >= 4:
result = ''.join(result[:4]) # 최대 2문장 출력
else:
result = decoded.strip()
- 자연스러운 문장 단위로 끊어서 간결하게 응답하도록 처리
6. 서버 실행
uvicorn.run(
SERVER["NAME"] + ":app",
host="0.0.0.0",
port=SERVER["PORT"],
reload=True,
workers=1
)
- server.py로 실행하면 "server:app"이 등록됩니다.
- 외부에서 접속하려면 host="0.0.0.0" 유지하세요.
7. 전체 코드
import uvicorn
import torch
import re
import logging
from fastapi import FastAPI, HTTPException
from transformers import AutoTokenizer, AutoModelForCausalLM
from peft import PeftModel, PeftConfig
from pydantic import BaseModel
from datetime import datetime
logging.basicConfig(level=logging.INFO)
# 설정
SERVER = {"NAME": "server", "PORT": 8000}
MAX_INPUT_LEN = 256
MAX_OUTPUT_LEN = 128
ADAPTER_PATH = "./model"
# ✅ 디바이스 설정
device = torch.device(
"mps" if torch.backends.mps.is_available()
else "cuda" if torch.cuda.is_available()
else "cpu"
)
# 서버 인스턴스 생성
app = FastAPI()
class PromptRequest(BaseModel):
prompt: str
def load_model():
peft_config = PeftConfig.from_pretrained(ADAPTER_PATH)
base_model = AutoModelForCausalLM.from_pretrained(
peft_config.base_model_name_or_path)
tokenizer = AutoTokenizer.from_pretrained(ADAPTER_PATH)
model = PeftModel.from_pretrained(base_model, ADAPTER_PATH)
return tokenizer, model
tokenizer, model = load_model()
model.to(device)
model.eval()
def log(type: str, txt: str):
now = datetime.now()
date_time = now.strftime("%Y-%m-%d %H:%M:%S")
with open("history.log", "a", encoding="utf-8") as file:
file.write("[" + date_time + " " + type + "]\n" + txt)
@app.post("/api/chat")
async def generate_text(payload: PromptRequest):
try:
# 1. 클라이언트로부터 프롬프트 수신
prompt = payload.prompt
logging.info(f"질문 : {prompt}")
log("질문", prompt + "\n")
# 2. 형식 정리
formatted_prompt = f"[질문] {prompt.strip()}\n[답변]"
# 3. 입력 토크나이즈
inputs = tokenizer(formatted_prompt, return_tensors="pt").to(device)
inputs.pop("token_type_ids", None)
# 4. 텍스트 생성
outputs = model.generate(
**inputs,
do_sample=False,
repetition_penalty=1.1,
no_repeat_ngram_size=2,
max_new_tokens=MAX_OUTPUT_LEN,
eos_token_id=tokenizer.eos_token_id,
pad_token_id=tokenizer.pad_token_id,
)
# 5. 결과 디코딩
decoded = tokenizer.decode(outputs[0], skip_special_tokens=True)
decoded = decoded.split("[답변]")[-1].strip()
decoded = re.sub(r'(\d+:)+', '', decoded).strip()
# 6. 첫 번째 완성된 문장 추출
result = re.split(r'(다\.|요\.|입니다\.|합니다\.|해요\.|예요\.|이예요\.)', decoded)
# 문장 자르되 최소 2문장 이상일 때만 자름
if len(result) >= 4:
result = ''.join(result[:4])
else:
result = decoded.strip()
logging.info(f"답변 : {result}")
log("답변", result + "\n\n")
return {"result": result}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# ✅ 서버 실행
if __name__ == "__main__":
uvicorn.run(
SERVER["NAME"] + ":app",
host="0.0.0.0",
port=SERVER["PORT"],
reload=True,
workers=1
)
8. 목적별 시스템 권장 사양
🎯 파인튜닝용 서버 (학습용)
수천~수만 개의 데이터로 LoRA 파인튜닝을 여러 번 반복할 경우
■ 운영체제:
- Ubuntu 22.04 LTS
■ GPU:
- NVIDIA A6000 (48GB) 또는 RTX 3090 (24GB) 이상
- polyglot-ko-1.3b는 약 13억 파라미터이므로 VRAM 16~24GB 이상 필요
- 학습 배치 사이즈를 높게 잡거나 추후 더 큰 모델을 쓰려면 A100(40/80GB)도 고려 가능
■ CPU:
- 최소: AMD Ryzen 7 / Intel i7 10세대 이상
- 권장: AMD EPYC / Intel Xeon (멀티 코어 16개 이상)
■ RAM:
- 최소: 32GB
- 권장: 64GB 이상 (Dataset 로딩, 토큰화 시 병목 방지)
■ 저장장치 (SSD):
- 최소: 512GB NVMe SSD
- 권장: 1TB 이상
- HuggingFace 모델 캐시, 학습 로그, 체크포인트 저장 등에 사용
■ CUDA 및 드라이버:
- CUDA 11.8 이상
- PyTorch 버전과 호환되도록 설치 필요 (torch==2.x 권장)
🧠 추론/서비스용 서버 (FastAPI 서버 운영)
학습된 LoRA 모델을 FastAPI로 배포할 목적
■ 운영체제:
- Ubuntu 22.04 LTS
■ GPU:
- 최소: NVIDIA T4 (16GB)
- 권장: RTX A4000 / 3060 12GB 이상
- 요청량이 많거나 속도를 중요시하면 24GB급 필요
■ CPU:
- 8코어 이상
■ RAM:
- 16~32GB
■ 저장장치:
- 256~512GB SSD
**📌 **요약
| 항목 | 파인튜닝용 | 추론용 |
|---|---|---|
| GPU | A6000 / 3090 / A100 | T4 / A4000 / 3060 이상 |
| RAM | 64GB 이상 | 16~32GB |
| 저장공간 | 1TB SSD | 256~512GB SSD |
| OS | Ubuntu 22.04 | Ubuntu 22.04 |
| 기타 | CUDA 11.8, PyTorch 2.x | FastAPI, Uvicorn, Tokenizer 저장 필요 |
💸 견적
| 항목 | 사양 | 금액 (원) | 참고 링크 |
|---|---|---|---|
| GPU | RTX A4000 16GB | 1,560,000 | |
| RAM | DDR4 16GB | 120,000 | |
| 저장공간 (SSD) | NVMe 512GB | 150,000 | |
| CPU | Ryzen 9 5900X 12코어 | 400,000 |