MINERVA Phase C-4 — 지능형 라우팅 (Thompson Sampling·Contextual Bandit)

A/B는 기간 정해 비교, Bandit은 보상 좋은 arm 쪽으로 트래픽을 흘린다 — explore 비용을 줄이는 두 번째 학습기

06편·22편의 A/B는 기간을 정해 비교하고 사후에 결정한다. 그러나 새 reranker·새 모델·여러 프롬프트 변형이 계속 들어오면 A/B를 여러 번 돌리는 비용(explore loss)이 커진다. Multi-Armed Bandit은 매 순간 보상이 좋은 arm 쪽으로 트래픽을 옮기며 연속 학습한다. 본 편은 ε-greedy·UCB1·Thompson Sampling을 정리하고, 사용자 context로 분기하는 Contextual Bandit(LinUCB·Logistic Thompson Sampling)으로 확장한다. MINERVA 스킬 라우팅·모델 선택·프롬프트 변형 관리에 어떻게 적용되는지 다룬다.

Agent
저자

Kwangmin Kim

공개

2026년 05월 06일

1 왜 Bandit이 필요한가 — A/B의 한계

22편 A/B 심화는 기간을 정해 두 그룹을 비교하고 끝나면 의사결정한다. MINERVA가 처음 1~2개 변형을 검증할 때는 A/B로 충분하다. 그러나 운영이 성숙해지면 다음 상황이 생긴다.

운영 상황 A/B의 부담 Bandit이 해결
reranker가 매주 후보 1~2개씩 들어옴 매번 4주짜리 A/B → 분기당 4~8회 24/7 연속 학습, 보상 좋은 arm 자동 가중
프롬프트 변형 5~10개 동시 비교 5×10 = 50개 페어비교 다중 비교 모든 arm을 한 번에 다루고 점진 수렴
사용자 부서·역할별 최적 모델이 다름 부서×모델 매트릭스 A/B 폭증 Contextual Bandit이 자동 분기
신규 LLM 출시 시 빨리 도입 4주 A/B 필수 → 도입 지연 작은 트래픽으로 explore, 좋으면 자동 확대

본 편은 22편의 통계 의사결정과 상호 보완한다 — Bandit은 자동 트래픽 분배, A/B는 신중한 사전·사후 검증.

2 Multi-Armed Bandit — 정의

정의: Multi-Armed Bandit (MAB)

K개의 arm(슬롯머신 팔, 모델·프롬프트 변형 등)이 각각 미지의 보상 분포를 가진다. 매 시점 한 arm을 선택해 보상을 받으며, 누적 후회(regret)를 최소화한다.

\[ \text{Regret}(T) = T \cdot \mu^* - \sum_{t=1}^{T} \mu_{a_t} \]

  • \(\mu^*\): 최적 arm의 평균 보상
  • \(\mu_{a_t}\): 시점 \(t\) 에 선택한 arm의 평균 보상
  • 좋은 알고리즘: \(\text{Regret}(T) = O(\log T)\)

A/B는 모든 arm에 동일 트래픽 (보상 정보를 활용 안 함). Bandit은 점차 좋은 arm에 집중 (explore vs exploit trade-off).

Trade-off:
- Explore — 잘 모르는 arm에 트래픽을 줘서 정보 수집
- Exploit — 지금까지 가장 좋은 arm에 트래픽 집중

너무 explore → regret 누적
너무 exploit → 더 좋은 arm 발견 못 함

3 ε-greedy — 가장 단순한 Bandit

import numpy as np
from collections import defaultdict


class EpsilonGreedy:
    def __init__(self, n_arms: int, epsilon: float = 0.1):
        self.n_arms = n_arms
        self.epsilon = epsilon
        self.counts = np.zeros(n_arms)
        self.values = np.zeros(n_arms)        # 각 arm의 평균 보상

    def select(self) -> int:
        if np.random.rand() < self.epsilon:
            return np.random.randint(self.n_arms)        # explore
        return int(np.argmax(self.values))                # exploit

    def update(self, arm: int, reward: float):
        self.counts[arm] += 1
        n = self.counts[arm]
        self.values[arm] += (reward - self.values[arm]) / n   # incremental mean


# 사용 예 — 5개 reranker 후보
bandit = EpsilonGreedy(n_arms=5, epsilon=0.1)

for query in stream_of_queries():
    arm = bandit.select()
    response = run_with_reranker(query, reranker_id=arm)
    reward = float(response.thumbs_up)        # 0/1 보상
    bandit.update(arm, reward)

강점: 코드 5줄, 운영 단순. 약점: - ε이 일정 → arm 수가 늘면 explore가 비효율적 - 분산 큰 arm을 일찍 버릴 수 있음 - non-stationary(시간 흐름에 따라 보상 분포 변함)에 약함

epsilon을 시간에 따라 줄이는 ε-decay 변형이 흔하다 (\(\epsilon_t = 1/\sqrt{t}\) 등).

4 UCB1 — 신뢰구간 상한 기반

각 arm의 신뢰구간 상한이 가장 큰 것을 선택. “낙관적인 의사결정” 원칙.

\[ a_t = \arg\max_a \left[ \hat{\mu}_a + \sqrt{\frac{2 \ln t}{n_a}} \right] \]

class UCB1:
    def __init__(self, n_arms: int):
        self.n_arms = n_arms
        self.counts = np.zeros(n_arms)
        self.values = np.zeros(n_arms)
        self.t = 0

    def select(self) -> int:
        self.t += 1
        # 모든 arm을 최소 1번 시도
        for a in range(self.n_arms):
            if self.counts[a] == 0:
                return a

        ucb = self.values + np.sqrt(2 * np.log(self.t) / self.counts)
        return int(np.argmax(ucb))

    def update(self, arm: int, reward: float):
        self.counts[arm] += 1
        n = self.counts[arm]
        self.values[arm] += (reward - self.values[arm]) / n

강점: - 이론적 보장 — \(O(\log T)\) regret - ε 같은 hyperparameter 없음 - 표본이 적은 arm을 자연 우대 (explore 보너스)

약점: - 보상이 베르누이가 아닌 경우 boundary 조정 필요 (UCB-V·KL-UCB) - 보상 분포 변화에 느리게 반응

5 Thompson Sampling — 베이지안

각 arm의 보상 분포를 사후 분포로 업데이트. 매 라운드 사후에서 sample 후 가장 큰 sample을 가진 arm 선택.

베르누이 보상(thumbs_up)의 경우 Beta-Bernoulli 켤레 사전:

\[ p_a \sim \text{Beta}(\alpha_a, \beta_a) \]

class ThompsonSampling:
    def __init__(self, n_arms: int, alpha_prior: float = 1.0, beta_prior: float = 1.0):
        self.n_arms = n_arms
        self.alpha = np.full(n_arms, alpha_prior)         # successes + prior
        self.beta = np.full(n_arms, beta_prior)           # failures + prior

    def select(self) -> int:
        samples = np.random.beta(self.alpha, self.beta)   # 각 arm에서 1 sample
        return int(np.argmax(samples))

    def update(self, arm: int, reward: float):
        if reward >= 0.5:                                  # 성공
            self.alpha[arm] += 1
        else:
            self.beta[arm] += 1


# 사용
ts = ThompsonSampling(n_arms=5)
for query in stream_of_queries():
    arm = ts.select()
    reward = float(run_with_reranker(query, arm).thumbs_up)
    ts.update(arm, reward)

강점: - 실증적으로 가장 좋은 성능 (Chapelle & Li 2011, Kaufmann 2012) - 분산 큰 arm을 자연스럽게 explore - 사전 분포로 사전 지식 주입 가능

약점: - 베르누이 외 분포는 sampling 구현 복잡 (Gaussian: 정규-역감마, Poisson: 감마-푸아송) - 여러 메트릭 조합은 별도 설계 필요

5.1 연속 보상의 Thompson Sampling

응답 시간·CTR·평균 점수 같은 연속 보상은 Gaussian 가정:

class GaussianThompsonSampling:
    """알려진 분산을 가진 Gaussian 보상."""

    def __init__(self, n_arms: int, sigma: float = 1.0,
                  mu_prior: float = 0.0, sigma_prior: float = 1.0):
        self.n_arms = n_arms
        self.sigma = sigma
        self.mu_prior = np.full(n_arms, mu_prior)
        self.sigma_prior = np.full(n_arms, sigma_prior)
        self.counts = np.zeros(n_arms)
        self.sums = np.zeros(n_arms)

    def select(self) -> int:
        # 사후: N(mu_post, sigma_post^2)
        n = self.counts
        mu_post = (self.mu_prior / self.sigma_prior**2 + self.sums / self.sigma**2) / \
                  (1 / self.sigma_prior**2 + n / self.sigma**2)
        sigma_post = np.sqrt(1 / (1 / self.sigma_prior**2 + n / self.sigma**2))
        samples = np.random.normal(mu_post, sigma_post)
        return int(np.argmax(samples))

    def update(self, arm: int, reward: float):
        self.counts[arm] += 1
        self.sums[arm] += reward

6 Contextual Bandit — context로 분기

지금까지의 Bandit은 모든 사용자에게 같은 arm을 추천. 사용자가 다르면 최적 arm이 다르다:

사용자 부서 reranker A reranker B reranker C
Sales (간단·빠른 응답 중요) 0.65 0.55 0.50
R&D (정확·근거 중요) 0.50 0.70 0.60
Support (긴 컨텍스트) 0.55 0.55 0.75

전역 Bandit은 평균 0.58 (B 또는 C)으로 수렴, 부서별 최적은 놓친다. Contextual Bandit은 사용자 context \(x\) (부서·역할·과거 행동 임베딩 등)를 입력 받아 arm 선택.

def select_arm(context: np.ndarray) -> int:
    """context 벡터 x를 받아 최적 arm 인덱스 반환."""
    ...

6.1 LinUCB — 선형 모델 + UCB

각 arm의 보상이 context에 대해 선형이라고 가정:

\[ r_{a, t} = \theta_a^\top x_t + \epsilon \]

class LinUCB:
    def __init__(self, n_arms: int, n_features: int, alpha: float = 1.0):
        self.n_arms = n_arms
        self.A = [np.eye(n_features) for _ in range(n_arms)]   # d x d
        self.b = [np.zeros(n_features) for _ in range(n_arms)] # d
        self.alpha = alpha

    def select(self, context: np.ndarray) -> int:
        ucbs = []
        for a in range(self.n_arms):
            A_inv = np.linalg.inv(self.A[a])
            theta = A_inv @ self.b[a]
            mu = theta @ context
            bonus = self.alpha * np.sqrt(context @ A_inv @ context)
            ucbs.append(mu + bonus)
        return int(np.argmax(ucbs))

    def update(self, arm: int, context: np.ndarray, reward: float):
        self.A[arm] += np.outer(context, context)
        self.b[arm] += reward * context

강점: 명확한 이론적 보장, 작은 context dim에서 효율적. 약점: 보상이 비선형이면 underfit. context dim 클 경우 행렬 연산 비용.

6.2 Logistic Thompson Sampling

보상이 0/1(클릭·thumbs_up)이면 logistic 모델이 자연:

\[ P(r = 1 | x, a) = \sigma(\theta_a^\top x) \]

Laplace approximation으로 사후 추정 후 sampling. 라이브러리 (vowpalwabbit, contextualbandits) 활용.

7 MINERVA 적용 — 어디에 라우팅이 필요한가

결정 지점 arm 후보 context 알고리즘
reranker 선택 BGE, Cohere, Voyage, GPT-4 rerank 질의 길이·언어 LinUCB
LLM 모델 선택 gpt-4o, claude-3.5, gemini-2 질의 복잡도·과거 비용 Contextual TS
프롬프트 변형 v1 conservative, v2 verbose, v3 step-by-step 사용자 부서·역할 TS (context 적으면)
RAG mode dense, sparse, hybrid 도메인·문서 유형 LinUCB
HITL 호출 여부 자동 답변 vs 사람 검토 신뢰도·도메인 민감도 logistic TS

MINERVA 06편 인프라 위에 Bandit 라우터를 추가:

# app/routing/bandit_router.py
from app.experiments.assignment import sticky_hash_assign

class BanditRouter:
    def __init__(self, exp_name: str, algorithm: str = "thompson"):
        self.exp_name = exp_name
        self.algorithm = self._build(algorithm)
        self._lock = threading.Lock()

    def assign(self, user_id: str, context: dict) -> str:
        # 일정 비율(예 5%)은 sticky A/B로 holdout — 검증용
        if sticky_hash_assign(user_id, exp=self.exp_name + "_holdout", ratio=0.05):
            return "control"
        with self._lock:
            arm = self.algorithm.select(self._features(context))
        return self._arm_id(arm)

    def update(self, arm: str, context: dict, reward: float):
        with self._lock:
            self.algorithm.update(self._arm_index(arm), self._features(context), reward)

핵심 운영 포인트: - 5% holdout 그룹 — A/B test로 baseline 추적해 Bandit이 정상 작동하는지 검증 - Lock — 동시 사용자 update의 race condition 방지 (낮은 부하면 lock 없이 가능) - Persistence — counts·values·posterior parameters를 DB에 주기적 저장 (재시작 시 복원)

8 자주 발생하는 운영 함정

8.1 Cold Start

신규 arm은 표본 0 — Thompson은 자동 explore하지만 초기 N round는 손실 누적. 해법:

# 강제 round-robin 초기 단계
if min(self.counts) < 50:
    return int(np.argmin(self.counts))    # 가장 적게 본 arm
return self._sample_select()

또는 사전 정보로 Beta(α=2, β=1) 같은 informed prior.

8.2 Delayed Feedback

사용자 thumbs_up이 즉시 안 옴. update를 일괄 처리하면 explore가 부족해짐. 해법: - in-flight 카운터 — 아직 보상 안 온 arm도 “사용 중”으로 카운트 - batched Thompson Sampling — 일정 주기 (시간·요청 수)로 사후 갱신

# 단순한 batched 패턴
class BatchedTS(ThompsonSampling):
    def __init__(self, n_arms, batch_size=100):
        super().__init__(n_arms)
        self.batch_size = batch_size
        self.pending = []

    def update(self, arm, reward):
        self.pending.append((arm, reward))
        if len(self.pending) >= self.batch_size:
            for a, r in self.pending:
                super().update(a, r)
            self.pending.clear()

8.3 Non-stationary

시간이 흐르면 최적 arm이 바뀐다 (모델 drift, 사용자 분포 변화). 해법: - Discounted UCB / Discounted TS — 과거 보상을 시간으로 가중 - Sliding Window — 최근 W개 round만 사용 - CUSUM 변화점 검출 — 분포 변화 감지 시 reset

# 단순 sliding window
class SlidingWindowTS(ThompsonSampling):
    def __init__(self, n_arms, window=10_000):
        super().__init__(n_arms)
        self.window = window
        self.history = deque(maxlen=window)

    def update(self, arm, reward):
        self.history.append((arm, reward))
        # 매 N round 재계산 (full rebuild는 비싸니 incremental 권장)
        ...

8.4 Off-policy 평가

라이브 트래픽 없이 새 알고리즘을 평가하고 싶다 — 로깅된 데이터로 어떻게? Inverse Propensity Score (IPS) estimator:

\[ \hat{V}(\pi) = \frac{1}{n} \sum_{i} \frac{\mathbb{1}[a_i = \pi(x_i)]}{p(a_i | x_i)} r_i \]

로깅 시점에 각 arm의 propensity (선택 확률)를 함께 기록해야 한다 — 22편 JSONL에 propensity 필드 추가 권장.

9 A/B와 Bandit 결합 운영 패턴

단계 A/B (06·22편) Bandit (본 편)
신규 arm 검증 작은 holdout으로 A/A 통과 확인 (해당 arm을 explore에 포함)
효과 크기 측정 사전 등록 후 4주 A/B (실시간 평균이지만 신뢰구간 넓음)
운영 트래픽 분배 50/50 고정 자동 가중 (보상 좋은 arm 쪽)
결과 의사결정 통계적 유의 + practical significance 자연 수렴 — 최적 arm이 트래픽 점유

권장 패턴: 1. 새 arm 후보가 들어오면 A/A test + sample size 계산 (22편) 2. Bandit에 등록, 5% holdout으로 baseline A/B 동시 운영 3. Bandit이 안정적으로 한 arm에 80%+ 트래픽 → 운영 진입 4. 분기마다 holdout을 사용해 post-hoc A/B 보고서 — 통계적 검증

이 결합이 빠른 학습(Bandit) + 검증(A/B) + 거버넌스(holdout)의 균형을 잡는다.

10 정리

영역 핵심
동기 A/B는 사전 비교, Bandit은 연속 학습 + explore 비용 절감
ε-greedy 가장 단순, 코드 5줄
UCB1 이론적 보장 \(O(\log T)\), hyperparameter 없음
Thompson Sampling 실증적 최강, 사후 sampling 기반
Contextual Bandit 사용자 context로 분기 — LinUCB·Logistic TS
함정 Cold start·Delayed feedback·Non-stationary·Off-policy 평가
결합 5% holdout으로 A/B 동시 운영, 분기 사후 보고서

11 응용 분야

시나리오 추천 알고리즘 비고
reranker 5개 후보 (베르누이 보상) Thompson Sampling (Beta) 가장 안정적
LLM 모델 선택 (비용·품질 trade-off) Contextual TS + cost penalty 보상 = 품질 - λ·cost
프롬프트 변형 50개 (대량) Thompson + sliding window sample size 분산 방지
부서별 reranker 분기 LinUCB context = 부서 one-hot + 질의 길이
신규 arm cold start informed prior + round-robin 초기 초기 N round 강제
사후 평가 IPS estimator + propensity 로깅 22편 JSONL 확장

12 관련 주제

선행 학습 (선수)

18-LangGraph 시리즈 cross-reference

  • #10 프롬프트 분류와 라우팅 — context 기반 라우팅 이론적 배경
  • #18 스킬 라우팅 확장성과 한계 — arm 수가 많을 때 검색 vs Bandit 비교

후속 (Phase C-4)

Subscribe

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