Case 12 : 품질 인증서를 만들어봅시다

페르소나

Tom은 FinanceBank Corp에 학습된 모델을 납품하는 현장 배포 엔지니어입니다.

클라이언트의 조달팀은 모델을 수락하기 전에 공식 모델 품질 인증서를 요구합니다.

• 사용된 학습 데이터 버전은 무엇인가요?
• 평가 메트릭의 정확한 값은 무엇인가요?
• 학습 환경은 무엇인가요?
• 합의된 품질 임계값을 통과했나요?
• 전체 판정은 PASS인가요, FAIL인가요?

상황

Tom은 보통 학습 로그, requirements.txt, 노트북 출력, Slack 메시지를 모아 인증서를 수동으로 작성합니다.

메트릭 하나가 잘못 복사되면 클라이언트 검토가 다시 시작되고, 납품 일정이 하루 이상 밀릴 수 있습니다.

이 케이스의 핵심은 학습 실행에 저장된 증거를 기반으로 품질 임계값을 검사하고, 납품용 인증서를 자동으로 구성하는 것입니다.

Contexta 없이 해결하려면

Contexta 없이 인증서를 만들려면 보통 아래 작업을 직접 해야 합니다.

1. 학습 데이터 버전을 로그나 노트북에서 찾습니다.
2. 평가 메트릭을 문서에 복사합니다.
3. Python 버전과 주요 패키지를 requirements.txt에서 추정합니다.
4. 품질 임계값과 실제 값을 손으로 비교합니다.
5. 전체 PASS/FAIL 판정을 문서에 작성합니다.

이 방식은 느리고, 사람이 복사한 값과 실제 실행 기록이 달라질 위험이 있습니다.

Contexta로 해결하기

Tom은 실행 스냅샷과 재현성 감사를 읽어 인증서를 만듭니다.

run:fraud-detection-financebank.fraud-model-v2-1
├─ dataset version
├─ metrics: accuracy, auc, f1, precision, recall
├─ environment snapshot
├─ threshold checks
└─ snapshot report

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

단계	Contexta API	얻는 정보
실행 증거 읽기	ctx.get_run_snapshot(RUN_REF)	평가 메트릭
환경 감사	ctx.audit_reproducibility(RUN_REF)	Python 버전과 패키지 수
임계값 검사	저장된 MetricRecord + QUALITY_THRESHOLDS	항목별 PASS/FAIL
인증서 리포트	ctx.build_snapshot_report(RUN_REF)	제출 가능한 구조화 리포트

예제 코드

아래 코드는 데이터 준비 단계에서 만든 납품 대상 실행을 읽고, 품질 임계값 검사와 인증서 요약을 출력합니다.

analyze_certificate.py

"""Create a quality-certificate summary from a recorded delivery run."""

from pathlib import Path

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

PROJECT_NAME = "fraud-detection-financebank"
CLIENT_NAME = "FinanceBank Corp"
QUALITY_THRESHOLDS: dict[str, float] = {
    "accuracy": 0.90,
    "auc": 0.93,
    "f1": 0.88,
    "precision": 0.87,
    "recall": 0.86,
}
RUN_REF = f"run:{PROJECT_NAME}.fraud-model-v2-1"

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

store = ctx.metadata_store
try:
    snapshot = ctx.get_run_snapshot(RUN_REF)
    metrics = {
        record.key: float(record.value)
        for record in snapshot.records
        if record.record_type == "metric" and record.value is not None
    }
    audit = ctx.audit_reproducibility(RUN_REF)

    print(f"MODEL QUALITY CERTIFICATE - {CLIENT_NAME}")
    print(f"Run: {snapshot.run.name}")
    print(f"Python: {audit.python_version}; packages: {audit.package_count}")

    overall = True
    for key, threshold in QUALITY_THRESHOLDS.items():
        passed = metrics[key] >= threshold
        overall = overall and passed
        status = "PASS" if passed else "FAIL"
        print(f"{key:<10} {metrics[key]:.4f} >= {threshold:.4f} [{status}]")

    report = ctx.build_snapshot_report(RUN_REF)
    print(f"Overall decision: {'PASS' if overall else 'FAIL'}")
    print(f"Report: {report.title}")
finally:
    store.close()

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

MODEL QUALITY CERTIFICATE - FinanceBank Corp
Run: fraud-model-v2-1
Python: 3.11.8; packages: 6
accuracy   0.9340 >= 0.9000 [PASS]
auc        0.9610 >= 0.9300 [PASS]
f1         0.9220 >= 0.8800 [PASS]
precision  0.9170 >= 0.8700 [PASS]
recall     0.9280 >= 0.8600 [PASS]
Overall decision: PASS
Report: Run Snapshot Report: run:fraud-detection-financebank.fraud-model-v2-1

코드 조각별로 이해하기

1. 실행 스냅샷에서 증거 읽기

snapshot = ctx.get_run_snapshot(RUN_REF)
metrics = {
    record.key: float(record.value)
    for record in snapshot.records
    if record.record_type == "metric" and record.value is not None
}

스냅샷은 실행에 연결된 평가 메트릭을 제공합니다.

인증서에 들어가는 값은 사람이 다시 입력한 값이 아니라 실행 기록에서 읽은 값입니다.

2. 재현성 감사 포함하기

audit = ctx.audit_reproducibility(RUN_REF)
print(f"Python: {audit.python_version}; packages: {audit.package_count}")

학습 환경은 납품 품질의 일부입니다.

Python 버전과 패키지 기록이 남아 있으면 클라이언트가 모델 재현 가능성을 확인할 수 있습니다.

3. 임계값을 프로그램으로 검사하기

for key, threshold in QUALITY_THRESHOLDS.items():
    passed = metrics[key] >= threshold
    status = "PASS" if passed else "FAIL"
    print(f"{key:<10} {metrics[key]:.4f} >= {threshold:.4f} [{status}]")

임계값 검사는 문서 안의 수동 비교가 아니라 코드로 수행됩니다.

동일한 기준을 모든 납품 실행에 반복 적용할 수 있습니다.

4. 인증서 리포트 만들기

report = ctx.build_snapshot_report(RUN_REF)
print(f"Report: {report.title}")

build_snapshot_report()는 실행 증거를 제출 가능한 구조로 묶습니다.

Tom은 인증서를 만들기 위해 로그와 노트북을 뒤지는 대신, 저장된 실행 기록에서 바로 품질 증거를 조립합니다.

최종 답변

Contexta를 통해, Tom은 조달팀의 질문에 이런 식으로 답변할 수 있습니다.

• Q1. 사용된 학습 데이터 버전은 무엇인가요?

  • A1. 사용된 학습 데이터 버전은 fraud-transactions-2025q1-v3입니다.

• Q2. 평가 메트릭의 정확한 값은 무엇인가요?

  • A2. accuracy=0.9340, auc=0.9610, f1=0.9220, precision=0.9170, recall=0.9280입니다.

• Q3. 학습 환경은 무엇인가요?

  • A3. Python 3.11.8이며, 패키지 6개가 기록되어 있습니다.

• Q4. 합의된 품질 임계값을 통과했나요?

  • A4. 네. accuracy, auc, f1, precision, recall이 모두 합의된 임계값을 통과했습니다.

• Q5. 전체 판정은 PASS인가요, FAIL인가요?

  • A5. 전체 판정은 PASS입니다.

따라서, Tom은 FinanceBank Corp에 다음과 같이 제출할 수 있습니다.

fraud-model-v2-1은 fraud-transactions-2025q1-v3 데이터로 학습됐고, 모든 합의 품질 기준을 통과했습니다.
평가 메트릭과 환경 정보는 실행 기록에서 생성된 것이므로, 이 모델은 납품 승인 후보로 제출할 수 있습니다.

선택: 예제 데이터 생성

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

아래 데이터 준비 코드는 .contexta 워크스페이스에 납품 대상 모델 실행과 환경, 데이터셋, 평가 메트릭 기록을 생성합니다.

seed_certificate_data.py

"""Create delivery-certificate records used by the certificate case study."""

from __future__ import annotations

import tempfile
from datetime import datetime
from pathlib import Path
from typing import Any

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


PROJECT_NAME = "fraud-detection-financebank"

# Quality thresholds agreed with the client
QUALITY_THRESHOLDS: dict[str, float] = {
    "accuracy":  0.90,
    "auc":       0.93,
    "f1":        0.88,
    "precision": 0.87,
    "recall":    0.86,
}

CLIENT_NAME = "FinanceBank Corp"
MODEL_NAME = "FraudShield v2.1"
DELIVERY_DATE = "2025-04-11"

_REC_COUNTER = 0


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


def run_example(workspace: Path | str | None = None) -> dict[str, Any]:
    """Create one run with metrics, dataset, and environment records."""

    if workspace is None:
        root = Path(tempfile.mkdtemp(prefix="contexta-case12-"))
        workspace_path = root / ".contexta"
    else:
        workspace_path = Path(workspace)

    ctx = Contexta(
        config=UnifiedConfig(
            project_name=PROJECT_NAME,
            workspace=WorkspaceConfig(root_path=workspace_path),
        )
    )

    store = ctx.metadata_store
    try:
        store.projects.put_project(
            Project(
                project_ref=f"project:{PROJECT_NAME}",
                name=PROJECT_NAME,
                created_at="2025-03-01T00:00:00Z",
                description=f"Fraud detection model for {CLIENT_NAME}",
            )
        )

        run_name = "fraud-model-v2-1"
        run_ref = f"run:{PROJECT_NAME}.{run_name}"
        started_at = "2025-04-08T09:00:00Z"
        ended_at = "2025-04-08T15:00:00Z"

        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,
            )
        )

        train_ref = f"stage:{PROJECT_NAME}.{run_name}.train"
        eval_ref = f"stage:{PROJECT_NAME}.{run_name}.evaluate"

        store.stages.put_stage_execution(
            StageExecution(
                stage_execution_ref=train_ref,
                run_ref=run_ref,
                stage_name="train",
                status="completed",
                started_at=started_at,
                ended_at="2025-04-08T13:00:00Z",
                order_index=0,
            )
        )
        store.stages.put_stage_execution(
            StageExecution(
                stage_execution_ref=eval_ref,
                run_ref=run_ref,
                stage_name="evaluate",
                status="completed",
                started_at="2025-04-08T13:00:00Z",
                ended_at=ended_at,
                order_index=1,
            )
        )

        # Evaluation metrics
        eval_metrics: dict[str, float] = {
            "accuracy":  0.934,
            "auc":       0.961,
            "f1":        0.922,
            "precision": 0.917,
            "recall":    0.928,
        }
        record_store = ctx.record_store
        obs_ts = ended_at
        for key, val in eval_metrics.items():
            record_store.append(
                MetricRecord(
                    envelope=RecordEnvelope(
                        record_ref=f"record:{PROJECT_NAME}.{run_name}.{_next_rid()}",
                        record_type="metric",
                        recorded_at=obs_ts,
                        observed_at=obs_ts,
                        producer_ref="contexta.case12",
                        run_ref=run_ref,
                        stage_execution_ref=eval_ref,
                        completeness_marker="complete",
                        degradation_marker="none",
                    ),
                    payload=MetricPayload(
                        metric_key=key,
                        value=val,
                        value_type="float64",
                    ),
                )
            )

        # Dataset version event (answers client Q1)
        dataset_version = "fraud-transactions-2025q1-v3"
        record_store.append(
            StructuredEventRecord(
                envelope=RecordEnvelope(
                    record_ref=f"record:{PROJECT_NAME}.{run_name}.{_next_rid()}",
                    record_type="event",
                    recorded_at=started_at,
                    observed_at=started_at,
                    producer_ref="contexta.case12",
                    run_ref=run_ref,
                    completeness_marker="complete",
                    degradation_marker="none",
                ),
                payload=StructuredEventPayload(
                    event_key="training.dataset-registered",
                    level="info",
                    message=f"Training dataset version: {dataset_version}",
                    attributes={
                        "dataset_version": dataset_version,
                        "record_count": 2_400_000,
                        "date_range": "2024-01-01 to 2025-03-31",
                    },
                    origin_marker="explicit_capture",
                ),
            )
        )

        # Environment snapshot (answers client Q3)
        env_packages = {
            "scikit-learn": "1.4.0",
            "xgboost": "2.0.3",
            "pandas": "2.2.1",
            "numpy": "1.26.4",
            "imbalanced-learn": "0.12.0",
            "shap": "0.45.0",
        }
        store.environments.put_environment_snapshot(
            EnvironmentSnapshot(
                environment_snapshot_ref=f"environment:{PROJECT_NAME}.{run_name}.snap",
                run_ref=run_ref,
                captured_at=started_at,
                python_version="3.11.8",
                platform="linux",
                packages=env_packages,
                environment_variables={},
            )
        )

        return {
            "client": CLIENT_NAME,
            "model": MODEL_NAME,
            "run_id": run_ref,
            "dataset_version": dataset_version,
        }
    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()

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

uv run seed_certificate_data.py