"""GROUNDING-V2 §4b — batch 크기 실험 **실행기**. 잠근 9 subject × batch 1/2/4/8 = **19 논리 호출** ★순서를 지킨다: manifest 고정 → acceptance 고정 → **그 다음에야** 유료 주행. ★프롬프트는 실험 중 **안 만진다.** 정하는 것은 batch 크기 하나뿐이고, 결과가 나쁘면 프롬프트를 고치는 게 아니라 **batch 를 줄인다** (Codex). 그래야 이 19회가 프롬프트 튜닝이 안 되고 §2-3c 의 held-out 도 깨끗하게 남는다. python -m tools.prompt_measure.grounding_batch_experiment --dry-run # 무료 python -m tools.prompt_measure.grounding_batch_experiment --run # ★유료 """ from __future__ import annotations import argparse import json import math import sys import time from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from app.modules.pipeline import grounding_claims_search as gcs # noqa: E402 from app.modules.pipeline.grounding_claims_acceptance import ( # noqa: E402 V_EMPTY, choose_batch_size, contract_survival, score_run, url_baseline) from app.modules.pipeline.grounding_claims_support import ( # noqa: E402 check_run as support_check_run) MANIFEST = "tests/fixtures/grounding/claims_search_manifest.json" #: ★비용 가드는 **세는 것**으로 건다. 「19회만」은 주석이 아니다. #: 실제로 나간 물리 전송이 이 수를 넘으면 **선다**. MAX_LOGICAL_CALLS = sum(math.ceil(gcs.EXPERIMENT_SAMPLE_SIZE / b) for b in gcs.EXPERIMENT_BATCH_SIZES) #: ★★★**승인된 신규 구매 상한.** 논리 19 중 **고유 payload 는 16** 이고 #: (꼬리 singleton 이 크기마다 겹친다), 그중 **하나는 이미 샀다**(진단 1회). #: 그래서 이 산출에서 새로 살 수 있는 것은 **15**다 (Codex 승인 범위). #: ★설명이 아니라 **문**이다 — 보내기 전에 세고, 넘으면 **한 번도 안 산다**. MAX_NEW_LOGICAL_CALLS = 15 #: ★이 산출에서는 **앞서 산 것의 재사용이 필수**다 (Codex). journal 이 #: 지워졌거나 다른 디렉토리면 이미 산 것을 다시 사게 되므로 선다. #: ★「첫 대상」이 아니다 — 진단 대상은 `--subject` 로 고르므로, 보는 것은 #: 「이 산출이 앞 주행을 잇는가」다. #: ★사용자가 켜고 끌 수 있는 문이 아니다 — 코드가 스스로 범위를 못 넓힌다. REQUIRE_DIAGNOSTIC_REUSE = True def load_manifest(root: Path) -> dict: """잠근 표본. ★content hash 를 **다시 계산해 맞춘다** — 잠갔다는 말이 맞는지 확인 없이 쓰면 그 사이 바뀐 것을 못 본다.""" import hashlib doc = json.loads((root / MANIFEST).read_text(encoding="utf-8")) # ★★**잠근 것 전부**로 다시 계산한다 — subjects 만 보면 era·검색 모델·팩· # 상한이 바뀌어도 「그때 그 실험」으로 읽힌다 (Codex). lock = doc.get("locked") if not isinstance(lock, dict) or "search_model" not in lock: raise SystemExit("manifest 에 잠근 목록(`locked`)이 없다 — 다시 만들어라") body = json.dumps(lock, ensure_ascii=False, sort_keys=True) got = hashlib.sha256(body.encode()).hexdigest()[:16] if got != doc.get("content_hash"): raise SystemExit( f"manifest 가 잠근 뒤 바뀌었다: {got} != {doc.get('content_hash')}") if len(lock["subjects"]) != gcs.EXPERIMENT_SAMPLE_SIZE: raise SystemExit( f"표본 수가 계약과 다르다: {len(lock['subjects'])} != " f"{gcs.EXPERIMENT_SAMPLE_SIZE}") # ★잠근 값이 **지금 코드와 같은지** 본다. 다르면 다른 실험이다. # ★top-level 에 잠근 값의 **사본이 남아 있으면** 선다 — 봉투가 둘이면 # 어느 쪽이 정본인지 모른다. dup = [k for k in ("era", "region", "search_model", "subjects") if k in doc] if dup: raise SystemExit(f"manifest top-level 에 사본이 있다: {dup} — 봉투는 하나다") now = {"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(gcs.EXPERIMENT_BATCH_SIZES)} # ★★**요청의 뜻도 전부** 대조한다 (Codex). 팩 좌표만 맞으면 통과하던 때는 # 잠근 것이 `results`(beta)·`store=null`·snippet 요구인데 코드는 # `action.sources`·`store=False`·HTTP 확인이었고, dry-run 이 초록이었다. # **거짓 잠금이 통과하면 잠근 의미가 없다.** now.update(gcs.request_contract()) drift = [k for k, v in now.items() if lock.get(k) != v] if drift: raise SystemExit(f"잠근 뒤 코드가 바뀌었다: {drift} — manifest 를 다시 만들어라") bad = [s for s in lock["subjects"] if s.get("quote_source") != "manuscript" or not str(s.get("source_quote") or "").strip()] if bad: # ★이 관문의 전부다 — 상상 묘사로 조사하지 않는다. raise SystemExit(f"원고 기반이 아닌 표본 {len(bad)}개") return doc def read_prior(out_dir: Path) -> list: """이 산출 디렉토리에 **이미 산 행 전부**. ★한 경로로 읽는다. ★★★진단 1회와 full 주행이 **서로 다른 파일만** 읽으면, 남은 실험을 재개할 때 이미 산 행을 **다시 산다** (Codex). 재사용 판정은 `payload_hash + 모델` 로 하므로, 여기서는 **다 읽어** 넘기고 신원이 다른 것은 `search_claims` 가 알아서 안 쓴다. """ rows = [] for f in sorted(out_dir.glob("*.jsonl")) if out_dir.exists() else (): for ln in f.read_text(encoding="utf-8").splitlines(): if not ln.strip(): continue try: rows.append(json.loads(ln)) except json.JSONDecodeError: # ★깨진 줄은 **버리지 않고 말한다** — 조용히 넘기면 # 「재사용했다」와 「못 읽었다」가 같아진다. print(f"★{f.name}: 못 읽는 줄이 있다 — 그 호출은 다시 산다") return rows def unique_forms(subjects) -> dict: """대상마다 **그 대상에만 있는** 표면형. ★남의 것이 근거 없이 claim 에 나타나면 미확정이다 (Codex ④). 겹치는 표기는 **빼야** 한다 — 안 빼면 정상 문장이 전부 미확정이 된다. """ forms = {s["research_subject_id"]: str(s.get("surface_form") or "") for s in subjects} out = {} for sid, f in forms.items(): if f and not any(f in other for k, other in forms.items() if k != sid): out[sid] = [f] return out def _diagnostic_one(args, doc, lock, subs, uf) -> int: """★유료 **1회만**. 잠근 subject **하나**를 **같은 프로덕션 경로**로 산다. ★어느 대상인지는 `--subject` 로 고른다(안 주면 첫 대상). 잠근 목록 밖은 못 고른다 — 승인 안 된 것을 조사하지 않는다. ★★그 결과는 **batch 후보·choice 에 안 넣는다** (Codex). known regression 진단일 뿐이고 §4b 통과·일반화 근거가 아니다. 나머지 18회는 **별도 승인** 없이는 못 산다. """ from app.core.openai_keys import openai_client want = str(getattr(args, "subject", "") or "").strip() if want: # ★잠근 목록 **안에서만** 고른다 — 밖이면 승인 범위 밖을 조사하는 것이다. picked = [s for s in subs if s["research_subject_id"] == want] if not picked: raise SystemExit( f"잠근 목록에 없는 대상이다: {want} — **선다**. " f"고를 수 있는 것: {[s['research_subject_id'] for s in subs]}") one = picked[:1] else: one = [subs[0]] n = gcs.plan_calls(one, 1, era=lock["era"], region=lock["region"]) if n != 1: raise SystemExit(f"진단은 1회여야 하는데 {n}회다 — 여기서 선다") print(f"\n★진단 1회: {one[0]['surface_form']!r} " f"({one[0]['owner_type']}, {one[0]['research_subject_id']}) · " f"팩 {gcs.PROMPT_PACK_VERSION} · strict={gcs.SCHEMA_STRICT}") args.out.mkdir(parents=True, exist_ok=True) live = args.out / "diagnostic_one.jsonl" prior = read_prior(args.out) with live.open("a", encoding="utf-8") as fh: def _append(row, _fh=fh): _fh.write(json.dumps(row, ensure_ascii=False) + "\n") _fh.flush() out = gcs.search_claims( openai_client(max_retries=gcs.SDK_RETRIES), one, model=lock["search_model"], era=lock["era"], region=lock["region"], batch_size=1, on_batch=_append, already=prior) row = out["batches"][0] shape = gcs.capture_shape(row) scored = score_run(out["batches"], unique_forms=uf) support = support_check_run(out["batches"]) result = { "mode": "diagnostic_one", "not_a_candidate": ("known regression 진단이다 — batch 후보·choice 에 " "안 넣는다. §4b 통과·일반화 근거가 아니다"), "bought_calls": out["bought_calls"], "reused_calls": out["reused_calls"], "subject": {k: one[0][k] for k in ("research_subject_id", "surface_form", "owner_type")}, "capture_shape": shape, # ★★**기계 채점**이다 — 「형식이 어긋났나」이지 「조사가 됐나」가 아니다. # 이름을 `scored` 로 두면 다음 사람이 그 수를 성과로 읽는다 (Codex). "mechanical_score": scored["counts"], "verdicts": scored["verdicts"], # ★★**연구 의미의 결론**은 프로덕션 판정기가 낸다. 빈손은 여기서 # `unresolved` 로 떨어진다 — 「못 찾음 → 차이 없음」은 계약이 금지한다. "semantic_delta": scored["semantic"], "support": support, "manifest_content_hash": doc["content_hash"], "pack_version": gcs.PROMPT_PACK_VERSION, "human_veto_required": ("이 (대상, claim) 쌍을 사람이 직접 봐야 한다. " "사람은 **거부만** 하고 통과를 만들지 않는다"), } # ★★**산 것과 결과를 먼저 남기고** 판정한다 — 실패여도 원형은 남는다. result["probe_ok"] = bool(shape["ok"]) (args.out / "diagnostic_one_result.json").write_text( json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") print(f"산 것 {out['bought_calls']}회 · 포착 {shape}") print(f"기계 채점 {scored['counts']} · 인용확인 {support['counts']}") print(f"의미 판정(프로덕션) {scored['semantic']['counts']}") print(f"★{result['not_a_candidate']}") print(f"→ {args.out}") if not shape["ok"]: # ★★★probe 요구를 어겼는데 **0 으로 끝내면 「진단 완료」로 읽힌다** # (Codex). 주소가 없거나 응답 원형을 잃은 것은 **실패**다. # ★같은 payload 재구매는 앞의 prior gate 가 막는다 — 여기서 # 다시 사지 않는다. print(f"★★probe 요구를 못 채웠다: {shape['why']} — **실패로 선다**") return 3 return 0 def preflight_new_calls(out_dir: Path, lock: dict, subs: list) -> dict: """**사기 전에** 이번 주행이 새로 살 호출 수를 센다. ★무료. ★신원 규칙은 `search_claims` 와 **같다** — `payload_hash + 모델`. 같은 payload 가 크기마다 겹치면(표본 9개의 꼬리 singleton) 그건 **한 번**만 산다. 이미 산 것도 뺀다. """ planned = [] for b in gcs.EXPERIMENT_BATCH_SIZES: planned += gcs.planned_payloads(subs, b, era=lock["era"], region=lock["region"]) unique = set(planned) model = str(lock["search_model"]) have = {str((r.get("provenance") or {}).get("payload_hash") or "") for r in read_prior(out_dir) if isinstance(r, dict) and not r.get("error") and not r.get("not_sent") and str((r.get("provenance") or {}).get("model") or "") == model} reused = unique & have return { "logical": len(planned), "unique": len(unique), "reused": len(reused), "new": len(unique - have), # ★★**이미 산 것이 실제로 재사용되나.** 전에는 「**첫** 대상의 # payload 가 있나」로 봤는데, 승인된 진단이 다른 대상이면 그 문이 # 멀쩡한 재개를 막는다(실측). 뜻은 「이 산출이 앞 주행을 잇는가」이지 # 「어느 대상이었나」가 아니다. "reused_payloads": sorted(reused), "diagnostic_reused": bool(reused), "why": (f"논리 {len(planned)} 중 고유 {len(unique)} · 앞 기록과 겹치는 것 " f"{len(reused)}. 겹치는 키는 **앞서 산 exact payload**(+모델)" f"이다. 앞 journal 이 없거나 다른 산출 디렉토리면 이미 산 것을 " f"다시 사게 된다"), } def run_experiment(args, doc, lock, subs, uf) -> int: """★유료 19 논리 호출. ★진단 1회를 이미 샀으면 **그 한 호출은 재사용**한다. 끝점에서 재려고 `main` 에서 떼어냈다 — 안에 있으면 하위 프로세스로만 잴 수 있고, 그러면 「같은 fake client 로 이어서 돌렸을 때」를 못 본다. """ # ★수는 **프로덕션 함수**에서 다시 받는다 — 손으로 넘기면 두 벌이 된다. plan = {b: gcs.plan_calls(subs, b, era=lock["era"], region=lock["region"]) for b in gcs.EXPERIMENT_BATCH_SIZES} # ★**프로덕션과 같은 문**으로 부른다 — bare `OpenAI()` 는 os.environ 만 # 읽어 .env 키도 2슬롯 failover 도 놓친다. 도구가 다른 경로로 부르면 # 그 도구는 프로덕션이 아닌 것을 잰다. from app.core.openai_keys import openai_client, slot_count # ★★**논리 호출과 물리 전송은 다르다.** 상한 19는 논리다 — SDK 재시도와 # 키 슬롯 failover 는 그 아래에서 더 보낸다 (Codex). # ①SDK 재시도를 **0으로** 끈다. 실패는 실패로 남는다 — 조용히 두 번 사지 # 않는다. # ②슬롯 failover 는 `_invoke` 안에서 도므로 여기서 못 센다. 대신 **상한을 # 적어 둔다**: 물리 ≤ 논리 × 슬롯 수. 「19회 샀다」로 쓰지 않는다. client = openai_client(max_retries=gcs.SDK_RETRIES) slots = slot_count() # ★★★**보내기 전에 신규 구매 수를 센다** (Codex). 기대값·설명이 아니라 # **문**이다. 앞 journal 이 없거나 못 읽히면 지금 코드도 16개를 새로 사서 # 승인 범위를 넘는다 — 그때 「15 이하일 것이다」라는 말은 아무것도 안 막는다. pre = preflight_new_calls(args.out, lock, subs) print(f"★preflight: 논리 {pre['logical']} · 고유 payload {pre['unique']} · " f"이미 산 것 {pre['reused']} · **신규 {pre['new']}** " f"(상한 {MAX_NEW_LOGICAL_CALLS})") if pre["new"] > MAX_NEW_LOGICAL_CALLS: raise SystemExit( f"신규 구매가 승인 범위를 넘는다: {pre['new']} > " f"{MAX_NEW_LOGICAL_CALLS} — **한 번도 사지 않고 선다**. " f"{pre['why']}") if REQUIRE_DIAGNOSTIC_REUSE and not pre["diagnostic_reused"]: raise SystemExit( "승인된 이 산출에서는 **앞서 산 것의 재사용이 필수**다. " f"앞 기록에서 재사용할 것을 못 찾았다 — " "journal 이 지워졌거나 다른 산출 디렉토리다. **선다**") # ★물리 상한은 **신규 수 × 슬롯**이다 — 논리 19 로 적으면 안 살 것까지 센다. phys_cap = pre["new"] * max(1, slots) print(f"★물리 전송 상한 = 신규 {pre['new']} × 슬롯 {slots} " f"= **{phys_cap}**. 보고에 「{MAX_LOGICAL_CALLS}회 샀다」로 쓰지 않는다") args.out.mkdir(parents=True, exist_ok=True) # ★★`batch=1` 을 **먼저** 돌려 소유권 기준선으로 쓴다 — 한 호출에 대상이 # 하나뿐이라 그 주소가 누구 것인지 **정확히** 안다 (Codex). # 기준선 없이 잰 batch>1 은 후보가 못 된다. order = sorted(gcs.EXPERIMENT_BATCH_SIZES) if order[0] != 1: raise SystemExit("batch 1 이 없다 — 소유권 기준선을 만들 수 없다") scored, spent, baseline = {}, 0, None for b in order: if spent + plan[b] > MAX_LOGICAL_CALLS: raise SystemExit(f"상한에 걸렸다 — {spent} + {plan[b]}") t0 = time.monotonic() # ★★**한 호출이 끝날 때마다 바로 디스크에 붙인다.** 전에는 그 batch # 크기가 다 끝난 뒤에야 썼다 — 9회 중 7회째에 끊기면 앞의 7회를 # 통째로 잃고 다시 사야 했다 (Codex). live = args.out / f"batch_{b}_live.jsonl" # ★★앞 주행이 끊겼으면 **이미 산 것을 다시 안 산다** (Codex). # ★진단이 남긴 줄도 **같이 읽는다** — 안 그러면 같은 payload 를 # 두 번 산다. prior = read_prior(args.out) if prior: print(f"★앞 주행에서 {len(prior)}줄 — 재사용을 시도한다") with live.open("a", encoding="utf-8") as fh: probed = {"done": False} def _append(row, _fh=fh, _p=probed): _fh.write(json.dumps(row, ensure_ascii=False) + "\n") _fh.flush() # ★★**첫 성공 호출을 모양 probe 로 쓴다.** snippet 을 못 잡으면 # `evidence_span` 을 확인할 수 없고 이 실험은 성립하지 않는다 — # **나머지를 사기 전에 선다** (Codex). 설치된 stable SDK 의 # `action.sources` 는 URL 만 주고 snippet 이 없다(실측). if _p["done"] or row.get("error") or row.get("not_sent"): return _p["done"] = True shape = gcs.capture_shape(row) print(f"★모양 probe: {shape}") if not shape["ok"]: raise SystemExit( f"첫 호출에서 {shape['why']} — 여기서 **선다**. " f"포착 모양={shape['shapes']}. 나머지를 사지 않는다") out = gcs.search_claims( client, subs, model=lock["search_model"], era=lock["era"], region=lock["region"], batch_size=b, on_batch=_append, already=prior) # ★**실제로 산 것만** 센다 — 재사용까지 세면 「19회 샀다」가 거짓이 된다. spent += out["bought_calls"] if out["logical_calls"] != plan[b]: # ★센 것과 나간 것이 다르면 **둘 중 하나가 거짓**이다. raise SystemExit( f"batch {b}: 센 것 {plan[b]} 회 vs 나간 것 " f"{out['logical_calls']} 회 — 가드가 뜻을 잃었다") # ★★**산 것을 판정보다 먼저 남긴다.** 게이트가 먼저 서서 유료 산출이 # 사라진 적이 있다. (args.out / f"batch_{b}_raw.json").write_text( json.dumps(out, ensure_ascii=False, indent=2), encoding="utf-8") if b == 1: baseline = url_baseline(out["batches"]) print(f"★소유권 기준선: 대상 {len(baseline)}개 · 주소 " f"{sum(len(v) for v in baseline.values())}개") r = score_run(out["batches"], unique_forms=uf, baseline=baseline) r["seconds"] = round(time.monotonic() - t0, 1) r["max_prompt_bytes_seen"] = out["max_prompt_bytes_seen"] r["oversized_subject_count"] = out["oversized_subject_count"] r["not_sent_count"] = out["not_sent_count"] # ★유료로 산 claim 이 **§4a 계약을 통과하나** — 채점과 다른 축이다. # 안 재면 「batch 는 깨끗한데 claim 은 하나도 못 쓴다」를 못 본다. r["contract"] = contract_survival(out["batches"]) # ★★★**인용이 그 페이지에 실제로 있나** — 여기서 **실제로 부른다**. # 안 부르면 지어낸 URL·인용이어도 choice 가 난다 (Codex). 유료 API 가 # 아니라 일반 HTTP 이고, SSRF 경계를 지난다. r["support"] = support_check_run(out["batches"]) scored[b] = r r["bought_calls"] = out["bought_calls"] r["reused_calls"] = out["reused_calls"] print(f"batch {b:2}: 산 것 {out['bought_calls']}/재사용 " f"{out['reused_calls']} · {r['counts']} · claim 계약 " f"통과 {r['contract']['passed']}/거부 {r['contract']['rejected']}" f" · {r['seconds']}초 · 최대 {r['max_prompt_bytes_seen']}바이트 · " f"후보={'예' if r['is_candidate'] else '아니오'}") # ★★★**기준선이 통째로 빈손이면 나머지를 안 산다** — 크기 비교가 # 성립하지 않는다. 결속을 잴 claim 이 하나도 없는데 2·4·8 을 사면 # 「빈손끼리 비교」가 되고, 그 수로 크기를 고르면 아무것도 아닌 것을 # 통과로 삼는다. 실측: strict 를 안 켠 진단에서 그 대상이 claim 0 이었다. # ★유료 6회를 아낀다. 산 9회는 그대로 남는다. # ★★★**기준선에 실패가 하나라도 있으면 거기서 끝낸다** (Codex). # ①실패한 행은 재사용 대상이 아니라서, 크기마다 겹치는 꼬리 # singleton 을 다음 크기가 **다시 산다** — 실측으로 preflight 15 인데 # 실제 16을 샀다. 승인 범위를 넘는 유일한 경로다. # ②그리고 출처 소유권 **기준선이 불완전**하다 — 그 상태로 잰 batch>1 # 은 어차피 후보가 못 된다. 돈과 실험 의미가 같은 방향이다. broken = [x for x in out["batches"] if x.get("error") or x.get("not_sent")] if b == gcs.EXPERIMENT_BATCH_SIZES[0] and broken: print(f"★★기준선(batch {b})에 실패한 호출이 {len(broken)}개 있다 — " f"소유권 기준선이 불완전하고, 그 행은 재사용이 안 돼 다음 " f"크기가 같은 것을 다시 산다. **나머지를 사지 않는다**") break # ★★★**§4b 가 재는 것은 batch 의 결속·누락·출처이지 A route 가 # 완결됐나가 아니다** (Codex). semantic 이 `unresolved` 여도 — # 꼭 알아야 할 구별점을 못 찾으면 계약대로 그렇게 된다 — 그 호출에 # **계약을 통과한 claim 과 확인된 인용**이 있으면 크기 비교는 성립한다. # 실측: strict 진단이 valid claim 4 · 인용 있다 3 인데도 semantic 은 # unresolved 였다. 그걸 「빈손」으로 읽으면 멀쩡한 판을 세운다. # ★semantic 은 **숨기지 않고 결과에 남긴다** — 다만 admission 을 # 막는 조건이 아니다. # ★★이 gate 는 **결과를 본 뒤에 고친 계약**이다. 그래서 앞선 # strict-false 판과 strict probe 는 development/regression 기록일 # 뿐이고, 일반화나 acceptance 독립성의 근거가 아니다 (Codex). valid = r["contract"]["passed"] found = r["support"]["counts"].get("있다", 0) if b == gcs.EXPERIMENT_BATCH_SIZES[0] and (valid < 1 or found < 1): print(f"★★기준선(batch {b})에 **잴 것이 없다** — 계약 통과 claim " f"{valid}개 · 확인된 인용 {found}개 (빈손 " f"{r['counts'][V_EMPTY]}/{len(subs)} · semantic " f"{r['semantic']['counts']}). 크기 비교가 성립하지 않는다. " f"**나머지를 사지 않는다**") break pick = choose_batch_size(scored) # ★★★**자동 결과는 여기까지다.** 결속을 기계로 다 못 닫았고(자가보고 + # 자기모순 + 남의 고유표기까지), 그 셋을 다 지나는 오결속이 실재한다 # (Codex 반례). 그래서 **기본값을 바꾸지 않는다**. pick["mechanical_candidate"] = pick.pop("chosen", None) pick["final_candidate"] = None pick["why_no_final"] = ( "자동으로는 **기계 후보**까지다. final 이 되려면 ①인용 support 에 " "`없다`·`못 열었다` 가 0 이고 ②고정 표본 전체의 (대상, claim) 쌍에 " "**사람의 거부권 검토가 끝났다**는 상태가 있어야 한다. " "★사람은 **거부만** 하고 통과를 만들지 않는다") bad_support = [b for b, r in scored.items() if r["support"]["counts"].get("없다") or r["support"]["counts"].get("못 열었다")] pick["support_blocked"] = sorted(bad_support) pick["default_change"] = "없음 — 사람 거부권 검토 전에는 기본값을 안 바꾼다" result = { "manifest_content_hash": doc["content_hash"], "sample_size": len(subs), "logical_calls_spent": spent, "logical_calls_planned": sum(plan.values()), # ★★**논리 19 ≠ 산 것 19.** 표본 9개를 2·4·8 로 자르면 **꼬리 묶음**이 # 매번 「9번째 대상 혼자」가 되어, batch 1 의 그 호출과 payload 가 # 똑같다. 같은 요청이라 다시 사지 않고 재사용한다 — 그래서 그 줄의 # 소유권은 **크기와 무관하게 subject 단위**다. 이걸 안 적으면 # 「batch 8 이 깨끗했다」를 크기 덕으로 읽는다. "reused_calls_total": sum(scored[b]["reused_calls"] for b in scored), "why_reused": ("표본 9개라 2·4·8 의 꼬리가 전부 「9번째 대상 혼자」다 — " "batch 1 의 같은 호출과 payload 가 같아 재사용된다. " "진단으로 산 행도 같은 규칙으로 쓰인다 — 키는 " "**앞서 산 exact payload**(+모델)이지 「첫 대상」이 " "아니다. 진단 대상은 `--subject` 로 고른다"), # ★물리 전송은 **못 셌다.** slot failover 가 client 안에서 돌아 도구가 # 못 본다 — 그러니 「샀다」가 아니라 **상한**으로 적는다. # ★★**승인 상한**과 **이번 주행 상한**은 다르다 (Codex). 30 은 승인 # 범위(신규 15×슬롯 2)이고, 실제로 산 것이 8이면 이번 주행의 상한은 # 16이다. 못 센 것을 「actual」이라고 쓰지 않는다. "approved_physical_cap": phys_cap, "this_run_physical_upper_bound": spent * max(1, slots), "sdk_retries": 0, "openai_slots": slots, "pack_version": gcs.PROMPT_PACK_VERSION, "by_batch": {str(b): scored[b] for b in scored}, "choice": pick, # ★한계를 결과에 **붙여 둔다** — 떼면 「실제 에피소드로 쟀다」가 된다. # ★★이 실험의 **입력 신원** — 없으면 나중에 무엇으로 쟀는지 못 되짚는다. "inputs": { "era": lock["era"], "region": lock["region"], "manuscript_hash": doc["manuscript_hash"], "a0_pack_version": doc["a0_pack_version"], "a0_model": doc.get("a0_model", ""), "search_pack_version": gcs.PROMPT_PACK_VERSION, "search_pack_manifest_hash": gcs.load_pack()["pack_manifest_hash"], "search_model": lock["search_model"], "max_prompt_bytes": gcs.MAX_PROMPT_BYTES, "hard_prompt_bytes": gcs.HARD_PROMPT_BYTES, }, # ★★출처 소유권을 **얼마나 세게** 쟀는지. batch>1 은 호출 단위까지다 — # 「남의 주소를 안 썼다」가 아니라 「호출 밖 주소를 안 썼다」. "ownership_by_batch": {str(b): scored[b]["ownership"] for b in scored}, "support_by_batch": {str(b): scored[b]["support"]["counts"] for b in scored}, "caveat": doc["sample_caveat"], # ★결과를 본 뒤에 고친 계약이 있다 — 그 사실을 산출에 **붙여 둔다**. "post_hoc_gate_note": ( "기준선 gate 를 「semantic 확정 0」에서 「계약 통과 claim ≥1 AND " "확인된 인용 ≥1」로 **결과를 본 뒤에** 고쳤다. 그래서 앞선 " "strict-false 판과 strict probe 는 development/regression 기록일 " "뿐이고, 일반화·acceptance 독립성의 근거가 아니다"), "why_not_twelve": doc["why_not_twelve"], "proves": scored[gcs.EXPERIMENT_BATCH_SIZES[0]]["proves"], } (args.out / "result.json").write_text( json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") print(f"\n기계 후보: {pick['mechanical_candidate']} — {pick['why']}") print(f"★final 후보: {pick['final_candidate']} — {pick['why_no_final']}") if pick["support_blocked"]: print(f"★인용 확인에서 막힌 크기: {pick['support_blocked']}") if pick["mechanical_candidate"] and pick["mechanical_candidate"] > 1: # ★고른 크기가 1이 아니면 **덜 쟀다**고 말한다. 안 말하면 보고가 # 증거보다 세진다. print("★그 크기의 「출처 소유권」은 **호출 단위**까지만 쟀다 — " "같은 호출 안에서 남의 주소를 쓴 것은 못 잡는다") # ★★**이번 주행 상한**을 낸다 — 승인 상한(30)을 여기 찍으면 8회를 사고도 # 「30까지 나갔다」로 읽힌다. 못 센 것을 actual 이라고 쓰지 않는다 (Codex). print(f"산 논리 호출 {spent}회 (이번 주행 물리 전송 ≤ " f"{spent * max(1, slots)} = 산 것 {spent} × 슬롯 {slots}) · " f"승인 상한은 {phys_cap} · 기록 → {args.out}") return 0 def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--root", type=Path, default=Path(__file__).resolve().parents[2]) # ★★**서로 배타다** (Codex). `--dry-run --run` 을 같이 주면 전에는 run 이 # 이겨 **실제로 샀다** — 사용자가 반대되는 두 플래그를 준 사실을 숨긴 채로. # 이제 충돌하면 provider 호출 **0 회**로 parser 가 선다. mode = ap.add_mutually_exclusive_group() mode.add_argument("--dry-run", action="store_true", help="무료 — 무엇을 얼마나 살지만 낸다 (기본값)") mode.add_argument("--run", action="store_true", help="★유료 — 19 논리 호출") mode.add_argument("--diagnostic-one", action="store_true", help="★유료 **1회만** — 잠근 subject 하나를 사고 끝낸다") ap.add_argument("--subject", default="", help="진단에 쓸 research_subject_id. 안 주면 잠근 첫 대상. " "★잠근 목록 밖이면 선다 — 승인 안 된 것을 조사하지 " "않는다") ap.add_argument("--out", type=Path, default=Path("../artifact/20260830_batch_experiment")) args = ap.parse_args() doc = load_manifest(args.root) # ★★**실행 입력은 잠근 것에서만 읽는다.** top-level 을 읽으면 잠근 사본과 # 쓰는 사본이 두 벌이 되고, 그러면 content hash 가 뜻을 잃는다 (Codex). lock = doc["locked"] subs = lock["subjects"] # ★★**보내기 전에 실제 호출 수를 센다.** `ceil(len/batch)` 로는 모자란다 — # 크기 상한에 걸리면 batch 가 더 쪼개져 **계획보다 많이 나간다**. 가드가 # 계획된 수만 보면 그 초과분을 못 막고, 이미 쓴 뒤에야 걸린다. plan = {b: gcs.plan_calls(subs, b, era=lock["era"], region=lock["region"]) for b in gcs.EXPERIMENT_BATCH_SIZES} nominal = {b: math.ceil(len(subs) / b) for b in gcs.EXPERIMENT_BATCH_SIZES} split = {b: (plan[b], nominal[b]) for b in plan if plan[b] != nominal[b]} print(f"표본 {len(subs)}개 · content_hash={doc['content_hash']}") print(f"era={lock['era']!r} region={lock['region']!r} " f"model={lock['search_model']!r}") print(f"A0 재사용: {doc['a0_reused_from']['ran_at']} " f"trace={doc['a0_reused_from']['opik_trace']}") for b, n in plan.items(): print(f" batch {b:2} → {n}회") print(f"합계 **{sum(plan.values())} 논리 호출** (상한 {MAX_LOGICAL_CALLS})") if split: # ★쪼개졌으면 **말한다.** 조용히 더 사면 「19회 샀다」가 거짓이 된다. print(f"★크기 상한에 걸려 쪼개진 것: {split} (실제, 계획)") if sum(plan.values()) > MAX_LOGICAL_CALLS: raise SystemExit( f"실제 호출 수가 상한을 넘는다: {sum(plan.values())} > " f"{MAX_LOGICAL_CALLS} — 크기 상한 분할 때문이다. 여기서 **선다**") uf = unique_forms(subs) print(f"고유 표기 대조 대상 {len(uf)}개 " f"(겹치는 표기는 뺐다 — 안 빼면 정상 문장이 전부 미확정이 된다)") if args.diagnostic_one: return _diagnostic_one(args, doc, lock, subs, uf) if not args.run: print("\n★유료 실행은 --run 이다. 지금은 아무것도 안 샀다.") return 0 return run_experiment(args, doc, lock, subs, uf) if __name__ == "__main__": raise SystemExit(main())