from __future__ import annotations """ 결합(bundle-v3) 대규모 랜덤 스트레스 시뮬레이션 — 의도 변화 적절성·이상 케이스 탐지. - 설문: 무작위 N_SURVEY건 (전수 1,870만은 불가 → 대규모 샘플) - 행동: N_SESS 세션 × 세션당 최대 MAX_ACT 클릭 (클릭 기반 윈도우 가정) - 진행바는 /tmp/cs_sim_progress.txt 에 in-place 기록 (tail -f 로 관람) - 요약은 stdout, 상세는 /tmp/cs_sim_report.json 탐지 이상: E1 범위/NaN (score∉[0,1]) E2 범용 intent(AI추천류) top-5 누출 E3 상위 동점 (top-1과 동률이 top-5 내 3개 이상) E4 행동 무반응 (클릭 entity의 신호 intent가 top-10에도 못 듦) E5 stuck (top-1이 ≥STUCK_N 연속 고정) + 분포 건전성(top-1 다양성/HHI), entity별 반응률 """ import json, random, sys, time from collections import Counter, defaultdict from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent)) from core.engines import config # noqa: E402 from core.extractor import get_extractor # noqa: E402 from core.inference import infer_batch, infer_with_behavior # noqa: E402 SID = "bundle-v3" N_SURVEY = 30_000 N_SESS = 8_000 MAX_ACT = 30 STUCK_N = 15 GENERIC = set() # bundle: 범용 intent 없음 (E2 N/A) PROG = "/tmp/bundle_sim_progress.txt" REPORT = "/tmp/bundle_sim_report.json" tax = config.get_taxonomy(SID)["intents"] NM = {i["id"]: i["name"] for i in tax} SURVEY = config.get_survey(SID)["questions"] OPTS = {q["id"]: [o["code"] for o in q["options"]] for q in SURVEY} SIG = config.get_behavior_signals(SID) # 행동 트리 _bc = config.get_behaviors(SID) STEP1 = [(b["event_type"], b["entity"], b["id"]) for b in _bc["step1"]["behaviors"]] STEP2 = {} for parent, items in _bc["step2"]["by_parent"].items(): STEP2[parent] = [(b["event_type"], b["entity"], b["id"]) for b in items] def _bar(frac, label): n = int(frac * 30) with open(PROG, "w") as f: f.write(f"[{'#'*n}{'.'*(30-n)}] {frac*100:5.1f}% {label}") def _rand_survey(rng): return {q: rng.choice(OPTS[q]) for q in OPTS} def _top(scores, k): return sorted(scores, key=lambda s: s.final_score, reverse=True)[:k] def _check_common(scores, anom, tag): # E1 범위/NaN for s in scores: v = s.final_score if v != v or v < -1e-9 or v > 1.0001: anom["E1"].append((tag, s.intent_id, v)); break top = _top(scores, 5) top_ids = [s.intent_id for s in top] # E2 범용 누출 g = GENERIC & set(top_ids) if g: anom["E2"].append((tag, list(g))) # E3 상위 동점 tv = round(top[0].final_score, 2) tie = sum(1 for s in top if round(s.final_score, 2) == tv) if tie >= 3: anom["E3"].append((tag, tv, tie, top_ids[:tie])) return top_ids def run(): ext = get_extractor() rng = random.Random(20240601) t0 = time.time() anom = defaultdict(list) # 설문 sweep top1_survey = Counter() for i in range(N_SURVEY): a = _rand_survey(rng) try: _, sc = infer_batch(a, SID) except Exception as e: anom["E1"].append(("survey", "EXCEPTION", str(e)[:80])); continue top_ids = _check_common(sc, anom, "survey") top1_survey[top_ids[0]] += 1 if i % 500 == 0: _bar(i / (N_SURVEY + N_SESS), f"설문 {i:,}/{N_SURVEY:,}") # 행동 sweep ent_rise = defaultdict(lambda: [0, 0]) # entity → [신호상승, 시도] stuck_runs = [] top1_beh = Counter() for j in range(N_SESS): a = _rand_survey(rng) sess = f"__st_{j}" ext.reset(sess) cur_parent = None last_top1 = None; run_len = 0; max_run = 0 n_act = rng.randint(5, MAX_ACT) for _ in range(n_act): # 트리 네비게이션 r = rng.random() if cur_parent is None or r < 0.35: et, en, bid = rng.choice(STEP1); cur_parent = bid elif r < 0.5 and cur_parent: cur_parent = None; ext.add_event(sess, "navigate_back", "back_to_step1"); continue else: items = STEP2.get(cur_parent) or STEP1 et, en, bid = rng.choice(items) ext.add_event(sess, et, en) try: _, sc = infer_with_behavior(a, sess, SID) except Exception as e: anom["E1"].append((f"beh_{j}", "EXCEPTION", str(e)[:80])); break top_ids = _check_common(sc, anom, f"beh_{j}") # E4 반응성 (신호 entity가 top-10에) targets = SIG.get(en, []) if targets: ent_rise[en][1] += 1 if set(targets) & set([s.intent_id for s in _top(sc, 10)]): ent_rise[en][0] += 1 # E5 stuck if top_ids[0] == last_top1: run_len += 1; max_run = max(max_run, run_len) else: last_top1 = top_ids[0]; run_len = 1 top1_beh[top_ids[0]] += 1 if max_run >= STUCK_N: stuck_runs.append((f"beh_{j}", last_top1, max_run)) ext.reset(sess) if j % 150 == 0: _bar((N_SURVEY + j) / (N_SURVEY + N_SESS), f"행동 {j:,}/{N_SESS:,}") _bar(1.0, "완료") dt = time.time() - t0 # ── 집계/요약 ── n_surv_top1 = sum(top1_survey.values()) hhi = sum((c / n_surv_top1) ** 2 for c in top1_survey.values()) if n_surv_top1 else 0 low_ent = {e: f"{v[0]}/{v[1]}={v[0]/v[1]*100:.0f}%" for e, v in ent_rise.items() if v[1] and v[0]/v[1] < 0.9} summary = { "elapsed_sec": round(dt, 1), "n_survey": N_SURVEY, "n_sess": N_SESS, "max_act": MAX_ACT, "E1_range_nan": len(anom["E1"]), "E2_generic_leak": len(anom["E2"]), "E3_top_tie(>=3)": len(anom["E3"]), "E5_stuck(>=%d)" % STUCK_N: len(stuck_runs), "survey_top1_distinct": len(top1_survey), "survey_top1_HHI": round(hhi, 4), "survey_top1_top8": [(NM[i][:14], f"{c/n_surv_top1*100:.1f}%") for i, c in top1_survey.most_common(8)], "entity_low_responsiveness(<90%)": low_ent, "stuck_samples": stuck_runs[:10], "E1_samples": anom["E1"][:5], "E2_samples": anom["E2"][:5], "E3_samples": anom["E3"][:5], } json.dump({"summary": summary, "anom_E1": anom["E1"][:200], "anom_E2": anom["E2"][:200], "anom_E3": anom["E3"][:200], "stuck": stuck_runs[:200]}, open(REPORT, "w"), ensure_ascii=False, indent=1) print("=" * 64) print(f"결합 스트레스 시뮬 완료 — {dt/60:.1f}분 (설문 {N_SURVEY:,} + 행동 {N_SESS:,}×≤{MAX_ACT})") print("=" * 64) print(f" [E1] 범위/NaN/예외 : {len(anom['E1'])} 건") print(f" [E2] 범용 intent 누출 : {len(anom['E2'])} 건") print(f" [E3] 상위 동점(≥3) : {len(anom['E3'])} 건") print(f" [E5] stuck(top1 ≥{STUCK_N}연속): {len(stuck_runs)} 세션") print(f" 설문 top-1 다양성: {len(top1_survey)}종, HHI={hhi:.3f}") print(f" 설문 top-1 최빈: " + ", ".join(f"{NM[i][:10]} {c/n_surv_top1*100:.0f}%" for i, c in top1_survey.most_common(5))) print(f" 행동 반응률<90% entity: {low_ent if low_ent else '없음 ✓'}") if stuck_runs: print(" stuck 샘플:", [(s[1], s[2]) for s in stuck_runs[:5]]) print(f"\n상세 리포트: {REPORT}") if __name__ == "__main__": run()