"""background_master_plan — Phase 7 Step 2. 그룹당 LLM 호출 1회로 floor_plans + backgrounds + gen_order + applies_to_shots 결정. """ from __future__ import annotations import logging import re import time from collections import defaultdict from typing import Any, Callable, Dict, List, Optional from app.modules.prompt_loader import load_prompt, load_schema logger = logging.getLogger(__name__) _NON_ASCII_TEXT_RE = re.compile( r"[ㄱ-ㆎ가-힣" r"一-鿿㐀-䶿豈-﫿" r"぀-ヿ]" ) _SAFE_ID_RE = re.compile(r"^[a-z0-9][a-z0-9_]*$") _MODULE = "background_master_plan" SURFACE_ROLE_ENUM = { "interior_room", "exterior_plate", "transition_zone", "site_surface", } class MasterPlanError(Exception): """retry 한도까지 실패.""" def build_master_plan_user_prompt( group: Dict[str, Any], scenes: List[Dict[str, Any]], visual_world_rules: str, location_profiles: Dict[str, Dict[str, Any]], ) -> str: """그룹의 master plan LLM에 보낼 user prompt — scene text 무절단 inject. .replace() 사용 (.format() brace 충돌 회피, T4와 동일 패턴). location_profiles: loc_id → entity_canon.metadata_json.location.space_profile. members block 에 space_profile marker (kind + allowed_space_keys) 를 주입해 LLM 이 single_space location 을 sub-room 으로 분할하지 않도록 한다 (D6 SOT 존중). """ template = load_prompt(_MODULE, "user_template") members_lines: List[str] = [] for m in group.get("members", []) or []: loc_id = m.get("loc_id", "") profile = (location_profiles or {}).get(loc_id) or {} kind = profile.get("kind", "unknown") allowed = profile.get("allowed_space_keys") or [] allowed_str = ", ".join(allowed) if allowed else "main" members_lines.append( f"- {loc_id}: {m.get('label', '')} " f"({'indoor' if m.get('is_indoor') else 'outdoor'}, {m.get('shot_count', 0)} shot) " f"[space_profile: {kind} — allowed_space_keys: {allowed_str}]" ) members_block = "\n".join(members_lines) or "(none)" scene_lines: List[str] = [] shot_lines: List[str] = [] for sc in scenes: si = sc.get("scene_index") if si is None: continue heading = (sc.get("heading") or "").strip() text = (sc.get("text") or "").strip() scene_lines.append(f"### Scene {si} — {heading}" if heading else f"### Scene {si}") scene_lines.append(text) scene_lines.append("") for sh in sc.get("shots", []) or []: shi = sh.get("shot_index", 0) shot_lines.append( f"- S{si}_Shot{shi} (loc={sh.get('location_id', '')}): {sh.get('description', '')}" ) scene_block = "\n".join(scene_lines).rstrip() or "(none)" shots_block = "\n".join(shot_lines) or "(none)" return ( template .replace("{group_id}", group.get("group_id", "")) .replace("{members_block}", members_block) .replace("{visual_world_rules}", visual_world_rules or "(none)") .replace("{scene_segments_block}", scene_block) .replace("{shots_block}", shots_block) ) def validate_master_plan_raw_intent( plan: Dict[str, Any], expected_group_id: str, group_loc_ids: set, group_shot_ids: set, location_profiles: Dict[str, Dict[str, Any]], ) -> None: """D6 T-pre-2 (B6) + T4-fix3 — LLM raw intent 단계 검증 (bg_id 부재). LLM 이 새 D6 schema 로 출력한 raw intent 를 검증. raw intent fields: loc_id / space_key_hint / time_phase / state_class / applies_to_shots / sub_location_label / state_label_raw / surface_role / depends_on_fp. floor_plans[]: fp_id / loc_id / space_key_hint / sub_location / scope / depends_on_fp. invariants (raw intent 한정): A. group_id matches + ASCII B. fp_id ASCII snake_case C. background 의 raw intent 필수 field 존재 D. state_class ∈ STATE_CLASS_ENUM E. interior_room depends_on_fp 는 1+ 자체 plan fp_id 참조. exterior/transition/site plate 는 fp-less raw intent 허용. E2. (T4-fix3) background.depends_on_fp[*] 가 가리키는 fp 의 declared (loc_id, normalized_space_key) 가 background 의 (loc_id, normalized_space_key) 와 일치. 불일치 시 FpLinkMismatchError raise (LLM retry path). F. background.loc_id ⊆ group_loc_ids F2. (T4-fix3) floor_plans[].loc_id ⊆ group_loc_ids G. applies_to_shots ⊆ group_shot_ids H. T4-fix I2: backgrounds[].bg_id 필드 존재 시 명시 reject — schema (additionalProperties: false) 가 1차 막지만 strict-off provider 안전망. LLM 이 bg_id 만들면 코드 부여 단계 (assign_bg_ids) 와 충돌. bg_id 정합성 / gen_order topological 등 코드-부여 후 ID-기반 invariant 는 `validate_master_plan_assigned` 가 별도 검사. location_profiles: loc_id → entity_canon.metadata_json.location.space_profile. normalize_space_key 의 SOT. floor_plans[].loc_id 가 이 dict 에 없으면 SemanticKeyError raise (entity_extractor 누락). """ from app.core.bg_state_vocab import LLM_INTENT_ID_RE, validate_state_class, StateClassError from app.core.bg_catalog import FpLinkMismatchError, SemanticKeyError, normalize_space_key if plan.get("group_id") != expected_group_id: raise ValueError( f"group_id mismatch: got {plan.get('group_id')!r}, expected {expected_group_id!r}" ) if _NON_ASCII_TEXT_RE.search(plan.get("group_id", "")): raise ValueError(f"group_id {plan.get('group_id')!r} contains non-ASCII") floor_plans = plan.get("floor_plans") or [] backgrounds = plan.get("backgrounds") or [] fp_ids = [fp.get("fp_id", "") for fp in floor_plans] fp_id_set = set(fp_ids) # invariant B: fp_id ASCII snake_case (LLM_INTENT_ID_RE) for fp in floor_plans: for field in ("fp_id",): v = fp.get(field, "") if _NON_ASCII_TEXT_RE.search(v): raise ValueError(f"floor_plan.{field}={v!r} contains non-ASCII") if not LLM_INTENT_ID_RE.match(v): raise ValueError(f"floor_plan.{field}={v!r} not ASCII snake_case") # fp_ids unique if len(fp_ids) != len(fp_id_set): raise ValueError(f"floor_plans.fp_id has duplicates: {fp_ids}") # T4-fix3: invariant F2 — floor_plans[].loc_id ⊆ group_loc_ids + fp_id → # (loc_id, normalized_space_key) 맵 build (E2 cross-check 의 input). fp_loc_space: Dict[str, tuple] = {} for fp in floor_plans: fp_id = fp.get("fp_id", "") fp_loc = fp.get("loc_id", "") fp_hint = fp.get("space_key_hint", "") if not fp_loc: raise ValueError( f"floor_plan.fp_id={fp_id!r} missing loc_id (T4-fix3 schema)" ) if fp_loc not in group_loc_ids: raise ValueError( f"floor_plan.fp_id={fp_id!r} loc_id={fp_loc!r} not in group " f"{sorted(group_loc_ids)}" ) if not fp_hint: raise ValueError( f"floor_plan.fp_id={fp_id!r} missing space_key_hint (T4-fix3 schema)" ) profile = location_profiles.get(fp_loc) if profile is None: raise SemanticKeyError( f"floor_plan.fp_id={fp_id!r} loc_id={fp_loc!r} has no location_profile " f"(available: {sorted(location_profiles.keys())})" ) fp_loc_space[fp_id] = (fp_loc, normalize_space_key(fp_loc, fp_hint, profile)) # invariant F3 (E2E v1): single_space location 은 floor_plan 정확히 1개. # normalize_space_key 가 single_space 의 모든 hint 를 main 으로 collapse → 2+ fp 면 # downstream assign_bg_ids 에서 같은 sem_key 의 다른 depends_on_fp 충돌 # (SemanticKeyError). raw 단계에서 retryable ValueError 로 차단 — silent crash 방지. fp_count_per_loc: Dict[str, int] = {} for fp in floor_plans: fp_loc = fp.get("loc_id", "") fp_count_per_loc[fp_loc] = fp_count_per_loc.get(fp_loc, 0) + 1 for fp_loc, count in fp_count_per_loc.items(): profile = location_profiles.get(fp_loc) or {} if profile.get("kind") == "single_space" and count > 1: raise ValueError( f"location {fp_loc!r} is single_space (space_profile SOT) but plan has " f"{count} floor_plans for it — single_space location 은 floor_plan 1개만 " f"허용. sub-room 이 보여도 단일 floor_plan 으로 통합할 것." ) # invariant C/D/E/F/G: background raw intent for bg in backgrounds: # T4-fix I2: bg_id field 존재 시 명시 reject — schema 가 막지만 (additionalProperties: # false), validator 도 명시 reject 하여 schema disabled / strict-off provider # 안전망. LLM 이 bg_id 만들면 코드 부여 단계 (assign_bg_ids) 와 충돌. if "bg_id" in bg: raise ValueError( f"background contains bg_id {bg.get('bg_id')!r} — D6 contract 위반: " f"raw intent 단계에서 LLM 이 bg_id 출력 금지 (코드가 부여)." ) loc_id = bg.get("loc_id", "") # invariant F: loc_id ⊆ group_loc_ids if loc_id not in group_loc_ids: raise ValueError( f"background loc_id {loc_id!r} not in group {sorted(group_loc_ids)}" ) # invariant C: required raw intent fields for field in ("space_key_hint", "time_phase", "state_class", "surface_role"): if not bg.get(field): raise ValueError( f"background loc_id={loc_id!r} missing required raw intent " f"field {field!r}" ) surface_role = bg.get("surface_role", "") if surface_role not in SURFACE_ROLE_ENUM: raise ValueError( f"background loc_id={loc_id!r} surface_role={surface_role!r} " f"not in {sorted(SURFACE_ROLE_ENUM)}" ) # invariant D: state_class enum try: validate_state_class(bg["state_class"]) except StateClassError as exc: raise ValueError( f"background loc_id={loc_id!r}: {exc}" ) from exc # invariant E: interior_room 은 floor_plan 의존 필수. exterior/transition/site # plate 는 fp-less raw intent 허용 (W21B-wave-2). deps = bg.get("depends_on_fp") or [] if surface_role == "interior_room" and not deps: raise ValueError( f"background loc_id={loc_id!r} surface_role='interior_room' " "requires non-empty depends_on_fp" ) for fp_ref in deps: if fp_ref not in fp_id_set: raise ValueError( f"background loc_id={loc_id!r} depends_on_fp {fp_ref!r} not in plan " f"(available: {sorted(fp_id_set)})" ) # invariant E2 (T4-fix3): fp link cross-check. # bg 의 (loc_id, normalized_space_key) 가 fp 가 declare 한 것과 일치 의무. # 다른 sub-room / 다른 location 의 fp 참조는 silent first-wins 차단. bg_profile = location_profiles.get(loc_id) if bg_profile is None: raise SemanticKeyError( f"background loc_id={loc_id!r} has no location_profile " f"(available: {sorted(location_profiles.keys())})" ) bg_space = normalize_space_key(loc_id, bg["space_key_hint"], bg_profile) for fp_ref in deps: fp_loc, fp_space = fp_loc_space[fp_ref] if (loc_id, bg_space) != (fp_loc, fp_space): raise FpLinkMismatchError( f"background loc_id={loc_id!r} space={bg_space!r} depends_on_fp " f"{fp_ref!r} which covers (loc_id={fp_loc!r}, space={fp_space!r}) — " "fp link mismatch (T4-fix3). Split fp by (loc_id, space_key_hint) " "or fix background.depends_on_fp." ) # invariant G: applies_to_shots ⊆ group_shot_ids for sid in bg.get("applies_to_shots") or []: if sid not in group_shot_ids: raise ValueError( f"background loc_id={loc_id!r} applies_to_shots {sid!r} " f"not in group" ) def validate_master_plan_assigned( catalog: Dict[str, Dict[str, Any]], ) -> None: """D6 T-pre-2 (B6) — code-assigned bg_id catalog 검증. `assign_bg_ids` 후처리 결과의 bg_id 가 D6 BG_ID_RE (`^L\\d{2,3}B\\d{2,3}$`) 패턴 따르고 unique 인지 확인. """ from app.core.bg_state_vocab import BG_ID_RE seen = set() for bg_id, entry in catalog.items(): if bg_id in seen: raise ValueError(f"duplicate bg_id {bg_id!r} in catalog") if not BG_ID_RE.match(bg_id): raise ValueError( f"assigned bg_id {bg_id!r} fails BG_ID_RE pattern " f"^L\\d{{2,3}}B\\d{{2,3}}$" ) # entry.bg_id 필드도 동일 (key 와 entry 내부 일치) if entry.get("bg_id") != bg_id: raise ValueError( f"catalog key {bg_id!r} mismatches entry.bg_id {entry.get('bg_id')!r}" ) seen.add(bg_id) def validate_master_plan_output( plan: Dict[str, Any], expected_group_id: str, group_loc_ids: set, group_shot_ids: set, ) -> None: """[Legacy pre-D6] 8 hard invariants: 1. group_id matches + ASCII; all fp_id/bg_id/sub_location/state_label ASCII 2. 모든 background.depends_on_fp는 비어있지 않고 자체 plan의 fp_id 참조 3. applies_to_shots ⊆ group_shot_ids 4. 같은 sub_location의 모든 background는 같은 fp_id를 ref (depends_on_fp 첫 entry 비교) 5. 같은 sub_location의 background N>1개일 때 후속 background는 depends_on_bg에 이전 같은 sub_location bg를 reference 6. gen_order는 위상정렬: 모든 dependency가 앞에 등장 7. fp_id/bg_id/gen_order 모두 unique + set 일치 8. background.loc_id ⊆ group_loc_ids D6 T-pre-2 (B6): 본 함수는 pre-D6 monolithic validator. D6 path 는 `validate_master_plan_raw_intent` (LLM 출력) + `validate_master_plan_assigned` (코드 부여 후) 로 split. 본 함수는 legacy compat / 옛 schema migration 용도로 남겨둠 — production runner 는 D6 path 만 호출. """ if plan.get("group_id") != expected_group_id: raise ValueError( f"group_id mismatch: got {plan.get('group_id')!r}, expected {expected_group_id!r}" ) if _NON_ASCII_TEXT_RE.search(plan.get("group_id", "")): raise ValueError(f"group_id {plan.get('group_id')!r} contains non-ASCII") floor_plans = plan.get("floor_plans") or [] backgrounds = plan.get("backgrounds") or [] gen_order = plan.get("gen_order") or [] fp_ids = [fp.get("fp_id", "") for fp in floor_plans] bg_ids = [bg.get("bg_id", "") for bg in backgrounds] fp_id_set = set(fp_ids) bg_id_set = set(bg_ids) # invariant 1: ASCII (check non-ASCII first for more specific error) for fp in floor_plans: for field in ("fp_id", "sub_location"): v = fp.get(field, "") if _NON_ASCII_TEXT_RE.search(v): raise ValueError(f"floor_plan.{field}={v!r} contains non-ASCII") if not _SAFE_ID_RE.match(v): raise ValueError(f"floor_plan.{field}={v!r} not ASCII snake_case") for bg in backgrounds: for field in ("bg_id", "sub_location", "state_label"): v = bg.get(field, "") if _NON_ASCII_TEXT_RE.search(v): raise ValueError(f"background.{field}={v!r} contains non-ASCII") if not _SAFE_ID_RE.match(v): raise ValueError(f"background.{field}={v!r} not ASCII snake_case") # invariant 7: unique + set 일치 if len(fp_ids) != len(fp_id_set): raise ValueError(f"floor_plans.fp_id has duplicates: {fp_ids}") if len(bg_ids) != len(bg_id_set): raise ValueError(f"backgrounds.bg_id has duplicates: {bg_ids}") expected_order_set = fp_id_set | bg_id_set if set(gen_order) != expected_order_set: raise ValueError( f"gen_order set mismatch: {set(gen_order)} vs {expected_order_set}" ) if len(gen_order) != len(set(gen_order)): raise ValueError(f"gen_order has duplicates: {gen_order}") # invariant 2 & 8: bg.depends_on_fp 비어있지 않고 fp_id 참조 + bg.loc_id 그룹 멤버 for bg in backgrounds: deps = bg.get("depends_on_fp") or [] if not deps: raise ValueError(f"background {bg.get('bg_id')!r} depends_on_fp is empty") for fp_ref in deps: if fp_ref not in fp_id_set: raise ValueError( f"background {bg.get('bg_id')!r} depends_on_fp {fp_ref!r} not in plan " f"(available: {sorted(fp_id_set)})" ) if bg.get("loc_id", "") not in group_loc_ids: raise ValueError( f"background {bg.get('bg_id')!r} loc_id {bg.get('loc_id')!r} not in group" ) # invariant 3: applies_to_shots ⊆ group_shot_ids for sid in bg.get("applies_to_shots") or []: if sid not in group_shot_ids: raise ValueError( f"background {bg.get('bg_id')!r} applies_to_shots {sid!r} not in group" ) # depends_on_bg는 자체 bg_id 참조여야 for bg_ref in bg.get("depends_on_bg") or []: if bg_ref not in bg_id_set: raise ValueError( f"background {bg.get('bg_id')!r} depends_on_bg {bg_ref!r} not in plan" ) # invariant 4: 같은 sub_location → 같은 fp_id (첫 entry 비교) + # 참조하는 floor_plan의 sub_location도 일치해야 함 (cross-validation). # 후자가 빠지면 LLM이 fp_living(sub=living_room)과 bg_x(sub=bedroom, # depends_on_fp=[fp_living])를 동시에 만들어 sub-room mismatch 버그 재현. fp_sub_map: Dict[str, str] = {fp.get("fp_id", ""): fp.get("sub_location", "") for fp in floor_plans} sub_to_fp: Dict[str, str] = {} for bg in backgrounds: sub = bg.get("sub_location", "") first_fp = (bg.get("depends_on_fp") or [""])[0] if sub in sub_to_fp: if sub_to_fp[sub] != first_fp: raise ValueError( f"sub_location {sub!r}: backgrounds reference different floor plans " f"({sub_to_fp[sub]} vs {first_fp})" ) else: sub_to_fp[sub] = first_fp # cross-check: bg.sub_location must equal fp.sub_location fp_sub = fp_sub_map.get(first_fp, "") if fp_sub != sub: raise ValueError( f"background {bg.get('bg_id')!r} sub_location {sub!r} mismatches " f"floor_plan {first_fp!r} sub_location {fp_sub!r}" ) # invariant 5: 같은 sub_location N>1개 → 후속 bg가 depends_on_bg에 이전 같은 sub_location bg 포함 sub_to_bgs: Dict[str, List[str]] = defaultdict(list) for bg in backgrounds: sub_to_bgs[bg.get("sub_location", "")].append(bg.get("bg_id", "")) for sub, lst in sub_to_bgs.items(): if len(lst) <= 1: continue # 첫 bg는 chain root, 나머지는 같은 sub의 이전 bg 중 1+개를 ref해야 bg_meta = {bg.get("bg_id", ""): bg for bg in backgrounds} for i, bid in enumerate(lst[1:], start=1): deps = set(bg_meta[bid].get("depends_on_bg") or []) prior_in_sub = set(lst[:i]) if not (deps & prior_in_sub): raise ValueError( f"background {bid!r} in sub_location {sub!r} missing chain " f"(should depends_on_bg one of {prior_in_sub})" ) # invariant 6: gen_order는 topological — 각 entry의 deps가 앞에 있어야 seen: set = set() fp_meta = {fp.get("fp_id", ""): fp for fp in floor_plans} bg_meta = {bg.get("bg_id", ""): bg for bg in backgrounds} for ent in gen_order: if ent in fp_meta: for dep in fp_meta[ent].get("depends_on_fp") or []: if dep not in seen: raise ValueError( f"topological: floor_plan {ent!r} dep {dep!r} appears later" ) elif ent in bg_meta: bg = bg_meta[ent] for dep in (bg.get("depends_on_fp") or []) + (bg.get("depends_on_bg") or []): if dep not in seen: raise ValueError( f"topological: background {ent!r} dep {dep!r} appears later" ) seen.add(ent) def run_background_master_plan( *, user_prompt: str, expected_group_id: str, group_loc_ids: List[str], group_shot_ids: List[str], location_profiles: Dict[str, Dict[str, Any]], call_structured_fn: Callable[..., Dict[str, Any]], project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, max_retries: int = 3, backoff_base_sec: float = 2.0, sleep_fn: Callable[[float], None] = time.sleep, ) -> Dict[str, Any]: system = load_prompt(_MODULE, "system") schema = load_schema(_MODULE, "schema") loc_set = set(group_loc_ids) shot_set = set(group_shot_ids) last_err: Optional[Exception] = None for attempt in range(max_retries): try: result = call_structured_fn( step="background_master_plan", system_prompt=system, user_prompt=user_prompt, response_schema=schema, project_config=project_config, schema_name="background_master_plan", opik_metadata=opik_metadata, ) # D6 T-pre-2 (B6) + T4-fix3: raw intent validator — bg_id 부재 + fp link # cross-check (location_profiles SOT). 코드 부여 후 validate_master_plan_assigned # 로 별도 검증 (BackgroundMasterPlanStep 의 post-processing 에서). validate_master_plan_raw_intent( result, expected_group_id, loc_set, shot_set, location_profiles, ) return result except Exception as exc: # T20 I3: LLM client errors (litellm/openai/httpx) 포함 last_err = exc logger.warning( "background_master_plan attempt %d/%d for %s failed: %s", attempt + 1, max_retries, expected_group_id, exc, ) if attempt + 1 < max_retries: sleep_fn(backoff_base_sec * (attempt + 1)) raise MasterPlanError( f"background_master_plan {expected_group_id!r} exhausted {max_retries} retries: {last_err}" )