diff --git a/agent/tools/test_autonomy_defects.py b/agent/tools/test_autonomy_defects.py new file mode 100644 index 0000000..cf51922 --- /dev/null +++ b/agent/tools/test_autonomy_defects.py @@ -0,0 +1,317 @@ +"""결함 회귀 게이트 — 실전에서 발견된 협상 결함을 시나리오로 재생해 서빙 번들을 검증한다. + +프로브(probe_serving_dqn)가 '눈으로 보는 행동 표'라면 이것은 '자동 합격/불합격'이다. +모든 검사 항목은 과거 실제 발생했던 결함이며, 하나라도 실패하면 exit 1 — 배포 금지. +재학습 번들은 반드시 이 게이트를 통과한 뒤에만 autonomy_serving.npz 로 교체한다. + +검사 대상은 서빙 실물이다: AutonomyPolicy.decide(봉투 마스크 포함) + ChatEngine._autonomy_next +(최종제안 전환) + ment_generator 가드(목표가 누설·할루시네이션). 시뮬 협력사는 스크립트라 +결정론적이고, 정책도 greedy 라 실행마다 같은 결과가 나온다. DB/LLM/도커 불필요. + +지형은 복수로 검사한다 — v3.4 가 실스케일(423,198)에선 통과하고 드라이브 지형(10,000)에서 +'첫 턴 목표가 통보'로 퇴화했던 사고: 한 지형 통과는 다른 지형을 보증하지 않는다. + +실행: agent 디렉터리에서 APP_ENV=local python -m tools.test_autonomy_defects [번들경로] +""" + +import os +import sys +from types import SimpleNamespace + +import numpy as np + +from negotiation.chat.service import ment_generator +from negotiation.chat.service.chat_engine import ChatEngine +from negotiation.policy import autonomy_store +from negotiation.policy.autonomy_store import AutonomyPolicy +from tenancy.config_loader import TenantConfigLoader + +# 검사 지형: 실제 견적(앵커율 ~1%) + 로컬 드라이브 견적(소액) — 스케일이 달라도 예절은 같아야 한다. +GEOS = { + "실스케일": dict(anchor=418_966, target=423_198, first=540_000, il=459_000), + "소액": dict(anchor=9_900, target=10_000, first=11_500, il=0), +} +MIN_PRESS = int(os.getenv("AUTONOMY_MIN_PRESS", "2")) + +_RESULTS = [] + + +def check(name: str, ok: bool, detail: str = ""): + _RESULTS.append((name, ok, detail)) + print(f" {'✔' if ok else '✘ FAIL'} {name}" + (f" — {detail}" if detail and not ok else "")) + + +# ---- 하니스: 서빙 실물 구동 (chat_service 의 ctx 관리 순서를 그대로 재현) ---------------- +def base_ctx(geo) -> dict: + return dict(revenue_amount=50_000_000, distribution_code="A", partner_count=3, + item_price=geo["first"], input_price=geo["first"], round=1, + anchor_price=geo["anchor"], target_price=geo["target"], + internet_lowest_price=geo["il"]) + + +def run_scenario(policy, supplier, geo, max_steps=30): + """정책 결정 → 스텝 전환 → 컨텍스트 부기(chat_service 순서) → 스크립트 협력사 반응 루프. + + trace 원소: (step, kind, q, 당시 제시가, autonomy_offer, 결정 시점 press_n) + """ + eng = ChatEngine.__new__(ChatEngine) # _autonomy_next 는 decider 와 ctx 만 쓴다 + ctx = base_ctx(geo) + + def decide(c): + act = policy.decide(c) + c["autonomy_pending"] = {"kind": act.kind, "q": act.counter_q, "s": act.strategy} + return act + + eng.autonomy_decider = decide + sess = SimpleNamespace(context=ctx) + trace, end = [], None + for _ in range(max_steps): + press_n_at = int(ctx.get("autonomy_press_n") or 0) + step = eng._autonomy_next(sess) + pending = ctx.pop("autonomy_pending", None) + if pending: # chat_service 부기: pending → last(+prev), 역제안 별도 보존, press 카운터 + if ctx.get("autonomy_last"): + ctx["autonomy_prev"] = ctx["autonomy_last"] + ctx["autonomy_last"] = dict(pending) + if pending["kind"] == "counter": + ctx["autonomy_last_counter"] = dict(pending) + if pending["kind"] == "press": + ctx["autonomy_press_n"] = press_n_at + 1 + trace.append((step, (pending or {}).get("kind"), (pending or {}).get("q"), + ctx["input_price"], ctx.get("autonomy_offer"), press_n_at)) + if step in ("협상완료", "협상실패"): + end = step + break + if step == "자율_최종제안": # 예→그 금액 타결 / 아니오→협상실패 (엔진 스텝 정의) + end = "협상완료" if supplier.final_yes(ctx) else "협상실패" + break + if step == "자율_역제안" and supplier.counter_yes(ctx): + ctx["input_price"] = ctx["autonomy_offer"] + end = "협상완료" + break + ctx["input_price"] = int(supplier.next_price(ctx)) + ctx["round"] = ctx.get("round", 1) + 1 + return trace, end, ctx + + +def fmt(trace): + out = [] + for step, kind, q, price, offer, _ in trace: + s = f"{price:,}→{step}" + if kind == "counter": + s += f"({offer:,})" + out.append(s) + return " ".join(out) + + +# ---- 스크립트 협력사 (결정론, 지형 비율로 정의) ----------------------------------------- +class Stubborn: + """조금씩 내리지만 하한이 목표가 위(×1.028) — 성사 불가능. 역제안·최종 전부 거절. + + 기대 궤적: 설득 ≥2회 → 앵커 이하 개시 → 단조 상향 사다리 → 최종제안(목표가) → 결렬.""" + def __init__(self, geo): + self.floor = int(geo["target"] * 1.028) + + def next_price(self, ctx): + return max(self.floor, int(ctx["input_price"] * 0.96)) + + def counter_yes(self, ctx): + return False + + def final_yes(self, ctx): + return False + + +class HoverNearTarget: + """목표가 +0.19% 고정 — 마무리 국면. 압박이 나오면 안 되는 구간.""" + def __init__(self, geo): + self.price = int(geo["target"] * 1.0019) + + def next_price(self, ctx): + return self.price + + def counter_yes(self, ctx): + return False + + def final_yes(self, ctx): + return False + + +class Dealable: + """4% 씩 내려와 목표가 바로 아래까지 협조 — 성사 가능 케이스.""" + def __init__(self, geo): + self.floor = int(geo["target"] * 0.9995) + self.accept_from = geo["anchor"] + 0.4 * (geo["target"] - geo["anchor"]) + + def next_price(self, ctx): + return max(self.floor, int(ctx["input_price"] * 0.96)) + + def counter_yes(self, ctx): + return ctx["autonomy_offer"] >= self.accept_from # 목표가 부근 제안은 수락 + + def final_yes(self, ctx): + return True + + +# ---- 시나리오 검사 (각 항목 = 과거 실제 결함) ------------------------------------------- +def assert_defects(tag, trace, end, geo): + anchor, target = geo["anchor"], geo["target"] + near = target * 1.005 + # '역제안' 검사는 일반 역제안 스텝만 센다 — 같은 금액 재시도가 자율_최종제안으로 전환된 것은 + # 반복이 아니라 설계된 최종 통보(제품 결정: 같은 금액 재호출 = 탄약 소진 → 마지막으로 묻고 종료). + counters = [(i, t) for i, t in enumerate(trace) if t[0] == "자율_역제안"] + presses = [t for t in trace if t[1] == "press"] + + if trace and trace[0][3] > near: + check(f"[{tag}] 개시 턴은 설득 (결함: v3.4 첫턴 walk→목표가 통보)", + trace[0][1] == "press", f"첫 결정이 {trace[0][1]}") + if counters: + i0, first = counters[0] + check(f"[{tag}] 첫 역제안은 앵커 이하 (결함: 사다리 꼭대기 개시)", + first[4] <= anchor, f"첫 역제안 {first[4]:,} > 앵커 {anchor:,}") + pressed_before = sum(1 for t in trace[:i0] if t[1] == "press") + if first[3] > near: # 마무리 국면은 해금 예외 + check(f"[{tag}] 역제시 해금 전 설득 ≥{MIN_PRESS}회 (결함: 첫턴 역제시)", + pressed_before >= MIN_PRESS, f"설득 {pressed_before}회 만에 역제안") + offers = [t[4] for _, t in counters] + check(f"[{tag}] 역제안 단조 상향 (결함: 제안 철회 423,198→420,024)", + all(b >= a for a, b in zip(offers, offers[1:])), f"철회 발생: {offers}") + check(f"[{tag}] 역제안 ≤ 목표가", all(o <= target for o in offers), f"{offers}") + check(f"[{tag}] 같은 금액 역제안 반복 없음 (결함: 421,082 반복)", + all(b != a for a, b in zip(offers, offers[1:])), f"{offers}") + check(f"[{tag}] 마무리 국면(≤목표가×1.005) 압박 없음 (결함: 802원 푼돈 흥정)", + all(t[3] > near for t in presses), "목표가 코앞에서 압박") + check(f"[{tag}] 목표가 초과 제시가 수락 없음 (결함: 목표가+14% 매입)", + not any(t[1] == "accept" and t[3] > target for t in trace), "") + finals = [t for t in trace if t[0] == "자율_최종제안"] + for f in finals: + check(f"[{tag}] 최종제안 금액 = 목표가 (결함: 직전 금액 재사용 60,548)", + f[4] == target, f"최종제안 {f[4]:,} ≠ 목표가 {target:,}") + # 결렬 의사(walk)로 끝났다면 반드시 최종제안을 거쳤어야 한다 (턴캡 종료는 예외) + walked_direct = any(t[1] == "walk" and t[0] == "협상실패" for t in trace) + capped = trace and trace[-1][0] == "협상실패" and trace[-1][1] is None + check(f"[{tag}] 결렬 전 최종제안 1회 보장 (결함: 최종 의사 확인 없이 종료)", + not walked_direct or capped or bool(finals), "walk 즉시 결렬") + check(f"[{tag}] 종료 보장 (무한 세션 없음)", end is not None, "max_steps 내 미종료") + + +# ---- 엔진 단위 검사 (정책 무관 — 전환 로직 자체) ---------------------------------------- +def engine_unit_tests(): + print("\n[엔진 전환 로직 단위 검사]") + geo = GEOS["실스케일"] + target = geo["target"] + + def force(kind, q=0.0, s=3): + eng = ChatEngine.__new__(ChatEngine) + eng.autonomy_decider = lambda c: SimpleNamespace(kind=kind, counter_q=q, strategy=s) + return eng + + # walk → 최종제안(목표가) → 재차 walk → 협상실패 + ctx = base_ctx(geo) + eng = force("walk") + sess = SimpleNamespace(context=ctx) + step1 = eng._autonomy_next(sess) + check("walk 1회차 → 자율_최종제안 전환", step1 == "자율_최종제안", f"got {step1}") + check("walk 전환 최종제안 금액 = 목표가", ctx.get("autonomy_offer") == target, + f"{ctx.get('autonomy_offer')}") + step2 = eng._autonomy_next(sess) + check("walk 2회차(최종 거절 후) → 협상실패", step2 == "협상실패", f"got {step2}") + + # 같은 q 역제안 반복 → 최종제안(목표가) 전환 + ctx = base_ctx(geo) + ctx["autonomy_last"] = ctx["autonomy_last_counter"] = {"kind": "counter", "q": 0.5, "s": 3} + sess = SimpleNamespace(context=ctx) + step = force("counter", q=0.5)._autonomy_next(sess) + check("같은 금액 재역제안 → 자율_최종제안 전환", step == "자율_최종제안", f"got {step}") + check("탄약소진 최종제안 금액 = 목표가", ctx.get("autonomy_offer") == target, + f"{ctx.get('autonomy_offer')}") + + # 턴 상한 — 캡 종료도 최종제안 보장을 우회하지 않는다 + ctx = base_ctx(geo) + ctx["round"] = 13 + sess = SimpleNamespace(context=ctx) + step = force("press")._autonomy_next(sess) + check("턴 상한 초과(최종 미실시) → 자율_최종제안", step == "자율_최종제안", f"got {step}") + check("턴캡 최종제안 금액 = 목표가", ctx.get("autonomy_offer") == target, + f"{ctx.get('autonomy_offer')}") + step = force("press")._autonomy_next(sess) + check("턴 상한 초과(최종 거절 후) → 협상실패", step == "협상실패", f"got {step}") + + +# ---- 봉투 마스크 단위 검사 (모델 무관 — 후보 필터 자체) ---------------------------------- +def envelope_unit_tests(policy): + print("\n[봉투 마스크 단위 검사]") + geo = GEOS["소액"] + ctx = base_ctx(geo) # 설득 0회, 제시가 목표가 위 → 설득만 가능해야 한다 + act = policy.decide(ctx) + check("설득 0회 상태의 결정은 press 만 가능 (walk·counter·accept 잠금)", + act.kind == "press", f"got {act.kind}") + + +# ---- 멘트 가드 검사 (목표가 누설·할루시네이션 — LLM 호출 없음) --------------------------- +def ment_guard_tests(): + print("\n[멘트 가드 검사]") + geo = GEOS["실스케일"] + target, anchor, il = geo["target"], geo["anchor"], geo["il"] + ctx = base_ctx(geo) + + prompt = ment_generator._prompt_for("자율_압박_3", ctx) + check("압박 프롬프트에 목표가 숫자 없음 (결함: 목표가 노출 멘트)", + str(target) not in prompt.replace(",", ""), "프롬프트가 목표가를 담고 있음") + + leak = f"저희 내부 산정 기준은 {target:,}원입니다. 이 가격에 맞춰 재검토 부탁드립니다." + check("목표가 포함 압박 멘트 → 폐기", not ment_generator._guard("자율_압박_3", ctx, leak), "") + + invented = "시장 상황을 고려해 400,000원 수준으로 재검토 부탁드립니다." + check("지어낸 금액 멘트 → 폐기 (할루시네이션)", + not ment_generator._guard("자율_압박_3", ctx, invented), "") + + ctx2 = dict(ctx, autonomy_offer=anchor) + ok_ment = f"내부 검토 결과 {anchor:,}원이면 즉시 진행이 가능합니다. 수락해 주시겠습니까?" + check("정상 역제안 멘트(제안가 포함) → 통과", + ment_generator._guard("자율_역제안", ctx2, ok_ment), "") + no_offer = "말씀하신 조건을 검토했고 조정이 필요합니다. 수락해 주시겠습니까?" + check("제안가 없는 역제안 멘트 → 폐기", + not ment_generator._guard("자율_역제안", ctx2, no_offer), "") + + ev = f"동일 품목 인터넷 최저가가 {il:,}원으로 확인됩니다. 재검토 부탁드립니다." + check("최저가 인용: 근거 있음(수집됨+제시가>최저가) → 허용", + ment_generator._guard("자율_압박_1", ctx, ev), "") + ctx3 = dict(ctx, internet_lowest_price=0) + ev0 = "동일 품목 인터넷 최저가 대비 높은 수준입니다. 재검토 부탁드립니다." + check("최저가 인용: 미수집 품목 → 폐기 (지어낸 시장 주장)", + not ment_generator._guard("자율_압박_1", ctx3, ev0), "") + + +def main(): + bundle = sys.argv[1] if len(sys.argv) > 1 else autonomy_store.BUNDLE_PATH + z = np.load(bundle, allow_pickle=False) + policy = AutonomyPolicy(z, TenantConfigLoader().load("ktcommerce").reward) + print(f"번들: {os.path.abspath(bundle)} (state_dim={int(z['state_dim'])})") + + for geo_name, geo in GEOS.items(): + print(f"\n{'─' * 60}\n지형 [{geo_name}] 앵커 {geo['anchor']:,} / 목표 {geo['target']:,} " + f"/ 첫 제시가 {geo['first']:,}") + for tag, sup_cls in (("완고", Stubborn), ("목표가위 고정", HoverNearTarget), + ("협조", Dealable)): + trace, end, _ = run_scenario(policy, sup_cls(geo), geo) + full_tag = f"{geo_name}·{tag}" + print(f"\n[{full_tag}] {fmt(trace)} ⇒ {end}") + assert_defects(full_tag, trace, end, geo) + + engine_unit_tests() + envelope_unit_tests(policy) + ment_guard_tests() + + fails = [(n, d) for n, ok, d in _RESULTS if not ok] + print(f"\n{'=' * 60}\n결과: {len(_RESULTS) - len(fails)}/{len(_RESULTS)} 통과") + if fails: + print("실패 항목 — 이 번들은 배포 금지:") + for n, d in fails: + print(f" ✘ {n} {d}") + sys.exit(1) + print("전 항목 통과 — 배포 가능.") + + +if __name__ == "__main__": + main()