#!/usr/bin/env python3 """검색어 저작이 「한 대상 = 한 물음」을 지키는가 — 팩 × 모델 2×2. 무엇이 문제였나 — 완주 판(da049582/fb7a883f)의 조사 기록 6건에서 **나간 질의가 준 검색어와 1:1로 대응하지 않았다**: de840c9c 검색어 3개가 서로 다른 것 → 이어붙인 문자열 통째로 1질의 ae2d4d15 검색어 4개가 서로 다른 것 → 뭉친 2질의 b364e2f1 검색어 3개가 **같은 것** → 그대로 2질의 ← 유일하게 성립 즉 뭉침은 검색 모델의 버릇이 아니라 **검색어가 여러 대상을 담았을 때** 일어난다. 그러면 고칠 자리는 검색층이 아니라 **저작층**이다. ★수정안은 지어내지 않았다 — 같은 저장소 seed 경로의 실증된 문안을 가져왔다 (`search_grounded_ref/10.../focus_query_system.md:27-28` "Give 2 or 3 short everyday queries for that one thing alone. Never fold another thing into a query.") 그 경로는 2026-08-03 에 같은 결함(12질의 전부 뭉쳐 나감)을 이 문장으로 고쳤다 — [[feedback-search-only-what-is-not-obvious]]. ★프로덕션 경로를 그대로 쓴다 — `assess_subjects` · `build_assess_schema` · `call_structured`. 팩 디렉토리와 모델만 갈아 끼운다(각각 모듈 전역이라 호출 시점에 읽힌다). 다시 짜지 않는다. ★입력은 프로덕션이 실제로 보낸 것을 쓴다 — 샷별 `place_en` (`still_recipe_service.py:3674-3677` · `:3997-4001` 의 `cls.get("place_en")`, plate·confined 두 갈래가 같은 값을 쓴다). 출처는 `shot_ref_classify` 체크포인트이고 **두 주행 판이 다 있어** 주행 간 흔들림까지 같이 잰다. ★`era_assess::` 레코드가 입력 문자열을 안 남겨(산출 subjects 만) 재현이 이 우회로를 거쳐야 했다 — 그 자체가 고칠 결함이다. ★덤으로 캐시 신원 흔들림도 같이 잰다 — 같은 장소의 샷들이 서로 다른 `place_en` 문장을 갖고, 캐시 키 `_cache_sha(subject_text, …)` 가 그 문장을 물기 때문에 「같은 장소 1회만 지출」 계약이 성립하지 않는다. usage: ab_era_terms.py # 4-arm × 4입력 × 1회 ab_era_terms.py --rounds 3 # 흔들림까지 ab_era_terms.py --arms v1:gemini-flash,v2:gpt ab_era_terms.py --json OUT.json # 2단계(실검색)에 넘길 재료 """ import argparse import json import sys from pathlib import Path sys.path.insert(0, "/Users/manta/Documents/Projects/TheRoad-I1/scratchpad") import _opik_env # noqa: E402,F401 ★cwd 를 backend 로 고정 ROOT = Path("/Users/manta/Documents/Projects/TheRoad-I1") CP = (ROOT / "projects/da049582-2c6d-492c-979d-f468d61bab6e/checkpoints" "/episodes/fb7a883f-baac-4145-9131-732ce628d474") PACKS = { "v1": "1.202608141330", # 현행 — 「한 대상」 절 없음 "v2": "2.202608251431", # 후보① seed 경로 문안을 **검색어 칸에** 포팅 # → 실측 실패: 대상이 이미 「A 및 B」라 무효 "v3": "3.202608251500", # 후보② **대상 칸**을 고친다 — 한 대상 = 사진 # 한 장이 혼자 보여줄 수 있는 것 하나 } # 완주 판이 실제로 쓴 세계 사실 (shot_ref_classify 체크포인트의 # world_anchor_en — still_recipe_service.py:527-528 이 " — " 를 앞에 붙인다) WORLD = (" — contemporary Korean-speaking setting; " "all people are Korean unless stated") def load_places(): """샷별 place_en 을 그대로 꺼낸다 — 프로덕션이 보내는 그 문자열.""" out = [] for f in sorted((CP / "shot_ref_classify").glob("manifest*.json")): d = json.loads(f.read_text(encoding="utf-8")) edition = d.get("updated_at", "")[:16] for sid, sh in (d.get("data", {}).get("shots") or {}).items(): txt = str(sh.get("place_en") or "").strip() if not txt: continue out.append({ "edition": edition, "shot": sid, "env": sh.get("environment"), "text": txt, "sha": _era_cache_sha(txt), }) return out def _era_cache_sha(txt: str) -> str: """프로덕션 캐시 키의 대상 성분만 — 같은 장소가 몇 벌로 갈리는지 본다.""" import hashlib return hashlib.sha256(txt.strip().encode("utf-8")).hexdigest()[:8] def report_drift(places): """캐시 신원 흔들림 — 서로 다른 place_en 이 몇 개인가.""" from collections import defaultdict by_env = defaultdict(set) for p in places: by_env[p["env"]].add(p["sha"]) print("── 캐시 신원 흔들림 (place_en 문장 수 = 조사 지출 상한)") for env, shas in sorted(by_env.items()): n = sum(1 for p in places if p["env"] == env) print(f" {env:9s} 샷·판 {n:2d}건 → 서로 다른 문장 {len(shas)}개") print(f" 전체 {len(places):2d}건 → " f"{len({p['sha'] for p in places})}개\n") def main() -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--arms", default="v1:gemini-flash,v1:gpt," "v2:gemini-flash,v2:gpt") ap.add_argument("--rounds", type=int, default=1) ap.add_argument("--env", default="", help="interior/exterior 로 좁힌다") ap.add_argument("--shots", default="", help="쉼표로 샷 id 지정") ap.add_argument("--json", default="", help="산출을 이 경로에 적는다") ap.add_argument("--drift-only", action="store_true", help="캐시 신원 흔들림만 — 모델 호출 없음(무료)") args = ap.parse_args() from app.modules.pipeline import era_research as ER places = load_places() report_drift(places) if args.drift_only: return 0 want = {s.strip() for s in args.shots.split(",") if s.strip()} places = [p for p in places if (not args.env or p["env"] == args.env) and (not want or p["shot"] in want)] print(f"입력 {len(places)}개 (shot_ref_classify 두 판의 place_en)\n") rows = [] for spec in args.arms.split(","): pack_sel, model = spec.strip().split(":") pack_dir = PACKS[pack_sel] ER.resolve_era_pack = lambda selector=None, _d=pack_dir: _d ER.ASSESS_MODEL = model print(f"══ arm {pack_sel} · {model} (팩 {pack_dir})") for p in places: for r in range(1, args.rounds + 1): try: subs = ER.assess_subjects( step_tag="era_terms_probe", subject_text=p["text"], world_facts_block=WORLD) except Exception as exc: # noqa: BLE001 print(f" {p['shot']} {p['edition']} r{r} " f"✘ {type(exc).__name__}: {exc}") continue if not subs: print(f" {p['shot']} {p['edition']} r{r} → 비대상") rows.append({**p, "arm": spec, "round": r, "subject": None, "terms": []}) continue s = subs[0] terms = [str(t) for t in (s.get("search_terms_native") or [])] print(f" {p['shot']} {p['edition']} r{r} → " f"{s.get('subject_native')}") for t in terms: print(f" · {t}") rows.append({**p, "arm": spec, "round": r, "subject": s.get("subject_native"), "terms": terms, "lock": s.get("language_lock_native")}) print() if args.json: Path(args.json).write_text( json.dumps(rows, ensure_ascii=False, indent=1), encoding="utf-8") print(f"→ {args.json} ({len(rows)}행)") return 0 if __name__ == "__main__": raise SystemExit(main())