From baafa40cdf96a76755e07dc4f849d293846e1091 Mon Sep 17 00:00:00 2001 From: Osama Hussein Date: Thu, 24 Sep 2026 01:06:32 +0300 Subject: [PATCH] fix(gemini/veo): move media inputs into instances[0] and validate params lastFrame, referenceImages and video were still sent under `parameters`, so Veo 3.1 rejected interpolation, reference-image and extension requests ("lastFrame isn't supported by this model"). The Gemini API reads all media inputs from instances[0]; parameters only carries generation config. - Route lastFrame, referenceImages and video into instances[0], base64 encoding file-like inputs and keeping their MIME type - Reject combinations the Gemini API documents as invalid with a 400 BadRequestError before the request reaches Google - Move the three fields from GeminiVideoGenerationParameters to GeminiVideoGenerationInstance --- litellm/llms/gemini/videos/transformation.py | 158 ++++++++- litellm/types/llms/gemini.py | 59 ++-- .../videos/test_gemini_video_media_inputs.py | 334 ++++++++++++++++++ 3 files changed, 510 insertions(+), 41 deletions(-) create mode 100644 tests/unit/llms/gemini/videos/test_gemini_video_media_inputs.py diff --git a/litellm/llms/gemini/videos/transformation.py b/litellm/llms/gemini/videos/transformation.py index 8a3894cf7fb..90eb466d225 100644 --- a/litellm/llms/gemini/videos/transformation.py +++ b/litellm/llms/gemini/videos/transformation.py @@ -1,9 +1,12 @@ import base64 +import mimetypes from collections.abc import Mapping -from typing import TYPE_CHECKING, Any, Final +from types import MappingProxyType +from typing import TYPE_CHECKING, Any, Final, Protocol, runtime_checkable import httpx from httpx._types import RequestFiles +from pydantic import ValidationError import litellm from litellm.constants import DEFAULT_GOOGLE_VIDEO_DURATION_SECONDS @@ -36,6 +39,9 @@ else: LiteLLMLoggingObj = Any BaseLLMException = Any +_VEO_MAX_REFERENCE_IMAGES: Final = 3 +_VEO_EIGHT_SECOND_RESOLUTIONS: Final = frozenset({"1080p", "4k"}) + def _convert_image_to_gemini_format(image_file) -> dict[str, str]: """ @@ -57,6 +63,111 @@ def _convert_image_to_gemini_format(image_file) -> dict[str, str]: return {"bytesBase64Encoded": base64_encoded, "mimeType": mime_type} +@runtime_checkable +class _BinaryFile(Protocol): + def read(self) -> bytes: ... + + +@runtime_checkable +class _Seekable(Protocol): + def seek(self, offset: int, /) -> int: ... + + +def _convert_video_to_gemini_format(video_file: _BinaryFile) -> Mapping[str, str]: + """The MIME type comes from the file name, falling back to video/mp4 (the format Veo generates)""" + guessed_type: Final = mimetypes.guess_type(str(getattr(video_file, "name", "")))[0] + mime_type: Final = guessed_type if guessed_type and guessed_type.startswith("video/") else "video/mp4" + if isinstance(video_file, _Seekable): + video_file.seek(0) + return MappingProxyType( + {"bytesBase64Encoded": base64.b64encode(video_file.read()).decode("utf-8"), "mimeType": mime_type} + ) + + +def _to_gemini_media(media: object, is_video: bool = False) -> object: + if not isinstance(media, _BinaryFile): + return media + return _convert_video_to_gemini_format(media) if is_video else _convert_image_to_gemini_format(media) + + +def _to_gemini_reference_image(reference: object) -> object: + if isinstance(reference, Mapping) and ("image" in reference or "referenceType" in reference): + return MappingProxyType({**reference, "image": _to_gemini_media(reference.get("image"))}) + return MappingProxyType({"image": _to_gemini_media(reference)}) + + +def _bad_request(message: str, model: str) -> litellm.BadRequestError: + return litellm.BadRequestError(message=f"Gemini Veo: {message}", model=model, llm_provider="gemini") + + +def _is_duration_eight_seconds(duration: object) -> bool: + if not isinstance(duration, (int, float, str)): + return False + try: + return float(duration) == 8 + except ValueError: + return False + + +def _validate_veo_request( + model: str, instance: GeminiVideoGenerationInstance, parameters: Mapping[str, object] +) -> None: + """ + Reject combinations the Gemini API documents as invalid before they reach Google. + See https://ai.google.dev/gemini-api/docs/veo#veo-model-parameters + """ + has_image: Final = instance.image is not None + has_video: Final = instance.video is not None + reference_images: Final = instance.referenceImages or () + has_reference_images: Final = len(reference_images) > 0 + resolution: Final = str(parameters.get("resolution") or "").strip().lower() + + if instance.lastFrame is not None and not has_image: + raise _bad_request("lastFrame requires image (the first frame) to be set.", model) + if has_reference_images and has_image: + raise _bad_request("referenceImages cannot be combined with image.", model) + if has_video and has_image: + raise _bad_request("video (extension) cannot be combined with image.", model) + if len(reference_images) > _VEO_MAX_REFERENCE_IMAGES: + raise _bad_request( + f"at most {_VEO_MAX_REFERENCE_IMAGES} referenceImages are allowed, got {len(reference_images)}.", model + ) + for reference in reference_images: + if reference.referenceType.lower() != "asset": + raise _bad_request( + f"referenceType '{reference.referenceType}' is not supported. The Gemini API only accepts 'asset' " + "reference images ('style' references are Vertex AI only).", + model, + ) + + if has_video and resolution not in ("", "720p"): + raise _bad_request(f"video extension only supports 720p resolution, got '{resolution}'.", model) + + duration: Final = parameters.get("durationSeconds") + if duration is not None and not _is_duration_eight_seconds(duration): + eight_second_reasons: Final = tuple( + reason + for reason, applies in ( + ("referenceImages", has_reference_images), + ("video extension", has_video), + (f"{resolution} resolution", resolution in _VEO_EIGHT_SECOND_RESOLUTIONS), + ) + if applies + ) + if eight_second_reasons: + raise _bad_request( + f"durationSeconds must be 8 when using {', '.join(eight_second_reasons)}, got {duration}.", model + ) + + person_generation: Final = parameters.get("personGeneration") + if person_generation is not None and (has_image or has_reference_images) and person_generation != "allow_adult": + raise _bad_request( + f"personGeneration must be 'allow_adult' for image-to-video, interpolation and " + f"referenceImages requests, got '{person_generation}'.", + model, + ) + + def _json_payload(raw_response: httpx.Response) -> object: """Read an HTTP response body as an opaque JSON payload.""" return raw_response.json() @@ -265,7 +376,10 @@ class GeminiVideoConfig(BaseVideoConfig): "image": { "bytesBase64Encoded": "...", "mimeType": "image/jpeg" - } + }, + "lastFrame": {...}, # interpolation, requires image + "referenceImages": [...], # up to 3 asset references + "video": {...} # extension of a Veo-generated video } ], "parameters": { @@ -274,25 +388,43 @@ class GeminiVideoConfig(BaseVideoConfig): "resolution": "720p" } } + + Media inputs (image, lastFrame, referenceImages, video) belong in + instances[0]; parameters only carries generation config. """ - instance: Final[GeminiVideoGenerationInstance] = {"prompt": prompt} - params_copy: Final = video_create_optional_request_params.copy() + image: Final = params_copy.pop("image", None) + last_frame: Final = params_copy.pop("lastFrame", None) + reference_images: Final = params_copy.pop("referenceImages", None) + video: Final = params_copy.pop("video", None) - if "image" in params_copy: - image: Final = params_copy.pop("image") - if image is not None: - if isinstance(image, dict): - image_data = image - else: - image_data = _convert_image_to_gemini_format(image) - instance["image"] = image_data + if reference_images is not None and not isinstance(reference_images, (list, tuple)): + raise _bad_request("referenceImages must be a list.", model) + + try: + instance: Final = GeminiVideoGenerationInstance.model_validate( + MappingProxyType( + { + "prompt": prompt, + "image": _to_gemini_media(image), + "lastFrame": _to_gemini_media(last_frame), + "referenceImages": tuple(_to_gemini_reference_image(r) for r in reference_images) + if reference_images + else None, + "video": _to_gemini_media(video, is_video=True), + } + ) + ) + except ValidationError as e: + raise _bad_request(f"invalid media input: {e}", model) from e + + _validate_veo_request(model=model, instance=instance, parameters=params_copy) parameters: Final = GeminiVideoGenerationParameters(**params_copy) request_body_obj: Final = GeminiVideoGenerationRequest(instances=[instance], parameters=parameters) - request_data: Final = request_body_obj.model_dump(exclude_none=True) + request_data: Final = request_body_obj.model_dump(mode="json", exclude_none=True) return request_data, [], api_base diff --git a/litellm/types/llms/gemini.py b/litellm/types/llms/gemini.py index 57fb8b5b0cd..1526f65f218 100644 --- a/litellm/types/llms/gemini.py +++ b/litellm/types/llms/gemini.py @@ -1,7 +1,8 @@ +from collections.abc import Mapping from enum import Enum from typing import Any, Literal -from typing_extensions import Required, TypedDict +from typing_extensions import TypedDict from .vertex_ai import ( GenerationConfig, @@ -151,7 +152,7 @@ class BidiGenerateContentSetup(TypedDict, total=False): # Image Generation Types -from pydantic import BaseModel +from pydantic import BaseModel, ConfigDict class GeminiImageGenerationInstance(TypedDict): @@ -231,11 +232,31 @@ class GeminiImageGenerationResponse(TypedDict): # Video Generation Types -class GeminiVideoGenerationInstance(TypedDict, total=False): - """Instance data for Gemini video generation request""" +class GeminiVideoReferenceImage(BaseModel): + """Asset reference image for Veo 3.1 video generation""" - prompt: Required[str] - image: dict[str, Any] + model_config = ConfigDict(frozen=True) + + image: Mapping[str, object] + referenceType: str = "asset" + + +class GeminiVideoGenerationInstance(BaseModel): + """Instance data for Gemini video generation request. All media inputs live here, not in parameters""" + + model_config = ConfigDict(frozen=True) + + prompt: str + image: Mapping[str, object] | None = None + + lastFrame: Mapping[str, object] | None = None + """The final image for an interpolation video. Must be used with 'image'.""" + + referenceImages: tuple[GeminiVideoReferenceImage, ...] | None = None + """Up to three asset reference images. Veo 3.1 and Veo 3.1 Fast only.""" + + video: Mapping[str, object] | None = None + """Veo-generated video to extend. Veo 3.1 and Veo 3.1 Fast only.""" class GeminiVideoGenerationParameters(BaseModel): @@ -250,37 +271,19 @@ class GeminiVideoGenerationParameters(BaseModel): durationSeconds: int | None = None """ - Length of the generated video in seconds (e.g., 4, 5, 6, 8). - Must be 8 when using extension/interpolation or referenceImages. + Length of the generated video in seconds (e.g., 4, 6, 8). + Must be 8 when using extension, referenceImages, or 1080p/4k resolution. """ resolution: str | None = None """ - Video resolution (e.g., '720p', '1080p'). - '1080p' only supports 8s duration; extension only supports '720p'. + Video resolution (e.g., '720p', '1080p', '4k'). + '1080p' and '4k' only support 8s duration; extension only supports '720p'. """ negativePrompt: str | None = None """Text describing what not to include in the video.""" - lastFrame: Any | None = None - """ - The final image for interpolation video to transition. - Should be used with the 'image' parameter. - """ - - referenceImages: list | None = None - """ - Up to three images to be used as style/content references. - Only supported in Veo 3.1 (list of VideoGenerationReferenceImage objects). - """ - - video: Any | None = None - """ - Video to be used for video extension (Video object). - Only supported in Veo 3.1 & Veo 3 Fast. - """ - personGeneration: str | None = None """ Controls the generation of people. diff --git a/tests/unit/llms/gemini/videos/test_gemini_video_media_inputs.py b/tests/unit/llms/gemini/videos/test_gemini_video_media_inputs.py new file mode 100644 index 00000000000..a8a644d0ee0 --- /dev/null +++ b/tests/unit/llms/gemini/videos/test_gemini_video_media_inputs.py @@ -0,0 +1,334 @@ +""" +Gemini Veo media inputs (image, lastFrame, referenceImages, video) belong in +instances[0]; parameters only carries generation config. + +See: https://ai.google.dev/gemini-api/docs/veo#veo-model-parameters +""" + +import base64 +import io +import json +from typing import Any + +import httpx +import pytest + +import litellm +from litellm.llms.custom_httpx.http_handler import HTTPHandler +from litellm.llms.gemini.videos.transformation import GeminiVideoConfig +from litellm.types.router import GenericLiteLLMParams + +VEO_31 = "veo-3.1-generate-preview" +VEO_31_LITE = "veo-3.1-lite-generate-preview" +API_BASE = f"https://generativelanguage.googleapis.com/v1beta/models/{VEO_31}:predictLongRunning" + +FIRST_FRAME = {"bytesBase64Encoded": "Zmlyc3Q=", "mimeType": "image/png"} +LAST_FRAME = {"bytesBase64Encoded": "bGFzdA==", "mimeType": "image/png"} +DRESS = {"bytesBase64Encoded": "ZHJlc3M=", "mimeType": "image/jpeg"} +VEO_VIDEO = {"uri": "https://generativelanguage.googleapis.com/v1beta/files/abc:download?alt=media"} + +PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"png-payload" +JPEG_BYTES = b"\xff\xd8\xff\xe0" + b"jpeg-payload" + + +def _request_body(params: dict[str, Any], model: str = VEO_31, prompt: str = "a prompt") -> dict: + data, files, _ = GeminiVideoConfig().transform_video_create_request( + model=model, + prompt=prompt, + api_base=API_BASE, + video_create_optional_request_params=params, + litellm_params=GenericLiteLLMParams(), + headers={}, + ) + assert files == [] + return data + + +def _file(content: bytes, name: str | None = None) -> io.BytesIO: + file = io.BytesIO(content) + if name is not None: + file.name = name + return file + + +def _b64(content: bytes) -> str: + return base64.b64encode(content).decode("utf-8") + + +@pytest.mark.parametrize( + ("params", "expected_body"), + [ + pytest.param( + {"aspectRatio": "16:9", "durationSeconds": 4}, + {"instances": [{"prompt": "a prompt"}], "parameters": {"aspectRatio": "16:9", "durationSeconds": 4}}, + id="text-to-video", + ), + pytest.param( + {"image": FIRST_FRAME, "durationSeconds": 6, "personGeneration": "allow_adult"}, + { + "instances": [{"prompt": "a prompt", "image": FIRST_FRAME}], + "parameters": {"durationSeconds": 6, "personGeneration": "allow_adult"}, + }, + id="image-to-video", + ), + pytest.param( + {"image": FIRST_FRAME, "lastFrame": LAST_FRAME, "aspectRatio": "9:16"}, + { + "instances": [{"prompt": "a prompt", "image": FIRST_FRAME, "lastFrame": LAST_FRAME}], + "parameters": {"aspectRatio": "9:16"}, + }, + id="interpolation", + ), + pytest.param( + {"referenceImages": [{"image": DRESS, "referenceType": "asset"}], "durationSeconds": 8}, + { + "instances": [{"prompt": "a prompt", "referenceImages": [{"image": DRESS, "referenceType": "asset"}]}], + "parameters": {"durationSeconds": 8}, + }, + id="reference-images", + ), + pytest.param( + {"video": VEO_VIDEO, "resolution": "720p"}, + {"instances": [{"prompt": "a prompt", "video": VEO_VIDEO}], "parameters": {"resolution": "720p"}}, + id="video-extension", + ), + ], +) +def test_media_inputs_go_to_instance_and_config_stays_in_parameters(params, expected_body): + assert _request_body(params) == expected_body + + +def test_file_like_last_frame_keeps_its_bytes_and_mime_type(): + body = _request_body({"image": _file(JPEG_BYTES), "lastFrame": _file(PNG_BYTES)}) + + assert body["instances"][0] == { + "prompt": "a prompt", + "image": {"bytesBase64Encoded": _b64(JPEG_BYTES), "mimeType": "image/jpeg"}, + "lastFrame": {"bytesBase64Encoded": _b64(PNG_BYTES), "mimeType": "image/png"}, + } + + +def test_reference_images_accept_file_likes_bare_or_wrapped(): + body = _request_body( + { + "referenceImages": [ + _file(PNG_BYTES), + {"image": _file(JPEG_BYTES), "referenceType": "ASSET"}, + DRESS, + ] + } + ) + + assert body["instances"][0]["referenceImages"] == [ + {"image": {"bytesBase64Encoded": _b64(PNG_BYTES), "mimeType": "image/png"}, "referenceType": "asset"}, + {"image": {"bytesBase64Encoded": _b64(JPEG_BYTES), "mimeType": "image/jpeg"}, "referenceType": "ASSET"}, + {"image": DRESS, "referenceType": "asset"}, + ] + + +@pytest.mark.parametrize( + ("video_file", "expected_mime_type"), + [ + pytest.param(_file(b"mp4-bytes", name="clip.mp4"), "video/mp4", id="named-mp4"), + pytest.param(_file(b"webm-bytes", name="clip.webm"), "video/webm", id="named-webm"), + pytest.param(_file(b"unnamed-bytes"), "video/mp4", id="unnamed-defaults-to-mp4"), + ], +) +def test_file_like_video_keeps_its_bytes_and_mime_type(video_file, expected_mime_type): + expected_bytes = video_file.getvalue() + + body = _request_body({"video": video_file}) + + assert body["instances"][0]["video"] == { + "bytesBase64Encoded": _b64(expected_bytes), + "mimeType": expected_mime_type, + } + + +class _NonIOBaseSeekableReader: + def __init__(self, content: bytes) -> None: + self._file = io.BytesIO(content) + + def read(self) -> bytes: + return self._file.read() + + def seek(self, offset: int) -> int: + return self._file.seek(offset) + + +def test_partly_read_video_wrapper_is_sent_from_the_start(): + video_file = _NonIOBaseSeekableReader(b"full-video-bytes") + video_file.read() + + body = _request_body({"video": video_file}) + + assert body["instances"][0]["video"] == {"bytesBase64Encoded": _b64(b"full-video-bytes"), "mimeType": "video/mp4"} + + +@pytest.mark.parametrize( + ("model", "params"), + [ + pytest.param( + VEO_31, {"image": FIRST_FRAME, "lastFrame": LAST_FRAME, "durationSeconds": 4}, id="interpolation-4s" + ), + pytest.param(VEO_31, {"personGeneration": "allow_adult"}, id="text-to-video-allow-adult-eu"), + pytest.param(VEO_31, {"resolution": "1080p", "durationSeconds": "8"}, id="1080p-8s-as-string"), + pytest.param(VEO_31, {"resolution": "4k", "durationSeconds": 8}, id="4k-8s"), + pytest.param(VEO_31, {"video": VEO_VIDEO, "resolution": "720p", "durationSeconds": 8}, id="extension-720p-8s"), + pytest.param(VEO_31_LITE, {"image": FIRST_FRAME, "lastFrame": LAST_FRAME}, id="lite-interpolation"), + pytest.param(VEO_31_LITE, {"resolution": "1080p", "durationSeconds": 8}, id="lite-1080p"), + pytest.param(f"gemini/{VEO_31}", {"referenceImages": [DRESS] * 3}, id="three-reference-images"), + ], +) +def test_documented_valid_combinations_are_not_rejected(model, params): + body = _request_body(params, model=model) + + assert body["instances"][0]["prompt"] == "a prompt" + + +@pytest.mark.parametrize( + ("model", "params", "expected_message"), + [ + pytest.param(VEO_31, {"lastFrame": LAST_FRAME}, "lastFrame requires image", id="last-frame-without-image"), + pytest.param( + VEO_31, + {"image": FIRST_FRAME, "referenceImages": [DRESS]}, + "referenceImages cannot be combined with image", + id="reference-images-with-image", + ), + pytest.param( + VEO_31, + {"image": FIRST_FRAME, "video": VEO_VIDEO}, + "video (extension) cannot be combined with image", + id="video-with-image", + ), + pytest.param( + VEO_31, + {"referenceImages": [DRESS] * 4}, + "at most 3 referenceImages are allowed, got 4", + id="four-reference-images", + ), + pytest.param( + VEO_31, + {"referenceImages": [{"image": DRESS, "referenceType": "style"}]}, + "referenceType 'style' is not supported", + id="style-reference", + ), + pytest.param( + VEO_31, {"referenceImages": DRESS}, "referenceImages must be a list", id="reference-images-not-list" + ), + pytest.param( + VEO_31, + {"referenceImages": [{"referenceType": "asset", "img": DRESS}]}, + "invalid media input", + id="reference-without-image-key", + ), + pytest.param( + VEO_31, + {"referenceImages": [DRESS], "durationSeconds": 4}, + "durationSeconds must be 8 when using referenceImages, got 4", + id="reference-images-4s", + ), + pytest.param( + VEO_31, + {"video": VEO_VIDEO, "durationSeconds": 6}, + "durationSeconds must be 8 when using video extension, got 6", + id="extension-6s", + ), + pytest.param( + VEO_31, + {"resolution": "1080p", "durationSeconds": 4}, + "durationSeconds must be 8 when using 1080p resolution, got 4", + id="1080p-4s", + ), + pytest.param( + VEO_31, + {"resolution": "4K", "durationSeconds": "6"}, + "durationSeconds must be 8 when using 4k resolution, got 6", + id="4k-6s", + ), + pytest.param( + VEO_31, + {"referenceImages": [DRESS], "durationSeconds": "eight"}, + "durationSeconds must be 8 when using referenceImages, got eight", + id="reference-images-non-numeric-duration", + ), + pytest.param( + VEO_31, + {"resolution": "1080p", "durationSeconds": [8]}, + "durationSeconds must be 8 when using 1080p resolution, got [8]", + id="1080p-list-duration", + ), + pytest.param( + VEO_31, + {"video": VEO_VIDEO, "resolution": "1080p", "durationSeconds": 8}, + "video extension only supports 720p resolution, got '1080p'", + id="extension-1080p", + ), + pytest.param( + VEO_31, + {"image": FIRST_FRAME, "personGeneration": "allow_all"}, + "personGeneration must be 'allow_adult'", + id="image-to-video-allow-all", + ), + pytest.param( + VEO_31, + {"referenceImages": [DRESS], "personGeneration": "allow_all"}, + "personGeneration must be 'allow_adult'", + id="reference-images-allow-all", + ), + pytest.param( + VEO_31, + {"image": FIRST_FRAME, "lastFrame": "/tmp/last.png"}, + "invalid media input", + id="last-frame-as-path-string", + ), + ], +) +def test_invalid_combinations_raise_bad_request(model, params, expected_message): + with pytest.raises(litellm.BadRequestError) as exc_info: + _request_body(params, model=model) + + assert exc_info.value.status_code == 400 + assert expected_message in str(exc_info.value), str(exc_info.value) + + +def _recording_client(requests: list[dict]) -> HTTPHandler: + def handler(request: httpx.Request) -> httpx.Response: + requests.append(json.loads(request.content)) + return httpx.Response(200, json={"name": "models/veo/operations/abc", "done": False}) + + return HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(handler))) + + +def test_video_generation_sends_last_frame_in_instance(): + requests: list[dict] = [] + + litellm.video_generation( + model=f"gemini/{VEO_31}", + prompt="a prompt", + input_reference=FIRST_FRAME, + lastFrame=LAST_FRAME, + api_key="test-key", + client=_recording_client(requests), + ) + + assert requests == [ + {"instances": [{"prompt": "a prompt", "image": FIRST_FRAME, "lastFrame": LAST_FRAME}], "parameters": {}} + ] + + +def test_video_generation_rejects_invalid_request_before_calling_google(): + requests: list[dict] = [] + + with pytest.raises(litellm.BadRequestError) as exc_info: + litellm.video_generation( + model=f"gemini/{VEO_31}", + prompt="a prompt", + lastFrame=LAST_FRAME, + api_key="test-key", + client=_recording_client(requests), + ) + + assert exc_info.value.status_code == 400 + assert requests == [], "invalid request must not reach the Gemini API"