1 Offline RL for Safe Policy Learning
1.1 왜 Offline RL인가
1.1.1 온라인 탐색의 위험
온라인 RL(Q-learning, DQN 등)은 학습 중 탐색이 필수다. 이는 일부 사용자에게 의도적으로 나쁜(또는 알 수 없는) 행동을 적용한다는 의미다.
온라인 RL의 탐색 비용:
의료: 환자에게 검증되지 않은 투약 → 생명 위험
금융: 고객에게 검증되지 않은 투자 전략 → 재산 손실
서비스: 사용자에게 나쁜 프롬프트 전략 → 이탈, 불만족
ε-greedy (ε=0.1):
→ 전체 상호작용의 10%에서 무작위 행동
→ 10,000명 중 1,000명이 무작위 전략 노출
→ 서비스 규모에서 허용 불가1.1.2 Offline RL의 핵심 아이디어
기존에 수집된 로그 데이터만으로 최적 정책을 학습:
\[ \mathcal{D} = \{(s_i, a_i, r_i, s_i')\}_{i=1}^{N} \]
- 데이터는 행동 정책(behavior policy) \(\mu\)로 수집됨
- 학습 목표: \(\mu\)와 다른 목표 정책(target policy) \(\pi\)를 찾아 성능 개선
- 환경과 추가 상호작용 없이 \(\mathcal{D}\)만 사용
기존 서비스 로그:
6개월간 수집된 (사용자 상태, 적용된 전략, 만족도, 다음 상태)
행동 정책 μ: 규칙 기반 시스템 (세그먼트별 고정 전략)
Offline RL 목표:
이 로그에서 μ보다 나은 π를 학습
→ 새로운 탐색 없이, 기존 데이터에서 최대한의 정보 추출1.2 Distribution Shift 문제
1.2.1 핵심 문제: 데이터에 없는 (s, a) 쌍
Offline RL의 가장 큰 도전은 distribution shift다.
\[ \pi(a|s) \neq \mu(a|s) \implies (s, a) \text{ 쌍의 분포가 다름} \]
행동 정책 μ의 데이터:
SI 세그먼트 → 항상 "기본" 전략 적용
→ (SI, 공감형)이나 (SI, 단계별) 데이터가 거의 없음
문제:
Q(SI, 공감형)을 추정할 데이터가 부족
→ Q가 부정확 → 과대 추정 가능
→ 학습된 정책이 "데이터에 없는 행동"을 자신 있게 선택
→ 실제 적용 시 실패1.2.2 Q값 과대 추정의 메커니즘
\[ \hat{Q}(s, a) = r + \gamma \max_{a'} \hat{Q}(s', a') \]
\(\max\) 연산자가 노이즈를 양의 방향으로 증폭:
데이터에 있는 행동: Q 추정이 비교적 정확 (데이터로 검증됨)
데이터에 없는 행동: Q 추정이 노이즈 투성이
→ max가 노이즈의 최대값을 선택
→ 과대 추정
→ 정책이 "데이터에 없는, Q가 높은" 행동을 선택
→ 실제로는 나쁜 행동1.3 Importance Sampling
1.3.1 기본 Off-Policy 보정
가장 단순한 접근: 행동 정책 \(\mu\)의 데이터로 목표 정책 \(\pi\)의 가치를 추정.
\[ \hat{V}^{\pi} = \frac{1}{n} \sum_{i=1}^{n} \frac{\pi(a_i | s_i)}{\mu(a_i | s_i)} r_i \]
- \(\pi(a|s) / \mu(a|s)\): 중요도 비율(importance weight)
- 직관: \(\mu\)가 잘 선택하지 않는 행동의 보상에 큰 가중치를 부여하여 \(\pi\) 하에서의 기대값을 복원
1.3.2 순차적 의사결정에서의 IS
에피소드 전체에 대한 importance weight:
\[ w_i = \prod_{t=0}^{T} \frac{\pi(a_t^{(i)} | s_t^{(i)})}{\mu(a_t^{(i)} | s_t^{(i)})} \]
1.3.3 높은 분산 문제
문제: 궤적이 길수록 w_i의 분산이 기하급수적으로 증가
예시 (T=8, 행동 3가지):
π가 선택하는 행동을 μ가 33% 확률로 선택한다면:
w = (1/0.33)^8 ≈ 6561 (한 궤적의 가중치)
대부분의 궤적: w ≈ 0 (거의 무시됨)
극소수 궤적: w >> 1 (극단적 가중치)
→ 추정량이 소수 궤적에 의존 → 분산 폭발해결책: - Weighted IS (WIS): \(w_i\)를 정규화하여 분산 감소 - Per-Decision IS: 단계별로 IS 적용, 미래 가중치 분리 - 그러나 근본적 한계 존재 → 가치 함수 기반 방법(CQL, BCQ) 선호
1.4 Conservative Q-Learning (CQL)
1.4.1 핵심 아이디어
데이터에 없는 행동의 Q값을 보수적으로 낮게 추정하여 distribution shift 방지.
1.4.2 CQL 손실 함수
\[ \mathcal{L}_{\text{CQL}} = \underbrace{\alpha \left( \mathbb{E}_{s \sim \mathcal{D}} \left[ \log \sum_a \exp Q(s, a) \right] - \mathbb{E}_{(s,a) \sim \mathcal{D}} [Q(s, a)] \right)}_{\text{CQL penalty}} + \underbrace{\frac{1}{2} \mathbb{E}_{(s,a,r,s') \sim \mathcal{D}} \left[ (Q(s,a) - \hat{\mathcal{B}}^\pi Q(s,a))^2 \right]}_{\text{Bellman error}} \]
CQL penalty 분해:
\[ \underbrace{\log \sum_a \exp Q(s, a)}_{\text{모든 행동의 Q를 올리면 페널티}} - \underbrace{Q(s, a_{\text{data}})}_{\text{데이터에 있는 행동의 Q는 유지}} \]
- \(\text{logsumexp}\): 모든 행동의 Q를 높이면 증가 → 억제
- \(-Q(s, a_{\text{data}})\): 데이터에 있는 행동의 Q는 높여도 됨
- 효과: 데이터에 없는 행동의 Q만 선택적으로 낮춤
1.4.3 α 하이퍼파라미터
\[ \alpha \text{ 크면:} \quad \text{매우 보수적 (데이터에 있는 행동만 선택)} \] \[ \alpha \text{ 작으면:} \quad \text{덜 보수적 (일부 외삽 허용)} \] \[ \alpha = 0: \quad \text{일반 Q-learning (Offline에서 발산 위험)} \]
실무에서 \(\alpha \in [0.1, 5.0]\) 범위로 탐색, 보통 \(\alpha \in [1.0, 2.0]\)에서 시작.
1.5 Batch Constrained Q-learning (BCQ)
1.5.1 핵심: 데이터에 있는 행동만 고려
CQL이 Q값을 낮추는 방식이라면, BCQ는 행동 선택 자체를 제한:
\[ \pi(s) = \arg\max_{a: \hat{\mu}(a|s) > \tau} Q(s, a) \]
- \(\hat{\mu}(a|s)\): 행동 정책의 추정 (생성 모델 또는 분류기)
- \(\tau\): 임계값 — 데이터에서 충분히 관찰된 행동만 후보
- 효과: 데이터에 없는 행동은 아예 후보에서 제외
BCQ의 직관:
데이터에서 "SI 세그먼트 + 공감형"이 거의 없으면:
→ μ̂(공감형|SI) < τ
→ 공감형은 후보에서 제외
→ Q(SI, 공감형)이 아무리 높아도 선택되지 않음
→ 안전한 행동만 선택1.5.2 CQL vs BCQ 비교
| 측면 | CQL | BCQ |
|---|---|---|
| 접근 | Q값을 보수적으로 | 행동 후보를 제한 |
| 추가 모델 | 없음 | 행동 정책 \(\hat{\mu}\) 추정 필요 |
| 외삽 허용 | 약간 (α 조절) | 거의 없음 |
| 성능 | 일반적으로 우세 | 보수적이지만 안전 |
| 구현 복잡도 | 손실 함수 추가 | 생성 모델 추가 |
1.6 PyTorch 구현: CQLAgent 전체 코드
import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
from collections import deque
import random
class QNetwork(nn.Module):
"""Q-function 신경망"""
def __init__(self, state_dim, n_actions, hidden_dims=(256, 128)):
super().__init__()
layers = []
prev_dim = state_dim
for h in hidden_dims:
layers.extend([nn.Linear(prev_dim, h), nn.ReLU()])
prev_dim = h
layers.append(nn.Linear(prev_dim, n_actions))
self.net = nn.Sequential(*layers)
def forward(self, x):
return self.net(x)
class CQLAgent:
"""
Conservative Q-Learning for Offline RL
CQL penalty: logsumexp(Q(s, ·)) - Q(s, a_data)
→ 데이터에 없는 행동의 Q를 보수적으로 추정
"""
def __init__(self, state_dim, n_actions, alpha=1.0, gamma=0.99,
lr=3e-4, tau=0.005):
self.n_actions = n_actions
self.alpha = alpha # CQL 보수성 강도
self.gamma = gamma
self.tau = tau # Target network soft update 비율
# Q-network 및 Target network
self.q_net = QNetwork(state_dim, n_actions)
self.target_net = QNetwork(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.train_steps = 0
self.loss_history = {"bellman": [], "cql": [], "total": []}
def select_action(self, state, greedy=True):
"""학습된 정책으로 행동 선택"""
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 train_step(self, batch):
"""CQL 학습 스텝"""
states, actions, rewards, next_states, dones = batch
states = torch.FloatTensor(states)
actions = torch.LongTensor(actions)
rewards = torch.FloatTensor(rewards)
next_states = torch.FloatTensor(next_states)
dones = torch.FloatTensor(dones)
# --- Bellman Loss ---
q_values = self.q_net(states)
q_selected = q_values.gather(1, actions.unsqueeze(1)).squeeze(1)
with torch.no_grad():
next_q = self.target_net(next_states).max(1)[0]
target = rewards + self.gamma * next_q * (1 - dones)
bellman_loss = F.mse_loss(q_selected, target)
# --- CQL Penalty ---
# logsumexp(Q(s, ·)): 모든 행동의 Q를 높이면 페널티
logsumexp_q = torch.logsumexp(q_values, dim=1).mean()
# Q(s, a_data): 데이터에 있는 행동의 Q는 유지
q_data = q_values.gather(1, actions.unsqueeze(1)).squeeze(1).mean()
cql_loss = logsumexp_q - q_data
# --- Total Loss ---
total_loss = bellman_loss + self.alpha * cql_loss
self.optimizer.zero_grad()
total_loss.backward()
# Gradient clipping for stability
torch.nn.utils.clip_grad_norm_(self.q_net.parameters(), max_norm=1.0)
self.optimizer.step()
# Target network soft update
self._soft_update()
# 기록
self.train_steps += 1
self.loss_history["bellman"].append(bellman_loss.item())
self.loss_history["cql"].append(cql_loss.item())
self.loss_history["total"].append(total_loss.item())
return total_loss.item()
def _soft_update(self):
"""Target network 소프트 업데이트: θ' ← τθ + (1-τ)θ'"""
for param, target_param in zip(
self.q_net.parameters(), self.target_net.parameters()
):
target_param.data.copy_(
self.tau * param.data + (1 - self.tau) * target_param.data
)
# --- 오프라인 데이터 로딩 및 학습 ---
class OfflineDataset:
"""서비스 로그를 오프라인 학습용으로 관리"""
def __init__(self, data_path=None):
self.buffer = []
def load_from_logs(self, logs):
"""
logs: list of dicts with keys:
state, action, reward, next_state, done
"""
for log in logs:
self.buffer.append((
np.array(log["state"]),
log["action"],
log["reward"],
np.array(log["next_state"]),
float(log["done"])
))
def sample(self, batch_size):
batch = random.sample(self.buffer, min(batch_size, len(self.buffer)))
states, actions, rewards, next_states, dones = zip(*batch)
return (
np.array(states),
np.array(actions),
np.array(rewards),
np.array(next_states),
np.array(dones)
)
def __len__(self):
return len(self.buffer)
# 학습 루프
def train_cql_offline(agent, dataset, n_epochs=500, batch_size=256,
eval_interval=50):
"""CQL 오프라인 학습"""
print(f"Offline dataset size: {len(dataset)}")
print(f"CQL alpha: {agent.alpha}")
print("=" * 50)
for epoch in range(1, n_epochs + 1):
epoch_losses = []
# 에폭당 여러 배치 학습
n_batches = max(1, len(dataset) // batch_size)
for _ in range(n_batches):
batch = dataset.sample(batch_size)
loss = agent.train_step(batch)
epoch_losses.append(loss)
if epoch % eval_interval == 0:
avg_loss = np.mean(epoch_losses)
avg_bellman = np.mean(agent.loss_history["bellman"][-n_batches:])
avg_cql = np.mean(agent.loss_history["cql"][-n_batches:])
print(f"Epoch {epoch:4d} | Loss: {avg_loss:.4f} | "
f"Bellman: {avg_bellman:.4f} | CQL: {avg_cql:.4f}")
return agent
# 사용 예시
dataset = OfflineDataset()
# dataset.load_from_logs(service_logs) # 실제 서비스 로그 로드
cql_agent = CQLAgent(
state_dim=6,
n_actions=3,
alpha=1.0, # CQL 보수성 (1.0 = 적당히 보수적)
gamma=0.9,
lr=3e-4,
tau=0.005
)
# trained_agent = train_cql_offline(cql_agent, dataset)1.7 정책 평가: Off-Policy Evaluation (OPE)
학습된 정책을 실제 배포하기 전에 오프라인 데이터로 성능을 추정.
1.7.1 Weighted Importance Sampling (WIS)
IS의 분산을 줄인 버전:
\[ \hat{V}_{\text{WIS}}^{\pi} = \frac{\sum_{i=1}^{n} w_i \cdot G_i}{\sum_{i=1}^{n} w_i} \]
- \(w_i = \prod_{t=0}^{T} \frac{\pi(a_t^{(i)}|s_t^{(i)})}{\mu(a_t^{(i)}|s_t^{(i)})}\): 중요도 가중치
- \(G_i = \sum_{t=0}^{T} \gamma^t r_t^{(i)}\): 할인 누적 보상
- 정규화로 분산 감소, 약간의 편향 발생
def weighted_importance_sampling(episodes, pi, mu, gamma=0.9):
"""
Weighted Importance Sampling for OPE
episodes: list of [(s, a, r), ...]
pi: 목표 정책 (학습된 CQL 정책)
mu: 행동 정책 (데이터 수집 시 사용된 정책)
"""
weights = []
returns = []
for episode in episodes:
w = 1.0
G = 0.0
for t, (s, a, r) in enumerate(episode):
w *= pi(a, s) / max(mu(a, s), 1e-8) # 0 나눗셈 방지
G += (gamma ** t) * r
weights.append(w)
returns.append(G)
weights = np.array(weights)
returns = np.array(returns)
# 가중 평균
if weights.sum() > 0:
return np.sum(weights * returns) / np.sum(weights)
return 0.01.7.2 Doubly Robust Estimator
IS와 가치 함수 추정을 결합하여 어느 한쪽이 맞으면 일관된 추정:
\[ \hat{V}_{\text{DR}}^{\pi} = \frac{1}{n} \sum_{i=1}^{n} \left[ \hat{V}^{\pi}(s_0^{(i)}) + \sum_{t=0}^{T} \gamma^t \rho_t^{(i)} \left( r_t^{(i)} + \gamma \hat{V}^{\pi}(s_{t+1}^{(i)}) - \hat{Q}^{\pi}(s_t^{(i)}, a_t^{(i)}) \right) \right] \]
- \(\rho_t = \prod_{\tau=0}^{t} \frac{\pi(a_\tau|s_\tau)}{\mu(a_\tau|s_\tau)}\): 누적 IS 비율
- \(\hat{V}^{\pi}, \hat{Q}^{\pi}\): 가치/행동-가치 함수 추정
- 이중 안전성: IS가 부정확해도 \(\hat{Q}\)가 맞으면 OK, 반대도 OK
def doubly_robust_estimator(episodes, pi, mu, q_hat, v_hat, gamma=0.9):
"""
Doubly Robust OPE Estimator
q_hat: Q-function 추정 (CQL의 Q-network)
v_hat: Value function 추정
"""
estimates = []
for episode in episodes:
s0 = episode[0][0]
dr_value = v_hat(s0)
rho = 1.0 # 누적 IS 비율
for t, (s, a, r) in enumerate(episode):
rho *= pi(a, s) / max(mu(a, s), 1e-8)
# 다음 상태의 가치
if t + 1 < len(episode):
next_v = v_hat(episode[t+1][0])
else:
next_v = 0.0
# DR 보정항
correction = rho * (r + gamma * next_v - q_hat(s, a))
dr_value += (gamma ** t) * correction
estimates.append(dr_value)
return np.mean(estimates)1.7.3 OPE 방법 비교
| 방법 | 편향 | 분산 | 필요 조건 |
|---|---|---|---|
| IS | 없음 (비편향) | 매우 높음 | \(\mu(a\|s) > 0\) |
| WIS | 약간 (편향) | 중간 | \(\mu(a\|s) > 0\) |
| Direct Method | 높음 (모델 의존) | 낮음 | \(\hat{Q}\) 추정 |
| Doubly Robust | 낮음 | 중간 | \(\hat{Q}\) + IS |
1.8 실무 예시: AI Agent 서비스 로그 → 최적 정책
1.8.1 전체 파이프라인
# Step 1: 서비스 로그 수집 (6개월)
# 행동 정책 μ: 세그먼트별 고정 전략
# SI → 기본, MIEP → 단계별, N → 공감형
# Step 2: 데이터 전처리
def preprocess_logs(raw_logs):
"""서비스 로그 → (s, a, r, s', done) 튜플"""
processed = []
for user_id, sessions in raw_logs.groupby("user_id"):
sessions = sessions.sort_values("week")
for i in range(len(sessions) - 1):
current = sessions.iloc[i]
next_row = sessions.iloc[i + 1]
state = np.array([
current["satisfaction"] / 5.0,
current["turn_count"] / 20.0,
current["emotion_score"],
current["seg_miep"],
current["seg_n"],
current["week"] / 8.0
])
next_state = np.array([
next_row["satisfaction"] / 5.0,
next_row["turn_count"] / 20.0,
next_row["emotion_score"],
next_row["seg_miep"],
next_row["seg_n"],
next_row["week"] / 8.0
])
processed.append({
"state": state,
"action": current["strategy_id"], # 0, 1, 2
"reward": current["satisfaction"],
"next_state": next_state,
"done": i == len(sessions) - 2
})
return processed
# Step 3: CQL 학습
dataset = OfflineDataset()
dataset.load_from_logs(preprocess_logs(raw_logs))
cql_agent = CQLAgent(state_dim=6, n_actions=3, alpha=1.5)
trained_agent = train_cql_offline(cql_agent, dataset, n_epochs=300)
# Step 4: OPE로 정책 평가 (배포 전)
behavior_policy = lambda a, s: { # 기존 고정 정책
0: 0.8 if s[3] < 0.5 and s[4] < 0.5 else 0.1, # SI → 기본
1: 0.8 if s[4] > 0.5 else 0.1, # N → 공감형
2: 0.8 if s[3] > 0.5 else 0.1 # MIEP → 단계별
}[a]
def learned_policy(a, s):
"""CQL로 학습된 정책"""
with torch.no_grad():
q = trained_agent.q_net(torch.FloatTensor(s).unsqueeze(0))
best_a = q.argmax().item()
return 1.0 if a == best_a else 0.0
# WIS로 정책 가치 추정
v_behavior = weighted_importance_sampling(
test_episodes, behavior_policy, behavior_policy)
v_learned = weighted_importance_sampling(
test_episodes, learned_policy, behavior_policy)
print(f"행동 정책 (기존) 추정 가치: {v_behavior:.3f}")
print(f"CQL 정책 (학습) 추정 가치: {v_learned:.3f}")
print(f"개선율: {(v_learned - v_behavior) / abs(v_behavior) * 100:.1f}%")
# Step 5: 소규모 A/B 테스트로 최종 검증
# CQL 정책 vs 기존 정책 → 5% 트래픽으로 1주 검증 후 확대1.9 A/B → Bandit → DTR → Offline RL 진화 경로
수준 1: A/B 테스트 + 통계 모델 (LMM/GEE)
┌─────────────────────────────────────────┐
│ 설계: 무작위 배정, 2~4주 관찰 │
│ 분석: LMM으로 처치 효과 추정 │
│ 결정: p-value < 0.05면 승자 채택 │
│ 장점: 인과 추론 엄밀, 해석 쉬움 │
│ 단점: 정적, 개인화 없음, 탐색 비용 │
└─────────────┬───────────────────────────┘
│ "사용자마다 최적 전략이 다르다"
▼
수준 2: Contextual Bandit (29번 파일)
┌─────────────────────────────────────────┐
│ 설계: 문맥 관찰 → 실시간 행동 선택 │
│ 알고리즘: LinUCB, Thompson Sampling │
│ 장점: 개인화, Regret 41~48% 감소 │
│ 단점: 순차적 의존성 무시 │
└─────────────┬───────────────────────────┘
│ "지금 행동이 미래 상태를 바꾼다"
▼
수준 3: DTR / Q-learning (30번 파일)
┌─────────────────────────────────────────┐
│ 설계: MDP, 순차적 최적 결정 │
│ 알고리즘: Q-learning, DQN │
│ 장점: 장기 보상 최적화 │
│ 단점: 온라인 탐색 위험 │
└─────────────┬───────────────────────────┘
│ "탐색 없이 기존 데이터로 학습하고 싶다"
▼
수준 4: Offline RL (이 파일)
┌─────────────────────────────────────────┐
│ 설계: 기존 로그만 사용, 추가 탐색 없음 │
│ 알고리즘: CQL, BCQ │
│ 검증: OPE (WIS, DR) → 소규모 A/B │
│ 장점: 안전, 기존 인프라 활용 │
│ 단점: 데이터 커버리지 한계 │
└─────────────────────────────────────────┘1.9.1 실무 권장 경로
Phase 1: A/B 테스트로 기본 처치 효과 확인 (LMM)
→ "공감형이 N 세그먼트에 효과적" 같은 기본 지식 확보
Phase 2: Contextual Bandit으로 실시간 최적화
→ LinUCB로 세그먼트 × 문맥별 최적 전략 탐색
→ 데이터 축적
Phase 3: 축적된 데이터로 Offline RL 학습
→ CQL로 순차적 최적 정책 학습
→ OPE로 안전 검증 후 배포
Phase 4: 지속적 개선
→ 새 로그 축적 → Offline RL 재학습 → OPE → 배포
→ 필요 시 소규모 온라인 탐색 (안전 범위 내)1.10 통계 모델 + RL 결합: CATE → 보상 함수
1.10.1 인과 추론으로 보상 함수 설계
통계 모델(인과 추론)로 조건부 평균 처치 효과(CATE)를 추정하고, 이를 RL의 보상 함수에 활용:
\[ \text{CATE}(x) = \mathbb{E}[Y(1) - Y(0) | X = x] \]
from sklearn.linear_model import LinearRegression
def estimate_cate_tlearner(df, treatment_col, outcome_col, feature_cols):
"""
T-learner로 CATE 추정
처치군과 대조군에 별도 모델을 학습하여 개인별 처치 효과 추정
"""
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]
)
# 개인별 CATE
df["cate"] = (
model_t.predict(df[feature_cols])
- model_c.predict(df[feature_cols])
)
return df
# RL 보상 함수에 CATE 통합
def cate_informed_reward(state, action, cate_model):
"""
CATE 기반 보상 함수:
처치 효과가 높은 사용자에게 적극적 전략 적용 시 보상 증가
"""
base_reward = observe_satisfaction(state, action)
if action > 0: # 개인화 전략 적용
cate = cate_model.predict(state.reshape(1, -1))[0]
# CATE가 높은 사용자 → 보상 보너스
bonus = max(0, cate) * 0.3
return base_reward + bonus
return base_reward1.10.2 왜 결합하는가
통계 모델 단독:
"이 사용자에게 공감형이 효과적" (CATE > 0)
→ 정적 결정, 순서/타이밍 고려 없음
RL 단독:
보상 함수를 직접 설계해야 함
→ 도메인 지식 없이 만족도만 사용하면 근시안적
CATE + RL:
1. CATE로 "어떤 사용자에게 어떤 전략이 효과적인지" 추정
2. CATE를 RL 보상에 반영 → 처치 효과가 높은 상황에서 적극 적용
3. RL이 순서/타이밍까지 최적화
→ 인과적 근거 있는 동적 최적화1.11 요약
| 개념 | 핵심 |
|---|---|
| Offline RL | 기존 로그만으로 정책 학습, 추가 탐색 없음 |
| Distribution Shift | 행동 정책 ≠ 목표 정책 → Q 과대 추정 위험 |
| Importance Sampling | IS 비율로 off-policy 보정, 높은 분산 한계 |
| CQL | logsumexp 페널티로 데이터 밖 Q를 보수적 추정 |
| BCQ | 데이터에 있는 행동만 후보로 제한 |
| OPE | WIS, Doubly Robust로 배포 전 정책 평가 |
| 진화 경로 | A/B → Bandit → DTR → Offline RL |
| CATE + RL | 인과 추론으로 보상 함수 강화 → 근거 있는 최적화 |