jwkim
d4acdd0ac5
- feature_dqn_policy: ScoreNet(상태+카드특징 → 점수) 학습 정책 (replay+타깃넷) - feature_builder: 이산화 없는 연속 상태 벡터(9) + 테넌트 성향 벡터(5) - dqn_store: numpy 전용 서빙(컨테이너 PyTorch 불필요), DQN_SERVING 플래그, 미지원 테넌트는 Q-table 자동 폴백 - 파이프라인: build_card_embeddings -> train_feature_dqn -> export_dqn_serving(npz) - retrain_from_logs: 실로그 재학습 + OPE(SNIPS) 게이트, 통과 시에만 번들 교체(.prev 백업) - probe_serving_dqn / compare_qtable_vs_dqn: 배포 전 행동 점검 도구 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
252 lines
12 KiB
Python
252 lines
12 KiB
Python
"""기존 Q-Table(UCB) vs action-as-feature DQN 공정 비교 — 고객사 성향 조건화 환경 (최종).
|
||
|
||
같은 환경(FeatureBuyer 2축 + 협력사·고객사성향 랜덤)에서 동일 에피소드로 학습·평가.
|
||
- Q-Table: 이산 state 162칸 + 카드=슬롯. 성향(고객사) 입력 자체가 불가능 → 평균 성향에 수렴
|
||
- DQN : 연속 상태 + 성향 벡터 + 카드 특징(임베딩+전략/톤 one-hot)
|
||
|
||
평가 4종:
|
||
① 학습 카드 9장 — 평균보상(진짜 목적함수) + top3 적중(MC 정답 기준)
|
||
② zero-shot 11장 — 안 본 카드 2장 포함
|
||
③ 새 카드 첫 턴 사용률 — 구조적 차이
|
||
④ 성향 극단 테스트 — 같은 협력사, 성향만 바꿨을 때 카드를 바꾸는가
|
||
|
||
실행: APP_ENV=local python -m tools.compare_qtable_vs_dqn
|
||
"""
|
||
|
||
import random
|
||
|
||
import numpy as np
|
||
import torch
|
||
|
||
from eval_harness.buyer import Scenario
|
||
from eval_harness.feature_buyer import FeatureBuyer, SupplierProfile, sample_supplier
|
||
from negotiation.policies.feature_dqn_policy import FeatureDQNPolicy
|
||
from negotiation.policies.qtable_policy import UCBQTablePolicy
|
||
from negotiation.policies.base import EpisodeState, PolicyContext, Transition
|
||
from negotiation.qtable.domain.model.q_table import QTable
|
||
from negotiation.qtable.domain.model.snapshot import NegotiationOutcome
|
||
from negotiation.qtable.domain.service.feature_builder import (
|
||
STATE_FEATURE_DIM, TENANT_FEATURE_DIM, build_state_features)
|
||
from negotiation.qtable.domain.service.reward_calculator import RewardCalculator
|
||
from negotiation.qtable.domain.service.state_calculator import state_index
|
||
from tenancy.config_loader import TenantConfigLoader
|
||
from tools.train_feature_dqn import (
|
||
ANCHOR, HOLDOUT, MAX_TURNS, TARGET, load_cards, make_snapshot, pref_config, sample_tenant_pref)
|
||
|
||
|
||
# ---- 정책 어댑터 ------------------------------------------------------------------
|
||
class DQNAdapter:
|
||
name = "feature_dqn"
|
||
|
||
def __init__(self, policy, feat):
|
||
self.p, self.feat = policy, feat
|
||
|
||
def _sf(self, snap, tf):
|
||
return np.concatenate([build_state_features(snap), tf])
|
||
|
||
def choose(self, snap, tf, avail, greedy):
|
||
self.p.greedy = greedy
|
||
i, _, _ = self.p.select(self._sf(snap, tf), np.stack([self.feat[c] for c in avail]))
|
||
return avail[i]
|
||
|
||
def learn(self, snap, tf, card, reward, next_snap, next_avail, done):
|
||
sf = self._sf(snap, tf)
|
||
if done or next_snap is None:
|
||
self.p.remember(sf, self.feat[card], reward, None, None, True)
|
||
else:
|
||
self.p.remember(sf, self.feat[card], reward, self._sf(next_snap, tf),
|
||
np.stack([self.feat[c] for c in next_avail]), False)
|
||
self.p.train_step()
|
||
|
||
|
||
class QTableAdapter:
|
||
"""기존 UCBQTablePolicy. 성향(tf)은 구조상 받을 수 없다 — 이산 state 162칸에 그 축이 없음."""
|
||
|
||
name = "qtable_ucb"
|
||
|
||
def __init__(self, all_numbers, state_cfg, lr=0.1, gamma=0.95):
|
||
self.numbers = list(all_numbers)
|
||
self.a_of = {n: i for i, n in enumerate(self.numbers)}
|
||
self.state_cfg = state_cfg
|
||
self.qt = QTable(162, len(self.numbers), learning_rate=lr, discount_factor=gamma)
|
||
self.pol = UCBQTablePolicy(self.qt)
|
||
|
||
def choose(self, snap, tf, avail, greedy):
|
||
idx = state_index(snap, self.state_cfg)
|
||
if greedy:
|
||
q = self.qt.row(idx)
|
||
return max(avail, key=lambda c: q[self.a_of[c]])
|
||
mask = np.zeros(len(self.numbers), dtype=bool)
|
||
for c in avail:
|
||
mask[self.a_of[c]] = True
|
||
ctx = PolicyContext(state_index=idx, snapshot=snap, action_space_size=len(self.numbers),
|
||
episode=EpisodeState(), available_mask=mask)
|
||
return self.numbers[self.pol.select(ctx).action_id]
|
||
|
||
def learn(self, snap, tf, card, reward, next_snap, next_avail, done):
|
||
idx = state_index(snap, self.state_cfg)
|
||
nidx = state_index(next_snap, self.state_cfg) if (next_snap is not None and not done) else None
|
||
self.pol.update(Transition(state_index=idx, action_id=self.a_of[card], reward=reward,
|
||
next_state_index=nidx, done=done))
|
||
|
||
|
||
class RandomAdapter:
|
||
name = "random"
|
||
|
||
def __init__(self, seed=0):
|
||
self.rng = np.random.default_rng(seed)
|
||
|
||
def choose(self, snap, tf, avail, greedy):
|
||
return avail[self.rng.integers(len(avail))]
|
||
|
||
def learn(self, *a, **k):
|
||
pass
|
||
|
||
|
||
# ---- 공용 에피소드 -----------------------------------------------------------------
|
||
def run_episode(adapter, sup, tf, pool, strat, rc, seed, learn=True, greedy=False, forced_first=None):
|
||
buyer = FeatureBuyer(sup, strat, seed=seed, max_turns=MAX_TURNS)
|
||
scenario = Scenario(anchor_price=ANCHOR, target_price=TARGET, revenue_amount=sup.revenue_amount,
|
||
distribution_code=sup.distribution_code, partner_count=sup.partner_count)
|
||
price0 = TARGET * 1.15
|
||
price, used, total_r, first_card = price0, set(), 0.0, None
|
||
|
||
for turn in range(1, MAX_TURNS + 1):
|
||
acceptance = max(0.0, (price0 - price) / price0)
|
||
snap = make_snapshot(sup, price, turn, acceptance)
|
||
avail = [c for c in pool if c not in used] or list(pool)
|
||
if turn == 1 and forced_first is not None:
|
||
card = forced_first
|
||
else:
|
||
card = adapter.choose(snap, tf, avail, greedy)
|
||
used.add(card)
|
||
if first_card is None:
|
||
first_card = card
|
||
|
||
resp = buyer.respond(card, scenario, turn, price)
|
||
price = resp.new_price
|
||
done = resp.accept or price <= ANCHOR or turn >= MAX_TURNS
|
||
success = resp.accept or price <= ANCHOR
|
||
outcome = (NegotiationOutcome.SUCCESS if success
|
||
else NegotiationOutcome.FAILURE if done else NegotiationOutcome.ONGOING)
|
||
# 채점은 최종 결과 시점만 (중간 턴 0 → γ 부트스트랩으로 전파).
|
||
# 진행 중 보상을 누적하면 '질질 끄는 전략'이 부당하게 유리해지는 인공물이 생긴다.
|
||
r = rc.calculate(make_snapshot(sup, price, turn, acceptance, outcome)).total if done else 0.0
|
||
total_r += r
|
||
|
||
if learn:
|
||
if done:
|
||
adapter.learn(snap, tf, card, r, None, None, True)
|
||
else:
|
||
acc2 = max(0.0, (price0 - price) / price0)
|
||
nsnap = make_snapshot(sup, price, turn + 1, acc2)
|
||
navail = [c for c in pool if c not in used] or list(pool)
|
||
adapter.learn(snap, tf, card, r, nsnap, navail, False)
|
||
if done:
|
||
return total_r, success, price, first_card
|
||
return total_r, False, price, first_card
|
||
|
||
|
||
# ---- MC 정답 랭킹: 이 (협력사, 성향)에서 진짜 좋은 첫 카드 top-k ---------------------
|
||
_rand = RandomAdapter(seed=1)
|
||
|
||
def rank_cards_mc(sup, tf, pool, strat, rc, seed, sims=6, k=3):
|
||
means = {}
|
||
for c in pool:
|
||
rs = [run_episode(_rand, sup, tf, pool, strat, rc, seed=seed + 17 * s,
|
||
learn=False, greedy=False, forced_first=c)[0] for s in range(sims)]
|
||
means[c] = np.mean(rs)
|
||
return sorted(means, key=lambda c: -means[c])[:k]
|
||
|
||
|
||
# ---- 학습/평가 ---------------------------------------------------------------------
|
||
def train(adapter, pool, strat, base_reward, state_cfg, episodes, seed):
|
||
rng = np.random.default_rng(seed)
|
||
for ep in range(1, episodes + 1):
|
||
sup = sample_supplier(rng)
|
||
rcfg, tf = sample_tenant_pref(rng, base_reward)
|
||
rc = RewardCalculator(rcfg, state_cfg)
|
||
run_episode(adapter, sup, tf, pool, strat, rc, seed=seed * 100 + ep, learn=True)
|
||
|
||
|
||
def evaluate(adapter, pool, strat, base_reward, state_cfg, n=300, seed0=777, label=""):
|
||
from negotiation.qtable.domain.service.feature_builder import build_tenant_features
|
||
rng = np.random.default_rng(seed0)
|
||
rewards, succ, ratios, hits, holdout_first = [], 0, [], 0, 0
|
||
for i in range(n):
|
||
sup = sample_supplier(rng)
|
||
rcfg, tf = sample_tenant_pref(rng, base_reward)
|
||
rc = RewardCalculator(rcfg, state_cfg)
|
||
good = rank_cards_mc(sup, tf, pool, strat, rc, seed=seed0 * 7 + i)
|
||
r, ok, price, first = run_episode(adapter, sup, tf, pool, strat, rc,
|
||
seed=seed0 * 1000 + i, learn=False, greedy=True)
|
||
rewards.append(r); succ += ok; ratios.append(price / TARGET)
|
||
hits += (first in good); holdout_first += (first in HOLDOUT)
|
||
m, ci = float(np.mean(rewards)), float(1.96 * np.std(rewards) / np.sqrt(n))
|
||
print(f"{label:<14} mean_rwd={m:.4f} ±{ci:.4f} success={succ/n:.3f} "
|
||
f"settled/tgt={np.mean(ratios):.3f} top3_hit={hits/n:.3f} 새카드첫턴={holdout_first/n:.3f}")
|
||
|
||
|
||
def pref_behavior_test(adapters, pool, strat, base_reward, state_cfg):
|
||
"""④ 같은 협력사, 성향만 바꿨을 때 카드를 바꾸는가 (greedy).
|
||
|
||
첫 턴은 '일단 깎기'가 공통 정답이라 성향 차이가 잘 안 드러난다.
|
||
→ 협상 중반(가격이 이미 target 근처, 3턴째) 상태를 함께 프로브: 여기서
|
||
성사중시는 '마무리(수락 잘 되는) 카드', 가격중시는 '더 깎는 카드'가 갈려야 한다.
|
||
"""
|
||
from negotiation.qtable.domain.service.feature_builder import build_tenant_features
|
||
sups = [SupplierProfile(5_000_000, 3, "A"), # 소형·경쟁多
|
||
SupplierProfile(200_000_000, 1, "A")] # 대형·단독
|
||
probes = [("첫턴", TARGET * 1.15, 1, 0.0),
|
||
("중반(3턴,가격↓)", TARGET * 1.02, 3, 0.11)]
|
||
for pr_name, price, turn, acc in probes:
|
||
print(f"\n ── 프로브: {pr_name} (price={price:.0f}) ──")
|
||
print(f" {'협력사':<13} {'성향':<9} " + " ".join(f"{a.name:<15}" for a in adapters))
|
||
for sup in sups:
|
||
row = {}
|
||
for p, pname in [(0.05, "성사중시"), (0.95, "가격중시")]:
|
||
rcfg = pref_config(base_reward, p)
|
||
tf = build_tenant_features(rcfg)
|
||
picks = []
|
||
for a in adapters:
|
||
snap = make_snapshot(sup, price, turn, acc)
|
||
picks.append(a.choose(snap, tf, pool, True))
|
||
seg = f"{sup.segment[0]}·{sup.segment[1]}"
|
||
print(f" {seg:<13} {pname:<9} " + " ".join(f"{c}(전략{strat[c]})".ljust(15) for c in picks))
|
||
|
||
|
||
def main(episodes=10000, seed=42):
|
||
random.seed(seed); np.random.seed(seed); torch.manual_seed(seed)
|
||
numbers, feat, strat = load_cards()
|
||
train_pool = [c for c in numbers if c not in HOLDOUT]
|
||
tcfg = TenantConfigLoader().load("ktcommerce")
|
||
card_dim = feat[numbers[0]].shape[0]
|
||
print(f"환경: 2축 FeatureBuyer + 성향 랜덤 · 학습 {episodes}ep · 카드특징 {card_dim}차원 "
|
||
f"(임베딩384+전략4+톤4) · 학습 {len(train_pool)}장 / 홀드아웃 {HOLDOUT}")
|
||
|
||
qt = QTableAdapter(numbers, tcfg.state)
|
||
dqn = DQNAdapter(FeatureDQNPolicy(state_dim=STATE_FEATURE_DIM + TENANT_FEATURE_DIM,
|
||
card_dim=card_dim, eps_decay=4000), feat)
|
||
|
||
print("\n[학습] qtable_ucb ...")
|
||
train(qt, train_pool, strat, tcfg.reward, tcfg.state, episodes, seed)
|
||
print("[학습] feature_dqn ...")
|
||
train(dqn, train_pool, strat, tcfg.reward, tcfg.state, episodes, seed)
|
||
|
||
print("\n=== ① 학습 카드 9장 풀 ===")
|
||
evaluate(RandomAdapter(seed), train_pool, strat, tcfg.reward, tcfg.state, label="random")
|
||
evaluate(qt, train_pool, strat, tcfg.reward, tcfg.state, label="qtable_ucb")
|
||
evaluate(dqn, train_pool, strat, tcfg.reward, tcfg.state, label="feature_dqn")
|
||
|
||
print("\n=== ② zero-shot 11장 풀 (안 본 카드 2장 포함) ===")
|
||
evaluate(RandomAdapter(seed), numbers, strat, tcfg.reward, tcfg.state, label="random")
|
||
evaluate(qt, numbers, strat, tcfg.reward, tcfg.state, label="qtable_ucb")
|
||
evaluate(dqn, numbers, strat, tcfg.reward, tcfg.state, label="feature_dqn")
|
||
|
||
print("\n=== ④ 성향 극단 테스트 — 같은 협력사, 성향만 바꾸면 카드를 바꾸는가 (11장 풀) ===")
|
||
pref_behavior_test([qt, dqn], numbers, strat, tcfg.reward, tcfg.state)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|