9장: 학습-평가-배포 자동화 사이클

수동 파인튜닝 파이프라인의 문제점:
 
  데이터 업데이트
       |
  "누군가" 학습 스크립트 수동 실행
       |
  결과 확인 ("이거 좋은 건가?")
       |
  수동으로 모델 서버에 배포
       |
  문제 발생 시 어떤 버전이었는지 모름
 
  문제:
    - 재현 불가능
    - 실수 발생 가능
    - 배포까지 시간이 오래 걸림
    - 롤백이 어려움

전체 자동화 아키텍처:
 
  트리거
    |-- 데이터 변경 (새 데이터 추가)
    |-- 코드 변경 (학습 설정 수정)
    |-- 스케줄 (주기적 재학습)
    |-- 수동 (버튼 클릭)
       |
  학습 파이프라인
    |-- 데이터 검증
    |-- 전처리
    |-- 모델 학습
    |-- 체크포인트 저장
       |
  평가 파이프라인
    |-- 자동 메트릭 평가
    |-- 벤치마크 테스트
    |-- 회귀 테스트
    |-- 품질 게이트 (통과/실패)
       |
  배포 파이프라인 (평가 통과 시)
    |-- 모델 레지스트리 등록
    |-- 스테이징 배포
    |-- 스모크 테스트
    |-- 프로덕션 배포
    |-- 모니터링 활성화
       |
  알림
    |-- Slack/이메일 통보
    |-- 대시보드 업데이트

# .github/workflows/fine-tuning-pipeline.yml
name: Fine-Tuning Pipeline
 
on:
  # 데이터 변경 시 자동 트리거
  push:
    paths:
      - 'data/**'
      - 'configs/**'
    branches:
      - main
 
  # 수동 트리거
  workflow_dispatch:
    inputs:
      config_file:
        description: 'Training config file'
        required: true
        default: 'configs/experiment_v1.yaml'
      run_evaluation:
        description: 'Run evaluation after training'
        required: true
        default: 'true'
 
  # 주간 스케줄 (매주 일요일 자정)
  schedule:
    - cron: '0 0 * * 0'
 
env:
  MODEL_REGISTRY: s3://my-bucket/model-registry
  HF_TOKEN: ${{ secrets.HF_TOKEN }}
  WANDB_API_KEY: ${{ secrets.WANDB_API_KEY }}

jobs:
  data-validation:
    runs-on: ubuntu-latest
    outputs:
      data_valid: ${{ steps.validate.outputs.valid }}
    steps:
      - uses: actions/checkout@v4
 
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
 
      - name: Install dependencies
        run: pip install -r requirements.txt
 
      - name: Validate training data
        id: validate
        run: |
          python scripts/validate_data.py \
            --input data/train.jsonl \
            --min-examples 100 \
            --max-examples 100000 \
            --check-format \
            --check-duplicates
 
  train:
    needs: data-validation
    if: needs.data-validation.outputs.data_valid == 'true'
    runs-on: [self-hosted, gpu]
    outputs:
      model_path: ${{ steps.train.outputs.model_path }}
      run_id: ${{ steps.train.outputs.run_id }}
    steps:
      - uses: actions/checkout@v4
 
      - name: Setup Python environment
        run: |
          python -m venv venv
          source venv/bin/activate
          pip install -r requirements.txt
 
      - name: Run training
        id: train
        run: |
          source venv/bin/activate
          python train.py \
            --config configs/experiment_v1.yaml \
            --output-dir ./output
          echo "model_path=./output/best-adapter" >> "$GITHUB_OUTPUT"
          echo "run_id=$(cat ./output/run_id.txt)" >> "$GITHUB_OUTPUT"
 
      - name: Upload model artifact
        uses: actions/upload-artifact@v4
        with:
          name: lora-adapter
          path: ./output/best-adapter/
          retention-days: 30

  evaluate:
    needs: train
    runs-on: [self-hosted, gpu]
    outputs:
      eval_passed: ${{ steps.quality-gate.outputs.passed }}
    steps:
      - uses: actions/checkout@v4
 
      - name: Download model artifact
        uses: actions/download-artifact@v4
        with:
          name: lora-adapter
          path: ./adapter/
 
      - name: Run evaluation
        id: evaluate
        run: |
          source venv/bin/activate
          python evaluate.py \
            --model-path ./adapter \
            --test-data data/test.jsonl \
            --output eval_results.json
 
      - name: Quality gate check
        id: quality-gate
        run: |
          source venv/bin/activate
          python scripts/quality_gate.py \
            --results eval_results.json \
            --min-rouge-l 0.70 \
            --min-bertscore 0.80 \
            --max-eval-loss 0.60
 
      - name: Upload evaluation results
        uses: actions/upload-artifact@v4
        with:
          name: eval-results
          path: eval_results.json

  deploy:
    needs: evaluate
    if: needs.evaluate.outputs.eval_passed == 'true'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
 
      - name: Download model artifact
        uses: actions/download-artifact@v4
        with:
          name: lora-adapter
          path: ./adapter/
 
      - name: Register model
        run: |
          python scripts/register_model.py \
            --adapter-path ./adapter \
            --registry-uri $MODEL_REGISTRY \
            --stage staging
 
      - name: Deploy to staging
        run: |
          python scripts/deploy.py \
            --environment staging \
            --model-path $MODEL_REGISTRY/latest
 
      - name: Smoke test
        run: |
          python scripts/smoke_test.py \
            --endpoint https://staging-api.example.com/v1/chat
 
      - name: Promote to production
        if: success()
        run: |
          python scripts/deploy.py \
            --environment production \
            --model-path $MODEL_REGISTRY/latest
          python scripts/register_model.py \
            --stage production
 
      - name: Notify
        if: always()
        run: |
          python scripts/notify.py \
            --channel fine-tuning \
            --status ${{ job.status }}

# scripts/quality_gate.py
import json
import sys
from argparse import ArgumentParser
 
 
def check_quality_gate(
    results_path: str,
    min_rouge_l: float = 0.70,
    min_bertscore: float = 0.80,
    max_eval_loss: float = 0.60,
) -> bool:
    """품질 게이트 검사"""
    with open(results_path) as f:
        results = json.load(f)
 
    checks = []
 
    # 1. ROUGE-L 점수 확인
    rouge_l = results.get("rouge_l", 0)
    rouge_pass = rouge_l >= min_rouge_l
    checks.append({
        "name": "ROUGE-L",
        "value": rouge_l,
        "threshold": min_rouge_l,
        "passed": rouge_pass,
    })
 
    # 2. BERTScore 확인
    bertscore = results.get("bertscore_f1", 0)
    bert_pass = bertscore >= min_bertscore
    checks.append({
        "name": "BERTScore F1",
        "value": bertscore,
        "threshold": min_bertscore,
        "passed": bert_pass,
    })
 
    # 3. 검증 손실 확인
    eval_loss = results.get("eval_loss", 1.0)
    loss_pass = eval_loss <= max_eval_loss
    checks.append({
        "name": "Eval Loss",
        "value": eval_loss,
        "threshold": max_eval_loss,
        "passed": loss_pass,
    })
 
    # 결과 출력
    print("품질 게이트 결과:")
    print("=" * 50)
    all_passed = True
    for check in checks:
        status = "PASS" if check["passed"] else "FAIL"
        print(
            "  " + check["name"] + ": "
            + str(round(check["value"], 4))
            + " (기준: " + str(check["threshold"]) + ") "
            + "[" + status + "]"
        )
        if not check["passed"]:
            all_passed = False
 
    overall = "PASS" if all_passed else "FAIL"
    print("=" * 50)
    print("종합: " + overall)
 
    return all_passed
 
 
if __name__ == "__main__":
    parser = ArgumentParser()
    parser.add_argument("--results", required=True)
    parser.add_argument("--min-rouge-l", type=float, default=0.70)
    parser.add_argument("--min-bertscore", type=float, default=0.80)
    parser.add_argument("--max-eval-loss", type=float, default=0.60)
    args = parser.parse_args()
 
    passed = check_quality_gate(
        args.results,
        args.min_rouge_l,
        args.min_bertscore,
        args.max_eval_loss,
    )
 
    if not passed:
        print("품질 게이트 실패. 배포가 중단됩니다.")
        sys.exit(1)
 
    # GitHub Actions 출력
    with open(sys.environ.get("GITHUB_OUTPUT", "/dev/null"), "a") as f:
        f.write("passed=true\n")

def regression_test(
    new_results: dict,
    prod_results: dict,
    tolerance: float = 0.02
) -> bool:
    """프로덕션 모델 대비 회귀 테스트"""
    regression_detected = False
 
    for metric in ["rouge_l", "bertscore_f1"]:
        new_val = new_results.get(metric, 0)
        prod_val = prod_results.get(metric, 0)
        diff = new_val - prod_val
 
        status = "OK"
        if diff < -tolerance:
            status = "REGRESSION"
            regression_detected = True
        elif diff > tolerance:
            status = "IMPROVED"
 
        print(
            metric + ": " + str(round(prod_val, 4))
            + " -> " + str(round(new_val, 4))
            + " (" + status + ")"
        )
 
    return not regression_detected

# scripts/data_pipeline.py
import json
from pathlib import Path
 
 
class DataPipeline:
    """학습 데이터 자동 처리 파이프라인"""
 
    def __init__(self, raw_dir: str, processed_dir: str):
        self.raw_dir = Path(raw_dir)
        self.processed_dir = Path(processed_dir)
 
    def run(self):
        """전체 데이터 파이프라인 실행"""
        print("데이터 파이프라인 시작")
 
        # 1. 원시 데이터 수집
        raw_data = self.collect_raw_data()
        print("원시 데이터: " + str(len(raw_data)) + "개")
 
        # 2. 형식 표준화
        normalized = self.normalize(raw_data)
 
        # 3. 정제 및 필터링
        cleaned = self.clean_and_filter(normalized)
        print("정제 후: " + str(len(cleaned)) + "개")
 
        # 4. 중복 제거
        deduped = self.deduplicate(cleaned)
        print("중복 제거 후: " + str(len(deduped)) + "개")
 
        # 5. 품질 검증
        validated = self.validate_quality(deduped)
        print("검증 통과: " + str(len(validated)) + "개")
 
        # 6. 분할 및 저장
        self.split_and_save(validated)
        print("데이터 파이프라인 완료")
 
    def collect_raw_data(self) -> list:
        all_data = []
        for file in self.raw_dir.glob("*.jsonl"):
            with open(file) as f:
                for line in f:
                    all_data.append(json.loads(line))
        return all_data
 
    def normalize(self, data: list) -> list:
        # 형식 표준화 (2장 참조)
        return data
 
    def clean_and_filter(self, data: list) -> list:
        # 정제 및 필터링 (3장 참조)
        return [d for d in data if self.passes_filter(d)]
 
    def passes_filter(self, example: dict) -> bool:
        messages = example.get("messages", [])
        if not messages:
            return False
        assistant_msgs = [
            m for m in messages if m["role"] == "assistant"
        ]
        if not assistant_msgs:
            return False
        response = assistant_msgs[0]["content"]
        return len(response) >= 50
 
    def deduplicate(self, data: list) -> list:
        seen = set()
        unique = []
        for d in data:
            key = json.dumps(d["messages"], sort_keys=True)
            import hashlib
            h = hashlib.md5(key.encode()).hexdigest()
            if h not in seen:
                seen.add(h)
                unique.append(d)
        return unique
 
    def validate_quality(self, data: list) -> list:
        return data  # 규칙 기반 + LLM 검증 적용
 
    def split_and_save(self, data: list):
        from sklearn.model_selection import train_test_split
 
        train_val, test = train_test_split(
            data, test_size=0.1, random_state=42
        )
        train, val = train_test_split(
            train_val, test_size=0.11, random_state=42
        )
 
        self.processed_dir.mkdir(parents=True, exist_ok=True)
        for split_name, split_data in [
            ("train", train), ("val", val), ("test", test)
        ]:
            path = self.processed_dir / (split_name + ".jsonl")
            with open(path, "w") as f:
                for d in split_data:
                    f.write(json.dumps(d, ensure_ascii=False) + "\n")
            print(
                "  " + split_name + ": "
                + str(len(split_data)) + "개 저장됨"
            )

# scripts/rollback.py
class RollbackManager:
    """모델 롤백 관리"""
 
    def __init__(self, registry, deployer):
        self.registry = registry
        self.deployer = deployer
 
    def rollback_to_previous(self, model_name: str):
        """이전 프로덕션 버전으로 롤백"""
        versions = self.registry.list_versions(model_name)
 
        # archived 상태의 가장 최근 버전 찾기
        archived = [
            v for v in versions if v["stage"] == "archived"
        ]
        if not archived:
            raise ValueError("롤백할 이전 버전이 없습니다.")
 
        prev_version = archived[-1]
        current = self.registry.get_production_model(model_name)
 
        print(
            "롤백: v" + str(current["version"])
            + " -> v" + str(prev_version["version"])
        )
 
        # 1. 이전 버전을 프로덕션으로 승격
        self.registry.promote_to_production(
            model_name, prev_version["version"]
        )
 
        # 2. 배포
        self.deployer.deploy(
            model_path=prev_version["model_path"],
            environment="production"
        )
 
        # 3. 현재 버전의 문제 기록
        self.registry.add_note(
            model_name,
            current["version"],
            "프로덕션 롤백됨. 문제 조사 필요."
        )
 
        print("롤백 완료")

# scripts/notify.py
import json
import os
from urllib.request import urlopen, Request
 
 
def send_slack_notification(
    channel: str,
    status: str,
    details: dict
):
    """Slack 웹훅을 통한 알림 전송"""
    webhook_url = os.environ["SLACK_WEBHOOK_URL"]
 
    color = "#36a64f" if status == "success" else "#ff0000"
    status_text = "성공" if status == "success" else "실패"
 
    blocks = [
        {
            "type": "header",
            "text": {
                "type": "plain_text",
                "text": "파인튜닝 파이프라인 " + status_text
            }
        },
        {
            "type": "section",
            "fields": [
                {
                    "type": "mrkdwn",
                    "text": "*모델:*\n" + details.get(
                        "model_name", "N/A"
                    )
                },
                {
                    "type": "mrkdwn",
                    "text": "*버전:*\n" + details.get(
                        "version", "N/A"
                    )
                },
            ]
        }
    ]
 
    if "metrics" in details:
        metrics_text = "\n".join(
            k + ": " + str(round(v, 4))
            for k, v in details["metrics"].items()
        )
        blocks.append({
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": "*평가 결과:*\n" + metrics_text
            }
        })
 
    payload = json.dumps({
        "channel": channel,
        "attachments": [{"color": color, "blocks": blocks}]
    }).encode()
 
    req = Request(webhook_url, data=payload)
    req.add_header("Content-Type", "application/json")
    urlopen(req)

자동화 파이프라인 체크리스트:
 
  학습 자동화:
    - 트리거 조건 정의 (데이터/코드 변경, 스케줄)
    - 데이터 검증 자동화
    - 재현 가능한 학습 설정
    - GPU 자원 자동 할당/해제
 
  평가 자동화:
    - 자동 메트릭 계산
    - 벤치마크 테스트 자동 실행
    - 품질 게이트 정의
    - 회귀 테스트 자동 실행
 
  배포 자동화:
    - 모델 레지스트리 자동 등록
    - 스테이징 자동 배포
    - 스모크 테스트 자동 실행
    - 프로덕션 배포 (수동 승인 또는 자동)
    - 롤백 스크립트 준비
 
  모니터링:
    - 파이프라인 상태 알림 (Slack/이메일)
    - 학습 메트릭 대시보드
    - 프로덕션 모델 성능 모니터링
    - 비용 추적