"""BackgroundPlannerStep — 에피소드 단위 배경 생성 플랜 (Phase 5 / Task 3). mode=floor_plan_anchored 시에 단일 LLM 호출로 - frequency 분석(저빈도 location 제외, ≥3 shot만 floor_plan) - building group 그루핑 - floor_plan_order / chain_bg_order 결정 - prev_shot_only fallback 분류 을 한꺼번에 결정한다. set_design / location-loop chain_bg_planning을 대체. 흐름: 1. 의존 체크포인트 6개 로드 (shot_validator/shot_selection/scene_director/ entity_merge/entity_detail/visual_world_rules) 2. selected shot 추출 (per scene) 3. location_lines + 전체 short_id 수집 (entity_merge가 source-of-truth, kind는 entity_detail에서 보강 가능. 누락 시 'unknown'.) 4. scene → primary_location 매핑 5. all_shot_ids 수집 (S{scene}_Shot{idx}) 6. build_planner_user_prompt → run_background_planner (3회 retry) 토글: - settings.background_mode == 'floor_plan_anchored' → 활성 - 그 외(off / chain_only) → applicable_count=0 즉시 반환 체크포인트 data 구조: - rationale_summary, floor_plans, floor_plan_order, chain_bg_groups, chain_bg_order, prev_shot_only (run_background_planner 결과 그대로) """ from __future__ import annotations import json import logging from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from app.core.step_runner import StepRunner logger = logging.getLogger(__name__) # Phase 5.2: needs_floor_plan boolean per fp (LLM-judged complexity). # v2→v3 prompt 롤로 stale checkpoint 자동 invalidation. SCHEMA_VERSION = 3 PROMPT_VERSION = "3" class BackgroundPlannerStep(StepRunner): """단일 LLM 호출로 episode 단위 배경 생성 플랜을 결정.""" def _config_hash(self) -> str: """SCHEMA_VERSION + PROMPT_VERSION 포함한 config_hash. resume 시 stale plan 회피.""" import hashlib, json as _json from app.core.config import settings payload = { "background_mode": settings.background_mode, "model": settings.openai_model, "schema_version": SCHEMA_VERSION, "prompt_version": PROMPT_VERSION, } return hashlib.sha256( _json.dumps(payload, sort_keys=True).encode("utf-8") ).hexdigest()[:16] # ── checkpoint loader (location_floor_plan / chain_bg_planning 패턴 동일) ── def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]: 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(): try: return json.loads(cp.read_text(encoding="utf-8")) except Exception as exc: logger.warning("background_planner: %s checkpoint parse failed: %s", step_id, exc) return None return None # ── _execute ── def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings # toggle: floor_plan_anchored가 아니면 즉시 not-applicable처럼 0 반환. # check_applicability(if_floor_plan_mode)가 not_applicable로 미리 거를 수도 있지만, # 직접 호출(테스트/스크립트)이나 manifest이 always인 환경에서도 안전하도록 이중 가드. if settings.background_mode != "floor_plan_anchored": logger.info( "background_planner: skipped — background_mode=%s (not floor_plan_anchored)", settings.background_mode, ) return { "applicable_count": 0, "completed_count": 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {}, } # 1. 의존 체크포인트 로드 shot_validator_cp = self._load_prev_checkpoint("shot_validator") shot_selection_cp = self._load_prev_checkpoint("shot_selection") scene_director_cp = self._load_prev_checkpoint("scene_director") entity_merge_cp = self._load_prev_checkpoint("entity_merge") entity_detail_cp = self._load_prev_checkpoint("entity_detail") rules_cp = self._load_prev_checkpoint("visual_world_rules") # scene_save segments — building_group 추론(어느 location들이 물리적으로 # 연결되는지)을 위해 LLM에 원본 시나리오 본문을 무절단 전달. scene_save_cp = self._load_prev_checkpoint("scene_save") # 2. selected shot 추출 selected_shots_by_scene = _extract_selected_shots( shot_validator_cp, shot_selection_cp, ) # 3. location lines + 전체 short_id (kind 보강 from entity_detail) location_lines, all_location_ids = _build_location_lines( entity_merge_cp, entity_detail_cp, ) # 4. scene → primary_location scene_primary: Dict[int, str] = {} if scene_director_cp: for sc in (scene_director_cp.get("data", {}) or {}).get("scenes", []) or []: si = sc.get("scene_index") primary = sc.get("primary_location", "") or "" if si is not None: scene_primary[int(si)] = primary # 5. all_shot_ids all_shot_ids: List[str] = [] for sc_idx in sorted(selected_shots_by_scene.keys()): for sh in selected_shots_by_scene[sc_idx]: shi = sh.get("shot_index", 0) all_shot_ids.append(f"S{sc_idx}_Shot{shi}") # selected shot이 0개면 LLM 호출도 생략 — invariant 10에 따라 빈 plan 반환. # 빈 input에 LLM을 부르면 비용/시간 낭비 + retry 폭주 위험. if not all_shot_ids: logger.info("background_planner: no selected shots — returning empty plan") empty_plan: Dict[str, Any] = { "rationale_summary": "선택된 shot이 없어 배경 생성 플랜을 비웁니다.", "floor_plans": [], "floor_plan_order": [], "chain_bg_groups": [], "chain_bg_order": [], "prev_shot_only": [], } return { "applicable_count": 1, "completed_count": 1, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": empty_plan, } # 6. user_prompt # CLAUDE.md no-truncation: visual_world_rules는 절대 자르지 않는다 (전문 그대로). rules_text = "" if rules_cp: data = rules_cp.get("data", {}) or {} rules_text = ( data.get("rules_text", "") or data.get("text", "") or _summarize_rules_data(data) or "" ) from app.modules.pipeline.background_planner import ( build_planner_user_prompt, run_background_planner, PlannerError, ) from app.modules.llm.llm_client import call_structured # scene_save.segments[] — 무절단 본문 (CLAUDE.md absolute rule) scene_segments: List[Dict[str, Any]] = [] if scene_save_cp: ss_data = scene_save_cp.get("data", {}) or {} scene_segments = list(ss_data.get("segments", []) or []) user_prompt = build_planner_user_prompt( selected_shots_by_scene=selected_shots_by_scene, location_lines=location_lines, visual_world_rules=rules_text, scene_primary_locations=scene_primary, scene_segments=scene_segments, ) # 7. LLM 호출 — PlannerError(retry exhausted)는 step을 fail이 아닌 partial로 보고 # (graceful degradation): 사용자가 에피소드 데이터 보강 후 force 재실행 가능. try: plan = run_background_planner( project_config=self.project_config, user_prompt=user_prompt, location_short_ids=all_location_ids, shot_ids=all_shot_ids, opik_metadata=self.build_opik_metadata(), call_structured_fn=call_structured, ) except PlannerError as exc: logger.error("background_planner: planner exhausted retries: %s", exc) return { "applicable_count": 1, "completed_count": 0, "failed_count": 1, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": { "error": str(exc), "rationale_summary": "", "floor_plans": [], "floor_plan_order": [], "chain_bg_groups": [], "chain_bg_order": [], "prev_shot_only": [], }, } return { "applicable_count": 1, "completed_count": 1, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": plan, } # ────────────────────────────────────────────── # 모듈 레벨 helper (step 인스턴스 의존 없음 → 직접 테스트 가능) # ────────────────────────────────────────────── def _extract_selected_shots( shot_validator_cp: Optional[Dict[str, Any]], shot_selection_cp: Optional[Dict[str, Any]], ) -> Dict[int, List[Dict[str, Any]]]: """shot_validator + shot_selection → {scene_index: [shot_dict, ...]}. 각 shot_dict에는 shot_index + description 포함 (가능하면 location_id도). 의존 체크포인트가 누락되면 빈 dict를 반환 — caller가 빈 plan으로 graceful 처리. """ if not shot_validator_cp or not shot_selection_cp: return {} sel_map: Dict[int, set] = {} for s in (shot_selection_cp.get("data", {}) or {}).get("scenes", []) or []: si = s.get("scene_index") if si is None: continue sel_map[int(si)] = set(s.get("selected_shot_indices", []) or []) selected: Dict[int, List[Dict[str, Any]]] = {} for s in (shot_validator_cp.get("data", {}) or {}).get("scenes", []) or []: si = s.get("scene_index") if si is None: continue si_int = int(si) sel = sel_map.get(si_int, set()) if not sel: continue for sh in s.get("shots", []) or []: shi = sh.get("shot_index") if shi is None or shi not in sel: continue entry: Dict[str, Any] = { "shot_index": shi, "description": sh.get("description", "") or "", } loc_id = sh.get("location_id", "") if loc_id: entry["location_id"] = loc_id selected.setdefault(si_int, []).append(entry) # shot_index 안정 정렬 (체크포인트가 이미 정렬되어 있어도 방어) if si_int in selected: selected[si_int].sort(key=lambda x: x.get("shot_index", 0)) return selected def _build_location_lines( entity_merge_cp: Optional[Dict[str, Any]], entity_detail_cp: Optional[Dict[str, Any]], ) -> Tuple[List[str], List[str]]: """entity_merge.locations[] → planner LLM에 줄 location_lines + 전체 short_id 리스트. entity_merge는 source-of-truth (short_id, name). entity_detail이 있으면 short_id 매칭으로 kind/description을 보강. kind 누락 시 'unknown' (LLM은 spec상 unknown을 outdoor로 처리). Returns: (location_lines, all_location_ids) — 둘 다 short_id 알파벳 정렬. """ if not entity_merge_cp: return [], [] locations = (entity_merge_cp.get("data", {}) or {}).get("locations", []) or [] # entity_detail에서 short_id → {kind, description} 매핑 detail_by_short: Dict[str, Dict[str, str]] = {} if entity_detail_cp: ed = entity_detail_cp.get("data", {}) or {} # entity_detail의 데이터 형태가 여러 변종 있을 수 있어 fallback 체이닝. candidates: List[Dict[str, Any]] = [] for key in ("locations", "entities", "items", "all"): val = ed.get(key) if isinstance(val, list): candidates.extend(val) for item in candidates: if not isinstance(item, dict): continue sid = item.get("short_id") or "" if not sid or not sid.startswith("L"): continue detail_by_short[sid] = { "kind": (item.get("kind") or item.get("location_kind") or "") or "", "description": item.get("description", "") or "", } lines: List[str] = [] short_ids: List[str] = [] for loc in locations: sid = loc.get("short_id") or "" name = loc.get("name") or sid if not sid: continue short_ids.append(sid) kind = ( detail_by_short.get(sid, {}).get("kind", "").strip() or loc.get("kind", "").strip() or "unknown" ) lines.append(f"{sid} ({kind}): {name}") # 안정 정렬: short_id 알파벳순 paired = sorted(zip(short_ids, lines), key=lambda t: t[0]) short_ids_sorted = [s for s, _ in paired] lines_sorted = [l for _, l in paired] return lines_sorted, short_ids_sorted def _summarize_rules_data(data: Dict[str, Any]) -> str: """visual_world_rules 결과 dict에서 사용 가능한 텍스트 발췌. upstream 모듈은 `{rules: [...], era, region}` 형태로 저장. rules_text 필드 없는 환경 호환을 위해 직접 합성한다. CLAUDE.md no-truncation: 모든 rule 항목 전체 본문 포함 (자르지 않음). """ rules = data.get("rules") or [] if not rules: return "" lines: List[str] = [] era = data.get("era", "") region = data.get("region", "") if era or region: lines.append(f"era={era} | region={region}") for r in rules: if not isinstance(r, dict): continue rt = r.get("rule_type", "") desc = r.get("description", "") guide = r.get("visual_guideline", "") bullet = f"- [{rt}] {desc}" if guide: bullet += f" / {guide}" lines.append(bullet) return "\n".join(lines)