9장: CI/CD 통합과 품질 게이트

name: Agent Quality Tests
 
on:
  pull_request:
    paths:
      - "src/agents/**"
      - "prompts/**"
      - "tools/**"
      - "tests/agent/**"
 
concurrency:
  group: agent-tests-${{ github.head_ref }}
  cancel-in-progress: true
 
env:
  ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
  OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
 
jobs:
  unit-tests:
    name: "Stage 1: Unit Tests (Tool Calls)"
    runs-on: ubuntu-latest
    timeout-minutes: 10
    steps:
      - uses: actions/checkout@v4
 
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
 
      - name: Install dependencies
        run: pip install -r requirements-test.txt
 
      - name: Run tool call tests
        run: |
          pytest tests/agent/unit/ \
            -v --tb=short \
            --junitxml=reports/unit-results.xml
 
      - name: Upload unit test results
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: unit-test-results
          path: reports/unit-results.xml
 
  evaluations:
    name: "Stage 2: Evaluations"
    needs: unit-tests
    runs-on: ubuntu-latest
    timeout-minutes: 30
    steps:
      - uses: actions/checkout@v4
 
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
 
      - name: Install dependencies
        run: pip install -r requirements-test.txt
 
      - name: Run evaluations (3x average)
        run: |
          python -m agent_eval.runner \
            --config eval_config.yaml \
            --num-runs 3 \
            --output reports/eval-results.json
 
      - name: Upload evaluation results
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: eval-results
          path: reports/eval-results.json
 
  regression-tests:
    name: "Stage 3: Regression Tests"
    needs: evaluations
    runs-on: ubuntu-latest
    timeout-minutes: 45
    steps:
      - uses: actions/checkout@v4
 
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
 
      - name: Install dependencies
        run: pip install -r requirements-test.txt
 
      - name: Load Golden Dataset
        run: |
          python -m agent_eval.dataset load \
            --version latest \
            --output golden_dataset.yaml
 
      - name: Run regression tests
        run: |
          python -m agent_eval.regression \
            --dataset golden_dataset.yaml \
            --baseline baselines/current.json \
            --output reports/regression-results.json
 
      - name: Upload regression results
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: regression-results
          path: reports/regression-results.json
 
  quality-gate:
    name: "Quality Gate"
    needs: [unit-tests, evaluations, regression-tests]
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
 
      - name: Download all results
        uses: actions/download-artifact@v4
        with:
          path: reports/
 
      - name: Evaluate quality gate
        id: gate
        run: |
          python -m agent_eval.quality_gate \
            --unit-results reports/unit-test-results/unit-results.xml \
            --eval-results reports/eval-results/eval-results.json \
            --regression-results reports/regression-results/regression-results.json \
            --output reports/gate-decision.json
 
      - name: Post PR comment
        if: always()
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const decision = JSON.parse(
              fs.readFileSync('reports/gate-decision.json', 'utf8')
            );
 
            const body = formatGateReport(decision);
 
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: body,
            });
 
      - name: Enforce gate
        run: |
          python -c "
          import json
          with open('reports/gate-decision.json') as f:
              decision = json.load(f)
          if decision['verdict'] != 'PASS':
              print(f'Quality gate FAILED: {decision[\"reason\"]}')
              exit(1)
          print('Quality gate PASSED')
          "

@dataclass
class QualityGateConfig:
    """품질 게이트 임계값 설정"""
 
    # Stage 1: 단위 테스트 — 100% 통과 필수
    unit_test_pass_rate: float = 1.0
 
    # Stage 2: 평가 — 평균 점수 기준
    eval_min_overall_score: float = 0.75
    eval_min_faithfulness: float = 0.85
    eval_min_hallucination: float = 0.90
 
    # Stage 3: 회귀 테스트 — 베이스라인 대비
    regression_max_drop: float = 0.05  # 5% 이상 하락 시 실패
    regression_min_pass_rate: float = 0.85
 
    # 종합 판정
    allow_eval_skip_if_unit_perfect: bool = False
    require_all_stages: bool = True
 
def evaluate_quality_gate(
    unit_results: dict,
    eval_results: dict,
    regression_results: dict,
    config: QualityGateConfig = QualityGateConfig(),
) -> dict:
    """품질 게이트 판정"""
 
    checks = []
 
    # Stage 1 확인
    unit_rate = unit_results["pass_rate"]
    checks.append({
        "stage": "unit_tests",
        "passed": unit_rate >= config.unit_test_pass_rate,
        "value": unit_rate,
        "threshold": config.unit_test_pass_rate,
        "message": f"단위 테스트 통과율: {unit_rate:.1%}",
    })
 
    # Stage 2 확인
    eval_score = eval_results["average_score"]
    faithfulness = eval_results["dimensions"]["faithfulness"]["mean"]
    hallucination = eval_results["dimensions"]["hallucination"]["mean"]
 
    checks.append({
        "stage": "eval_overall",
        "passed": eval_score >= config.eval_min_overall_score,
        "value": eval_score,
        "threshold": config.eval_min_overall_score,
        "message": f"평가 종합 점수: {eval_score:.3f}",
    })
 
    checks.append({
        "stage": "eval_faithfulness",
        "passed": faithfulness >= config.eval_min_faithfulness,
        "value": faithfulness,
        "threshold": config.eval_min_faithfulness,
        "message": f"충실성 점수: {faithfulness:.3f}",
    })
 
    checks.append({
        "stage": "eval_hallucination",
        "passed": hallucination >= config.eval_min_hallucination,
        "value": hallucination,
        "threshold": config.eval_min_hallucination,
        "message": f"환각 방지 점수: {hallucination:.3f}",
    })
 
    # Stage 3 확인
    baseline_rate = regression_results["baseline_success_rate"]
    current_rate = regression_results["current_success_rate"]
    drop = baseline_rate - current_rate
 
    checks.append({
        "stage": "regression",
        "passed": drop <= config.regression_max_drop,
        "value": current_rate,
        "threshold": baseline_rate - config.regression_max_drop,
        "message": f"회귀 성공률: {current_rate:.1%} (베이스라인: {baseline_rate:.1%}, 변동: {-drop:+.1%})",
    })
 
    # 종합 판정
    all_passed = all(c["passed"] for c in checks)
    failed_checks = [c for c in checks if not c["passed"]]
 
    verdict = "PASS" if all_passed else "FAIL"
    reason = (
        "모든 품질 기준을 충족했습니다"
        if all_passed
        else f"{len(failed_checks)}개 기준 미달: " +
             ", ".join(c["stage"] for c in failed_checks)
    )
 
    return {
        "verdict": verdict,
        "reason": reason,
        "checks": checks,
        "failed_checks": failed_checks,
    }

def format_pr_report(gate_decision: dict) -> str:
    """품질 게이트 결과를 PR 코멘트용 마크다운으로 포맷"""
 
    verdict_icon = "[PASS]" if gate_decision["verdict"] == "PASS" else "[FAIL]"
    header = f"## Agent Quality Report {verdict_icon}\n\n"
 
    # 요약
    summary = f"**판정**: {gate_decision['verdict']}\n"
    summary += f"**사유**: {gate_decision['reason']}\n\n"
 
    # 상세 체크 테이블
    table = "| 단계 | 결과 | 값 | 기준 | 설명 |\n"
    table += "|------|------|-----|------|------|\n"
 
    for check in gate_decision["checks"]:
        status = "Pass" if check["passed"] else "Fail"
        table += (
            f"| {check['stage']} | {status} | "
            f"{check['value']:.3f} | {check['threshold']:.3f} | "
            f"{check['message']} |\n"
        )
 
    # 실패한 테스트 상세
    details = ""
    if gate_decision["failed_checks"]:
        details = "\n### 실패 항목 상세\n\n"
        for check in gate_decision["failed_checks"]:
            details += f"- **{check['stage']}**: {check['message']}\n"
            details += f"  - 현재 값: {check['value']:.3f}, 기준: {check['threshold']:.3f}\n"
 
    return header + summary + table + details

## Agent Quality Report [PASS]
 
**판정**: PASS
**사유**: 모든 품질 기준을 충족했습니다
 
| 단계 | 결과 | 값 | 기준 | 설명 |
|------|------|-----|------|------|
| unit_tests | Pass | 1.000 | 1.000 | 단위 테스트 통과율: 100.0% |
| eval_overall | Pass | 0.823 | 0.750 | 평가 종합 점수: 0.823 |
| eval_faithfulness | Pass | 0.891 | 0.850 | 충실성 점수: 0.891 |
| eval_hallucination | Pass | 0.934 | 0.900 | 환각 방지 점수: 0.934 |
| regression | Pass | 0.912 | 0.862 | 회귀 성공률: 91.2% |

import random
from datetime import datetime
 
class OnlineEvaluator:
    """프로덕션 트래픽의 온라인 평가"""
 
    def __init__(
        self,
        sample_rate: float = 0.10,  # 10% 트래픽 샘플링
        eval_dimensions: list[str] | None = None,
    ):
        self.sample_rate = sample_rate
        self.eval_dimensions = eval_dimensions or [
            "faithfulness", "relevance", "hallucination",
        ]
 
    def should_evaluate(self) -> bool:
        """샘플링 결정"""
        return random.random() < self.sample_rate
 
    async def evaluate_async(
        self,
        conversation: list[dict],
        tool_calls: list[dict],
        metadata: dict,
    ) -> dict | None:
        """비동기로 프로덕션 응답을 평가"""
        if not self.should_evaluate():
            return None
 
        # 비동기 평가 (응답 지연에 영향 없음)
        result = await comprehensive_evaluation(
            conversation=conversation,
            tool_results=tool_calls,
        )
 
        # 메트릭 저장
        metric_entry = {
            "timestamp": datetime.now().isoformat(),
            "conversation_id": metadata.get("conversation_id"),
            "scores": result["dimensions"],
            "weighted_score": result["weighted_score"],
            "metadata": metadata,
        }
 
        await self.store_metric(metric_entry)
        await self.check_alerts(metric_entry)
 
        return metric_entry
 
    async def check_alerts(self, metric: dict):
        """실시간 알림 확인"""
        if metric["weighted_score"] < 0.5:
            await self.send_alert(
                severity="critical",
                message=f"프로덕션 응답 품질 심각 저하: {metric['weighted_score']:.3f}",
                conversation_id=metric["conversation_id"],
            )

def check_eval_alignment(
    offline_scores: list[float],
    online_scores: list[float],
    max_divergence: float = 0.15,
) -> dict:
    """오프라인 평가와 온라인 평가의 점수가 정렬되어 있는지 확인"""
    import statistics
 
    offline_mean = statistics.mean(offline_scores)
    online_mean = statistics.mean(online_scores)
 
    divergence = abs(offline_mean - online_mean)
    is_aligned = divergence <= max_divergence
 
    return {
        "offline_mean": offline_mean,
        "online_mean": online_mean,
        "divergence": divergence,
        "is_aligned": is_aligned,
        "message": (
            "오프라인과 온라인 평가가 정렬되어 있습니다"
            if is_aligned
            else f"평가 불일치 감지: 차이 {divergence:.3f} (허용 {max_divergence})"
        ),
    }

CI_COST_STRATEGIES = {
    "tiered_execution": {
        "description": "변경 범위에 따라 테스트 수준을 조절",
        "rules": [
            "코드만 변경: 단위 테스트만 실행",
            "프롬프트 변경: 단위 + 평가",
            "모델/도구 변경: 전체 테스트",
        ],
    },
    "smart_sampling": {
        "description": "Golden Dataset에서 관련 케이스만 실행",
        "rules": [
            "변경된 도구와 관련된 테스트 케이스만 선택",
            "최근 실패 이력이 있는 케이스 우선",
            "전체 실행은 야간 배치로",
        ],
    },
    "caching": {
        "description": "변경되지 않은 부분의 결과를 캐싱",
        "rules": [
            "도구 코드 미변경 시 도구 테스트 결과 재사용",
            "프롬프트 해시 기반 평가 결과 캐싱",
        ],
    },
}

NOTIFICATION_RULES = {
    "critical": {
        "channels": ["slack_urgent", "pagerduty"],
        "conditions": [
            "단위 테스트 실패",
            "환각 점수 0.7 미만",
            "프로덕션 온라인 평가 0.5 미만",
        ],
        "throttle": "즉시 알림, 1시간 내 중복 방지",
    },
    "warning": {
        "channels": ["slack_team"],
        "conditions": [
            "평가 점수 하락 (임계값 근접)",
            "회귀 성공률 5% 이상 하락",
            "비용 베이스라인 50% 이상 초과",
        ],
        "throttle": "하루 최대 3회",
    },
    "info": {
        "channels": ["slack_team"],
        "conditions": [
            "품질 게이트 통과",
            "베이스라인 갱신",
            "새 테스트 케이스 졸업",
        ],
        "throttle": "하루 1회 요약",
    },
}