Case 05 : 배포 게이트를 자동화합시다

페르소나

Priya는 모델 배포 과정을 운영하는 MLOps 엔지니어입니다.

Priya의 팀은 Slack 체크리스트로 배포를 승인합니다.

• 메트릭을 확인했나요?
• 이전 버전과 비교했나요?
• 데이터 품질을 검증했나요?
• 평가 스테이지가 실제로 실행됐나요?

문제는 체크리스트가 사람이 기억하고 확인하는 과정에 의존한다는 점입니다.

상황

Priya의 팀은 한 분기 동안 세 번의 배포 사고를 겪었습니다.

• 3월: 잘못된 스테이지의 평가 메트릭으로 배포했습니다.
• 4월: 데이터셋 버전이 바뀐 실행을 배포해 CTR 하락을 유발했습니다.
• 5월: evaluate 스테이지가 건너뛰어졌는데도 “메트릭 확인” 체크박스가 통과됐습니다.

이 케이스의 핵심은 배포 전 확인 항목을 사람의 체크리스트가 아니라 실행 기록을 읽는 프로그램으로 바꾸는 것입니다.

Contexta 없이 해결하려면

Contexta 없이 배포 게이트를 운영하면 보통 아래처럼 진행됩니다.

1. 후보 실행의 메트릭을 사람이 직접 확인합니다.
2. 이전 배포와 비교했는지 Slack에 댓글로 남깁니다.
3. 데이터 품질 이슈가 있었는지 담당자 기억에 의존합니다.
4. 평가 스테이지가 누락되지 않았는지 별도 로그를 확인합니다.
5. 승인자가 체크박스와 댓글을 보고 최종 승인합니다.

이 방식은 과거 사고 패턴을 기억하지 못합니다. 메트릭이 없는 실행도 체크박스 하나로 통과될 수 있습니다.

Contexta로 해결하기

배포 게이트는 후보 실행의 스냅샷과 진단 결과를 읽어 자동으로 PASS/FAIL을 결정합니다.

candidate-v2-good       PASS  metrics present, no regression
candidate-v3-no-eval    FAIL  missing: accuracy, auc, f1
candidate-v4-regressed  FAIL  regression detected

이 해결 흐름은 세 단계입니다.

단계	Contexta API	얻는 정보
진단 이슈 확인	ctx.diagnose_run(candidate_run_id)	에러 수준 이슈 존재 여부
필수 메트릭 확인	ctx.get_run_snapshot(candidate_run_id)	accuracy, auc, f1 기록 여부
회귀 검사	ctx.compare_runs(previous_deploy_run_id, candidate_run_id)	이전 배포 대비 허용 범위 초과 하락 여부

예제 코드

아래 코드는 데이터 준비 단계에서 만든 좋은 후보, 평가 누락 후보, 성능 회귀 후보를 게이트 함수로 검사합니다. 회귀 임계값은 예제 코드에서 REGRESSION_THRESHOLD = 0.02로 정하고, 게이트 함수 호출 시 전달합니다.

deployment_gate_review.py

"""Run a deployment gate against previously recorded candidates."""

from pathlib import Path
from dataclasses import dataclass

from contexta import Contexta
from contexta.config import UnifiedConfig, WorkspaceConfig


PROJECT_NAME = "product-ranker"
REQUIRED_METRICS = ["accuracy", "auc", "f1"]
BASELINE = f"run:{PROJECT_NAME}.baseline-v1"
CANDIDATES = [
    f"run:{PROJECT_NAME}.candidate-v2-good",
    f"run:{PROJECT_NAME}.candidate-v3-no-eval",
    f"run:{PROJECT_NAME}.candidate-v4-regressed",
]
REGRESSION_THRESHOLD = 0.02


@dataclass
class GateResult:
    passed: bool
    run_id: str
    checks: list[tuple[str, bool, str]]

    def print_report(self) -> None:
        status = "PASS" if self.passed else "FAIL"
        print(f"\n  Pre-deployment gate for '{self.run_id}': [{status}]")
        for name, ok, detail in self.checks:
            icon = "  [OK]  " if ok else "  [NG]  "
            print(f"    {icon}  [{name}] {detail}")


def pre_deployment_gate(
    ctx: Contexta,
    candidate_run_id: str,
    previous_deploy_run_id: str | None,
    regression_threshold: float,
) -> GateResult:
    checks: list[tuple[str, bool, str]] = []

    diag = ctx.diagnose_run(candidate_run_id)
    errors = [issue for issue in diag.issues if issue.severity == "error"]
    warnings = [issue for issue in diag.issues if issue.severity == "warning"]
    if errors:
        checks.append(
            ("diagnostics", False, f"{len(errors)} error(s): {', '.join(issue.code for issue in errors)}")
        )
    else:
        checks.append(("diagnostics", True, f"clean ({len(warnings)} warning(s))"))

    snapshot = ctx.get_run_snapshot(candidate_run_id)
    metric_keys = {record.key for record in snapshot.records if record.record_type == "metric"}
    missing = [metric for metric in REQUIRED_METRICS if metric not in metric_keys]
    if missing:
        checks.append(("required_metrics", False, f"missing: {missing}"))
    else:
        values = {
            record.key: record.value
            for record in snapshot.records
            if record.record_type == "metric"
        }
        summary = ", ".join(f"{key}={values[key]:.4f}" for key in REQUIRED_METRICS)
        checks.append(("required_metrics", True, summary))

    comparison = ctx.compare_runs(previous_deploy_run_id, candidate_run_id)
    regressions: list[str] = []
    for stage in comparison.stage_comparisons:
        for delta in stage.metric_deltas:
            if (
                delta.metric_key in REQUIRED_METRICS
                and delta.left_value is not None
                and delta.right_value is not None
            ):
                change = (delta.right_value - delta.left_value) / max(abs(delta.left_value), 1e-9)
                if change < -regression_threshold:
                    regressions.append(
                        f"{delta.metric_key}: {delta.left_value:.4f}->{delta.right_value:.4f} ({change:+.1%})"
                    )

    if regressions:
        checks.append(("regression_check", False, f"regression detected: {'; '.join(regressions)}"))
    else:
        checks.append(
            ("regression_check", True, f"no significant regression vs previous deploy (threshold={regression_threshold:.0%})")
        )

    return GateResult(
        passed=all(ok for _, ok, _ in checks),
        run_id=candidate_run_id,
        checks=checks,
    )


ctx = Contexta(
    config=UnifiedConfig(
        project_name=PROJECT_NAME,
        workspace=WorkspaceConfig(root_path=Path(".contexta")),
    )
)

store = ctx.metadata_store
try:
    for candidate in CANDIDATES:
        result = pre_deployment_gate(
            ctx,
            candidate,
            BASELINE,
            regression_threshold=REGRESSION_THRESHOLD,
        )
        result.print_report()
finally:
    store.close()

실행하면 다음과 같은 결과를 얻습니다.

Pre-deployment gate for 'run:product-ranker.candidate-v2-good': [PASS]
    [OK]    [diagnostics] clean (0 warning(s))
    [OK]    [required_metrics] accuracy=0.9010, auc=0.9330, f1=0.8890
    [OK]    [regression_check] no significant regression vs previous deploy (threshold=2%)

Pre-deployment gate for 'run:product-ranker.candidate-v3-no-eval': [FAIL]
    [OK]    [diagnostics] clean (0 warning(s))
    [NG]    [required_metrics] missing: ['accuracy', 'auc', 'f1']
    [OK]    [regression_check] no significant regression vs previous deploy (threshold=2%)

Pre-deployment gate for 'run:product-ranker.candidate-v4-regressed': [FAIL]
    [OK]    [diagnostics] clean (0 warning(s))
    [OK]    [required_metrics] accuracy=0.8410, auc=0.8910, f1=0.8590
    [NG]    [regression_check] regression detected: accuracy: 0.8930->0.8410 (-5.8%); auc: 0.9270->0.8910 (-3.9%); f1: 0.8810->0.8590 (-2.5%)

코드 조각별로 이해하기

분석 스크립트가 호출하는 pre_deployment_gate() 함수는 아래 세 가지 검사를 순서대로 수행합니다.

1. 진단 이슈 확인하기

diag = ctx.diagnose_run(candidate_run_id)
errors = [i for i in diag.issues if i.severity == "error"]

배포 후보에 에러 수준 진단 이슈가 있으면 게이트는 실패합니다.

이번 예제의 세 후보는 모두 error-level diagnostics가 없어 이 검사는 [OK]로 통과합니다.

이 검사는 데이터 품질 저하나 스테이지 내부 이상을 사람이 놓치는 상황을 줄입니다.

2. 필수 메트릭 확인하기

snapshot = ctx.get_run_snapshot(candidate_run_id)
obs_keys = {o.key for o in snapshot.records if o.record_type == "metric"}
missing = [m for m in REQUIRED_METRICS if m not in obs_keys]

evaluate 스테이지가 건너뛰어진 candidate-v3-no-eval에는 필수 메트릭이 존재하지 않습니다.

Slack 체크박스의 “메트릭 확인” 대신, 실제 저장된 기록을 기준으로 확인합니다.

3. 이전 배포 대비 회귀 검사하기

REGRESSION_THRESHOLD = 0.02

comparison = ctx.compare_runs(previous_deploy_run_id, candidate_run_id)

if change < -REGRESSION_THRESHOLD:
    regressions.append(...)

후보 실행이 이전 배포보다 허용 범위 이상 나빠졌다면 게이트는 실패합니다.

candidate-v4-regressed는 accuracy, auc, f1이 각각 5.8%p, 3.9%p, 2.5%p 하락했습니다.

이 예제에서는 REGRESSION_THRESHOLD = 0.02로 설정해 2%를 넘는 하락을 막습니다.

최종 답변

Contexta를 통해, Priya는 앞서 제시된 질문에 이런 식으로 답변할 수 있습니다.

• Q1. 메트릭을 확인했나요?

  • A1. 네. candidate-v2-good과 candidate-v4-regressed에는 accuracy, auc, f1이 모두 있습니다.
    candidate-v3-no-eval은 세 메트릭이 모두 없어 FAIL입니다.

• Q2. 이전 버전과 비교했나요?

  • A2. 네. candidate-v2-good은 이전 배포 대비 유의미한 회귀가 없어 PASS입니다.
    candidate-v4-regressed는 accuracy, auc, f1 모두 회귀가 있어 FAIL입니다.

• Q3. 데이터를 검증했나요?

  • A3. 네.진단 결과 기준으로 세 후보 모두 error-level diagnostics는 없습니다.

• Q4. 평가 스테이지가 실제로 실행됐나요?

  • A4. 네. 하지만 candidate-v3-no-eval은 평가 메트릭이 없기 때문에 평가 단계가 배포 승인에 필요한 증거를 남기지 못했습니다.

따라서, Priya는 다음과 같이 답할 수 있습니다.

이번 배포 후보 중 승인 가능한 것은 candidate-v2-good뿐입니다.
candidate-v3-no-eval은 필수 메트릭이 없고, candidate-v4-regressed는 이전 배포 대비 회귀가 커서 배포를 막아야 합니다.

선택: 예제 데이터 생성

이 섹션은 직접 코드를 실행해 보고 싶은 경우에만 필요합니다.

아래 데이터 준비 코드는 .contexta 워크스페이스에 기준 배포 실행과 세 가지 배포 후보를 생성합니다.

deployment_gate_seed.py

"""Create deployment-gate candidate records used by the gate case study."""

from __future__ import annotations

import tempfile
from pathlib import Path
from typing import Any

from contexta import Contexta
from contexta.config import UnifiedConfig, WorkspaceConfig
from contexta.contract import (
    DeploymentExecution,
    MetricPayload,
    MetricRecord,
    Project,
    RecordEnvelope,
    Run,
    StageExecution,
)

PROJECT_NAME = "product-ranker"
_rid = 0


def _next_rid() -> str:
    global _rid
    _rid += 1
    return f"r{_rid:04d}"


def _make_run(
    store: Any,
    record_store: Any,
    run_name: str,
    accuracy: float,
    auc: float,
    f1: float,
    has_evaluate_stage: bool = True,
    started_at: str = "2025-05-01T09:00:00Z",
    ended_at: str = "2025-05-01T11:00:00Z",
) -> str:
    run_ref = f"run:{PROJECT_NAME}.{run_name}"
    store.runs.put_run(
        Run(
            run_ref=run_ref,
            project_ref=f"project:{PROJECT_NAME}",
            name=run_name,
            status="completed",
            started_at=started_at,
            ended_at=ended_at,
        )
    )
    stage_ref = f"stage:{PROJECT_NAME}.{run_name}.evaluate"
    if has_evaluate_stage:
        store.stages.put_stage_execution(
            StageExecution(
                stage_execution_ref=stage_ref,
                run_ref=run_ref,
                stage_name="evaluate",
                status="completed",
                started_at=started_at,
                ended_at=ended_at,
                order_index=0,
            )
        )
        for key, val in [("accuracy", accuracy), ("auc", auc), ("f1", f1)]:
            record_store.append(
                MetricRecord(
                    envelope=RecordEnvelope(
                        record_ref=f"record:{PROJECT_NAME}.{run_name}.{_next_rid()}",
                        record_type="metric",
                        recorded_at=ended_at,
                        observed_at=ended_at,
                        producer_ref="contexta.case05",
                        run_ref=run_ref,
                        stage_execution_ref=stage_ref,
                        completeness_marker="complete",
                        degradation_marker="none",
                    ),
                    payload=MetricPayload(
                        metric_key=key,
                        value=val,
                        value_type="float64",
                    ),
                )
            )
    return run_ref


def run_example(workspace: Path | str | None = None) -> dict[str, Any]:
    if workspace is None:
        workspace = Path(tempfile.mkdtemp(prefix="contexta-case05-")) / ".contexta"

    ctx = Contexta(
        config=UnifiedConfig(
            project_name=PROJECT_NAME,
            workspace=WorkspaceConfig(root_path=Path(workspace)),
        )
    )
    store = ctx.metadata_store

    try:
        store.projects.put_project(
            Project(
                project_ref=f"project:{PROJECT_NAME}",
                name=PROJECT_NAME,
                created_at="2025-01-01T00:00:00Z",
            )
        )

        # ── Scenario A: safe baseline (previous production model) ────────────
        prev_run_ref = _make_run(
            store, ctx.record_store, "baseline-v1",
            accuracy=0.893, auc=0.927, f1=0.881,
            started_at="2025-04-28T09:00:00Z",
            ended_at="2025-04-28T11:00:00Z",
        )
        store.deployments.put_deployment_execution(
            DeploymentExecution(
                deployment_execution_ref=f"deployment:{PROJECT_NAME}.prod-v1",
                project_ref=f"project:{PROJECT_NAME}",
                deployment_name="prod-v1",
                status="completed",
                started_at="2025-04-28T12:00:00Z",
                ended_at="2025-04-28T12:10:00Z",
                run_ref=prev_run_ref,
                order_index=0,
            )
        )

        # ── Scenario B: GOOD candidate — should pass ────────────────────────
        good_run_ref = _make_run(
            store, ctx.record_store, "candidate-v2-good",
            accuracy=0.901, auc=0.933, f1=0.889,
            started_at="2025-05-05T09:00:00Z",
            ended_at="2025-05-05T11:00:00Z",
        )

        # ── Scenario C: MISSING METRICS — should fail ───────────────────────
        no_eval_run_ref = _make_run(
            store, ctx.record_store, "candidate-v3-no-eval",
            accuracy=0.0, auc=0.0, f1=0.0,
            has_evaluate_stage=False,
            started_at="2025-05-06T09:00:00Z",
            ended_at="2025-05-06T11:00:00Z",
        )

        # ── Scenario D: REGRESSION — accuracy dropped 5% ────────────────────
        regressed_run_ref = _make_run(
            store, ctx.record_store, "candidate-v4-regressed",
            accuracy=0.841, auc=0.891, f1=0.859,
            started_at="2025-05-07T09:00:00Z",
            ended_at="2025-05-07T11:00:00Z",
        )

        return {
            "baseline_run_id": prev_run_ref,
            "candidate_run_ids": [good_run_ref, no_eval_run_ref, regressed_run_ref],
        }

    finally:
        store.close()


def main() -> None:
    from contextlib import redirect_stdout
    import io

    with redirect_stdout(io.StringIO()):
        run_example(Path(".contexta"))

    print(f"Seeded {PROJECT_NAME} data in .contexta.")


if __name__ == "__main__":
    main()

코드를 deployment_gate_seed.py로 저장한 뒤, Contexta가 설치된 환경에서 실행하세요.

uv run deployment_gate_seed.py