Dynamic Treatment Regime (DTR)

순차적 최적 처치 결정 — Q-learning과 장기 보상 최적화

Contextual Bandit이 각 시점을 독립으로 보는 반면, DTR은 현재 행동이 미래 상태에 영향을 미치는 순차적 구조를 명시적으로 모델링한다. MDP 프레임워크, Q-learning의 수학적 원리, 환경 시뮬레이션 구현, DQN 확장, 그리고 SMART 임상시험 설계까지 다룬다.

Statistics
Reinforcement Learning
저자

Kwangmin Kim

공개

2026년 03월 08일

1 Dynamic Treatment Regime (DTR)

1.1 Bandit → DTR: 순차적 의존성

1.1.1 Contextual Bandit의 가정과 한계

Contextual Bandit은 각 시점이 독립이라고 가정한다:

\[ x_t \perp a_{t-1} \quad \text{(이전 행동이 현재 상태에 영향 없음)} \]

현실에서는 이 가정이 깨진다:

Week 1: 공감형 프롬프트 → 사용자 만족도 3.2 → 4.1 (상승)
Week 2: 상승된 만족도(4.1)가 새로운 상태 → 다음 행동 선택에 영향
Week 3: 단계별 가이드 적용 → 만족도 유지 4.0 → 개인화 비율 상승

→ 전략 순서가 중요: (공감 → 단계별)과 (단계별 → 공감)의 장기 결과가 다름

DTR은 이 순차적 의존성을 명시적으로 모델링한다.

1.2 DTR 수학적 프레임워크

1.2.1 Markov Decision Process (MDP)

DTR은 MDP로 공식화된다:

\[ \mathcal{M} = (\mathcal{S}, \mathcal{A}, P, R, \gamma, T) \]

요소 정의 AI Agent 예시
\(\mathcal{S}\) 상태 공간 만족도, 턴수, 감정, 세그먼트, 주차, 개인화비율
\(\mathcal{A}\) 행동 공간 {기본, 공감형, 단계별 가이드}
\(P(s_{t+1} \| s_t, a_t)\) 전이 확률 현재 전략이 다음 주 상태에 미치는 영향
\(R(s_t, a_t)\) 보상 함수 해당 주의 만족도
\(\gamma \in [0, 1)\) 할인율 미래 보상의 현재 가치 (보통 0.9~0.99)
\(T\) 에피소드 길이 8주

1.2.2 정책 (Policy)

정책은 상태에서 행동으로의 매핑:

\[ \pi: \mathcal{S} \to \mathcal{A}, \quad a_t = \pi(s_t) \]

최적 정책은 누적 보상을 최대화:

\[ \pi^* = \arg\max_\pi \mathbb{E}\left[\sum_{t=0}^{T} \gamma^t r_t \;\middle|\; s_0, \pi\right] \]

1.2.3 가치 함수 (Value Function)

정책 \(\pi\) 하에서 상태 \(s\)의 가치:

\[ V^\pi(s) = \mathbb{E}\left[\sum_{t=0}^{T} \gamma^t r_t \;\middle|\; s_0 = s, \pi\right] \]

행동-가치 함수 (Q-function):

\[ Q^\pi(s, a) = \mathbb{E}\left[r_0 + \gamma V^\pi(s_1) \;\middle|\; s_0 = s, a_0 = a\right] \]

최적 정책과 Q-function의 관계:

\[ \pi^*(s) = \arg\max_a Q^*(s, a) \]

1.3 Q-learning for DTR

1.3.1 Bellman Equation

최적 Q-function은 Bellman 최적 방정식을 만족:

\[ Q^*(s, a) = \mathbb{E}\left[r + \gamma \max_{a'} Q^*(s', a') \;\middle|\; s, a\right] \]

직관: 지금 보상 + 다음 상태에서의 최대 미래 가치

1.3.2 선형 Q-function 근사

상태 공간이 연속적일 때, Q를 선형 함수로 근사:

\[ Q(s, a; \mathbf{w}_a) = \mathbf{w}_a^T \phi(s) \]

  • \(\phi(s) \in \mathbb{R}^d\): 상태 특징 벡터
  • \(\mathbf{w}_a \in \mathbb{R}^d\): arm \(a\)의 가중치

1.3.3 Temporal Difference (TD) Error

매 시점 관찰 \((s_t, a_t, r_t, s_{t+1})\)에서:

\[ \delta_t = r_t + \gamma \max_{a'} Q(s_{t+1}, a'; \mathbf{w}) - Q(s_t, a_t; \mathbf{w}) \]

가중치 업데이트:

\[ \mathbf{w}_{a_t} \leftarrow \mathbf{w}_{a_t} + \alpha \cdot \delta_t \cdot \phi(s_t) \]

  • \(\alpha\): 학습률
  • \(\delta_t\): TD error — 양수면 Q를 올리고, 음수면 내린다

1.4 환경 시뮬레이션: AgentPersonalizationEnv

1.4.1 상태 공간 설계

import numpy as np

class AgentPersonalizationEnv:
    """
    AI Agent 개인화 시뮬레이션 환경

    상태: [만족도, 턴수(정규화), 감정점수, 세그먼트_MIEP, 세그먼트_N, 주차(정규화)]
    행동: 0=기본, 1=공감형, 2=단계별 가이드
    에피소드: 8주 (T=8)
    """
    def __init__(self, max_weeks=8):
        self.max_weeks = max_weeks
        self.state_dim = 6
        self.n_actions = 3
        self.action_names = ["기본", "공감형", "단계별"]

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

        # 사용자 프로필 샘플링
        segment = np.random.choice(["SI", "MIEP", "N"], p=[0.4, 0.35, 0.25])
        self.segment = segment
        self.seg_miep = 1.0 if segment == "MIEP" else 0.0
        self.seg_n = 1.0 if segment == "N" else 0.0

        # 초기 상태
        self.satisfaction = np.random.uniform(2.5, 4.0)
        self.emotion = np.random.uniform(-0.5, 0.5)
        self.turn_count = np.random.randint(3, 15)
        self.personalization_ratio = 0.0

        return self._get_state()

    def _get_state(self):
        return np.array([
            self.satisfaction / 5.0,         # 정규화
            self.turn_count / 20.0,          # 정규화
            self.emotion,                     # [-1, 1]
            self.seg_miep,                    # 이진
            self.seg_n,                       # 이진
            self.week / self.max_weeks        # 정규화
        ])

    def step(self, action):
        """한 주 진행: 행동 → 보상 + 상태 전이"""
        reward = self._compute_reward(action)

        # 상태 전이: 현재 행동이 다음 상태에 영향
        self._transition(action, reward)

        self.week += 1
        done = self.week > self.max_weeks

        next_state = self._get_state() if not done else np.zeros(self.state_dim)
        return next_state, reward, done

    def _compute_reward(self, action):
        """세그먼트 × 전략 × 현재 상태 → 보상"""
        base = self.satisfaction * 0.3 + self.emotion * 0.2

        if action == 0:  # 기본
            effect = 0.0
        elif action == 1:  # 공감형
            # N 세그먼트에 효과적, 감정이 낮을 때 더 효과적
            effect = 0.5 * self.seg_n + 0.15 * self.seg_miep
            effect += 0.2 * max(0, -self.emotion)  # 부정 감정일 때 보너스
        else:  # 단계별 가이드
            # MIEP에 효과적, 주차가 진행될수록 효과 증가
            effect = 0.4 * self.seg_miep + 0.05 * self.seg_n
            effect += 0.1 * (self.week / self.max_weeks)  # 후반부 보너스

        noise = np.random.normal(0, 0.2)
        return np.clip(base + effect + noise, 0, 5)

    def _transition(self, action, reward):
        """상태 전이: 행동과 보상이 다음 상태를 결정"""
        # 만족도 업데이트 (지수 이동 평균)
        self.satisfaction = 0.7 * self.satisfaction + 0.3 * reward

        # 감정 업데이트
        if reward > 3.0:
            self.emotion = min(1.0, self.emotion + 0.1)
        else:
            self.emotion = max(-1.0, self.emotion - 0.15)

        # 턴수 증가
        self.turn_count += np.random.randint(2, 8)

        # 개인화 비율 (공감형/단계별 사용 시 증가)
        if action > 0:
            self.personalization_ratio = min(
                1.0, self.personalization_ratio + 0.15
            )

1.4.2 핵심 설계 포인트

1. 순차적 의존성:
   - 만족도는 지수 이동 평균으로 업데이트 → 과거 행동의 누적 효과
   - 감정은 보상에 따라 변동 → 행동 선택의 결과가 다음 상태에 반영

2. 세그먼트별 차별적 효과:
   - SI: 기본 전략이 무난 (추가 효과 없음)
   - MIEP: 단계별 가이드에 보너스 +0.4
   - N: 공감형에 보너스 +0.5, 부정 감정일 때 추가 +0.2

3. 시간 의존 효과:
   - 단계별 가이드는 후반부에 효과 증가 (주차/max_weeks 보너스)

1.5 Q-learning 학습 코드

1.5.1 전체 학습 루프

class QLearningDTR:
    """
    Tabular-style Q-learning with linear function approximation
    """
    def __init__(self, state_dim, n_actions,
                 lr=0.01, gamma=0.9, epsilon_start=1.0, epsilon_end=0.05,
                 epsilon_decay=0.995):
        self.n_actions = n_actions
        self.gamma = gamma
        self.lr = lr
        self.epsilon = epsilon_start
        self.epsilon_end = epsilon_end
        self.epsilon_decay = epsilon_decay

        # 선형 Q-function: Q(s, a) = W[a] @ s
        self.W = np.random.randn(n_actions, state_dim) * 0.01

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

    def select_action(self, state):
        if np.random.random() < self.epsilon:
            return np.random.randint(self.n_actions)
        return np.argmax(self.q_values(state))

    def update(self, state, action, reward, next_state, done):
        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] += self.lr * td_error * state

    def decay_epsilon(self):
        self.epsilon = max(self.epsilon_end, self.epsilon * self.epsilon_decay)


# 학습 실행
env = AgentPersonalizationEnv(max_weeks=8)
agent = QLearningDTR(state_dim=6, n_actions=3)

n_episodes = 10000
episode_rewards = []

for ep in range(n_episodes):
    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

    agent.decay_epsilon()
    episode_rewards.append(total_reward)

    if (ep + 1) % 2000 == 0:
        avg = np.mean(episode_rewards[-500:])
        print(f"Episode {ep+1:5d} | Avg Reward (last 500): {avg:.3f} | ε: {agent.epsilon:.3f}")

1.5.2 수렴 확인

import matplotlib.pyplot as plt

# 이동 평균으로 수렴 확인
window = 200
moving_avg = np.convolve(episode_rewards, np.ones(window)/window, mode="valid")

plt.figure(figsize=(10, 5))
plt.plot(moving_avg, linewidth=1.5)
plt.xlabel("Episode")
plt.ylabel("Total Reward (8-week)")
plt.title("Q-learning Training Convergence")
plt.axhline(y=np.mean(moving_avg[-1000:]), color="r", linestyle="--",
            label=f"Final Avg: {np.mean(moving_avg[-1000:]):.2f}")
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()

1.6 학습된 최적 정책 분석

1.6.1 세그먼트 × 주차별 최적 전략

strategy_names = ["기본", "공감형", "단계별"]
segments = {
    "SI":   [0.0, 0.0],
    "MIEP": [1.0, 0.0],
    "N":    [0.0, 1.0]
}

print("학습된 최적 전략 (Q-learning)")
print("=" * 55)
print(f"{'세그먼트':>8} | {'Week':>4} | {'전략':>8} | {'Q-values':>24}")
print("-" * 55)

for seg_name, (seg_m, seg_n) in segments.items():
    for week in [1, 2, 4, 6, 8]:
        state = np.array([3.5/5, 5/20, 0.0, seg_m, seg_n, week/8])
        q_vals = agent.q_values(state)
        best_action = np.argmax(q_vals)

        q_str = ", ".join(f"{q:.2f}" for q in q_vals)
        print(f"{seg_name:>8} | {week:>4} | {strategy_names[best_action]:>8} | [{q_str}]")
    print("-" * 55)

1.6.2 기대 출력

학습된 최적 전략 (Q-learning)
=======================================================
 세그먼트 | Week |       전략 |               Q-values
-------------------------------------------------------
      SI |    1 |       기본 | [1.23, 0.98, 0.87]
      SI |    2 |       기본 | [1.31, 1.05, 0.92]
      SI |    4 |       기본 | [1.45, 1.12, 1.01]
      SI |    6 |       기본 | [1.52, 1.18, 1.08]
      SI |    8 |       기본 | [1.55, 1.20, 1.10]
-------------------------------------------------------
    MIEP |    1 |       기본 | [1.15, 1.08, 1.12]
    MIEP |    2 |     단계별 | [1.22, 1.15, 1.35]
    MIEP |    4 |     단계별 | [1.28, 1.20, 1.58]
    MIEP |    6 |     단계별 | [1.30, 1.22, 1.72]
    MIEP |    8 |     단계별 | [1.32, 1.24, 1.85]
-------------------------------------------------------
       N |    1 |       기본 | [1.10, 1.05, 0.88]
       N |    2 |     공감형 | [1.18, 1.42, 0.95]
       N |    4 |     공감형 | [1.25, 1.68, 1.02]
       N |    6 |     공감형 | [1.28, 1.80, 1.05]
       N |    8 |     공감형 | [1.30, 1.88, 1.08]
-------------------------------------------------------

해석:

  • SI: 전 기간 기본 전략이 최적 (추가 전략의 효과 미미)
  • MIEP: 1주차 기본으로 시작 → 2주차부터 단계별 가이드가 우세 (후반부로 갈수록 Q값 격차 확대)
  • N: 1주차 기본 → 2주차부터 공감형이 압도적 (부정 감정 보정 효과)

1.7 Deep Q-Network (DQN)

1.7.1 선형 근사 → 신경망 확장

상태 공간이 복잡하거나 비선형 관계가 있을 때, Q-function을 신경망으로 근사:

\[ Q(s, a; \theta) \approx Q^*(s, a) \]

import torch
import torch.nn as nn
from collections import deque
import random

class DQN(nn.Module):
    def __init__(self, state_dim, n_actions):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(state_dim, 128),
            nn.ReLU(),
            nn.Linear(128, 64),
            nn.ReLU(),
            nn.Linear(64, n_actions)
        )

    def forward(self, x):
        return self.net(x)


class DQNAgent:
    def __init__(self, state_dim, n_actions, lr=1e-3, gamma=0.9,
                 buffer_size=10000, batch_size=64, target_update=100):
        self.n_actions = n_actions
        self.gamma = gamma
        self.batch_size = batch_size
        self.target_update = target_update
        self.step_count = 0

        self.q_net = DQN(state_dim, n_actions)
        self.target_net = DQN(state_dim, n_actions)
        self.target_net.load_state_dict(self.q_net.state_dict())

        self.optimizer = torch.optim.Adam(self.q_net.parameters(), lr=lr)
        self.buffer = deque(maxlen=buffer_size)
        self.epsilon = 1.0

    def select_action(self, state):
        if random.random() < self.epsilon:
            return random.randint(0, self.n_actions - 1)

        with torch.no_grad():
            state_t = torch.FloatTensor(state).unsqueeze(0)
            q_values = self.q_net(state_t)
            return q_values.argmax(dim=1).item()

    def store(self, state, action, reward, next_state, done):
        self.buffer.append((state, action, reward, next_state, done))

    def train_step(self):
        if len(self.buffer) < self.batch_size:
            return 0.0

        batch = random.sample(self.buffer, self.batch_size)
        states, actions, rewards, next_states, dones = zip(*batch)

        states = torch.FloatTensor(np.array(states))
        actions = torch.LongTensor(actions)
        rewards = torch.FloatTensor(rewards)
        next_states = torch.FloatTensor(np.array(next_states))
        dones = torch.FloatTensor(dones)

        # 현재 Q
        q_values = self.q_net(states).gather(1, actions.unsqueeze(1)).squeeze()

        # Target Q
        with torch.no_grad():
            next_q = self.target_net(next_states).max(1)[0]
            target = rewards + self.gamma * next_q * (1 - dones)

        loss = nn.MSELoss()(q_values, target)

        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()

        # Target network 업데이트
        self.step_count += 1
        if self.step_count % self.target_update == 0:
            self.target_net.load_state_dict(self.q_net.state_dict())

        return loss.item()

1.7.2 DQN vs Linear Q-learning 비교

측면 Linear Q-learning DQN
모델 \(Q = \mathbf{w}^T \phi(s)\) \(Q = f_\theta(s)\) (신경망)
비선형 관계 불가 가능
안정성 높음 (수렴 보장) Experience Replay + Target Net 필요
데이터 효율 좋음 많은 데이터 필요
해석 가능성 \(\mathbf{w}\)에서 직접 해석 블랙박스
추천 상황 상태 차원 낮고, 관계 선형 상태 복잡, 비선형 관계

1.8 DTR과 임상시험: SMART

1.8.1 Sequential Multiple Assignment Randomized Trial

DTR을 임상시험으로 평가하는 표준 설계가 SMART:

SMART (Sequential Multiple Assignment Randomized Trial):

Stage 1: 모든 환자를 치료 A 또는 B에 무작위 배정
  ↓
중간 평가: 반응자(Responder) vs 비반응자(Non-responder) 분류
  ↓
Stage 2:
  반응자 → 유지(Maintain) 또는 강화(Intensify) 무작위 배정
  비반응자 → 전환(Switch) 또는 추가(Augment) 무작위 배정

1.8.2 SMART와 Q-learning의 연결

SMART 데이터에서 DTR 학습:

1. Stage 2부터 역방향 Q-learning (Backward Induction):
   Q₂(s₂, a₂) = E[Y | s₂, a₂]
   → Stage 2의 최적 행동 결정

2. Stage 1 Q-learning:
   Q₁(s₁, a₁) = E[r₁ + max_{a₂} Q₂(s₂, a₂) | s₁, a₁]
   → Stage 1의 최적 행동 결정 (미래 최적을 고려)

3. 최적 DTR:
   π* = (π₁*, π₂*) = (argmax Q₁, argmax Q₂)

1.8.3 AI Agent 맥락의 SMART

Stage 1 (Week 1~4): 기본 vs 공감형 프롬프트 무작위 배정
  ↓
중간 평가 (Week 4): 만족도 3.5 이상 = 반응자
  ↓
Stage 2 (Week 5~8):
  반응자:  현재 전략 유지 vs 단계별 가이드로 전환
  비반응자: 다른 전략으로 전환 vs 현재 전략 + 추가 개인화

→ 4가지 내장 DTR 비교:
  DTR 1: 기본 → (반응) 유지 / (비반응) 전환
  DTR 2: 기본 → (반응) 단계별 / (비반응) 추가
  DTR 3: 공감형 → (반응) 유지 / (비반응) 전환
  DTR 4: 공감형 → (반응) 단계별 / (비반응) 추가

1.9 DTR의 한계와 실무 고려

1.9.1 한계

1. 환경 모델 필요:
   - 전이 확률 P(s'|s,a)를 알거나 시뮬레이션할 수 있어야 함
   - 실제 서비스에서는 환경이 복잡하고 비정상적

2. 온라인 탐색의 위험:
   - Q-learning은 탐색 중 일부 사용자에게 나쁜 전략 적용
   - 의료/금융에서는 허용 불가

3. 데이터 효율성:
   - 상태 × 행동 공간이 크면 수렴까지 많은 에피소드 필요
   - 8주 × 사용자 1명 = 에피소드 1개 (비쌈)

4. 비정상 환경:
   - 사용자 선호가 시간에 따라 변할 수 있음
   - 정적 정책이 최적이 아닐 수 있음

→ 해결: Offline RL — 기존 서비스 로그로 안전하게 학습

1.10 요약

개념 핵심
Bandit → DTR 현재 행동이 미래 상태에 영향 → 순차적 의존성
MDP \((S, A, P, R, \gamma)\) — 상태, 행동, 전이, 보상, 할인
Q-function \(Q(s,a) = r + \gamma \max_{a'} Q(s', a')\)
TD error 현재 추정 vs 실제 관찰의 차이 → 업데이트 신호
선형 Q-learning \(Q = \mathbf{w}^T \phi(s)\), 해석 가능, 안정적
DQN 신경망 근사, 비선형 관계, Experience Replay 필요
SMART DTR 평가를 위한 순차적 무작위 배정 임상시험
한계 온라인 탐색 위험 → Offline RL로 해결

다음 글: 31 — Offline RL for Safe Policy Learning — 기존 서비스 로그에서 안전하게 최적 정책을 학습하는 방법을 다룬다.

Subscribe

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