6장: 회귀 테스트 자동화

from dataclasses import dataclass
 
@dataclass
class GraduationCriteria:
    """평가를 회귀 테스트로 졸업시키기 위한 기준"""
 
    # 최소 연속 통과 횟수
    min_consecutive_passes: int = 5
 
    # 최소 평균 점수
    min_average_score: float = 0.85
 
    # 최대 허용 표준편차 (안정성)
    max_score_stdev: float = 0.10
 
    # 최소 실행 기간 (일)
    min_observation_days: int = 7
 
    # 최소 총 실행 횟수
    min_total_runs: int = 10
 
def check_graduation(
    eval_history: list[dict],
    criteria: GraduationCriteria,
) -> dict:
    """졸업 기준 충족 여부를 확인"""
    import statistics
    from datetime import datetime, timedelta
 
    scores = [h["score"] for h in eval_history]
    dates = [h["date"] for h in eval_history]
    passed = [h["passed"] for h in eval_history]
 
    # 기준 1: 최소 실행 횟수
    enough_runs = len(eval_history) >= criteria.min_total_runs
 
    # 기준 2: 최소 관찰 기간
    if dates:
        observation_days = (max(dates) - min(dates)).days
        enough_days = observation_days >= criteria.min_observation_days
    else:
        enough_days = False
 
    # 기준 3: 연속 통과
    consecutive = 0
    max_consecutive = 0
    for p in passed:
        if p:
            consecutive += 1
            max_consecutive = max(max_consecutive, consecutive)
        else:
            consecutive = 0
    enough_consecutive = max_consecutive >= criteria.min_consecutive_passes
 
    # 기준 4: 평균 점수
    avg_score = statistics.mean(scores) if scores else 0
    high_enough = avg_score >= criteria.min_average_score
 
    # 기준 5: 안정성
    stdev = statistics.stdev(scores) if len(scores) > 1 else 0
    stable_enough = stdev <= criteria.max_score_stdev
 
    ready = all([enough_runs, enough_days, enough_consecutive,
                 high_enough, stable_enough])
 
    return {
        "ready_to_graduate": ready,
        "criteria_met": {
            "min_total_runs": enough_runs,
            "min_observation_days": enough_days,
            "min_consecutive_passes": enough_consecutive,
            "min_average_score": high_enough,
            "max_score_stdev": stable_enough,
        },
        "stats": {
            "total_runs": len(eval_history),
            "average_score": avg_score,
            "stdev": stdev,
            "max_consecutive_passes": max_consecutive,
        },
    }

from dataclasses import dataclass, field
 
@dataclass
class GoldenTestCase:
    """Golden Dataset의 개별 테스트 케이스"""
    id: str
    category: str
    input_message: str
    expected_tools: list[str]           # 호출해야 할 도구 목록
    expected_tool_params: dict | None   # 핵심 파라미터
    quality_criteria: list[str]         # LLM-as-Judge 평가 기준
    min_score: float                    # 최소 통과 점수
    reference_response: str | None      # 참조 답변 (선택)
    tags: list[str] = field(default_factory=list)
    created_at: str = ""
    last_passed: str = ""
    pass_rate: float = 0.0
 
@dataclass
class GoldenDataset:
    """회귀 테스트용 Golden Dataset"""
    name: str
    version: str
    test_cases: list[GoldenTestCase]
    metadata: dict = field(default_factory=dict)
 
    def filter_by_category(self, category: str) -> list[GoldenTestCase]:
        return [tc for tc in self.test_cases if tc.category == category]
 
    def filter_by_tags(self, tags: list[str]) -> list[GoldenTestCase]:
        return [
            tc for tc in self.test_cases
            if any(tag in tc.tags for tag in tags)
        ]

name: customer-support-agent-golden
version: "1.3.0"
metadata:
  agent: customer-support-v2
  created: "2026-03-01"
  last_updated: "2026-03-28"
 
test_cases:
  - id: "CS-001"
    category: "order_inquiry"
    input_message: "주문번호 ORD-12345의 배송 상태를 확인해 주세요"
    expected_tools:
      - "lookup_order"
      - "get_shipping_status"
    expected_tool_params:
      lookup_order:
        order_id: "ORD-12345"
    quality_criteria:
      - "주문번호를 정확히 조회했는가"
      - "배송 상태 정보가 포함되어 있는가"
      - "예상 배송일이 안내되었는가"
    min_score: 0.85
    tags: ["order", "shipping", "core"]
 
  - id: "CS-002"
    category: "refund"
    input_message: "지난주 구매한 상품을 환불하고 싶습니다"
    expected_tools:
      - "get_recent_orders"
      - "check_refund_eligibility"
    quality_criteria:
      - "최근 주문 목록을 조회했는가"
      - "환불 정책을 안내했는가"
      - "환불 절차를 단계별로 설명했는가"
    min_score: 0.80
    tags: ["refund", "core"]

import hashlib
import json
 
def compute_dataset_hash(dataset: GoldenDataset) -> str:
    """데이터셋의 해시를 계산하여 변경을 추적"""
    content = json.dumps(
        [vars(tc) for tc in dataset.test_cases],
        sort_keys=True,
        ensure_ascii=False,
    )
    return hashlib.sha256(content.encode()).hexdigest()[:12]
 
def track_dataset_change(old_dataset, new_dataset) -> dict:
    """두 데이터셋 버전 간의 차이를 분석"""
    old_ids = {tc.id for tc in old_dataset.test_cases}
    new_ids = {tc.id for tc in new_dataset.test_cases}
 
    return {
        "added": new_ids - old_ids,
        "removed": old_ids - new_ids,
        "modified": {
            tc.id for tc in new_dataset.test_cases
            if tc.id in old_ids and tc != get_by_id(old_dataset, tc.id)
        },
        "unchanged": old_ids & new_ids,
    }

from datetime import datetime, timedelta
from collections import defaultdict
 
class RollingMetrics:
    """롤링 윈도우 기반 성공률 모니터링"""
 
    def __init__(self, results: list[dict]):
        self.results = sorted(results, key=lambda r: r["date"])
 
    def success_rate(self, window_days: int) -> float:
        """지정된 기간의 성공률 계산"""
        cutoff = datetime.now() - timedelta(days=window_days)
        recent = [r for r in self.results if r["date"] >= cutoff]
 
        if not recent:
            return 0.0
 
        passed = sum(1 for r in recent if r["passed"])
        return passed / len(recent)
 
    def by_category(self, window_days: int) -> dict:
        """카테고리별 성공률"""
        cutoff = datetime.now() - timedelta(days=window_days)
        recent = [r for r in self.results if r["date"] >= cutoff]
 
        categories = defaultdict(lambda: {"passed": 0, "total": 0})
        for r in recent:
            cat = r["category"]
            categories[cat]["total"] += 1
            if r["passed"]:
                categories[cat]["passed"] += 1
 
        return {
            cat: data["passed"] / data["total"]
            for cat, data in categories.items()
        }
 
    def detect_regression(
        self,
        baseline_window: int = 30,
        current_window: int = 7,
        threshold_drop: float = 0.10,
    ) -> dict:
        """현재 성공률이 베이스라인 대비 하락했는지 감지"""
        baseline_rate = self.success_rate(baseline_window)
        current_rate = self.success_rate(current_window)
 
        drop = baseline_rate - current_rate
        is_regression = drop >= threshold_drop
 
        return {
            "baseline_rate": baseline_rate,
            "current_rate": current_rate,
            "drop": drop,
            "is_regression": is_regression,
            "message": (
                f"성공률이 {drop:.1%} 하락했습니다 "
                f"({baseline_rate:.1%} -> {current_rate:.1%})"
                if is_regression
                else "정상 범위입니다"
            ),
        }

@dataclass
class Baseline:
    """에이전트 성능 베이스라인"""
    agent_version: str
    created_at: str
    overall_success_rate: float
    category_rates: dict
    average_scores: dict
    sample_size: int
 
    @classmethod
    def from_results(cls, agent_version: str, results: list[dict]):
        """실행 결과로부터 베이스라인 생성"""
        import statistics
 
        passed = sum(1 for r in results if r["passed"])
        scores = [r["score"] for r in results]
 
        # 카테고리별 성공률
        categories = defaultdict(list)
        for r in results:
            categories[r["category"]].append(r["passed"])
 
        return cls(
            agent_version=agent_version,
            created_at=datetime.now().isoformat(),
            overall_success_rate=passed / len(results),
            category_rates={
                cat: sum(vals) / len(vals)
                for cat, vals in categories.items()
            },
            average_scores={
                "mean": statistics.mean(scores),
                "stdev": statistics.stdev(scores),
                "p50": sorted(scores)[len(scores) // 2],
            },
            sample_size=len(results),
        )
 
def should_update_baseline(
    current_baseline: Baseline,
    new_results: list[dict],
    improvement_threshold: float = 0.05,
) -> bool:
    """베이스라인을 갱신해야 하는지 판단"""
    new_rate = sum(1 for r in new_results if r["passed"]) / len(new_results)
    improvement = new_rate - current_baseline.overall_success_rate
 
    # 유의미한 개선이 있을 때만 갱신
    return improvement >= improvement_threshold

async def analyze_change_impact(
    agent_before,
    agent_after,
    golden_dataset: GoldenDataset,
    num_runs: int = 3,
) -> dict:
    """변경 전후의 에이전트 성능을 비교 분석"""
 
    before_results = []
    after_results = []
 
    for test_case in golden_dataset.test_cases:
        # 변경 전 에이전트 실행
        for _ in range(num_runs):
            result = await run_test(agent_before, test_case)
            before_results.append({**result, "test_id": test_case.id})
 
        # 변경 후 에이전트 실행
        for _ in range(num_runs):
            result = await run_test(agent_after, test_case)
            after_results.append({**result, "test_id": test_case.id})
 
    # 테스트 케이스별 비교
    comparison = {}
    for tc in golden_dataset.test_cases:
        before = [r for r in before_results if r["test_id"] == tc.id]
        after = [r for r in after_results if r["test_id"] == tc.id]
 
        before_rate = sum(r["passed"] for r in before) / len(before)
        after_rate = sum(r["passed"] for r in after) / len(after)
 
        comparison[tc.id] = {
            "before": before_rate,
            "after": after_rate,
            "delta": after_rate - before_rate,
            "status": (
                "improved" if after_rate > before_rate
                else "degraded" if after_rate < before_rate
                else "unchanged"
            ),
        }
 
    # 전체 요약
    improved = sum(1 for c in comparison.values() if c["status"] == "improved")
    degraded = sum(1 for c in comparison.values() if c["status"] == "degraded")
    unchanged = sum(1 for c in comparison.values() if c["status"] == "unchanged")
 
    return {
        "summary": {
            "improved": improved,
            "degraded": degraded,
            "unchanged": unchanged,
            "total": len(comparison),
        },
        "details": comparison,
        "recommendation": (
            "APPROVE" if degraded == 0
            else "REVIEW" if degraded <= 2
            else "BLOCK"
        ),
    }

async def regression_pipeline(
    agent,
    golden_dataset: GoldenDataset,
    baseline: Baseline,
) -> dict:
    """회귀 테스트 자동화 파이프라인"""
 
    # 1단계: 회귀 테스트 실행
    results = []
    for tc in golden_dataset.test_cases:
        result = await run_regression_test(agent, tc)
        results.append(result)
 
    # 2단계: 롤링 메트릭 계산
    metrics = RollingMetrics(results)
    regression = metrics.detect_regression()
 
    # 3단계: 베이스라인 비교
    current_rate = sum(r["passed"] for r in results) / len(results)
    baseline_delta = current_rate - baseline.overall_success_rate
 
    # 4단계: 판정
    verdict = "PASS"
    if regression["is_regression"]:
        verdict = "FAIL"
    elif baseline_delta < -0.05:
        verdict = "WARNING"
 
    # 5단계: 리포트 생성
    report = {
        "verdict": verdict,
        "current_success_rate": current_rate,
        "baseline_success_rate": baseline.overall_success_rate,
        "delta": baseline_delta,
        "regression_detected": regression["is_regression"],
        "failed_tests": [r for r in results if not r["passed"]],
        "total_tests": len(results),
    }
 
    return report