"""GROUNDING-V2 §4b.0 — **A0 입력 적격성 관문** + 12-subject manifest 잠그기. ★왜 이 관문이 있나 (Codex, 2026-08-30). `grounding_a0` 가 **어떤 에피소드에서도 안 돌았다**. 그래서 저장 체크포인트로 만든 subject 는 전부 ``quote_source="entity_description"`` — LLM 이 상상해 쓴 묘사이지 **원고 문장이 아니다**. 그 payload 로 검색하면 프로덕션이 보낼 입력과 다른 것을 재게 되고, `subject_payload_hash` 가 입력 신원에 들어가므로 나중에 A0 를 돌리면 그 표본은 **전부 다른 조사**가 되어 재사용도 0이다. ## 순서 (바꾸지 않는다) ① A0 (유료·소액 / 재사용이면 0) → ② NEED 개가 전부 manuscript 인지 확인 → ③ manifest content hash 로 잠금 → ④ 그 다음에야 검색 실험(9 × 1/2/4/8 = 19회) ★**결과를 본 뒤 편한 에피소드로 바꾸면 안 된다.** 대상과 fallback 순서를 아래에 미리 박아 둔다. ## 왜 이 에피소드인가 혼합-era 에피소드는 **본 acceptance 표본으로 못 쓴다** — era 문자열이 하나라도 실제 시간대가 여럿이면 batch leakage 와 시대 귀속 오류를 구별할 수 없다 (Codex). 그래서 **단일 시대·지역** 모집단만 쓴다. python -m tools.prompt_measure.grounding_a0_manifest --dry-run # 무료 python -m tools.prompt_measure.grounding_a0_manifest --run # 유료 1회 """ from __future__ import annotations import argparse import hashlib import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) #: ★**이번 주행에서 승인된 대상**. 결과를 본 뒤 바꾸면 표본이 아니라 결과에 #: 맞춘 것이 된다. #: ★★한 개다 — 「모자라면 다음 에피소드」는 **이번 범위가 아니다** #: (Codex 2026-08-30). 그건 그 자리에서 **중단하고 보고**하는 것이고, 다른 #: 에피소드는 **별도 승인**을 받는다. 자동 fallback 으로 두면 승인 범위를 #: 코드가 스스로 넓힌다. EPISODE_ORDER = [ # (project 접두어, episode 접두어, 왜) ("da049582", "97375a4b", "1983년 늦가을 단일 시대. 이전 판에서 A0 후보 13개"), ] #: ★다음 후보 — **승인 대상이 아니다.** 여기 적어 두는 이유는, 표본을 못 #: 채웠을 때 「그럼 어디로 가느냐」를 결과를 본 뒤 고르지 않기 위해서다. #: 옮기려면 `EPISODE_ORDER` 로 **손으로** 옮기고 승인을 다시 받는다. NEXT_CANDIDATES = [ ("da049582", "0aea12c2", "같은 1983년. 별도 승인 뒤에만"), ] #: ★실험 계약은 **앱 모듈이 정본**이다 — 여기서 다시 적으면 두 곳이 갈린다. from app.modules.pipeline.grounding_claims import ( # noqa: E402 subject_payload_hash as _subject_payload_hash) from app.modules.pipeline.grounding_claims_search import ( # noqa: E402 EXPERIMENT_BATCH_SIZES as BATCH_CANDIDATES, EXPERIMENT_SAMPLE_SIZE as NEED, ) #: ★비용 가드 — **승인된 대상 수**만큼. 재시도는 `strict_single_attempt` 가 0으로. #: ★전에 `1` 로 박아 두고 목록에 2개를 넣었더니 **선언한 fallback 이 닿을 수 #: 없는 코드**가 됐다. 가드와 목록이 **따로 놀면** 둘 중 하나가 거짓말이 된다 — #: 그래서 목록에서 유도한다. MAX_A0_CALLS = len(EPISODE_ORDER) def _episode_dir(root: Path, proj: str, epi: str) -> Path | None: for d in sorted(root.glob(f"{proj}*/checkpoints/episodes/{epi}*")): if (d / "entity_merge" / "manifest.json").exists(): return d return None def _cleaned_text(ep: Path) -> str: """A0 가 읽는 것과 **같은 원문**. ★자르지 않는다.""" for step in ("text_cleanup", "scene_save"): p = ep / step / "manifest.json" if not p.exists(): continue data = json.loads(p.read_text(encoding="utf-8")).get("data") or {} for key in ("cleaned_text", "text", "fulltext"): v = data.get(key) if isinstance(v, str) and v.strip(): return v scenes = data.get("scenes") if isinstance(scenes, list) and scenes: joined = "\n\n".join( str((s or {}).get("content") or (s or {}).get("text") or "") for s in scenes) if joined.strip(): return joined raise SystemExit(f"원문을 못 찾았다: {ep}") def _rules(ep: Path) -> dict: return json.loads((ep / "visual_world_rules" / "manifest.json") .read_text(encoding="utf-8"))["data"] def survey(root: Path) -> list[dict]: """무료 — A0 **없이** 지금 무엇이 있는지만 본다.""" from app.modules.pipeline.grounding_shadow import ( build_subjects_from_saved_episode) out = [] for proj, epi, why in EPISODE_ORDER: ep = _episode_dir(root, proj, epi) if ep is None: out.append({"coord": f"{proj}/{epi}", "error": "디렉토리 없음", "why": why}) continue r = _rules(ep) b = build_subjects_from_saved_episode( ep, project_id=ep.parents[2].name, episode_id=ep.name) srcs: dict[str, int] = {} for s in b["subjects"]: k = s.get("quote_source") or "없음" srcs[k] = srcs.get(k, 0) + 1 out.append({ "coord": f"{proj}/{epi}", "why": why, "era": r.get("era"), "region": r.get("region"), "subjects_now": len(b["subjects"]), "quote_sources_now": srcs, "text_chars": len(_cleaned_text(ep)), }) return out def run_a0(root: Path) -> dict: """★유료 — A0 를 **한 에피소드에 한 번**. 재시도 없음.""" from app.modules.pipeline.grounding_a0 import collect_candidates calls = 0 for proj, epi, why in EPISODE_ORDER: ep = _episode_dir(root, proj, epi) if ep is None: continue if calls >= MAX_A0_CALLS: # ★가드는 **세는 것**으로 건다. 「한 번만 부르겠다」는 주석이 아니라. raise SystemExit( f"A0 호출 상한 {MAX_A0_CALLS} 에 걸렸다 — 회차를 늘리려면 " "EPISODE_ORDER 를 먼저 늘려라(대상은 미리 박는다)") pid, eid = ep.parents[2].name, ep.name r = _rules(ep) calls += 1 out = collect_candidates( _cleaned_text(ep), project_id=pid, episode_id=eid, era=str(r.get("era") or "").strip(), region=str(r.get("region") or "").strip(), # ★재시도 0. 실패하면 실패로 남는다 — 조용히 두 번 사지 않는다. strict_single_attempt=True) cands = out.get("candidates") or [] print(f"{proj}/{epi}: 후보 {len(cands)}개 · " f"지어낸 인용 {len(out.get('hallucinated') or [])}건") if len(cands) >= NEED: return {"coord": f"{proj}/{epi}", "why": why, "project_id": pid, "episode_id": eid, "era": str(r.get("era") or ""), "region": str(r.get("region") or ""), "a0": out, "episode_dir": str(ep)} print(f" → {NEED}개 미만이라 다음 fallback 으로") raise SystemExit( f"승인된 대상이 {NEED}개를 못 채웠다 — 여기서 **선다**. " "entity_description 이나 합성 fixture 로 메우지 않고, 다른 에피소드로 " f"알아서 넘어가지도 않는다. 다음 후보는 {NEXT_CANDIDATES} 이고 " "EPISODE_ORDER 로 손으로 옮긴 뒤 승인을 다시 받는다") def build_manifest(picked: dict) -> dict: """★`NEED` 개를 고르고 **content hash 로 잠근다**. ★수를 여기 다시 적지 않는다 — 정본은 `grounding_claims_search.EXPERIMENT_SAMPLE_SIZE` 다. """ from app.modules.pipeline.grounding_shadow import ( build_subjects_from_saved_episode) ep = Path(picked["episode_dir"]) built = build_subjects_from_saved_episode( ep, project_id=picked["project_id"], episode_id=picked["episode_id"], a0_candidates=picked["a0"].get("candidates") or []) subs = built["subjects"] # ★**원고 기반만 쓴다.** 이게 이 관문의 전부다. manuscript = [s for s in subs if s.get("quote_source") == "manuscript"] if len(manuscript) < NEED: # ★**왜 모자란지**까지 적는다. 「2개뿐이다」만으로는 A0 가 못 건진 # 것인지, 건졌는데 결속이 안 된 것인지 구분이 안 된다. cands = len(picked["a0"].get("candidates") or []) # ★★**사유별 개수**를 낸다. `build_subjects` 가 이미 세고 있는데 안 # 찍고 있었다 — 「2개뿐이다」와 「ambiguous 5건이라 못 붙였다」는 # 다음에 고칠 곳이 완전히 다르다. raise SystemExit( f"원고 기반이 {len(manuscript)}개뿐이다 — 여기서 **선다**. " f"A0 후보 {cands}개 · entity_merge 대상 {len(subs)}개 · " f"결속된 것 {len(manuscript)}개 · 못 붙은 대상 " f"{len(built.get('unbound') or [])}개 · " f"사유별 {built.get('carry_reasons') or {}}. " "entity_description 으로 메우지 않는다") # ★선정 규칙: **id 오름차순 앞에서부터**. 「좋아 보이는 것」을 고르지 않는다. manuscript.sort(key=lambda s: s["research_subject_id"]) rows = [] for s in manuscript[:NEED]: pv = s.get("provenance") or {} rows.append({ "research_subject_id": s["research_subject_id"], "surface_form": s.get("surface_form") or "", "source_anchor": s.get("source_anchor") or "", "source_quote": s.get("source_quote") or "", "quote_source": s.get("quote_source") or "", "owner_type": s.get("owner_type") or "", "short_id": pv.get("short_id") or "", # ★규칙은 **프로덕션 것**을 쓴다 — 여기서 따로 계산하면 도구가 # 잠근 신원과 프로덕션이 저장한 신원이 갈린다. "subject_payload_hash": _subject_payload_hash(s), }) text = _cleaned_text(ep) from app.modules.pipeline import grounding_a0 as a0 from app.modules.pipeline import grounding_claims_search as gcs # ★★content_hash 는 **subjects 만**으로는 모자란다 (Codex). era·region· # 검색 모델·검색 팩·크기 상한이 바뀌면 그건 **다른 실험**이다. # ★검색 모델은 A0 것에서 **빌리지 않는다** — 조사 모델과 후보 추출 모델은 # 별개 축이고, 빌리면 A0 를 바꾼 날 검색 모델이 따라 움직인다. lock = { "subjects": rows, "era": picked["era"], "region": picked["region"], "search_model": SEARCH_MODEL, "search_pack_version": gcs.PROMPT_PACK_VERSION, "search_pack_manifest_hash": gcs.load_pack()["pack_manifest_hash"], "max_prompt_bytes": gcs.MAX_PROMPT_BYTES, "hard_prompt_bytes": gcs.HARD_PROMPT_BYTES, "batch_candidates": list(BATCH_CANDIDATES), # ★★**request 의 의미도 잠근다** — 이것들이 바뀌면 같은 표본이라도 # 다른 요청을 보낸 것이라 다른 실험이다 (Codex). "text_only_search": True, "schema_strict": False, "source_capture": "results(beta) + action.sources(stable) · " "snippet 없으면 첫 호출에서 선다", "include": ["web_search_call.results"], "sdk_retries": 0, } body = json.dumps(lock, ensure_ascii=False, sort_keys=True) return { "contract": "grounding-v2 §4b batch 실험 표본", "gate": (f"A0 입력 적격성 — {NEED}개가 **전부 원고 기반**이어야 한다. " "entity_description 이나 합성 fixture 로 메우지 않는다"), "selection_rule": f"research_subject_id 오름차순 앞에서부터 {NEED}개", "batch_candidates": list(BATCH_CANDIDATES), "why_not_twelve": ("9개를 한 묶음으로 보내는 것은 12-subject 결속·" "leakage 검증이 아니다. 12는 원고 기반 고유 subject 가 " "12개 생긴 뒤 별도 재검증한다"), "logical_calls": sum(-(-NEED // b) for b in BATCH_CANDIDATES), "sample_caveat": ("1.3KB 검증 원고에서 A0 승격을 거쳐 나온 9개다 — " "실제 에피소드 대표성은 없다"), "episode_order": [{"coord": f"{p}/{e}", "why": w} for p, e, w in EPISODE_ORDER], "picked": picked["coord"], "why_picked": picked["why"], "project_id": picked["project_id"], "episode_id": picked["episode_id"], "manuscript_hash": hashlib.sha256(text.encode()).hexdigest()[:16], # ★★**A0 좌표는 실제로 그 산출을 낸 판의 것**이어야 한다. 재사용인데 # 지금 코드의 팩 버전을 적으면, 팩이 바뀐 날 「이 후보는 새 팩이 낸 # 것」이라는 거짓말이 된다. 오늘은 우연히 둘 다 같아서 안 보였다. # ★현재 팩은 `builder_pack_version` 으로 **따로** 남긴다 — 둘이 다르면 # 그 사실 자체가 읽는 사람에게 필요한 정보다. "a0_pack_version": (picked.get("a0_reused_from") or {}).get( "pack_version") or a0.PROMPT_PACK_VERSION, "a0_model": (picked.get("a0_reused_from") or {}).get("model") or "", "builder_pack_version": a0.PROMPT_PACK_VERSION, "a0_candidates": len(picked["a0"].get("candidates") or []), "manuscript_subjects": len(manuscript), "count": len(rows), # ★★era·region·search_model·subjects 는 **`locked` 에만 있다.** # 같은 값을 top-level 에도 두면 잠근 사본과 쓰는 사본이 두 벌이 되고, # 실측으로 top-level 만 바꿔도 content_hash 검사를 통과했다 (Codex). "locked": lock, "content_hash": hashlib.sha256(body.encode()).hexdigest()[:16], } #: ★2026-08-30 유료 A0 실행분. Opik trace 에서 복구해 **추적되게** 둔 것이고, #: 같은 에피소드·같은 팩·같은 모델이다. 재구매 없이 manifest 를 만들 때 쓴다. RECOVERED_FIXTURE = ("tests/fixtures/grounding/a0_recovered_97375a4b.json") #: ★**검색 모델은 따로 둔다.** A0(후보 추출)와 검색(조사)은 별개 축이라, #: A0 모델에서 빌리면 A0 를 바꾼 날 검색 모델이 따라 움직인다 (Codex). SEARCH_MODEL = "gpt-5.6-sol" def from_recovered(root: Path) -> dict: """★유료 A0 를 **다시 사지 않고** 2026-08-30 실행분으로 manifest 를 만든다. 검색과 달리 A0 는 같은 입력에 대해 거의 같은 것을 낸다 — 그걸 확인하려고 또 사는 것은 낭비다. 대신 **재사용했다는 사실을 manifest 에 적는다.** """ _fp = Path(RECOVERED_FIXTURE) if not _fp.is_absolute(): _fp = Path(__file__).resolve().parents[2] / RECOVERED_FIXTURE fx = json.loads(_fp.read_text(encoding="utf-8")) proj, epi, why = EPISODE_ORDER[0] # ★★fixture 가 **승인된 대상의 것**인지. 다른 에피소드의 산출로 이 판의 # manifest 를 만들면 표본이 통째로 딴것이 된다. if str(fx.get("episode") or "") != f"{proj}/{epi}": raise SystemExit( f"fixture 가 다른 에피소드 것이다: {fx.get('episode')!r} vs " f"승인 대상 {proj}/{epi}") ep = _episode_dir(root, proj, epi) if ep is None: raise SystemExit(f"에피소드 디렉토리를 못 찾았다: {proj}/{epi}") pid, eid = ep.parents[2].name, ep.name text = _cleaned_text(ep) # ★★재사용은 「같은 결과가 나올 것」이 아니라 **입력이 그대로인가**로 # 판단한다 (Codex). 글자 수만 보면 같은 길이의 다른 원고를 통과시킨다. ents = json.loads((ep / "entity_merge" / "manifest.json") .read_text(encoding="utf-8"))["data"] now = { "manuscript_sha256": hashlib.sha256(text.encode()).hexdigest(), "entity_merge_sha256": hashlib.sha256(json.dumps( {k: ents.get(k) or [] for k in ("characters", "locations", "props")}, ensure_ascii=False, sort_keys=True).encode()).hexdigest(), } drift = [k for k, v in now.items() if fx.get(k) != v] if drift: raise SystemExit( f"재사용 입력이 바뀌었다: {drift} — 그 산출은 **다른 판의 것**이다. " "A0 를 다시 사야 한다") from app.modules.pipeline.grounding_subject import build_subject kept, dropped = [], [] for c in fx["a0_candidates"]: q = (c.get("source_quote") or "").strip() if not (q and q in text): # ★★**조용히 버리지 않는다.** 입력 hash 가 맞는데 인용이 원문에 # 없다면 fixture 나 후처리가 어긋난 것이다 — 그걸 건너뛰면 표본이 # 줄어든 채 「9개로 쟀다」가 된다. 오늘만 이 부류를 여러 번 만났다. dropped.append({"surface_form": c.get("surface_form"), "why": "source_quote 가 원문에 없다"}) continue kept.append({**c, **build_subject( project_id=pid, episode_id=eid, source_anchor=c["source_anchor"], surface_form=c["surface_form"], owner_type=c["owner_type"], provenance={"a0_pack_version": fx["a0_pack_version"], "a0_physical_model": fx["model"], "source_step": "grounding_a0"}), "source_quote": q}) if dropped: raise SystemExit( f"복구 후보 {len(dropped)}개의 인용이 원문에 없다: " f"{[d['surface_form'] for d in dropped][:5]} — 입력 hash 는 맞는데 " "인용이 안 맞으면 fixture 나 후처리가 어긋난 것이다. 조용히 " "버리고 줄어든 표본으로 재지 않는다") if len(kept) != len(fx["a0_candidates"]): raise SystemExit( f"재사용 후보 수가 다르다: {len(kept)} != " f"{len(fx['a0_candidates'])}") print(f"{proj}/{epi}: 재사용 후보 {len(kept)}개 " f"(2026-08-30 실행분 · trace {fx['opik_trace']})") return {"coord": f"{proj}/{epi}", "why": why, "project_id": pid, "episode_id": eid, "era": str(_rules(ep).get("era") or ""), "region": str(_rules(ep).get("region") or ""), "a0": {"candidates": kept, "hallucinated": []}, "episode_dir": str(ep), "a0_reused_from": {"fixture": RECOVERED_FIXTURE, "opik_trace": fx["opik_trace"], "ran_at": "2026-08-30", "pack_version": fx["a0_pack_version"], "model": fx["model"]}} def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--projects-root", type=Path, default=Path("../projects")) ap.add_argument("--dry-run", action="store_true", help="무료 — A0 없이 지금 무엇이 있는지만 본다") ap.add_argument("--run", action="store_true", help="★유료 — A0 1회") ap.add_argument("--reuse", action="store_true", help="무료 — 2026-08-30 A0 실행분을 재사용해 manifest 를 만든다") ap.add_argument("--out", type=Path, default=Path("tests/fixtures/grounding/" "claims_search_manifest.json")) args = ap.parse_args() if args.reuse: picked = from_recovered(args.projects_root) doc = build_manifest(picked) # ★재사용했다는 사실을 **manifest 안에** 남긴다 — 나중에 「그때 A0 를 # 샀나」를 못 되짚으면 이 표본의 출처가 흐려진다. doc["a0_reused_from"] = picked["a0_reused_from"] args.out.parent.mkdir(parents=True, exist_ok=True) args.out.write_text(json.dumps(doc, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(f"\n{doc['count']}개 · content_hash={doc['content_hash']} → {args.out}") for r in doc["locked"]["subjects"]: print(f" {r['owner_type']:9} {r['surface_form'][:38]}") return 0 if args.dry_run or not args.run: for row in survey(args.projects_root): print(json.dumps(row, ensure_ascii=False, indent=2)) if not args.run: print("\n★유료 실행은 --run 이다. 지금은 아무것도 안 샀다.") return 0 picked = run_a0(args.projects_root) # ★★**게이트 전에 저장한다.** 처음에는 게이트가 먼저 서서, 유료로 산 # A0 후보 23개가 통째로 사라졌다 — 다시 사야만 무엇이 나왔는지 볼 수 # 있었다. 산 것은 **판정과 무관하게** 먼저 남긴다. raw = args.out.with_name(args.out.stem + "_a0_raw.json") raw.parent.mkdir(parents=True, exist_ok=True) raw.write_text(json.dumps( {k: v for k, v in picked.items() if k != "episode_dir"}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(f"A0 원산출 저장 → {raw}") doc = build_manifest(picked) args.out.parent.mkdir(parents=True, exist_ok=True) args.out.write_text(json.dumps(doc, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(f"\n{doc['count']}개 · content_hash={doc['content_hash']} → {args.out}") for r in doc["subjects"]: print(f" {r['short_id']:5} {r['owner_type']:9} {r['surface_form'][:34]}") return 0 if __name__ == "__main__": raise SystemExit(main())