jwkim
94c7a6568d
실전에서 발견된 결함 전부를 자동 재생하는 배포 게이트 — 실패 시 exit 1(배포 금지). 지형 2종(실스케일 423,198 / 소액 10,000) x 협력사 시나리오 3종(완고/목표가위 고정/협조) + 엔진 전환·봉투 마스크·멘트 가드 단위검사 = 76+항목. 검사 대상은 서빙 실물(AutonomyPolicy.decide + ChatEngine._autonomy_next + ment 가드). 결정론(스크립트 상대 + greedy 정책)이라 실행마다 동일 결과. DB/LLM/도커 불필요. v3.4 교훈 반영: 한 지형 통과가 다른 지형을 보증하지 않는다 — 게이트 후 궤적 눈 비교 병행. 실행: APP_ENV=local python -m tools.test_autonomy_defects [번들경로] Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
318 lines
15 KiB
Python
318 lines
15 KiB
Python
"""결함 회귀 게이트 — 실전에서 발견된 협상 결함을 시나리오로 재생해 서빙 번들을 검증한다.
|
||
|
||
프로브(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()
|