MINERVA A/B 실험 프레임워크

1 왜 A/B 실험이 필요한가

RAG 파이프라인에는 조정할 수 있는 파라미터가 많다: 청크 크기, top-k, reranker 모델, LLM temperature 등. 이 파라미터를 변경하면 성능이 향상될 수도, 저하될 수도 있다. “더 나은 설정”을 찾으려면 두 설정을 동시에 실행하고 사용자 피드백과 메트릭으로 비교해야 한다.

MINERVA의 A/B 실험 프레임워크가 해결하는 문제:

• 코드 변경 없이 파이프라인 파라미터를 변형한다 (YAML override)
• 같은 사용자는 항상 같은 실험군에 할당된다 (sticky hash)
• 실험 결과를 자동으로 기록한다 (JSONL 메트릭)
• 실험군별 에이전트 인스턴스를 효율적으로 관리한다 (캐싱)

2 실험 정의: YAML Override

2.1 기본 설정 파일

# data/configs/qna_chatbot.yaml (기본 설정)
chunking:
  chunk_size: 1500
  child_chunk_size: 400
retriever:
  top_k: 20
  search_type: hybrid
reranker:
  model: flashrank
  top_n: 3
llm:
  model: gpt-4.1
  temperature: 0.7

2.2 실험 정의 파일

# data/experiments/reranker_topn_experiment.yaml
name: reranker_topn_experiment
description: "Reranker top_n을 3 vs 5로 비교"
status: active
start_date: 2026-05-01
end_date: 2026-05-31

arms:
  control:
    weight: 0.5
    overrides: {}    # 기본 설정 그대로

  treatment:
    weight: 0.5
    overrides:
      reranker.top_n: 5
      retriever.top_k: 30

overrides는 dotted-key 형식이다. reranker.top_n: 5는 기본 설정의 reranker.top_n을 3에서 5로 변경한다. 나머지 설정은 기본값을 유지한다.

2.3 Dotted-key Override 구현

from copy import deepcopy

def apply_overrides(base_config: dict, overrides: dict) -> dict:
    config = deepcopy(base_config)
    for dotted_key, value in overrides.items():
        keys = dotted_key.split(".")
        target = config
        for key in keys[:-1]:
            target = target[key]
        target[keys[-1]] = value
    return config

# 사용 예
base = {"reranker": {"model": "flashrank", "top_n": 3}, "retriever": {"top_k": 20}}
overrides = {"reranker.top_n": 5, "retriever.top_k": 30}

result = apply_overrides(base, overrides)
# {"reranker": {"model": "flashrank", "top_n": 5}, "retriever": {"top_k": 30}}

이 방식의 이점은 실험 정의가 간결하다는 것이다. 변경할 파라미터만 명시하면 되고, 나머지는 기본값을 상속한다.

3 사용자 할당: Sticky Hash

3.1 왜 Sticky인가

A/B 실험에서 같은 사용자가 요청할 때마다 다른 실험군에 배정되면 결과가 오염된다. Sticky hash는 사용자 ID와 실험 이름을 해시하여 항상 같은 실험군에 할당한다.

import hashlib

def sticky_hash(user_id: str, experiment_name: str, arms: dict) -> str:
    seed = f"{user_id}:{experiment_name}"
    hash_value = int(hashlib.sha256(seed.encode()).hexdigest(), 16)
    normalized = (hash_value % 10000) / 10000  # 0.0 ~ 1.0

    cumulative = 0.0
    for arm_id, arm_config in arms.items():
        cumulative += arm_config["weight"]
        if normalized < cumulative:
            return arm_id

    return list(arms.keys())[-1]

# 동일 사용자는 항상 같은 결과
sticky_hash("user_42", "reranker_topn_experiment", arms)  # "control"
sticky_hash("user_42", "reranker_topn_experiment", arms)  # "control" (동일)

# 다른 사용자는 다를 수 있다
sticky_hash("user_99", "reranker_topn_experiment", arms)  # "treatment"

3.2 할당 비율 검증

해시 기반 할당이 실제로 의도한 비율에 근사하는지 확인한다.

from collections import Counter

results = Counter()
for i in range(10000):
    arm = sticky_hash(f"user_{i}", "test_experiment", arms)
    results[arm] += 1

# Counter({'control': 5023, 'treatment': 4977})
# 50:50에 근사

SHA-256 해시의 균일성 덕분에 사용자 수가 충분하면 의도한 비율에 수렴한다.

4 실험 플로우

전체 요청 처리 흐름에서 실험이 어떻게 작동하는지 정리한다.

1. 사용자 요청 도착 (user_id: "user_42")
2. 활성 실험 목록 확인
3. sticky_hash("user_42", "reranker_topn_experiment") → "treatment"
4. treatment arm의 overrides 적용 → 변형 RAGConfig 생성
5. AgentCache에서 ("reranker_topn_experiment", "treatment") 키로 에이전트 조회
6. 캐시 미스 → 변형 config로 에이전트 생성 + warmup + 캐시 저장
7. 에이전트 실행 → Response 생성
8. 메트릭 기록 (experiment, arm, latency, user_feedback)
9. Response + experiment_id + arm_id를 프론트엔드에 반환

프론트엔드는 experiment_id와 arm_id를 받아 피드백 전송 시 함께 전달한다. 이로써 어떤 실험군의 응답에 대한 피드백인지 추적할 수 있다.

5 메트릭 수집: JSONL

5.1 기록 형식

{"timestamp": "2026-05-05T14:30:00", "experiment": "reranker_topn_experiment", "arm": "control", "user_id": "user_42", "run_id": "abc-123", "latency_ms": 1500, "input_tokens": 500, "output_tokens": 200}
{"timestamp": "2026-05-05T14:30:05", "experiment": "reranker_topn_experiment", "arm": "treatment", "user_id": "user_99", "run_id": "def-456", "latency_ms": 1800, "input_tokens": 700, "output_tokens": 250}

JSONL(JSON Lines) 형식을 사용하는 이유:

• append-only: 파일 끝에 추가만 하므로 동시 쓰기에 안전하다
• 스트리밍 처리: 파일 전체를 메모리에 올리지 않고 한 줄씩 처리한다
• pandas 호환: pd.read_json(path, lines=True)로 즉시 DataFrame이 된다

5.2 메트릭 기록 구현

import json
from datetime import datetime
from pathlib import Path

class MetricsLogger:
    def __init__(self, base_dir: str = "data/experiments"):
        self.base_dir = Path(base_dir)

    def log(self, experiment: str, arm: str, user_id: str,
            run_id: str, latency_ms: int, **kwargs):
        record = {
            "timestamp": datetime.now().isoformat(),
            "experiment": experiment,
            "arm": arm,
            "user_id": user_id,
            "run_id": run_id,
            "latency_ms": latency_ms,
            **kwargs,
        }

        log_file = self.base_dir / experiment / "metrics.jsonl"
        log_file.parent.mkdir(parents=True, exist_ok=True)

        with open(log_file, "a") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")

5.3 피드백 기록

class FeedbackLogger:
    def log(self, run_id: str, feedback_type: str,
            experiment: str | None = None, arm: str | None = None):
        record = {
            "timestamp": datetime.now().isoformat(),
            "run_id": run_id,
            "feedback": feedback_type,  # "up" or "down"
            "experiment": experiment,
            "arm": arm,
        }

        log_file = self.base_dir / "feedback.jsonl"
        with open(log_file, "a") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")

6 분석

실험 결과를 분석하는 패턴이다.

import pandas as pd
from scipy import stats

# 메트릭 로드
metrics = pd.read_json("data/experiments/reranker_topn_experiment/metrics.jsonl", lines=True)
feedback = pd.read_json("data/experiments/feedback.jsonl", lines=True)

# 실험별 집계
summary = metrics.groupby("arm").agg(
    count=("run_id", "count"),
    avg_latency=("latency_ms", "mean"),
    p95_latency=("latency_ms", lambda x: x.quantile(0.95)),
).round(1)

# 피드백 join
merged = metrics.merge(feedback[["run_id", "feedback"]], on="run_id", how="left")
satisfaction = merged.groupby("arm")["feedback"].apply(
    lambda x: (x == "up").sum() / x.notna().sum()
).round(3)

# 통계 검정
control = metrics[metrics["arm"] == "control"]["latency_ms"]
treatment = metrics[metrics["arm"] == "treatment"]["latency_ms"]
t_stat, p_value = stats.ttest_ind(control, treatment)

6.1 메트릭 유형별 검정 선택

scipy.stats.ttest_ind만으로는 모든 실험 메트릭을 다룰 수 없다. MINERVA의 core/stats.py는 메트릭 유형별 검정을 분리한다.

메트릭 유형	예시	검정 함수
비율 (이진 결과)	인용률(citation_rate), 만족률, 포맷 준수율	two_proportion_z_test
시간 (연속)	latency_ms, ttft_ms	welch_t_test (등분산 가정 없이)
트래픽 분배 검증	50:50 의도 vs 실제 관측 비율	srm_chi_square
효과 크기	처치-대조 차이의 상대 변화	relative_lift

# core/stats.py 사용 예
from core.stats import two_proportion_z_test, welch_t_test, srm_chi_square, relative_lift

# 1. 인용률 비교 (비율 메트릭)
control_cited = (control_df["citation_count"] > 0).sum()
treatment_cited = (treatment_df["citation_count"] > 0).sum()
z, p = two_proportion_z_test(
    success_a=control_cited, n_a=len(control_df),
    success_b=treatment_cited, n_b=len(treatment_df),
)

# 2. 지연 비교 (연속 메트릭, 등분산 미보장)
t, p = welch_t_test(control_df["latency_ms"], treatment_df["latency_ms"])

# 3. 트래픽 분배 검증 (의도 50:50인데 관측 60:40이면 sticky 누수 의심)
chi2, p = srm_chi_square(observed=[len(control_df), len(treatment_df)], expected_ratio=[0.5, 0.5])

# 4. 효과 크기
lift = relative_lift(treatment_mean=t_mean, control_mean=c_mean)  # (t-c)/c

6.2 왜 SRM 검정이 필요한가

Sample Ratio Mismatch는 가장 흔히 놓치는 실험 무결성 문제다. 의도가 50:50인데 실제 관측이 52:48이면 그냥 우연이지만, 60:40이면 sticky_hash 구현 버그·캐시 누수·일부 트래픽 우회를 의심해야 한다. SRM이 깨진 실험은 효과 크기 추정 자체를 신뢰할 수 없으므로, 다른 검정을 보기 전에 먼저 실행한다. core/stats.py의 srm_chi_square는 df=1은 erfc 정밀 계산, df>2는 Wilson-Hilferty 근사로 처리한다.

6.3 답변 품질 자동 신호화

A/B 실험에서 사람의 피드백(helpful/unhelpful 응답)만으로는 표본이 부족하다. MINERVA는 core/answer_features.py에서 정규식 기반 자동 신호를 추출해 인용률·표 포함률·원리근거 포함률·포맷 준수율을 매 응답마다 계산한다. 이 신호들이 runs.jsonl에 함께 기록되므로, A/B 분석 시 feedback.jsonl join 없이도 비율 메트릭으로 즉시 비교할 수 있다.

# answer_features.py 정규식 5패턴 (간략)
_PRINCIPLE_KEYWORD_RE = r"§|규칙|원칙|근거|결정\s*트리|구조\s*분해|..."
_PRINCIPLE_CITATION_RE = r"\[\s*\d+\s*,\s*§\s*[\d.]+\s*\]"

# 매 응답마다 계산되어 runs.jsonl에 기록
features = {
    "has_citation": bool(_PRINCIPLE_CITATION_RE.search(answer)),
    "has_principle": bool(_PRINCIPLE_KEYWORD_RE.search(answer)),
    "has_table": "|" in answer,  # 마크다운 표
    # ...
}

6.4 비용 메트릭

core/pricing.py는 모델별 토큰 단가표(gpt-4.1: (2.5, 10.0) 등 — input/output USD per 1M tokens)를 보유하며, runs.jsonl의 input_tokens/output_tokens/model_name으로 호출별 비용을 사후 계산할 수 있다. A/B 실험에서 품질이 같다면 저렴한 arm을 채택해야 하므로, 비용은 검정 결과와 함께 보고되어야 한다.

7 실험 관리 API

A/B 라우터는 운영 대시보드(Tests.tsx)와 같은 화면에서 사용량 지표(Monitoring.tsx)와 함께 노출되도록 /monitoring/ab prefix 아래 둔다. 두 화면이 같은 prefix의 데이터를 소비하면 메트릭 변화와 실험 결과를 한 번에 대조할 수 있다.

# routers/ab.py
router = APIRouter(prefix="/monitoring/ab", tags=["A/B Experiments"])

@router.get("/experiments")
def list_experiments():
    """등록된 실험 목록 (active/paused/stopped 메타 포함)"""
    return load_all_experiments()

@router.get("/{experiment_name}")
def get_experiment(experiment_name: str, days: int = 7):
    """arm별 메트릭 + 통계 검정 결과"""
    metrics_df = load_metrics(experiment_name, days=days)
    feedback_df = load_feedback(experiment_name, days=days)

    arms = {}
    for arm_id, arm_df in metrics_df.groupby("arm"):
        arms[arm_id] = {
            "n": len(arm_df),
            "avg_latency_ms": arm_df["latency_ms"].mean(),
            "p95_latency_ms": arm_df["latency_ms"].quantile(0.95),
            "citation_rate": (arm_df["citation_count"] > 0).mean(),
            "input_tokens_avg": arm_df["input_tokens"].mean(),
            "output_tokens_avg": arm_df["output_tokens"].mean(),
            "estimated_cost_usd": estimate_cost(arm_df),
        }

    # SRM 검증: 의도 비율 vs 관측 비율
    expected_ratio = [a["weight"] for a in load_arms(experiment_name)]
    observed = [arms[k]["n"] for k in arms]
    srm_chi2, srm_p = srm_chi_square(observed, expected_ratio)

    # 핵심 메트릭 검정 (control vs treatment)
    if "control" in arms and "treatment" in arms:
        ctrl, trt = metrics_df[metrics_df["arm"] == "control"], metrics_df[metrics_df["arm"] == "treatment"]
        t_stat, t_p = welch_t_test(ctrl["latency_ms"], trt["latency_ms"])
        z_stat, z_p = two_proportion_z_test(
            (ctrl["citation_count"] > 0).sum(), len(ctrl),
            (trt["citation_count"] > 0).sum(), len(trt),
        )
    else:
        t_stat = t_p = z_stat = z_p = None

    return {
        "experiment": experiment_name,
        "days": days,
        "arms": arms,
        "tests": {
            "srm": {"chi2": srm_chi2, "p_value": srm_p, "ok": srm_p > 0.001},
            "latency_welch_t": {"t": t_stat, "p_value": t_p},
            "citation_rate_z": {"z": z_stat, "p_value": z_p},
        },
    }

이 엔드포인트의 핵심 설계: SRM 검정을 모든 실험 응답에 함께 포함시켜, 대시보드 사용자가 효과 크기 해석 전에 무결성 점검을 먼저 보게 한다. srm_p가 0.001 미만이면 실험 자체가 의심스럽다는 신호이므로 ok=false로 표시한다.

8 관련 주제

선행 지식

• Pydantic 심화 – RAGConfig 스키마
• BaseAgent 계약 패턴 – 에이전트 캐싱 구조

후속 주제

• 프로덕션 배포 – 실험 설정 배포 관리

다른 카테고리 연결

• FastAPI 서빙 레이어 – 실험 라우팅과 에이전트 캐시
• MINERVA 아키텍처 개요 – 전체 구조에서 실험의 위치