#!/usr/bin/env python """GROUNDING-V2 §2-3 — 저장된 에피소드에 `shadow_plan` 을 돌려 본다. 무엇을 답하나 ① 분류·계획 기구가 **저장된 실데이터**에서 도는가 ② **검색 호출이 0인가** ③ **production 산출이 한 파일도 안 바뀌는가** (manifest 만이 아니라 **전 파일**) 무엇을 답하지 **않나** ★「회수권이 후보로 살아 있나」는 여기서 못 본다 — A0(원문 후보 수집)가 §2-3.5 라, 지금 재생기는 **이미 살아남은 엔티티**만 읽는다. 여기서 안 보이는 것을 「없다」로 읽으면 안 된다. ★``--classify`` 는 **유료다.** 검색은 0이지만 에피소드마다 Sol 을 부르고 Router 재시도·fallback 도 열려 있다. 그래서 ``--i-know-this-costs-money`` 를 같이 줘야 돌고, 논리 호출 수를 따로 찍는다. .venv/bin/python tools/prompt_measure/shadow_plan_replay.py [--limit N] [--project ID] 종료코드: 0 통과 · 1 어긋남 · 2 잴 것이 없음 """ from __future__ import annotations import argparse import collections import sys from pathlib import Path _REPO = Path(__file__).resolve().parents[3] sys.path.insert(0, str(_REPO / "backend")) from app.core.grounding_mode import ( # noqa: E402 GROUNDING_MODE_SHADOW_PLAN, buys_no_research_at_all, buys_v2_research, ) from app.modules.pipeline.grounding_shadow import ( # noqa: E402 ShadowSourceError, replay_shadow_plan, ) def _default_projects_root() -> Path: """★symlink 를 만들지 않는다. worktree 에는 ``projects/`` 가 안 따라오므로 (gitignore) **경로를 인자로 받는다** — 저장소에 절대경로 링크를 커밋하면 fresh clone 에서 깨지고 개인 경로가 노출된다.""" return _REPO / "projects" def _episode_dirs(root: Path, limit: int | None, project: str | None): """★project 로 먼저 거른 **뒤에** limit 을 건다. 반대로 하면 「전체 첫 N개를 뽑고 그중 이 프로젝트」가 되어 0건이 나온다 — 실제로 한 번 그렇게 재고 「없다」로 읽을 뻔했다. """ seen = [] for manifest in sorted(root.glob("*/checkpoints/episodes/*/entity_filter/manifest.json")): ep_dir = manifest.parent.parent project_id = ep_dir.parents[2].name if project and not project_id.startswith(project): continue seen.append((project_id, ep_dir.name, ep_dir)) if limit and len(seen) >= limit: break return seen def _effective_target() -> tuple[str, str, str, str]: """★실효 설정에서 읽되 **백엔드별로 갈라야 한다.** 처음엔 backend 와 무관하게 언제나 gemini 를 돌려주고 **backend alias 를 version 이라고 불렀다.** 둘 다 틀렸다 — ``STILL_IMAGE_BACKEND=grok2`` 면 실제로 굽는 것은 ``x-ai/grok-imagine-image-2.0`` 이고, ``nb2``/``grok2`` 는 버전이 아니라 **백엔드 이름**이다. 이 provider 들은 **버전을 모델 id 에 담는다**(``gemini-3.1-flash-image`` 는 GA, ``…-2.0`` 은 2.0). 따로 version 칸이 없으므로 model id 를 그대로 version 자리에 둔다 — 없는 버전을 지어내지 않는다. Returns: (provider, model, version, backend) """ from app.core.config import settings backend = getattr(settings, "still_image_backend", "") if backend == "grok2": provider, model = "xai", getattr(settings, "grok_image_model", "") elif backend == "nb2": provider, model = "gemini", getattr(settings, "gemini_image_model", "") else: raise SystemExit( f"STILL_IMAGE_BACKEND={backend!r} 을 모른다 — nb2|grok2 만 안다. " "모르는 백엔드로 capability 를 판정하면 실제로 굽는 모델과 어긋난다") if not model: raise SystemExit(f"{backend} 의 이미지 모델 설정이 비어 있다") return (provider, model, model, backend) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--projects-root", type=Path, default=None, help="projects/ 디렉토리 (기본: 저장소 루트)") ap.add_argument("--limit", type=int, default=None) ap.add_argument("--project", default=None, help="이 project id 만") ap.add_argument("--classify", action="store_true", help="★Sol 분류기를 실제로 부른다 — 유료다") ap.add_argument("--i-know-this-costs-money", action="store_true", help="--classify 와 함께 줘야 실제로 돈다") ap.add_argument("--write-shadow", action="store_true", help="shadow 체크포인트를 남긴다 (production 은 안 건드린다)") args = ap.parse_args() assert buys_no_research_at_all(GROUNDING_MODE_SHADOW_PLAN) and \ not buys_v2_research(GROUNDING_MODE_SHADOW_PLAN), \ "shadow_plan 이 조사를 사는 모드로 바뀌었다 — 이 도구를 돌리면 안 된다" if args.classify and not args.i_know_this_costs_money: print("--classify 는 에피소드마다 Sol 을 부른다 (검색은 0이지만 provider 호출은 유료).") print("정말 돌리려면 --i-know-this-costs-money 를 같이 주세요.") return 1 root = args.projects_root or _default_projects_root() if not root.is_dir(): print(f"projects 디렉토리가 없다: {root}\n --projects-root 로 경로를 주세요.") return 2 eps = _episode_dirs(root, args.limit, args.project) if not eps: print(f"잴 에피소드가 없다 — {root} 아래 entity_filter 체크포인트를 못 찾았다") return 2 classify_fn = None target = _effective_target() if args.classify: from app.modules.pipeline.grounding_classifier import classify as _classify def classify_fn(subjects, *, era="", region=""): # noqa: F811 return _classify( subjects, era=era, region=region, ) # ★분류는 **이미지 모델을 안 본다**(2026-08-30 폐기 축 정리). 좌표를 # 찍어 두는 것은 「이 판이 어느 backend 설정에서 돌았나」 기록일 뿐, # 판정 입력이 아니다. print(f"이 주행의 이미지 backend(참고): {target[3]} · " f"provider={target[0]} · model={target[1]} — 분류 입력 아님") total = collections.Counter() searches = logical_calls = 0 judges: set = set() changed, broken = [], [] for project_id, episode_id, ep_dir in eps: label = f"{project_id[:8]}/{episode_id[:8]}" try: out = replay_shadow_plan( ep_dir, project_id=project_id, episode_id=episode_id, classify_fn=classify_fn, write_checkpoint=args.write_shadow) except ShadowSourceError as exc: # ★깨진 에피소드를 조용히 건너뛰지 않는다. 그러면 일부가 깨져도 전체가 통과다. broken.append(f"{label}: {exc}") continue total.update(out["counts"]) searches += out["search_calls"] logical_calls += out["classifier_logical_calls"] j = out.get("classifier_judge") or {} if j.get("physical_model"): judges.add(f"{j.get('alias')}/{j['physical_model']}") if not out["production_unchanged"]: changed.append(label) if args.classify: print(f" {label} 시대={out['era']} 지역={out['region']}") for d in sorted(out["decided"], key=lambda x: x.get("route", "")): print(f" {d.get('route','?'):10} {(d.get('_short_id') or '--'):5} " f"{(d.get('_surface_form') or '')[:30]:30} " f"{d.get('reason','')[:44]}") subjects = sum(total.values()) print(f"\n에피소드 {len(eps)}개 (읽음 {len(eps) - len(broken)}) · 후보 {subjects}개") print(f" 검색 호출 {searches}") print(f" 분류기 **논리** 호출 {logical_calls} ★검색은 아니지만 유료다") print(f" (Router 재시도·fallback 때문에 **실제 전송은 이보다 많을 수 있다**)") if judges: print(f" 실제 판정 모델 {sorted(judges)}") print(f" production 변경 {len(changed)}건") print(f" 읽기 실패 {len(broken)}건") print(f" route {dict(total)}") bad = [] if searches: bad.append(f"검색 호출이 {searches}건 났다 — shadow_plan 은 0이어야 한다") if changed: bad.append(f"production 산출이 바뀐 에피소드 {len(changed)}건: {changed[:5]}") if broken: bad.append(f"저장 CP 를 못 읽은 에피소드 {len(broken)}건: {broken[:3]}") if subjects == 0: # ★0 은 「깨끗함」이 아니라 **재는 도구가 아무것도 못 읽었다**는 뜻일 수 있다. bad.append("후보가 하나도 안 나왔다 — 결함보다 재는 도구를 먼저 의심하라") if not args.classify and total.get("unresolved", 0) != subjects: bad.append("분류를 안 돌렸는데 미확정이 아닌 route 가 나왔다") if bad: print("\n어긋남:") for b in bad: print(f" · {b}") return 1 print("\n통과 — 검색 0 · production 불변") return 0 if __name__ == "__main__": raise SystemExit(main())