10장: CI/CD 통합과 품질 게이트 구축

from dataclasses import dataclass
from enum import Enum
 
 
class GateStatus(Enum):
    PASSED = "passed"
    FAILED = "failed"
    WARNING = "warning"
 
 
@dataclass
class GateCondition:
    """품질 게이트의 개별 조건."""
    name: str
    metric: str
    operator: str       # "gte", "lte", "eq", "delta_gte", "delta_lte"
    threshold: float
    blocking: bool       # True이면 실패 시 배포 차단
    description: str
 
 
@dataclass
class GateResult:
    """품질 게이트 평가 결과."""
    condition: GateCondition
    actual_value: float
    status: GateStatus
    message: str
 
 
class QualityGate:
    """품질 게이트를 정의하고 평가합니다."""
    
    def __init__(self, conditions: list[GateCondition]):
        self.conditions = conditions
    
    def evaluate(
        self,
        current_metrics: dict[str, float],
        baseline_metrics: dict[str, float] | None = None,
    ) -> tuple[GateStatus, list[GateResult]]:
        """모든 게이트 조건을 평가합니다."""
        
        results: list[GateResult] = []
        overall_status = GateStatus.PASSED
        
        for condition in self.conditions:
            result = self._check_condition(condition, current_metrics, baseline_metrics)
            results.append(result)
            
            if result.status == GateStatus.FAILED and condition.blocking:
                overall_status = GateStatus.FAILED
            elif result.status == GateStatus.WARNING and overall_status == GateStatus.PASSED:
                overall_status = GateStatus.WARNING
        
        return overall_status, results
    
    def _check_condition(
        self,
        condition: GateCondition,
        current: dict[str, float],
        baseline: dict[str, float] | None,
    ) -> GateResult:
        """단일 조건을 확인합니다."""
        
        current_value = current.get(condition.metric, 0.0)
        
        if condition.operator.startswith("delta_"):
            if baseline is None:
                return GateResult(
                    condition=condition,
                    actual_value=current_value,
                    status=GateStatus.WARNING,
                    message="베이스라인 없음 - 델타 비교 불가",
                )
            baseline_value = baseline.get(condition.metric, 0.0)
            delta = current_value - baseline_value
            actual = delta
        else:
            actual = current_value
        
        passed = self._compare(actual, condition.operator, condition.threshold)
        
        return GateResult(
            condition=condition,
            actual_value=actual,
            status=GateStatus.PASSED if passed else GateStatus.FAILED,
            message=f"{condition.metric}: {actual:.4f} {'>=<'} {condition.threshold}",
        )
    
    @staticmethod
    def _compare(value: float, operator: str, threshold: float) -> bool:
        op = operator.replace("delta_", "")
        if op == "gte":
            return value >= threshold
        elif op == "lte":
            return value <= threshold
        elif op == "eq":
            return abs(value - threshold) < 1e-6
        return False
 
 
# 실전 품질 게이트 정의
PRODUCTION_GATE = QualityGate(conditions=[
    GateCondition(
        name="최소 정확도",
        metric="accuracy",
        operator="gte",
        threshold=0.85,
        blocking=True,
        description="정확도가 85% 미만이면 배포를 차단합니다.",
    ),
    GateCondition(
        name="정확도 회귀",
        metric="accuracy",
        operator="delta_gte",
        threshold=-0.02,
        blocking=True,
        description="이전 대비 정확도가 2%p 이상 하락하면 배포를 차단합니다.",
    ),
    GateCondition(
        name="PII 노출 없음",
        metric="pii_leak_rate",
        operator="lte",
        threshold=0.0,
        blocking=True,
        description="PII 노출이 하나라도 발생하면 배포를 차단합니다.",
    ),
    GateCondition(
        name="지연시간",
        metric="p99_latency_ms",
        operator="lte",
        threshold=3000,
        blocking=False,
        description="P99 지연시간 3초 초과 시 경고합니다.",
    ),
])

name: LLM Evaluation Pipeline
 
on:
  pull_request:
    paths:
      - 'prompts/**'
      - 'src/llm/**'
      - 'config/model_config.yaml'
 
jobs:
  detect-changes:
    runs-on: ubuntu-latest
    outputs:
      prompts_changed: ${{ steps.changes.outputs.prompts }}
      model_changed: ${{ steps.changes.outputs.model }}
    steps:
      - uses: actions/checkout@v4
      - uses: dorny/paths-filter@v3
        id: changes
        with:
          filters: |
            prompts:
              - 'prompts/**'
            model:
              - 'config/model_config.yaml'
 
  run-evaluation:
    needs: detect-changes
    if: needs.detect-changes.outputs.prompts_changed == 'true' || needs.detect-changes.outputs.model_changed == 'true'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      
      - name: Install dependencies
        run: pip install -r requirements-eval.txt
      
      - name: Load baseline metrics
        id: baseline
        run: |
          python scripts/load_baseline.py \
            --branch ${{ github.base_ref }} \
            --output baseline_metrics.json
      
      - name: Run evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          python scripts/run_eval.py \
            --config eval_config.yaml \
            --output current_metrics.json \
            --golden-dataset data/golden_v1.2.json
      
      - name: Check quality gate
        id: gate
        run: |
          python scripts/check_gate.py \
            --current current_metrics.json \
            --baseline baseline_metrics.json \
            --gate-config quality_gate.yaml \
            --output gate_result.json
      
      - name: Post PR comment
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const result = JSON.parse(fs.readFileSync('gate_result.json', 'utf8'));
            
            let body = `## LLM 평가 결과\n\n`;
            body += `상태: **${result.overall_status}**\n\n`;
            body += `| 조건 | 결과 | 값 | 임계값 |\n`;
            body += `|------|------|-----|--------|\n`;
            
            for (const r of result.details) {
              const icon = r.status === 'passed' ? '[PASS]' : r.status === 'failed' ? '[FAIL]' : '[WARN]';
              body += `| ${r.name} | ${icon} | ${r.actual_value.toFixed(4)} | ${r.threshold} |\n`;
            }
            
            await github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: body
            });
      
      - name: Enforce gate
        if: steps.gate.outputs.status == 'failed'
        run: |
          echo "품질 게이트 실패 - 차단 조건이 충족되지 않았습니다."
          exit 1

name: Prompt Evaluation
 
on:
  pull_request:
    paths:
      - 'prompts/**'
 
jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Run promptfoo evaluation
        uses: promptfoo/promptfoo-action@v1
        with:
          config: promptfooconfig.yaml
          cache: false
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
      
      - name: Comment results
        uses: promptfoo/promptfoo-action/comment@v1
        with:
          github-token: ${{ secrets.GITHUB_TOKEN }}

import json
from pathlib import Path
from datetime import datetime
 
 
class BaselineManager:
    """평가 베이스라인을 관리합니다."""
    
    def __init__(self, storage_dir: str):
        self.storage_dir = Path(storage_dir)
        self.storage_dir.mkdir(parents=True, exist_ok=True)
    
    def save_baseline(
        self,
        metrics: dict[str, float],
        branch: str,
        commit_sha: str,
    ) -> None:
        """현재 메트릭을 베이스라인으로 저장합니다."""
        baseline = {
            "metrics": metrics,
            "branch": branch,
            "commit_sha": commit_sha,
            "timestamp": datetime.now().isoformat(),
        }
        
        # 브랜치별 최신 베이스라인
        path = self.storage_dir / f"{branch}_latest.json"
        with open(path, "w") as f:
            json.dump(baseline, f, indent=2)
        
        # 이력 보관
        history_path = self.storage_dir / f"{branch}_{commit_sha[:8]}.json"
        with open(history_path, "w") as f:
            json.dump(baseline, f, indent=2)
    
    def load_baseline(self, branch: str) -> dict[str, float] | None:
        """브랜치의 최신 베이스라인을 로드합니다."""
        path = self.storage_dir / f"{branch}_latest.json"
        if not path.exists():
            return None
        
        with open(path) as f:
            data = json.load(f)
        return data["metrics"]
    
    def compare_with_baseline(
        self,
        current: dict[str, float],
        branch: str,
    ) -> dict[str, dict]:
        """현재 메트릭을 베이스라인과 비교합니다."""
        baseline = self.load_baseline(branch)
        if baseline is None:
            return {"status": "no_baseline", "details": {}}
        
        comparison = {}
        for metric, value in current.items():
            baseline_value = baseline.get(metric)
            if baseline_value is not None:
                delta = value - baseline_value
                comparison[metric] = {
                    "current": value,
                    "baseline": baseline_value,
                    "delta": delta,
                    "relative_change": delta / baseline_value if baseline_value != 0 else 0,
                    "regression": delta < -0.01,  # 1%p 이상 하락
                }
        
        return {"status": "compared", "details": comparison}

import subprocess
from pathlib import Path
 
 
class PromptChangeDetector:
    """프롬프트 변경을 감지하고 영향 범위를 분석합니다."""
    
    def __init__(self, prompt_dir: str = "prompts/"):
        self.prompt_dir = Path(prompt_dir)
        # 프롬프트 파일과 관련 평가 태스크의 매핑
        self.prompt_task_mapping: dict[str, list[str]] = {
            "prompts/customer_support.txt": ["support_accuracy", "support_tone"],
            "prompts/summarization.txt": ["summary_quality", "summary_faithfulness"],
            "prompts/system_common.txt": ["all"],  # 공통 프롬프트는 전체 평가
        }
    
    def get_changed_prompts(self, base_ref: str = "main") -> list[str]:
        """기준 브랜치 대비 변경된 프롬프트 파일 목록을 반환합니다."""
        result = subprocess.run(
            ["git", "diff", "--name-only", base_ref, "--", str(self.prompt_dir)],
            capture_output=True,
            text=True,
        )
        return [f.strip() for f in result.stdout.strip().split("\n") if f.strip()]
    
    def get_required_evaluations(self, changed_files: list[str]) -> set[str]:
        """변경된 파일에 따라 실행해야 할 평가 태스크를 결정합니다."""
        tasks: set[str] = set()
        
        for file in changed_files:
            mapped_tasks = self.prompt_task_mapping.get(file, ["unknown_prompt_eval"])
            if "all" in mapped_tasks:
                return {"all"}
            tasks.update(mapped_tasks)
        
        return tasks

from datetime import datetime, timedelta
 
 
class DriftMonitor:
    """프로덕션 환경의 품질 드리프트를 모니터링합니다."""
    
    def __init__(self, alert_threshold: float = 0.05):
        self.alert_threshold = alert_threshold
        self.metric_history: list[dict] = []
    
    def record_metric(self, metric_name: str, value: float, timestamp: datetime | None = None) -> None:
        """메트릭 값을 기록합니다."""
        self.metric_history.append({
            "metric": metric_name,
            "value": value,
            "timestamp": (timestamp or datetime.now()).isoformat(),
        })
    
    def check_drift(self, metric_name: str, window_days: int = 7) -> dict:
        """최근 윈도우 기간의 메트릭 변화를 분석합니다."""
        cutoff = datetime.now() - timedelta(days=window_days)
        
        recent = [
            entry["value"]
            for entry in self.metric_history
            if entry["metric"] == metric_name
            and datetime.fromisoformat(entry["timestamp"]) >= cutoff
        ]
        
        older = [
            entry["value"]
            for entry in self.metric_history
            if entry["metric"] == metric_name
            and datetime.fromisoformat(entry["timestamp"]) < cutoff
        ]
        
        if not recent or not older:
            return {"status": "insufficient_data"}
        
        import numpy as np
        recent_mean = np.mean(recent)
        older_mean = np.mean(older)
        drift = recent_mean - older_mean
        
        return {
            "status": "drift_detected" if abs(drift) > self.alert_threshold else "stable",
            "recent_mean": recent_mean,
            "baseline_mean": older_mean,
            "drift": drift,
            "sample_sizes": {"recent": len(recent), "baseline": len(older)},
        }