MINERVA Phase C-4 — 실험 파이프라인 자동화 (가설→실험→분석→배포 루프)

C15~C18을 묶는 운영 루프 — 가설 등록부터 사후 보고서까지 자동화하고, 거버넌스 게이트로 안전을 잡는다

C15(A/B 심화)·C16(Bandit)·C17(세그멘테이션)·C18(개인화)을 운영에서 끊김 없이 돌리려면 파이프라인 자동화가 필수다. 본 편은 9단계 루프(가설 등록·표본 크기 산출·실험 등록·할당·모니터링· 분석·의사결정·배포·사후 보고서)를 정리하고, 자동·반자동·사람 게이트의 분리 기준, 실험 카탈로그·거버넌스, 자주 발생 함정을 다룬다. Phase C-4의 마무리.

Agent
저자

Kwangmin Kim

공개

2026년 05월 06일

1 왜 파이프라인 자동화인가

C15 A/B 심화·C16 Bandit·C17 세그멘테이션·C18 개인화이 모두 잘 돌아가면 한 가지 새 문제가 생긴다 — 운영 부담.

수동 운영 시 마찰 결과
가설 1개당 표본 크기·MDE·검정 선택을 사람이 수동 결정 결정 품질 들쭉날쭉
실험마다 YAML·Bandit·라우팅을 따로 셋업 실험 시작 마찰 → 실험 수 감소
SRM·guardrail 모니터링이 사람 손 놓침 → 실험 결론 신뢰도 ↓
다중 비교 보정·세그먼트별 분석을 ad-hoc으로 분석 품질 비일관
결과 보고서 수기 작성 의사결정 지연
실험 결과가 카탈로그에 기록되지 않음 학습 누적 안 됨

자동화는 결정 품질을 표준화하고 실험 처리량을 늘린다. 작은 회사에서 분기당 5개 실험이 자동화 후 분기당 30개로 늘어난 사례가 흔하다.

2 9단계 파이프라인

[1] 가설 정의      → YAML hypothesis spec
[2] 사전 검증      → 표본 크기·MDE·power 자동 계산
[3] 실험 등록      → 검토자 승인 + 카탈로그 등록
[4] 할당           → sticky hash + holdout + audience filter (C17 segment)
[5] 실행 + 라우팅   → Bandit (C16) 또는 고정 비율 A/B
[6] 모니터링       → SRM·Sequential·Guardrail 실시간 알림
[7] 분석           → 검정 자동 선택 + 다중 비교 보정 + CUPED
[8] 의사결정       → 자동/반자동/사람 게이트 분기
[9] 사후 보고서    → 카탈로그 archive + 학습 누적

각 단계가 중단점: 다음 단계로 넘어갈 때 명시적 검증을 통과해야 함.

3 단계 1 — 가설 정의 (YAML)

# experiments/exp_2026Q2_001.yaml
id: exp_2026Q2_001
title: "BGE-large reranker 도입 (R&D 부서 한정)"
hypothesis: |
  R&D 엔지니어 세그먼트에서 reranker를 BGE-base → BGE-large로 교체하면
  답변 정확도(thumbs_up_rate)가 baseline 대비 +3pp 개선된다.

owner: kmkim@example.com
reviewers: [team-lead, exp-governance]

audience:
  segment: rnd_engineer            # C17 세그먼트로 한정
  exclude_users: [internal_test_*]
  min_segment_size: 800            # 표본 부족 시 실험 등록 거부

metrics:
  primary: thumbs_up_rate
  guardrail:
    - p95_latency_ms
    - cost_per_query_usd
  exploratory:
    - feedback_rate
    - response_length

design:
  alpha: 0.05
  power: 0.8
  mde: 0.03
  variance_baseline: 0.247         # p(1-p) for p=0.55
  test_method: welch_t_test
  multiple_correction: bonferroni  # guardrail 보정
  cuped_covariate: thumbs_up_rate_pre_30d   # CUPED 적용

arms:
  control:
    reranker: bge_base
  treatment:
    reranker: bge_large

allocation:
  type: bandit                     # bandit | fixed_ab
  bandit_algo: thompson
  holdout_fraction: 0.05           # A/B 검증용

duration:
  min_days: 7
  max_days: 21
  early_stop:
    method: pocock
    n_looks: 5

스키마 자체가 거버넌스 — 누락된 필드는 자동 거부.

4 단계 2 — 사전 검증

# scripts/exp_precheck.py
from pydantic import BaseModel, ValidationError
import yaml

class ExperimentSpec(BaseModel):
    id: str
    hypothesis: str
    audience: dict
    metrics: dict
    design: dict
    arms: dict
    allocation: dict
    duration: dict


def precheck(path: str):
    with open(path) as f:
        spec = ExperimentSpec(**yaml.safe_load(f))

    # 1. 표본 크기 계산
    n_required = required_sample_per_arm(
        p_baseline=0.55, mde=spec.design["mde"],
        alpha=spec.design["alpha"], power=spec.design["power"],
    )

    # 2. audience 크기 확인
    audience_size = count_segment_users(spec.audience["segment"])
    if audience_size < n_required * len(spec.arms):
        raise ValueError(
            f"audience {audience_size} < required {n_required * len(spec.arms)}"
        )

    # 3. guardrail이 primary와 같은 메트릭이면 안 됨
    if spec.metrics["primary"] in spec.metrics["guardrail"]:
        raise ValueError("primary와 guardrail이 같은 메트릭")

    # 4. arm 이름 충돌
    if "control" not in spec.arms:
        raise ValueError("control arm 필수")

    return {
        "audience_size": audience_size,
        "required_per_arm": n_required,
        "estimated_duration_days": estimate_duration(audience_size, n_required),
    }

PR에 이 결과를 자동 코멘트 → 검토자가 통과 후 머지.

5 단계 3 — 실험 등록과 거버넌스

# scripts/exp_register.py
def register(spec_path: str):
    spec = load_spec(spec_path)
    precheck_result = precheck(spec_path)

    # DB 저장
    db.experiments.insert({
        "id": spec.id,
        "spec": spec.model_dump(),
        "status": "pending_approval",
        "precheck": precheck_result,
        "owner": spec.owner,
        "registered_at": datetime.utcnow(),
    })

    # 검토자에게 Slack 알림
    notify(spec.reviewers, spec)

승인 절차: - Auto-approve: 작은 변경 (프롬프트 미세 조정, 카탈로그 라벨) - Lead review: 표준 실험 (새 모델·새 reranker) - Governance review: 위험 실험 (가격 변경·외부 노출·데이터 정책 영향)

승인 매트릭스를 spec metadata로 자동 분류 — 사람이 결정하지 않아도 라우팅.

6 단계 4 — 할당

# app/experiments/allocator.py
def assign_arm(user_id: str, spec: ExperimentSpec) -> str:
    # 1. audience filter (C17 segment)
    segment = load_segment(user_id)
    if segment.get(spec.audience["segment"]) is None:
        return None                    # 실험 대상 아님

    # 2. holdout 분리
    if sticky_hash(user_id, spec.id + "_holdout") < spec.allocation["holdout_fraction"]:
        return "control"               # holdout = 항상 control

    # 3. Bandit 또는 고정 A/B
    if spec.allocation["type"] == "bandit":
        ctx = context_vector(segment)
        return spec.arms_list[bandit_router(spec.id).select(ctx)]

    return sticky_hash_arm(user_id, spec.id, spec.arms_list)

audience에 들지 않으면 None — 그 사용자는 일반 default 처리. 실험 외부 사용자가 noise로 들어가지 않음.

7 단계 5 — 실행과 라우팅 통합

06편 라우터에 실험 등록 hook:

# app/agents/router.py
def route_request(query: Query) -> Response:
    active_exps = db.experiments.find({"status": "running"})
    for spec in active_exps:
        arm = assign_arm(query.user_id, spec)
        if arm:
            apply_arm_overrides(query, spec, arm)   # YAML override 주입

    return run_agent(query)

여러 실험이 동시 실행될 때 arm 충돌 방지 — 같은 컴포넌트(reranker)에 두 실험이 동시에 처치를 주려고 하면 등록 단계에서 거부.

def precheck_conflicts(spec):
    overlapping = db.experiments.find({
        "status": "running",
        "audience.segment": spec.audience["segment"],
        "arms_modify": {"$in": list(spec.arms_modify)},   # 동일 컴포넌트 변경
    })
    if overlapping:
        raise ValueError(f"conflict with {[e.id for e in overlapping]}")

8 단계 6 — 모니터링 루프

매시간(또는 실시간) 점검:

# scripts/exp_monitor.py — cron 또는 Airflow DAG
def monitor_running_experiments():
    for spec in db.experiments.find({"status": "running"}):
        snapshot = collect_snapshot(spec.id)

        # 1. SRM 검정
        srm_p = chi_square_srm(snapshot["arm_counts"], spec.allocation["expected_ratios"])
        if srm_p < 0.001:
            page(spec.owner, f"SRM violation in {spec.id}: p={srm_p:.4f}")
            update_status(spec.id, "paused_srm")

        # 2. Guardrail 검정 (Bonferroni)
        for metric in spec.metrics["guardrail"]:
            p = welch_test(snapshot["arms"], metric)
            if p < 0.05 / len(spec.metrics["guardrail"]):
                if is_worse(snapshot, metric):
                    page(spec.owner, f"Guardrail breach: {metric}")
                    update_status(spec.id, "paused_guardrail")

        # 3. Sequential test (Pocock)
        if spec.duration["early_stop"]["method"] == "pocock":
            n_looks = spec.duration["early_stop"]["n_looks"]
            boundary = 0.05 / np.sqrt(n_looks)
            p = welch_test(snapshot["arms"], spec.metrics["primary"])
            if p < boundary:
                signal_early_stop(spec.id, p)

이 루프가 사람의 모니터링 부담을 0에 가깝게 만든다 — 알림이 올 때만 개입.

9 단계 7 — 자동 분석

실험 종료(시간 만료 또는 early stop) 시:

# app/analysis/auto_analyze.py
def analyze(spec_id: str) -> dict:
    spec = load_spec(spec_id)
    data = load_jsonl(f"results/{spec_id}.jsonl")

    report = {"spec_id": spec_id}

    # 1. CUPED 보정
    if covariate := spec.design.get("cuped_covariate"):
        data = apply_cuped(data, covariate)

    # 2. 검정 선택 (분포 따라)
    test_fn = select_test(data, spec.metrics["primary"], spec.design["test_method"])
    report["primary"] = test_fn(data, spec.metrics["primary"])

    # 3. Guardrail (Bonferroni 보정)
    report["guardrail"] = {}
    for metric in spec.metrics["guardrail"]:
        p = welch_test(data, metric)
        report["guardrail"][metric] = {
            "p": p,
            "p_adj": min(p * len(spec.metrics["guardrail"]), 1.0),
        }

    # 4. Exploratory (BH-FDR 보정)
    if spec.metrics.get("exploratory"):
        ps = [welch_test(data, m) for m in spec.metrics["exploratory"]]
        _, p_adj, _, _ = multipletests(ps, method="fdr_bh")
        report["exploratory"] = dict(zip(spec.metrics["exploratory"], p_adj))

    # 5. 세그먼트별 (C17·Simpson's paradox 방어)
    report["by_segment"] = {
        seg: welch_test(data[data.segment == seg], spec.metrics["primary"])
        for seg in segments_in_audience(spec)
    }

    # 6. holdout 검증 (Bandit이 결과를 왜곡 안 했는지)
    holdout = data[data.is_holdout]
    if len(holdout) > 100:
        report["holdout_check"] = welch_test(holdout, spec.metrics["primary"])

    return report

모든 검정·보정·CUPED·세그먼트별 분석이 자동. 분석가는 결과를 해석하는 데 집중.

10 단계 8 — 의사결정 게이트

def decide(report: dict, spec: ExperimentSpec) -> str:
    primary = report["primary"]
    guardrail = report["guardrail"]

    # 1. Guardrail 위반 → reject
    if any(g["p_adj"] < 0.05 and g["worse"] for g in guardrail.values()):
        return "reject"

    # 2. Primary 통과 + 효과 크기 충분 → ship
    if primary["p"] < 0.05 and primary["effect_size"] >= spec.design["mde"]:
        if spec.governance == "auto_ship":
            return "ship_auto"
        return "ship_pending_review"   # 사람 최종 승인

    # 3. 검정력 부족 + 표본 가능 → extend
    if primary["p"] < 0.2 and current_sample < required_sample * 1.5:
        return "extend"

    # 4. 그 외 → reject (효과 없음)
    return "reject"
게이트 조건
auto_ship 카탈로그 라벨·프롬프트 미세 변경 (저위험)
pending_review 모델 변경·라우팅 변경 (중위험)
governance_review 가격·외부·정책 (고위험)

위험도는 spec.governance가 자동 결정.

11 단계 9 — 사후 보고서와 카탈로그

def archive(spec_id: str, report: dict, decision: str):
    db.experiments.update(spec_id, {
        "status": "completed",
        "report": report,
        "decision": decision,
        "completed_at": datetime.utcnow(),
    })

    # Markdown 보고서 생성 + Confluence·GitBook 자동 publish
    markdown = render_report(spec_id, report, decision)
    publish_to_knowledge_base(markdown)

    # 학습 누적 — 실험 카탈로그 업데이트
    update_catalog(spec_id, report)


def render_report(spec_id, report, decision) -> str:
    return f"""
# 실험 {spec_id}

## 결정: {decision}

## Primary metric
- {report['primary']['metric']}: {report['primary']['effect_size']:+.3f}
  (p={report['primary']['p']:.4f}, 95% CI [...])

## Guardrail
{format_guardrail_table(report['guardrail'])}

## Segment-level analysis
{format_segment_table(report['by_segment'])}

## Recommendations
{generate_recommendations(report, decision)}
"""

이 자동 보고서가 회의 시간을 줄인다 — 모두가 같은 기준으로 결과를 본다.

12 거버넌스 — 자동·반자동·사람 분리

# config/governance.yaml
auto_ship:
  - prompt_microcopy
  - fewshot_example
  - retrieval_param_tune

pending_review:
  - reranker_swap
  - llm_model_swap
  - new_skill

governance_review:
  - pricing_logic
  - external_facing_change
  - data_retention_policy
  - security_scope

YAML 룰이 자동 분류. 룰에 없는 카테고리는 자동으로 governance_review로 escalation.

13 자주 발생하는 함정

13.1 Test 인프라 자체의 회귀

파이프라인 코드가 바뀐 후 SRM·CUPED·검정 결과가 미세 달라짐. 영향 큼.

해법: - A/A test를 매주 자동 실행 — 거짓 양성률이 5%인지 확인 - 파이프라인 변경 PR에는 회귀 테스트 강제 (12-1 snapshot 테스트)

13.2 Multi-experiment Interaction

여러 실험이 같은 사용자에게 처치를 동시 적용 → arm 효과가 뒤섞임.

해법: 단계 5의 conflict precheck. 또는 factorial design으로 의도적 결합.

13.3 Drift in baseline

control arm의 baseline이 시간에 따라 변함. 옛 실험 보고서의 baseline과 새 실험의 baseline이 다름.

해법: 공통 holdout — 모든 실험이 5% holdout을 공유. 절대값이 아니라 holdout 대비 lift로 보고.

13.4 Human Override 남용

거버넌스 게이트를 사람이 자주 덮어쓰면(force_ship) 자동화의 신뢰 무너짐. 매분기 override 통계를 리뷰.

13.5 Catalog rotting

오래된 실험 보고서가 카탈로그에 그대로 — 코드는 이미 변경. 새 실험 입안 시 옛 결과를 신뢰하면 안 됨.

해법: - 보고서에 expiration 자동 표시 (예: 90일 후 “재검증 필요”) - “active 결정”은 별도 카탈로그에 (롤백 시 빠른 참조)

14 MINERVA 적용

app/
├── experiments/
│   ├── spec.py              # Pydantic ExperimentSpec
│   ├── precheck.py          # 단계 2
│   ├── register.py          # 단계 3
│   ├── allocator.py         # 단계 4
│   ├── monitor.py           # 단계 6 (cron)
│   ├── auto_analyze.py      # 단계 7
│   └── decide.py            # 단계 8
├── analysis/
│   ├── tests.py             # Welch·MWU·bootstrap
│   ├── cuped.py             # CUPED
│   ├── multipletests.py     # Bonferroni·BH-FDR
│   └── sequential.py        # Pocock·always-valid
└── routing/
    ├── bandit_router.py     # C16 Bandit
    └── personalization.py   # C18 카탈로그 적용

scripts/
├── exp_register.py          # CLI 진입점
├── exp_monitor.py           # cron job
└── exp_analyze.py           # 종료 시 자동 호출

experiments/                  # YAML spec 저장소 (git tracked)
├── exp_2026Q2_001.yaml
├── exp_2026Q2_002.yaml
└── ...

11-0 환경변수·12-0 테스트 fixture 위에 자연스럽게 얹힌다. CI에서 spec validation·precheck·dry-run을 자동 실행 (07-1 GitHub Actions).

15 정리

영역 핵심
9단계 가설→사전 검증→등록→할당→실행→모니터링→분석→의사결정→보고서
표준화 YAML spec이 거버넌스 — 누락 필드 자동 거부
사람 게이트 auto_ship·pending_review·governance_review 위험도별 라우팅
모니터링 SRM·Guardrail·Sequential 자동 → 알림으로만 개입
분석 CUPED·Bonferroni·BH-FDR·세그먼트별 자동
카탈로그 보고서 자동 publish + expiration 표시
함정 인프라 회귀·multi-exp interaction·baseline drift·override 남용·catalog rotting

16 응용 분야

시나리오 본 편 단계
새 reranker 도입 자동 검증 1·2·3·4·5·6·7·8·9 전체
작은 프롬프트 변경 빠른 ship auto_ship 게이트
외부 노출 변경 보수적 검증 governance_review + 14일 min duration
인프라 점검 (A/A 자동) 단계 6 cron의 별도 룰
분기 실험 회고 카탈로그 expiration·세그먼트별 lift 차트

17 관련 주제

선행 학습 (선수)

18-LangGraph 시리즈 cross-reference

  • #22 시스템 프롬프트 평가 — 본 편 단계 7 분석 방법론

후속 (Phase C-5 진입)

  • C20~C23 발화 데이터 분석 — 단계 6·9에서 수집된 대화 로그를 더 깊이 활용
  • Phase C-9 관측성과 비용 — 본 편 단계 6 모니터링이 운영 가시성 토대

Cross-reference (운영)

Subscribe

Enjoy this blog? Get notified of new posts by email: