Spaces:
Runtime error
Runtime error
File size: 7,848 Bytes
a1e2ff8 732b14f a1e2ff8 732b14f a1e2ff8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 | """GPT-4o Vision analyser for RICS section photos.
Extracts professional survey observations from uploaded images.
Has a hard timeout and retry logic to handle transient API failures gracefully.
"""
from __future__ import annotations
import base64
import json
import logging
from pathlib import Path
from typing import TYPE_CHECKING
if TYPE_CHECKING:
pass
logger = logging.getLogger(__name__)
_VISION_SYSTEM_PROMPT = """You are a UK Chartered Building Surveyor (MRICS) examining a photograph taken during a residential property inspection.
Analyse the image and extract factual, professional observations suitable for a RICS survey report. Focus on:
- Visible condition (good, fair, poor, defective)
- Materials and construction type
- Defects (cracks, damp stains, decay, displacement)
- Maintenance items or recommendations
Do NOT invent or assume defects that are not clearly visible. State only what can be observed.
Output a JSON object with a single key "observations" containing a list of concise bullet strings (max 8 items)."""
_VISION_USER_TEMPLATE = "This photo is from report section: {section_label}. Describe the relevant observations."
_VISION_TIMEOUT_SECONDS: float = 45.0
_VISION_MAX_ATTEMPTS: int = 2
class VisionAnalysisResult:
"""Outcome of a vision-analysis attempt.
Attributes:
observations: Extracted bullets (may be empty).
ok: ``True`` if the call succeeded.
error: Short, human-readable failure reason (``""`` if ok).
"""
__slots__ = ("observations", "ok", "error")
def __init__(self, observations: list[str], ok: bool, error: str = "") -> None:
self.observations = observations
self.ok = ok
self.error = error
def analyze_section_photo(
image_path: str | Path,
mime_type: str,
section_label: str,
openai_api_key: str,
vision_model: str = "gpt-4o",
) -> VisionAnalysisResult:
"""Analyse a section photo and return observation bullets.
Has a hard timeout of ~45 seconds and retries once on transient errors
(timeout / connection / 5xx). The return type is a small structured object
so the caller can surface a useful error message to the UI when vision
fails — avoiding silent skips on a paid API call.
Args:
image_path: Absolute path to the stored image file.
mime_type: MIME type of the image (``image/jpeg``, ``image/png``, etc.).
section_label: Human-readable section name (e.g. "Main Roof").
openai_api_key: OpenAI API key.
vision_model: Model that supports vision (default ``gpt-4o``).
Returns:
VisionAnalysisResult with observations list, ok flag, and error message.
"""
if not openai_api_key:
logger.debug("No OpenAI key — skipping vision analysis")
return VisionAnalysisResult([], ok=False, error="OpenAI API key not configured.")
try:
image_bytes = Path(image_path).read_bytes()
except OSError as exc:
logger.warning("Could not read photo %s: %s", image_path, exc)
return VisionAnalysisResult([], ok=False, error=f"Could not read image file: {exc}")
b64 = base64.standard_b64encode(image_bytes).decode("ascii")
data_url = f"data:{mime_type};base64,{b64}"
from app.config import settings as app_settings
from openai import APIConnectionError, APIStatusError, APITimeoutError, OpenAI
last_error: str = ""
for attempt in range(_VISION_MAX_ATTEMPTS):
try:
messages = [
{"role": "system", "content": _VISION_SYSTEM_PROMPT},
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {"url": data_url, "detail": "high"},
},
{
"type": "text",
"text": _VISION_USER_TEMPLATE.format(section_label=section_label),
},
],
},
]
from app.llm.openai_chat import chat_completions_create_raw
async def _vision_call() -> str:
return await chat_completions_create_raw(
messages=messages,
model=vision_model,
max_tokens=400,
temperature=0.0,
response_format={"type": "json_object"},
phase="section_photo_vision",
section_id=section_label,
api_key=openai_api_key,
)
import asyncio
import concurrent.futures
def _run_in_fresh_loop() -> str:
return asyncio.run(_vision_call())
try:
asyncio.get_running_loop()
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
raw = pool.submit(_run_in_fresh_loop).result(
timeout=_VISION_TIMEOUT_SECONDS + 10
)
except RuntimeError:
raw = _run_in_fresh_loop()
data = json.loads(raw)
# Distinguish three cases that the previous .get(key, []) collapsed:
# 1. key absent → contract violation, surface as ok=False
# 2. key present, !list → contract violation, surface as ok=False
# 3. key present, list → genuine result (empty list is "model saw nothing", ok=True)
if "observations" not in data:
logger.warning(
"Vision response missing 'observations' key: %s",
raw[:200],
)
return VisionAnalysisResult(
[],
ok=False,
error="Vision response missing 'observations' key.",
)
obs = data["observations"]
if not isinstance(obs, list):
logger.warning(
"Vision response 'observations' is not a list (got %s): %s",
type(obs).__name__,
raw[:200],
)
return VisionAnalysisResult(
[],
ok=False,
error=f"Invalid 'observations' field type: {type(obs).__name__}.",
)
clean = [str(o).strip() for o in obs if isinstance(o, str) and o.strip()][:8]
logger.info("Vision analysis OK: %d observations from %s", len(clean), image_path)
return VisionAnalysisResult(clean, ok=True, error="")
except (APITimeoutError, APIConnectionError) as exc:
last_error = f"Network error: {exc}"
logger.warning("Vision API transient error (attempt %d/%d): %s", attempt + 1, _VISION_MAX_ATTEMPTS, exc)
continue
except APIStatusError as exc:
if exc.status_code >= 500:
last_error = f"Server error ({exc.status_code})"
logger.warning("Vision API 5xx (attempt %d/%d): %s", attempt + 1, _VISION_MAX_ATTEMPTS, exc)
continue
else:
last_error = f"API error ({exc.status_code}): {exc.message}"
logger.warning("Vision API client error: %s", exc)
break
except json.JSONDecodeError as exc:
last_error = f"Invalid JSON response: {exc}"
logger.warning("Vision response not valid JSON: %s", exc)
break
except Exception as exc:
last_error = f"Unexpected error: {exc}"
logger.exception("Vision analysis unexpected failure")
break
return VisionAnalysisResult([], ok=False, error=last_error or "Vision analysis failed after retries.")
|