본문으로 건너뛰기

Case 06 : 감사 증거를 모아봅시다


페르소나​


Elena는 규제 대상 보험 클라이언트에게 AI 솔루션을 납품하는 솔루션 아키텍트입니다.

클라이언트의 감사관은 프로덕션 모델에 대해 문서화된 증거를 요구합니다.

  • 어떤 데이터셋 버전으로 학습했나요?
  • 평가 메트릭의 정확한 값은 무엇인가요?
  • 학습 당시 Python과 라이브러리 환경은 무엇이었나요?
  • 이전 모델과 무엇이 달라졌나요?
  • 제출 가능한 감사 문서를 만들 수 있나요?

상황​


팀은 Git 로그, 개인 노트북, Slack 스레드, 공유 드라이브를 이틀 동안 검색했습니다.

일부 정보는 찾지 못해 추정으로 채웠고, 감사관은 이를 거부했습니다.

“추정이 아니라 학습 시점에 기록된 증거를 제출하세요.”


이 케이스의 핵심은 감사 질문의 답을 실행 시점에 저장된 레코드에서 재구성하는 것입니다.


Contexta 없이 해결하려면​


Contexta 없이 감사 패키지를 만들려면 보통 아래 작업을 직접 해야 합니다.

  1. 데이터셋 버전이 적힌 노트북 셀이나 Slack 메시지를 찾습니다.
  2. 평가 메트릭이 출력된 로그나 스크린샷을 찾습니다.
  3. 학습 당시 환경을 requirements.txt와 Dockerfile로 추정합니다.
  4. 이전 모델과 현재 모델의 메트릭 차이를 수동으로 비교합니다.
  5. Word나 PDF 문서에 근거를 복사해 붙입니다.

이 방식은 감사 가능성이 약합니다. 검색된 파일이 실제 학습 실행과 연결된 증거인지 확인하기 어렵기 때문입니다.


Contexta로 해결하기​


Elena는 실행 스냅샷, 환경 감사, 실행 비교, 스냅샷 리포트를 사용해 감사 질문에 답합니다.

run:production-model
├─ event: training.dataset-registered
├─ metric: auc, precision, recall, f1
├─ environment: python, packages
└─ report: snapshot report

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

단계Contexta API얻는 정보
실행 증거 읽기ctx.get_run_snapshot(CURRENT_RUN)데이터셋 이벤트와 평가 메트릭
환경 감사ctx.audit_reproducibility(CURRENT_RUN)Python 버전과 패키지 수
이전 버전과 비교ctx.compare_runs(...)이전 모델 대비 메트릭 차이
제출 문서 생성ctx.build_snapshot_report(CURRENT_RUN)감사 패키지의 구조화된 섹션

예제 코드​


아래 코드는 데이터 준비 단계에서 만든 이전 실행과 현재 실행을 읽고, 감사 질문에 답할 수 있는 증거를 출력합니다.

analyze_compliance_audit.py
"""Assemble audit evidence from previously recorded model runs."""

from pathlib import Path

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


PROJECT_NAME = "loss-ratio-predictor"
PREVIOUS_RUN = f"run:{PROJECT_NAME}.model-v1"
CURRENT_RUN = f"run:{PROJECT_NAME}.model-v2"

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

store = ctx.metadata_store
try:
    snapshot = ctx.get_run_snapshot(CURRENT_RUN)
    print(f"Audit target: {snapshot.run.name}")
    for record in snapshot.records:
        if record.record_type == "event" and record.key == "training.dataset-registered":
            print(f"Dataset: {record.message}")
        if record.record_type == "metric":
            print(f"{record.key}: {record.value:.4f}")

    audit = ctx.audit_reproducibility(CURRENT_RUN)
    print(f"\nEnvironment: python={audit.python_version}, packages={audit.package_count}")

    print("\nMetric deltas: model-v1 -> model-v2")
    comparison = ctx.compare_runs(PREVIOUS_RUN, CURRENT_RUN)
    for stage in comparison.stage_comparisons:
        for delta in stage.metric_deltas:
            if delta.delta is not None:
                print(f"{stage.stage_name}/{delta.metric_key}: {delta.delta:+.4f}")

    report = ctx.build_snapshot_report(CURRENT_RUN)
    print(f"\nReport: {report.title}")
finally:
    store.close()

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

Audit target: model-v2
auc: 0.8910
precision: 0.8520
recall: 0.8410
f1: 0.8460
Dataset: Training dataset version: claims-2025q1

Environment: python=3.11.5, packages=4

Metric deltas: model-v1 -> model-v2
evaluate/auc: +0.0200
evaluate/f1: +0.0200
evaluate/precision: +0.0180
evaluate/recall: +0.0220

Report: Run Snapshot Report: run:loss-ratio-predictor.model-v2

코드 조각별로 이해하기​


1. 데이터셋과 메트릭 증거 읽기​


snapshot = ctx.get_run_snapshot(CURRENT_RUN)
for record in snapshot.records:
    if record.record_type == "event" and record.key == "training.dataset-registered":
        print(f"Dataset: {record.message}")
    if record.record_type == "metric":
        print(f"{record.key}: {record.value:.4f}")

스냅샷은 실행에 연결된 이벤트와 메트릭을 함께 제공합니다.

감사 답변은 별도 문서의 기억이 아니라, 해당 실행의 원본 기록에서 나옵니다.


2. 환경 감사하기​


audit = ctx.audit_reproducibility(CURRENT_RUN)
print(f"\nEnvironment: python={audit.python_version}, packages={audit.package_count}")

환경 감사 결과는 Python 버전과 기록된 패키지 수를 보여 줍니다.

규제 대상 프로젝트에서는 재현 가능 여부 자체가 품질 증거가 됩니다.


3. 이전 버전과 비교하기​


comparison = ctx.compare_runs(PREVIOUS_RUN, CURRENT_RUN)
for stage in comparison.stage_comparisons:
    for delta in stage.metric_deltas:
        if delta.delta is not None:
            print(f"{stage.stage_name}/{delta.metric_key}: {delta.delta:+.4f}")

모델이 이전 버전보다 어떻게 달라졌는지 평가 메트릭 기준으로 설명할 수 있습니다.


4. 공식 리포트 만들기​


report = ctx.build_snapshot_report(CURRENT_RUN)
print(f"\nReport: {report.title}")

build_snapshot_report()는 저장된 실행 기록을 제출 가능한 구조로 묶습니다.

원본 증거를 대체하는 문서가 아니라, 증거를 읽기 쉬운 형태로 정리한 결과입니다.


최종 답변​


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

  • Q1. 어떤 데이터셋 버전으로 학습했나요?

    • A1. 감사 대상 model-v2는 claims-2025q1 데이터셋으로 학습되었습니다.

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

    • A2. auc=0.8910, precision=0.8520, recall=0.8410, f1=0.8460입니다.

  • Q3. 학습 당시 Python과 라이브러리 환경은 무엇이었나요?

    • A3. Python 버전은 3.11.5이고, 패키지 4개가 기록되어 있습니다.

  • Q4. 이전 모델과 무엇이 달라졌나요?

    • A4. model-v1 대비 auc +0.0200, f1 +0.0200, precision +0.0180, recall +0.0220입니다.

  • Q5. 제출 가능한 감사 문서를 만들 수 있나요?

    • A5. 네. Run Snapshot Report: run:loss-ratio-predictor.model-v2가 생성됩니다.

따라서, Elena는 감사관에게 다음과 같이 답할 수 있습니다.

model-v2는 claims-2025q1 데이터셋으로 학습됐고, 평가 메트릭과 환경 정보가 실행 기록에 남아 있습니다.
이전 모델 대비 주요 메트릭이 개선됐으며, 제출용 snapshot report도 생성할 수 있습니다.


선택: 예제 데이터 생성​


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

아래 데이터 준비 코드는 .contexta 워크스페이스에 이전 모델과 현재 감사 대상 모델의 실행, 배포, 환경 기록을 생성합니다.


seed_compliance_audit_data.py
"""Create compliance-audit records used by the audit 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,
    EnvironmentSnapshot,
    MetricPayload,
    MetricRecord,
    Project,
    RecordEnvelope,
    Run,
    StageExecution,
    StructuredEventPayload,
    StructuredEventRecord,
)

PROJECT_NAME = "loss-ratio-predictor"
_rid = 0


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


def _put_metric(record_store: Any, run_name: str, stage_ref: str, key: str, val: float, ts: str) -> None:
    run_ref = f"run:{PROJECT_NAME}.{run_name}"
    record_store.append(
        MetricRecord(
            envelope=RecordEnvelope(
                record_ref=f"record:{PROJECT_NAME}.{run_name}.{_next_rid()}",
                record_type="metric",
                recorded_at=ts,
                observed_at=ts,
                producer_ref="contexta.case06",
                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",
            ),
        )
    )


def _make_run_with_env(
    store: Any,
    record_store: Any,
    run_name: str,
    dataset_version: str,
    metrics: dict[str, float],
    python_version: str,
    packages: dict[str, str],
    started_at: str,
    ended_at: str,
) -> 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,
        )
    )

    # Training stage
    train_ref = f"stage:{PROJECT_NAME}.{run_name}.train"
    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=f"{started_at[:10]}T14:00:00Z",
            order_index=0,
        )
    )

    # Evaluate stage
    eval_ref = f"stage:{PROJECT_NAME}.{run_name}.evaluate"
    store.stages.put_stage_execution(
        StageExecution(
            stage_execution_ref=eval_ref,
            run_ref=run_ref,
            stage_name="evaluate",
            status="completed",
            started_at=f"{started_at[:10]}T14:00:00Z",
            ended_at=ended_at,
            order_index=1,
        )
    )

    for key, val in metrics.items():
        _put_metric(record_store, run_name, eval_ref, key, val, ended_at)

    # Record dataset version as a structured event (answers auditor question 1)
    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.case06",
                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},
                origin_marker="explicit_capture",
            ),
        )
    )

    # Environment snapshot (answers auditor question 3)
    all_packages = {**packages, "scikit-learn": "1.3.0", "pandas": "2.0.3"}
    env = EnvironmentSnapshot(
        environment_snapshot_ref=f"environment:{PROJECT_NAME}.{run_name}.snap",
        run_ref=run_ref,
        captured_at=started_at,
        python_version=python_version,
        platform="linux",
        packages=all_packages,
        environment_variables={},
    )
    store.environments.put_environment_snapshot(env)

    return run_ref


def run_example(workspace: Path | str | None = None) -> dict[str, Any]:
    if workspace is None:
        workspace = Path(tempfile.mkdtemp(prefix="contexta-case06-")) / ".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",
            )
        )

        # Previous model (v1) -- deployed March
        prev_run_ref = _make_run_with_env(
            store, ctx.record_store,
            run_name="model-v1",
            dataset_version="claims-2024q4",
            metrics={"auc": 0.871, "precision": 0.834, "recall": 0.819, "f1": 0.826},
            python_version="3.10.12",
            packages={"torch": "1.13.0", "xgboost": "1.7.6"},
            started_at="2025-03-01T09:00:00Z",
            ended_at="2025-03-01T16: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-03-02T10:00:00Z",
                ended_at="2025-03-02T10:15:00Z",
                run_ref=prev_run_ref,
                order_index=0,
            )
        )

        # Current production model (v2) -- deployed June, under audit
        curr_run_ref = _make_run_with_env(
            store, ctx.record_store,
            run_name="model-v2",
            dataset_version="claims-2025q1",
            metrics={"auc": 0.891, "precision": 0.852, "recall": 0.841, "f1": 0.846},
            python_version="3.11.5",
            packages={"torch": "2.0.1", "xgboost": "2.0.0"},
            started_at="2025-06-01T09:00:00Z",
            ended_at="2025-06-01T16:00:00Z",
        )
        store.deployments.put_deployment_execution(
            DeploymentExecution(
                deployment_execution_ref=f"deployment:{PROJECT_NAME}.prod-v2",
                project_ref=f"project:{PROJECT_NAME}",
                deployment_name="prod-v2",
                status="completed",
                started_at="2025-06-02T10:00:00Z",
                ended_at="2025-06-02T10:15:00Z",
                run_ref=curr_run_ref,
                order_index=1,
            )
        )

        return {
            "previous_run_id": prev_run_ref,
            "current_run_id": curr_run_ref,
            "deployment_ids": [
                f"deployment:{PROJECT_NAME}.prod-v1",
                f"deployment:{PROJECT_NAME}.prod-v2",
            ],
        }

    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_compliance_audit_data.py로 저장한 뒤, Contexta가 설치된 환경에서 실행하세요.

uv run seed_compliance_audit_data.py