본문으로 건너뛰기

Case 04 : 배포 모델의 출처를 찾아봅시다


페르소나​


Carlos는 CTR 랭킹 모델을 담당하는 머신러닝 엔지니어입니다.

금요일 오후에 모델을 배포했고, 월요일 아침 프로덕트 매니저 Johnny에게서 CTR이 18% 하락했다는 연락을 받았습니다.

Carlos가 즉시 답해야 하는 질문은 네 가지입니다.

  • 어떤 학습 실행이 현재 배포된 모델을 만들었나요?
  • 그 실행의 학습 메트릭은 무엇이었나요?
  • 어떤 데이터셋 버전으로 학습했나요?
  • 롤백한다면 정확히 어떤 실행과 배포로 돌아가야 하나요?

상황​


Carlos의 배포 메모에는 model_20250401.pkl이라는 파일명만 남아 있습니다.

파일명만으로는 학습 실행, 메트릭, 데이터셋 버전, 이전 안전 배포를 알 수 없습니다.


이 케이스의 핵심은 배포 레코드를 학습 실행과 연결해, 사고 시점에 모델 출처와 롤백 대상을 즉시 확인하는 것입니다.


Contexta 없이 해결하려면​


Contexta 없이 월요일 사고에 답하려면, Carlos는 보통 아래 작업을 직접 해야 합니다.

  1. 배포 노트에서 모델 파일명을 확인합니다.
  2. 해당 파일을 만든 학습 스크립트나 노트북을 찾습니다.
  3. Git 로그와 Slack 대화를 검색해 어떤 실행이 배포됐는지 추정합니다.
  4. 학습 당시 메트릭과 데이터셋 버전을 다시 찾습니다.
  5. 이전 배포 파일이 정말 안전한 기준선인지 확인합니다.

이 과정은 시간이 걸릴 뿐 아니라, 롤백이 단순히 직전의 파일로 되돌리는 작업으로 이어질 수 있습니다.


Contexta로 해결하기​


Carlos는 배포를 DeploymentExecution으로 기록하고, 각 배포를 정확한 학습 실행과 연결합니다.

deployment:prod-deploy-april
└─ run:friday-run-c
   ├─ train/accuracy=0.901
   ├─ train/auc=0.938
   └─ event: training.dataset-registered = v2025-03-31

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

단계Contexta API얻는 정보
배포 목록 읽기ctx.list_deployments(PROJECT_NAME)어떤 배포가 어떤 실행을 참조하는지
배포 실행 스냅샷 읽기ctx.get_run_snapshot(DEPLOYED_RUN)메트릭, 스테이지, 데이터셋 이벤트
기준선과 비교하기ctx.compare_runs(SAFE_RUN, DEPLOYED_RUN)이전 안전 배포와 현재 배포의 차이

예제 코드​


아래 코드는 이미 저장된 배포 기록과 학습 실행을 읽어, 현재 배포의 출처를 추적합니다.

deployment_traceability_review.py
"""Trace a previously recorded deployment back to its training run."""

from pathlib import Path

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


PROJECT_NAME = "ctr-ranking-model"
DEPLOYED_RUN = f"run:{PROJECT_NAME}.friday-run-c"
SAFE_RUN = f"run:{PROJECT_NAME}.friday-run-b"

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

store = ctx.metadata_store
try:
    print("Deployments:")
    for deployment in ctx.list_deployments(PROJECT_NAME):
        print(f"{deployment.deployment_id} -> {deployment.run_id}")

    snapshot = ctx.get_run_snapshot(DEPLOYED_RUN)
    print(f"\nCurrently deployed run: {snapshot.run.name}")
    for record in snapshot.records:
        if record.record_type == "metric":
            print(f"{record.key}: {record.value:.4f}")
        if record.record_type == "event" and record.key == "training.dataset-registered":
            print(record.message)

    print("\nComparison: safe baseline -> deployed")
    comparison = ctx.compare_runs(SAFE_RUN, DEPLOYED_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}")
finally:
    store.close()

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

Deployments:
deployment:ctr-ranking-model.prod-deploy-april -> run:ctr-ranking-model.friday-run-c
deployment:ctr-ranking-model.prod-deploy-march -> run:ctr-ranking-model.friday-run-b

Currently deployed run: friday-run-c
accuracy: 0.9010
auc: 0.9380
loss: 0.2610
Training dataset version: v2025-03-31

Comparison: safe baseline -> deployed
train/accuracy: +0.0080
train/auc: +0.0110
train/loss: -0.0200

코드 조각별로 이해하기​


1. 비교할 실행 참조 정하기​


PROJECT_NAME = "ctr-ranking-model"
DEPLOYED_RUN = f"run:{PROJECT_NAME}.friday-run-c"
SAFE_RUN = f"run:{PROJECT_NAME}.friday-run-b"

분석 코드는 현재 배포된 실행과 이전 안전 기준선 실행을 명시적으로 지정합니다.

DEPLOYED_RUN은 월요일 사고 조사 대상이고, SAFE_RUN은 이전 프로덕션 배포가 참조하던 실행입니다.


2. 배포 목록에서 실행 연결 확인하기​


print("Deployments:")
for deployment in ctx.list_deployments(PROJECT_NAME):
    print(f"{deployment.deployment_id} -> {deployment.run_id}")

list_deployments()는 저장된 배포 기록을 반환합니다.

출력에서 prod-deploy-april이 friday-run-c를, prod-deploy-march가 friday-run-b를 가리키는 것을 확인할 수 있습니다.


3. 배포된 실행의 증거 읽기​


snapshot = ctx.get_run_snapshot(DEPLOYED_RUN)
print(f"\nCurrently deployed run: {snapshot.run.name}")
for record in snapshot.records:
    if record.record_type == "metric":
        print(f"{record.key}: {record.value:.4f}")
    if record.record_type == "event" and record.key == "training.dataset-registered":
        print(record.message)

스냅샷에는 실행 메트릭과 데이터셋 등록 이벤트가 함께 들어 있습니다.

Carlos는 “현재 배포 모델은 friday-run-c에서 왔고, v2025-03-31 데이터셋으로 학습됐다”고 말할 수 있습니다.


4. 이전 안전 배포와 비교하기​


print("\nComparison: safe baseline -> deployed")
comparison = ctx.compare_runs(SAFE_RUN, DEPLOYED_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}")

compare_runs()는 현재 배포 실행과 이전 안전 기준선의 메트릭 차이를 보여 줍니다.

이 예제에서는 오프라인 AUC는 더 좋아졌지만 데이터셋 버전이 바뀌었기 때문에, CTR 하락 원인 후보를 데이터 품질 쪽으로 좁힐 수 있습니다.


최종 답변​


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

  • Q1. 어떤 학습 실행이 현재 배포된 모델을 만들었나요?

    • A1. 현재 배포 prod-deploy-april은 run:ctr-ranking-model.friday-run-c를 가리킵니다.

  • Q2. 그 실행의 학습 메트릭은 무엇이었나요?

    • A2. friday-run-c의 메트릭은 accuracy=0.9010, auc=0.9380, loss=0.2610입니다.

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

    • A3. 현재 배포된 실행은 v2025-03-31 데이터셋으로 학습되었습니다.

  • Q4. 롤백한다면 정확히 어떤 실행과 배포로 돌아가야 하나요?

    • A4. 이전 안전 배포는 prod-deploy-march이고, 이 배포는 run:ctr-ranking-model.friday-run-b를 가리킵니다.

따라서, Carlos는 월요일 사고 대응에서 다음과 같이 답할 수 있습니다.

현재 모델은 friday-run-c에서 왔고, 오프라인 메트릭은 이전 안전 기준선보다 나쁘지 않습니다.
다만 데이터셋이 v2025-03-31로 바뀌었으므로, 우선 조사 대상은 새 데이터셋이며 필요하면 prod-deploy-march로 롤백하겠습니다.


선택: 예제 데이터 생성​


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

아래 데이터 준비 코드는 .contexta 워크스페이스에 금요일 학습 실행 3개와 배포 레코드 2개를 생성합니다.


deployment_traceability_seed.py
"""Create deployment-traceability records used by the traceability 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,
    StructuredEventPayload,
    StructuredEventRecord,
)


PROJECT_NAME = "ctr-ranking-model"

_REC_COUNTER = 0


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


def _create_training_run(
    store: Any,
    record_store: Any,
    project_name: str,
    run_name: str,
    accuracy: float,
    auc: float,
    loss: float,
    dataset_version: 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,
        )
    )

    feat_stage_ref  = f"stage:{project_name}.{run_name}.feature-engineering"
    train_stage_ref = f"stage:{project_name}.{run_name}.train"

    # feat stage ends halfway between started_at and ended_at (simple midpoint by string isn't safe — use fixed offsets)
    from datetime import datetime, timedelta, timezone
    _s = datetime.fromisoformat(started_at.replace("Z", "+00:00"))
    _e = datetime.fromisoformat(ended_at.replace("Z", "+00:00"))
    _mid = _s + (_e - _s) / 2
    feat_ended = _mid.strftime("%Y-%m-%dT%H:%M:%SZ")
    store.stages.put_stage_execution(
        StageExecution(
            stage_execution_ref=feat_stage_ref,
            run_ref=run_ref,
            stage_name="feature-engineering",
            status="completed",
            started_at=started_at,
            ended_at=feat_ended,
            order_index=0,
        )
    )
    store.stages.put_stage_execution(
        StageExecution(
            stage_execution_ref=train_stage_ref,
            run_ref=run_ref,
            stage_name="train",
            status="completed",
            started_at=feat_ended,
            ended_at=ended_at,
            order_index=1,
        )
    )

    obs_ts = ended_at
    for key, val in [("accuracy", accuracy), ("auc", auc), ("loss", loss)]:
        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.case04",
                    run_ref=run_ref,
                    stage_execution_ref=train_stage_ref,
                    completeness_marker="complete",
                    degradation_marker="none",
                ),
                payload=MetricPayload(
                    metric_key=key,
                    value=val,
                    value_type="float64",
                    aggregation_scope="run",
                ),
            )
        )

    # Log dataset version as a structured event on the run (no stage context)
    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.case04",
                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",
            ),
        )
    )

    return run_ref


def _create_deployment(
    store: Any,
    project_name: str,
    deploy_name: str,
    run_ref: str,
    started_at: str,
    ended_at: str,
    order_index: int,
) -> str:
    deploy_ref = f"deployment:{project_name}.{deploy_name}"
    store.deployments.put_deployment_execution(
        DeploymentExecution(
            deployment_execution_ref=deploy_ref,
            project_ref=f"project:{project_name}",
            deployment_name=deploy_name,
            status="completed",
            started_at=started_at,
            ended_at=ended_at,
            order_index=order_index,
            run_ref=run_ref,
        )
    )
    return deploy_ref


def run_example(workspace: Path | str | None = None) -> dict[str, Any]:
    """Create three training runs and two deployment records."""

    if workspace is None:
        root = Path(tempfile.mkdtemp(prefix="contexta-case04-"))
        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="Click-through rate ranking model",
            )
        )

        # Three training runs from Friday (a/b/c experiments before final deploy)
        run_a_ref = _create_training_run(
            store, ctx.record_store, PROJECT_NAME, "friday-run-a",
            accuracy=0.881, auc=0.912, loss=0.308,
            dataset_version="v2025-03-28",
            started_at="2025-04-01T08:00:00Z",
            ended_at="2025-04-01T09:30:00Z",
        )
        run_b_ref = _create_training_run(
            store, ctx.record_store, PROJECT_NAME, "friday-run-b",
            accuracy=0.893, auc=0.927, loss=0.281,
            dataset_version="v2025-03-28",
            started_at="2025-04-01T09:45:00Z",
            ended_at="2025-04-01T11:15:00Z",
        )
        # run-c is what actually got deployed - best offline AUC
        run_c_ref = _create_training_run(
            store, ctx.record_store, PROJECT_NAME, "friday-run-c",
            accuracy=0.901, auc=0.938, loss=0.261,
            dataset_version="v2025-03-31",  # newer dataset - might explain the CTR drop
            started_at="2025-04-01T12:00:00Z",
            ended_at="2025-04-01T13:45:00Z",
        )

        # Previous (safe) deployment linked to run-b
        _create_deployment(
            store, PROJECT_NAME, "prod-deploy-march",
            run_b_ref,
            started_at="2025-03-28T17:00:00Z",
            ended_at="2025-03-28T17:10:00Z",
            order_index=0,
        )

        # Friday deployment linked to run-c (the one with CTR drop)
        friday_deploy_ref = _create_deployment(
            store, PROJECT_NAME, "prod-deploy-april",
            run_c_ref,
            started_at="2025-04-01T16:00:00Z",
            ended_at="2025-04-01T16:08:00Z",
            order_index=1,
        )

        return {
            "run_ids": [run_a_ref, run_b_ref, run_c_ref],
            "deployment_ids": [
                f"deployment:{PROJECT_NAME}.prod-deploy-march",
                friday_deploy_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_traceability_seed.py로 저장한 뒤, Contexta가 설치된 환경에서 실행하세요.

uv run deployment_traceability_seed.py