"""AutonomyStore — 완전 자율 협상 정책 서빙 (룰 대체, numpy 전용). AUTONOMY_MODE=1 이면 가격협상 판정 룰(앵커 이하 타결 / 와일드카드 존 / 라운드 상한)과 카드 선택을 전부 이 정책의 행동 결정으로 대체한다: accept → 협상완료 (제시가 타결) walk → 협상실패 counter → "C원이면 수락" 역제안 스텝 press → 전략별 압박 멘트 스텝 행동의 유일한 유인은 보상 함수다. 남는 제한은 두 가지뿐이며 비즈니스 룰이 아니다: - 역제안 후보 격자가 [anchor−5%span, target] 안 (행동 공간 정의) - 세션 턴 상한(엔지니어링 타임아웃, ChatEngine._AUTONOMY_TURN_CAP) 번들: artifacts/autonomy_serving.npz (tools/export_autonomy_serving.py). 불가(플래그 꺼짐/번들 없음)면 None → 기존 룰 엔진 그대로 (즉시 롤백 경로). """ import os from typing import Optional import numpy as np from common.logger import LOG from negotiation.policies.autonomy_actions import ( ACTIONS, Action, extra_state, internet_gap_feat, settle_norm) from negotiation.qtable.domain.model.snapshot import NegotiationSnapshot from negotiation.qtable.domain.service.feature_builder import ( build_state_features, build_tenant_features) _HERE = os.path.dirname(os.path.abspath(__file__)) BUNDLE_PATH = os.path.join(_HERE, "..", "..", "artifacts", "autonomy_serving.npz") class AutonomyPolicy: """세션 컨텍스트 → 상태특징 → 행동(greedy). ChatEngine 에 decider 로 주입된다.""" name = "full_autonomy" def __init__(self, z, reward_cfg): self._W = (z["W0"], z["b0"], z["W1"], z["b1"], z["W2"], z["b2"]) self._state_dim = int(z["state_dim"]) self._tenant_feat = build_tenant_features(reward_cfg) @staticmethod def _acceptance(ctx: dict) -> float: base = ctx.get("item_price") or ctx.get("first_offer_price") or 0 cur = ctx.get("input_price") or 0 if base <= 0 or cur <= 0: return 0.0 return max(0.0, (base - cur) / base) def decide(self, ctx: dict) -> Action: """ChatSession.context → Action. 상태 구성은 ChatService._snapshot 과 동일 규칙.""" snap = NegotiationSnapshot( revenue_amount=ctx["revenue_amount"], distribution_code=ctx["distribution_code"], partner_count=ctx["partner_count"], acceptance_ratio=self._acceptance(ctx), input_price=ctx.get("input_price", ctx["anchor_price"]), anchor_price=ctx["anchor_price"], target_price=ctx["target_price"], round_number=ctx.get("round", 0), ) # v3 추가 특징: 직전 역제안 기억 + 마감 잔여율 + 협력사 이력 + 인터넷최저가 갭. # 소스가 없으면 전부 중립값(0.5/0) — 학습 시뮬의 '미상' 표현과 동일해야 한다. # 역제안 기억은 autonomy_last_counter(역제안만 갱신) — autonomy_last(마지막 행동)를 쓰면 # 사이에 낀 설득이 기억을 지워 단조 봉투가 뚫린다(counter→press→counter 철회 실버그). # 시뮬의 last_kind/last_q 도 역제안만 추적하므로 이쪽이 학습 분포와도 일치한다. last = ctx.get("autonomy_last_counter") or {} deadline = 0.5 end_ts, total_s = ctx.get("deadline_end_ts"), ctx.get("deadline_total_s") if end_ts and total_s: import time deadline = float(np.clip((end_ts - time.time()) / total_s, 0.0, 1.0)) hist_n = int(ctx.get("hist_n") or 0) hist_success = float(ctx["hist_success"]) if ctx.get("hist_success") is not None else 0.5 hist_settle = (settle_norm(float(ctx["hist_settle_ratio"])) if ctx.get("hist_settle_ratio") is not None else 0.5) sf = np.concatenate([build_state_features(snap), self._tenant_feat, extra_state( last.get("kind", ""), float(last.get("q", 0.0)), deadline=deadline, hist_n=min(hist_n, 5) / 5.0, hist_success=hist_success if hist_n else 0.5, hist_settle=hist_settle, internet_gap=internet_gap_feat(float(ctx.get("internet_lowest_price") or 0), float(snap.anchor_price)), )]) span = max(snap.target_price - snap.anchor_price, 1.0) pos = (snap.input_price - snap.anchor_price) / span # 행동 봉투 (학습 available_actions 와 동일해야 한다): # ① 목표가 초과 제시가는 '수락' 제외 — 매입 승인 범위(v3.1 착취 방지) # ② 직전 역제안보다 낮은 금액의 역제안 제외 — 단조 양보 원칙(제안 철회는 협상 예절 위반; # 올리는 '속도'는 정책 학습, 후퇴 '금지'만 구조로 보장) # ③ 역제시 해금 조건 — 옛 제품 의미론 복원(제품 결정 2026-07-10): 일반 카드는 설득만, # 역제시(숫자 제안)는 와일드카드처럼 마무리 수단. 최소 AUTONOMY_MIN_PRESS(기본 2)회 # 설득 이후에만 역제시 후보가 열린다. 해금 후의 타이밍·금액은 정책 학습. # ④ 마무리 국면 — 제시가가 목표가 0.5% 이내로 붙으면 압박 제외(+역제시 잠금 해제): # 푼돈 차이에서 '재검토 부탁' 반복은 상대만 지치게 한다. 클로징(역제안/최종제안)하거나 끝내거나. min_press = int(os.getenv("AUTONOMY_MIN_PRESS", "2")) near_target = snap.input_price <= snap.target_price * 1.005 counter_locked = (int(ctx.get("autonomy_press_n") or 0) < min_press) and not near_target last_counter_q = float(last["q"]) if last.get("kind") == "counter" else None if last_counter_q is not None: counter_locked = False # 이미 역제시를 시작했으면 잠그지 않는다(단조 봉투가 관리) # ⑤ 첫 역제안은 앵커가 이하(q ≤ 0)만 — 낮게 개시해 목표가까지 천천히 올라간다 # (제품 결정: 사다리를 다 쓰는 앵커링 개시. 이후 단조 봉투가 상향을 관리). # ⑥ 결렬(walk)도 해금 전 금지 — 설득 0회에 walk 를 고르면 최종제안 보장(엔진)과 결합해 # '첫 턴 목표가 통보'가 된다(v3.4 라이브 결함). 해금 전에는 설득만 가능. cands = [a for a in ACTIONS if not (a.kind == "accept" and snap.input_price > snap.target_price) and not (a.kind == "counter" and counter_locked) and not (a.kind == "walk" and counter_locked) and not (a.kind == "press" and near_target) and not (a.kind == "counter" and last_counter_q is None and a.counter_q > 1e-9) and not (a.kind == "counter" and last_counter_q is not None and a.counter_q < last_counter_q - 1e-9)] feats = [] for a in cands: cut = 0.0 if a.kind == "counter": c = snap.anchor_price + a.counter_q * span cut = max(0.0, (snap.input_price - c) / max(snap.input_price, 1.0)) feats.append(a.feat(pos, cut)) feats = np.stack(feats) W0, b0, W1, b1, W2, b2 = self._W x = np.concatenate([np.repeat(sf[None, :], feats.shape[0], axis=0), feats], axis=1) h = np.maximum(x @ W0.T + b0, 0.0) h = np.maximum(h @ W1.T + b1, 0.0) scores = (h @ W2.T + b2).squeeze(-1) return cands[int(np.argmax(scores))] @staticmethod def counter_price(ctx: dict, act: Action) -> int: span = max(ctx["target_price"] - ctx["anchor_price"], 1.0) return int(round(ctx["anchor_price"] + act.counter_q * span)) class AutonomyStore: _z = None _load_failed = False @classmethod def enabled(cls) -> bool: return os.getenv("AUTONOMY_MODE", "0").lower() in ("1", "true", "yes") @classmethod def policy_for(cls, engine) -> Optional[AutonomyPolicy]: """engine: tenancy.registry.TenantEngine. 비활성/번들 없음 → None (룰 엔진 유지).""" if not cls.enabled() or cls._load_failed: return None if cls._z is None: try: cls._z = np.load(BUNDLE_PATH, allow_pickle=False) LOG.i("[Autonomy] 완전 자율 정책 번들 로드 완료 — 협상 판정 룰 대체 모드") except Exception as ex: cls._load_failed = True LOG.e_no_callstack(f"[Autonomy] 번들 로드 실패 → 룰 엔진 유지: {ex}") return None return AutonomyPolicy(cls._z, engine.config.reward)