From 0653e1725023324ee29646da381f8cdd41213ac5 Mon Sep 17 00:00:00 2001 From: Iuri de Silvio Date: Thu, 29 Oct 2026 19:45:03 +0100 Subject: [PATCH] Add reference examples to the Gemini v6 and OpenAI v7 detection prompts google_gemini@v6 and open_ai@v7 take an optional reference_examples input for object-detection: up to 10 labeled example images that go before the image to label in every request, so the model sees what each class looks like and how tightly it is boxed. Gemini gets the examples with their boxes and labels drawn on them; OpenAI gets up to 3 crops per label, grouped by label. Without the input the requests are unchanged. Co-Authored-By: Claude Opus 5.5 --- workflows/CHANGELOG.md | 1 + .../common/vlm_reference_examples.py | 500 ++++++++++++++++++ .../models/foundation/google_gemini/v6.py | 58 ++ .../core_steps/models/foundation/openai/v7.py | 72 ++- .../common/test_vlm_reference_examples.py | 337 ++++++++++++ .../foundation/test_google_gemini_v6.py | 135 +++++ .../models/foundation/test_openai_v7.py | 154 ++++++ 7 files changed, 1255 insertions(+), 2 deletions(-) create mode 100644 workflows/roboflow_workflows/core_steps/common/vlm_reference_examples.py create mode 100644 workflows/tests/unit_tests/core_steps/common/test_vlm_reference_examples.py diff --git a/workflows/CHANGELOG.md b/workflows/CHANGELOG.md index 09ac8d73e7..8b1ff664cc 100644 --- a/workflows/CHANGELOG.md +++ b/workflows/CHANGELOG.md @@ -18,6 +18,7 @@ for contributor and maintainer responsibilities. ### Added +- Google Gemini v6 (`roboflow_core/google_gemini@v6`) and OpenAI v7 (`roboflow_core/open_ai@v7`) gain an optional `reference_examples` input for `object-detection`: up to 10 labeled example images, each `{"image": ..., "boxes": [{"x", "y", "width", "height", "class"}]}` in pixels of that image. The examples go before the image to label in every request, so the model sees what each class looks like and how tightly it is boxed. Gemini gets the example images with their boxes and labels drawn on them; OpenAI gets up to 3 crops per label, grouped by label (the output labels, so `output_classes` when set). Example images load through the host's image codec, with the same URL policy as any workflow image. Other tasks reject the input. Without it the requests are unchanged. - Vision Events (`roboflow_core/roboflow_vision_events@v1`) and Vision Event Bundle (`roboflow_core/vision_event_bundle@v1`) blocks gain an optional `value_base_units` input: what the event is worth, in base units of the use case's currency (e.g. cents for USD). The currency always comes from the use case. The value is sent as `valueBaseUnits` to the Vision Events API and in the bundle `payload.json`, and as `value_base_units` to the local event store. A typed value or workflow input must be a whole number (`true`/`false` are rejected). A value from another step is rounded to whole base units; one that is not a finite number or falls outside ±10,000,000,000 is dropped with a warning and the event is still recorded. Left empty, the use case's default value applies. `operator_feedback` events never carry a value, so the blocks omit it for them. ## `0.2.14` diff --git a/workflows/roboflow_workflows/core_steps/common/vlm_reference_examples.py b/workflows/roboflow_workflows/core_steps/common/vlm_reference_examples.py new file mode 100644 index 0000000000..ec78734ac9 --- /dev/null +++ b/workflows/roboflow_workflows/core_steps/common/vlm_reference_examples.py @@ -0,0 +1,500 @@ +"""Labeled example images for VLM object detection ("label by example"). + +A reference example is an image with its objects boxed and labeled. Detection +blocks send the examples before the image to label, so the model sees what each +label looks like and how tightly it is boxed. There are two renderings because +the providers read them differently: whole images with the boxes drawn on them +(Gemini) and crops of the boxes grouped by label (OpenAI). The examples go first +in the request, so every image of a batch shares the same prefix and the +provider can cache it. +""" + +import base64 +import json +from dataclasses import dataclass +from functools import lru_cache +from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union + +import cv2 +import numpy as np +from PIL import Image, ImageDraw, ImageFont +from pydantic import BaseModel, ConfigDict, Field, ValidationError +from roboflow_workflows.utils.images import encode_image_to_jpeg_bytes, load_image + +REFERENCE_EXAMPLES_TASK = "object-detection" +MAX_REFERENCE_EXAMPLES = 10 +MAX_CROPS_PER_LABEL = 3 +BOXED_IMAGE_MAX_EDGE = 1024 +CROP_MARGIN = 0.15 +CROP_MIN_MARGIN = 8 +CROP_MIN_EDGE = 128 +CROP_MAX_EDGE = 384 + +REFERENCE_CROPS_INTRO = ( + "Reference examples: crops of labeled objects taken from other images of the " + "same dataset, grouped by label. They show what each label looks like." +) +REFERENCE_BOXED_INTRO = ( + "Reference examples: other images of the same dataset where every object with " + "one of the labels is outlined by a colored box with its label written on it. " + "They show what each label looks like and how tightly it is boxed." +) +TARGET_IMAGE_INTRO = "Target image:" +TARGET_ONLY_RULE = ( + " The reference images are not part of the task: return detections for the " + "target image only." +) + +REFERENCE_EXAMPLES_DESCRIPTION = ( + "Labeled example images that show the model what each class looks like and how " + "tightly it is boxed (object-detection only). Each example is " + '`{"image": , "boxes": ' + '[{"x": ..., "y": ..., "width": ..., "height": ..., "class": ...}]}`, with box ' + "centers and sizes in pixels of that example image and `class` one of the " + "block's labels. Box every object of those labels in each example. At most " + f"{MAX_REFERENCE_EXAMPLES} examples. They are sent with every image, so each " + "request costs more input tokens." +) +REFERENCE_EXAMPLES_EXAMPLE = [ + { + "image": {"type": "url", "value": "https://example.com/labeled.jpg"}, + "boxes": [{"x": 120, "y": 80, "width": 40, "height": 60, "class": "cat"}], + } +] + +# RGB, picked to stay apart from each other on photos. +_BOX_COLORS = [ + (230, 25, 75), + (60, 180, 75), + (255, 225, 25), + (0, 130, 200), + (245, 130, 48), + (145, 30, 180), + (70, 240, 240), + (240, 50, 230), + (210, 245, 60), + (250, 190, 212), + (0, 128, 128), + (220, 190, 255), + (170, 110, 40), + (128, 0, 0), + (0, 0, 128), +] + + +class ReferenceBox(BaseModel): + """One labeled box of a reference example, in the Roboflow prediction format.""" + + model_config = ConfigDict(populate_by_name=True) + + x: float = Field(description="Box center x, in pixels of the example image.") + y: float = Field(description="Box center y, in pixels of the example image.") + width: float = Field(gt=0, description="Box width in pixels.") + height: float = Field(gt=0, description="Box height in pixels.") + class_name: str = Field( + alias="class", + min_length=1, + description="Label of the object, one of the block's labels.", + examples=["cat"], + ) + + +class ReferenceExample(BaseModel): + """An example image with every object of the block's labels boxed.""" + + image: Union[str, Dict[str, Any]] = Field( + description="Image as a URL, a base64 string or an inference image dict.", + examples=[{"type": "url", "value": "https://example.com/labeled.jpg"}], + ) + boxes: List[ReferenceBox] = Field( + description="Every object of the block's labels in the image." + ) + + +@dataclass(frozen=True) +class LoadedReferenceExample: + """A decoded example image (BGR) with boxes as (label, x_min, y_min, x_max, y_max).""" + + image: np.ndarray + boxes: Tuple[Tuple[str, float, float, float, float], ...] + + +def validate_reference_examples( + value: Optional[Sequence[Any]], + *, + task_type: str, + labels: Optional[Sequence[str]], +) -> List[ReferenceExample]: + """Validate reference examples against the task and the labels of the block. + + Args: + value: Manifest literal or resolved selector value; ``None`` and an empty + list mean no examples. + task_type: Task the block runs; examples are only supported for + object-detection. + labels: Labels the block returns, or ``None`` when they are only known at + runtime (a selector). + + Returns: + Parsed examples, empty when there are none. + + Raises: + ValueError: If an example is malformed, there are too many, the task is not + object-detection, or a box uses a label the block does not have. + """ + examples = _parse_reference_examples(value) + if not examples: + return [] + + if task_type != REFERENCE_EXAMPLES_TASK: + raise ValueError( + f"`reference_examples` are only supported for the {REFERENCE_EXAMPLES_TASK} task" + ) + if labels is not None: + used_labels = {box.class_name for example in examples for box in example.boxes} + unknown = sorted(used_labels - set(labels)) + if unknown: + raise ValueError( + f"`reference_examples` use labels that the block does not have: {unknown}" + ) + + return examples + + +def load_reference_examples( + examples: Sequence[ReferenceExample], +) -> List[LoadedReferenceExample]: + """Decode the example images and convert their boxes to clipped corners. + + Images go through the installed image codec, so URLs follow the same policy as + any other workflow image. + + Args: + examples: Validated examples. + + Returns: + One loaded example per input, in order. + + Raises: + ValueError: If a box lies entirely outside its image. + """ + loaded = [] + for index, example in enumerate(examples): + pixels, _ = load_image(example.image) + height, width = pixels.shape[:2] + boxes = [] + for box in example.boxes: + x_min = max(0.0, box.x - box.width / 2) + y_min = max(0.0, box.y - box.height / 2) + x_max = min(float(width), box.x + box.width / 2) + y_max = min(float(height), box.y + box.height / 2) + if x_max <= x_min or y_max <= y_min: + raise ValueError( + f"`reference_examples[{index}]` has a box outside its " + f"{width}x{height} image" + ) + boxes.append((box.class_name, x_min, y_min, x_max, y_max)) + loaded.append(LoadedReferenceExample(image=pixels, boxes=tuple(boxes))) + return loaded + + +def render_boxed_examples( + examples: Sequence[LoadedReferenceExample], + *, + labels: Sequence[str], +) -> List[str]: + """Draw every box and its label on each example image. + + Args: + examples: Loaded examples. + labels: Block labels; a label's position picks its box color. + + Returns: + One base64 JPEG per example, downscaled to ``BOXED_IMAGE_MAX_EDGE``. + """ + label_positions = {label: index for index, label in enumerate(labels)} + rendered = [] + for example in examples: + height, width = example.image.shape[:2] + scale = min(1.0, BOXED_IMAGE_MAX_EDGE / max(height, width)) + pixels = example.image + if scale < 1.0: + pixels = cv2.resize( + pixels, + (max(1, round(width * scale)), max(1, round(height * scale))), + interpolation=cv2.INTER_AREA, + ) + canvas = Image.fromarray(np.ascontiguousarray(pixels[:, :, ::-1])) + draw = ImageDraw.Draw(canvas) + line_width = max(2, round(max(canvas.size) / 400)) + font = _label_font(max(12, round(max(canvas.size) / 70))) + + for label, x_min, y_min, x_max, y_max in example.boxes: + color = _BOX_COLORS[label_positions[label] % len(_BOX_COLORS)] + box = [x_min * scale, y_min * scale, x_max * scale, y_max * scale] + draw.rectangle(box, outline=color, width=line_width) + left, top, right, bottom = draw.textbbox((0, 0), label, font=font) + tag_top = box[1] - (bottom - top) - 4 + if tag_top < 0: + tag_top = box[1] + line_width + draw.rectangle( + [ + box[0], + tag_top, + box[0] + right - left + 4, + tag_top + bottom - top + 4, + ], + fill=color, + ) + draw.text( + (box[0] + 2 - left, tag_top + 2 - top), + label, + fill=_text_color(color), + font=font, + ) + + bgr = np.ascontiguousarray(np.asarray(canvas)[:, :, ::-1]) + rendered.append(_jpeg_base64(bgr)) + return rendered + + +def render_example_crops( + examples: Sequence[LoadedReferenceExample], + *, + labels: Sequence[str], +) -> List[Tuple[str, List[str]]]: + """Cut crops of the example boxes, grouped by label. + + Args: + examples: Loaded examples. + labels: Block labels; groups follow this order. + + Returns: + ``(label, crops)`` pairs with up to ``MAX_CROPS_PER_LABEL`` base64 JPEG crops + each, largest boxes first. Labels without boxes are left out. + """ + by_label: Dict[str, List[Tuple[np.ndarray, Tuple[float, float, float, float]]]] = { + label: [] for label in labels + } + for example in examples: + for label, x_min, y_min, x_max, y_max in example.boxes: + by_label[label].append((example.image, (x_min, y_min, x_max, y_max))) + + groups = [] + for label in labels: + largest_first = sorted(by_label[label], key=lambda item: -_area(item[1])) + chosen = largest_first[:MAX_CROPS_PER_LABEL] + if chosen: + groups.append( + (label, [_crop_base64(image, box=box) for image, box in chosen]) + ) + return groups + + +def gemini_reference_parts_for( + value: Optional[Sequence[Any]], + *, + labels: Sequence[str], +) -> List[dict]: + """Build Gemini content parts for validated examples: boxed images. + + Results are cached, because a job runs the same examples on every image. + + Args: + value: Examples accepted by ``validate_reference_examples``. + labels: Block labels. + + Returns: + Content parts to place before the target image. + """ + parts = list(_cached_parts("gemini", _cache_key(value), tuple(labels))) + return parts + + +def openai_reference_parts_for( + value: Optional[Sequence[Any]], + *, + labels: Sequence[str], +) -> List[dict]: + """Build OpenAI Responses API content parts for validated examples: crops. + + Results are cached, because a job runs the same examples on every image. + + Args: + value: Examples accepted by ``validate_reference_examples``. + labels: Labels the model returns. + + Returns: + Content parts to place before the target image. + """ + parts = list(_cached_parts("openai", _cache_key(value), tuple(labels))) + return parts + + +def with_reference_parts( + parts: Sequence[dict], + *, + reference_parts: Sequence[dict], + target_intro: dict, + is_image_part: Callable[[dict], bool], +) -> List[dict]: + """Put the references first and point the detection prompt at the target image. + + Args: + parts: Content parts of the detection request without references. + reference_parts: Parts built for the examples. + target_intro: Text part placed right before the image to label. + is_image_part: Tells the image part apart from text parts. + + Returns: + New content parts. The last text part, which holds the detection + instructions, has "in this image" pointed at the target image and ends with + a sentence that rules out labeling the references. + """ + text_indices = [index for index, part in enumerate(parts) if "text" in part] + instructions_index = text_indices[-1] if text_indices else None + combined = list(reference_parts) + for index, part in enumerate(parts): + if is_image_part(part): + combined.append(target_intro) + combined.append(part) + elif index == instructions_index: + text = part["text"].replace("in this image", "in the target image", 1) + combined.append({**part, "text": text + TARGET_ONLY_RULE}) + else: + combined.append(part) + return combined + + +def _parse_reference_examples(value: Optional[Sequence[Any]]) -> List[ReferenceExample]: + if not value: + return [] + + if not isinstance(value, (list, tuple)): + raise ValueError("`reference_examples` must be a list of examples") + if len(value) > MAX_REFERENCE_EXAMPLES: + raise ValueError( + f"`reference_examples` accepts at most {MAX_REFERENCE_EXAMPLES} examples, " + f"got {len(value)}" + ) + + examples = [] + for index, item in enumerate(value): + if isinstance(item, ReferenceExample): + examples.append(item) + continue + try: + examples.append(ReferenceExample.model_validate(item)) + except ValidationError as error: + problems = "; ".join( + f"{'.'.join(str(part) for part in detail['loc'])}: {detail['msg']}" + for detail in error.errors() + ) + raise ValueError( + f"`reference_examples[{index}]` is not a valid example: {problems}" + ) from None + return examples + + +def _gemini_parts(boxed_images: Sequence[str]) -> List[dict]: + parts = [{"text": REFERENCE_BOXED_INTRO}] + for index, data in enumerate(boxed_images, start=1): + parts.append({"text": f"Reference image {index}:"}) + parts.append({"inline_data": {"mime_type": "image/jpeg", "data": data}}) + return parts + + +def _openai_parts(crop_groups: Sequence[Tuple[str, Sequence[str]]]) -> List[dict]: + parts = [{"type": "input_text", "text": REFERENCE_CROPS_INTRO}] + for label, crops in crop_groups: + parts.append({"type": "input_text", "text": f'Examples of "{label}":'}) + parts.extend( + {"type": "input_image", "image_url": f"data:image/jpeg;base64,{crop}"} + for crop in crops + ) + return parts + + +def _cache_key(value: Optional[Sequence[Any]]) -> str: + items = [ + item.model_dump(by_alias=True) if isinstance(item, ReferenceExample) else item + for item in value or [] + ] + key = json.dumps(items, sort_keys=True) + return key + + +# Without the cache every block run would fetch and render the same examples again. +# The key is the full examples JSON, so a hit needs the same images and boxes. +@lru_cache(maxsize=4) +def _cached_parts( + provider: str, examples_json: str, labels: Tuple[str, ...] +) -> Tuple[dict, ...]: + examples = load_reference_examples( + _parse_reference_examples(json.loads(examples_json)) + ) + if provider == "gemini": + parts = _gemini_parts(render_boxed_examples(examples, labels=labels)) + else: + parts = _openai_parts(render_example_crops(examples, labels=labels)) + return tuple(parts) + + +def _area(box: Tuple[float, float, float, float]) -> float: + x_min, y_min, x_max, y_max = box + return (x_max - x_min) * (y_max - y_min) + + +def _crop_base64(image: np.ndarray, *, box: Tuple[float, float, float, float]) -> str: + height, width = image.shape[:2] + x_min, y_min, x_max, y_max = box + margin_x = max(CROP_MIN_MARGIN, (x_max - x_min) * CROP_MARGIN) + margin_y = max(CROP_MIN_MARGIN, (y_max - y_min) * CROP_MARGIN) + left, top = max(0, int(x_min - margin_x)), max(0, int(y_min - margin_y)) + right = min(width, int(np.ceil(x_max + margin_x))) + bottom = min(height, int(np.ceil(y_max + margin_y))) + crop = image[top:bottom, left:right] + + longest = max(crop.shape[:2]) + if longest < CROP_MIN_EDGE: + crop = _resize_longest_edge( + crop, edge=CROP_MIN_EDGE, interpolation=cv2.INTER_CUBIC + ) + elif longest > CROP_MAX_EDGE: + crop = _resize_longest_edge( + crop, edge=CROP_MAX_EDGE, interpolation=cv2.INTER_AREA + ) + + encoded = _jpeg_base64(np.ascontiguousarray(crop)) + return encoded + + +def _resize_longest_edge( + image: np.ndarray, *, edge: int, interpolation: int +) -> np.ndarray: + height, width = image.shape[:2] + scale = edge / max(height, width) + size = (max(1, round(width * scale)), max(1, round(height * scale))) + resized = cv2.resize(image, size, interpolation=interpolation) + return resized + + +def _jpeg_base64(image: np.ndarray) -> str: + encoded = base64.b64encode(encode_image_to_jpeg_bytes(image)).decode("ascii") + return encoded + + +def _label_font(size: int) -> ImageFont.ImageFont: + # The scalable default font needs FreeType; without it Pillow has only a + # fixed-size bitmap font. + try: + font = ImageFont.load_default(size=size) + except (TypeError, OSError): + font = ImageFont.load_default() + return font + + +def _text_color(color: Tuple[int, int, int]) -> Tuple[int, int, int]: + red, green, blue = color + luminance = 0.299 * red + 0.587 * green + 0.114 * blue + return (0, 0, 0) if luminance > 150 else (255, 255, 255) diff --git a/workflows/roboflow_workflows/core_steps/models/foundation/google_gemini/v6.py b/workflows/roboflow_workflows/core_steps/models/foundation/google_gemini/v6.py index 8a6bb70448..f3f2d00fc9 100644 --- a/workflows/roboflow_workflows/core_steps/models/foundation/google_gemini/v6.py +++ b/workflows/roboflow_workflows/core_steps/models/foundation/google_gemini/v6.py @@ -44,6 +44,16 @@ ARGON_SEGMENTATION_SCHEMA, decode_argon_segmentations, ) +from roboflow_workflows.core_steps.common.vlm_reference_examples import ( + MAX_REFERENCE_EXAMPLES, + REFERENCE_EXAMPLES_DESCRIPTION, + REFERENCE_EXAMPLES_EXAMPLE, + REFERENCE_EXAMPLES_TASK, + TARGET_IMAGE_INTRO, + gemini_reference_parts_for, + validate_reference_examples, + with_reference_parts, +) from roboflow_workflows.core_steps.common.vlms import VLM_TASKS_METADATA from roboflow_workflows.environment import ( WORKFLOWS_REMOTE_EXECUTION_MAX_STEP_CONCURRENT_REQUESTS, @@ -265,6 +275,16 @@ Separate `roboflow_core/vlm_as_detector@v2` and `roboflow_core/vlm_as_classifier@v2` steps are no longer needed. +### Reference examples + +For `object-detection`, `reference_examples` takes up to {MAX_REFERENCE_EXAMPLES} +labeled example images. The block draws each example's boxes and labels on the +image (longest edge 1024 pixels) and sends the examples before the image to label, +so Gemini sees what each class looks like and how tightly it is boxed. Box every +object of the listed classes in each example. The examples go with every image, so +they add input tokens to each request; they come first, so the provider can cache +them across a batch. Without examples the request is unchanged. + ### API Key Options This block supports two API key modes: @@ -374,6 +394,21 @@ class BlockManifest(WorkflowBlockManifest): }, }, ) + reference_examples: Optional[ + Union[Selector(kind=[LIST_OF_VALUES_KIND]), List[Dict[str, Any]]] + ] = Field( + default=None, + description=REFERENCE_EXAMPLES_DESCRIPTION, + examples=[REFERENCE_EXAMPLES_EXAMPLE, "$inputs.reference_examples"], + json_schema_extra={ + "relevant_for": { + "task_type": { + "values": {REFERENCE_EXAMPLES_TASK}, + "required": False, + }, + }, + }, + ) api_key: Union[ Selector(kind=[STRING_KIND, SECRET_KIND, ROBOFLOW_MANAGED_KEY]), str ] = Field( @@ -492,6 +527,12 @@ def validate(self) -> "BlockManifest": raise ValueError( f"`output_structure` parameter required to be set for task `{self.task_type}`" ) + if isinstance(self.reference_examples, list): + validate_reference_examples( + self.reference_examples, + task_type=self.task_type, + labels=self.classes if isinstance(self.classes, list) else None, + ) validate_reasoning_level( model=self.model_version, level=self.thinking_level, @@ -585,6 +626,7 @@ def run( google_code_execution: Optional[bool], max_concurrent_requests: Optional[int], api_key: str = "rf_key:account", + reference_examples: Optional[List[Any]] = None, ) -> BlockResult: inference_images = [i.to_inference_format() for i in images] raw_outputs = run_gemini_prompting( @@ -602,6 +644,7 @@ def run( thinking_level=thinking_level, google_code_execution=google_code_execution, max_concurrent_requests=max_concurrent_requests, + reference_examples=reference_examples, ) results = [] for image, (content, input_tokens, output_tokens) in zip(images, raw_outputs): @@ -651,6 +694,7 @@ def run_gemini_prompting( thinking_level: Optional[str], google_code_execution: Optional[bool], max_concurrent_requests: Optional[int], + reference_examples: Optional[List[Any]] = None, ) -> List[Tuple[str, Optional[int], Optional[int]]]: if task_type not in PROMPT_BUILDERS: raise ValueError(f"Task type: {task_type} not supported.") @@ -660,6 +704,13 @@ def run_gemini_prompting( raise ValueError( "Code execution is not supported with segmentation JSON output" ) + reference_parts = [] + if reference_examples: + validate_reference_examples( + reference_examples, task_type=task_type, labels=classes + ) + reference_parts = gemini_reference_parts_for(reference_examples, labels=classes) + gemini_prompts = [] for image in images: loaded_image, _ = load_image(image) @@ -683,6 +734,13 @@ def run_gemini_prompting( thinking_level=thinking_level, max_tokens=max_tokens, ) + if reference_parts: + generated_prompt["contents"]["parts"] = with_reference_parts( + generated_prompt["contents"]["parts"], + reference_parts=reference_parts, + target_intro={"text": TARGET_IMAGE_INTRO}, + is_image_part=lambda part: "inline_data" in part, + ) if ( google_code_execution diff --git a/workflows/roboflow_workflows/core_steps/models/foundation/openai/v7.py b/workflows/roboflow_workflows/core_steps/models/foundation/openai/v7.py index 634b3d386f..fa26ee0b65 100644 --- a/workflows/roboflow_workflows/core_steps/models/foundation/openai/v7.py +++ b/workflows/roboflow_workflows/core_steps/models/foundation/openai/v7.py @@ -50,6 +50,17 @@ decode_vlm_output, describe_vlm_prediction_outputs, ) +from roboflow_workflows.core_steps.common.vlm_reference_examples import ( + MAX_CROPS_PER_LABEL, + MAX_REFERENCE_EXAMPLES, + REFERENCE_EXAMPLES_DESCRIPTION, + REFERENCE_EXAMPLES_EXAMPLE, + REFERENCE_EXAMPLES_TASK, + TARGET_IMAGE_INTRO, + openai_reference_parts_for, + validate_reference_examples, + with_reference_parts, +) from roboflow_workflows.core_steps.common.vlms import VLM_TASKS_METADATA from roboflow_workflows.environment import ( WORKFLOWS_REMOTE_EXECUTION_MAX_STEP_CONCURRENT_REQUESTS, @@ -457,6 +468,15 @@ def get_detection_box_format(model_version: str) -> str: returned label to `output_classes`, and decoded predictions use those stable labels. Omitting `output_classes` preserves the original `classes` contract. +For `object-detection`, `reference_examples` takes up to {MAX_REFERENCE_EXAMPLES} +labeled example images. The block cuts up to {MAX_CROPS_PER_LABEL} crops per label +from the examples' boxes (largest first) and sends them, grouped by label, before +the image to label, so the model sees what each label looks like. Box labels are +the output labels: `output_classes` when set, `classes` otherwise. The crops go +with every image, so they add input tokens to each request; they come first, so +the provider can cache them across a batch. Without examples the request is +unchanged. + The `instance-segmentation` task asks for one outline polygon per instance of the requested classes - a flat `[x1, y1, x2, y2, ...]` vertex list in absolute pixel coordinates of the uploaded image, enforced via structured outputs - and @@ -650,6 +670,21 @@ class BlockManifest(WorkflowBlockManifest): }, ) ) + reference_examples: Optional[ + Union[Selector(kind=[LIST_OF_VALUES_KIND]), List[Dict[str, Any]]] + ] = Field( + default=None, + description=REFERENCE_EXAMPLES_DESCRIPTION, + examples=[REFERENCE_EXAMPLES_EXAMPLE, "$inputs.reference_examples"], + json_schema_extra={ + "relevant_for": { + "task_type": { + "values": {REFERENCE_EXAMPLES_TASK}, + "required": False, + }, + }, + }, + ) api_key: Union[ Selector(kind=[STRING_KIND, SECRET_KIND, ROBOFLOW_MANAGED_KEY]), str ] = Field( @@ -767,6 +802,16 @@ def validate(self) -> "BlockManifest": raise ValueError( f"`output_structure` parameter required to be set for task `{self.task_type}`" ) + if isinstance(self.reference_examples, list): + if isinstance(self.output_classes, list): + labels = self.output_classes + elif self.output_classes is None and isinstance(self.classes, list): + labels = self.classes + else: + labels = None + validate_reference_examples( + self.reference_examples, task_type=self.task_type, labels=labels + ) validate_reasoning_level( model=self.model_version, level=self.reasoning_effort, @@ -865,6 +910,7 @@ def run( output_classes: Optional[List[str]] = None, api_key: str = "rf_key:account", service_tier: Optional[str] = None, + reference_examples: Optional[List[Any]] = None, ) -> BlockResult: resolved_output_classes = validate_output_classes(classes, output_classes) inference_images = [i.to_inference_format() for i in images] @@ -885,6 +931,7 @@ def run( temperature=temperature, max_concurrent_requests=max_concurrent_requests, service_tier=service_tier, + reference_examples=reference_examples, ) box_format = get_detection_box_format(model_version) results = [] @@ -964,13 +1011,15 @@ def run_openai_prompting( temperature: Optional[float], max_concurrent_requests: Optional[int], service_tier: Optional[str] = None, + reference_examples: Optional[List[Any]] = None, ) -> List[Tuple[str, Optional[int], Optional[int]]]: """Encode images, build per-task prompts and execute OpenAI requests. Each image is preprocessed according to the task type (see ``encode_image_for_task``), turned into a request payload by the prompt builder registered for the task in ``PROMPT_BUILDERS`` and executed via - ``execute_openai_requests``. + ``execute_openai_requests``. With ``reference_examples``, crops of the + examples go before the target image in every detection request. Args: roboflow_api_key: Roboflow API key, required only for proxied @@ -989,16 +1038,27 @@ def run_openai_prompting( temperature: Sampling temperature. max_concurrent_requests: Cap on concurrent OpenAI requests. service_tier: OpenAI service tier, omitted from the request when ``None``. + reference_examples: Optional labeled example images for object detection, + in the format ``validate_reference_examples`` accepts. Returns: ``(output_text, input_tokens, output_tokens)`` tuples, one per input image; token counts are ``None`` when the provider omits usage data. Raises: - ValueError: If the task type has no registered prompt builder. + ValueError: If the task type has no registered prompt builder, or the + reference examples are invalid for the task or the labels. """ if task_type not in PROMPT_BUILDERS: raise ValueError(f"Task type: {task_type} not supported.") + reference_parts = [] + if reference_examples: + labels = validate_output_classes(classes, output_classes) + validate_reference_examples( + reference_examples, task_type=task_type, labels=labels + ) + reference_parts = openai_reference_parts_for(reference_examples, labels=labels) + openai_prompts = [] for image in images: loaded_image, _ = load_image(image) @@ -1016,6 +1076,14 @@ def run_openai_prompting( image_height=image_height, model_version=model_version, ) + if reference_parts: + message = generated_prompt["input"][0] + message["content"] = with_reference_parts( + message["content"], + reference_parts=reference_parts, + target_intro={"type": "input_text", "text": TARGET_IMAGE_INTRO}, + is_image_part=lambda part: part.get("type") == "input_image", + ) openai_prompts.append(generated_prompt) return execute_openai_requests( platform_client=platform_client, diff --git a/workflows/tests/unit_tests/core_steps/common/test_vlm_reference_examples.py b/workflows/tests/unit_tests/core_steps/common/test_vlm_reference_examples.py new file mode 100644 index 0000000000..a60ff49148 --- /dev/null +++ b/workflows/tests/unit_tests/core_steps/common/test_vlm_reference_examples.py @@ -0,0 +1,337 @@ +import base64 +from typing import List, Optional, Tuple +from unittest.mock import patch + +import cv2 +import numpy as np +import pytest +from roboflow_workflows.core_steps.common import vlm_reference_examples +from roboflow_workflows.core_steps.common.vlm_reference_examples import ( + CROP_MAX_EDGE, + CROP_MIN_EDGE, + REFERENCE_BOXED_INTRO, + REFERENCE_CROPS_INTRO, + TARGET_IMAGE_INTRO, + TARGET_ONLY_RULE, + gemini_reference_parts_for, + load_reference_examples, + openai_reference_parts_for, + render_boxed_examples, + render_example_crops, + validate_reference_examples, + with_reference_parts, +) + + +@pytest.fixture(autouse=True) +def clear_reference_parts_cache() -> None: + vlm_reference_examples._cached_parts.cache_clear() + + +def _image_base64( + width: int, height: int, color: Tuple[int, int, int] = (255, 255, 255) +) -> str: + image = np.full((height, width, 3), color, dtype=np.uint8) + _, encoded = cv2.imencode(".png", image) + return base64.b64encode(encoded.tobytes()).decode("ascii") + + +def _example( + width: int = 200, + height: int = 100, + boxes: Optional[List[dict]] = None, +) -> dict: + return { + "image": {"type": "base64", "value": _image_base64(width, height)}, + "boxes": ( + boxes + if boxes is not None + else [{"x": 50, "y": 50, "width": 40, "height": 20, "class": "cat"}] + ), + } + + +def _decode(data: str) -> np.ndarray: + return cv2.imdecode( + np.frombuffer(base64.b64decode(data), dtype=np.uint8), cv2.IMREAD_COLOR + ) + + +@pytest.mark.parametrize("value", [None, []]) +def test_validate_reference_examples_treats_missing_examples_as_none(value) -> None: + # when + result = validate_reference_examples( + value, task_type="object-detection", labels=["cat"] + ) + + # then + assert result == [] + + +def test_validate_reference_examples_reads_roboflow_prediction_boxes() -> None: + # when + result = validate_reference_examples( + [_example()], task_type="object-detection", labels=["cat"] + ) + + # then + assert len(result) == 1 + box = result[0].boxes[0] + assert (box.x, box.y, box.width, box.height, box.class_name) == ( + 50, + 50, + 40, + 20, + "cat", + ) + + +def test_validate_reference_examples_rejects_more_than_ten_examples() -> None: + with pytest.raises(ValueError, match="at most 10 examples, got 11"): + validate_reference_examples( + [_example()] * 11, task_type="object-detection", labels=["cat"] + ) + + +def test_validate_reference_examples_rejects_other_tasks() -> None: + with pytest.raises(ValueError, match="only supported for the object-detection"): + validate_reference_examples( + [_example()], task_type="classification", labels=["cat"] + ) + + +def test_validate_reference_examples_rejects_labels_the_block_does_not_have() -> None: + with pytest.raises(ValueError, match=r"does not have: \['cat'\]"): + validate_reference_examples( + [_example()], task_type="object-detection", labels=["dog"] + ) + + +def test_validate_reference_examples_skips_the_label_check_for_unknown_labels() -> None: + # when + result = validate_reference_examples( + [_example()], task_type="object-detection", labels=None + ) + + # then + assert len(result) == 1 + + +def test_validate_reference_examples_names_the_malformed_field() -> None: + # given + example = _example(boxes=[{"x": 50, "y": 50, "height": 20, "class": "cat"}]) + + # when + with pytest.raises(ValueError) as error: + validate_reference_examples( + [example], task_type="object-detection", labels=["cat"] + ) + + # then + assert "reference_examples[0]" in str(error.value) + assert "boxes.0.width" in str(error.value) + + +def test_load_reference_examples_clips_boxes_to_the_image() -> None: + # given + examples = validate_reference_examples( + [ + _example( + boxes=[{"x": 190, "y": 10, "width": 40, "height": 40, "class": "cat"}] + ) + ], + task_type="object-detection", + labels=["cat"], + ) + + # when + loaded = load_reference_examples(examples) + + # then + assert loaded[0].image.shape == (100, 200, 3) + assert loaded[0].boxes == (("cat", 170.0, 0.0, 200.0, 30.0),) + + +def test_load_reference_examples_rejects_a_box_outside_its_image() -> None: + # given + examples = validate_reference_examples( + [ + _example( + boxes=[{"x": 500, "y": 50, "width": 40, "height": 20, "class": "cat"}] + ) + ], + task_type="object-detection", + labels=["cat"], + ) + + # when / then + with pytest.raises(ValueError, match="outside its 200x100 image"): + load_reference_examples(examples) + + +def test_render_example_crops_keeps_the_largest_boxes_in_label_order() -> None: + # given + boxes = [ + {"x": 30, "y": 30, "width": 10, "height": 10, "class": "cat"}, + {"x": 100, "y": 50, "width": 60, "height": 40, "class": "cat"}, + {"x": 150, "y": 50, "width": 30, "height": 30, "class": "cat"}, + {"x": 50, "y": 70, "width": 20, "height": 20, "class": "cat"}, + {"x": 180, "y": 20, "width": 20, "height": 10, "class": "dog"}, + ] + examples = load_reference_examples( + validate_reference_examples( + [_example(boxes=boxes)], + task_type="object-detection", + labels=["dog", "cat", "bird"], + ) + ) + + # when + groups = render_example_crops(examples, labels=["dog", "cat", "bird"]) + + # then + assert [label for label, _ in groups] == ["dog", "cat"] + cat_crops = [_decode(crop) for crop in groups[1][1]] + assert len(cat_crops) == 3 + # The 60x40 box with its margin is the widest crop, upscaled to the minimum edge. + assert cat_crops[0].shape[1] == CROP_MIN_EDGE + for crop in cat_crops: + assert CROP_MIN_EDGE <= max(crop.shape[:2]) <= CROP_MAX_EDGE + + +def test_render_boxed_examples_downscales_and_draws_in_the_label_color() -> None: + # given + examples = load_reference_examples( + validate_reference_examples( + [ + _example( + width=2048, + height=1024, + boxes=[ + { + "x": 1024, + "y": 512, + "width": 800, + "height": 400, + "class": "dog", + } + ], + ) + ], + task_type="object-detection", + labels=["cat", "dog"], + ) + ) + + # when + rendered = render_boxed_examples(examples, labels=["cat", "dog"]) + + # then + image = _decode(rendered[0]) + assert image.shape == (512, 1024, 3) + # The second label takes the second color, green (60, 180, 75) in RGB; the + # right edge of the box sits at x = 1424 / 2 in the downscaled image. + blue, green, red = (int(v) for v in image[256, 711]) + assert green > 150 and red < 110 and blue < 110 + assert min(int(v) for v in image[256, 512]) >= 250 + + +def test_gemini_reference_parts_number_the_boxed_images() -> None: + # when + parts = gemini_reference_parts_for([_example(), _example()], labels=["cat"]) + + # then + assert parts[0] == {"text": REFERENCE_BOXED_INTRO} + assert parts[1] == {"text": "Reference image 1:"} + assert parts[2]["inline_data"]["mime_type"] == "image/jpeg" + assert parts[3] == {"text": "Reference image 2:"} + assert len(parts) == 5 + + +def test_openai_reference_parts_group_crops_by_label() -> None: + # given + boxes = [ + {"x": 50, "y": 50, "width": 40, "height": 20, "class": "cat"}, + {"x": 150, "y": 50, "width": 40, "height": 20, "class": "dog"}, + ] + + # when + parts = openai_reference_parts_for([_example(boxes=boxes)], labels=["cat", "dog"]) + + # then + assert parts[0] == {"type": "input_text", "text": REFERENCE_CROPS_INTRO} + assert parts[1] == {"type": "input_text", "text": 'Examples of "cat":'} + assert parts[2]["type"] == "input_image" + assert parts[2]["image_url"].startswith("data:image/jpeg;base64,") + assert parts[3] == {"type": "input_text", "text": 'Examples of "dog":'} + assert len(parts) == 5 + + +def test_reference_parts_are_built_once_for_the_same_examples() -> None: + # given + examples = [_example()] + real_load_image = vlm_reference_examples.load_image + + # when + with patch.object( + vlm_reference_examples, "load_image", side_effect=real_load_image + ) as load_image: + first = gemini_reference_parts_for(examples, labels=["cat"]) + second = gemini_reference_parts_for(examples, labels=["cat"]) + + # then + assert first == second + assert load_image.call_count == 1 + + +def test_with_reference_parts_puts_references_first_and_points_at_the_target() -> None: + # given + image_part = {"inline_data": {"mime_type": "image/jpeg", "data": "target"}} + parts = [ + image_part, + {"text": "Detect all objects in this image. Only use these labels: cat"}, + ] + reference_parts = [{"text": "references"}] + + # when + result = with_reference_parts( + parts, + reference_parts=reference_parts, + target_intro={"text": TARGET_IMAGE_INTRO}, + is_image_part=lambda part: "inline_data" in part, + ) + + # then + assert result == [ + {"text": "references"}, + {"text": TARGET_IMAGE_INTRO}, + image_part, + { + "text": "Detect all objects in the target image. Only use these labels: cat" + + TARGET_ONLY_RULE + }, + ] + assert parts[1]["text"].startswith("Detect all objects in this image.") + + +def test_with_reference_parts_keeps_text_before_the_image_in_place() -> None: + # given + parts = [ + {"type": "input_text", "text": "List of all classes: cat"}, + {"type": "input_image", "image_url": "data:image/png;base64,target"}, + ] + + # when + result = with_reference_parts( + parts, + reference_parts=[], + target_intro={"type": "input_text", "text": TARGET_IMAGE_INTRO}, + is_image_part=lambda part: part.get("type") == "input_image", + ) + + # then + assert result == [ + {"type": "input_text", "text": "List of all classes: cat" + TARGET_ONLY_RULE}, + {"type": "input_text", "text": TARGET_IMAGE_INTRO}, + parts[1], + ] diff --git a/workflows/tests/unit_tests/core_steps/models/foundation/test_google_gemini_v6.py b/workflows/tests/unit_tests/core_steps/models/foundation/test_google_gemini_v6.py index dccc948a3d..2daefbb15a 100644 --- a/workflows/tests/unit_tests/core_steps/models/foundation/test_google_gemini_v6.py +++ b/workflows/tests/unit_tests/core_steps/models/foundation/test_google_gemini_v6.py @@ -6,10 +6,12 @@ the block from Gemini's native ``yxyx_0_1000`` boxes. """ +import base64 import json from typing import List, Optional from unittest.mock import patch +import cv2 import numpy as np import pytest from roboflow_workflows.core_steps.models.foundation.google_gemini.v6 import ( @@ -489,3 +491,136 @@ def test_segmentation_runtime_rejects_resolved_other_model() -> None: image=_build_image(), classes=["ring"], ) + + +def _reference_example(box_class: str = "cat") -> dict: + image = np.full((100, 200, 3), 255, dtype=np.uint8) + _, encoded = cv2.imencode(".png", image) + return { + "image": { + "type": "base64", + "value": base64.b64encode(encoded.tobytes()).decode("ascii"), + }, + "boxes": [{"x": 50, "y": 50, "width": 40, "height": 20, "class": box_class}], + } + + +def _run_detection_with_mock(reference_examples: Optional[list]) -> tuple: + block = GoogleGeminiBlockV6(api_key="rf-key") + with patch(EXECUTE_REQUESTS_SEAM) as mock_execute: + mock_execute.return_value = [(DETECTION_OUTPUT, 11, 3)] + results = block.run( + images=Batch(content=[_build_image()], indices=[(0,)]), + task_type="object-detection", + prompt=None, + output_structure=None, + classes=["cat", "dog"], + model_version="gemini-4-argon", + max_tokens=16384, + temperature=None, + thinking_level="low", + google_code_execution=None, + max_concurrent_requests=None, + api_key="rf_key:account", + reference_examples=reference_examples, + ) + parts = mock_execute.call_args[1]["gemini_prompts"][0]["contents"]["parts"] + return results[0], parts + + +@pytest.mark.parametrize( + "reference_examples", [[_reference_example()], "$inputs.reference_examples"] +) +def test_manifest_accepts_reference_examples(reference_examples) -> None: + # when + result = BlockManifest.model_validate( + { + "type": "roboflow_core/google_gemini@v6", + "name": "gemini", + "images": "$inputs.image", + "task_type": "object-detection", + "classes": ["cat", "dog"], + "reference_examples": reference_examples, + } + ) + + # then + assert result.reference_examples == reference_examples + + +def test_manifest_rejects_reference_examples_for_other_tasks() -> None: + with pytest.raises(ValueError, match="only supported for the object-detection"): + BlockManifest.model_validate( + { + "type": "roboflow_core/google_gemini@v6", + "name": "gemini", + "images": "$inputs.image", + "task_type": "classification", + "classes": ["cat", "dog"], + "reference_examples": [_reference_example()], + } + ) + + +def test_manifest_rejects_reference_examples_with_unknown_labels() -> None: + with pytest.raises(ValueError, match=r"does not have: \['bird'\]"): + BlockManifest.model_validate( + { + "type": "roboflow_core/google_gemini@v6", + "name": "gemini", + "images": "$inputs.image", + "task_type": "object-detection", + "classes": ["cat", "dog"], + "reference_examples": [_reference_example(box_class="bird")], + } + ) + + +def test_run_sends_boxed_reference_images_before_the_target() -> None: + # when + result, parts = _run_detection_with_mock([_reference_example()]) + + # then + assert parts[0]["text"].startswith("Reference examples: other images") + assert parts[1] == {"text": "Reference image 1:"} + assert parts[2]["inline_data"]["mime_type"] == "image/jpeg" + assert parts[3] == {"text": "Target image:"} + assert parts[4]["inline_data"]["mime_type"] == "image/jpeg" + assert parts[5]["text"].startswith("Detect all objects in the target image. ") + assert parts[5]["text"].endswith( + "The reference images are not part of the task: return detections for " + "the target image only." + ) + assert len(parts) == 6 + assert detection_boxes(result["predictions"])[0] == EXPECTED_XYXY + + +@pytest.mark.parametrize("reference_examples", [None, []]) +def test_run_without_reference_examples_keeps_the_request(reference_examples) -> None: + # when + _, parts = _run_detection_with_mock(reference_examples) + + # then + assert len(parts) == 2 + assert "inline_data" in parts[0] + assert parts[1]["text"].startswith("Detect all objects in this image. ") + + +def test_run_rejects_resolved_reference_examples_for_other_tasks() -> None: + with pytest.raises(ValueError, match="only supported for the object-detection"): + block = GoogleGeminiBlockV6(api_key="rf-key") + with patch(EXECUTE_REQUESTS_SEAM): + block.run( + images=Batch(content=[_build_image()], indices=[(0,)]), + task_type="classification", + prompt=None, + output_structure=None, + classes=["cat", "dog"], + model_version="gemini-4-argon", + max_tokens=None, + temperature=None, + thinking_level=None, + google_code_execution=None, + max_concurrent_requests=None, + reference_examples=[_reference_example()], + ) diff --git a/workflows/tests/unit_tests/core_steps/models/foundation/test_openai_v7.py b/workflows/tests/unit_tests/core_steps/models/foundation/test_openai_v7.py index 4817184122..c1625dc769 100644 --- a/workflows/tests/unit_tests/core_steps/models/foundation/test_openai_v7.py +++ b/workflows/tests/unit_tests/core_steps/models/foundation/test_openai_v7.py @@ -6,10 +6,12 @@ each of the three per-model detection prompt styles. """ +import base64 import json from typing import List, Optional from unittest.mock import MagicMock, Mock, patch +import cv2 import numpy as np import pytest from pydantic import ValidationError @@ -770,3 +772,155 @@ def test_run_hands_the_service_tier_to_every_request() -> None: ) assert mock_execute.call_args[1]["service_tier"] == "ultrafast" + + +def _reference_example(box_class: str = "cat") -> dict: + image = np.full((100, 200, 3), 255, dtype=np.uint8) + _, encoded = cv2.imencode(".png", image) + return { + "image": { + "type": "base64", + "value": base64.b64encode(encoded.tobytes()).decode("ascii"), + }, + "boxes": [{"x": 50, "y": 50, "width": 40, "height": 20, "class": box_class}], + } + + +def _run_detection_with_mock( + reference_examples: Optional[list], + model_version: str = "gpt-6.1-sol", + output_classes: Optional[List[str]] = None, +) -> tuple: + block = OpenAIBlockV7(api_key="rf-key") + raw_output = ( + NORMALIZED_LEGACY_OUTPUT + if model_version == "gpt-5.1" + else STRUCTURED_ABSOLUTE_OUTPUT + ) + with patch(EXECUTE_REQUESTS_SEAM) as mock_execute: + mock_execute.return_value = [(raw_output, 11, 3)] + results = block.run( + images=Batch( + content=[ + _build_image( + width=DETECTION_IMAGE_WIDTH, height=DETECTION_IMAGE_HEIGHT + ) + ], + indices=[(0,)], + ), + task_type="object-detection", + prompt=None, + output_structure=None, + classes=["small household feline", "domestic canine"], + output_classes=output_classes, + model_version=model_version, + reasoning_effort=None, + image_detail="auto", + max_tokens=None, + temperature=None, + max_concurrent_requests=None, + api_key="rf_key:account", + reference_examples=reference_examples, + ) + prompt = mock_execute.call_args[1]["openai_prompts"][0] + return results[0], prompt + + +@pytest.mark.parametrize( + "reference_examples", [[_reference_example()], "$inputs.reference_examples"] +) +def test_manifest_accepts_reference_examples(reference_examples) -> None: + # when + result = BlockManifest.model_validate( + { + "type": "roboflow_core/open_ai@v7", + "name": "open_ai", + "images": "$inputs.image", + "task_type": "object-detection", + "classes": ["small household feline", "domestic canine"], + "output_classes": ["cat", "dog"], + "reference_examples": reference_examples, + } + ) + + # then + assert result.reference_examples == reference_examples + + +def test_manifest_checks_reference_labels_against_output_classes() -> None: + with pytest.raises(ValueError, match="small household feline"): + BlockManifest.model_validate( + { + "type": "roboflow_core/open_ai@v7", + "name": "open_ai", + "images": "$inputs.image", + "task_type": "object-detection", + "classes": ["small household feline", "domestic canine"], + "output_classes": ["cat", "dog"], + "reference_examples": [ + _reference_example(box_class="small household feline") + ], + } + ) + + +def test_manifest_rejects_reference_examples_for_segmentation() -> None: + with pytest.raises(ValueError, match="only supported for the object-detection"): + BlockManifest.model_validate( + { + "type": "roboflow_core/open_ai@v7", + "name": "open_ai", + "images": "$inputs.image", + "task_type": "instance-segmentation", + "classes": ["cat"], + "reference_examples": [_reference_example()], + } + ) + + +def test_run_sends_reference_crops_before_the_target() -> None: + # when + result, prompt = _run_detection_with_mock( + [_reference_example()], output_classes=["cat", "dog"] + ) + + # then + content = prompt["input"][0]["content"] + assert content[0]["text"].startswith("Reference examples: crops of labeled") + assert content[1] == {"type": "input_text", "text": 'Examples of "cat":'} + assert content[2]["image_url"].startswith("data:image/jpeg;base64,") + assert content[3] == {"type": "input_text", "text": "Target image:"} + assert content[4]["image_url"].startswith("data:image/png;base64,") + assert content[5]["text"].startswith("Detect all objects in the target image. ") + assert content[5]["text"].endswith("return detections for the target image only.") + assert len(content) == 6 + assert prompt["text"]["format"]["name"] == "detections" + assert detection_boxes(result["predictions"])[0] == EXPECTED_XYXY + + +def test_run_adds_reference_crops_to_the_legacy_prompt_style() -> None: + # when + _, prompt = _run_detection_with_mock( + [_reference_example(box_class="small household feline")], + model_version="gpt-5.1", + ) + + # then + content = prompt["input"][0]["content"] + assert content[-3]["text"].startswith("List of all classes to be recognised") + assert content[-3]["text"].endswith("return detections for the target image only.") + assert content[-2] == {"type": "input_text", "text": "Target image:"} + assert content[-1]["type"] == "input_image" + assert "You act as object-detection model" in prompt["instructions"] + + +@pytest.mark.parametrize("reference_examples", [None, []]) +def test_run_without_reference_examples_keeps_the_request(reference_examples) -> None: + # when + _, prompt = _run_detection_with_mock(reference_examples) + + # then + content = prompt["input"][0]["content"] + assert len(content) == 2 + assert content[0]["type"] == "input_image" + assert content[1]["text"].startswith("Detect all objects in this image. ")