"""background_planner — 에피소드 단위 배경 생성 플랜 (Phase 5). set_design / location-loop chain_bg_planning 교체용 신규 step. - 입력: 선택된 shots + 모든 location + visual_world_rules - 출력: floor_plans + chain_bg_groups + prev_shot_only (전부 LLM 결정) - 모델: gpt-5.5 (text 분석 주력) 흐름: 1. build_planner_user_prompt — 입력을 user prompt로 직렬화 2. _inject_runtime_enums — schema에 location_short_ids / shot_ids enum 주입 3. call_structured_fn — gpt-5.5 호출 (3회 retry) 4. validate_planner_output — 10개 semantic invariant 검증 """ from __future__ import annotations import copy import logging import re import time from typing import Any, Callable, Dict, List, Optional from app.modules.prompt_loader import load_prompt, load_schema logger = logging.getLogger(__name__) _MODULE = "background_planner" # 한글/한자/kana 검출 — universal noun 검증용 (background_chain_planning과 동일). _NON_ASCII_TEXT_RE = re.compile( r"[ㄱ-ㆎ가-힣" r"一-鿿㐀-䶿豈-﫿" r"぀-ヿ]" ) class PlannerError(Exception): """background_planner 호출이 retry 한도까지 실패했을 때 발생.""" # ────────────────────────────────────────────── # user prompt 빌더 # ────────────────────────────────────────────── def build_planner_user_prompt( selected_shots_by_scene: Dict[int, List[Dict[str, Any]]], location_lines: List[str], visual_world_rules: str, scene_primary_locations: Dict[int, str], scene_segments: Optional[List[Dict[str, Any]]] = None, ) -> str: """planner LLM에 보낼 user prompt 빌드. 각 shot 라인은 가능한 경우 `(loc=Lxx)` 마커를 포함한다 — system prompt가 이 마커를 권위(authoritative) location attribution으로 사용한다. 마커가 없으면 LLM은 scene primary_location으로 fallback한다. Args: selected_shots_by_scene: scene_index → [{shot_index, description, location_id?}, ...] location_lines: ["L01 (indoor): 옥탑방", "L02 (outdoor): 골목", ...] visual_world_rules: 에피소드의 visual_world_rules 텍스트 전문 (절대 자르지 않음) scene_primary_locations: scene_index → primary location short_id (예: "L01") scene_segments: scene_save.data.segments 원본 — 무절단 텍스트. [{scene_index, heading, text, ...}, ...] 형태. 빈/None이면 빈 블록. Returns: format된 user prompt 문자열 """ template = load_prompt(_MODULE, "user_template") scene_blocks: List[str] = [] for scene_idx in sorted(selected_shots_by_scene.keys()): shots = selected_shots_by_scene[scene_idx] primary = scene_primary_locations.get(scene_idx, "") block_lines: List[str] = [ f"Scene {scene_idx} (primary_location={primary}):" ] for sh in shots: shot_idx = sh.get("shot_index", 0) description = sh.get("description", "") or "" # (loc=Lxx) 마커 — shot에 location_id가 있으면 우선, 없으면 scene primary loc_marker = sh.get("location_id") or primary or "" if loc_marker: block_lines.append( f" S{scene_idx}_Shot{shot_idx} (loc={loc_marker}): {description}" ) else: block_lines.append( f" S{scene_idx}_Shot{shot_idx}: {description}" ) scene_blocks.append("\n".join(block_lines)) # scene_texts_block — 원본 시나리오 segment 무절단 inject (CLAUDE.md absolute rule) scene_texts_lines: List[str] = [] for seg in (scene_segments or []): si = seg.get("scene_index") if not isinstance(si, int): continue heading = (seg.get("heading") or "").strip() text = (seg.get("text") or "").strip() if heading: scene_texts_lines.append(f"### Scene {si} — {heading}") else: scene_texts_lines.append(f"### Scene {si}") scene_texts_lines.append(text) scene_texts_lines.append("") scene_texts_block = "\n".join(scene_texts_lines).rstrip() or "(none)" # v1 template은 {scene_texts_block} placeholder가 없어도 .format()이 무시 — backward-compat. return template.format( visual_world_rules=visual_world_rules or "(none)", location_lines="\n".join(location_lines) or "(none)", scene_blocks="\n\n".join(scene_blocks) or "(none)", scene_texts_block=scene_texts_block, ) # ────────────────────────────────────────────── # semantic invariant validator (10 rules) # ────────────────────────────────────────────── # 한국어 허용 필드 (rationale, rationale_summary는 한국어 OK). # 그 외 ID/snake_case 필드는 ASCII만 허용. _ID_FIELDS_FLOOR_PLAN = ("id", "building_group", "primary_location_id") _ID_FIELDS_CHAIN_BG = ("id", "floor_plan_id", "location_id", "kind", "parent_id", "time") _ID_FIELDS_PREV_SHOT = ("location_id", "kind") def _check_no_korean(value: str, field_label: str) -> Optional[str]: """ID/snake_case 필드에 한글/한자/kana가 있으면 에러 메시지 반환.""" if value and _NON_ASCII_TEXT_RE.search(value): return f"{field_label}={value!r} contains non-ASCII text (snake_case ASCII required)" return None def validate_planner_output( plan: Dict[str, Any], all_location_ids: List[str], all_shot_ids: List[str], ) -> None: """planner LLM 출력에 대한 10개 semantic invariant 검증. schema가 못 잡는 의미적 규칙을 모두 검사하고, 위반 시 ValueError를 raise한다. rationale_summary와 per-item rationale은 한국어 허용; ID/snake_case 필드는 ASCII only. Invariants: 1. floor_plans[].location_ids ⊆ all_location_ids 2. chain_bg_groups[].shot_ids ⊆ all_shot_ids 3. floor_plan_order = floor_plans IDs (set equality) 4. chain_bg_order = chain_bg_groups IDs (set equality) 5. parent_id이 비지 않으면 chain_bg_order에서 자신보다 앞에 있어야 6. frequency rule: floor_plans[].shot_count ≥ 3 7. building_group이 같은 floor_plans는 floor_plan_order에서 인접 8. 한국어 0건 in 모든 ID/snake_case 필드 9. cycle 검출 — chain_bg parent chain에서 자기 자신이 ancestor가 되면 안 됨 10. episode-level: floor_plans=[]이면 chain_bg_groups=[]이어야 (역도 성립) """ location_set = set(all_location_ids) shot_set = set(all_shot_ids) floor_plans = plan.get("floor_plans") or [] floor_plan_order = list(plan.get("floor_plan_order") or []) chain_bg_groups = plan.get("chain_bg_groups") or [] chain_bg_order = list(plan.get("chain_bg_order") or []) prev_shot_only = plan.get("prev_shot_only") or [] # ── invariant 10 (early): episode-level — floor_plans=[] iff chain_bg_groups=[] ── # episode-level invariant은 다른 검사보다 먼저 — floor_plans/chain_bg_groups 한쪽만 # 비어있으면 downstream invariant들이 의미 없는 에러를 내기 전에 즉시 reject. if bool(floor_plans) != bool(chain_bg_groups): raise ValueError( f"invariant 10: floor_plans (count={len(floor_plans)}) and " f"chain_bg_groups (count={len(chain_bg_groups)}) must both be empty or both non-empty" ) # ── invariant 1: floor_plans[].location_ids ⊆ all_location_ids ── for fp in floor_plans: for loc in fp.get("location_ids") or []: if loc not in location_set: raise ValueError( f"invariant 1: floor_plan {fp.get('id')!r} location {loc!r} " f"not in all_location_ids" ) primary = fp.get("primary_location_id", "") if primary and primary not in location_set: raise ValueError( f"invariant 1: floor_plan {fp.get('id')!r} primary_location_id " f"{primary!r} not in all_location_ids" ) # primary는 location_ids에도 포함되어야 if primary and primary not in (fp.get("location_ids") or []): raise ValueError( f"invariant 1: floor_plan {fp.get('id')!r} primary_location_id " f"{primary!r} must be in its own location_ids" ) # ── invariant 2: chain_bg_groups[].shot_ids ⊆ all_shot_ids ── for cb in chain_bg_groups: for sid in cb.get("shot_ids") or []: if sid not in shot_set: raise ValueError( f"invariant 2: chain_bg {cb.get('id')!r} shot {sid!r} " f"not in all_shot_ids" ) cb_loc = cb.get("location_id", "") if cb_loc and cb_loc not in location_set: raise ValueError( f"invariant 2: chain_bg {cb.get('id')!r} location_id " f"{cb_loc!r} not in all_location_ids" ) # ── invariant 3: floor_plan_order = floor_plans IDs (set equality) ── fp_ids = [fp.get("id", "") for fp in floor_plans] if set(fp_ids) != set(floor_plan_order): raise ValueError( f"invariant 3: floor_plan_order {floor_plan_order} does not match " f"floor_plans IDs {fp_ids}" ) if len(floor_plan_order) != len(set(floor_plan_order)): raise ValueError( f"invariant 3: floor_plan_order has duplicates: {floor_plan_order}" ) # ── invariant 4: chain_bg_order = chain_bg_groups IDs (set equality) ── cb_ids = [cb.get("id", "") for cb in chain_bg_groups] if set(cb_ids) != set(chain_bg_order): raise ValueError( f"invariant 4: chain_bg_order {chain_bg_order} does not match " f"chain_bg_groups IDs {cb_ids}" ) if len(chain_bg_order) != len(set(chain_bg_order)): raise ValueError( f"invariant 4: chain_bg_order has duplicates: {chain_bg_order}" ) # ── invariant 5: parent_id이 비지 않으면 chain_bg_order에서 자신보다 앞에 있어야 ── cb_by_id: Dict[str, Dict[str, Any]] = {cb.get("id", ""): cb for cb in chain_bg_groups} seen_in_order: List[str] = [] for cid in chain_bg_order: cb = cb_by_id.get(cid) if cb is None: # invariant 4가 이미 잡았어야 한다 — 방어적 skip seen_in_order.append(cid) continue pid = (cb.get("parent_id") or "").strip() if pid and pid not in seen_in_order: raise ValueError( f"invariant 5: chain_bg {cid!r} comes before its parent {pid!r} " f"in chain_bg_order" ) seen_in_order.append(cid) # invariant 11: floor_plan_id가 floor_plans에 존재해야 함 fp_id = cb.get("floor_plan_id", "") if fp_id and fp_id not in fp_ids: raise ValueError( f"invariant 11: chain_bg {cid!r} floor_plan_id {fp_id!r} " f"not in floor_plans" ) # ── invariant 6: frequency rule: floor_plans[].shot_count ≥ 3 ── for fp in floor_plans: shot_count = fp.get("shot_count", 0) if shot_count < 3: raise ValueError( f"invariant 6: floor_plan {fp.get('id')!r} has shot_count=" f"{shot_count} (< 3) — {shot_count} shot location forbidden in floor_plans" ) # ── invariant 7: same building_group floor_plans는 floor_plan_order에서 인접 ── fp_meta = {fp.get("id", ""): fp for fp in floor_plans} seen_groups: Dict[str, int] = {} # building_group → 마지막 등장 index for i, fp_id in enumerate(floor_plan_order): fp = fp_meta.get(fp_id) if fp is None: continue bg = fp.get("building_group", "") if not bg: continue if bg in seen_groups: # 인접하지 않으면 위반 (현재 index와 마지막 index 차이 > 1) if i - seen_groups[bg] != 1: raise ValueError( f"invariant 7: building_group {bg!r} is non-adjacent in " f"floor_plan_order (gap between index {seen_groups[bg]} and {i})" ) seen_groups[bg] = i # ── invariant 8: 한국어 0건 in 모든 ID/snake_case 필드 ── for fp in floor_plans: for fname in _ID_FIELDS_FLOOR_PLAN: err = _check_no_korean(fp.get(fname, ""), f"floor_plan.{fname}") if err: raise ValueError(f"invariant 8: {err}") for loc in fp.get("location_ids") or []: err = _check_no_korean(loc, "floor_plan.location_ids[]") if err: raise ValueError(f"invariant 8: {err}") for cb in chain_bg_groups: for fname in _ID_FIELDS_CHAIN_BG: err = _check_no_korean(cb.get(fname, ""), f"chain_bg.{fname}") if err: raise ValueError(f"invariant 8: {err}") for sid in cb.get("shot_ids") or []: err = _check_no_korean(sid, "chain_bg.shot_ids[]") if err: raise ValueError(f"invariant 8: {err}") for ps in prev_shot_only: for fname in _ID_FIELDS_PREV_SHOT: err = _check_no_korean(ps.get(fname, ""), f"prev_shot_only.{fname}") if err: raise ValueError(f"invariant 8: {err}") # floor_plan_order / chain_bg_order ID 자체도 ASCII 필요 for fp_id in floor_plan_order: err = _check_no_korean(fp_id, "floor_plan_order[]") if err: raise ValueError(f"invariant 8: {err}") for cb_id in chain_bg_order: err = _check_no_korean(cb_id, "chain_bg_order[]") if err: raise ValueError(f"invariant 8: {err}") # ── invariant 9: cycle 검출 — chain_bg parent chain에서 자기 자신이 ancestor가 되면 안 됨 ── for cb in chain_bg_groups: cid = cb.get("id", "") visited: List[str] = [] cursor = cid while True: cur_node = cb_by_id.get(cursor) if cur_node is None: break pid = (cur_node.get("parent_id") or "").strip() if not pid: break if pid == cid: raise ValueError( f"invariant 9: chain_bg {cid!r} has cycle — " f"self appears as ancestor through {visited + [pid]}" ) if pid in visited: raise ValueError( f"invariant 9: chain_bg parent chain has a cycle starting from " f"{cid!r}: {visited + [pid]}" ) visited.append(pid) cursor = pid # safety bound (graph는 최대 노드 수만큼만 traverse) if len(visited) > len(chain_bg_groups) + 1: raise ValueError( f"invariant 9: chain_bg parent chain exceeds bound from {cid!r}" ) # NOTE: invariant 10은 함수 진입 직후 early-check로 처리됨. # ────────────────────────────────────────────── # runtime enum injection # ────────────────────────────────────────────── def _inject_runtime_enums( schema: Dict[str, Any], location_short_ids: List[str], shot_ids: List[str], ) -> Dict[str, Any]: """schema의 location/shot 관련 string field에 enum 제약을 주입. 원본 schema는 mutate하지 않는다 (deepcopy). 주입 대상: - floor_plans[].location_ids[] → enum=location_short_ids - floor_plans[].primary_location_id → enum=location_short_ids - chain_bg_groups[].location_id → enum=location_short_ids - chain_bg_groups[].shot_ids[] → enum=shot_ids - prev_shot_only[].location_id → enum=location_short_ids """ new_schema = copy.deepcopy(schema) props = new_schema.get("properties") or {} # floor_plans[].location_ids[] / primary_location_id fp = props.get("floor_plans") or {} fp_items = (fp.get("items") or {}).get("properties") or {} if location_short_ids: if "location_ids" in fp_items: inner = fp_items["location_ids"] inner["items"] = {"type": "string", "enum": list(location_short_ids)} if "primary_location_id" in fp_items: fp_items["primary_location_id"] = { **fp_items["primary_location_id"], "enum": list(location_short_ids), } # chain_bg_groups[].location_id / shot_ids[] cb = props.get("chain_bg_groups") or {} cb_items = (cb.get("items") or {}).get("properties") or {} if location_short_ids and "location_id" in cb_items: cb_items["location_id"] = { **cb_items["location_id"], "enum": list(location_short_ids), } if shot_ids and "shot_ids" in cb_items: cb_items["shot_ids"]["items"] = { "type": "string", "enum": list(shot_ids), } # prev_shot_only[].location_id ps = props.get("prev_shot_only") or {} ps_items = (ps.get("items") or {}).get("properties") or {} if location_short_ids and "location_id" in ps_items: ps_items["location_id"] = { **ps_items["location_id"], "enum": list(location_short_ids), } return new_schema # ────────────────────────────────────────────── # run_background_planner 진입점 (3회 retry) # ────────────────────────────────────────────── _DEFAULT_MAX_RETRIES = 3 _DEFAULT_BACKOFF_BASE_SEC = 2 # attempt별 sleep = base * (attempt + 1) def run_background_planner( *, project_config: Optional[Dict[str, Any]] = None, user_prompt: str, location_short_ids: List[str], shot_ids: List[str], opik_metadata: Optional[Dict[str, Any]] = None, call_structured_fn: Callable[..., Dict[str, Any]], max_retries: int = _DEFAULT_MAX_RETRIES, backoff_base_sec: float = _DEFAULT_BACKOFF_BASE_SEC, sleep_fn: Callable[[float], None] = time.sleep, ) -> Dict[str, Any]: """planner LLM을 호출해 background plan을 생성하고 invariant까지 검증. 실패 시 backoff로 retry. ``max_retries``회까지 실패하면 PlannerError raise. Args: project_config: project_llm_config (provider/model 결정) user_prompt: build_planner_user_prompt 결과 location_short_ids: schema enum + invariant 검증용 location ids shot_ids: schema enum + invariant 검증용 shot ids opik_metadata: Opik trace metadata (optional) call_structured_fn: LLM 호출 함수 (보통 llm_client.call_structured) max_retries: 총 시도 횟수 (default 3) backoff_base_sec: 시도 간 대기 시간 base (default 2초) sleep_fn: 테스트 주입용 sleep 함수 Returns: validate_planner_output을 통과한 plan dict. Raises: PlannerError: max_retries 모두 실패. """ system = load_prompt(_MODULE, "system") schema = load_schema(_MODULE, "schema") schema = _inject_runtime_enums(schema, location_short_ids, shot_ids) last_err: Optional[Exception] = None for attempt in range(max_retries): try: result = call_structured_fn( step="background_planner", system_prompt=system, user_prompt=user_prompt, response_schema=schema, project_config=project_config, schema_name="background_planner", opik_metadata=opik_metadata, ) validate_planner_output(result, location_short_ids, shot_ids) logger.info( "background_planner: ok on attempt %d/%d (floor_plans=%d, chain_bg=%d, prev_shot=%d)", attempt + 1, max_retries, len(result.get("floor_plans") or []), len(result.get("chain_bg_groups") or []), len(result.get("prev_shot_only") or []), ) return result except (ValueError, RuntimeError, TimeoutError, ConnectionError) as exc: # ValueError = invariant 위반(retry로 LLM 재요청), 나머지 = 일시적 LLM/네트워크 오류. # TypeError/AttributeError/KeyError 같은 프로그래밍 버그는 fail-fast. last_err = exc logger.warning( "background_planner: attempt %d/%d failed: %s", attempt + 1, max_retries, exc, ) if attempt + 1 < max_retries: sleep_fn(backoff_base_sec * (attempt + 1)) raise PlannerError( f"background_planner failed after {max_retries} retries: {last_err}" )