Pandago/graphstrike-model-training
0
1"""2Round 2 shared evaluation runner.3 4Deterministic loop + two LLM decision points per suspicious account.5 6Framework logic (deterministic, no LLM):7 - GET_POLICY as step 08 - Loop over visible accounts sorted by risk_score desc9 - INSPECT if not yet inspected10 - INVESTIGATE_NETWORK on very high-risk suspects (risk >= 0.80, once)11 - SUBMIT at end12 13Per-account LLM decision points:14 - DP1 (tool selection): "reverse_image_search" | "analyze_bio" | "check_ip" | "done"15 - DP2 (flag decision): "flag" | "skip"16 17Entry point:18 run_evaluation(model_name, call_llm, platform, base_url, tasks, seeds, out_path)19"""20 21from __future__ import annotations22 23import json24import os25import sys26import time27from dataclasses import dataclass, asdict, field28from pathlib import Path29from typing import Callable, Dict, List, Optional30 31# Make parent (fake_gang_env/) importable so the local client + models resolve32# AHEAD of any same-named modules in site-packages. Always insert at index 033# (and drop any pre-cached aliases) — a no-op `if not in sys.path` check is34# unsafe because the path may already be present at lower priority.35_PARENT = Path(__file__).resolve().parent.parent36sys.path.insert(0, str(_PARENT))37for _stale in ("models", "client"):38 sys.modules.pop(_stale, None)39 40from client import FakeGangEnvClient, StepResult # noqa: E40241from models import ActionType, FakeGangAction # noqa: E40242 43 44# ---------------------------------------------------------------------------45# Prompts for the two decision points46# ---------------------------------------------------------------------------47 48DP1_PROMPT = """You are evaluating a suspicious account. Pick the next tool to call, or 'done' if evidence is sufficient.49 50PLATFORM: {platform} | primary signal: {primary_signal} | threshold θ*: {threshold:.3f}51ACCOUNT: {account_id} | risk_score: {risk:.3f} | hub_legitimacy: {hub:.2f}52 53Revealed signals (None = not yet revealed):54 photo_reuse_score: {photo}55 bio_template_score: {bio}56 ip_cluster_signal: {ip}57 shared_ip_count: {shared_ip}58 59Budget: steps_remaining={steps_left} | tool costs: reverse_image_search=1, analyze_bio=1, check_ip=260 61Reply with EXACTLY one token (no punctuation, no explanation):62 reverse_image_search (reveal photo_reuse_score)63 analyze_bio (reveal bio_template_score)64 check_ip (reveal ip_cluster_signal, costs 2 steps)65 done (evidence is sufficient, move to flag decision)66"""67 68DP2_PROMPT = """You must decide whether to FLAG this account as a coordinated fake.69 70PLATFORM: {platform} | threshold θ*: {threshold:.3f} | fp_penalty: {fp_weight}71ACCOUNT: {account_id} | risk_score: {risk:.3f} | hub_legitimacy: {hub:.2f}72 73Revealed signals:74 photo_reuse_score: {photo}75 bio_template_score: {bio}76 ip_cluster_signal: {ip}77 shared_ip_count: {shared_ip}78 79Running totals: flagged={n_flagged}/10 | steps_remaining={steps_left}80 81Guidance:82 - Flag ONLY if risk >= threshold and hub_legitimacy < 0.70 (celebrities are hubs).83 - High photo_reuse + bio_template or shared_ip_count >= 5 is strong evidence.84 - Skip on ambiguous cases when fp_penalty is high.85 86Reply with EXACTLY one token:87 flag (mark this account as a coordinated fake)88 skip (leave it alone, move on)89"""90 91 92# ---------------------------------------------------------------------------93# Episode log record94# ---------------------------------------------------------------------------95 96@dataclass97class EpisodeLog:98 model: str99 platform: str100 task: str101 seed: int102 episode_id: str = ""103 threshold: float = 0.0104 primary_signal: str = ""105 steps_taken: int = 0106 inspected: int = 0107 tool_calls: Dict[str, int] = field(default_factory=lambda: {108 "reverse_image_search": 0, "analyze_bio": 0, "check_ip": 0,109 "get_policy": 0, "investigate_network": 0,110 })111 flagged: int = 0112 dp1_calls: int = 0113 dp2_calls: int = 0114 dp1_invalid: int = 0115 dp2_invalid: int = 0116 reward: Optional[float] = None117 grader_score: Optional[float] = None118 final_message: str = ""119 wall_seconds: float = 0.0120 121 122# ---------------------------------------------------------------------------123# Helpers124# ---------------------------------------------------------------------------125 126def _find_account(obs, account_id: str):127 for a in obs.visible_accounts:128 if a.account_id == account_id:129 return a130 return None131 132 133def _render_signal(val) -> str:134 if val is None:135 return "None"136 if isinstance(val, float):137 return f"{val:.3f}" if val > 0 else "0.000"138 return str(val)139 140 141def _parse_dp1(text: str) -> Optional[str]:142 t = text.strip().lower().split("\n")[0].strip().strip("'\"`.,")143 valid = {"reverse_image_search", "analyze_bio", "check_ip", "done"}144 for tok in valid:145 if tok in t:146 return tok147 return None148 149 150def _parse_dp2(text: str) -> Optional[str]:151 t = text.strip().lower().split("\n")[0].strip().strip("'\"`.,")152 if "flag" in t and "unflag" not in t:153 return "flag"154 if "skip" in t or "no" == t or "keep" in t:155 return "skip"156 return None157 158 159def _seeds_for_platform(seeds: List[int], platform: Optional[str]) -> List[int]:160 """The env assigns platform by seed%2 (even=Instagram, odd=Snapchat).161 Offset seeds so the requested platform is actually used."""162 if platform is None:163 return seeds164 p = platform.lower()165 if p == "instagram":166 return [s if s % 2 == 0 else s + 1 for s in seeds]167 if p == "snapchat":168 return [s if s % 2 == 1 else s + 1 for s in seeds]169 return seeds # unknown platform → env will fall back to its own mapping170 171 172def _policy_from_message(msg: str) -> Dict[str, float]:173 """Parse the message returned by GET_POLICY into a dict.174 Message format: 'Policy compiled: Platform: X | Threshold: 0.081 | Primary Signal: photo_reuse | FP Penalty: 0.5x | ...'175 """176 out = {"threshold": 0.0, "primary_signal": "", "platform": "", "fp_weight": "?"}177 try:178 body = msg.split("Policy compiled:", 1)[-1]179 for part in body.split("|"):180 k, _, v = part.strip().partition(":")181 k = k.strip().lower()182 v = v.strip()183 if k == "platform":184 out["platform"] = v185 elif k == "threshold":186 out["threshold"] = float(v)187 elif k == "primary signal":188 out["primary_signal"] = v189 elif k == "fp penalty":190 out["fp_weight"] = v191 except Exception:192 pass193 return out194 195 196# ---------------------------------------------------------------------------197# Per-account loop (DP1 + DP2)198# ---------------------------------------------------------------------------199 200def _gather_and_flag(201 client: FakeGangEnvClient,202 obs,203 account_id: str,204 policy: Dict,205 call_llm: Callable[[str], str],206 log: EpisodeLog,207 max_dp1_iters: int = 4,208 tuples: Optional[List[Dict]] = None,209) -> StepResult:210 """Run DP1 tool-gathering loop then DP2 flag decision for one account.211 Returns the latest StepResult (observation reflects any actions taken).212 213 If `tuples` is provided, each LLM decision appends a dict with keys:214 prompt, completion, step_reward, decision_type, platform, threshold,215 fp_penalty, step_index216 """217 last = StepResult(observation=obs, done=False, reward=None, message="")218 last.observation = obs219 plat = policy.get("platform") or log.platform220 fp_w = policy.get("fp_weight", "?")221 thr = float(policy.get("threshold", 0.0) or 0.0)222 223 # DP1 loop224 for _ in range(max_dp1_iters):225 acc = _find_account(last.observation, account_id)226 if acc is None:227 return last228 229 photo = acc.photo_reuse_score if acc.photo_reuse_score > 0 else None230 bio = acc.bio_template_score if acc.bio_template_score > 0 else None231 ip_signal = None # ip_cluster_signal is returned only in the step message; treat as unknown until check_ip called232 233 prompt = DP1_PROMPT.format(234 platform=policy.get("platform") or log.platform,235 primary_signal=policy.get("primary_signal", "?"),236 threshold=policy.get("threshold", 0.0),237 account_id=account_id,238 risk=acc.fake_risk_score,239 hub=acc.hub_legitimacy_score,240 photo=_render_signal(photo),241 bio=_render_signal(bio),242 ip=_render_signal(ip_signal),243 shared_ip=acc.shared_ip_count,244 steps_left=last.observation.steps_remaining,245 )246 log.dp1_calls += 1247 resp = call_llm(prompt)248 choice = _parse_dp1(resp)249 if choice is None:250 log.dp1_invalid += 1251 if tuples is not None:252 tuples.append({253 "prompt": prompt, "completion": resp, "step_reward": 0.0,254 "decision_type": "dp1", "platform": plat, "threshold": thr,255 "fp_penalty": fp_w, "step_index": log.steps_taken,256 })257 break258 if choice == "done":259 if tuples is not None:260 tuples.append({261 "prompt": prompt, "completion": resp, "step_reward": 0.0,262 "decision_type": "dp1", "platform": plat, "threshold": thr,263 "fp_penalty": fp_w, "step_index": log.steps_taken,264 })265 break266 267 atype = {268 "reverse_image_search": ActionType.REVERSE_IMAGE_SEARCH,269 "analyze_bio": ActionType.ANALYZE_BIO,270 "check_ip": ActionType.CHECK_IP,271 }[choice]272 last = client.step(FakeGangAction(action_type=atype, account_id=account_id))273 log.tool_calls[choice] = log.tool_calls.get(choice, 0) + 1274 log.steps_taken += 1275 if tuples is not None:276 tuples.append({277 "prompt": prompt, "completion": resp,278 "step_reward": float(last.reward) if last.reward is not None else 0.0,279 "decision_type": "dp1", "platform": plat, "threshold": thr,280 "fp_penalty": fp_w, "step_index": log.steps_taken,281 })282 if last.done or last.observation.steps_remaining <= 1:283 return last284 285 # Stop early if all cheap signals revealed286 acc2 = _find_account(last.observation, account_id)287 if acc2 and acc2.photo_reuse_score > 0 and acc2.bio_template_score > 0:288 break289 290 # DP2291 acc = _find_account(last.observation, account_id)292 if acc is None:293 return last294 prompt = DP2_PROMPT.format(295 platform=policy.get("platform") or log.platform,296 threshold=policy.get("threshold", 0.0),297 fp_weight=policy.get("fp_weight", "?"),298 account_id=account_id,299 risk=acc.fake_risk_score,300 hub=acc.hub_legitimacy_score,301 photo=_render_signal(acc.photo_reuse_score if acc.photo_reuse_score > 0 else None),302 bio=_render_signal(acc.bio_template_score if acc.bio_template_score > 0 else None),303 ip=_render_signal(None),304 shared_ip=acc.shared_ip_count,305 n_flagged=len(last.observation.flagged_ids),306 steps_left=last.observation.steps_remaining,307 )308 log.dp2_calls += 1309 resp = call_llm(prompt)310 choice = _parse_dp2(resp)311 step_reward = 0.0312 if choice is None:313 log.dp2_invalid += 1314 elif choice == "flag":315 last = client.step(FakeGangAction(action_type=ActionType.FLAG, account_id=account_id))316 log.flagged = len(last.observation.flagged_ids)317 step_reward = float(last.reward) if last.reward is not None else 0.0318 if tuples is not None:319 tuples.append({320 "prompt": prompt, "completion": resp, "step_reward": step_reward,321 "decision_type": "dp2", "platform": plat, "threshold": thr,322 "fp_penalty": fp_w, "step_index": log.steps_taken,323 })324 return last325 326 327# ---------------------------------------------------------------------------328# Single episode329# ---------------------------------------------------------------------------330 331def _run_episode(332 client: FakeGangEnvClient,333 model: str,334 platform: str,335 task: str,336 seed: int,337 call_llm: Callable[[str], str],338 max_accounts_per_episode: int = 15,339 collect_tuples: bool = False,340):341 """Run one episode. Returns EpisodeLog by default, or342 (EpisodeLog, list[tuple_dict]) when collect_tuples=True."""343 log = EpisodeLog(model=model, platform=platform, task=task, seed=seed)344 tuples: List[Dict] = [] if collect_tuples else None # type: ignore[assignment]345 t0 = time.time()346 347 res = client.reset(task=task, seed=seed)348 obs = res.observation349 350 # Deterministic step 0: GET_POLICY351 res = client.step(FakeGangAction(action_type=ActionType.GET_POLICY))352 log.tool_calls["get_policy"] += 1353 policy = _policy_from_message(res.message or res.observation.message)354 log.threshold = float(policy.get("threshold", 0.0) or 0.0)355 log.primary_signal = str(policy.get("primary_signal", "") or "")356 obs = res.observation357 358 investigated: set[str] = set()359 processed: set[str] = set()360 361 while not res.done and obs.steps_remaining > 1 and len(processed) < max_accounts_per_episode:362 # Candidate pool: suspects first, then visible not-yet-processed, ranked by risk.363 candidates = [a for a in obs.visible_accounts if a.account_id not in processed]364 suspects = set(obs.suspect_ids)365 candidates.sort(366 key=lambda a: (a.account_id in suspects, a.fake_risk_score),367 reverse=True,368 )369 if not candidates:370 break371 acc = candidates[0]372 aid = acc.account_id373 374 # Ensure inspected375 if aid not in obs.inspected_ids:376 res = client.step(FakeGangAction(action_type=ActionType.INSPECT, account_id=aid))377 log.steps_taken += 1378 log.inspected += 1379 obs = res.observation380 if res.done or obs.steps_remaining <= 1:381 break382 383 # Expand network once for very risky hubs384 acc_now = _find_account(obs, aid)385 if (386 acc_now387 and acc_now.fake_risk_score >= 0.80388 and aid not in investigated389 and obs.steps_remaining >= 5390 ):391 res = client.step(FakeGangAction(action_type=ActionType.INVESTIGATE_NETWORK, account_id=aid))392 log.tool_calls["investigate_network"] += 1393 log.steps_taken += 2394 investigated.add(aid)395 obs = res.observation396 if res.done or obs.steps_remaining <= 1:397 break398 399 # DP1 + DP2 per account400 res = _gather_and_flag(client, obs, aid, policy, call_llm, log, tuples=tuples)401 obs = res.observation402 processed.add(aid)403 if res.done:404 break405 406 # Final submit (if not already done)407 if not res.done:408 res = client.step(FakeGangAction(action_type=ActionType.SUBMIT))409 410 log.flagged = len(res.observation.flagged_ids)411 log.reward = res.reward412 log.final_message = (res.message or "")[:400]413 log.episode_id = getattr(res.observation, "episode_id", "") or f"{task}_{seed:03d}_{platform}"414 log.wall_seconds = round(time.time() - t0, 2)415 416 # Grader endpoint (optional)417 try:418 import requests419 g = requests.get(f"{client.base_url}/grader", timeout=30).json()420 log.grader_score = g.get("score")421 except Exception:422 pass423 424 if collect_tuples:425 # Attach episode-level identifiers to each decision tuple.426 for t in tuples:427 t["episode_id"] = log.episode_id428 t["grader_score"] = log.grader_score429 return log, tuples430 return log431 432 433# ---------------------------------------------------------------------------434# Public entry point435# ---------------------------------------------------------------------------436 437def run_evaluation(438 model_name: str,439 call_llm: Callable[[str], str],440 platform: str,441 base_url: str = "http://localhost:8000",442 tasks: Optional[List[str]] = None,443 seeds: Optional[List[int]] = None,444 out_path: Optional[str] = None,445) -> List[EpisodeLog]:446 tasks = tasks or ["easy", "medium", "hard"]447 seeds = seeds or [0, 1, 2]448 seeds = _seeds_for_platform(seeds, platform)449 450 out_path = out_path or str(451 _PARENT / "eval-models" / "results" /452 f"{model_name.replace('/', '_')}_{platform.lower()}_results.jsonl"453 )454 Path(out_path).parent.mkdir(parents=True, exist_ok=True)455 456 print(f"\n{'='*70}")457 print(f"Round 2 evaluation | model={model_name} | platform={platform}")458 print(f"Target: {base_url} | tasks={tasks} | seeds={seeds}")459 print(f"Log: {out_path}")460 print(f"{'='*70}")461 462 logs: List[EpisodeLog] = []463 client = FakeGangEnvClient(base_url=base_url)464 465 with open(out_path, "w") as f:466 for task in tasks:467 for seed in seeds:468 print(f"\n--- episode: task={task} seed={seed} ---")469 try:470 log = _run_episode(client, model_name, platform, task, seed, call_llm)471 except Exception as e:472 print(f" ✗ episode failed: {e}")473 log = EpisodeLog(474 model=model_name, platform=platform, task=task, seed=seed,475 final_message=f"EXCEPTION: {e}",476 )477 logs.append(log)478 f.write(json.dumps(asdict(log)) + "\n")479 f.flush()480 print(481 f" → steps={log.steps_taken} inspected={log.inspected} "482 f"flagged={log.flagged} dp1={log.dp1_calls}(bad={log.dp1_invalid}) "483 f"dp2={log.dp2_calls}(bad={log.dp2_invalid}) "484 f"reward={log.reward} grader={log.grader_score} ({log.wall_seconds}s)"485 )486 487 # Summary488 print(f"\n{'='*70}")489 print("SUMMARY")490 print(f"{'='*70}")491 rewards = [l.reward for l in logs if l.reward is not None]492 graders = [l.grader_score for l in logs if l.grader_score is not None]493 if rewards:494 print(f" mean reward: {sum(rewards)/len(rewards):.4f} (n={len(rewards)})")495 if graders:496 print(f" mean grader: {sum(graders)/len(graders):.4f} (n={len(graders)})")497 total_dp1 = sum(l.dp1_calls for l in logs)498 total_dp2 = sum(l.dp2_calls for l in logs)499 bad_dp1 = sum(l.dp1_invalid for l in logs)500 bad_dp2 = sum(l.dp2_invalid for l in logs)501 print(f" DP1 calls: {total_dp1} (invalid: {bad_dp1})")502 print(f" DP2 calls: {total_dp2} (invalid: {bad_dp2})")503 print(f" Logged to: {out_path}")504 return logs505 506 507# ---------------------------------------------------------------------------508# Standard CLI argument parser (reused by every model shim)509# ---------------------------------------------------------------------------510 511def build_cli():512 import argparse513 parser = argparse.ArgumentParser(description="Round 2 eval runner")514 parser.add_argument("--url", default=os.getenv("API_BASE_URL_ENV", "http://localhost:8000"),515 help="Environment server URL")516 parser.add_argument("--platform", default="Instagram", help="Platform name (Instagram/Snapchat/...)")517 parser.add_argument("--tasks", nargs="+", default=["easy", "medium", "hard"])518 parser.add_argument("--seeds", nargs="+", type=int, default=[0, 1, 2])519 parser.add_argument("--out", default=None, help="Output JSONL path")520 return parser521