1 RL for Longitudinal Data — Overview

1.1 이 시리즈의 파일들

파일	주제	핵심
29 — Contextual Bandit	LinUCB, Thompson Sampling	실시간 개인화, 탐색-활용 균형
30 — DTR	Dynamic Treatment Regime	순차적 최적 처치, Q-learning
31 — Offline RL	Conservative Q-Learning	기존 로그로 안전하게 정책 학습

1.2 A/B 테스트의 한계: 왜 RL이 필요한가

A/B 테스트는 정적(static)이다:

A/B 테스트:
  실험 기간: 전체 사용자를 두 그룹으로 나눠 2~4주 관찰
  결정: 실험 종료 후 승자 전략 채택
  한계:
    - 실험 기간 동안 열등한 전략(Arm B)을 받는 사용자 손해 (Regret)
    - 사용자 상태 변화 반영 불가 (같은 전략이 모든 상황에 동일하게 적용)
    - 최적 전략이 사용자 특성마다 다를 수 있음

RL의 관점:

RL 기반 개인화:
  매 상호작용마다 사용자 상태를 관찰
  현재 상태에서 최적 행동(전략) 선택
  결과(보상)로 정책 업데이트
  → 탐색(Exploration)과 활용(Exploitation)의 균형

장점:
  - 실시간 적응
  - 개인별 최적화
  - 누적 보상 최대화 (장기 리텐션)

1.3 기법 1: Contextual Bandit

1.3.1 개념

A/B 테스트와 RL의 중간 단계. 문맥(Context)을 보고 행동을 선택하되, 순차적 의존성은 없다.

A/B 테스트: 모든 사용자에게 동일한 전략
단순 Bandit: 전체 평균 보상이 가장 높은 전략
Contextual Bandit: 사용자 상태(문맥)를 보고 그 사람에게 최적 전략 선택

구조:

at each interaction:
  1. 사용자 상태 관찰: x_t (세그먼트, 과거 만족도, 현재 감정 등)
  2. 행동 선택: a_t = π(x_t)  (어떤 프롬프트 전략?)
  3. 보상 관찰: r_t = R(x_t, a_t)  (만족도, 전환 여부)
  4. 정책 업데이트: π ← π + ∇r_t

다음 상태로의 전환은 고려하지 않음 (Bandit의 한계)

1.3.2 예시: AI Agent 프롬프트 전략 최적화

import numpy as np
from sklearn.linear_model import Ridge

class LinUCB:
    """
    Linear Upper Confidence Bound (LinUCB)
    — 선형 보상 모델 + 불확실성 탐색
    """
    def __init__(self, n_arms, n_features, alpha=1.0):
        self.n_arms = n_arms      # 행동 수 (프롬프트 전략 수)
        self.n_features = n_features
        self.alpha = alpha        # 탐색 강도

        # 각 arm별 파라미터
        self.A = [np.eye(n_features) for _ in range(n_arms)]     # (n_feat, n_feat)
        self.b = [np.zeros(n_features) for _ in range(n_arms)]    # (n_feat,)

    def select_arm(self, context):
        """문맥 x를 보고 최적 행동 선택"""
        ucb_values = []
        for a in range(self.n_arms):
            A_inv = np.linalg.inv(self.A[a])
            theta = A_inv @ self.b[a]              # 추정 보상 계수

            # UCB = 예측 보상 + 탐색 보너스
            predicted = theta @ context
            uncertainty = self.alpha * np.sqrt(context @ A_inv @ context)
            ucb_values.append(predicted + uncertainty)

        return np.argmax(ucb_values)

    def update(self, arm, context, reward):
        """관찰된 보상으로 파라미터 업데이트"""
        self.A[arm] += np.outer(context, context)
        self.b[arm] += reward * context


# 프롬프트 전략 3가지
ARMS = ["기본", "공감형", "단계별 가이드"]

# 사용자 문맥 피처
def get_context(user):
    return np.array([
        user["segment_MIEP"],
        user["segment_N"],
        user["avg_satisfaction"],
        user["emotion_score"],
        user["session_count"],
        user["personalized_ratio"],
        1.0  # 편향 항
    ])

# 시뮬레이션
bandit = LinUCB(n_arms=3, n_features=7, alpha=1.5)

history = []
for t, user in enumerate(user_stream):
    context = get_context(user)
    arm = bandit.select_arm(context)
    reward = observe_satisfaction(user, ARMS[arm])  # 실제 서비스에서 측정
    bandit.update(arm, context, reward)

    history.append({
        "t": t, "user_id": user["user_id"],
        "arm": ARMS[arm], "reward": reward
    })

# 누적 후회(Regret) 계산
best_reward = max([np.mean([h["reward"] for h in history if h["arm"]==a])
                   for a in ARMS])
cumulative_regret = np.cumsum([best_reward - h["reward"] for h in history])

1.3.3 A/B 테스트 vs Contextual Bandit 비교

1000번의 상호작용 시뮬레이션:

A/B 테스트 (500번씩 탐색 후 승자 적용):
  탐색 기간 손실: ~150 (열등한 전략 노출)
  최적 전략 채택 후: 안정
  총 후회: 320

LinUCB:
  초기에 탐색, 점차 최적 전략으로 수렴
  총 후회: 185   ← 41% 감소

Thompson Sampling:
  베이지안 탐색, 더 부드러운 수렴
  총 후회: 165   ← 48% 감소

1.3.4 Thompson Sampling (베이지안 대안)

class ThompsonSamplingBandit:
    """
    Thompson Sampling: 각 arm의 보상 분포를 베이지안으로 추정
    보상이 이진(전환 여부)일 때: Beta-Binomial 사용
    """
    def __init__(self, n_arms):
        self.alpha = np.ones(n_arms)   # 성공 횟수 + 1 (Beta prior)
        self.beta  = np.ones(n_arms)   # 실패 횟수 + 1

    def select_arm(self, context=None):
        # 각 arm의 Beta 분포에서 샘플링 → 가장 높은 arm 선택
        samples = np.random.beta(self.alpha, self.beta)
        return np.argmax(samples)

    def update(self, arm, reward):
        if reward == 1:
            self.alpha[arm] += 1
        else:
            self.beta[arm] += 1

1.4 기법 2: Dynamic Treatment Regime (DTR)

1.4.1 개념

Contextual Bandit과 달리 순차적 의사결정을 고려한다. 현재 행동이 미래 상태에 영향을 미친다.

Bandit:   x_t → a_t → r_t  (각 시점 독립)

DTR:      x_1 → a_1 → r_1 → x_2 → a_2 → r_2 → ... → x_T → a_T → r_T
          (이전 행동이 다음 상태에 영향)

목표: 전체 기간의 누적 보상 최대화
     E[r_1 + r_2 + ... + r_T]

AI Agent 맥락:

Week 1: 사용자 상태 관찰 → 기본 프롬프트 적용 → 만족도 관찰
Week 2: (Week 1 결과를 반영한) 상태 관찰 → 공감형 프롬프트 → 만족도 관찰
...
Week 8: 상태 관찰 → 단계별 가이드 → 전환 여부 관찰

→ 어떤 주에 어떤 전략을 써야 장기 전환율이 최대화되는가?

1.4.2 Q-learning 기반 DTR

import numpy as np

class QLearningDTR:
    """
    Q-learning으로 동적 처치 결정 학습
    Q(s, a) = 상태 s에서 행동 a를 취할 때의 기대 누적 보상
    """
    def __init__(self, n_actions, n_state_features,
                 learning_rate=0.01, discount=0.9, epsilon=0.1):
        self.n_actions = n_actions
        self.gamma = discount    # 미래 보상 할인율
        self.eps = epsilon       # 탐색 확률

        # Q 함수를 선형 근사 (심층 Q-learning으로 확장 가능)
        self.W = np.zeros((n_actions, n_state_features))

    def q_values(self, state):
        return self.W @ state   # (n_actions,)

    def select_action(self, state):
        if np.random.random() < self.eps:
            return np.random.randint(self.n_actions)   # 탐색
        return np.argmax(self.q_values(state))          # 활용

    def update(self, state, action, reward, next_state, done):
        """벨만 방정식으로 Q 업데이트"""
        q_current = self.q_values(state)[action]
        if done:
            q_target = reward
        else:
            q_target = reward + self.gamma * np.max(self.q_values(next_state))

        # 그래디언트 업데이트
        td_error = q_target - q_current
        self.W[action] += 0.01 * td_error * state


# 환경 시뮬레이션
class AgentPersonalizationEnv:
    """AI Agent 개인화 시뮬레이션 환경"""
    def __init__(self):
        self.state_dim = 6      # [만족도, 턴수, 감정, 세그먼트, 주차, 개인화비율]
        self.n_actions = 3      # [기본, 공감형, 단계별]

    def reset(self):
        """새 에피소드 (새 사용자) 시작"""
        self.week = 1
        self.user_profile = self._sample_user()
        return self._get_state()

    def step(self, action):
        """한 주 진행"""
        reward = self._get_reward(action)
        self.week += 1
        done = self.week > 8
        next_state = self._get_state() if not done else None
        return next_state, reward, done

    def _get_reward(self, action):
        """프롬프트 전략 × 사용자 상태 → 만족도 보상"""
        base = self.user_profile["base_satisfaction"]
        if action == 0:   # 기본
            return base + np.random.normal(0, 0.3)
        elif action == 1: # 공감형: N계열에 효과적
            bonus = 0.5 if self.user_profile["segment"] == "N" else 0.1
            return base + bonus + np.random.normal(0, 0.3)
        else:             # 단계별: MIEP에 효과적
            bonus = 0.4 if self.user_profile["segment"] == "MIEP" else 0.05
            return base + bonus + np.random.normal(0, 0.3)


# 학습
env = AgentPersonalizationEnv()
agent = QLearningDTR(n_actions=3, n_state_features=6)

for episode in range(10000):
    state = env.reset()
    total_reward = 0
    while True:
        action = agent.select_action(state)
        next_state, reward, done = env.step(action)
        agent.update(state, action, reward, next_state, done)
        total_reward += reward
        if done:
            break
        state = next_state

    if (episode + 1) % 1000 == 0:
        print(f"Episode {episode+1}: Avg Reward = {total_reward/8:.3f}")

1.4.3 최적 정책 분석

# 학습된 정책: 각 상태에서 어떤 전략을 선택하는가?
segment_types = ["SI", "MIEP", "N"]
strategy_names = ["기본", "공감형", "단계별"]

print("학습된 최적 전략:")
print("=" * 40)
for seg in segment_types:
    for week in [1, 4, 8]:
        state = build_state(segment=seg, week=week, satisfaction=3.5)
        action = agent.select_action(state)
        print(f"{seg} 세그먼트, Week {week}: {strategy_names[action]}")

학습된 최적 전략:
========================================
SI 세그먼트, Week 1: 기본         ← 탐색 초반: 기본 전략
SI 세그먼트, Week 4: 기본
SI 세그먼트, Week 8: 기본         ← SI는 일관되게 기본
MIEP 세그먼트, Week 1: 기본       ← 초반: 기본으로 파악
MIEP 세그먼트, Week 4: 단계별     ← 중반: 단계별로 심화
MIEP 세그먼트, Week 8: 단계별     ← 후반: 단계별 유지
N 세그먼트, Week 1: 기본
N 세그먼트, Week 4: 공감형        ← 불만 누적 시점에 공감형
N 세그먼트, Week 8: 공감형        ← 이탈 방지 집중

1.5 기법 3: Offline RL (역강화학습 + 배치 정책 최적화)

1.5.1 왜 Offline RL인가

온라인 RL의 문제: - 실제 서비스에서 “탐색” = 일부 사용자에게 나쁜 전략 노출 - 안전 제약 (의료, 금융 등) - 초기 데이터 없이 시작 불가

Offline RL: 기존 관찰 데이터(로그)만으로 최적 정책 학습

수집된 데이터: (상태, 행동, 보상, 다음 상태) 튜플
현재 정책(행동 로그)과 다른 정책을 학습하는 것이 핵심 도전
→ Distribution Shift 문제

1.5.2 Conservative Q-Learning (CQL)

오프라인 데이터에서 과대 추정을 방지하는 방법:

import torch
import torch.nn as nn

class CQLAgent:
    """
    Conservative Q-Learning for Offline RL
    데이터에 없는 (상태, 행동) 쌍의 Q값을 보수적으로 낮게 추정
    """
    def __init__(self, state_dim, n_actions, alpha=1.0):
        self.q_net = nn.Sequential(
            nn.Linear(state_dim, 128), nn.ReLU(),
            nn.Linear(128, 64), nn.ReLU(),
            nn.Linear(64, n_actions)
        )
        self.target_net = nn.Sequential(
            nn.Linear(state_dim, 128), nn.ReLU(),
            nn.Linear(128, 64), nn.ReLU(),
            nn.Linear(64, n_actions)
        )
        self.alpha = alpha   # 보수성 강도
        self.optimizer = torch.optim.Adam(self.q_net.parameters(), lr=1e-3)

    def train_step(self, batch):
        states, actions, rewards, next_states, dones = batch

        # 표준 Bellman 손실
        q_values = self.q_net(states).gather(1, actions.unsqueeze(1))
        with torch.no_grad():
            next_q = self.target_net(next_states).max(1)[0]
            target = rewards + 0.99 * next_q * (1 - dones)
        bellman_loss = nn.MSELoss()(q_values.squeeze(), target)

        # CQL 정규화: 데이터에 없는 행동의 Q를 낮게
        # 데이터에 있는 행동의 Q를 높게 유지
        q_all = self.q_net(states)
        cql_loss = (
            torch.logsumexp(q_all, dim=1).mean()    # 모든 행동 Q 높이면 페널티
            - q_all.gather(1, actions.unsqueeze(1)).mean()  # 관찰 행동 Q는 보상
        )

        total_loss = bellman_loss + self.alpha * cql_loss
        self.optimizer.zero_grad()
        total_loss.backward()
        self.optimizer.step()
        return total_loss.item()


# 기존 서비스 로그로 학습
offline_data = load_service_logs()  # 수집된 (s, a, r, s') 데이터
cql_agent = CQLAgent(state_dim=6, n_actions=3)

for epoch in range(1000):
    batch = sample_batch(offline_data, batch_size=256)
    loss = cql_agent.train_step(batch)

1.6 A/B 테스트 → Bandit → DTR → Offline RL: 진화 경로

단계 1: A/B 테스트 (현재 표준)
  특징: 정적, 그룹 단위, 실험 기간 필요
  장점: 해석 쉬움, 통계적 검정 가능
  단점: 적응 없음, 탐색 비용

단계 2: Contextual Bandit
  특징: 실시간, 개인 단위, 탐색-활용 균형
  장점: A/B보다 낮은 후회
  단점: 순차적 의존성 무시

단계 3: Dynamic Treatment Regime (DTR)
  특징: 순차적, 장기 보상 최적화
  장점: 미래 상태까지 고려
  단점: 환경 모델 필요, 복잡함

단계 4: Offline RL
  특징: 기존 로그 활용, 안전한 탐색
  장점: 실제 서비스 위험 없음
  단점: Distribution shift 문제

1.7 통계 모델과 RL의 결합

1.7.1 인과 추론 + RL

통계 모델(인과 추론)로 처치 효과를 추정하고, 그것을 RL의 보상 함수로 사용:

# Step 1: LMM으로 조건부 처치 효과 (CATE) 추정
from sklearn.linear_model import LinearRegression

def estimate_cate(df, treatment_col, outcome_col, feature_cols):
    """조건부 평균 처치 효과 추정 (T-learner)"""
    treated = df[df[treatment_col] == 1]
    control = df[df[treatment_col] == 0]

    model_t = LinearRegression().fit(
        treated[feature_cols], treated[outcome_col]
    )
    model_c = LinearRegression().fit(
        control[feature_cols], control[outcome_col]
    )

    # 각 개인의 처치 효과 추정
    df["cate"] = (
        model_t.predict(df[feature_cols])
        - model_c.predict(df[feature_cols])
    )
    return df

# Step 2: CATE를 Bandit의 보상 함수로 사용
# → 처치 효과가 높은 사용자에게 더 적극적으로 처치 적용

1.8 정리: 종단 데이터 × 의사결정 프레임워크

프레임워크	핵심 질문	데이터 구조	추천 상황
A/B 테스트 + LMM	처치 효과가 있는가?	반복 측정	인과 추론, 통계적 검정
Contextual Bandit	이 사람에게 지금 최적 전략은?	실시간 스트림	실시간 개인화, 낮은 위험
DTR / Q-learning	장기적으로 최적 전략 순서는?	종단 궤적	의료, 교육, 장기 최적화
Offline RL	기존 데이터로 최적 정책은?	히스토리컬 로그	탐색 비용/위험이 클 때

Agent Personalization과의 연결:

현재 (Agent 22-Personalization 시리즈):
  Segmentation → A/B 테스트로 전략 검증 (정적)
  Personalization → 프로필 기반 템플릿 (반정적)
  Hyperpersonalization → 실시간 감정 반응 (동적이지만 규칙 기반)

RL로 진화 시:
  Contextual Bandit → 실시간 프롬프트 전략 최적화
  DTR → 8주 개인화 여정 전체 계획 최적화
  Offline RL → 기존 서비스 로그로 안전하게 학습