""" src/evaluation/eval_router.py Router accuracy evaluation — intent classification Precision / Recall / F1 Test set: ACTION_ONLY 30 + QA_ONLY 30 + BOTH 30 = 90 total Target : ≥ 99% accuracy across all three classes Usage (from project root): python -m src.evaluation.eval_router python -m src.evaluation.eval_router --test-cases data/eval/router_test_cases.json python -m src.evaluation.eval_router --out reports/router_eval.json """ import argparse import json import logging import random import sys import time from pathlib import Path from typing import Optional import sys # Windows 환경에서 특수문자 출력(✓, ✗, 이모지 등) 시 cp949 인코딩 에러 방지 if sys.stdout.encoding.lower() != 'utf-8' and hasattr(sys.stdout, 'reconfigure'): sys.stdout.reconfigure(encoding='utf-8') from sklearn.metrics import classification_report, accuracy_score from src.logging_config import setup_logging from src.graph.router import router_node setup_logging() logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # 전처리 플래그(mentions_action / mentions_question) → intent 환산 # --------------------------------------------------------------------------- def _flags_to_intent(email_ctx: dict | None) -> Optional[str]: """전처리 플래그를 router intent 라벨로 환산. 전처리 실패/둘 다 False면 None. (action=T, question=T) → BOTH (action=T, question=F) → ACTION_ONLY (action=F, question=T) → QA_ONLY (action=F, question=F) → None (전처리가 둘 다 못 잡음 → 환산 불가) """ if not email_ctx or not email_ctx.get("preprocess_ok"): return None a = bool(email_ctx.get("mentions_action")) q = bool(email_ctx.get("mentions_question")) if a and q: return "BOTH" if a: return "ACTION_ONLY" if q: return "QA_ONLY" return None # --------------------------------------------------------------------------- # Default test-cases path # --------------------------------------------------------------------------- _DEFAULT_TEST_CASES = ( Path(__file__).resolve().parent.parent.parent / "data" / "eval" / "router_test_cases_gen.json" ) # --------------------------------------------------------------------------- # Core evaluation function # --------------------------------------------------------------------------- def eval_router( test_cases: list[dict], *, verbose: bool = True, delay: float = 2.0, # 요청 사이 대기 시간(초) - rate limit 방지용 ) -> dict: """ Run every test case through router_node and compute classification metrics. Parameters ---------- test_cases : list[dict] Each element must have: "user_input" : str — email text fed to the router "label" : str — ground-truth intent ("ACTION_ONLY" | "QA_ONLY" | "BOTH") verbose : bool If True, print per-sample progress and the final report to stdout. delay : float Seconds to sleep between each API call. Increase if hitting 429 rate limits. Returns ------- dict with keys: "accuracy" : float "report" : dict (sklearn classification_report output_dict=True) "errors" : list (samples where router returned None) "latency_s" : dict {"mean": float, "total": float} """ labels_true: list[str] = [] labels_pred: list[str] = [] errors: list[dict] = [] mismatches: list[dict] = [] latencies: list[float] = [] # 전처리 플래그 ↔ router 결과 일치도 추적 flag_total = 0 # flag_intent 환산 가능한 케이스 수 (전처리 OK & 플래그로 intent 결정됨) flag_agree = 0 # flag_intent == router 예측 일치 수 flag_correct = 0 # flag_intent == 정답 라벨 (전처리 플래그 자체 정확도) flag_disagreements: list[dict] = [] total = len(test_cases) logger.info("Starting router evaluation on %d samples …", total) # 운영 경로와 동일하게 진입 전 이메일 전처리 — 라우터가 preprocess hint도 함께 보도록 한다. from src.preprocess import preprocess_email for i, case in enumerate(test_cases, start=1): user_input: str = case["user_input"] true_label: str = case["label"] email_ctx = preprocess_email(user_input).model_dump() # Build a minimal AgentState (only fields router_node reads are required) state = { "user_input": user_input, "email_context": email_ctx, "error_messages": [], } t0 = time.perf_counter() result = router_node(state) elapsed = time.perf_counter() - t0 latencies.append(elapsed) predicted = result.get("intent") if predicted is None: logger.warning("[%d/%d] router returned None for: %.80s", i, total, user_input) errors.append({"index": i, "input": user_input, "true_label": true_label}) # Treat as wrong prediction — use a sentinel so metrics reflect the failure predicted = "__ERROR__" labels_true.append(true_label) labels_pred.append(predicted) # ── 전처리 플래그 ↔ router 결과 일치 검사 ────────────────────────────── flag_intent = _flags_to_intent(email_ctx) if flag_intent is not None: flag_total += 1 if flag_intent == predicted: flag_agree += 1 else: flag_disagreements.append({ "index": i, "id": case.get("id", f"#{i}"), "true_label": true_label, "router_pred": predicted, "flag_intent": flag_intent, "mentions_action": bool(email_ctx.get("mentions_action")), "mentions_question": bool(email_ctx.get("mentions_question")), }) if flag_intent == true_label: flag_correct += 1 status = "✓" if predicted == true_label else "✗" if predicted != true_label and predicted != "__ERROR__": mismatches.append({ "index": i, "id": case.get("id", f"#{i}"), "input": user_input, # 잘라내지 않고 전체 텍스트 저장 "true_label": true_label, "predicted": predicted, }) if verbose: print( f"[{i:>3}/{total}] {status} true={true_label:<12} pred={predicted:<12}" f" ({elapsed:.2f}s) {user_input[:60]!r}" ) # 다음 요청 전 대기 (마지막 샘플은 불필요) if i < total and delay > 0: time.sleep(delay) # Filter out __ERROR__ entries for sklearn (it can't handle unknown labels cleanly) clean_true = [t for t, p in zip(labels_true, labels_pred) if p != "__ERROR__"] clean_pred = [p for t, p in zip(labels_true, labels_pred) if p != "__ERROR__"] target_names = ["ACTION_ONLY", "QA_ONLY", "BOTH"] report_dict: dict = classification_report( clean_true, clean_pred, labels=target_names, target_names=target_names, output_dict=True, zero_division=0, ) accuracy: float = accuracy_score(clean_true, clean_pred) mean_latency = sum(latencies) / len(latencies) if latencies else 0.0 total_latency = sum(latencies) # ── Print summary ──────────────────────────────────────────────────────── if verbose: print("\n" + "=" * 70) print(classification_report( clean_true, clean_pred, labels=target_names, target_names=target_names, zero_division=0, )) print(f" Overall accuracy : {accuracy * 100:.2f}% (target ≥ 99%)") print(f" Errors (None) : {len(errors)}") print(f" Latency (mean) : {mean_latency:.2f}s") print(f" Latency (total) : {total_latency:.1f}s") print("-" * 70) # 전처리 플래그(mentions_action/question) ↔ router 결과 정합성 agree_pct = (flag_agree / flag_total * 100) if flag_total else 0.0 fcorr_pct = (flag_correct / flag_total * 100) if flag_total else 0.0 print(f" Preprocess-flag ↔ router agreement : {flag_agree}/{flag_total} = {agree_pct:.1f}%") print(f" Preprocess-flag accuracy (vs label): {flag_correct}/{flag_total} = {fcorr_pct:.1f}%") print(f" Flag-derivable cases : {flag_total}/{total} (rest: preprocess failed or both flags False)") if flag_disagreements: print(f" Flag↔router disagreements : {len(flag_disagreements)}") for d in flag_disagreements[:10]: print(f" - [{d['id']}] router={d['router_pred']:<11} flag={str(d['flag_intent']):<11}" f" (action={d['mentions_action']}, question={d['mentions_question']}, true={d['true_label']})") print("=" * 70) if accuracy >= 0.99: print("🎉 PASS — accuracy target met!") else: gap = (0.99 - accuracy) * total print(f"⚠️ FAIL — need {gap:.1f} more correct predictions to reach 99%.") return { "accuracy": accuracy, "report": report_dict, "errors": errors, "mismatches": mismatches, "latency_s": {"mean": mean_latency, "total": total_latency}, "preprocess_flag_consistency": { "flag_total": flag_total, "agree_with_router": flag_agree, "agree_rate": (flag_agree / flag_total) if flag_total else None, "flag_accuracy_vs_label": (flag_correct / flag_total) if flag_total else None, "disagreements": flag_disagreements, }, } # --------------------------------------------------------------------------- # CLI entry point # --------------------------------------------------------------------------- def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Evaluate the router node on a labelled test set." ) parser.add_argument( "--test-cases", type=Path, default=_DEFAULT_TEST_CASES, help=f"Path to JSON test cases file (default: {_DEFAULT_TEST_CASES})", ) parser.add_argument( "--out", type=Path, default=None, help="Optional path to save the evaluation report as JSON.", ) parser.add_argument( "--quiet", action="store_true", help="Suppress per-sample progress output.", ) parser.add_argument( "--delay", type=float, default=2.0, help="Seconds to wait between API calls to avoid 429 rate limits (default: 2.0).", ) return parser.parse_args() if __name__ == "__main__": args = _parse_args() # ── Load test cases ────────────────────────────────────────────────────── test_cases_path: Path = args.test_cases if not test_cases_path.exists(): logger.error( "Test cases file not found: %s\n" " Generate it with: python -m src.data.generate_router_dataset", test_cases_path, ) sys.exit(1) with open(test_cases_path, encoding="utf-8") as f: test_cases: list[dict] = json.load(f) # Shuffle the test cases to ensure mixed intent evaluation during the run # Using a fixed seed guarantees reproducibility of the shuffle order random.seed(42) random.shuffle(test_cases) logger.info("Loaded and shuffled %d test cases from %s", len(test_cases), test_cases_path) # ── Run evaluation ─────────────────────────────────────────────────────── results = eval_router(test_cases, verbose=not args.quiet, delay=args.delay) # ── Optionally save report ─────────────────────────────────────────────── if args.out: out_path: Path = args.out out_path.parent.mkdir(parents=True, exist_ok=True) with open(out_path, "w", encoding="utf-8") as f: json.dump(results, f, indent=2, ensure_ascii=False) logger.info("Report saved to %s", out_path) # ── Exit code reflects pass/fail ───────────────────────────────────────── sys.exit(0 if results["accuracy"] >= 0.99 else 1)