"""retrain_from_logs — experience_logs 실데이터로 feature_dqn 오프라인 재학습 + OPE 게이트. 파이프라인: ① learning.experience_logs 로드(전 테넌트 — 범용 에이전트는 테넌트를 특징으로 조건화하므로 통합 학습) ② 세션별 에피소드 재구성: 카드턴(done=False) N개 + 종료행(done=True) 1개. 보상은 학습 규약(최종 결과 시점만 채점)에 맞춰 종료행 reward 만 쓰고 중간턴은 0. ③ 현재 체크포인트에서 fine-tune (낮은 lr — 시뮬 사전학습 망각 방지) ④ OPE(SNIPS, 궤적 IS): 후보 모델 vs 현재 서빙 번들. 후보가 못 넘으면 배포하지 않는다. ⑤ 통과 시 dqn_serving.npz 원자적 교체(직전본 .prev 백업) → `docker compose build agent && up -d agent` 로 배포. 실행(호스트, torch+DB 필요): APP_ENV=local python -m tools.retrain_from_logs 환경변수: MIN_EPISODES(기본 200) 재학습 최소 에피소드 수 — 미달 시 skip (과적합 방지) EPOCHS(기본 20) / LR(기본 1e-4) / FORCE_DEPLOY=1 (OPE 게이트 무시 — 테스트 전용) 주의: 서빙이 greedy(탐색 없음)라 로그가 선택 편향됨 — OPE 의 유효표본(ESS)이 작으면 게이트가 보수적으로 배포를 막는다. 이는 의도된 동작이다(조용한 성능저하 방지). """ import asyncio import json import os from collections import defaultdict import numpy as np import torch from common.database.db_session_manager import DB_SESSION_MNG from common.database.model.models import ExperienceLog from common.enums import DBType, DBWRType from negotiation.policies.feature_dqn_policy import FeatureDQNPolicy from negotiation.qtable.domain.model.snapshot import NegotiationSnapshot from negotiation.qtable.domain.service.feature_builder import ( STATE_FEATURE_DIM, TENANT_FEATURE_DIM, build_state_features, build_tenant_features) from sqlalchemy import select from tenancy.config_loader import TenantConfigLoader from tools.export_dqn_serving import CKPT_PATH, OUT_PATH, export_bundle from tools.train_feature_dqn import load_cards _HERE = os.path.dirname(os.path.abspath(__file__)) RETRAIN_CKPT = os.path.join(_HERE, "..", "artifacts", "feature_dqn_retrained.pt") REPORT_PATH = os.path.join(_HERE, "..", "artifacts", "retrain_report.json") MIN_EPISODES = int(os.getenv("MIN_EPISODES", "200")) EPOCHS = int(os.getenv("EPOCHS", "20")) LR = float(os.getenv("LR", "1e-4")) FORCE_DEPLOY = os.getenv("FORCE_DEPLOY") == "1" PROPENSITY_FALLBACK = 0.9 # 구로그 propensity 누락 시 (UCB/DQN 모두 greedy≈(1-ε)+ε/n) # ---- ① 로그 로드 ------------------------------------------------------------- async def fetch_logs(): def _q(s): q = (select(ExperienceLog.company_id, ExperienceLog.session_id, ExperienceLog.card_id, ExperienceLog.reward, ExperienceLog.done, ExperienceLog.snapshot, ExperienceLog.propensity, ExperienceLog.turn, ExperienceLog.id) .where(ExperienceLog.is_invalidated == False) # noqa: E712 .order_by(ExperienceLog.company_id, ExperienceLog.session_id, ExperienceLog.id)) return DB_SESSION_MNG.execute(s, q) err, rows = await DB_SESSION_MNG.execute_lambda(DBType.MAIN.value, DBWRType.DB_READ.value, _q) return rows # ---- ② 에피소드 재구성 -------------------------------------------------------- def build_episodes(rows, known_cards: set): """→ [{tenant, steps:[(snapshot, card, propensity)], terminal_reward}], 스킵 사유 카운트.""" by_session = defaultdict(list) for r in rows: if r[1] is not None: by_session[(r[0], str(r[1]))].append(r) episodes, skipped = [], defaultdict(int) for (company_id, _sid), items in by_session.items(): selects = [r for r in items if not r[4] and r[5]] # done=False, snapshot 有 terminals = [r for r in items if r[4] and r[3] is not None] # done=True, reward 有 if not selects or not terminals: skipped["종료행/카드턴 없음(미완결 세션)"] += 1 continue if any(str(r[2] or "").startswith("AUT|") for r in selects): skipped["완전 자율 세션(카드 재학습 대상 아님)"] += 1 continue if any(r[2] not in known_cards for r in selects): skipped["임베딩 없는 카드(파일매핑 테넌트 등)"] += 1 continue episodes.append(dict( tenant=company_id, steps=[(r[5], r[2], r[6] if r[6] else PROPENSITY_FALLBACK) for r in selects], terminal_reward=float(terminals[-1][3]), )) return episodes, skipped def tenant_feat_for(cache: dict, loader: TenantConfigLoader, company_id: str) -> np.ndarray: """테넌트 보상설정 → 성향 특징. 미온보딩/로드 실패는 _base 폴백.""" if company_id not in cache: try: cfg = loader.load(company_id) except Exception: cfg = loader.load("_base") cache[company_id] = build_tenant_features(cfg.reward) return cache[company_id] def to_transitions(episodes, feat, tenant_feats): """학습 규약(train_feature_dqn 과 동일): 중간턴 r=0, 종료턴만 terminal_reward. 다음 후보 = 전체 − 사용분.""" all_cards = list(feat.keys()) out = [] for ep in episodes: tf = tenant_feats[ep["tenant"]] used = set() n = len(ep["steps"]) for i, (snap_d, card, _p) in enumerate(ep["steps"]): sf = np.concatenate([build_state_features(NegotiationSnapshot.from_dict(snap_d)), tf]) used.add(card) if i == n - 1: out.append((sf, feat[card], ep["terminal_reward"], None, None, True)) else: s2_d = ep["steps"][i + 1][0] s2 = np.concatenate([build_state_features(NegotiationSnapshot.from_dict(s2_d)), tf]) cands = [c for c in all_cards if c not in used] or all_cards out.append((sf, feat[card], 0.0, s2, np.stack([feat[c] for c in cands]), False)) return out # ---- ④ OPE (SNIPS, 궤적 단위 IS) ---------------------------------------------- def _greedy_match(score_fn, ep, feat, tf) -> float: """궤적 IS 가중치: Π 1[greedy(sᵢ)=aᵢ]/pᵢ. 한 턴이라도 불일치면 0.""" all_cards = list(feat.keys()) w, used = 1.0, set() for snap_d, card, p in ep["steps"]: sf = np.concatenate([build_state_features(NegotiationSnapshot.from_dict(snap_d)), tf]) cands = [c for c in all_cards if c not in used] or all_cards sc = score_fn(sf, np.stack([feat[c] for c in cands])) if cands[int(np.argmax(sc))] != card: return 0.0 w /= max(p, 1e-3) used.add(card) return w def snips(score_fn, episodes, feat, tenant_feats): """SNIPS 추정치 + 유효표본크기(ESS). 매치 0건이면 (None, 0).""" ws, rs = [], [] for ep in episodes: w = _greedy_match(score_fn, ep, feat, tenant_feats[ep["tenant"]]) ws.append(w) rs.append(ep["terminal_reward"]) ws, rs = np.array(ws), np.array(rs) if ws.sum() <= 0: return None, 0.0 est = float((ws * rs).sum() / ws.sum()) ess = float(ws.sum() ** 2 / (ws ** 2).sum()) return est, ess def np_scorer_from_bundle(path): """현재 서빙 번들(npz) → score_fn (dqn_store 와 동일 forward).""" z = np.load(path, allow_pickle=False) W0, b0, W1, b1, W2, b2 = z["W0"], z["b0"], z["W1"], z["b1"], z["W2"], z["b2"] def score(sf, card_feats): x = np.concatenate([np.repeat(sf[None, :], card_feats.shape[0], axis=0), card_feats], axis=1) h = np.maximum(x @ W0.T + b0, 0.0) h = np.maximum(h @ W1.T + b1, 0.0) return (h @ W2.T + b2).squeeze(-1) return score # ---- 메인 --------------------------------------------------------------------- async def run(): numbers, feat, _ = load_cards() rows = await fetch_logs() episodes, skipped = build_episodes(rows, set(numbers)) print(f"로그 {len(rows)}행 → 에피소드 {len(episodes)}개 (스킵: {dict(skipped) or '없음'})") report = dict(rows=len(rows), episodes=len(episodes), skipped=dict(skipped), min_episodes=MIN_EPISODES, deployed=False) if len(episodes) < MIN_EPISODES and not FORCE_DEPLOY: print(f"[skip] 에피소드 {len(episodes)} < MIN_EPISODES {MIN_EPISODES} — 과적합 위험으로 재학습 안 함") report["result"] = "skipped_insufficient_data" return report loader = TenantConfigLoader() tenant_feats = {} for ep in episodes: tenant_feat_for(tenant_feats, loader, ep["tenant"]) # ③ fine-tune (시뮬 사전학습 체크포인트에서 이어서, 낮은 lr) transitions = to_transitions(episodes, feat, tenant_feats) batch = min(64, max(8, len(transitions) // 4)) policy = FeatureDQNPolicy(state_dim=STATE_FEATURE_DIM + TENANT_FEATURE_DIM, card_dim=feat[numbers[0]].shape[0], lr=LR, batch_size=batch) if os.path.exists(CKPT_PATH): policy.load(CKPT_PATH) print(f"[fine-tune] 시작점: {os.path.basename(CKPT_PATH)} lr={LR} batch={batch}") policy.buf.extend(transitions) steps = EPOCHS * max(1, len(transitions) // batch) losses = [l for _ in range(steps) if (l := policy.train_step()) is not None] print(f"[fine-tune] {steps} step loss {losses[0]:.4f} → {losses[-1]:.4f}" if losses else "[fine-tune] 스텝 없음") # ④ OPE 게이트: 후보 vs 현재 서빙 def cand_score(sf, cf): return policy.scores(sf, cf) cand_est, cand_ess = snips(cand_score, episodes, feat, tenant_feats) cur_est, cur_ess = (snips(np_scorer_from_bundle(OUT_PATH), episodes, feat, tenant_feats) if os.path.exists(OUT_PATH) else (None, 0.0)) print(f"[OPE/SNIPS] 후보 {cand_est} (ESS {cand_ess:.1f}) vs 현재 {cur_est} (ESS {cur_ess:.1f})") report.update(ope_candidate=cand_est, ope_candidate_ess=cand_ess, ope_current=cur_est, ope_current_ess=cur_ess) min_ess = max(3.0, 0.02 * len(episodes)) passed = (cand_est is not None and cand_ess >= min_ess and (cur_est is None or cand_est >= cur_est - 0.01)) if not passed and not FORCE_DEPLOY: print(f"[게이트 불통과] 배포하지 않음 (필요 ESS ≥ {min_ess:.1f}). 현재 번들 유지.") report["result"] = "gate_failed" return report # ⑤ 배포: 후보 저장 + 번들 교체 (.prev 백업) policy.save(RETRAIN_CKPT) path = export_bundle(policy.q.state_dict(), OUT_PATH) print(f"[배포] {path} (직전본 → dqn_serving.npz.prev)") print(" 적용: docker compose build agent && docker compose up -d agent") report.update(result="deployed" if passed else "force_deployed", deployed=True, ckpt=os.path.abspath(RETRAIN_CKPT)) return report def main(): report = asyncio.run(run()) with open(REPORT_PATH, "w", encoding="utf-8") as f: json.dump(report, f, ensure_ascii=False, indent=2) print(f"[리포트] {os.path.abspath(REPORT_PATH)}") if __name__ == "__main__": main()