"""W21B-wave-4: real-VLM provider for floor-plan marker SEMANTIC readback. Production-adjacent helper that, when explicitly invoked behind an opt-in selector, performs a vision-LLM inspection of a base floor-plan PNG and returns a validated semantic readback (status='ok') describing, per numbered base marker, whether the object actually drawn at that marker matches the expected object class implied by the dossier inventory's label/layer. This module mirrors ``floor_plan_vlm_provider`` (the W20A2 geometry provider) in structure and fail-closed discipline, but asks a different question: geometry reads marker *position* (cell/kind); this reads marker *content fidelity*. It is deliberately separate so the two gates carry independent cost / failure / acceptance policies. **This wave wires the function body but invokes no real API by default.** ``litellm`` is imported lazily inside the function, so importing this module touches no external SDK. The step wrapper resolves to this helper only when its opt-in selector is True; the default keeps it out of every code path. Output is validated by the core ``validate_semantic_output`` — the same exact-ID / shape / enum contract the dry-run tests pin. Any non-conforming model output raises ``SemanticReadbackError`` (the core fail-closed signal), so a direct caller and the ``compute_semantic_readback`` dispatcher both reject bad output. """ from __future__ import annotations import base64 import json import logging import os from pathlib import Path from typing import Any, Dict, List, Optional from app.modules.pipeline.floor_plan_semantic_readback import ( SEMANTIC_READBACK_SCHEMA, SemanticReadbackError, validate_semantic_output, ) logger = logging.getLogger(__name__) # Telemetry label distinguishing real-provider readbacks from synthetic # fixtures. Not required in the readback shape (the step counts real # calls via its own counter), kept for labels/audit. # NOTE: 'gpt55' 는 역사적 안정 식별자(provenance 매칭/lineage 호환) — 모델 교체와 # 무관하게 유지. 실제 물리 모델은 별도 model 인자/PROVIDER_MODEL_DEFAULT 에 기록되며 # 이 식별자에서 추론하지 않는다. WIRED_PROVIDER_NAME: str = "litellm_gpt55_semantic_vision_v1" # ★#92 (2026-08-27): VLM 은 **gemini 3.1 pro + grok 최신 둘만**. # 사용자 지시를 네 번째로 받았다 — 「무조건 gemini 3.1 pro 와 grok # 최신 모델 둘을 사용해야해 / 아홉 전부 바꿔」. # # ★**기본 OFF 라고 안 바꾸면 안 된다.** 앞 판에서 「지금 안 도니까 # 놔둔다」고 내가 판단했는데, 그건 내 판단으로 사용자 명시 지시를 # 뒤집은 것이다. 켜는 순간 지시를 어긴 상태로 돌아간다. # # ★물리 이름을 직접 쓴다 — 이 자리는 Router 를 안 거치고 litellm 을 # 직접 친다. 접두 `gemini/` 는 Router 등록부와 같다 # (`llm_client.py:538`) — 지어낸 것이 아니다. PROVIDER_MODEL_DEFAULT: str = "gemini/gemini-3.1-pro-preview" MAX_COMPLETION_TOKENS_DEFAULT: int = 16000 # Generic inspector prompt. Carries NO scenario-specific tokens — every # fact about the current dwelling is injected as a structured JSON # inventory in the user prompt. The system prompt locks the role # (content-fidelity inspector, not classifier), forbids re-labelling, # and requires evidence in generic visual terms only. _SYSTEM_PROMPT_GENERIC: str = ( "You are a vision-only inspector of a rendered top-down floor-plan " "diagram. The diagram draws small numbered circle markers; each " "number corresponds to a base structural element, opening, fixed " "fixture, or anchor furniture item that the dossier inventory lists " "with an expected label and an expected layer. " "For EVERY marker number in the dossier inventory, locate that " "marker in the image and judge whether the object/shape actually " "drawn at its position matches the object class implied by its " "expected_label and expected_layer. " "Set semantic_match='match' when the drawn glyph clearly depicts the " "expected object class; 'mismatch' when it clearly depicts a " "different object class, sits on an unrelated area, or the expected " "object is absent from the image; 'uncertain' when you genuinely " "cannot tell. " "You do NOT re-label or re-classify markers — carry expected_label " "and expected_layer through verbatim from the dossier inventory. " "Describe observed_object_summary and source_ref in generic visual " "terms only (shape, position, size) — no proper nouns, no " "scenario-specific words. mismatch_reason is empty unless the " "verdict is mismatch or uncertain. " "Respond with one JSON document matching the provided response " "schema; emit no prose, no markdown." ) def _build_dossier_facts(*, dossier: Dict[str, Any]) -> Dict[str, Any]: """Project the dossier into the JSON inventory the VLM sees. Carries each base marker's number / label / category / layer / position_hint verbatim so the model can locate and judge it without re-classifying. Overlay/transient markers are intentionally excluded — only base structural fidelity is gated here. """ inventory_out: List[Dict[str, Any]] = [] for entry in (dossier or {}).get("base_marker_inventory") or []: if not isinstance(entry, dict): continue decision = entry.get("base_layer_decision") if decision not in { "base_structural_unit", "base_opening", "base_persistent_fixture", "base_persistent_furniture", }: continue inventory_out.append( { "number": entry.get("number"), "expected_label": entry.get("label"), "expected_layer": decision, "category": entry.get("category"), "position_hint": entry.get("position_hint"), } ) return { "fp_id": dossier.get("fp_id"), "base_marker_inventory": inventory_out, } def _build_messages( *, dossier_facts: Dict[str, Any], image_data_url: str ) -> List[Dict[str, Any]]: user_intro = ( "Dossier facts (JSON, generic schema — fp_id + base_marker_" "inventory with number / expected_label / expected_layer / " "category / position_hint). Judge each numbered marker's drawn " "content against its expected_label/expected_layer. Carry " "expected_label and expected_layer verbatim into your output." ) user_outro = ( "Now respond with the JSON semantic readback per the response " "schema. status MUST be 'ok'. fp_id MUST equal the dossier " "fp_id. Emit one observed_marker_semantics entry for every " "marker number in the dossier inventory. No commentary outside " "the JSON document." ) return [ {"role": "system", "content": _SYSTEM_PROMPT_GENERIC}, { "role": "user", "content": [ {"type": "text", "text": user_intro}, { "type": "text", "text": json.dumps(dossier_facts, sort_keys=True), }, { "type": "image_url", "image_url": {"url": image_data_url, "detail": "original"}, }, {"type": "text", "text": user_outro}, ], }, ] def _build_response_format() -> Dict[str, Any]: return { "type": "json_schema", "json_schema": { "name": "floor_plan_semantic_readback", "strict": True, "schema": SEMANTIC_READBACK_SCHEMA, }, } def litellm_semantic_vlm_provider( *, dossier: Dict[str, Any], fp_image_path: Optional[str], model: str = PROVIDER_MODEL_DEFAULT, max_completion_tokens: int = MAX_COMPLETION_TOKENS_DEFAULT, timeout_seconds: int = 120, ) -> Dict[str, Any]: """Real-VLM semantic readback via a single litellm vision call. Fail-closed at every joint. The completion call count is 0 whenever the helper fail-closes during preflight (arg/env/capability/fs). Output is validated through the core ``validate_semantic_output``; any non-conforming model output raises ``SemanticReadbackError``. """ # 1. Arg-shape guards (pre-env, pre-network). if fp_image_path is None: raise SemanticReadbackError( "litellm_semantic_vlm_provider requires fp_image_path; got None" ) fp_id = dossier.get("fp_id") if isinstance(dossier, dict) else None if not isinstance(fp_id, str) or not fp_id: raise SemanticReadbackError("dossier.fp_id missing or non-string") # 1b. Empty-inventory preflight (pre-env, pre-network). A dossier # with zero base_* markers gives the VLM nothing to attest to; the # validator would only reject it AFTER a costly completion call, so # we fail closed here. Matches the core synthetic fixture's guard. dossier_facts = _build_dossier_facts(dossier=dossier) if not dossier_facts["base_marker_inventory"]: raise SemanticReadbackError( f"dossier fp_id={fp_id!r} has zero base_* markers; refusing " f"the VLM call (costly no-op)" ) # 2. Env preflight — ★#92 (2026-08-27): **모델이 정하는 provider** 기준. from app.modules.pipeline.vlm_auth import auth_kwargs, provider_of _auth_kw = auth_kwargs(model, SemanticReadbackError) # ★호출당 한 번만 # 3. Lazy import 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 — env-dependent raise SemanticReadbackError( f"litellm import failed: {type(exc).__name__}: {exc}" ) from exc # 4. Capability preflight. try: supports_ok = bool(litellm.supports_response_schema(model=model)) except Exception as exc: raise SemanticReadbackError( f"litellm.supports_response_schema raised: " f"{type(exc).__name__}: {exc}" ) from exc if not supports_ok: raise SemanticReadbackError( f"model {model!r} does not advertise response_schema support" ) try: supported_params = ( litellm.get_supported_openai_params( model=model, custom_llm_provider=provider_of(model) ) or [] ) except Exception as exc: raise SemanticReadbackError( f"litellm.get_supported_openai_params raised: " f"{type(exc).__name__}: {exc}" ) from exc if "response_format" not in supported_params: raise SemanticReadbackError( f"model {model!r} does not list response_format in supported " f"openai params; refusing to call" ) # 5. Filesystem preflight. path_obj = Path(fp_image_path) if not path_obj.exists() or not path_obj.is_file(): raise SemanticReadbackError( f"fp_image_path does not exist on disk: {fp_image_path!r}" ) try: image_bytes = path_obj.read_bytes() except Exception as exc: raise SemanticReadbackError( f"fp_image_path read failed: {type(exc).__name__}: {exc}" ) from exc if not image_bytes: raise SemanticReadbackError(f"fp_image_path is empty: {fp_image_path!r}") # 6. Base64 data URL. image_b64 = base64.b64encode(image_bytes).decode("ascii") image_data_url = f"data:image/png;base64,{image_b64}" # 7. Assemble request. ``dossier_facts`` was built at the # empty-inventory preflight above and is reused here. messages = _build_messages( dossier_facts=dossier_facts, image_data_url=image_data_url ) response_format = _build_response_format() # 8. Single completion call (num_retries=0, no temperature). try: response = _llm_completion( model=model, **_auth_kw, messages=messages, response_format=response_format, timeout=timeout_seconds, max_completion_tokens=max_completion_tokens, num_retries=0, ) except Exception as exc: raise SemanticReadbackError( f"litellm.completion raised: {type(exc).__name__}: {exc}" ) from exc # 9. Response guards (fail-closed at every joint). choices = getattr(response, "choices", None) or [] if not choices: raise SemanticReadbackError("response.choices is empty") choice = choices[0] msg = getattr(choice, "message", None) if msg is None: raise SemanticReadbackError("response.choices[0].message is missing") refusal = getattr(msg, "refusal", None) if refusal: raise SemanticReadbackError( f"VLM emitted a refusal: {str(refusal)[:200]!r}" ) content = getattr(msg, "content", None) if not content: raise SemanticReadbackError( "response.choices[0].message.content is empty" ) finish_reason = getattr(choice, "finish_reason", None) if finish_reason != "stop": raise SemanticReadbackError( f"finish_reason must be 'stop' exactly (got {finish_reason!r}); " f"length/content_filter fail closed" ) # 10. Parse JSON content. try: parsed = json.loads(content) except (TypeError, ValueError) as exc: raise SemanticReadbackError( f"response content is not valid JSON: {type(exc).__name__}: {exc}" ) from exc # 11. Validate against dossier exact-ID + shape contract. res = validate_semantic_output(output=parsed, dossier=dossier, fp_id=fp_id) if not res["ok"]: joined = "; ".join(res["blockers"]) raise SemanticReadbackError(f"validator_failed: {joined[:400]}") return res["readback"]