Validate and recover structured captions

examples/worker.openai-compatible.yaml

@@ -20,6 +20,21 @@ worker:
     #   - "i cannot describe"
     #   - "unable to provide a caption"
 
+    # Structured-output validation is optional. When enabled, invalid JSON is
+    # treated like an empty/refused result and can use inference.retry_prompt.
+    # The remaining transforms require validate_json_output: true.
+    # validate_json_output: true
+    # repair_invalid_json_escapes: true
+    # canonicalize_json_output: true
+    # normalize_yxyx_bboxes: true
+    # deduplicate_json_elements: true
+
+    # Provider-specific fields applied only to fallback requests. This is
+    # useful when the retry needs a stricter or smaller response schema.
+    # retry_extra_body:
+    #   response_format:
+    #     type: "json_object"
+
     endpoints:
       - name: "zai-coding-plan"
         base_url: "https://api.z.ai/api/coding/paas/v4"
@@ -33,6 +48,11 @@ worker:
         # requests_per_minute: 600
         timeout_seconds: 120
         max_retries: 6
+        # Provider-specific fields shared by primary and fallback requests can
+        # be placed here, including response_format or chat-template options.
+        # extra_body:
+        #   response_format:
+        #     type: "json_object"
 
       # Add more accounts or providers to aggregate their capacity. Model names
       # can differ between endpoints.

recipes/STRUCTURED_JSON.md

@@ -0,0 +1,104 @@
+# Structured JSON captions
+
+CaptionFlow's OpenAI-compatible worker can request provider-constrained JSON,
+reject malformed responses, and retry failed items with a smaller schema or
+different sampling settings. This works for Ideogram-style captions and other
+JSON caption formats without coupling CaptionFlow to one schema.
+
+## Configure the worker
+
+Provider-specific request fields remain local to the worker. Put the primary
+structured-output constraint in an endpoint's `extra_body` and enable local
+validation on the worker:
+
+```yaml
+worker:
+  server: "ws://orchestrator.example:8765"
+  token: "replace-with-captionflow-worker-token"
+
+  openai_compatible:
+    validate_json_output: true
+    canonicalize_json_output: true
+
+    # Optional repairs for common model failures. Bounding boxes are expected
+    # in [ymin, xmin, ymax, xmax] order.
+    repair_invalid_json_escapes: true
+    normalize_yxyx_bboxes: true
+    deduplicate_json_elements: true
+
+    endpoints:
+      - name: "local-vllm"
+        base_url: "http://inference.example:8000/v1"
+        api_key_env: "VLLM_API_KEY"
+        model: "vision-model"
+        initial_concurrency: 16
+        max_concurrency: 16
+        extra_body:
+          response_format:
+            type: "json_schema"
+            json_schema:
+              name: "caption"
+              strict: true
+              schema:
+                type: "object"
+                additionalProperties: false
+                required: ["description"]
+                properties:
+                  description: {type: "string"}
+```
+
+`validate_json_output` checks strict JSON syntax, including rejecting
+non-standard `NaN` and `Infinity` values. It does not itself implement JSON
+Schema validation; use the endpoint's `response_format` when the provider
+supports constrained decoding. The optional transforms require
+`validate_json_output: true`.
+
+`repair_invalid_json_escapes` only escapes backslashes that cannot begin a
+valid JSON escape. `normalize_yxyx_bboxes` corrects inverted coordinate pairs;
+it does not infer boxes, clamp their range, or assess localization quality.
+`deduplicate_json_elements` removes exact duplicate objects from arrays named
+`elements`.
+
+## Configure recovery
+
+The orchestrator owns the shared prompt and standard sampling settings. A
+response that is empty, refused, rejected by a provider safety filter, or
+invalid JSON can receive one semantic retry:
+
+```yaml
+orchestrator:
+  inference:
+    inference_prompts:
+      - >-
+        Describe the image as one JSON object matching the requested schema.
+        Return JSON only.
+    sampling:
+      temperature: 0.3
+      max_tokens: 2048
+    retry_prompt: >-
+      Try again. Return exactly one valid JSON object with a concise,
+      visually grounded description and no surrounding prose.
+    retry_sampling:
+      temperature: 0.1
+      max_tokens: 1024
+```
+
+`retry_sampling` accepts the standard OpenAI sampling fields supported by the
+worker. It can be configured globally under `inference` or per stage. Changes
+received from the orchestrator apply without restarting the worker.
+
+Some providers need a different constrained-decoding body for the retry. Set
+that locally so provider details and credentials never pass through the
+orchestrator:
+
+```yaml
+worker:
+  openai_compatible:
+    retry_extra_body:
+      response_format:
+        type: "json_object"
+```
+
+Retry request fields override the endpoint's primary `extra_body`, except for
+`model` and `messages`, which CaptionFlow reserves. Non-standard sampling such
+as vLLM's `repetition_penalty` can also be placed in `retry_extra_body`.

src/caption_flow/models.py

@@ -184,6 +184,7 @@ class ProcessingStage:
     # provider rejects the image before inference.
     retry_prompt: Optional[str] = None
     retry_without_image: bool = False
+    retry_sampling: Optional[Dict[str, Any]] = None
 
 
 @dataclass

src/caption_flow/utils/vllm_config.py

@@ -43,6 +43,7 @@ class VLLMConfigManager:
         "inference_prompts",
         "refusal_markers",
         "retry_prompt",
+        "retry_sampling",
         "retry_without_image",
     }
 

src/caption_flow/workers/caption.py

@@ -505,6 +505,9 @@ async def _execute_post_hook(self):
     def _parse_stages_config(self, vllm_config: Dict[str, Any]) -> List[ProcessingStage]:
         """Parse stages configuration from vLLM config."""
         stages_config = vllm_config.get("stages", [])
+        default_retry_sampling = vllm_config.get("retry_sampling")
+        if default_retry_sampling is not None and not isinstance(default_retry_sampling, dict):
+            raise ValueError("retry_sampling must be a mapping")
 
         if not stages_config:
             # Backward compatibility
@@ -517,12 +520,18 @@ def _parse_stages_config(self, vllm_config: Dict[str, Any]) -> List[ProcessingSt
                     requires=[],
                     retry_prompt=vllm_config.get("retry_prompt"),
                     retry_without_image=bool(vllm_config.get("retry_without_image", False)),
+                    retry_sampling=(
+                        dict(default_retry_sampling) if default_retry_sampling is not None else None
+                    ),
                 )
             ]
 
         # Parse stages
         stages = []
         for stage_cfg in stages_config:
+            retry_sampling = stage_cfg.get("retry_sampling", default_retry_sampling)
+            if retry_sampling is not None and not isinstance(retry_sampling, dict):
+                raise ValueError(f"Stage '{stage_cfg['name']}' retry_sampling must be a mapping")
             stage = ProcessingStage(
                 name=stage_cfg["name"],
                 model=stage_cfg.get("model", vllm_config.get("model")),
@@ -541,6 +550,7 @@ def _parse_stages_config(self, vllm_config: Dict[str, Any]) -> List[ProcessingSt
                         vllm_config.get("retry_without_image", False),
                     )
                 ),
+                retry_sampling=dict(retry_sampling) if retry_sampling is not None else None,
             )
             stages.append(stage)