본문으로 건너뛰기

Case 11 : 프로젝트 온보딩 자료를 만듭시다


페르소나​


Alex는 이탈률 예측 모델 팀의 리드입니다.

이번 채용에서 신규 ML 엔지니어 Jamie가 합류했고, 첫날에 프로젝트의 현재 상태를 빠르게 이해해야 합니다.

Jamie가 알고 싶은 질문은 다섯 가지입니다.

  • 학습 실행이 몇 개이고 이름은 무엇인가요?
  • 시간에 따라 정확도는 어떻게 변했나요?
  • 어떤 실행이 프로덕션에 배포됐나요?
  • 객관적으로 가장 좋은 실행은 무엇인가요?
  • 전체 실행을 비교한 구조화된 리포트가 있나요?

상황​


프로젝트는 4개월 동안 운영됐습니다.

6개의 학습 실행과 2번의 배포가 있었고, 성능은 여러 번의 재학습을 거치며 변화했습니다.


도구가 없다면 Alex는 Git 로그, Confluence, Slack, 오래된 노트북을 뒤져 반나절짜리 요약 문서를 만들어야 합니다.

그리고 그 문서는 새 실행이 추가되는 순간 오래된 문서가 됩니다.


이 케이스의 핵심은 프로젝트 이력을 수동 문서가 아니라 저장된 실행 기록에서 온디맨드로 재생성하는 것입니다.


Contexta 없이 해결하려면​


Contexta 없이 온보딩 자료를 만들려면 보통 아래 작업을 직접 해야 합니다.

  1. 모든 학습 실행의 이름과 날짜를 찾습니다.
  2. 각 노트북이나 로그에서 최종 정확도를 복사합니다.
  3. 어떤 실행이 배포됐는지 배포 기록과 Slack을 확인합니다.
  4. 가장 좋은 실행을 수동으로 정렬합니다.
  5. 전체 이력을 문서로 정리하고 계속 최신 상태로 유지합니다.

수동 문서는 스냅샷입니다. 프로젝트가 움직이면 문서는 바로 낡습니다.


Contexta로 해결하기​


Alex는 Contexta에서 현재 등록된 실행과 배포를 읽고, 비교 리포트를 즉시 만듭니다.

project:churn-prediction
├─ 6 training runs
├─ 2 deployments
├─ best run by accuracy
└─ multi-run report

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

단계Contexta API얻는 정보
실행 목록 읽기ctx.list_runs(PROJECT_NAME)전체 학습 실행과 상태
배포 목록 읽기ctx.list_deployments(PROJECT_NAME)어떤 실행이 프로덕션에 갔는지
최적 실행 선택ctx.select_best_run(run_ids, "accuracy", stage_name="evaluate")정확도 기준 최고 실행
비교 리포트 생성ctx.build_multi_run_report(run_ids)공유 가능한 프로젝트 이력 요약

예제 코드​


아래 코드는 데이터 준비 단계에서 만든 6개 학습 실행과 2개 배포 기록을 읽고, 신규 엔지니어에게 줄 요약을 생성합니다.

analyze_onboarding.py
"""Build an onboarding summary from previously recorded project history."""

from pathlib import Path

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


PROJECT_NAME = "churn-prediction"

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

store = ctx.metadata_store
try:
    runs = ctx.list_runs(PROJECT_NAME)
    deployments = ctx.list_deployments(PROJECT_NAME)
    run_ids = [run.run_id for run in runs]

    print(f"Runs: {len(runs)}")
    for run in runs:
        print(f"  {run.name}")

    print(f"\nDeployments: {len(deployments)}")
    for deployment in deployments:
        print(f"  {deployment.deployment_id} -> {deployment.run_id}")

    best_run = ctx.select_best_run(run_ids, "accuracy", stage_name="evaluate", higher_is_better=True)
    print(f"\nBest run by accuracy: {best_run}")

    report = ctx.build_multi_run_report(run_ids)
    print(f"Report: {report.title}")
finally:
    store.close()

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

Runs: 6
  churn-v1-jan
  churn-v2-feb
  churn-v3-feb
  churn-v4-mar
  churn-v5-apr
  churn-v6-apr

Deployments: 2
  deployment:churn-prediction.prod-deploy-v3 -> run:churn-prediction.churn-v3-feb
  deployment:churn-prediction.prod-deploy-v6 -> run:churn-prediction.churn-v6-apr

Best run by accuracy: run:churn-prediction.churn-v6-apr
Report: Multi Run Report: 6 runs

코드 조각별로 이해하기​


1. 현재 실행 목록 읽기​


runs = ctx.list_runs(PROJECT_NAME)
for run in runs:
    print(f"  {run.name}")

실행 목록은 별도 온보딩 문서가 아니라 현재 Contexta 워크스페이스의 상태에서 나옵니다.

새 실행이 추가되면 다음 요약에도 자연스럽게 반영됩니다.


2. 배포 이력 확인하기​


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

배포 이력을 보면 어떤 실험이 실제 프로덕션까지 갔는지 구분할 수 있습니다.

신규 엔지니어는 “좋았던 실험”과 “실제로 운영된 모델”을 혼동하지 않게 됩니다.


3. 최적 실행과 전체 리포트 만들기​


run_ids = [run.run_id for run in runs]
best_run = ctx.select_best_run(run_ids, "accuracy", stage_name="evaluate", higher_is_better=True)
report = ctx.build_multi_run_report(run_ids)

select_best_run()은 기준 메트릭을 명시해 최고 실행을 선택합니다.

build_multi_run_report()는 실행 전체를 비교 가능한 구조로 묶어 온보딩 자료로 사용할 수 있게 합니다.


최종 답변​


Contexta를 통해, Alex는 Jamie가 첫날 물어본 질문에 이런 식으로 답변할 수 있습니다.

  • Q1. 학습 실행이 몇 개이고 이름은 무엇인가요?

    • A1. 총 6개입니다.
      이름은 churn-v1-jan, churn-v2-feb, churn-v3-feb, churn-v4-mar, churn-v5-apr, churn-v6-apr입니다.

  • Q2. 시간에 따라 정확도는 어떻게 변했나요?

    • A2. 정확도는 churn-v1-jan의 0.821에서 시작해 churn-v2-feb 0.843, churn-v3-feb 0.861,
      churn-v4-mar 0.878, churn-v5-apr 0.894, churn-v6-apr 0.902로 상승했습니다.

  • Q3. 어떤 실행이 프로덕션에 배포됐나요?

    • A3. 배포는 2번 있었습니다. prod-deploy-v3는 churn-v3-feb를, prod-deploy-v6는 churn-v6-apr를 가리킵니다.

  • Q4. 객관적으로 가장 좋은 실행은 무엇인가요?

    • A4. 정확도를 기준으로 가장 높았던 실행은 run:churn-prediction.churn-v6-apr입니다.

  • Q5. 전체 실행을 비교한 구조화된 리포트가 있나요?

    • A5. 네. Multi Run Report: 6 runs가 생성됩니다.

따라서, Alex는 Jamie에게 다음과 같이 설명할 수 있습니다.

이 프로젝트에는 6개의 학습 실행과 2개의 배포가 있습니다.
정확도는 0.821에서 0.902까지 꾸준히 올랐고, 가장 최신 배포와 정확도 기준 최고 실행은 모두 churn-v6-apr입니다.


선택: 예제 데이터 생성​


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

아래 데이터 준비 코드는 .contexta 워크스페이스에 6개 학습 실행과 2개 배포 기록을 생성합니다.


seed_onboarding_data.py
"""Create project-history records used by the onboarding 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 = "churn-prediction"

_REC_COUNTER = 0

# Chronological run history: (run_name, month_label, started_at, ended_at, accuracy, auc, f1)
_RUN_HISTORY = [
    ("churn-v1-jan", "Jan", "2025-01-10T09:00:00Z", "2025-01-10T12:00:00Z", 0.821, 0.854, 0.811),
    ("churn-v2-feb", "Feb", "2025-02-07T09:00:00Z", "2025-02-07T12:00:00Z", 0.843, 0.872, 0.836),
    ("churn-v3-feb", "Feb", "2025-02-21T09:00:00Z", "2025-02-21T12:00:00Z", 0.861, 0.889, 0.854),
    ("churn-v4-mar", "Mar", "2025-03-14T09:00:00Z", "2025-03-14T12:00:00Z", 0.878, 0.907, 0.871),
    ("churn-v5-apr", "Apr", "2025-04-03T09:00:00Z", "2025-04-03T12:00:00Z", 0.894, 0.921, 0.888),
    ("churn-v6-apr", "Apr", "2025-04-18T09:00:00Z", "2025-04-18T12:00:00Z", 0.902, 0.933, 0.896),  # best
]

# Deployments: (deploy_name, linked_run_index 0-based, started_at, ended_at, order_index)
_DEPLOYMENT_HISTORY = [
    ("prod-deploy-v3", 2, "2025-02-22T15:00:00Z", "2025-02-22T15:12:00Z", 0),
    ("prod-deploy-v6", 5, "2025-04-19T14:00:00Z", "2025-04-19T14:10:00Z", 1),
]


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


def _build_run(
    store: Any,
    record_store: Any,
    project_name: str,
    run_name: str,
    started_at: str,
    ended_at: str,
    accuracy: float,
    auc: float,
    f1: float,
) -> 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,
        )
    )

    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=f"{started_at[:10]}T10:30: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=f"{started_at[:10]}T10:30:00Z",
            ended_at=ended_at,
            order_index=1,
        )
    )

    obs_ts = ended_at
    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=obs_ts,
                    observed_at=obs_ts,
                    producer_ref="contexta.case11",
                    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",
                ),
            )
        )

    # Log a notes event describing what changed in this run
    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.case11",
                run_ref=run_ref,
                completeness_marker="complete",
                degradation_marker="none",
            ),
            payload=StructuredEventPayload(
                event_key="training.run-registered",
                level="info",
                message=f"Training run {run_name} started.",
                origin_marker="explicit_capture",
            ),
        )
    )

    return run_ref


def run_example(workspace: Path | str | None = None) -> dict[str, Any]:
    """Create 6 runs over 4 months and 2 deployment records."""

    if workspace is None:
        root = Path(tempfile.mkdtemp(prefix="contexta-case11-"))
        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-01-01T00:00:00Z",
                description="Customer churn prediction model",
            )
        )

        run_refs: list[str] = []
        for run_name, _month, started, ended, acc, auc, f1 in _RUN_HISTORY:
            ref = _build_run(
                store, ctx.record_store, PROJECT_NAME,
                run_name=run_name,
                started_at=started,
                ended_at=ended,
                accuracy=acc, auc=auc, f1=f1,
            )
            run_refs.append(ref)

        for deploy_name, run_idx, started, ended, order in _DEPLOYMENT_HISTORY:
            store.deployments.put_deployment_execution(
                DeploymentExecution(
                    deployment_execution_ref=f"deployment:{PROJECT_NAME}.{deploy_name}",
                    project_ref=f"project:{PROJECT_NAME}",
                    deployment_name=deploy_name,
                    status="completed",
                    started_at=started,
                    ended_at=ended,
                    run_ref=run_refs[run_idx],
                    order_index=order,
                )
            )

        return {
            "run_ids": run_refs,
            "deployment_count": len(_DEPLOYMENT_HISTORY),
        }
    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_onboarding_data.py로 저장한 뒤, Contexta가 설치된 환경에서 실행하세요.

uv run seed_onboarding_data.py