Update vlm_caption.py

vlm_caption.py

@@ -1,17 +1,20 @@
 """
-VLM captioning: image + structured YOLO detections -> safety-aware observation.
+VLM captioning: image [+ optional YOLO detections] -> safety-aware observation.
 
-This is the LANGUAGE half of the pipeline. The VLM receives:
-  - the image itself (so it can describe scene/context YOLO can't)
-  - the deterministic detections + compliance findings from detector.py
+This is the LANGUAGE half of the pipeline. It supports MULTIPLE models behind a
+single interface, and two modes:
 
-It is instructed to treat the detector's compliance findings as the source of
-truth for PPE claims, and to use the image only for context/description. That
-boundary is what keeps the safety judgement auditable.
+  1. WITH detector (default): the VLM receives the image AND the deterministic
+     detections + compliance findings, and is told to treat the detector's
+     compliance findings as authoritative. This is the AiGiS design.
 
-Supports MULTIPLE models behind one interface:
-  - Claude models (Anthropic SDK)         -> require ANTHROPIC_API_KEY
-  - Qwen3-VL models (DashScope, OpenAI-compatible) -> require DASHSCOPE_API_KEY
+  2. IMAGE-ONLY (use_detector=False): the VLM receives the image alone, with no
+     detector JSON, and judges PPE from the image itself. This is the baseline
+     that shows what the detector adds.
+
+Providers:
+  - Claude models (Anthropic SDK)                  require ANTHROPIC_API_KEY
+  - Qwen3-VL models (DashScope, OpenAI-compatible) require DASHSCOPE_API_KEY
 
 Add a new model by adding one entry to MODEL_REGISTRY.
 """
@@ -47,7 +50,20 @@ DEFAULT_MODEL = "sonnet"
 DASHSCOPE_BASE_URL = "https://dashscope-intl.aliyuncs.com/compatible-mode/v1"
 
 
-SYSTEM_PROMPT = """You are a workplace-safety observation assistant for the AiGiS platform.
+# ---------------------------------------------------------------------------
+# Prompts
+# ---------------------------------------------------------------------------
+# Shared output contract for both modes.
+_SYSTEM_COMMON = """Return ONLY a JSON object with these keys, nothing else:
+  "caption": one-sentence description of the scene,
+  "safety_observation": 2-4 sentences on PPE compliance and visible hazards,
+  "ppe_summary": object mapping each assessed body part to "present"/"missing"/"uncertain",
+  "severity": one of "none", "low", "medium", "high" — your read of overall risk.
+
+Be concise, factual, and neutral. This is a safety record, not marketing copy."""
+
+# Mode 1: detector output provided and authoritative.
+SYSTEM_WITH_DETECTOR = """You are a workplace-safety observation assistant for the AiGiS platform.
 
 You are given (1) an image of a work environment and (2) structured detections
 from a deterministic object-detection model, including PPE-compliance findings.
@@ -62,14 +78,28 @@ Rules:
   that the detector does not cover.
 - Do not invent objects or people not supported by the detections or clearly
   visible in the image.
-- Be concise, factual, and neutral. This is a safety record, not marketing copy.
 
-Return ONLY a JSON object with these keys, nothing else:
-  "caption": one-sentence description of the scene,
-  "safety_observation": 2-4 sentences on PPE compliance and visible hazards,
-  "ppe_summary": object mapping each assessed body part to "present"/"missing"/"uncertain",
-  "severity": one of "none", "low", "medium", "high" — your read of overall risk.
-"""
+""" + _SYSTEM_COMMON
+
+# Mode 2: image only, no detector.
+SYSTEM_IMAGE_ONLY = """You are a workplace-safety observation assistant for the AiGiS platform.
+
+You are given an image of a work environment. There is no object-detection input;
+assess the scene from the image alone.
+
+Rules:
+- Identify the people and safety-relevant PPE you can see (helmet, safety vest,
+  gloves, glasses/eye protection, footwear, etc.) and judge whether required PPE
+  appears present or missing for the visible activity.
+- Note visible hazards (heights, machinery, sparks, traffic, etc.).
+- Be explicit that these judgements come from the image alone and may need
+  on-site verification.
+- Do not invent objects or people not clearly visible in the image.
+
+""" + _SYSTEM_COMMON
+
+# Backwards-compatible alias (older code referenced SYSTEM_PROMPT).
+SYSTEM_PROMPT = SYSTEM_WITH_DETECTOR
 
 
 # ---------------------------------------------------------------------------
@@ -94,7 +124,9 @@ def _resolve(model: str) -> tuple[str, str, str]:
     return model, provider, model
 
 
-def _build_user_text(detection_result: dict) -> str:
+def _build_user_text(detection_result: dict | None) -> str:
+    if detection_result is None:
+        return "Write the safety observation for this image as specified."
     return (
         "Structured detector output (authoritative for PPE claims):\n"
         + json.dumps(detection_result, indent=2)
@@ -114,14 +146,14 @@ def _parse(text: str) -> dict:
 # ---------------------------------------------------------------------------
 # Provider back-ends
 # ---------------------------------------------------------------------------
-def _caption_anthropic(model_id, media_type, b64, user_text) -> dict:
+def _caption_anthropic(model_id, system, media_type, b64, user_text) -> dict:
     import anthropic
 
     client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
     msg = client.messages.create(
         model=model_id,
         max_tokens=1024,
-        system=SYSTEM_PROMPT,
+        system=system,
         messages=[
             {
                 "role": "user",
@@ -143,7 +175,7 @@ def _caption_anthropic(model_id, media_type, b64, user_text) -> dict:
     return _parse(text)
 
 
-def _caption_dashscope(model_id, media_type, b64, user_text) -> dict:
+def _caption_dashscope(model_id, system, media_type, b64, user_text) -> dict:
     # DashScope is OpenAI-compatible; the OpenAI SDK is the client.
     from openai import OpenAI
 
@@ -155,7 +187,7 @@ def _caption_dashscope(model_id, media_type, b64, user_text) -> dict:
         model=model_id,
         max_tokens=1024,
         messages=[
-            {"role": "system", "content": SYSTEM_PROMPT},
+            {"role": "system", "content": system},
             {
                 "role": "user",
                 "content": [
@@ -181,37 +213,55 @@ _BACKENDS = {
 # ---------------------------------------------------------------------------
 # Public entry points
 # ---------------------------------------------------------------------------
-def caption(image_path: str, detection_result: dict, model: str = DEFAULT_MODEL) -> dict:
+def caption(
+    image_path: str,
+    detection_result: dict | None = None,
+    model: str = DEFAULT_MODEL,
+    use_detector: bool = True,
+) -> dict:
     """
-    detection_result: the dict from detector.DetectionResult.to_dict()
+    detection_result: the dict from detector.DetectionResult.to_dict(), or None.
     model:            registry key ("sonnet", "qwen-32b", ...) or a raw model id.
-    Returns the parsed JSON dict from the VLM.
+    use_detector:     True  -> send detections, treat as authoritative.
+                      False -> image-only (no detector JSON in the prompt).
     """
     media_type, b64 = _encode_image(image_path)
-    return _caption_from_b64(media_type, b64, detection_result, model)
+    return _caption_from_b64(media_type, b64, detection_result, model, use_detector)
 
 
 def caption_bytes(
     image_bytes: bytes,
     media_type: str,
-    detection_result: dict,
+    detection_result: dict | None = None,
     model: str = DEFAULT_MODEL,
+    use_detector: bool = True,
 ) -> dict:
     """
     Same as caption(), but for an image already in memory instead of one on disk.
     """
     b64 = base64.standard_b64encode(image_bytes).decode()
-    return _caption_from_b64(media_type, b64, detection_result, model)
+    return _caption_from_b64(media_type, b64, detection_result, model, use_detector)
 
 
-def _caption_from_b64(media_type: str, b64: str, detection_result: dict, model: str) -> dict:
+def _caption_from_b64(
+    media_type: str,
+    b64: str,
+    detection_result: dict | None,
+    model: str,
+    use_detector: bool,
+) -> dict:
     key, provider, model_id = _resolve(model)
-    user_text = _build_user_text(detection_result)
+
+    # Only use detector mode if we were actually given detections.
+    with_detector = use_detector and detection_result is not None
+    system = SYSTEM_WITH_DETECTOR if with_detector else SYSTEM_IMAGE_ONLY
+    user_text = _build_user_text(detection_result if with_detector else None)
 
     backend = _BACKENDS.get(provider)
     if backend is None:
         raise ValueError(f"Unknown provider '{provider}' for model '{model}'")
 
-    result = backend(model_id, media_type, b64, user_text)
+    result = backend(model_id, system, media_type, b64, user_text)
     result["_model"] = key
-    return result
+    result["_use_detector"] = with_detector
+    return result