"""C(c) — **구간마다 한 번 읽는** 판독의 payload 조립과 좌표 매김. ★이 모듈은 **아무것도 사지 않는다.** provider 를 부르지 않고 payload 만 짓는다. 실제 호출은 스텝이 하고, 지금은 그 스텝이 없다. ## 왜 인용은 모델이, 좌표는 코드가 `grounding_chunk_merge._assert_rows` 는 `text[start:end] != source_quote` 면 **fail-closed** 한다. 그러면 모델에게 문자 좌표를 세라고 하는 순간, 한 글자만 어긋나도 그 행이 통째로 버려진다 — 그리고 그것은 **모델이 못 본 것**이 아니라 **셈을 틀린 것**이다. 두 실패를 한 통에 담으면 무엇을 고쳐야 할지 안 보인다. 그래서 모델은 **원문 조각**과 **그것이 몇 번째인지**만 적고, 좌표는 여기서 찾아 매긴다. 못 찾으면 그때 서는데, 그건 **지어낸 인용**이라는 뜻이라 서는 것이 맞다. ## 이 모듈이 안 하는 것 - **판정** — 등록 여부·동일성은 `grounding_chunk_merge` 가 한다 - **ID 발급 규칙** — `grounding_chunk_merge.local_id` 하나를 쓴다 - **구간 나누기 규칙** — 생산 관례(`bundles`)를 쓰고 여기서 새로 안 만든다 """ from __future__ import annotations import hashlib import json from pathlib import Path from typing import Any, Dict, List, Optional, Sequence from app.modules.pipeline import grounding_shot_catalog as _sc from app.modules.pipeline.grounding_chunk_merge import OWNERS, local_id _MODULE = "grounding_chunk" #: ★팩 버전은 **한 곳**에서. 덮어쓰지 않고 새 디렉토리로 올린다. # ★6 (2026-09-02): 장소는 씬이 벌어지는 자리마다(업소 ≠ 길) · 부분은 자기 업소에 bound_parent · # 부류·검색어는 **전체의 뼈대**(작은 부속·마모·색은 visual_brief 로) — Codex/사용자 계약. CHUNK_PACK_VERSION = "7.202609022250" #: 모델이 `location_part` 행에 싣는 **맥락 칸**. ★이름은 한 곳에서만 정한다. _HOST_CONTEXT_KEY = "host_context" #: ★★**후처리 계약** 버전. 팩(획득)과 **따로** 둔다 — 실데이터를 본 뒤 파서를 #: 고친 것이라, 이것을 획득 신원·잠금에 접으면 **이미 산 것을 다시 사야 한다**. #: 대신 파생 장부·채점에 남겨 「무엇으로 해석했는지」를 되짚게 한다 (Codex). # ★★해석 계약. 2026-09-01 — `host_context` 를 행에 싣고 # `grounding_chunk_merge._settle_host_context` 가 계약대로 내리기 시작했다. # **읽는 법이 바뀌면 여기가 움직여야** 같은 raw 를 새 코드로 다시 읽은 # 산출이 옛것과 구별된다 (사는 것은 pack 판이 지배한다 — 둘은 다른 축). PROCESSING_CONTRACT_VERSION = "4.202609022340" #: 호출에 붙는 schema 이름. ★**여기가 한 곳**이다 — 스텝과 감사 도구가 각자 #: 지으면 Opik 에서 같은 호출이 두 이름으로 보인다. CHUNK_SCHEMA_NAME = "grounding_chunk_read" MERGE_SCHEMA_NAME = "grounding_chunk_merge" #: 격리 사유. Q_MENTION = "mention_not_in_source" Q_EVIDENCE = "evidence_not_in_source" Q_SHOT = "shot_binding_invalid" Q_EVIDENCE_SCENE = "evidence_outside_mention_scenes" _PACK_ROOT = (Path(__file__).resolve().parents[4] / "prompts" / "_base" / _MODULE) def pack_dir(version: Optional[str] = None) -> Path: """팩 디렉토리. ★없으면 **선다** — 빈 프롬프트로 사는 길을 안 둔다.""" d = _PACK_ROOT / (version or CHUNK_PACK_VERSION) if not d.is_dir(): raise FileNotFoundError(f"팩이 없다: {d}") return d def load_text(name: str, version: Optional[str] = None) -> str: p = pack_dir(version) / name if not p.is_file(): raise FileNotFoundError(f"팩에 {name} 이 없다: {p}") t = p.read_text(encoding="utf-8").strip() if not t: raise ValueError(f"{p} 가 비었다 — 빈 지시문으로 부르지 않는다") return t def load_schema(name: str, version: Optional[str] = None) -> Dict[str, Any]: return json.loads((pack_dir(version) / name).read_text(encoding="utf-8")) def acquisition_identity(payload: Dict[str, Any], *, model_alias: str, model_physical: str, request_contract: Dict[str, Any]) -> str: """**나가는 것**의 신원. ★이것이 바뀌면 **다시 사야 한다**. 접는 것은 실제로 provider 로 나가는 것뿐이다 — system · user/parts · schema · 요청 모델 좌표 · 요청 계약. ★원문·세계 사실·샷 전문·샷 ID 는 **payload 안에** 이미 들어 있으므로 그것이 바뀌면 여기가 움직인다. 따로 세지 않는다(두 벌이 된다). ★★**후처리 계약을 여기 섞지 않는다** (Codex 2026-08-31). 섞으면 파서를 고친 것만으로 **이미 산 것을 다시 사게** 된다. 후처리는 아래 `processing_stamp` 이 나른다. """ h = hashlib.sha256() for part in (str(payload.get("system") or ""), # ★**parts 전체**를 접는다 (Codex NON-BLOCK). 지금은 text # 하나뿐이지만, 「나가는 것 그 자체」가 신원이라면 첫 칸만 # 보면 안 된다 — 뒤에 무엇을 더해도 안 움직인다. json.dumps(payload.get("parts") or [], sort_keys=True, ensure_ascii=False), json.dumps(payload.get("schema") or {}, sort_keys=True, ensure_ascii=False), str(model_alias), str(model_physical), json.dumps(request_contract or {}, sort_keys=True, ensure_ascii=False)): h.update(part.encode("utf-8")) h.update(b"\x00") return h.hexdigest()[:24] def processing_stamp(acquisition: str, contract: Optional[str] = None) -> str: """**해석**의 지문. ★이것만 바뀌면 **재구매 0** — 저장 응답을 다시 푼다. 하류(스텝·재개·config)는 이 값을 보고 stale 을 판단한다. 획득 신원이 같으면 살 필요가 없고, 후처리가 바뀌었으니 **다시 풀기만** 하면 된다. """ h = hashlib.sha256() h.update(acquisition.encode("utf-8")) h.update(b"\x00") h.update((contract or PROCESSING_CONTRACT_VERSION).encode("utf-8")) return h.hexdigest()[:24] def build_chunk_payload(segment_ids: Sequence[str], segments: Dict[str, str], world_facts: str, *, version: Optional[str] = None, shot_catalog: Optional[Sequence[Dict[str, Any]]] = None ) -> Dict[str, Any]: """한 구간(여러 씬일 수 있다)의 호출 payload. ★**원문을 자르지 않는다.** 구간에 속한 씬의 본문을 통째로 넣는다. Returns: `{"system": str, "parts": [...], "schema": {...}}` — 실제 전송은 호출부가 한다. 이 함수는 **보내지 않는다.** """ if not segment_ids: raise ValueError("구간에 씬이 하나도 없다") missing = [s for s in segment_ids if s not in segments] if missing: raise KeyError(f"모르는 segment {missing}") body = "\n\n".join( f"[{sid}]\n{segments[sid]}" for sid in segment_ids) text = ("SEGMENTS IN THIS CHUNK:\n" + body + "\n\nWORLD FACTS (region and era — authoritative):\n" + (world_facts or "").strip()) schema = load_schema("chunk_schema.json", version) if shot_catalog is not None: # ★샷 목록은 **실제 체크포인트에서** 온다. 설명을 자르지 않는다. text += ("\n\nSHOTS IN THIS CHUNK (choose by id):\n" + json.dumps(list(shot_catalog), ensure_ascii=False, indent=1)) schema = _sc.patch_schema_with_shot_ids(schema, shot_catalog) return { "system": load_text("system.md", version), "parts": [{"type": "text", "text": text}], "schema": schema, } def resolve_rows(model_rows: Sequence[Dict[str, Any]], *, chunk_id: str, segment_ids: Sequence[str], segments: Dict[str, str], shot_catalog: Optional[Sequence[Dict[str, Any]]] = None ) -> Dict[str, Any]: """모델이 낸 행에 **local_id 와 정본 span** 을 매긴다. 모델은 `mentions`(이름 덩어리 + 몇 번째)와 `evidence_quotes`(겉모습 문장)를 따로 냈다. 여기서 **언급만** 좌표로 바꿔 `occurrences` 로 삼고, 근거는 따로 실어 **등장 횟수에 안 센다**. ★**어느 씬에 있는지도 여기서 찾는다.** 모델에게 물으면 틀릴 수 있고, 틀린 것을 우리가 못 가른다 — 원문에 있으면 찾아지고 없으면 안 찾아진다. ★조각이 여러 씬에 **같이** 있으면 `occurrence_index` 를 **구간 전체**에서 센다 — 씬 순서대로 이어 센다. 모델이 본 것도 이어진 본문이다. Returns: `{"rows": [...], "quarantined": [...], "processing_contract": str}`. ★검증된 언급이 **하나도** 없거나 근거 문장을 지어낸 행만 **행째로** `quarantined` 로 간다 (2026-09-02 밤). 일부만 어긋난 행은 검증된 자리·검증된 씬의 근거·검증된 샷 ID 만 남기고 뺀 것을 `salvage_problems` 에 적은 채 산다. 격리 행은 자동 merge· 출현 수·등록·최종 ID·자동 합격에 **하나도** 기여하지 않는다. Raises: ValueError: owner 가 다섯 갈래 밖이거나 `mentions` 가 아예 없을 때. """ out: List[Dict[str, Any]] = [] quarantined: List[Dict[str, Any]] = [] for i, r in enumerate(model_rows or ()): lid = local_id(chunk_id, i) owner = str((r or {}).get("owner_type") or "") if owner not in OWNERS: raise ValueError(f"자리 {i}: 모르는 owner {owner!r} — {OWNERS}") mentions = (r or {}).get("mentions") or [] if not mentions: raise ValueError(f"자리 {i}: `mentions` 가 비었다") # ★★★**검증된 것만 남기고, 뺀 것을 적는다** (2026-09-02 실측으로 뒤집음). # 앞 판(Codex 2026-08-31)은 인용 하나가 안 맞으면 **행째** 격리했다 — # 「틀린 것만 떼면 서로 다른 대상을 합친 오염이 깨끗한 척 통과한다」는 # 이유였다. 실측은 달랐다: 두 유료 주행에서 22행 중 9행 · 29행 중 12행이 # 「언급 번호 하나 초과」(이발사 3번째 — 2번뿐) · 「씬 밖 샷 결속 하나」로 # 통째로 사라져 **장소 셋·주인공·부분 둘**이 등록 자체를 못 했고, 그 빈자리를 # 남은 장소 하나가 덮어 「장소 신원 붕괴」로 오진됐다. 지어낸 인용은 여전히 # **한 자리도 남지 않는다** — 남는 것은 원문에서 찾은 자리뿐이고, 뺀 것은 # `salvage_problems` 에 사유째 남아 감사가 본다. 검증된 언급이 **하나도** # 없거나 근거 문장을 지어냈으면 지금처럼 행째 격리다. bad: List[Dict[str, Any]] = [] fatal: List[Dict[str, Any]] = [] occ_out: List[Dict[str, Any]] = [] for j, m in enumerate(mentions): quote = str((m or {}).get("mention_quote") or "") idx = (m or {}).get("occurrence_index") if not quote: bad.append({"kind": Q_MENTION, "at": j, "quote": quote, "occurrence_index": idx, "why": "이름 덩어리가 비었다"}) continue if not isinstance(idx, int) or isinstance(idx, bool) or idx < 1: bad.append({"kind": Q_MENTION, "at": j, "quote": quote, "occurrence_index": idx, "why": "`occurrence_index` 가 1 이상 int 가 아니다"}) continue try: occ_out.append({ "source_span": _find_span(quote, idx, segment_ids, segments), "source_quote": quote, }) except ValueError as exc: bad.append({"kind": Q_MENTION, "at": j, "quote": quote, "occurrence_index": idx, "why": str(exc)}) ev = [] verified_scenes = {o["source_span"]["segment_id"] for o in occ_out} for j, q in enumerate((r or {}).get("evidence_quotes") or []): q = str(q or "") where = [sid for sid in segment_ids if q and q in segments[sid]] if not where: # ★근거가 틀린 행은 **두 축과 의무 판정**이 살아남으면 안 된다 # — 그래서 이 행은 여전히 **행째** 격리한다 (Codex). fatal.append({"kind": Q_EVIDENCE, "at": j, "quote": q, "why": "이 구간 원문에 없다 — 지어낸 근거다"}) continue if verified_scenes and not (set(where) & verified_scenes): # ★★근거가 **다른 씬**에만 있다 (Codex BLOCK 2026-09-02 밤): 검증된 언급은 # 씬 1 인데 근거 문장은 씬 2 것이면, 그 문장은 이 행의 것이 아닐 수 있다 # (다른 자리의 겉모습이 이 행의 payload 에 붙는다). 좌표로만 가른다 — # 그 문장이 이 행의 검증된 언급 씬 어디에도 없으면 **그 문장만** 뺀다. # ★행째 격리는 안 한다: 실측(골목)에서 판독기가 씬 4 언급을 안 적고 씬 4 # 근거만 적는 **덜 적음**이 실제로 있었다 — 행을 죽이면 그 자리가 빈다. bad.append({"kind": Q_EVIDENCE_SCENE, "at": j, "quote": q, "where": where, "why": (f"근거는 {where} 에 있는데 이 행의 검증된 언급은 " f"{sorted(verified_scenes)} 에만 있다 — 다른 씬의 " f"근거를 이 행에 붙이지 않는다")}) continue ev.append(q) # ★★샷 결속 — 상태 **셋**을 그대로 나른다. 나체 빈 배열은 금지. # 씬 밖 결속·catalog 밖 ID·중복은 **그 ID 만** 사유째 빠지고 검증된 ID 는 남는다. bind_state, bind_ids = _sc.BIND_UNRESOLVED, [] if shot_catalog is not None: row_scenes = [o["source_span"]["segment_id"] for o in occ_out] bind_state, bind_ids, why = _sc.verify_binding( (r or {}).get("shot_binding_status"), (r or {}).get("shot_appearance_ids"), shot_catalog, row_scenes) if why: bad.append({"kind": Q_SHOT, "at": -1, "quote": str((r or {}).get("shot_appearance_ids")), "kept": list(bind_ids), "why": why}) if not occ_out: fatal.append({"kind": Q_MENTION, "at": -1, "quote": "", "why": f"검증된 언급이 하나도 없다 ({len(bad)}개 전부 원문 밖)"}) if fatal: quarantined.append({ "chunk_id": chunk_id, "row_index": i, "local_id": lid, "owner_type": owner, "surface_form": str((r or {}).get("surface_form") or ""), "problems": fatal + bad, "raw_mentions": mentions, "raw_evidence_quotes": list( (r or {}).get("evidence_quotes") or []), "processing_contract": PROCESSING_CONTRACT_VERSION, }) continue hard = (r or {}).get("hard_to_generate") notice = (r or {}).get("viewers_would_notice") # ★★**의무는 같은 행에서 둘 다 참일 때만 선다** (사용자 확정 두 축). obliged = (hard is True and notice is True) out.append({ "local_id": lid, "owner_type": owner, "surface_form": str((r or {}).get("surface_form") or ""), "occurrences": occ_out, "evidence_quotes": ev, "shot_binding_status": bind_state, "shot_appearance_ids": bind_ids, "hard_to_generate": hard, "viewers_would_notice": notice, # ★뺀 것이 있으면 사유째 싣는다 — 「다 맞았다」와 「일부 뺐다」를 가른다 **({"salvage_problems": bad} if bad else {}), "visual_brief": str((r or {}).get("visual_brief") or ""), # ★사진을 고를 때 **이것만** 준다 — 시대·형태를 VLM 에게 안 묻는다 "coarse_type_label": str((r or {}).get("coarse_type_label") or ""), "search_terms_native": ( list((r or {}).get("search_terms_native") or []) if obliged else []), "language_lock_native": ( str((r or {}).get("language_lock_native") or "") if obliged else ""), # ★★그 부분이 **어디에 달렸나** — `location_part` 만 갖는다. # 여기서 판정하지 않는다. `grounding_host_context.normalize` 가 # 계약대로 내리고, 어기면 `unresolved` 로 떨어뜨린다. # ★없으면 칸도 안 만든다 — 「없다」와 「칸이 없다」를 가른다. **({_HOST_CONTEXT_KEY: (r or {})[_HOST_CONTEXT_KEY]} if isinstance((r or {}).get(_HOST_CONTEXT_KEY), dict) else {}), }) return {"rows": out, "quarantined": quarantined, "processing_contract": PROCESSING_CONTRACT_VERSION} def _find_span(quote: str, index: int, segment_ids: Sequence[str], segments: Dict[str, str]) -> Dict[str, Any]: """구간 전체에서 `index` 번째 `quote` 를 찾아 씬 좌표로 돌려준다.""" seen = 0 for sid in segment_ids: text = segments[sid] pos = -1 while True: pos = text.find(quote, pos + 1) if pos < 0: break seen += 1 if seen == index: return {"segment_id": sid, "start": pos, "end": pos + len(quote)} raise ValueError( f"인용 {quote!r} 의 {index}번째가 구간 {list(segment_ids)} 에 없다 " f"— 지어낸 인용이거나 다듬어 적은 것이다 (구간 전체에서 {seen}번 나옴)") def build_merge_payload(rows: Sequence[Dict[str, Any]], *, version: Optional[str] = None) -> Dict[str, Any]: """동일성 판정 payload. ★**원문을 안 보낸다** — compact 행만. C(c) 의 이득이 여기서 난다. merge 가 원문을 다시 읽으면 「원문이 한 번」이 깨진다. """ compact = [{ "local_id": r.get("local_id"), "owner_type": r.get("owner_type"), "surface_form": r.get("surface_form"), "visual_brief": r.get("visual_brief"), "coarse_type_label": r.get("coarse_type_label"), "occurrence_count": len(r.get("occurrences") or []), # ★★**근거를 같이 준다** (Codex 2026-08-31). 앞 판은 이름과 설명만 # 보냈는데, 그러면 **같은 말이 서로 다른 것을 가리키는 경우**를 # 모델이 구조적으로 못 가른다 — 가를 정보가 payload 에 없다. # 없는 정보로 판정하라고 하면 그건 짐작을 시키는 것이고, # C(c) 가 줄이려던 바로 그것이 는다. # ★원문 **전문**은 여전히 안 보낸다. 인용 조각과 어느 씬인지만이다. # ★언급(이름 덩어리)과 근거(겉모습 문장)를 **갈라서** 준다. # 이름이 같아도 근거가 다르면 다른 실물일 수 있고, 이름이 달라도 # 근거가 같으면 같은 실물일 수 있다 — 둘 다 있어야 가른다. "mentions": [{"segment_id": o["source_span"]["segment_id"], "quote": o["source_quote"]} for o in (r.get("occurrences") or [])], "evidence": list(r.get("evidence_quotes") or []), } for r in rows] return { "system": load_text("merge_system.md", version), "parts": [{"type": "text", "text": "ROWS:\n" + json.dumps( compact, ensure_ascii=False, indent=1)}], "schema": load_schema("merge_schema.json", version), }