Structured output returns 4xx error

#### Environment details

- Programming language: python
- OS: ubuntu / docker
- Language runtime version: 3.11.14
- Package version: 1.31.0

#### Problem:

We try to use the structured output mode of gemini-2.5-flash like [here](https://ai.google.dev/gemini-api/docs/structured-output?example=recipe) described in the docs. We want to use the model for visual scene analysis, object detection, object segmentation. For this we need a structured json output to handle the data.

Data Models for Structured Output:

```
from pydantic import BaseModel, Field, RootModel
from text2action import BoundingBox, Point2d, SegmentationMaskAndBoundingBox

class SceneDescriptionResult(BaseModel):
description: str = Field(..., description="A detailed description of the scene in the image.")

class GeminiBoundingBox(BaseModel):
"""Bounding box of a object with coordinates normalized between 0 and 1000."""

xmin: int = Field(..., description="The minimum x coordinate of the bounding box, normalized between 0 and 1000.")
xmax: int = Field(..., description="The maximum x coordinate of the bounding box, normalized between 0 and 1000.")
ymin: int = Field(..., description="The minimum y coordinate of the bounding box, normalized between 0 and 1000.")
ymax: int = Field(..., description="The maximum y coordinate of the bounding box, normalized between 0 and 1000.")

class GeminiPoint2d(BaseModel):
"""A 2D point with coordinates normalized between 0 and 1000."""

point: list[int] = Field(..., description="A 2D point in [y, x] format normalized between 0 and 1000.")
label: str = Field(..., description="The label associated with the point.")

class DetectedObject(BaseModel):
"""A detected object in an image."""

label: str = Field(..., description="Class name of the detected object, not a description.")
box_2d: GeminiBoundingBox = Field(..., description="2D bounding box of the detected object.")

class ObjectDetectionResult(RootModel[list[DetectedObject]]):
"""The result of an object detection model, containing a list of detected objects."""

root: list[DetectedObject] = Field(..., description="A list of detected objects in the image.")

class SegmentedObject(BaseModel):
"""A segmented object in an image."""

label: str = Field(..., description="Class name of the detected object, not a description.")
box_2d: GeminiBoundingBox = Field(..., description="2D bounding box of the detected object.")
mask: str = Field(..., description="Segmentation mask.")

class ObjectSegementationResult(RootModel[list[SegmentedObject]]):
"""The result of a segmentation model, containing a list of segmented objects."""

root: list[SegmentedObject] = Field(..., description="A list of segmented objects in the image.")

class TrajectoryResult(RootModel[list[GeminiPoint2d]]):
"""The result of a point detection model, containing a list of 2D points."""

root: list[GeminiPoint2d] = Field(..., description="A list of detected 2D points in the image.")
```

Code for model:

```
import json
import re
from abc import ABCMeta, abstractmethod
from typing import Any, Dict, List, Type, TypeVar, Union

from google.genai import Client
from google.genai.types import (
AutomaticFunctionCallingConfig,
GenerateContentConfig,
GenerateContentResponse,
HttpOptionsDict,
ThinkingConfig,
)
from google.oauth2.service_account import Credentials
from PIL import Image
from pydantic import BaseModel
from text2action import (
Base64Image,
BoundingBox,
CameraFrame,
Point2d,
SegmentationMaskAndBoundingBox,
get_logger,
get_secret,
)

from app.utils import remove_base64_prefix, scale_and_clamp, to_bounding_box
from app.vision.gemini_schema import (
ObjectDetectionResult,
SceneDescriptionResult,
ObjectSegementationResult,
TrajectoryResult,
)
from app.vision.image_processing import get_pre_post_processor

logger = get_logger("vision")

TSchema = TypeVar("TSchema", bound=BaseModel)

class GenerativeAiClient(metaclass=ABCMeta):

@abstractmethod
async def send_image_request(self, image: Image.Image, prompt: str, schema: Type[TSchema]) -> TSchema:
"""Run a query on a multi modal model using an image and a prompt.

Args:
image (Image.Image): The image to send to the model.
prompt (str): The prompt we want to query the model with.

Returns:
Union[List[Dict[str, Any]], Dict[str, Any]]: The data type that is requested in the prompt template.
"""

class Gemini25Flash(GenerativeAiClient):

def __init__(self) -> None:
self.async_client = Client(
vertexai=True,
location=get_secret("text-to-action/dev/GOOGLE_VERTEX_LOCATION", "eu-central-1")["GOOGLE_VERTEX_LOCATION"],
project=get_secret("text-to-action/dev/GOOGLE_VERTEX_PROJECT_ID", "eu-central-1")[
"GOOGLE_VERTEX_PROJECT_ID"
],
credentials=self.load_credentials(),
).aio
self.model = "gemini-2.5-flash"
self.NORMALIZATION_FACTOR = 1000.0

def load_credentials(self) -> Credentials:
"""Get the credentials from secrets manager.

Returns:
Credentials: The service account credentials.
"""
secret = get_secret("text-to-action/dev/GOOGLE_APPLICATION_CREDENTIALS", "eu-central-1")
credentials = json.loads(secret["GOOGLE_APPLICATION_CREDENTIALS"])

# Define the required scope for Vertex AI
scopes = ["https://www.googleapis.com/auth/cloud-platform"]

return Credentials.from_service_account_info(info=credentials, scopes=scopes)

def _clean_response_text(self, text: str) -> str:
"""Clean the text from unwanted unicode characters.
- Gemini returns sometimes characters in a json string that are not allowed and
make json parsing fail.

Args:
text (str): The text to clean.

Returns:
str: Cleaned text.
"""

# Remove or replace invalid control characters except for \n, \r, \t
# Control characters are in Unicode range 0-31
return re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", "", text)

async def query_gemini(
self,
image: Image.Image,
prompt: str,
schema: Type[TSchema],
) -> GenerateContentResponse:
"""Run a query on gemini flash 2.5 using a image and a prompt.

Args:
image (Image.Image): The image to send to gemini.
prompt (str): The prompt we want to query gemini with.

Returns:
GenerateContentResponse: The response object holding all necessary data.
"""
config = GenerateContentConfig(
response_mime_type="application/json",
# response_json_schema=schema.model_json_schema(),
response_schema=schema,
thinking_config=ThinkingConfig(thinking_budget=0),
temperature=0.0,
automatic_function_calling=AutomaticFunctionCallingConfig(disable=True),
http_options=HttpOptionsDict(timeout=10_000),
)

logger.info("Sending generate_content request.")
response = await self.async_client.models.generate_content(
model=self.model,
contents=[image, prompt],
config=config,
)
logger.info("Received generate_content response.")

response.parsed = schema.model_validate_json(response.text)
# if response.parsed is None:
# logger.warning("Gemini response did not parse into schema %s", schema.__name__)
# response.parsed = schema.model_validate_json(response.text)

return response

def parse_response(self, response: GenerateContentResponse) -> Union[List[Dict[str, Any]], Dict[str, Any]]:
"""Try to parse the response to a json data type.
We encountered multiple problems.
- Ill formatted base64 encoding
- Missing junks in the text
- Ill formatted json strings

Args:
response (GenerateContentResponse): The response from Gemini.

Raises:
RuntimeError: If we cannot parse the responses text.

Returns:
Union[List[Dict[str, Any]], Dict[str, Any]]: The data type that is requested in the prompt template.
"""
data = []
if not response.text is None:
try:
logger.info("Parsing response")
data = json.loads(response.text)
except json.decoder.JSONDecodeError as e:
logger.warning("JSON parsing failed: %s. Attempting to clean response text.", e)
cleaned_text = self._clean_response_text(response.text)
try:
data = json.loads(cleaned_text)
except json.decoder.JSONDecodeError as e2:
logger.error(f"JSON parsing failed after cleaning: {e2}")
# Log a snippet of the problematic response for debugging
snippet_start = max(0, e2.pos - 50)
snippet_end = e2.pos + 50
logger.error(f"Response text snippet around error:\n{cleaned_text[snippet_start:snippet_end]}\n")
raise RuntimeError(f"Gemini model response JSON decode error after cleaning: {e2}") from e2
return data

def denormalize_coordinates(
self, data: Union[List[Dict], Dict], width: int, height: int
) -> Union[List[Dict], Dict]:
"""Recursively find and convert normalized coordinates in the response data to absolute pixel coordinates.
This method modifies the data in-place.

Args:
data (Union[List[Dict], Dict]): The data from the gemini model.
width (int): Original image width.
height (int): Original image height.

Returns:
Union[List[Dict], Dict]: Data converted to absolute coordinates in pixel space.
"""
if isinstance(data, list):
for item in data:
self.denormalize_coordinates(item, width, height)

elif isinstance(data, dict):
# Denormalize bounding boxes
if "box_2d" in data and isinstance(data["box_2d"], list) and len(data["box_2d"]) == 4:
y_min_norm, x_min_norm, y_max_norm, x_max_norm = data["box_2d"]
data["box_2d"] = [
scale_and_clamp(y_min_norm, height, self.NORMALIZATION_FACTOR),
scale_and_clamp(x_min_norm, width, self.NORMALIZATION_FACTOR),
scale_and_clamp(y_max_norm, height, self.NORMALIZATION_FACTOR),
scale_and_clamp(x_max_norm, width, self.NORMALIZATION_FACTOR),
]

# Denormalize points
if "point" in data and isinstance(data["point"], list) and len(data["point"]) == 2:
y_norm, x_norm = data["point"]
data["point"] = [
scale_and_clamp(y_norm, height, self.NORMALIZATION_FACTOR),
scale_and_clamp(x_norm, width, self.NORMALIZATION_FACTOR),
]

return data

async def send_image_request(self, image: Image.Image, prompt: str, schema: Type[TSchema]) -> TSchema:
width, height = image.size
response = await self.query_gemini(image=image, prompt=prompt, schema=schema)
# data = self.parse_response(response)
# return self.denormalize_coordinates(data, width=width, height=height)
assert isinstance(response.parsed, schema)
return response.parsed

class ObjectDetection:
def __init__(self, client: GenerativeAiClient):
self.client = client
self.pre_and_postprocessor = get_pre_post_processor()

def parse_bounding_boxes(self, bounding_boxes: ObjectDetectionResult, width: int, height: int) -> List[BoundingBox]:
"""Convert the dictionary of normalized bounding boxes to a Pydantic model of absolute image coordinates.

Args:
bounding_boxes (List[Dict[str, Any]]): Normalized bounding boxes of type
{
"box_2d": [y_min, x_min, y_max, x_max],
"label": str
}
Returns:
List[BoundingBox]: List of not degenerated and not normalized bounding boxes in absolute pixel space.
"""
predictions = []
for detected_object in bounding_boxes.root:
bbox = to_bounding_box(detected_object, width=width, height=height)
if not bbox.is_degenerated():
predictions.append(self.pre_and_postprocessor.postprocess_bounding_box(bbox))
return predictions

async def detect_objects(self, camera_frame: CameraFrame, queries: List[str]) -> List[BoundingBox]:
"""
Detect specified objects in an image and return absolute bounding boxes.

Args:
image_base64 (str): Base64-encoded input image.
queries (List[str]): List of object user queries for object detection.

Returns:
List[BoundingBox]: Detection results in the format
"""
image = self.pre_and_postprocessor.preprocess_image(camera_frame.color_image.to_pillow_image())
prompt = (
f"Detect objects in the image that match the following user queries: {queries}.\n\n"
"For each detected object, output its true class label (e.g., 'elephant', 'cat', 'car').\n\n"
"Requirements:\n"
"- IMPORTANT: 'label' must be the object's actual class name (e.g., 'car', 'dog', 'elephant'), "
"never a descriptive phrase from the query (e.g., not 'biggest animal', 'left object').\n"
"- Descriptive attributes (biggest, leftmost, red, etc.) are used only to choose which object to detect, "
"but must not appear in the 'label'.\n"
"- Bounding box coordinates must be strictly normalized to the range [0, 1000].\n"
"- The output must be valid JSON with no extra text or commentary.\n"
)
bounding_box_response = await self.client.send_image_request(
image=image, prompt=prompt, schema=ObjectDetectionResult
)
result = self.parse_bounding_boxes(bounding_box_response, width=image.width, height=image.height)
return result

class SemanticSegmentation:
def __init__(self, client: GenerativeAiClient):
self.client = client
self.pre_and_postprocessor = get_pre_post_processor()

def has_image_prefix(self, png_str: str) -> bool:
"""Check if the mask has the correct prefix.
- Sometimes gemini returns a ill formatted base64 string.

Args:
prediction (dict): The prediction dictionary.

Returns:
bool: Wether the prefix is missing.
"""
if not png_str.startswith("data:image/png;base64,"):
logger.warning("Mask is missing mime-type prefix")
return False
else:
return True

def is_ill_formatted_base64_string(self, png_str: str) -> bool:
"""Check if base64 data length is multiple of 4 and non-empty (correct padding).
- Gemini sometimes returns incomplete chinks. We identify this wether the string si a multiple of 4.

Args:
prediction (dict): The prediction dictionary.

Returns:
bool: Wether the base64 string is ill formatted.
"""
# Check if base64 data length is multiple of 4 and non-empty (correct padding)
png_str_without_prefix = png_str.replace("data:image/png;base64,", "")
if not png_str_without_prefix or len(png_str_without_prefix) % 4 != 0:
logger.warning("Ill formatted base64 string for mask")
return True
else:
return False

def can_be_parsed(self, png_str: str) -> bool:
"""Check if the base64 string can be encoded as an png image.

Args:
prediction (dict): The prediction dictionary.

Returns:
bool: Wether the base64 string can be encoded as png image.
"""
png_str_without_prefix = remove_base64_prefix(png_str)
try:
Base64Image(image_data=png_str_without_prefix, data_type="uint8").to_pillow_image()
return True
except:
logger.warning("Parsing of segmentation mask failed due to invalid base64.")
return False

def is_valid_segmentation_mask(self, mask: str) -> bool:
"""Sanitize the segmentation mask. Eg if the mask validates the image dimensions.

Args:
prediction (dict[any, any]): Prediction dictionary

Returns:
bool: If it is a valid prediction
"""
return (
self.has_image_prefix(mask) and not self.is_ill_formatted_base64_string(mask) and self.can_be_parsed(mask)
)

def create_binary_overlay(
self, mask: str, bounding_box: BoundingBox, width: int, height: int, threshold: int = 128
) -> Base64Image:
"""
Create a binary overlay image from a segmentation mask.

Args:
mask (str): Base64-encoded segmentation mask image.
bounding_box (BoundingBox): Absolute bounding box in image coordinate system.
width (int): Target overlay image width in pixels.
height (int): Target overlay image height in pixels.
threshold (int): The threshold to create a binary image from.

Returns:
Base64Image: Base64-encoded PNG overlay with white region for the mask and black background.
"""
mask_img = Base64Image(image_data=remove_base64_prefix(mask), data_type="uint8").to_pillow_image()
mask_resized = mask_img.resize(
(bounding_box.x_max - bounding_box.x_min, bounding_box.y_max - bounding_box.y_min),
Image.Resampling.BILINEAR,
)
mask_binary = mask_resized.point(lambda p: 255 if p > threshold else 0)
binary_overlay = Image.new("L", (width, height), 0)
binary_overlay.paste(
255, (bounding_box.x_min, bounding_box.y_min, bounding_box.x_max, bounding_box.y_max), mask_binary
)
binary_overlay = self.pre_and_postprocessor.postprocess_image(binary_overlay)
return Base64Image.from_pillow_image(binary_overlay)

def parse_segmentation_masks(
self, segmentation_masks: ObjectSegementationResult, width: int, height: int
) -> List[SegmentationMaskAndBoundingBox]:
"""Parse the segmentation mask in a base64 representative data type.

Args:
segmentation_masks (List[Dict[str, Any]]): The segmentation masks from model.
width (int): Original height of the image.
height (int): Original Width of the image.

Returns:
List[SegmentationMaskAndBoundingBox]: The parsed data type.
"""
predictions = []
for segmented_object in segmentation_masks.root:
bbox = to_bounding_box(segmented_object, width, height)
if not bbox.is_degenerated() and self.is_valid_segmentation_mask(segmented_object.mask):
mask = self.create_binary_overlay(segmented_object.mask, bounding_box=bbox, width=width, height=height)
bbox = self.pre_and_postprocessor.postprocess_bounding_box(bbox)
predictions.append(SegmentationMaskAndBoundingBox(mask=mask, **bbox.model_dump()))
return predictions

async def segment_objects(
self, camera_frame: CameraFrame, queries: List[str]
) -> List[SegmentationMaskAndBoundingBox]:
"""
Detect specified objects in an image and return absolute bounding boxes and the corresponding segmentation masks.

Args:
image_base64 (str): Base64-encoded input image.
queries (List[str]): List of object user queries for object detection.

Returns:
List[SegmentationMaskAndBoundingBox]: Segmentation results.
"""
image = self.pre_and_postprocessor.preprocess_image(camera_frame.color_image.to_pillow_image())
width, height = image.size
prompt = (
f"Segment objects in the image that match the following user queries: {queries}.\n"
"For each segmented object, output its true class label (e.g., 'elephant', 'cat', 'car').\n\n"
"Requirements:\n"
"- Only segment objects that correspond to the user queries.\n"
"- 'label' must be the object's actual class name (e.g., 'elephant'), never a descriptive phrase from the query "
"(e.g., not 'biggest animal', 'left one', or 'red object').\n"
"- Segmentation mask must be only as big as the bounding box.\n"
"- Descriptive attributes (biggest, smallest, leftmost, etc.) must guide which object is segmented, "
"but must not appear in the 'label'.\n"
"- Bounding box coordinates must be normalized to the range [0, 1000].\n"
"- The output must be valid JSON with no extra text or commentary.\n"
)
segmentation_result = await self.client.send_image_request(image=image, prompt=prompt, schema=ObjectSegementationResult)
result = self.parse_segmentation_masks(segmentation_result, width=width, height=height)
return result

class SceneAnalysis:
def __init__(self, client: GenerativeAiClient):
self.client = client
self.pre_and_postprocessor = get_pre_post_processor()

async def describe_scene(self, camera_frame: CameraFrame, query: str) -> str:
"""
Perform scene analysis on an image.

Args:
image_base64 (str): Base64-encoded input image.
queries (List[str]): List of object user queries for object detection.

Returns:
str: Scene analysis result in the format
"""
image = self.pre_and_postprocessor.preprocess_image(camera_frame.color_image.to_pillow_image())
prompt = (
f"Analyze the scene in the image and answer the following user query: {query}.\n\n"
"Requirements:\n"
"- The output must be valid JSON with no extra text or commentary.\n"
)
description_result = await self.client.send_image_request(
image=image, prompt=prompt, schema=SceneDescriptionResult
)
return description_result.description

class TrajectoryPlanning:
def __init__(self, client: GenerativeAiClient):
self.client = client
self.pre_and_postprocessor = get_pre_post_processor()

def parse_point(self, point: Dict[str, Any]) -> Point2d:
"""Convert the point to absolute image coordinates.

Args:
point (Dict[str, Any]): Normalized Point.

Returns:
Point2d: Point in absolute image coordinates
"""
y, x = point["point"]
label = point["label"]
point = Point2d(x=x, y=y, label=label)
return self.pre_and_postprocessor.postprocess_point(point)

def parse_points(self, points: TrajectoryResult, width: int, height: int) -> List[Point2d]:
"""Convert the point to absolute image coordinates.

Args:
points (List[Dict[str, Any]]): Normalized Points.

Returns:
List[Point2d]: Points to absolute image coordinates
"""
trajectory = []
for point in points.root:
parsed_point = point.to_point2d(width, height)
trajectory.append(self.pre_and_postprocessor.postprocess_point(parsed_point))
return trajectory

async def plan_trajectory(self, camera_frame: CameraFrame, query: str) -> List[Point2d]:
"""Perform the planning of a trajectory of single points or a point to a spcific object

Args:
image_base64 (str): Base64-encoded input image.
query (str): A user query to perform the trajectory

Returns:
TrajectoryModelListResponse: List of points with the label in the trajectory
"""
image = self.pre_and_postprocessor.preprocess_image(camera_frame.color_image.to_pillow_image())
prompt = (
f"Generate a trajectory of single points or point to a specific object based on the following user query: {query}.\n\n"
"Output Format (strictly JSON, no extra text, no markdown, no extra marks):\n"
"The answer should follow the json format: [{'point': , 'label': }, ...]."
"The points are in [y, x] format normalized to 0-1000"
)
points_result = await self.client.send_image_request(image=image, prompt=prompt, schema=TrajectoryResult)
result = self.parse_points(points_result, width=image.width, height=image.height)
return result
```

We try here in the config:

```
config = GenerateContentConfig(
response_mime_type="application/json",
response_schema=schema,
thinking_config=ThinkingConfig(thinking_budget=0),
temperature=0.0,
automatic_function_calling=AutomaticFunctionCallingConfig(disable=True),
http_options=HttpOptionsDict(timeout=10_000),
)
```

to pass the model schema. Problem is that we receive an 400 error (`400 INVALID_ARGUMENT. {'error': {'code': 400, 'message': 'Request contains an invalid argument.', 'status': 'INVALID_ARGUMENT'}}`) with the information that parameters are not valid (but not which one). For object detection it works fine but for object segmentation not.