"""요소 Phase StepRunner — entity_extract, entity_review, entity_detail, entity_t2i.""" import copy import json import logging import time from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from typing import Any, Dict, List, Optional from app.core.prompt_assembly import assemble_user_prompt from app.core.step_runner import StepRunner from app.modules.pipeline import grounding_binding as _gb from app.modules.llm.llm_client import call_structured from app.modules.prompt_loader import load_prompt from app.modules.pipeline import episode_carry as _ec logger = logging.getLogger(__name__) def _traits_with_scale(src: Dict[str, Any]) -> List[str]: """`visual_traits` + 채택한 크기 견줌 — 중복은 넣지 않는다.""" base = src.get("visual_traits") out = list(base) if isinstance(base, list) else [] for line in scale_trait_sentences(src.get("build_scale")): if line and line not in out: out.append(line) return out def scale_trait_sentences( build_scale: Optional[List[Dict[str, Any]]], ) -> List[str]: """채택한 견줌을 **결정적 문장**으로 — `visual_traits` 에 실을 형태. ★왜 `visual_traits` 인가 (2026-09-20 Codex ㉢): 샷 문안은 `stable_traits` 를 읽고, 그것은 `visual_traits` 에서 온다 (`checkpoint_sync/entity_sync_service`). T2I context 에만 넣으면 **참조 그림 한 장까지만** 닿고 샷마다 크기가 유지되지 않는다. 새 DB 칼럼을 만들지 않고 **기존 배선**을 쓰는 최소안이다. ★의미를 더하지 않는다 — 모델이 낸 `compared_to`·`relation` 을 정해진 꼴로 잇기만 한다. `metadata_json` 은 쓸 수 없다(`entity_metadata` 가 character 를 닫힌 모양으로 접어 버린다). ★원자료(`build_scale`)는 따로 보존한다 — 근거와 렌더 문장은 다르다. """ woven, _ = _scale_rows(build_scale) return [ f"{d.get('compared_to', '')}과(와) 견주어 {d.get('relation', '')}" for d in woven ] def _scale_rows( build_scale: Optional[List[Dict[str, Any]]], ) -> tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: """근거 있는 견줌만 고른다 — `distinctive` 와 **같은 구조 게이트**. ★`source_quote` 가 비어 있지 않은지만 본다. 「원문에 그 말이 있는지」를 대조한 것이 아니다 — 그 표현으로 넓히지 않는다. """ woven: List[Dict[str, Any]] = [] excluded: List[Dict[str, Any]] = [] for d in (build_scale or []): if not isinstance(d, dict): continue ok = (bool((d.get("source_quote") or "").strip()) and bool((d.get("relation") or "").strip())) (woven if ok else excluded).append(d) return woven, excluded def build_build_scale_block( build_scale: Optional[List[Dict[str, Any]]], ) -> tuple[str, List[Dict[str, Any]]]: """크기(키·덩치) **견줌** 블록 — 2026-09-20 사용자 지시. > "대략적으로 비교될만한 형태로 묘사하는 정도로… 표준 크기 및 덩치 > 등으로 비교하기 쉽게… 또한 프롬프트에서 비교적인 부분은 필요할듯해" ## 왜 있나 정본에 **인물 크기가 아예 없었다.** 그래서 키가 성인 남자를 넘지 않는 로봇이 화면에서 3m 병기로 그려졌고, 사람이 정본을 손으로 고쳐야 했다. ★막고 있던 것은 추출 팩의 「**바디 묘사 절대 금지**」였다. 그 금지는 몸매·체형 묘사를 막으려던 것인데 크기까지 같이 지웠다. 새 팩(v19)은 `build_scale` 을 **구조 칸**으로 받고 그 금지와 다른 축임을 명시한다. ## 게이트 — `distinctive` 와 **같은 구조 신호** `source_quote` 가 비어 있지 않은 항목만 채택한다(본문·기획서 직접 근거). 글자로 뜻을 재지 않는다 — 이 저장소 금지 규칙. ★**수치가 아니라 견줌**을 싣는다. T2I 는 센티미터를 안 받는다(실측) — 「보통 성인보다 머리 하나 작다」는 그릴 수 있다. Returns: (block, excluded) — `block` 은 t2i context 에 붙일 문자열(채택 항목이 없으면 빈 문자열), `excluded` 는 근거 없이 들어온 항목. """ woven, excluded = _scale_rows(build_scale) if not woven: return "", excluded lines = "\n".join( f"- {d.get('compared_to', '')} 과(와) 견주어: {d.get('relation', '')}" f" (원문 인용: {d.get('source_quote', '')})" for d in woven ) return ( "\n\n[크기 — 견줌으로]\n" "아래는 이 요소의 크기를 다른 것과 견준 것이다. 수치가 아니라 " "견줌이므로 **그 견줌이 한눈에 보이게** 그린다.\n" + lines ), excluded def build_distinctive_trait_block( distinctive_visual_traits: Optional[List[Dict[str, Any]]], ) -> tuple[str, List[Dict[str, Any]]]: """② P0a generic fix (2026-06-18) — evidence-gated distinctive 외형 블록. entity_detail 의 ``distinctive_visual_traits`` 중 **렌더용으로 채택할 항목만** 골라 entity_t2i context 로 넘길 별도 블록 문자열을 만든다. 채택 조건은 두 개의 **구조적 신호**다 (trait 텍스트를 글자/regex/substring 으로 의미 판정하지 않음 — feedback-no-literal-substring-meaning. literal/metaphor 판단은 LLM 이 ``interpretation_kind`` 로 내려준 구조화 결과를 코드가 그대로 신뢰한다): 1. ``source_quote`` 가 비어 있지 않다 (본문 직접 근거 존재). 2. ``interpretation_kind == "literal_visual"`` (신체 부위의 색/형태/표식을 평이하게 사실로 진술한 경우. metaphorical_impression / ambiguous 는 제외). 이로써 (a) 본문 근거 없는 invention(빈 quote), (b) 근거는 있으나 시적·인상적 비유(예: '금빛 물결이 일렁이는 눈동자')는 둘 다 t2i_prompt(ref/scene SOT)로 승격되지 않는다. 반대로 평이한 색 진술(예: '붉은 눈동자')은 보존된다. Returns: (block, excluded): ``block`` 은 t2i context 에 append 할 문자열(채택 항목이 없으면 빈 문자열), ``excluded`` 는 채택되지 않은 distinctive 항목 dict 목록 (각 trait/interpretation_kind/interpretation_reason 포함 — '근거 있는데 왜 빠졌나' 추적용 diagnostic). """ dvts = distinctive_visual_traits or [] woven: List[Dict[str, Any]] = [] excluded: List[Dict[str, Any]] = [] for d in dvts: if not isinstance(d, dict): continue has_quote = bool((d.get("source_quote") or "").strip()) is_literal = d.get("interpretation_kind") == "literal_visual" (woven if (has_quote and is_literal) else excluded).append(d) if not woven: return "", excluded lines = "\n".join( f"- {d.get('trait', '')} (원문 인용: {d.get('source_quote', '')}" + ( f"; 연결근거: {d['attribution_note']}" if (d.get("attribution_note") or "").strip() else "" ) + ")" for d in woven ) return f"\n\n[원문 인용으로 확인된 문자적 특이 외형 — 반드시 반영]\n{lines}", excluded from app.modules.pipeline import grounding_entity_sync_ext as _sync_ext # noqa: E402 def _ENTITY_KEYS(): """갈래 표 — **계약 한 곳**에서 온다. ★여기서 다시 적지 않는다.""" from app.modules.pipeline.grounding_carry import ENTITY_KEY_TO_OWNER return ENTITY_KEY_TO_OWNER #: ★★★인스턴스 신원 계약 (Codex BLOCK 2026-09-03 06:40 · 실측 f7cc45c576c0): producer(entity_merge)가 short_id 를 준 행의 #: 신원은 entity_detail → entity_t2i → sync 끝까지 **short_id** 다. (name, type) 은 legacy(short_id 없는 옛 CP)에만 쓴다. #: 실측: 같은 이름 「흙바닥」 location_part 둘(LP05→L03 · LP07→L02)을 (name, type) 큐가 하나로 접어 LP07 canon 이 안 생겼고 #: part_of 동기화가 fail-closed 로 섰다. 두 스텝의 config hash 에 실려 옛 CP 를 다시 굽는다. ENTITY_INSTANCE_IDENTITY_CONTRACT_VERSION = "2.202609030640" def _ikey(name: str, etype: str, sid: str = "") -> str: """인스턴스 열쇠 — short_id 가 있으면 그것, 없으면(legacy) `_qkey`.""" return str(sid) if sid else _qkey(name, etype) def _identity_hash(base: str) -> str: import hashlib return hashlib.sha256(f"{base}|{ENTITY_INSTANCE_IDENTITY_CONTRACT_VERSION}".encode("utf-8")).hexdigest() def _qkey(name: str, etype: str) -> str: """이 스텝의 **열쇠**. ★이름 하나로는 갈래가 갈리지 않는다. 같은 원문 낱말이 `location` 과 `location_part` 로 **둘 다** 잡힐 수 있다 — 「이발소」와 「이발소」(그 안의 고정 설비). 이름만으로 열쇠를 쓰면 하나가 다른 하나를 덮거나 건너뛴다. """ return f"{etype}\u0000{name}" def _VALIDATED_TYPES() -> frozenset: """산출 모양을 **검증하는** 갈래. ★검증기가 아는 것에서 파생한다. ★★`location_part` 가 빠져 있었다 — 그 갈래만 검증 없이 sync 까지 갔다 (Codex 재현 2026-09-01). 손으로 적으면 갈래가 늘 때 또 빠진다. """ from app.core.entity_metadata import validated_entity_types return validated_entity_types() def _carry_chunk_marker(runner) -> Dict[str, Any]: """앞 CP 의 **새 모양 표식**을 그대로 나른다. ★없으면 안 만든다. ★sync(`entity_sync_service`)가 읽는 것은 `entity_t2i` CP 다. 표식이 거기 없으면 `owner_keys` 가 옛 셋만 내고 **`location_part` 행이 조용히 버려진다** (Codex 재현 2026-09-01). """ from app.core.grounding_mode import (resolve_grounding_mode, uses_chunk_producer) from app.modules.pipeline import grounding_entity_sync_ext as _x chunk = uses_chunk_producer(resolve_grounding_mode( getattr(runner, "project_config", None))) try: prev = runner._load_prev_checkpoint("entity_merge") or {} except Exception as exc: # noqa: BLE001 if chunk: from app.core.errors import AppError raise AppError( code="entity_t2i.marker_unreadable", message=f"C(c) 판인데 앞 CP 를 못 읽었다 ({exc})", status_code=400) from exc return {} got = prev.get(_x.CHUNK_SCHEMA_MARKER) if chunk and got != _x.CHUNK_SCHEMA_VERSION: # ★★★C(c) 판에서 표식이 없거나 틀리면 **선다**. 조용히 legacy 로 # 접으면 저장·직렬화 한 칸이 빠져도 C/L/P 만 성공하고 **LP 만 # 소리 없이 사라진다** (Codex 2026-09-01). from app.core.errors import AppError raise AppError( code="entity_t2i.marker_missing", message=(f"C(c) 판인데 앞 CP 의 `{_x.CHUNK_SCHEMA_MARKER}` 가 " f"{got!r} 다 — 아는 판은 " f"{_x.CHUNK_SCHEMA_VERSION!r} 뿐이다. 이것을 legacy 로 " "접으면 location_part 가 조용히 사라진다"), status_code=400) return {_x.CHUNK_SCHEMA_MARKER: got} if got else {} def _by_lane(done: Dict[str, Any]) -> Dict[str, Any]: """`entity_type` 별로 나눈다. ★**있는 갈래만** 칸을 만든다. ★★빈 칸을 늘 만들면 legacy 체크포인트의 **모양이 바뀐다**. 지금 갈래가 셋뿐인 주행은 앞과 **한 글자도 안 달라야** 한다 (Codex D-inert · 09-01). ★기존 세 칸은 **비어도 만든다** — 옛 소비자가 그 칸을 그냥 읽는다. """ always = {"characters", "locations", "props"} out: Dict[str, Any] = {} for key, owner in _ENTITY_KEYS(): got = [v for v in done.values() if v.get("entity_type") == owner] if got or key in always: out[key] = got return out class _EntityStepMixin: """요소 단계 공통 — fulltext 로드, 이전 단계 결과 로드.""" def _load_fulltext(self) -> str: from app.models.project import Episode from sqlalchemy.orm import undefer ep = ( self.db.query(Episode) .options(undefer(Episode.fulltext)) .filter(Episode.id == self.episode_id) .first() ) if not ep or not ep.fulltext: from app.core.errors import AppError raise AppError( code="step.no_fulltext", message="시나리오 텍스트가 없습니다.", status_code=400, ) return ep.fulltext def _a0_candidates(self): """A0 가 건진 후보. ★없으면 ``None`` — legacy 가 한 글자도 안 달라진다. ★``shadow_plan`` 에서도 ``None`` 이다. A0 는 돌지만 **나가는 프롬프트에 안 실린다** — 관측이 하류를 바꾸면 shadow 가 아니다(계약 §12). 후보를 프롬프트에 넣으면 추출 산출이 달라지고, 그러면 「shadow 를 켠 것만으로 하류가 stale 되지 않는다」가 거짓이 된다. """ from app.core.grounding_mode import buys_v2_research, resolve_grounding_mode if not buys_v2_research(resolve_grounding_mode(self.project_config)): return None cp = self._load_prev_checkpoint("grounding_a0") cands = ((cp or {}).get("data") or {}).get("candidates") or [] return cands or None def _config_hash(self) -> str: """★★결속 계약·팩 bytes 를 지문에 접는다 (Codex BLOCK-2). 안 접으면 이미 완료된 `entity_all`/`entity_extract`/`entity_merge` 체크포인트가 그대로 SKIP 되어 **새 ID 칸이 영영 안 생긴다** — 배포해도 고친 것이 아무 데도 안 닿는 부류다. ★후보가 없는 판(legacy 포함)은 **옛 값 그대로** 돌려준다. 그래야 legacy 체크포인트가 안 깨진다. ★저장과 비교가 **이 함수 하나**를 봐야 한다. `_execute` 가 이 값을 `data["config_hash"]` 로 안 돌려주면, 저장은 project_config 만 보고 비교는 이것을 봐서 **매번 어긋난다**. """ from app.core.step_runner import compute_config_hash base = compute_config_hash(self.project_config) extra = {} if self._a0_candidates(): from app.modules.pipeline import grounding_binding as _gb extra.update(_gb.pack_fingerprint()) # ★★앞 화 명부가 **실제로 붙는 판**에서만 접는다 (2026-09-04). # 첫 화는 명부가 비어 블록도 스키마 패치도 안 붙으므로 지문이 # 안 움직여야 한다 — 움직이면 「아무것도 안 바뀐다」가 거짓이 되고 # 멀쩡한 옛 체크포인트를 전부 다시 태운다. # ★반대로 명부가 붙는 판에서 이걸 안 접으면, 계약이나 팩을 고쳐도 # resume 이 옛 산출을 그대로 쓴다 — 고친 것이 안 도는 부류다. # ★★팩 버전만이 아니라 **명부 내용**의 지문도 접는다. 이름·별명· # 앵커가 바뀌어도 안 움직이면 옛 체크포인트가 그대로 재사용된다. # ★명부를 실는 스텝은 `entity_all_*` 셋뿐이다. 다른 스텝(및 이 mixin # 을 단독으로 쓰는 시험)은 `step_id` 가 없을 수 있으니 물어만 본다. _owner = {"entity_all_character": "character", "entity_all_location": "location", "entity_all_prop": "prop"}.get( getattr(self, "step_id", "")) _digest = (_ec.roster_digest(self.db, self.project_id, _owner, self.episode_id) if _owner else "") if _digest: extra.update(_ec.pack_fingerprint()) extra["carry_roster"] = _digest if not extra: return base import hashlib import json raw = json.dumps({"base": base, **extra}, sort_keys=True, ensure_ascii=False) return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16] def _load_cleaned_text(self) -> str: cp = self._load_prev_checkpoint("text_cleanup") if cp and cp.get("data", {}).get("cleaned_text"): return cp["data"]["cleaned_text"] return self._load_fulltext() def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict]: from app.core.config import settings cp = ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" ) if cp.exists(): return json.loads(cp.read_text(encoding="utf-8")) return None # ── entity_all: 리스팅 단계 (이름 + 출현 씬 + short_id 부여) ── def _prefix_for(owner: str) -> str: """갈래 → `short_id` 접두. ★**계약 모듈 한 곳**에서 온다. `"C"`·`"L"`·`"P"` 를 호출부에 박아 두면 갈래가 늘어도 여기는 모른다. """ from app.modules.pipeline.grounding_entity_contract import OWNER_PREFIX return OWNER_PREFIX[owner] def _prior_roster(runner, owner: str, mode: str = "resume"): """앞 화 명부 `(블록, 허용 ID)`. ★첫 화는 `("", [])` 라 바이트 불변. ★세 갈래가 **같은 함수**를 쓴다 — 손으로 세 번 적으면 한 곳이 빠진다. 실제로 그 부류의 결함이 있었다(`binding_out` 이 한 갈래에만 빠져 `NameError` 였다). """ from app.modules.pipeline.episode_carry import build_roster_block # ★화 범위를 준다 — 프로젝트 전체를 실으면 1화 재분석에 2·3화 신원이 # **과거로 샌다** (Codex BLOCK 2026-09-04). # ★`force` 일 때만 스냅샷을 다시 뜬다. resume 은 **얼린 것**을 봐야 # 이 화가 제 canon 을 만든 뒤에도 지문이 안 흔들린다. return build_roster_block(runner.db, runner.project_id, owner, runner.episode_id, refresh=(mode == "force")) def _assign_short_ids(runner, entities: List[Dict], owner: str) -> List[Dict]: """엔티티 리스트에 `short_id` 를 부여한다 — **프로젝트 장부에서**. ★★★종전에는 `f"{prefix}{i:02d}"` 로 **이 목록 안 위치**를 번호로 썼다. 그래서 화가 달라도 늘 `C01` 부터 시작했고, `entity_canon` 의 유일성이 `(project_id, short_id)` 라 2화의 `C01` 이 **1화의 행을 찾아 덮었다.** 실측(골목 끝 `da049582`): 1화 C01=민수 였는데 지금 DB 의 C01 은 2·3화의 정임이고, 민수는 C04 로 밀려 있다. ★이미 `short_id` 가 있는 것은 **안 건드린다** — 앞 화에서 물려받은 신원이거나 producer 가 준 것이다. """ from app.core.entity_identity import assign_short_ids return assign_short_ids(runner.db, runner.project_id, owner, entities) class EntityAllCharacterStep(_EntityStepMixin, StepRunner): """인물 리스팅 — shot 기반 1회 호출 (fallback: 씬 체이닝).""" def _execute(self, mode="resume") -> Dict[str, Any]: rules_cp = self._load_prev_checkpoint("visual_world_rules") visual_rules = json.dumps(rules_cp.get("data", {}), ensure_ascii=False) if rules_cp else "" # shot 기반 추출 shot_cp = self._load_prev_checkpoint("shot_validator") if shot_cp: from app.core.steps.shot_validator_step import assert_no_failed_scenes assert_no_failed_scenes(shot_cp, self.project_config, consumer_step="entity_all_character") shots_scenes = shot_cp.get("data", {}).get("scenes", []) if shot_cp else [] total_shots = sum(len(s.get("shots", [])) for s in shots_scenes) # ★★결속 장부가 앉을 자리. **여기가 빠져 있었다** — 두 갈래 모두 # provider 호출 **전에** `NameError` 였다(Codex). 세 갈래에 같은 줄을 # 손으로 넣다가 한 곳만 빠진 부류다. _binding: Dict[str, Any] = {} # ★★앞 화 명부 + 이관 장부. **두 경로 모두**에 실어야 한다 — # 한쪽만 주면 그 갈래를 탄 프로젝트에서 이관이 통째로 없다. _roster = _prior_roster(self, "character", mode) _carry: Dict[str, Any] = {} # ★이름 근거 장부가 앉을 자리 (2026-09-19). 근거가 안 맞는 이름을 # **버리지는 않지만**, 체크포인트에 남겨야 사람이 볼 수 있다. # 로그로만 두면 「어긋난 게 없었다」와 「아무도 안 봤다」가 같아진다. _prov: Dict[str, Any] = {} if total_shots > 0: from app.modules.pipeline.entity_lister import list_characters_from_shots characters = list_characters_from_shots( shots_scenes, visual_rules, self.project_config, self.build_opik_metadata(), a0_candidates=self._a0_candidates(), binding_out=_binding, prior_roster=_roster, carry_out=_carry, provenance_out=_prov, ) else: logger.warning("entity_all_character: no shots found, falling back to scene chaining") save_cp = self._load_prev_checkpoint("scene_save") scenes = save_cp.get("data", {}).get("segments", []) if save_cp else [] from app.modules.pipeline.entity_lister import list_entities_by_type characters = list_entities_by_type( "", "character", visual_rules, self.project_config, self.build_opik_metadata(), scenes=scenes, # ★shot 경로와 **같은 결속 계약**. 한쪽만 주면 # 이 갈래를 탄 프로젝트에서 결속이 통째로 없다. a0_candidates=self._a0_candidates(), binding_out=_binding, prior_roster=_roster, carry_out=_carry) _assign_short_ids(self, characters, "character") # ★★장부를 **short_id 로 잇는다** (Codex 지적). 이름으로 이으면 # 나중에 이름이 바뀌는 순간 표식이 끊긴다 — 그런데 이 장부는 # 바로 「이름이 수상하다」는 기록이라 이름이 바뀔 가능성이 가장 높다. # short_id 는 위 발급기가 방금 붙였으므로 여기서만 이을 수 있다. if _prov.get("unverified"): _by_name = {c.get("name"): c.get("short_id") for c in characters} for _u in _prov["unverified"]: _u["short_id"] = _by_name.get(_u.get("name")) return {"completed_count": len(characters), "applicable_count": len(characters), "failed_count": 0, # ★저장과 비교가 **같은 함수**를 봐야 한다 — 여기서 안 # 실으면 저장은 project_config 만 보고 비교는 local 을 # 봐서 매번 어긋난다. "config_hash": self._config_hash(), "data": {"characters": characters, # ★★후보마다 어디로 갔는지. 없으면 다음 소비자가 # 「다퉜다」와 「아무것도 아니다」를 못 가른다. **({_gb.LEDGER_KEY: _binding.get("ledger") or {}} if _binding else {}), # ★★앞 화에서 물려받았는지 새것인지. 같은 이유로 # 남긴다 — 로그로만 두면 「이관이 안 됐다」와 # 「이관할 것이 없었다」가 구별이 안 된다. **({_ec.LEDGER_KEY: _carry} if _carry else {}), # ★★이름 근거 장부. 「전부 확인」과 「아무도 안 봤다」를 # 가르는 유일한 자리다 — 어긋난 이름을 버리지 않으므로 # 여기 안 남기면 산출만 보고는 알 길이 없다. **({"name_provenance": _prov} if _prov else {})}} class EntityAllLocationStep(_EntityStepMixin, StepRunner): """배경 리스팅 — shot 기반 1회 호출.""" def _execute(self, mode="resume") -> Dict[str, Any]: rules_cp = self._load_prev_checkpoint("visual_world_rules") visual_rules = json.dumps(rules_cp.get("data", {}), ensure_ascii=False) if rules_cp else "" shot_cp = self._load_prev_checkpoint("shot_validator") shots_scenes = shot_cp.get("data", {}).get("scenes", []) if shot_cp else [] total_shots = sum(len(s.get("shots", [])) for s in shots_scenes) # ★★결속 장부가 앉을 자리. 로그로만 남기면 다음 소비자가 못 읽는다. _binding: Dict[str, Any] = {} # ★★앞 화 명부 + 이관 장부. **두 경로 모두**에 실어야 한다 — # 한쪽만 주면 그 갈래를 탄 프로젝트에서 이관이 통째로 없다. _roster = _prior_roster(self, "location", mode) _carry: Dict[str, Any] = {} if total_shots > 0: from app.modules.pipeline.entity_lister import list_entities_from_shots locations = list_entities_from_shots( shots_scenes, "location", visual_rules, self.project_config, self.build_opik_metadata(), a0_candidates=self._a0_candidates(), binding_out=_binding, prior_roster=_roster, carry_out=_carry) else: logger.warning("entity_all_location: no shots, falling back to scene chaining") save_cp = self._load_prev_checkpoint("scene_save") scenes = save_cp.get("data", {}).get("segments", []) if save_cp else [] from app.modules.pipeline.entity_lister import list_entities_by_type locations = list_entities_by_type( "", "location", visual_rules, self.project_config, self.build_opik_metadata(), scenes=scenes, # ★shot 경로와 **같은 결속 계약**. 한쪽만 주면 # 이 갈래를 탄 프로젝트에서 결속이 통째로 없다. a0_candidates=self._a0_candidates(), binding_out=_binding, prior_roster=_roster, carry_out=_carry) _assign_short_ids(self, locations, "location") return {"completed_count": len(locations), "applicable_count": len(locations), "failed_count": 0, # ★저장과 비교가 **같은 함수**를 봐야 한다 — 여기서 안 # 실으면 저장은 project_config 만 보고 비교는 local 을 # 봐서 매번 어긋난다. "config_hash": self._config_hash(), "data": {"locations": locations, # ★★후보마다 어디로 갔는지. 없으면 다음 소비자가 # 「다퉜다」와 「아무것도 아니다」를 못 가른다. **({_gb.LEDGER_KEY: _binding.get("ledger") or {}} if _binding else {}), # ★★앞 화에서 물려받았는지 새것인지. 같은 이유로 # 남긴다 — 로그로만 두면 「이관이 안 됐다」와 # 「이관할 것이 없었다」가 구별이 안 된다. **({_ec.LEDGER_KEY: _carry} if _carry else {})}} class EntityAllPropStep(_EntityStepMixin, StepRunner): """소품 리스팅 — shot 기반 1회 호출.""" def _execute(self, mode="resume") -> Dict[str, Any]: rules_cp = self._load_prev_checkpoint("visual_world_rules") visual_rules = json.dumps(rules_cp.get("data", {}), ensure_ascii=False) if rules_cp else "" shot_cp = self._load_prev_checkpoint("shot_validator") shots_scenes = shot_cp.get("data", {}).get("scenes", []) if shot_cp else [] total_shots = sum(len(s.get("shots", [])) for s in shots_scenes) # ★★결속 장부가 앉을 자리. 로그로만 남기면 다음 소비자가 못 읽는다. _binding: Dict[str, Any] = {} # ★★앞 화 명부 + 이관 장부. **두 경로 모두**에 실어야 한다 — # 한쪽만 주면 그 갈래를 탄 프로젝트에서 이관이 통째로 없다. _roster = _prior_roster(self, "prop", mode) _carry: Dict[str, Any] = {} if total_shots > 0: from app.modules.pipeline.entity_lister import list_entities_from_shots props = list_entities_from_shots( shots_scenes, "prop", visual_rules, self.project_config, self.build_opik_metadata(), a0_candidates=self._a0_candidates(), binding_out=_binding, prior_roster=_roster, carry_out=_carry) else: logger.warning("entity_all_prop: no shots, falling back to scene chaining") save_cp = self._load_prev_checkpoint("scene_save") scenes = save_cp.get("data", {}).get("segments", []) if save_cp else [] from app.modules.pipeline.entity_lister import list_entities_by_type props = list_entities_by_type( "", "prop", visual_rules, self.project_config, self.build_opik_metadata(), scenes=scenes, # ★shot 경로와 **같은 결속 계약**. 한쪽만 주면 # 이 갈래를 탄 프로젝트에서 결속이 통째로 없다. a0_candidates=self._a0_candidates(), binding_out=_binding, prior_roster=_roster, carry_out=_carry) _assign_short_ids(self, props, "prop") return {"completed_count": len(props), "applicable_count": len(props), "failed_count": 0, # ★저장과 비교가 **같은 함수**를 봐야 한다 — 여기서 안 # 실으면 저장은 project_config 만 보고 비교는 local 을 # 봐서 매번 어긋난다. "config_hash": self._config_hash(), "data": {"props": props, # ★★후보마다 어디로 갔는지. 없으면 다음 소비자가 # 「다퉜다」와 「아무것도 아니다」를 못 가른다. **({_gb.LEDGER_KEY: _binding.get("ledger") or {}} if _binding else {}), # ★★앞 화에서 물려받았는지 새것인지. 같은 이유로 # 남긴다 — 로그로만 두면 「이관이 안 됐다」와 # 「이관할 것이 없었다」가 구별이 안 된다. **({_ec.LEDGER_KEY: _carry} if _carry else {})}} class EntityExtractStep(_EntityStepMixin, StepRunner): """Step 8 (legacy): 요소 추출 (3턴: 인물 -> 배경 -> 소품).""" def _execute(self, mode="resume") -> Dict[str, Any]: fulltext = self._load_cleaned_text() rules_cp = self._load_prev_checkpoint("visual_world_rules") visual_rules = ( json.dumps(rules_cp.get("data", {}), ensure_ascii=False) if rules_cp else "" ) save_cp = self._load_prev_checkpoint("scene_save") segments_json = ( json.dumps( save_cp.get("data", {}).get("segments", []), ensure_ascii=False, ) if save_cp else "" ) from app.modules.pipeline.entity_extractor_v4 import extract_all_entities result = extract_all_entities( fulltext, visual_rules, segments_json, self.project_config, self.build_opik_metadata(), ) total = ( len(result.get("characters", [])) + len(result.get("locations", [])) + len(result.get("props", [])) ) return { "completed_count": total, "applicable_count": total, "failed_count": 0, "data": result, } def _inject_short_ids(entities: List[Dict], sid_map: Dict[str, str]) -> List[Dict]: """entity_all의 name→short_id 매핑을 extract 결과에 주입.""" for e in entities: if not e.get("short_id"): e["short_id"] = sid_map.get(e.get("name", ""), "") return entities class EntityExtractCharacterStep(_EntityStepMixin, StepRunner): """Step 8: 인물 추출 — entity_all_character 리스트 기반으로 상세 설명 추가.""" def _execute(self, mode="resume") -> Dict[str, Any]: fulltext = self._load_cleaned_text() rules_cp = self._load_prev_checkpoint("visual_world_rules") visual_rules = json.dumps(rules_cp.get("data", {}), ensure_ascii=False) if rules_cp else "" all_cp = self._load_prev_checkpoint("entity_all_character") # ★★앞 단계 장부를 물려받고, 여기서 끊긴 것을 **되쓴다**. _prev_ledger = (all_cp or {}).get("data", {}).get(_gb.LEDGER_KEY) or {} _binding: Dict[str, Any] = {} if all_cp and all_cp.get("data", {}).get("characters"): name_list = all_cp["data"]["characters"] sid_map = {e["name"]: e.get("short_id", "") for e in name_list} from app.modules.pipeline.entity_extractor_v4 import extract_entities_by_type_with_list characters = extract_entities_by_type_with_list( fulltext, "character", name_list, visual_rules, self.project_config, self.build_opik_metadata(), a0_candidates=self._a0_candidates(), binding_out=_binding) _inject_short_ids(characters, sid_map) else: from app.modules.pipeline.entity_extractor_v4 import extract_entities_by_type characters = extract_entities_by_type(fulltext, "character", visual_rules, "", self.project_config, self.build_opik_metadata()) return {"completed_count": len(characters), "applicable_count": len(characters), "failed_count": 0, # ★저장과 비교가 **같은 함수**를 봐야 한다 — 여기서 안 # 실으면 저장은 project_config 만 보고 비교는 local 을 # 봐서 매번 어긋난다. "config_hash": self._config_hash(), "data": {"characters": characters, # ★★「붙었다가 끊긴 것」을 `lost` 로 되쓴다. 그냥 # `unbound` 로 두면 「아무도 안 불렀다」와 같아져 # 승격된다. _gb.LEDGER_KEY: _gb.merge_ledger( {"ledger": _prev_ledger}, lost=_binding.get("lost") or [], contested=[])}} class EntityExtractLocationStep(_EntityStepMixin, StepRunner): """Step 9: 배경 추출 — entity_all_location 리스트 기반으로 상세 설명 추가.""" def _execute(self, mode="resume") -> Dict[str, Any]: fulltext = self._load_cleaned_text() rules_cp = self._load_prev_checkpoint("visual_world_rules") visual_rules = json.dumps(rules_cp.get("data", {}), ensure_ascii=False) if rules_cp else "" all_cp = self._load_prev_checkpoint("entity_all_location") # ★★앞 단계 장부를 물려받고, 여기서 끊긴 것을 **되쓴다**. _prev_ledger = (all_cp or {}).get("data", {}).get(_gb.LEDGER_KEY) or {} _binding: Dict[str, Any] = {} if all_cp and all_cp.get("data", {}).get("locations"): name_list = all_cp["data"]["locations"] sid_map = {e["name"]: e.get("short_id", "") for e in name_list} from app.modules.pipeline.entity_extractor_v4 import extract_entities_by_type_with_list locations = extract_entities_by_type_with_list( fulltext, "location", name_list, visual_rules, self.project_config, self.build_opik_metadata(), a0_candidates=self._a0_candidates(), binding_out=_binding) _inject_short_ids(locations, sid_map) else: from app.modules.pipeline.entity_extractor_v4 import extract_entities_by_type locations = extract_entities_by_type(fulltext, "location", visual_rules, "", self.project_config, self.build_opik_metadata()) return {"completed_count": len(locations), "applicable_count": len(locations), "failed_count": 0, # ★저장과 비교가 **같은 함수**를 봐야 한다 — 여기서 안 # 실으면 저장은 project_config 만 보고 비교는 local 을 # 봐서 매번 어긋난다. "config_hash": self._config_hash(), "data": {"locations": locations, # ★★「붙었다가 끊긴 것」을 `lost` 로 되쓴다. 그냥 # `unbound` 로 두면 「아무도 안 불렀다」와 같아져 # 승격된다. _gb.LEDGER_KEY: _gb.merge_ledger( {"ledger": _prev_ledger}, lost=_binding.get("lost") or [], contested=[])}} class EntityExtractPropStep(_EntityStepMixin, StepRunner): """Step 10: 소품 추출 — entity_all_prop 리스트 기반으로 상세 설명 추가.""" def _execute(self, mode="resume") -> Dict[str, Any]: fulltext = self._load_cleaned_text() rules_cp = self._load_prev_checkpoint("visual_world_rules") visual_rules = json.dumps(rules_cp.get("data", {}), ensure_ascii=False) if rules_cp else "" all_cp = self._load_prev_checkpoint("entity_all_prop") # ★★앞 단계 장부를 물려받고, 여기서 끊긴 것을 **되쓴다**. _prev_ledger = (all_cp or {}).get("data", {}).get(_gb.LEDGER_KEY) or {} _binding: Dict[str, Any] = {} if all_cp and all_cp.get("data", {}).get("props"): name_list = all_cp["data"]["props"] sid_map = {e["name"]: e.get("short_id", "") for e in name_list} from app.modules.pipeline.entity_extractor_v4 import extract_entities_by_type_with_list props = extract_entities_by_type_with_list( fulltext, "prop", name_list, visual_rules, self.project_config, self.build_opik_metadata(), a0_candidates=self._a0_candidates(), binding_out=_binding) _inject_short_ids(props, sid_map) else: from app.modules.pipeline.entity_extractor_v4 import extract_entities_by_type props = extract_entities_by_type(fulltext, "prop", visual_rules, "", self.project_config, self.build_opik_metadata()) return {"completed_count": len(props), "applicable_count": len(props), "failed_count": 0, # ★저장과 비교가 **같은 함수**를 봐야 한다 — 여기서 안 # 실으면 저장은 project_config 만 보고 비교는 local 을 # 봐서 매번 어긋난다. "config_hash": self._config_hash(), "data": {"props": props, # ★★「붙었다가 끊긴 것」을 `lost` 로 되쓴다. 그냥 # `unbound` 로 두면 「아무도 안 불렀다」와 같아져 # 승격된다. _gb.LEDGER_KEY: _gb.merge_ledger( {"ledger": _prev_ledger}, lost=_binding.get("lost") or [], contested=[])}} # ── Step 11: entity_filter ── class EntityFilterStep(_EntityStepMixin, StepRunner): """Step 11: 저빈도 요소 필터링 (3씬 이하 -> LLM 판단).""" def _chunk_projection(self) -> Dict[str, Any]: """C(c) 판 — **호출 0 투영**. ★저빈도 필터를 다시 안 돌린다. ★★★producer 가 한 판독에서 **등록 문턱**(반복 또는 두 축 예외)을 이미 걸었다. 여기서 저빈도 LLM 을 또 돌리면 — ①같은 뜻을 두 번 판단하고(그 판정이 갈리면 정본이 없다) ②`location_parts` 를 아예 못 보고 **네 번째 갈래가 사라진다** (앞 판이 실제로 그랬다 — Codex 재현 2026-09-01) 그래서 등록된 줄을 **그대로** 통과시킨다. """ from app.core.errors import AppError from app.modules.pipeline.grounding_carry import ENTITY_KEY_TO_OWNER cp = self._load_prev_checkpoint("entity_merge") data = (cp or {}).get("data") or {} if not data: raise AppError( code="step.no_input", message=("entity_merge 산출 없음 — C(c) 판에서 필터는 그 " "투영을 그대로 통과시킨다"), status_code=400) lanes = {k: list(data.get(k) or []) for k, _o in ENTITY_KEY_TO_OWNER if data.get(k) or k in ("characters", "locations", "props")} total = sum(len(v) for v in lanes.values()) return { "completed_count": total, "applicable_count": total, "failed_count": 0, "config_hash": self._config_hash(), "data": { "filtered_entities": lanes, # ★아무것도 안 지웠다 — producer 문턱이 이미 걸렀다 "removed_entities": [], "removed": [], "projected_from": "entity_merge", "projection_reason": ( "C(c) producer 가 등록 문턱을 이미 걸었다 — " "같은 뜻을 다시 판단하지 않는다"), }, } def _execute(self, mode="resume") -> Dict[str, Any]: from app.core.grounding_mode import (resolve_grounding_mode, uses_chunk_producer) # ★★★C(c) 판이면 **호출 0 투영**이다. if uses_chunk_producer(resolve_grounding_mode(self.project_config)): return self._chunk_projection() fulltext = self._load_cleaned_text() # entity_merge 결과 로드 (병합 후 필터링) merge_cp = self._load_prev_checkpoint("entity_merge") if merge_cp and merge_cp.get("data"): entities = { "characters": merge_cp["data"].get("characters", []), "locations": merge_cp["data"].get("locations", []), "props": merge_cp["data"].get("props", []), } else: # fallback: 개별 추출 결과 char_cp = self._load_prev_checkpoint("entity_extract_character") loc_cp = self._load_prev_checkpoint("entity_extract_location") prop_cp = self._load_prev_checkpoint("entity_extract_prop") entities = { "characters": (char_cp or {}).get("data", {}).get("characters", []), "locations": (loc_cp or {}).get("data", {}).get("locations", []), "props": (prop_cp or {}).get("data", {}).get("props", []), } # Load segments for scene context save_cp = self._load_prev_checkpoint("scene_save") segments = save_cp.get("data", {}).get("segments", []) if save_cp else [] # entity_all에서 shot_count 복원 (extract/merge에서 유실됨) for etype, prefix in [("characters", "entity_all_character"), ("locations", "entity_all_location"), ("props", "entity_all_prop")]: all_cp = self._load_prev_checkpoint(prefix) if all_cp and all_cp.get("data"): key = etype from app.modules.pipeline.entity_filter import _appearance_count count_map = {e["name"]: _appearance_count(e) for e in all_cp["data"].get(key, [])} for e in entities.get(etype, []): if "shot_count" not in e: e["shot_count"] = count_map.get(e["name"], 0) # entity_relation에서 변형 관계(visual_similarity=true)가 있는 요소는 보호 protected_sids: set = set() rel_cp = self._load_prev_checkpoint("entity_relation") if rel_cp and rel_cp.get("data", {}).get("relations"): for rel in rel_cp["data"]["relations"]: if rel.get("visual_similarity"): protected_sids.add(rel.get("base_short_id", "")) protected_sids.add(rel.get("variant_short_id", "")) protected_sids.discard("") # ★GROUNDING-V2 §2-3.5 — 조사·저작·미확정 대상을 **기존 통로에 union** 한다. # 새 필터 기구를 만들지 않는다(계획 §1.8). 체크포인트가 없으면 # 아무것도 안 더해져 **legacy 가 그대로 돈다.** # # ★그런데 `v2` 에서까지 「없으면 안 더한다」로 두면 **fail-open** 이다 — # 계획이 안 돌았을 뿐인데 조사 대상이 조용히 지워진다. `entity_filter` 는 # `grounding_plan` 을 `depends_on` 으로 걸 수 없다(기존 프로젝트가 전부 # `gate.blocked` 가 된다). 그래서 **모드로 가려 여기서 닫는다.** from app.core.grounding_mode import buys_v2_research, resolve_grounding_mode _g_mode = resolve_grounding_mode(self.project_config) _g_cp = self._load_prev_checkpoint("grounding_plan") _g_data = (_g_cp or {}).get("data") or {} if buys_v2_research(_g_mode): from app.core.errors import AppError if not _g_cp: raise AppError( code="entity_filter.grounding_plan_missing", message=(f"grounding_mode={_g_mode} 인데 고증 계획 체크포인트가 " f"없다 — 보호 목록 없이 거르면 조사 대상이 지워진다"), status_code=400, ) _cp_mode = _g_data.get("mode") if _cp_mode != _g_mode: raise AppError( code="entity_filter.grounding_plan_stale", message=(f"고증 계획이 mode={_cp_mode!r} 로 돌았는데 지금은 " f"{_g_mode!r} 다 — 다시 돌려야 한다"), status_code=400, ) _g_decided = _g_data.get("decided") or [] if _g_decided: from app.modules.pipeline.grounding_overlay import ( materialize_missing_entities, protected_short_ids) # ★★★**어려운 단발 대상을 실제 행으로 등록한다** (사용자 확정). # `protected_short_ids` 는 **이미 있는 행**만 지킨다 — 추출이 # 지워 버린 것(옛 화폐·옛 브랜드 제품·알려진 장소)은 지킬 행 # 자체가 없다. 그러면 조사 대상에는 있는데 엔티티·카드에는 없는 # **반쪽**이 된다. # ★대상은 `generation_difficulty` 가 여는 것만이다 — `route` 로 # 고르면 실측상 **전부** 등록된다(아홉 축이 전부 research 였다). # ★★A0 후보는 `grounding_plan` 체크포인트에 **없다** — 거기엔 # `decided`·`completeness`·`candidate_ledger` 만 있다. 짐작으로 # `_g_data.get("candidates")` 를 읽었더니 늘 빈 목록이어서 # **아무것도 안 하면서 조용히 통과**했다. # ★`grounding_plan` 이 읽는 **바로 그 자리**에서 읽는다 # (`grounding_steps`: `a0_cp["data"]["candidates"]`). _a0 = (((self._load_prev_checkpoint("grounding_a0") or {}) .get("data") or {}).get("candidates") or []) if _a0: _new = materialize_missing_entities(_a0, _g_decided, entities) for _key, _rows in _new.items(): # ★함수가 **받은 키 그대로** 돌려준다 — 변환표를 또 두면 # 한쪽만 고쳐진다. entities.setdefault(_key, []) entities[_key].extend(_rows) # ★만든 행은 **보호도 같이** 받는다. 안 그러면 방금 만든 # 것을 저빈도 필터가 그 자리에서 도로 지운다. protected_sids |= {r["short_id"] for r in _rows} if _new: logger.info( "grounding: 어려운 단발 대상 %d개를 엔티티로 등록했다", sum(len(v) for v in _new.values())) _g_protected = protected_short_ids(_g_decided, entities) if _g_protected: logger.info( "grounding: 저빈도 필터 보호에 %d개 추가 (research/design/unresolved)", len(_g_protected)) protected_sids |= _g_protected from app.modules.pipeline.entity_filter import filter_low_frequency_entities # ★★문턱은 **계약 모듈 한 곳**에서 온다. 여기 숫자를 적으면 두 벌이 # 되고, 실제로 그랬다 — 상수를 만들어 놓고 아무도 안 썼다 (Codex). from app.modules.pipeline.grounding_entity_contract import ( ENTITY_MIN_OCCURRENCES) result = filter_low_frequency_entities( entities=entities, segments=segments, fulltext=fulltext, max_scenes=ENTITY_MIN_OCCURRENCES, protected_short_ids=protected_sids if protected_sids else None, project_config=self.project_config, opik_metadata=self.build_opik_metadata(), ) return { "completed_count": 1, "applicable_count": 1, "failed_count": 0, "data": result, } # ── Step 12: entity_review ── class EntityReviewStep(_EntityStepMixin, StepRunner): """Step 12: 요소 교차 검증 — 다른 모델로 추출 결과 교차 확인.""" def _execute(self, mode="resume") -> Dict[str, Any]: fulltext = self._load_cleaned_text() # Try entity_filter checkpoint first (has filtered results) filter_cp = self._load_prev_checkpoint("entity_filter") if filter_cp and filter_cp.get("data", {}).get("filtered_entities"): extract_data = filter_cp["data"]["filtered_entities"] else: # Fallback: Load from 3 separate checkpoints (v3 split) or legacy single checkpoint char_cp = self._load_prev_checkpoint("entity_extract_character") loc_cp = self._load_prev_checkpoint("entity_extract_location") prop_cp = self._load_prev_checkpoint("entity_extract_prop") if char_cp or loc_cp or prop_cp: extract_data = { "characters": (char_cp or {}).get("data", {}).get("characters", []), "locations": (loc_cp or {}).get("data", {}).get("locations", []), "props": (prop_cp or {}).get("data", {}).get("props", []), } else: # Legacy fallback: single entity_extract checkpoint prev = self._load_prev_checkpoint("entity_extract") if not prev or not prev.get("data"): from app.core.errors import AppError raise AppError( code="step.no_input", message="entity_extract 결과 없음", status_code=400, ) extract_data = prev["data"] from app.modules.pipeline.entity_reviewer import review_entities review_result = review_entities( entities=extract_data, fulltext=fulltext, project_config=self.project_config, opik_metadata=self.build_opik_metadata(), ) # 불필요 항목 필터링 — 일시 비활성화 (review 결과만 기록, 제거 안 함) filtered = { "characters": list(extract_data.get("characters", [])), "locations": list(extract_data.get("locations", [])), "props": list(extract_data.get("props", [])), "review": review_result, } total = ( len(filtered["characters"]) + len(filtered["locations"]) + len(filtered["props"]) ) return { "completed_count": 1, "applicable_count": 1, "failed_count": 0, "data": filtered, } # ── Step 13.5: entity_merge ── class EntityMergeStep(_EntityStepMixin, StepRunner): """요소 중복 병합 — 타입 간 중복/유사 요소를 LLM으로 판별하여 제거.""" def _chunk_projection(self) -> Dict[str, Any]: """C(c) 판 — 새 producer 산출을 **그대로 투영**한다. ★호출 0. ★★앞 판은 옛 추출 여섯을 끄고 **대체 산출을 안 이었다** — 그러면 `total=0` 이라 빈 배열을 정상 completed 로 내고, 하류(`entity_detail`· `entity_t2i`·DB sync)에서 **엔티티가 통째로 사라진다** (Codex 재현 2026-09-01). 「중복 호출을 막았다」가 아니라 **전부 잃은 것**이다. ★여기서 판단하지 않는다 — producer 가 낸 갈래 줄을 옮기고 병합 칸만 빈 값으로 맞춘다. 모델을 **한 번도 안 부른다**. """ from app.core.errors import AppError from app.modules.pipeline.grounding_carry import ENTITY_KEY_TO_OWNER cp = self._load_prev_checkpoint("grounding_chunk") data = (cp or {}).get("data") or {} if not data: raise AppError( code="step.no_input", message=("grounding_chunk 산출 없음 — C(c) 판에서 엔티티는 " "그 producer 가 낸 줄이 정본이다"), status_code=400) lanes = {k: list(data.get(k) or []) for k, _o in ENTITY_KEY_TO_OWNER if data.get(k) or k in ("characters", "locations", "props")} total = sum(len(v) for v in lanes.values()) return { "completed_count": total, "applicable_count": total, "failed_count": 0, "config_hash": self._config_hash(), # ★★sync 가 `location_parts` 를 **열려면** 이 표식이 CP **최상위** # 에 있어야 한다(`is_chunk_marked` 가 거기를 본다). `data` 안에 # 넣으면 못 찾고 LP 행이 조용히 버려진다. _sync_ext.CHUNK_SCHEMA_MARKER: _sync_ext.CHUNK_SCHEMA_VERSION, "data": { **lanes, # ★병합·삭제는 producer 가 이미 한 판독 안에서 끝냈다 "removed": [], "merges": [], "merge_provenance_asked": False, "merge_refused": [], # ★어디서 왔는지 남긴다 — 감사와 무효화의 근거다 "projected_from": "grounding_chunk", "projection_contract": str( (data.get("contracts") or {}).get("adapter") or ""), }, } def _execute(self, mode="resume") -> Dict[str, Any]: from app.core.grounding_mode import (resolve_grounding_mode, uses_chunk_producer) # ★★★C(c) 판이면 **호출 0 투영**이다 — 옛 추출은 안 돌았다. if uses_chunk_producer(resolve_grounding_mode(self.project_config)): return self._chunk_projection() # 3개 extract 결과 로드 char_cp = self._load_prev_checkpoint("entity_extract_character") loc_cp = self._load_prev_checkpoint("entity_extract_location") prop_cp = self._load_prev_checkpoint("entity_extract_prop") characters = (char_cp or {}).get("data", {}).get("characters", []) locations = (loc_cp or {}).get("data", {}).get("locations", []) props = (prop_cp or {}).get("data", {}).get("props", []) total = len(characters) + len(locations) + len(props) if total == 0: return {"completed_count": 0, "applicable_count": 0, "failed_count": 0, "config_hash": self._config_hash(), "data": {"characters": [], "locations": [], "props": [], "removed": []}} # visual_world_rules vwr_cp = self._load_prev_checkpoint("visual_world_rules") vwr_data = vwr_cp.get("data", {}) if vwr_cp else {} era = vwr_data.get("era", "") region = vwr_data.get("region", "") # scene_summary sum_cp = self._load_prev_checkpoint("scene_summary") summaries = sum_cp.get("data", {}).get("summaries", []) if sum_cp else [] summary_text = "\n".join( f"씬{s.get('scene_index', '?')}: {s.get('scene_summary', '')}" for s in summaries ) # 요소 목록 구성 entity_lines = [] for c in characters: entity_lines.append(f"{c.get('short_id','')} {c['name']} (character): {c.get('description','')}") for l in locations: entity_lines.append(f"{l.get('short_id','')} {l['name']} (location): {l.get('description','')}") for p in props: entity_lines.append(f"{p.get('short_id','')} {p['name']} (prop): {p.get('description','')}") user_prompt = ( f"[세계관] 시대: {era}, 지역: {region}\n\n" f"[씬별 요약]\n{summary_text}\n\n" f"[전체 요소 목록]\n" + "\n".join(entity_lines) + "\n\n" "위 요소 목록에서 타입이 다르지만 실질적으로 같은 대상인 항목을 찾으세요.\n" "예: 같은 물체가 배경(location)과 소품(prop)에 동시 등록된 경우.\n" "중복이 있으면 제거할 short_id를 선택하세요. 없으면 빈 배열을 반환하세요.\n" "각 타입 쌍을 고르게 검토하되, 실질적으로 같은 대상이 서로 다른 타입으로 등록된 경우를 찾으세요." ) remove_schema = { "type": "object", "properties": { "remove": { "type": "array", "items": {"type": "string"}, "description": "제거할 short_id 목록", }, }, "required": ["remove"], "additionalProperties": False, } # ★★**지울 것만 받으면 그 행의 고증 근거가 그대로 사라진다** (3b 계약 6). # 「무엇을 지우나」와 함께 「어디로 합치나」를 받아, 지운 행의 후보 ID 를 # 남는 행에 옮긴다. 옛 `remove` 는 그대로 두어 되돌려 읽을 수 있게 한다. # ★**옮길 근거가 있을 때만 묻는다.** 늘 물으면 후보가 없는 판(legacy # 포함)의 나가는 지문·schema 가 달라진다 — 이 단계와 무관한 변화다. from app.modules.pipeline import grounding_binding as _gb from app.modules.pipeline.grounding_binding import FIELD as _GB_FIELD _has_provenance = any( e.get(_GB_FIELD) for e in characters + locations + props) if _has_provenance: remove_schema["properties"]["merges"] = { "type": "array", "description": "제거하는 것마다 어느 것으로 합치는지", "items": { "type": "object", "properties": { "remove_short_id": {"type": "string"}, "keep_short_id": {"type": "string"}, }, "required": ["remove_short_id", "keep_short_id"], "additionalProperties": False, }, } remove_schema["required"] = ["remove", "merges"] # ★지시문은 **팩**에서 온다. user_prompt += "\n\n" + _gb.merge_instruction().strip() removed = [] merges: List[Dict] = [] _refused: set = set() try: result = call_structured( step="entity_merge", system_prompt="당신은 시나리오 요소 분석 전문가입니다. 중복 요소를 찾아 제거할 항목을 판별합니다.", user_prompt=user_prompt, response_schema=remove_schema, project_config=self.project_config, schema_name="entity_merge", opik_metadata=self.build_opik_metadata(), ) removed = result.get("remove", []) merges = result.get("merges", []) or [] except Exception as exc: logger.warning("Entity merge LLM call failed: %s", exc) # 제거 적용 if removed: all_rows = characters + locations + props all_sids = {e.get("short_id", "") for e in all_rows} unknown = [sid for sid in removed if sid not in all_sids] if unknown: logger.warning("Entity merge: LLM returned unknown short_ids: %s", unknown) remove_set = set(removed) & all_sids # 존재하는 것만 # ★★**지우기 전에** 고증 근거를 남는 행으로 옮긴다. 지운 뒤에는 # 옮길 것이 없다. 옮길 데가 없으면 **그 행은 안 지운다**. _refused = _carry_merge_provenance(all_rows, merges, remove_set) remove_set -= _refused characters = [c for c in characters if c.get("short_id", "") not in remove_set] locations = [l for l in locations if l.get("short_id", "") not in remove_set] props = [p for p in props if p.get("short_id", "") not in remove_set] logger.info("Entity merge: removed %s", list(remove_set)) new_total = len(characters) + len(locations) + len(props) return { "completed_count": new_total, "applicable_count": total, "failed_count": 0, "config_hash": self._config_hash(), "data": { "characters": characters, "locations": locations, "props": props, "removed": removed, "merges": merges, "merge_provenance_asked": _has_provenance, # ★삭제를 거부한 것을 **기록에 남긴다** — 조용히 살려 두면 # 다음 사람이 「LLM 이 안 지우라 했다」로 읽는다. "merge_refused": sorted(_refused), }, } def _carry_merge_provenance(all_rows: List[Dict], merges: List[Dict], remove_set: set) -> set: """★★지우는 행의 **고증 후보 ID 를 남는 행에 옮긴다** (3b 계약 6). 안 옮기면 병합 한 번에 그 대상의 원문 근거가 사라지고, 뒤에서 「조사 대상이었는데 아무 엔티티에도 없다」가 된다. ★★**대응이 성립하지 않으면 그 행을 안 지운다** (Codex BLOCK-4). schema 가 허용하는 반례들 — remove=[P01], merges=[] 근거를 든 행이 그냥 사라진다 keep 이 목록에 없다 옮길 데가 없다 keep 도 remove 중이다 옮겨 봐야 같이 사라진다 keep == remove 자기 자신으로 합친다 한 remove 가 두 keeper 로 같은 후보가 두 행에 생긴다 「지우고 나서 경고」는 이미 사라진 뒤라 아무 소용이 없다. Returns: ★**지우면 안 되는 short_id 들.** 호출부가 `remove_set` 에서 뺀다. """ from app.modules.pipeline.grounding_binding import FIELD, carry_into by_sid = {str(e.get("short_id") or ""): e for e in all_rows} # ★근거를 **든** 행만 본다. 안 든 행은 옛 계약 그대로 지운다. bearing = {sid for sid in remove_set if (by_sid.get(sid) or {}).get(FIELD)} if not bearing: return set() plan: Dict[str, str] = {} refused: set = set() seen_from: set = set() for m in merges or (): src = str((m or {}).get("remove_short_id") or "") dst = str((m or {}).get("keep_short_id") or "") if src not in bearing: continue if src in seen_from: # ★한 행을 두 곳으로 합칠 수는 없다 — 같은 후보가 둘이 된다. refused.add(src) plan.pop(src, None) continue seen_from.add(src) if not dst or dst == src or dst not in by_sid or dst in remove_set: refused.add(src) continue plan[src] = dst # ★대응이 **아예 없는** 것도 거부다. 「merges 가 required 배열」인 것만으로는 # 빈 배열을 못 막는다. refused |= (bearing - set(plan)) for src, dst in plan.items(): carry_into(by_sid[dst], by_sid[src]) if refused: logger.warning( "Entity merge: 근거를 옮길 데가 없어 **삭제를 거부한** 행 %s", sorted(refused)) return refused # ── Step 10: entity_detail ── #: ★[2026-09-17] entity_detail 한 호출에 싣는 요소 수. #: 컨트리로드(요소 241개)를 한 번에 보내자 gpt-6-astra 가 **1개만 쓰고 #: `finish_reason=stop` 으로 끝냈다**(출력 689토큰). 빠진 240개를 다시 #: 보낸 재시도도 1개(489토큰). 잘린 게 아니라 긴 목록에서 첫 줄만 답한 것이다. #: 같은 입력(대본 전문·세계관·기획서 인물 절 그대로)에 목록만 잘라 보내니 #: **60개 → 60/60 · 30개 → 30/30** 이 돌아왔다. 파청(104개)은 한 번에 됐다. #: 검증된 60 보다 여유를 둬 50. ★대본 전문은 묶음마다 **자르지 않고** 보낸다. _DETAIL_CHUNK_SIZE = 50 #: 묶음을 같이 부르는 수 — beat_extract 와 같은 방식(스레드 풀 + call_structured). _DETAIL_CHUNK_WORKERS = 4 class EntityDetailStep(_EntityStepMixin, StepRunner): """Step 10: 요소 상세 — 시각적 상세 정보 + short_id 확정.""" def _execute(self, mode="resume") -> Dict[str, Any]: fulltext = self._load_cleaned_text() # entity_filter 우선 → entity_merge fallback → entity_extract fallback filter_cp = self._load_prev_checkpoint("entity_filter") if filter_cp and filter_cp.get("data", {}).get("filtered_entities"): filtered = filter_cp["data"]["filtered_entities"] else: merge_cp = self._load_prev_checkpoint("entity_merge") if merge_cp and "removed" in merge_cp.get("data", {}): # ★★갈래 표를 **여기서 다시 적지 않는다** (2026-09-01). # 세 갈래를 손으로 적어 둬서 v2_chunk producer 가 낸 # `location_parts` 가 여기서 통째로 사라졌다 — 같은 규칙이 # 두 곳이면 한쪽만 고쳐진다. # ★**없는 칸은 안 만든다** — 그래야 legacy CP 가 무변이다. filtered = {k: merge_cp["data"][k] for k, _o in _ENTITY_KEYS() if merge_cp["data"].get(k)} else: char_cp = self._load_prev_checkpoint("entity_extract_character") loc_cp = self._load_prev_checkpoint("entity_extract_location") prop_cp = self._load_prev_checkpoint("entity_extract_prop") filtered = { "characters": (char_cp or {}).get("data", {}).get("characters", []), "locations": (loc_cp or {}).get("data", {}).get("locations", []), "props": (prop_cp or {}).get("data", {}).get("props", []), } if not any(filtered.values()): from app.core.errors import AppError raise AppError( code="step.no_input", message="entity_extract 결과 없음", status_code=400, ) # 엔티티 큐 구성 (name, type, short_id) — ★신원은 short_id (계약 ENTITY_INSTANCE_IDENTITY_CONTRACT_VERSION). # 같은 (name, type) 이라도 short_id 가 다르면 **다른 실체**다(실측 f7cc45c576c0: 흙바닥 LP05/LP07). 이름으로 접지 않는다. # chunk producer 모드(v2_chunk)는 short_id 없음·중복을 provider 전에 fail-closed. legacy 만 (name, type) 로 접는다. from app.core.grounding_mode import resolve_grounding_mode, uses_chunk_producer _strict = uses_chunk_producer(resolve_grounding_mode(self.project_config)) entity_queue: List[tuple] = [] seen_keys: set = set() for etype_key, etype_label in _ENTITY_KEYS(): for e in filtered.get(etype_key, []): name = e["name"] sid = str(e.get("short_id") or "") if _strict and not sid: from app.core.errors import AppError raise AppError(code="entity_detail.identity_missing", status_code=422, message=f"producer 가 short_id 를 주지 않았다: {name!r} ({etype_label}) — 신원 없이 사지 않는다") key = sid if sid else (name, etype_label) if key in seen_keys: if _strict: from app.core.errors import AppError raise AppError(code="entity_detail.identity_duplicate", status_code=422, message=f"short_id 가 겹친다: {sid!r} ({name!r}, {etype_label}) — 사지 않고 선다") logger.warning("Entity queue: 중복 제거(legacy · short_id 없음) — %s (%s)", name, etype_label) continue seen_keys.add(key) entity_queue.append((name, etype_label, sid)) _sid_set = {str(q[2]) for q in entity_queue if len(q) >= 3 and q[2]} # ★(name, type) 짝이 큐에서 **유일**하면 표식 없이 온 답도 그 short_id 로 귀속한다 — 짝이 둘 이상인 것만 표식으로 결속 from collections import Counter as _Counter _pair_n = _Counter((q[0], q[1]) for q in entity_queue) _unique_sid = {(q[0], q[1]): str(q[2]) for q in entity_queue if len(q) >= 3 and q[2] and _pair_n[(q[0], q[1])] == 1} def _detail_key(ename: str, etype: str, eid: str) -> str: if eid in _sid_set: return eid if (ename, etype) in _unique_sid: return _unique_sid[(ename, etype)] return f"{ename}:{etype}" if etype else ename from app.modules.pipeline.entity_extractor_v3 import _load_prompt, _load_schema # visual_world_rules 로드 — description 생성에 시대/의상 맥락 반영 vwr_cp = self._load_prev_checkpoint("visual_world_rules") vwr_data = vwr_cp.get("data", {}) if vwr_cp else {} world_block = "" if vwr_data.get("era") or vwr_data.get("region"): world_block = f"\n[세계관]\n시대: {vwr_data.get('era', '')}\n지역: {vwr_data.get('region', '')}" for r in vwr_data.get("rules", []): if r.get("rule_type") == "costume": world_block += f"\n의상규칙: {r.get('description', '')}" # 기획서 인물 정보 보강 (첫 에피소드만) from app.core.planning_doc_context import get_planning_context pctx = get_planning_context(self.project_id, self.episode_id, self.db) planning_block = pctx.inject_if_available("characters_text", "## 기획서 인물 참고 정보") # ★장소도 같이 (2026-09-18) — 이 단계가 배경 description 도 만든다. # 기획서에 장소 절이 없으면 빈 문자열이라 조립이 안 바뀐다. planning_block += pctx.inject_if_available("locations_text", "## 기획서 장소 참고 정보") detail_schema = _load_schema("turn1_7_detail_batch_schema.json") # ★줄마다 불투명 표식(short_id)을 싣고 응답이 그 표식으로 결속된다 — 모델은 표식의 뜻을 판단하지 않는다(팩 v10) entity_list_text = "\n".join( (f"- [{sid}] {name} ({etype})" if sid else f"- {name} ({etype})") for name, etype, sid in ((q[0], q[1], (q[2] if len(q) >= 3 else "")) for q in entity_queue) ) detail_prompt = _load_prompt( "turn1_7_detail_batch", entity_list=entity_list_text, fulltext=fulltext, ) + world_block + planning_block # key = "name:type" (동명 다른 타입 구분) entity_details: Dict[str, Dict] = {} _system = (self._load_prompt("entity_extractor_v2", "system") if hasattr(self, '_load_prompt') else load_prompt("entity_extractor_v2", "system")) def _record(ent: Dict[str, Any]) -> Dict[str, Any]: # ★묶음 답과 재시도 답이 **같은 모양**으로 담긴다. 종전 재시도 갈래는 # distinctive_visual_traits 를 버려서, 다시 받은 요소만 특이 외형이 빠졌다. return { "description": ent.get("description", ""), "visual_traits": ent.get("visual_traits", []), # ② P0a generic fix (2026-06-18): evidence-gated distinctive # 외형. EntityT2iStep 이 source_quote + interpretation_kind 로 # 게이트하므로 여기선 raw 보존 (gate 는 build 시점). 누락 시 t2i # 가 distinctive 를 못 봄. "distinctive_visual_traits": ent.get("distinctive_visual_traits", []), # ★크기 견줌도 **원자료로 보존한다** (2026-09-20 Codex BLOCK). # 이 함수가 칸을 안 담으면 스키마·소비 helper 를 다 만들어도 # CP 에 값이 없어 T2I 가 `None` 을 받는다 — 게이트는 # `build_build_scale_block` 이 build 시점에 건다. "build_scale": ent.get("build_scale", []), } def _missing(rows, have: Dict[str, Dict]) -> List[tuple]: # ★short_id 가 있는 줄은 short_id 로만 센다(이름이 같은 다른 줄의 답을 빌리지 않는다) return [ (n, t, (r[0] if r else "")) for n, t, *r in rows if ((r and r[0]) and r[0] not in have) or (not (r and r[0]) and f"{n}:{t}" not in have and n not in have) ] def _ask(rows, schema_name: str, tags: List[str]) -> Dict[str, Dict]: # ★재시도도 세계관·기획서 인물 절을 **같이** 싣는다 — 종전 재시도는 둘 다 # 빠져 있어, 다시 받은 인물에는 기획서 내용이 안 들어갔다. listing = "\n".join( (f"- [{sid}] {n} ({t})" if sid else f"- {n} ({t})") for n, t, sid in ((q[0], q[1], (q[2] if len(q) >= 3 else "")) for q in rows) ) prompt = _load_prompt( "turn1_7_detail_batch", entity_list=listing, fulltext=fulltext, ) + world_block + planning_block # ★★표식 칸은 **이 호출에 실린 표식 중 하나**만 받는다 (2026-09-17 컨트리로드). # 장소 50·소품 41 묶음에서 gpt-6-astra 가 표식을 베끼다 글자가 뭉개지고 # ('P41ள', 'L64 "location"') 1개만 쓰고 끝냈다 — 첫 호출·재시도 모두. # 같은 입력에 표식만 묶으니 50/50 · 41/41, 표식↔이름 짝도 전부 맞았다. # ★표식 없는 줄이 섞이면 빈 값도 허용한다 — 막으면 남의 표식을 골라 그 답을 덮는다. # ★팩 스키마는 묶음끼리 같이 쓰므로 **복사본에만** 적는다. schema = detail_schema sids = sorted({str(q[2]) for q in rows if len(q) >= 3 and q[2]}) id_prop = (detail_schema.get("properties", {}).get("entities", {}) .get("items", {}).get("properties", {}).get("entity_id")) if sids and id_prop is not None: schema = copy.deepcopy(detail_schema) schema["properties"]["entities"]["items"]["properties"]["entity_id"]["enum"] = ( sids + ([""] if any(not (len(q) >= 3 and q[2]) for q in rows) else [])) result = call_structured( step="entity_detail_batch", system_prompt=_system, user_prompt=prompt, response_schema=schema, project_config=self.project_config, schema_name=schema_name, opik_metadata=self.build_opik_metadata(tags), ) got: Dict[str, Dict] = {} for ent in result.get("entities", []): key = _detail_key(ent["name"], ent.get("entity_type", ""), str(ent.get("entity_id") or "").strip()) got[key] = _record(ent) return got def _must_stop(exc: BaseException) -> bool: # ★★정지·주인 잃음·예산 초과·마감은 **이 묶음 하나의 실패가 아니다** # (Codex BLOCK 2026-09-17). 여기서 삼키면 재시도가 **정지 뒤에 새로 # 보내고**, 다른 묶음도 계속 돈다. 판정은 한 곳(run_control.is_abort)에. from app.core.research_call_budget import ResearchCallBudgetExceeded from app.core.run_control import is_abort return (is_abort(exc) or isinstance(exc, ResearchCallBudgetExceeded) or bool(getattr(exc, "is_run_deadline", False))) def _run_chunk(ci: int, total: int, rows: List[tuple]) -> Dict[str, Dict]: got: Dict[str, Dict] = {} try: got.update(_ask(rows, "entity_detail_batch", [f"chunk:{ci}/{total}"])) except Exception as exc: if _must_stop(exc): raise logger.warning("Entity detail chunk %d/%d failed: %s", ci, total, exc) miss = _missing(rows, got) if miss: try: got.update(_ask(miss, "entity_detail_retry", ["retry", f"chunk:{ci}/{total}"])) except Exception as exc: if _must_stop(exc): raise logger.warning("Entity detail chunk %d/%d retry failed: %s", ci, total, exc) left = len(_missing(rows, got)) if left: # ★조용히 넘기지 않는다 — 2/241 이 로그 한 줄 없이 partial 로 넘어갔다. logger.warning("Entity detail chunk %d/%d: %d/%d 요소가 끝내 비었다", ci, total, left, len(rows)) return got chunks = [entity_queue[i:i + _DETAIL_CHUNK_SIZE] for i in range(0, len(entity_queue), _DETAIL_CHUNK_SIZE)] logger.info("Entity detail: %d요소 → %d묶음(최대 %d개씩)", len(entity_queue), len(chunks), _DETAIL_CHUNK_SIZE) if len(chunks) <= 1: for ci, rows in enumerate(chunks, 1): entity_details.update(_run_chunk(ci, len(chunks), rows)) else: from concurrent.futures import ThreadPoolExecutor from app.core.research_call_budget import bind_current_research_budget from app.modules.llm.opik_trace import bind_current_trace # ★★정지 표·텍스트 예산·Opik trace 는 **스레드마다 따로**다. 그냥 넘기면 # worker 안에서 정지가 안 들리고 예산이 안 세어지고 기록 계층이 끊긴다 # (Codex BLOCK 2026-09-17). 본보기 = grounding_chunk_step._read_chunks. # ★감싸기는 **여기(부모 스레드)**에서 한다 — 캡처 시점이 부모다. send = bind_current_trace(bind_current_research_budget(_run_chunk)) with ThreadPoolExecutor(max_workers=min(_DETAIL_CHUNK_WORKERS, len(chunks))) as pool: futs = [pool.submit(send, ci, len(chunks), rows) for ci, rows in enumerate(chunks, 1)] # ★제출 순서대로 합친다 — 같은 열쇠가 겹칠 일은 없지만 결과가 실행 순서에 안 흔들리게. try: for f in futs: entity_details.update(f.result()) except BaseException: # 세워야 하는 예외가 올라왔다 — 아직 안 시작한 묶음은 보내지 않는다. for f in futs: f.cancel() raise # ★회계는 **큐의 인스턴스마다**(short_id 별) — 이름이 같은 다른 줄의 답으로 채워지지 않는다 _have = sum( 1 for n, t, *r in entity_queue if ((r and r[0]) and r[0] in entity_details) or (not (r and r[0]) and (f"{n}:{t}" in entity_details or n in entity_details))) return { "completed_count": int(_have), "applicable_count": len(entity_queue), "failed_count": max(0, len(entity_queue) - int(_have)), "config_hash": self._config_hash(), "data": {"entity_queue": entity_queue, "entity_details": entity_details, "identity_contract": ENTITY_INSTANCE_IDENTITY_CONTRACT_VERSION}, } def _config_hash(self) -> str: """★인스턴스 신원 계약 + **실제로 고른 팩**(prompt·schema 의 버전과 원본 bytes 해시)을 지문에 싣는다 (Codex 06:50 NON-BLOCK · 이번 live 뒤). 실측: 제가 만든 팩이 loader 에 안 실렸는데 지문은 그대로라 알 길이 없었다 — 다음 팩 변경 때 옛 CP 가 조용히 되쓰이지 않게.""" from app.core.grounding_mode import resolve_grounding_mode, uses_chunk_producer # ★★legacy/v2 지문은 **옛 값 그대로**(byte-identical) — PR #82 리뷰 2026-09-03: `_identity_hash` 가 mode 검사보다 # 먼저 감싸서 legacy 의 completed CP(md5 16자)가 전부 어긋났고 runner 는 그것을 BLOCK(409)으로 다뤄 기존 # 에피소드의 재개가 막혔다. 신원 계약·팩 digest 는 v2_chunk 에서만 접는다. if not uses_chunk_producer(resolve_grounding_mode(self.project_config)): return super()._config_hash() base = _identity_hash(super()._config_hash()) # ★resolve_effective 가 실제로 고른 **내용 bytes 의 digest**(raw_content_hash)만 접는다 — 디렉터리 번호가 아니다. from app.modules.prompt_loader import resolve_effective digests = [] for stem, kind in (("turn1_7_detail_batch", "prompt"), ("turn1_7_detail_batch_schema", "schema")): r = resolve_effective("entity_extractor_v2", stem, kind=kind) digests.append(f"{stem}:{r.get('raw_content_hash')}") import hashlib return hashlib.sha256((base + "|" + "|".join(digests)).encode("utf-8")).hexdigest() # ── Step 11: entity_t2i ── class EntityT2iStep(_EntityStepMixin, StepRunner): """Step 11: T2I 프롬프트 병렬 생성.""" def _config_hash(self) -> str: """★인스턴스 신원 계약을 지문에 싣는다 — 옛 (name, type) 열쇠로 구운 CP 는 어긋나 다시 굽는다(entity_detail 과 같은 상수). ★legacy/v2 는 옛 값 그대로(byte-identical) — 기존 에피소드의 재개를 막지 않는다(PR #82 리뷰 2026-09-03).""" from app.core.grounding_mode import resolve_grounding_mode, uses_chunk_producer if not uses_chunk_producer(resolve_grounding_mode(self.project_config)): return super()._config_hash() return _identity_hash(super()._config_hash()) def _execute(self, mode="resume") -> Dict[str, Any]: prev = self._load_prev_checkpoint("entity_detail") if not prev or not prev.get("data"): from app.core.errors import AppError raise AppError( code="step.no_input", message="entity_detail 결과 없음", status_code=400, ) entity_queue = prev["data"]["entity_queue"] # 하위 호환: gpt_details → entity_details entity_details = prev["data"].get("entity_details") or prev["data"].get("gpt_details", {}) # ★★**상세가 빠진 것은 `entity_merge` 에서 메운다** (2026-09-21). # # 왜 필요한가 — 실측: `entity_detail` 이 다섯 묶음 중 한 호출에서 # 50개 중 1개만 쓰고 끝냈다(`finish_reason=stop`, 재시도 3회 동일). # 그 48개는 `description`·`visual_traits` 가 **빈 채로** 정본까지 # 내려갔고, 소품 참조 문안은 `{entity_description}` 하나로 끝나는 # 템플릿이라 **무엇을 그릴지 없는 상품 사진 지시**가 나간다 # (`prop_ref.md`). 장소 설명은 시대 조사 입력으로도 간다. # # ★**모델이 지어낸 말로 메우지 않는다.** 이 스텝의 LLM 출력은 # 여전히 안 쓴다(`feedback_no_silent_fallback`). 메우는 것은 # **앞 단계가 대본에서 뽑아 둔 것**(`entity_merge`)이고, 그 # 사실을 `description_source` 로 남긴다. # ★상세가 있으면 **상세가 이긴다** — 이 길은 빈 칸에만 닿는다. _merge_cp = self._load_prev_checkpoint("entity_merge") or {} # ★갈래 표는 **계약 한 곳**에서 온다(`_ENTITY_KEYS`) — 여기서 손으로 # 적으면 네 번째 갈래가 통째로 사라진다. 그리고 **그 갈래가 곧 # 소유 종류**라, 행에 `entity_type` 이 없는 옛 merge 판도 종류를 # 얻는다(Codex BLOCK 2026-09-21). _mf_by_sid: Dict[str, Dict[str, Any]] = {} #: 열쇠 → (rec, 원자료 short_id) · ★**모호하면 `None`**. #: 같은 이름 다른 short_id 가 둘이면 **아무도 못 쓴다** — 이름으로 #: 남의 설명을 집어 오면 신원이 어긋난 채 정본에 실린다 #: (`ENTITY_INSTANCE_IDENTITY_CONTRACT` 와 같은 이유). _mf_by_name: Dict[str, Optional[Any]] = {} for _kind, _owner in _ENTITY_KEYS(): for _e in ((_merge_cp.get("data") or {}).get(_kind) or []): if not isinstance(_e, dict): continue _nm = str(_e.get("name") or "").strip() if not _nm: continue # ★설명이 **없는 행도 색인한다**. 「어느 개체인가」와 「쓸 # 설명이 있는가」는 다른 물음이다 — 빈 설명 행을 여기서 # 버리면 같은 이름 후보가 둘인데 하나로 세어져 모호성 # 봉인이 안 걸린다(Codex BLOCK 2026-09-21 둘째). # 설명이 없으면 **고른 뒤에** 빈 것을 돌려준다. # ★`visual_traits` 가 **문자열**로 저장된 판이 있다(`'[]'`). # 그대로 실으면 하류가 목록으로 읽다 깨진다 — 여기서 푼다. _vt = _e.get("visual_traits") or [] if isinstance(_vt, str): try: _vt = json.loads(_vt) except Exception: # noqa: BLE001 _vt = [] _rec = { "description": _e.get("description"), "visual_traits": _vt if isinstance(_vt, list) else [], "description_source": "entity_merge", } _sid = str(_e.get("short_id") or "").strip() _et = str(_e.get("entity_type") or "").strip() or _owner if _sid: _mf_by_sid[_sid] = _rec # ★열쇠는 **(이름, 종류)** 하나다. 이름 단독은 **두지 # 않는다** — 갈래에서 종류를 얻어 놓고 마지막에 이름으로 # 내려가면 그 종류를 다시 버리는 것이다(장소 설명이 # 부분 장소 정본에 실린다. Codex BLOCK 2026-09-21 첫째). _key = f"{_nm}:{_et}" _prev = _mf_by_name.get(_key, "∅") if _prev == "∅": _mf_by_name[_key] = (_rec, _sid) elif _prev is not None and _prev[0] is not _rec: _mf_by_name[_key] = None # ★모호 — 봉인 def _merge_source(ename, etype, sid=""): """`entity_merge` 원자료 — **신원이 맞을 때만** 돌려준다. ★short_id 가 신원이다(`ENTITY_INSTANCE_IDENTITY_CONTRACT`). 대상에 short_id 가 있고 원자료 쪽도 제 short_id 를 가졌는데 서로 다르면 **다른 개체**다 — 이름이 같아도 안 쓴다. ★(이름, 종류)가 원자료에서 **유일하지 않으면** 빈 채로 둔다. 모호한 자리를 메우는 것보다 비워 두는 것이 낫다. ★이름 **단독**으로는 안 찾는다 — 종류가 다르면 다른 개체다. """ def _usable(rec): """설명이 **있을 때만** 원자료로 친다 — 없으면 「없음」.""" if rec and str(rec.get("description") or "").strip(): return rec return {} if sid: got = _mf_by_sid.get(sid) if got: return _usable(got) if not etype: return {} # ★종류를 모르면 못 고른다 pair = _mf_by_name.get(f"{ename}:{etype}") if not pair: return {} # 없음 · 또는 **모호** rec, src_sid = pair if sid and src_sid and src_sid != sid: return {} # ★다른 개체의 설명이다 return _usable(rec) def _source_detail(ename, etype, sid=""): """이 엔티티의 **원자료**. 상세가 이기고, 없으면 `entity_merge`. ★열쇠 순서는 조립부와 같다(short_id → 이름:종류 → 이름). ★빈 dict 는 「없음」이다 — 그때만 대체 원천으로 간다. """ got = ((entity_details.get(sid) if sid else None) or entity_details.get(f"{ename}:{etype}") or entity_details.get(ename, {})) if got: return got return _merge_source(ename, etype, sid) # ★★★열쇠는 **(이름, 갈래)** 다 — 이름 하나로 열쇠를 쓰면 같은 이름의 # 장소와 그 장소의 부분이 **서로 덮는다**(Codex 재현 2026-09-01). # sync 는 `(entity_type, name)` 계약을 지키는데 그 앞에서 무너졌다. # ★옛 열쇠도 남긴다 — 이 스텝 밖(하위 호환 소비자)이 이름으로 읽는다. name_to_sid = {} for item in entity_queue: if len(item) >= 3 and item[2]: name_to_sid[_qkey(item[0], item[1])] = item[2] name_to_sid.setdefault(item[0], item[2]) # resume: 이미 완료된 것 스킵 cp = self.load_checkpoint() done = ( cp.get("data", {}).get("completed", {}) if cp and mode == "resume" else {} ) # ★현재 대상 밖의 완료분은 버린다 (2026-08-07 실측·Codex 3차 리뷰). # # 체크포인트가 archive 에서 자동 복원될 수 있고(상류 force 가 하류를 # 지운 뒤에도 그랬다), 그때 복원본에는 **지금은 없는 엔티티**의 완료 # 기록이 들어 있다. 그것을 그대로 받으면 `len(done)` 이 현재 대상 # 수를 넘어 `failed_count = total - len(done)` 가 음수가 되고 # (실측 154/97, failed=-57), 그 partial 이 entity sync 의 stale # cleanup 을 막아 UniqueViolation 으로 파이프라인이 멈췄다. # # 표식 층(`invalidate_downstream` 의 marker)과 **둘 다** 막는다 — # 한 층만 막으면 다른 경로로 같은 오염이 들어온다. # ★★★열쇠가 **두 모양**이다 — 새 CP 는 `_qkey`, 옛 CP 는 이름 하나. # 판정을 한 모양으로만 하면 다른 모양이 통째로 stale 이 되어 # **끝낸 유료 호출을 다시 산다**(Codex 재현 2026-09-01 — 내가 새 # 열쇠를 넣고 이 자리를 안 고쳤다). want_q = {_qkey(n, t) for (n, t, *_r) in entity_queue} # ★새 열쇠 — short_id. 같은 (name, type) 이 둘 이상이면 이름 열쇠로는 어느 쪽인지 모른다. _sids = {str(_r[0]) for (_n, _t, *_r) in entity_queue if _r and _r[0]} from collections import Counter as _Counter _pair_count = _Counter((n, t) for (n, t, *_r) in entity_queue) # ★★★**짝의 집합**이지 이름→갈래 표가 아니다. 표로 두면 같은 이름의 # 장소와 그 부분에서 **뒤엣것이 앞엣것을 덮어** 멀쩡히 끝낸 쪽이 # stale 로 지워지고 재개가 그 유료 호출을 다시 산다 # (Codex 재현 2026-09-01). want_pairs = {(n, t) for (n, t, *_r) in entity_queue} def _pair_of(_key, _val): """열쇠(short_id · _qkey · 이름)에서 (name, type) 짝을 되찾는다.""" got = str((_val or {}).get("entity_type") or "") if "\u0000" in str(_key): _t, _n = str(_key).split("\u0000", 1) return (_n, _t) _nm = str((_val or {}).get("name") or _key) return (_nm, got) def _kept(_key, _val) -> bool: if _key in _sids: return True # ★short_id 열쇠 — 그 인스턴스가 지금 대상에 있다 pair = _pair_of(_key, _val) if _key in want_q or pair in want_pairs: # ★같은 짝이 둘 이상이면 옛 값이 **자기 short_id 를 밝힌 것**만 되쓴다 — 아니면 어느 쪽인지 모른다 if _pair_count.get(pair, 0) > 1: return str((_val or {}).get("short_id") or "") in _sids return True return False if done: stale = [k for k, v in done.items() if not _kept(k, v)] if stale: logger.warning( "entity_t2i: 현재 대상에 없는 완료 기록 %d건 폐기 " "(archive 복원 잔존 추정) — 표본 %s", len(stale), stale[:8], ) done = {k: v for k, v in done.items() if _kept(k, v)} # ★★**이미 끝난 기록의 빈 칸은 메운다 — 다시 사지 않는다** (2026-09-21). # # `entity_detail` 이 한 묶음을 통째로 못 내면 그 엔티티들은 # `description`·`visual_traits` 가 **빈 채로** 정본까지 내려간다. # 그러면 소품 참조가 깨진다 — 문안 템플릿이 `{entity_description}` # 하나로 끝나고(`prop_ref.md`), `ref_image_pipeline` 은 **템플릿이 # 있으면 `t2i_prompt` 를 안 쓴다**. 장소 설명은 시대 조사로도 간다. # # ★**다시 만들지 않는다.** `t2i_prompt` 는 이미 있고 쓸 만하다 — # 깨진 것은 설명 칸뿐이라 그 칸만 메운다(LLM 호출 0). # ★메우는 것은 모델이 지어낸 말이 아니라 **앞 단계가 대본에서 뽑아 # 둔 것**(`entity_merge`)이고, 출처를 `description_source` 로 남긴다. # ★원자료가 없으면 **그대로 둔다** — 없는 것을 지어내지 않는다. if done and (_mf_by_sid or _mf_by_name): _filled = [] for _k, _v in list(done.items()): if not isinstance(_v, dict): continue if str(_v.get("description") or "").strip(): continue # ★기록이 **이미 비어 있다**는 것은 그 원천에 설명이 없었다는 # 뜻이다. 그러니 상세를 다시 거치지 않고 대체 원천을 바로 # 본다 — `_source_detail` 을 태우면 빈 상세가 이겨서 메우기가 # 통째로 안 돈다(내 시험이 이 실수를 잡았다). _nm = str(_v.get("name") or "") _et = str(_v.get("entity_type") or "") _sd = str(_v.get("short_id") or "") _src = _merge_source(_nm, _et, _sd) if not str(_src.get("description") or "").strip(): continue # ★특징이 **이미 있으면 그대로 둔다** — 빈 칸만 메우는 # 일인데 멀쩡한 특징을 원자료로 갈아치우면 안 된다 # (Codex NON-BLOCK 2026-09-21). _keep_vt = _v.get("visual_traits") or [] done[_k] = { **_v, "description": _src.get("description") or "", "visual_traits": (_keep_vt if _keep_vt else (_src.get("visual_traits") or [])), "description_source": "entity_merge", } _filled.append(_k) if _filled: logger.warning( "entity_t2i: 설명이 빈 완료 기록 %d건을 `entity_merge` " "원자료로 메웠다(호출 0) — 표본 %s", len(_filled), _filled[:8]) # ★재개 열쇠도 **(이름, 갈래)** 다. 옛 체크포인트는 이름만 갖고 # 있으므로 둘 다 본다 — 그래야 재개가 안 깨진다. def _already(_n, _t, _sid=""): if _sid and _sid in done: return True # ★새 CP — short_id 열쇠 got = done.get(_qkey(_n, _t)) if got is None: got = done.get(_n) if got is not None and str((got or {}).get("entity_type") or "") != _t: got = None if got is None: return False # ★옛 열쇠 — 같은 짝이 둘 이상이면 옛 값의 short_id 가 **이 인스턴스**일 때만 되쓴다 if _pair_count.get((_n, _t), 0) > 1: return bool(_sid) and str((got or {}).get("short_id") or "") == _sid return True remaining = [ (i, n, t, (r[0] if r else "")) for i, (n, t, *r) in enumerate(entity_queue) if not _already(n, t, (r[0] if r else "")) ] total = len(entity_queue) from app.modules.pipeline.entity_extractor_v3 import ( ENTITY_DETAIL_SCHEMA, _load_system, _load_prompt, ) system_prompt = _load_system() # visual_world_rules에서 era/region + costume rules 로드 vwr_cp = self._load_prev_checkpoint("visual_world_rules") vwr_data = vwr_cp.get("data", {}) if vwr_cp else {} era = vwr_data.get("era", "") region = vwr_data.get("region", "") world_context = "" if era or region: world_context = f"\n\n[세계관]\n시대: {era}\n지역: {region}" for r in vwr_data.get("rules", []): if r.get("rule_type") == "costume": world_context += f"\n의상규칙: {r.get('description', '')}" t2i_ctx = vwr_data.get("t2i_context", "") if t2i_ctx: world_context += f"\n\n[T2I 시각 컨텍스트]\n{t2i_ctx}" def _gen_t2i(idx, ename, etype, sid=""): ent_detail = _source_detail(ename, etype, sid) # 덧붙임 블록을 둘로 나눠 둔다 — world_context 는 프로젝트·주행 # 단위로 고정이고 detail_block 은 호출마다 변한다. 축 A1 재배열이 # 그 경계를 쓴다. 이어 붙이는 순서는 예전과 같다(OFF=무변화). detail_block = "" if ent_detail: desc = ent_detail.get("description", "") traits = ", ".join(ent_detail.get("visual_traits", [])) detail_block += ( f"\n\n[시나리오 기반 상세 정보]\n" f"설명: {desc}\n시각적 특징: {traits}" ) # ② P0a generic fix (2026-06-18): evidence-gated distinctive 외형. # literal_visual + source_quote 항목만 별도 블록으로 t2i context weave. _dblock, _excluded = build_distinctive_trait_block( ent_detail.get("distinctive_visual_traits") ) detail_block += _dblock # ★크기 견줌도 같은 자리에서 싣는다 (2026-09-20). _sblock, _s_excluded = build_build_scale_block( ent_detail.get("build_scale") ) detail_block += _sblock if _s_excluded: logger.info( "entity_t2i %s: build_scale %d개 렌더 제외 " "(근거 인용 또는 견줌 문장 없음): %s", ename, len(_s_excluded), [{"compared_to": d.get("compared_to"), "relation": d.get("relation"), "has_quote": bool( (d.get("source_quote") or "").strip())} for d in _s_excluded]) if _excluded: logger.info( "entity_t2i %s: distinctive_visual_traits %d개 렌더 제외 " "(evidence-gate, literal_visual+source_quote 아님): %s", ename, len(_excluded), [ { "trait": d.get("trait"), "kind": d.get("interpretation_kind"), "has_quote": bool((d.get("source_quote") or "").strip()), "reason": d.get("interpretation_reason"), } for d in _excluded ], ) turn_msg = assemble_user_prompt( "entity_t2i", render=lambda **kw: _load_prompt("turn_entity_detail", **kw), fields={"entity_name": ename, "entity_type": etype}, project_block=world_context, call_block=detail_block, ) for attempt in range(3): try: detail = call_structured( step="entity_t2i", system_prompt=system_prompt, user_prompt=turn_msg, response_schema=ENTITY_DETAIL_SCHEMA, project_config=self.project_config, schema_name="entity_t2i", opik_metadata=self.build_opik_metadata(), ) # description / visual_traits 는 source detail # (entity_detail 결과) 에서 forward — LLM 의 unsourced # trait 이 final output 에 도입되는 path 를 봉쇄. ent_detail # 이 빈 dict 라도 LLM detail 로 silent fallback 안 함 # (feedback_no_silent_fallback). LLM 출력의 description / # visual_traits 는 무시. t2i_prompt 만 LLM 출력에서 take — # entity_t2i 의 단일 책임. short_description 은 후속 fix 로 # schema 및 출력 dict 에서 제거 (Codex MIN 1, consumer 0). # # D6 T2 (B4 #2): metadata_json 은 LLM 출력에서 take. # entity_detail batch 가 만들지 않는 field 라 source-forward # 정책 외 — location 의 space_profile 분류는 entity_t2i LLM # 호출 시점에 system prompt 의 schema 가이드대로 LLM 이 출력. # T2-fix (I3): schema required 라 항상 {} 또는 dict 보장. # I7: character/prop 은 prompt 가 `{}` 출력 명시 (location 만 # 의미 있음). EntitySyncService 가 entity_type 으로 분기. src = ent_detail or {} md = detail.get("metadata_json") or {} # Area B (Task 3 / C3) + D6 carry: location + prop 양쪽 # metadata SOT post-validation. 3 attempts 모두 fail 시 done # 에 안 넣음 → failed_count 증가. character 는 semantic # post-validation 부재 (기존 빈-marker 패턴 유지 — image gen # preflight 가 catch). # # location 의 D6 space_profile 내부 검증은 # validate_entity_metadata_shape helper 안에서 통합 호출 # (Task 2 fix commit 8383c20). 본 site 별도 호출 불필요 — # single point of validation. SpaceProfileError 는 raise 시 # attempt loop 의 except 가 catch → retry. # ★★2026-09-01 — `location_part` 도 검증한다. 앞 판은 # 성공 응답의 `metadata_json` 을 **검증 없이** 통과시켜, # 틀린 모양이 sync 까지 갔다(Codex 재현). # ★갈래 목록을 손으로 안 적는다 — 검증기가 아는 것을 쓴다. # ★★중립 모양 갈래(character·location_part)는 검증 **앞**에서 # 계약 모양으로 접는다 (실측 2026-09-02: location_part 에 # 모델이 dict 를 지어내 3회 실패 → partial · 결정적). # 값이 곧 SOT 인 prop·location 은 그대로 검증한다. from app.core.entity_metadata import ( normalize_metadata_for_type, ) md = normalize_metadata_for_type(etype, md) if etype in _VALIDATED_TYPES(): from app.core.entity_metadata import ( validate_entity_metadata_shape, ) from app.core.errors import AppError from app.core.bg_state_vocab import SpaceProfileError try: validate_entity_metadata_shape( etype, md, short_id=sid or name_to_sid.get(_qkey(ename, etype), "") or ename, ) except (AppError, SpaceProfileError) as exc: # Area B Task 3 quality review M2 fix: # location 의 D6 space_profile 위반은 SpaceProfileError — # `except AppError` 만으로는 catch 안 됨 → outer Exception # 으로 떨어져 log warning 누락 (silent retry). 양쪽 모두 # catch + warn + raise 로 logging 대칭 보장. exc_msg = getattr(exc, "message", None) or str(exc) logger.warning( "entity_t2i %s post-validate failed (attempt %d/3) " "for %r: %s — retry", etype, attempt + 1, ename, exc_msg, ) raise return ( idx, ename, etype, { "name": ename, "short_id": sid or name_to_sid.get(_qkey(ename, etype), ""), "description": src.get("description", ""), "visual_traits": _traits_with_scale(src), "t2i_prompt": detail.get("t2i_prompt", ""), # 원자료는 그대로 남긴다 — 근거와 렌더 문장은 다르다. "build_scale": src.get("build_scale", []), "metadata_json": md, # ★어디서 온 설명인가 — 상세가 빠져 `entity_merge` # 로 메운 것이면 그 사실을 남긴다(감사). **({"description_source": "entity_merge"} if src.get("description_source") == "entity_merge" else {}), }, ) except Exception: if attempt < 2: time.sleep(2 * (attempt + 1)) # 3 attempts 모두 실패: source detail 보존 + t2i_prompt 빈 문자열 # marker. downstream 이 빈 prompt 를 빈 image 로 흘리지 않도록 # consumer 는 check 필요 (image gen 단계의 ref_image_pipeline 가 # asset_readiness preflight 로 catch 가능). # # Area B (Task 3 / C3) + D6 carry: location + prop entity 는 failure # 시 fail-fast — SOT (space_profile / visual_identity) 손실은 silent # absorb 안 함. data=None 반환으로 caller 가 done 에 안 넣고 # failed_count 증가. character 는 기존 빈-marker pattern 유지 # (image gen preflight 가 catch). if etype in _VALIDATED_TYPES(): return (idx, ename, etype, None) src = ent_detail or {} # Area B (Task 4 review C1 fix): character failure marker 도 normalized # closed shape `{"location": None, "visual_identity": None}` 출력 — Task 4 # sync (Boundary 2) 의 strict shape contract 정합. 빈 dict `{}` 는 sync # 의 keys check 에서 fail-fast 발생 → character LLM 실패 시 entire # episode fail cascade 위험 차단. 의미는 동일 — t2i_prompt="" marker 가 # image preflight 의 catch path 보존. return ( idx, ename, etype, { "name": ename, "short_id": name_to_sid.get(_qkey(ename, etype), ""), "description": src.get("description", ""), "visual_traits": src.get("visual_traits", []) if isinstance(src.get("visual_traits"), list) else [], "t2i_prompt": "", "metadata_json": {"location": None, "visual_identity": None}, }, ) if remaining: max_workers = min(10, len(remaining)) with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = {} _sid_of_idx = {i: sid for (i, _n, _t, sid) in remaining} for i, ename, etype, sid in remaining: if futures: time.sleep(1) futures[executor.submit(_gen_t2i, i, ename, etype, sid)] = i for future in as_completed(futures): result = future.result() idx, ename, etype, data = result # T2-fix (review iter3 I1): location post-validation 실패 시 # data=None — done 에 안 넣음 → failed_count 증가 (line 851). if data is None: logger.error( "entity_t2i: %r (%s) failed all 3 attempts — " "excluded from done (failed_count++)", ename, etype, ) self.update_progress(len(done), total) continue data["entity_type"] = etype # ★열쇠는 **short_id**(있으면) — 같은 이름의 다른 실체가 덮지 않게. 없으면(legacy) (이름, 갈래). done[_ikey(ename, etype, _sid_of_idx.get(idx, ""))] = data self.update_progress(len(done), total) # 증분 체크포인트 (crash resume용) — 분류 배열도 포함 self.save_checkpoint( { "status": "running", "data": { "completed": done, "entity_queue": entity_queue, **_by_lane(done), }, } ) return { "completed_count": len(done), "applicable_count": total, "failed_count": total - len(done), "config_hash": self._config_hash(), # ★저장과 비교가 같은 지문(신원 계약 포함) # ★sync 가 읽는 CP 는 **여기**다 — 표식을 최상위로 나른다 **_carry_chunk_marker(self), "data": {"completed": done, **_by_lane(done), "identity_contract": ENTITY_INSTANCE_IDENTITY_CONTRACT_VERSION}, }