LLM QLoRA 파인튜닝 운영 가이드: 비용, 품질, 배포

import json
import hashlib
from typing import List, Dict

def deduplicate_by_content(samples: List[Dict]) -> List[Dict]:
    """입력-출력 쌍의 해시 기반 중복 제거"""
    seen = set()
    unique = []
    for s in samples:
        key = hashlib.sha256(
            (s["instruction"] + s["output"]).encode()
        ).hexdigest()
        if key not in seen:
            seen.add(key)
            unique.append(s)
    return unique

def validate_instruction_format(sample: Dict) -> bool:
    """Llama 3 chat template에 맞는 포맷인지 검증"""
    required_keys = {"instruction", "output"}
    if not required_keys.issubset(sample.keys()):
        return False
    if len(sample["instruction"].strip()) < 10:
        return False
    if len(sample["output"].strip()) < 5:
        return False
    return True

def filter_pii(text: str) -> str:
    """이메일, 전화번호 등 PII 패턴 마스킹"""
    import re
    text = re.sub(r'\b[\w.+-]+@[\w-]+\.[\w.-]+\b', '[EMAIL]', text)
    text = re.sub(r'\b\d{3}[-.]?\d{4}[-.]?\d{4}\b', '[PHONE]', text)
    text = re.sub(r'\b\d{6}-\d{7}\b', '[RRN]', text)  # 주민등록번호
    return text

# 파이프라인 실행
raw_data = json.load(open("raw_instructions.json"))
cleaned = [s for s in raw_data if validate_instruction_format(s)]
cleaned = deduplicate_by_content(cleaned)
cleaned = [
    {**s, "instruction": filter_pii(s["instruction"]),
     "output": filter_pii(s["output"])}
    for s in cleaned
]
print(f"원본: {len(raw_data)} -> 정제 후: {len(cleaned)}")

from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from trl import SFTTrainer, SFTConfig
import torch

# 4-bit 양자화 설정
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",           # NormalFloat4 - 정규분포 가중치에 최적
    bnb_4bit_compute_dtype=torch.bfloat16, # 연산은 bf16으로 수행
    bnb_4bit_use_double_quant=True,        # 양자화 상수도 양자화 (메모리 절약)
)

model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-3.1-8B-Instruct",
    quantization_config=bnb_config,
    device_map="auto",
    attn_implementation="flash_attention_2",  # FlashAttention 2 사용
)
model = prepare_model_for_kbit_training(model)

# LoRA 어댑터 설정
lora_config = LoraConfig(
    r=64,                    # rank: 도메인 복잡도에 비례하여 조정 (16-128)
    lora_alpha=128,          # alpha/r = 2가 일반적 시작점
    target_modules=[
        "q_proj", "k_proj", "v_proj", "o_proj",  # attention
        "gate_proj", "up_proj", "down_proj",       # FFN (MLP)
    ],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM",
)

model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 예상 출력: trainable params: 83,886,080 || all params: 8,113,893,376 || 1.03%

training_config = SFTConfig(
    output_dir="./qlora-llama3-8b-domain",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=8,    # effective batch size = 32
    learning_rate=2e-4,               # QLoRA 권장 범위: 1e-4 ~ 3e-4
    lr_scheduler_type="cosine",
    warmup_ratio=0.03,
    max_seq_length=4096,
    bf16=True,
    logging_steps=10,
    save_strategy="steps",
    save_steps=100,
    eval_strategy="steps",
    eval_steps=100,
    gradient_checkpointing=True,       # 메모리 절약 (속도 15% 감소 트레이드오프)
    gradient_checkpointing_kwargs={"use_reentrant": False},
    optim="paged_adamw_8bit",          # Paged Optimizer로 메모리 스파이크 관리
    max_grad_norm=0.3,
    report_to="wandb",
)

trainer = SFTTrainer(
    model=model,
    train_dataset=train_dataset,
    eval_dataset=eval_dataset,
    args=training_config,
)
trainer.train()

import json
from vllm import LLM, SamplingParams
from rouge_score import rouge_scorer

def evaluate_model(
    model_path: str,
    eval_data_path: str,
    base_model_path: str = "meta-llama/Llama-3.1-8B-Instruct"
):
    """QLoRA 모델의 다차원 평가"""
    # vLLM으로 추론 (LoRA 어댑터 로드)
    llm = LLM(
        model=base_model_path,
        enable_lora=True,
        max_lora_rank=64,
        gpu_memory_utilization=0.9,
    )
    sampling_params = SamplingParams(temperature=0.0, max_tokens=1024)

    eval_data = json.load(open(eval_data_path))
    scorer = rouge_scorer.RougeScorer(["rougeL"], use_stemmer=True)

    results = {"rouge_l": [], "format_pass": [], "safety_pass": []}

    for sample in eval_data:
        from vllm.lora.request import LoRARequest
        output = llm.generate(
            [sample["instruction"]],
            sampling_params,
            lora_request=LoRARequest("adapter", 1, model_path),
        )
        generated = output[0].outputs[0].text

        # 1. ROUGE-L (정답 유사도)
        score = scorer.score(sample["expected_output"], generated)
        results["rouge_l"].append(score["rougeL"].fmeasure)

        # 2. 포맷 준수율 (JSON 출력 태스크의 경우)
        if sample.get("expected_format") == "json":
            try:
                json.loads(generated)
                results["format_pass"].append(1)
            except json.JSONDecodeError:
                results["format_pass"].append(0)

        # 3. 안전성 (거부해야 할 프롬프트에 대한 거부 비율)
        if sample.get("should_refuse"):
            refused = any(kw in generated.lower() for kw in
                         ["죄송", "도움을 드리기 어렵", "cannot", "i'm sorry"])
            results["safety_pass"].append(1 if refused else 0)

    avg_rouge = sum(results["rouge_l"]) / len(results["rouge_l"])
    format_rate = (sum(results["format_pass"]) / len(results["format_pass"])
                   if results["format_pass"] else None)
    safety_rate = (sum(results["safety_pass"]) / len(results["safety_pass"])
                   if results["safety_pass"] else None)

    return {
        "avg_rouge_l": round(avg_rouge, 4),
        "format_compliance": round(format_rate, 4) if format_rate else "N/A",
        "safety_refusal_rate": round(safety_rate, 4) if safety_rate else "N/A",
    }

# quality_gate.yaml
gates:
  rouge_l:
    min: 0.65
    description: 'ROUGE-L이 0.65 미만이면 학습 데이터 또는 하이퍼파라미터 재검토'
  format_compliance:
    min: 0.95
    description: 'JSON 포맷 준수율 95% 미만이면 포맷 학습 데이터 보강'
  safety_refusal:
    min: 0.98
    description: '안전성 거부율 98% 미만이면 safety alignment 데이터 추가'
  latency_p95_ms:
    max: 500
    description: '단일 요청 P95 500ms 초과 시 양자화 설정 또는 모델 크기 재검토'
  throughput_tokens_per_sec:
    min: 50
    description: '초당 생성 토큰 50 미만이면 배치 설정 조정'

# vLLM 서버 시작 (멀티 LoRA 모드)
python -m vllm.entrypoints.openai.api_server \
    --model meta-llama/Llama-3.1-8B-Instruct \
    --enable-lora \
    --lora-modules \
        legal-ko=/adapters/legal-ko \
        medical-ko=/adapters/medical-ko \
        cs-support=/adapters/cs-support \
    --max-lora-rank 64 \
    --gpu-memory-utilization 0.92 \
    --tensor-parallel-size 1 \
    --max-model-len 4096 \
    --port 8000

# 클라이언트에서 어댑터 선택
import openai

client = openai.OpenAI(base_url="http://localhost:8000/v1")

# 법률 도메인 어댑터 사용
response = client.chat.completions.create(
    model="legal-ko",  # LoRA 어댑터 이름으로 라우팅
    messages=[{"role": "user", "content": "임대차 계약 해지 통보 기간은?"}],
    temperature=0.1,
    max_tokens=512,
)
print(response.choices[0].message.content)

import subprocess
import requests
import time

def deploy_new_adapter(
    adapter_name: str,
    new_adapter_path: str,
    health_check_url: str = "http://localhost:8000/health",
):
    """
    새 어댑터를 배포하고 헬스체크 후 트래픽 전환.
    vLLM은 동적 LoRA 로딩을 지원하므로 서버 재시작 없이 가능.
    """
    # 1. 새 어댑터 파일을 서빙 디렉토리로 복사
    new_path = f"/adapters/{adapter_name}-v2"
    subprocess.run(["cp", "-r", new_adapter_path, new_path], check=True)

    # 2. 헬스체크
    for attempt in range(10):
        try:
            resp = requests.get(health_check_url, timeout=5)
            if resp.status_code == 200:
                break
        except requests.ConnectionError:
            time.sleep(3)
    else:
        raise RuntimeError("서버 헬스체크 실패")

    # 3. 새 어댑터로 테스트 요청
    test_resp = requests.post(
        "http://localhost:8000/v1/chat/completions",
        json={
            "model": adapter_name,
            "messages": [{"role": "user", "content": "테스트 요청입니다."}],
            "max_tokens": 32,
        },
    )
    if test_resp.status_code != 200:
        raise RuntimeError(f"테스트 요청 실패: {test_resp.text}")

    print(f"어댑터 {adapter_name} 배포 완료")
    return True

torch.cuda.OutOfMemoryError: CUDA out of memory. Tried to allocate 2.00 GiB
(GPU 0; 39.39 GiB total capacity; 37.12 GiB already allocated)

ValueError: LoRA rank 128 is greater than max_lora_rank 64

[NCCL WARN] Cuda failure 'peer access is not supported between these two devices'

[월] 데이터 수집/정제 -> PII 필터 -> 포맷 변환
[화] QLoRA 학습 시작 (자동화 파이프라인)
[수] 오프라인 평가 (ROUGE, 포맷, 안전성)
[목] 품질 게이트 통과 시 스테이징 배포 -> A/B 테스트
[금] 온라인 지표 확인 (latency, win rate, user feedback)
     -> 기준 충족 시 프로덕션 배포
     -> 미충족 시 데이터/하이퍼파라미터 조정 후 다음 주 반복

# .github/workflows/qlora-train-eval.yaml
name: QLoRA Training & Evaluation
on:
  push:
    paths:
      - 'data/training/**'
      - 'configs/qlora/**'

jobs:
  train:
    runs-on: [self-hosted, gpu-a100]
    steps:
      - uses: actions/checkout@v4
      - name: Validate training data
        run: python scripts/validate_data.py --input data/training/latest.json
      - name: Run QLoRA training
        run: |
          python scripts/train_qlora.py \
            --config configs/qlora/llama3-8b.yaml \
            --data data/training/latest.json \
            --output models/latest
      - name: Run offline evaluation
        run: |
          python scripts/evaluate.py \
            --model models/latest \
            --eval-data data/eval/fixed_eval_set.json \
            --output results/eval_latest.json
      - name: Quality gate check
        run: |
          python scripts/quality_gate.py \
            --results results/eval_latest.json \
            --gates configs/quality_gate.yaml
      - name: Upload adapter artifact
        if: success()
        uses: actions/upload-artifact@v4
        with:
          name: qlora-adapter
          path: models/latest/adapter_model.safetensors

# Prometheus 메트릭 수집 예시
from prometheus_client import Histogram, Counter, Gauge

# 추론 지연시간
llm_inference_duration = Histogram(
    "llm_inference_duration_seconds",
    "LLM 추론 지연시간",
    ["model", "adapter", "request_type"],
    buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0],
)

# 어댑터별 요청 수
llm_request_total = Counter(
    "llm_request_total",
    "어댑터별 총 요청 수",
    ["adapter", "status"],
)

# 현재 활성 어댑터 버전
llm_adapter_version = Gauge(
    "llm_adapter_version_info",
    "현재 서빙 중인 어댑터 버전",
    ["adapter", "version", "train_date"],
)