"""W21B-wave-4 v8 Phase A1: floor_plan_layout_plan LLM provider. ★DEAD (2026-08-15 확인): 호출자 없음, 축 D 정리 대상. ``litellm_layout_plan_provider`` 를 부르는 코드가 저장소 전체(app·tests· scripts·tools)에 하나도 없고, 이 모듈을 import 하는 곳도 없다. 따라서 ``_SYSTEM_PROMPT`` 는 어떤 실행 경로에서도 모델에 나가지 않는 죽은 무게다. 프로젝트 규칙(모듈 삭제 금지)에 따라 표식만 남기고 제거 여부는 사람 결정으로 넘긴다 — 다시 배선하려면 이 표식부터 지울 것. Real (text) LLM provider that asks a layout planner to emit the Path-S coordinate scaffold (``element_layout[]`` rects on a normalized grid) from the FP numbered_elements inventory + the marker numbers selected camera views use. Output is validated through the pure core ``validate_layout``; any non-conforming output raises ``LayoutPlanError``. Mirrors ``floor_plan_semantic_vlm_provider`` discipline: lazy ``import litellm``, env/capability preflight, strict json_schema response_format, fail-closed at every joint. NO image/VLM call (text-only). This wave wires the body but the step resolves to it only behind an opt-in selector; the default keeps it out of every code path. Scenario-leakage guard: the system prompt carries no scenario nouns; all dwelling facts arrive as a structured JSON inventory in the user message. """ from __future__ import annotations import json import logging import os from typing import Any, Dict, List, Optional from app.modules.pipeline.floor_plan_layout_plan import ( DEFAULT_GRID, DEFAULT_RENDER_CAP, LAYOUT_SCHEMA, LayoutPlanError, validate_layout, ) logger = logging.getLogger(__name__) # NOTE: 'gpt55' 는 역사적 안정 식별자(provenance 매칭/lineage 호환) — 모델 교체와 # 무관하게 유지. 실제 물리 모델은 별도 model 인자/PROVIDER_MODEL_DEFAULT 에 기록되며 # 이 식별자에서 추론하지 않는다. WIRED_PROVIDER_NAME = "litellm_gpt55_layout_plan_v1" PROVIDER_MODEL_DEFAULT = "openai/gpt-6-astra" MAX_COMPLETION_TOKENS_DEFAULT = 16000 # DEAD (2026-08-15 확인): 호출자 없음, 축 D 정리 대상 — 모듈 docstring 참조. _SYSTEM_PROMPT = ( "You are a spatial layout planner for a top-down floor plan. You do NOT " "draw art; you emit structured coordinates that code will render as a " "simple box diagram. Given a numbered_elements inventory (number, label, " "category, position_hint, base_layer_decision) and the marker numbers " "that selected camera views actually use, output a layout JSON. Place " f"every RENDERED element as an axis-aligned rect on a {DEFAULT_GRID}x" f"{DEFAULT_GRID} grid (origin top-left, x right, y down, 0..{DEFAULT_GRID - 1}). " "Rules: structural-unit areas may contain/touch other elements, but two " "non-area objects must NOT fully overlap; keep stairs/circulation as a " "simple line or narrow rect, never a maze; preserve the dwelling scale " "implied by the inventory (do NOT invent extra rooms; a compact dwelling " "stays one compact plan). Set render_on_plan=true ONLY for the most " f"important structural units, openings, fixed fixtures, and anchor " f"furniture (cap about {DEFAULT_RENDER_CAP}). Low-salience or decorative " "items go to metadata_only_elements (NOT drawn). Set importance 0..1; " "mark items used by selected camera views as high importance. Carry place " "richness in place_semantic_tags / expected_visual_density / " "key_fixture_groups, NOT by over-detailing the plan. NEVER place a " "transient / state_overlay marker in element_layout. Use ONLY marker " "numbers present in the inventory, and carry each marker's " "base_layer_decision verbatim (do not re-classify). No prose outside JSON." ) def _build_messages(*, fp_id: str, numbered_elements: List[Dict[str, Any]], used_numbers: List[int]) -> List[Dict[str, Any]]: user = ( "numbered_elements (JSON):\n" + json.dumps(numbered_elements, ensure_ascii=False, sort_keys=True) + "\n\nmarker numbers used by selected camera views: " + json.dumps(sorted(used_numbers)) + f"\n\nEmit the layout JSON for fp_id={fp_id}. render_on_plan cap " f"about {DEFAULT_RENDER_CAP}; low-salience items to " "metadata_only_elements." ) return [{"role": "system", "content": _SYSTEM_PROMPT}, {"role": "user", "content": user}] def _response_format() -> Dict[str, Any]: return {"type": "json_schema", "json_schema": { "name": "floor_plan_layout_plan", "strict": True, "schema": LAYOUT_SCHEMA}} def litellm_layout_plan_provider( *, fp_id: str, numbered_elements: List[Dict[str, Any]], used_numbers: List[int], dossier_inventory: List[Dict[str, Any]], model: str = PROVIDER_MODEL_DEFAULT, max_completion_tokens: int = MAX_COMPLETION_TOKENS_DEFAULT, timeout_seconds: int = 180, ) -> Dict[str, Any]: """Single text-LLM completion → validated layout. Fail-closed. ``dossier_inventory`` is the exact-ID authority for validation (number → base_layer_decision). Returns the validated layout dict or raises ``LayoutPlanError``. """ if not isinstance(fp_id, str) or not fp_id: raise LayoutPlanError("fp_id missing or non-string") if not numbered_elements: raise LayoutPlanError(f"fp_id={fp_id!r}: empty numbered_elements") from app.core.openai_keys import has_openai_key if not has_openai_key(): raise LayoutPlanError("OPENAI_API_KEY missing; refusing to call litellm") try: import litellm # type: ignore from app.core.openai_keys import ( # type: ignore llm_completion as _llm_completion, ) except Exception as exc: # pragma: no cover raise LayoutPlanError(f"litellm import failed: {exc}") from exc try: if not bool(litellm.supports_response_schema(model=model)): raise LayoutPlanError( f"model {model!r} lacks response_schema support") params = litellm.get_supported_openai_params( model=model, custom_llm_provider="openai") or [] except LayoutPlanError: raise except Exception as exc: raise LayoutPlanError(f"litellm capability preflight raised: {exc}") from exc if "response_format" not in params: raise LayoutPlanError( f"model {model!r} lacks response_format support") try: resp = _llm_completion( model=model, messages=_build_messages(fp_id=fp_id, numbered_elements=numbered_elements, used_numbers=used_numbers), response_format=_response_format(), timeout=timeout_seconds, max_completion_tokens=max_completion_tokens, num_retries=0) except Exception as exc: raise LayoutPlanError(f"litellm.completion raised: {exc}") from exc choices = getattr(resp, "choices", None) or [] if not choices: raise LayoutPlanError("response.choices empty") ch = choices[0] msg = getattr(ch, "message", None) if msg is None: raise LayoutPlanError("response message missing") if getattr(msg, "refusal", None): raise LayoutPlanError("LLM refusal") content = getattr(msg, "content", None) if not content: raise LayoutPlanError("response content empty") if getattr(ch, "finish_reason", None) != "stop": raise LayoutPlanError( f"finish_reason must be 'stop' (got {getattr(ch, 'finish_reason', None)!r})") try: parsed = json.loads(content) except (TypeError, ValueError) as exc: raise LayoutPlanError(f"content not valid JSON: {exc}") from exc res = validate_layout(output=parsed, dossier_inventory=dossier_inventory, fp_id=fp_id) if not res["ok"]: raise LayoutPlanError("validator_failed: " + "; ".join(res["blockers"])[:400]) return res["layout"]