From 1b1ddef1b2aab936e33f2ccbc1dc145f76940885 Mon Sep 17 00:00:00 2001 From: debanshd Date: Fri, 16 Oct 2026 01:00:48 -0700 Subject: [PATCH] fix: preserve non-null anyOf branches when stripping null types for Vertex (#41683) * fix: preserve non-null anyOf branches when stripping null types for Vertex * refactor: build anyOf branches immutably and drop explanatory comments * style: apply ruff format * fix: annotate anyOf list construction for the LIT002 budget * fix(vertex_ai): send a null-only anyOf as one nullable null instead of raising * test(vertex_ai): pin empty anyOf passthrough * fix(vertex_ai): forward a non-list anyOf unchanged and add wire cells for the null fold * fix(vertex_ai): forward a non-object anyOf branch for Google to judge instead of crashing * fix(vertex_ai): fold null out of type arrays and union branches before sending to Gemini --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com> --- litellm/llms/vertex_ai/common_utils.py | 42 +- .../test_gemini_anyof_null_branches_wire.py | 681 ++++++++++++++++++ tests/unit/llms/vertex_ai/test_vertex.py | 2 +- .../vertex_ai/test_vertex_ai_common_utils.py | 143 ++++ 4 files changed, 843 insertions(+), 25 deletions(-) create mode 100644 tests/integration/providers/test_gemini_anyof_null_branches_wire.py diff --git a/litellm/llms/vertex_ai/common_utils.py b/litellm/llms/vertex_ai/common_utils.py index 98bc679a78e..581e2cb15f7 100644 --- a/litellm/llms/vertex_ai/common_utils.py +++ b/litellm/llms/vertex_ai/common_utils.py @@ -738,14 +738,14 @@ def build_vertex_schema(parameters: dict, add_property_ordering: bool = False): # refs recursively and correctly detects/skips circular references. unpack_defs(parameters, defs) + _convert_schema_types(parameters) + # 5. Nullable fields: # * https://github.com/pydantic/pydantic/issues/1270 # * https://stackoverflow.com/a/58841311 # * https://github.com/pydantic/pydantic/discussions/4872 convert_anyof_null_to_nullable(parameters) - _convert_schema_types(parameters) - # Handle empty strings in enum values - Gemini doesn't accept empty strings in enums _fix_enum_empty_strings(parameters) @@ -926,34 +926,28 @@ def convert_anyof_null_to_nullable(schema, depth=0): ) """ Converts null objects within anyOf by removing them and adding nullable to all remaining objects """ anyof: Final = schema.get("anyOf", None) - if anyof is not None: - contains_null = False + if isinstance(anyof, list): + non_null_branches: Final = [ + atype for atype in anyof if not (isinstance(atype, dict) and atype.get("type") == "null") + ] + contains_null: Final = len(non_null_branches) < len(anyof) + anyof[:] = non_null_branches or anyof[:1] for atype in anyof: - if isinstance(atype, dict) and atype.get("type") == "null": - # remove null type - anyof.remove(atype) - contains_null = True - elif "type" not in atype and len(atype) == 0: - # Handle empty object case + if not isinstance(atype, dict): + continue + if not atype: atype["type"] = "object" - - if len(anyof) == 0: - # Edge case: response schema with only null type present is invalid in Vertex AI - raise ValueError( - "Invalid input: AnyOf schema with only null type is not supported. Please provide a non-null type." - ) - - if contains_null: - # set all types to nullable following guidance found here: https://cloud.google.com/vertex-ai/generative-ai/docs/samples/generativeaionvertexai-gemini-controlled-generation-response-schema-3#generativeaionvertexai_gemini_controlled_generation_response_schema_3-python - # Empty `items: {}` on array branches is left in place; downstream - # process_items() converts it to {"type": "object"}, which Vertex - # requires whenever type == "array" (even inside anyOf). - for atype in anyof: + if contains_null: + # set all types to nullable following guidance found here: https://cloud.google.com/vertex-ai/generative-ai/docs/samples/generativeaionvertexai-gemini-controlled-generation-response-schema-3#generativeaionvertexai_gemini_controlled_generation_response_schema_3-python + # Empty `items: {}` on array branches is left in place; downstream + # process_items() converts it to {"type": "object"}, which Vertex + # requires whenever type == "array" (even inside anyOf). atype["nullable"] = True + convert_anyof_null_to_nullable(atype, depth=depth + 1) properties: Final = schema.get("properties", None) if properties is not None: - for name, value in properties.items(): + for value in properties.values(): convert_anyof_null_to_nullable(value, depth=depth + 1) items: Final = schema.get("items", None) diff --git a/tests/integration/providers/test_gemini_anyof_null_branches_wire.py b/tests/integration/providers/test_gemini_anyof_null_branches_wire.py new file mode 100644 index 00000000000..9fa99cb69f8 --- /dev/null +++ b/tests/integration/providers/test_gemini_anyof_null_branches_wire.py @@ -0,0 +1,681 @@ +import asyncio +import base64 +import json +import re +import threading +import uuid +from collections.abc import AsyncIterator, Iterator, Mapping +from concurrent.futures import ThreadPoolExecutor +from itertools import product +from types import MappingProxyType +from typing import Final, Literal + +import anthropic +import httpx +import openai +import pytest +import websockets +from integration._support.client import ( + JSON_OBJECT, + Gateway, + Scenario, + gateway_from_environment, + object_value, + string_value, +) +from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process +from integration._support.vertex import service_account_json +from integration._support.wire import Reply, Request, Wire, wire_server +from integration._support.upstream import GEMINI_LIVE_PATH +from integration.providers._gemini_tts_voice_support import live_deployment, live_scenario +from integration.providers._vertex_batch_rows import encoded, gcs_peer, is_gcs_upload, jsonl, upload, vertex_model +from pydantic import JsonValue, TypeAdapter + +Provider = Literal["vertex_ai", "gemini"] +Endpoint = Literal["chat", "messages", "responses"] + +_BACKEND: Final = "gemini-3.8-flash" +_NO_JSON_SCHEMA_BACKEND: Final = "gemini-flash-latest" +_IMAGE_BACKEND: Final = "gemini-3.1-flash-image" +_PROJECT: Final = "scripted-project" +_LOCATION: Final = "us-central1" +_GEMINI_KEY: Final = "sk-scripted-gemini" +_COUNT: Final = 777 +_FAIL_MARKER: Final = "scripted-invalid-argument" +_STRING_ANYOF: Final = re.compile(rb'"anyOf":\s*"') +_SCALAR_BRANCH: Final = re.compile(rb'"anyOf":\s*\[\s*(?:true|false|-?\d)') +_PNG: Final = base64.b64encode(b"\x89PNG\r\n\x1a\nscripted").decode() +_JSON_LIST: Final = TypeAdapter(list[JsonValue]) + +_NULL: Final[Mapping[str, JsonValue]] = {"type": "null"} +_STRING: Final[Mapping[str, JsonValue]] = {"type": "string"} +_INTEGER: Final[Mapping[str, JsonValue]] = {"type": "integer"} +_NULLABLE_STRING: Final[JsonValue] = {"anyOf": [{"type": "string", "nullable": True}]} +_NULLABLE_NULL: Final[JsonValue] = {"anyOf": [{"type": "null", "nullable": True}]} + +_UNIONS: Final[tuple[tuple[str, JsonValue, JsonValue], ...]] = ( + ("null-null-string", {"anyOf": [_NULL, _NULL, _STRING]}, _NULLABLE_STRING), + ("null-only", {"anyOf": [_NULL]}, _NULLABLE_NULL), + ("two-nulls-only", {"anyOf": [_NULL, _NULL]}, _NULLABLE_NULL), + ("string-null", {"anyOf": [_STRING, _NULL]}, _NULLABLE_STRING), + ( + "integer-null-null-string", + {"anyOf": [_INTEGER, _NULL, _NULL, _STRING]}, + {"anyOf": [{"type": "integer", "nullable": True}, {"type": "string", "nullable": True}]}, + ), + ("null-empty", {"anyOf": [_NULL, {}]}, {"anyOf": [{"type": "object", "nullable": True}]}), + ("empty-only", {"anyOf": [{}]}, {"anyOf": [{"type": "object"}]}), + ("empty-list", {"anyOf": []}, {"anyOf": []}), + ("object-anyof", {"anyOf": _STRING}, {"anyOf": _STRING}), + ("null-anyof", {"anyOf": None}, {"anyOf": None}), + ("five-hundred-nulls-then-string", {"anyOf": [*([_NULL] * 500), _STRING]}, _NULLABLE_STRING), + ("string-null-type-array", {"type": ["string", "null"]}, _NULLABLE_STRING), + ( + "null-inside-a-union-branch", + {"anyOf": [{"type": "object", "properties": {"x": {"anyOf": [_INTEGER, _NULL]}}}, _NULL]}, + { + "anyOf": [ + { + "type": "object", + "properties": {"x": {"anyOf": [{"type": "integer", "nullable": True}]}}, + "nullable": True, + } + ] + }, + ), +) +_FANOUT_UNIONS: Final = tuple(case for case in _UNIONS if case[0] in {"null-null-string", "null-only", "object-anyof"}) + + +def _text_reply() -> Mapping[str, JsonValue]: + return { + "candidates": [{"content": {"role": "model", "parts": [{"text": "OK"}]}, "finishReason": "STOP"}], + "usageMetadata": {"promptTokenCount": 9, "candidatesTokenCount": 1, "totalTokenCount": 10}, + "modelVersion": _BACKEND, + } + + +def _image_reply() -> Mapping[str, JsonValue]: + return { + "candidates": [ + {"content": {"role": "model", "parts": [{"inlineData": {"mimeType": "image/png", "data": _PNG}}]}} + ], + "usageMetadata": {"promptTokenCount": 9, "candidatesTokenCount": 1, "totalTokenCount": 10}, + } + + +def _rejection(message: str) -> Reply: + error: Final = {"code": 400, "message": message, "status": "INVALID_ARGUMENT"} + return Reply(status=400, body=json.dumps({"error": error}).encode()) + + +def _google(request: Request) -> Reply: + if is_gcs_upload(request): + return gcs_peer(request) + if _FAIL_MARKER.encode() in request.body: + return _rejection(f"{_FAIL_MARKER}: the scripted tool schema was refused") + if _STRING_ANYOF.search(request.body): + return _rejection("Invalid value at 'tools[0].function_declarations[0].parameters' (anyOf), not a list") + if _SCALAR_BRANCH.search(request.body): + return _rejection( + "Invalid value at 'tools[0].function_declarations[0].parameters.properties[0].value.any_of[0]'" + ) + path: Final = request.target.split("?")[0] + if path.endswith(":countTokens"): + return Reply(body=json.dumps({"totalTokens": _COUNT}).encode()) + if path.endswith(":streamGenerateContent"): + return Reply(content_type="text/event-stream", chunks=(f"data: {json.dumps(_text_reply())}\n\n".encode(),)) + if path.endswith(f"{_IMAGE_BACKEND}:generateContent"): + return Reply(body=json.dumps(_image_reply()).encode()) + if path.endswith(":generateContent"): + return Reply(body=json.dumps(_text_reply()).encode()) + return Reply(status=404, body=json.dumps({"error": {"code": 404, "message": request.target}}).encode()) + + +def _deployment(gateway: Gateway, scenario: Scenario, wire: Wire, provider: Provider, backend: str = _BACKEND) -> str: + if provider == "gemini": + return scenario.model(model=f"gemini/{backend}", api_base=wire.url, api_key=_GEMINI_KEY) + return scenario.model( + model=f"vertex_ai/{backend}", + api_base=wire.url, + api_key=None, + vertex_project=_PROJECT, + vertex_location=_LOCATION, + vertex_credentials=service_account_json(_PROJECT, gateway.upstream_url.rstrip("/")), + ) + + +def _parameters(value: JsonValue) -> dict[str, JsonValue]: + return {"type": "object", "properties": {"value": value}, "required": ["value"]} + + +def _declaration(description: str, value: JsonValue) -> dict[str, JsonValue]: + return {"name": "record", "description": description, "parameters": _parameters(value)} + + +def _tool(endpoint: Endpoint, description: str, parameters: JsonValue) -> dict[str, JsonValue]: + if endpoint == "messages": + return {"name": "record", "description": description, "input_schema": parameters} + if endpoint == "responses": + return {"type": "function", "name": "record", "description": description, "parameters": parameters} + return {"type": "function", "function": {"name": "record", "description": description, "parameters": parameters}} + + +def _request_body( + endpoint: Endpoint, model: str, description: str, parameters: JsonValue, stream: bool +) -> tuple[str, dict[str, JsonValue]]: + prompt: Final = f"Record the value hello by calling the record tool. {description}" + tools: Final[list[JsonValue]] = [_tool(endpoint, description, parameters)] + streaming: Final[dict[str, JsonValue]] = {"stream": True} if stream else {} + if endpoint == "messages": + messages_body: Final[dict[str, JsonValue]] = { + "model": model, + "max_tokens": 16, + "messages": [{"role": "user", "content": prompt}], + "tools": tools, + } + return "/v1/messages", {**messages_body, **streaming} + if endpoint == "responses": + return "/v1/responses", {"model": model, "input": prompt, "tools": tools, **streaming} + chat_body: Final[dict[str, JsonValue]] = { + "model": model, + "max_tokens": 16, + "messages": [{"role": "user", "content": prompt}], + "tools": tools, + } + return "/v1/chat/completions", {**chat_body, **streaming} + + +def _body(request: Request) -> dict[str, JsonValue]: + return JSON_OBJECT.validate_json(request.body) + + +def _json(response: httpx.Response) -> dict[str, JsonValue]: + return JSON_OBJECT.validate_json(response.content) + + +def _generate_requests(wire: Wire) -> tuple[Request, ...]: + return tuple( + request + for request in wire.drain() + if request.target.split("?")[0].endswith((":generateContent", ":streamGenerateContent")) + ) + + +def _sent_declarations(body: Mapping[str, JsonValue]) -> JsonValue: + tools: Final = _JSON_LIST.validate_python(body["tools"]) + assert len(tools) == 1, tools + tool: Final = object_value(tools[0]) + assert len(tool) == 1, tool + return next(iter(tool.values())) + + +def _description() -> str: + return f"Record a value {uuid.uuid4().hex[:12]}" + + +def _assert_caller_saw_ok(endpoint: Endpoint, response: httpx.Response, stream: bool) -> None: + assert response.status_code == 200, response.text + if stream: + assert "OK" in response.text, response.text + return + payload: Final = _json(response) + assert string_value(payload["id"]), payload + if endpoint == "chat": + choice: Final = object_value(_JSON_LIST.validate_python(payload["choices"])[0]) + assert object_value(choice["message"])["content"] == "OK", payload + return + assert "OK" in response.text, payload + + +@pytest.mark.parametrize("provider", ("vertex_ai", "gemini")) +@pytest.mark.parametrize(("union", "expected"), [case[1:] for case in _UNIONS], ids=[case[0] for case in _UNIONS]) +def test_chat_tool_union_reaches_google_with_null_branches_folded_into_nullable( + gateway: Gateway, provider: Provider, union: JsonValue, expected: JsonValue +) -> None: + with wire_server(_google) as wire, gateway.scenario() as scenario: + model: Final = _deployment(gateway, scenario, wire, provider) + description: Final = _description() + path, body = _request_body("chat", model, description, _parameters(union), False) + response: Final = gateway.request("POST", path, body) + _assert_caller_saw_ok("chat", response, False) + requests: Final = _generate_requests(wire) + assert len(requests) == 1, [request.target for request in requests] + assert requests[0].target.split("?")[0].endswith(f"{_BACKEND}:generateContent"), requests[0].target + assert _sent_declarations(_body(requests[0])) == [_declaration(description, expected)] + + +@pytest.mark.parametrize("stream", (False, True), ids=("whole", "stream")) +@pytest.mark.parametrize("endpoint", ("chat", "messages", "responses")) +@pytest.mark.parametrize("provider", ("vertex_ai", "gemini")) +@pytest.mark.parametrize( + ("union", "expected"), [case[1:] for case in _FANOUT_UNIONS], ids=[case[0] for case in _FANOUT_UNIONS] +) +def test_every_unified_endpoint_sends_the_same_folded_union( + gateway: Gateway, provider: Provider, endpoint: Endpoint, stream: bool, union: JsonValue, expected: JsonValue +) -> None: + with wire_server(_google) as wire, gateway.scenario() as scenario: + model: Final = _deployment(gateway, scenario, wire, provider) + description: Final = _description() + path, body = _request_body(endpoint, model, description, _parameters(union), stream) + response: Final = gateway.request("POST", path, body) + _assert_caller_saw_ok(endpoint, response, stream) + requests: Final = _generate_requests(wire) + assert len(requests) == 1, [request.target for request in requests] + method: Final = "streamGenerateContent" if stream else "generateContent" + assert requests[0].target.split("?")[0].endswith(f"{_BACKEND}:{method}"), requests[0].target + assert _sent_declarations(_body(requests[0])) == [_declaration(description, expected)] + + +def test_nested_item_and_property_unions_are_folded_at_every_depth(gateway: Gateway) -> None: + parameters: Final[dict[str, JsonValue]] = { + "type": "object", + "properties": { + "values": {"type": "array", "items": {"anyOf": [_NULL, _NULL, _INTEGER]}}, + "nested": {"type": "object", "properties": {"inner": {"anyOf": [_NULL, _NULL, _STRING]}}}, + }, + "required": ["values", "nested"], + } + with wire_server(_google) as wire, gateway.scenario() as scenario: + model: Final = _deployment(gateway, scenario, wire, "vertex_ai") + description: Final = _description() + path, body = _request_body("chat", model, description, parameters, False) + response: Final = gateway.request("POST", path, body) + _assert_caller_saw_ok("chat", response, False) + requests: Final = _generate_requests(wire) + assert len(requests) == 1, [request.target for request in requests] + assert _sent_declarations(_body(requests[0])) == [ + { + "name": "record", + "description": description, + "parameters": { + "type": "object", + "properties": { + "values": {"type": "array", "items": {"anyOf": [{"type": "integer", "nullable": True}]}}, + "nested": {"type": "object", "properties": {"inner": _NULLABLE_STRING}}, + }, + "required": ["values", "nested"], + }, + } + ] + + +def test_a_string_anyof_reaches_google_and_its_400_reaches_the_caller(gateway: Gateway) -> None: + with wire_server(_google) as wire, gateway.scenario() as scenario: + model: Final = _deployment(gateway, scenario, wire, "gemini") + description: Final = _description() + path, body = _request_body("chat", model, description, _parameters({"anyOf": "string"}), False) + response: Final = gateway.request("POST", path, body) + assert response.status_code == 400, response.text + assert "not a list" in response.text, response.text + requests: Final = _generate_requests(wire) + assert len(requests) == 1, [request.target for request in requests] + assert _sent_declarations(_body(requests[0])) == [_declaration(description, {"anyOf": "string"})] + + +@pytest.mark.parametrize("provider", ("vertex_ai", "gemini")) +def test_a_google_rejection_of_a_folded_union_reaches_the_caller(gateway: Gateway, provider: Provider) -> None: + with wire_server(_google) as wire, gateway.scenario() as scenario: + model: Final = _deployment(gateway, scenario, wire, provider) + description: Final = f"{_FAIL_MARKER} {uuid.uuid4().hex[:12]}" + path, body = _request_body("chat", model, description, _parameters({"anyOf": [_NULL, _NULL, _STRING]}), False) + response: Final = gateway.request("POST", path, body) + assert response.status_code == 400, response.text + assert _FAIL_MARKER in response.text, response.text + requests: Final = _generate_requests(wire) + assert len(requests) == 1, [request.target for request in requests] + assert _sent_declarations(_body(requests[0])) == [_declaration(description, _NULLABLE_STRING)] + + +@pytest.mark.parametrize( + ("union", "sent"), + (({"anyOf": [True, _NULL]}, {"anyOf": [True]}), ({"anyOf": [_NULL, 5]}, {"anyOf": [5]})), + ids=("true-null", "null-int"), +) +@pytest.mark.parametrize("provider", ("vertex_ai", "gemini")) +def test_a_union_branch_that_is_not_an_object_reaches_google_and_its_400_reaches_the_caller( + gateway: Gateway, provider: Provider, union: JsonValue, sent: JsonValue +) -> None: + with wire_server(_google) as wire, gateway.scenario() as scenario: + model: Final = _deployment(gateway, scenario, wire, provider) + rejected_description: Final = _description() + path, broken = _request_body("chat", model, rejected_description, _parameters(union), False) + failed: Final = gateway.request("POST", path, broken) + assert failed.status_code == 400, failed.text + assert "any_of[0]" in failed.text, failed.text + rejected: Final = _generate_requests(wire) + assert len(rejected) == 1, [request.target for request in rejected] + assert _sent_declarations(_body(rejected[0])) == [_declaration(rejected_description, sent)] + description: Final = _description() + _, valid = _request_body("chat", model, description, _parameters({"anyOf": [_NULL, _NULL, _STRING]}), False) + recovered: Final = gateway.request("POST", path, valid) + _assert_caller_saw_ok("chat", recovered, False) + requests: Final = _generate_requests(wire) + assert len(requests) == 1, [request.target for request in requests] + assert _sent_declarations(_body(requests[0])) == [_declaration(description, _NULLABLE_STRING)] + + +def test_a_repeated_identical_request_sends_the_same_folded_union_each_time(gateway: Gateway) -> None: + with wire_server(_google) as wire, gateway.scenario() as scenario: + model: Final = _deployment(gateway, scenario, wire, "vertex_ai") + description: Final = _description() + path, body = _request_body("chat", model, description, _parameters({"anyOf": [_NULL, _NULL, _STRING]}), False) + first: Final = gateway.request("POST", path, body) + second: Final = gateway.request("POST", path, body) + _assert_caller_saw_ok("chat", first, False) + _assert_caller_saw_ok("chat", second, False) + assert _json(first)["id"] != _json(second)["id"] + requests: Final = _generate_requests(wire) + assert len(requests) == 2, [request.target for request in requests] + assert [_sent_declarations(_body(request)) for request in requests] == [ + [_declaration(description, _NULLABLE_STRING)] + ] * 2 + + +def test_openai_sdk_chat_sends_the_folded_union(gateway: Gateway) -> None: + with wire_server(_google) as wire, gateway.scenario() as scenario: + model: Final = _deployment(gateway, scenario, wire, "gemini") + description: Final = _description() + client: Final = openai.OpenAI( + base_url=f"{str(gateway.client.base_url).rstrip('/')}/v1", api_key=gateway.key, max_retries=0 + ) + completion: Final = client.chat.completions.create( + model=model, + max_tokens=16, + messages=[{"role": "user", "content": "Record the value hello by calling the record tool."}], + tools=[_tool("chat", description, _parameters({"anyOf": [_NULL, _NULL, _STRING]}))], # pyright: ignore[reportArgumentType] # SDK TypedDicts reject a JsonValue mapping + ) + assert completion.choices[0].message.content == "OK" + requests: Final = _generate_requests(wire) + assert len(requests) == 1, [request.target for request in requests] + assert _sent_declarations(_body(requests[0])) == [_declaration(description, _NULLABLE_STRING)] + + +async def _async_openai_stream_text(gateway: Gateway, model: str, description: str) -> str: + client: Final = openai.AsyncOpenAI( + base_url=f"{str(gateway.client.base_url).rstrip('/')}/v1", api_key=gateway.key, max_retries=0 + ) + stream: Final = await client.chat.completions.create( + model=model, + max_tokens=16, + stream=True, + messages=[{"role": "user", "content": "Record the value hello by calling the record tool."}], + tools=[_tool("chat", description, _parameters({"anyOf": [_NULL]}))], # pyright: ignore[reportArgumentType] # SDK TypedDicts reject a JsonValue mapping + ) + return "".join([chunk.choices[0].delta.content or "" async for chunk in stream if chunk.choices]) + + +def test_async_openai_sdk_stream_sends_the_null_only_union_as_one_nullable_null(gateway: Gateway) -> None: + with wire_server(_google) as wire, gateway.scenario() as scenario: + model: Final = _deployment(gateway, scenario, wire, "vertex_ai") + description: Final = _description() + assert asyncio.run(_async_openai_stream_text(gateway, model, description)) == "OK" + requests: Final = _generate_requests(wire) + assert len(requests) == 1, [request.target for request in requests] + assert requests[0].target.split("?")[0].endswith(":streamGenerateContent"), requests[0].target + assert _sent_declarations(_body(requests[0])) == [_declaration(description, _NULLABLE_NULL)] + + +def test_anthropic_sdk_messages_sends_the_folded_union(gateway: Gateway) -> None: + with wire_server(_google) as wire, gateway.scenario() as scenario: + model: Final = _deployment(gateway, scenario, wire, "vertex_ai") + description: Final = _description() + client: Final = anthropic.Anthropic(base_url=str(gateway.client.base_url), api_key=gateway.key, max_retries=0) + message: Final = client.messages.create( + model=model, + max_tokens=16, + messages=[{"role": "user", "content": "Record the value hello by calling the record tool."}], + tools=[_tool("messages", description, _parameters({"anyOf": [_NULL, _NULL]}))], # pyright: ignore[reportArgumentType] # SDK TypedDicts reject a JsonValue mapping + ) + assert [block.type for block in message.content] == ["text"] + requests: Final = _generate_requests(wire) + assert len(requests) == 1, [request.target for request in requests] + assert _sent_declarations(_body(requests[0])) == [_declaration(description, _NULLABLE_NULL)] + + +@pytest.mark.parametrize( + ("union", "expected"), [case[1:] for case in _FANOUT_UNIONS], ids=[case[0] for case in _FANOUT_UNIONS] +) +def test_vertex_count_tokens_sends_the_folded_union(gateway: Gateway, union: JsonValue, expected: JsonValue) -> None: + with wire_server(_google) as wire, gateway.scenario() as scenario: + model: Final = _deployment(gateway, scenario, wire, "vertex_ai") + description: Final = _description() + response: Final = gateway.request( + "POST", + "/v1/messages/count_tokens", + { + "model": model, + "messages": [{"role": "user", "content": "Record the value hello by calling the record tool."}], + "tools": [_tool("messages", description, _parameters(union))], + }, + ) + assert response.status_code == 200, response.text + assert _json(response)["input_tokens"] == _COUNT, response.text + requests: Final = tuple(request for request in wire.drain() if request.target.endswith(":countTokens")) + assert len(requests) == 1, [request.target for request in requests] + assert _sent_declarations(_body(requests[0])) == [_declaration(description, expected)] + + +@pytest.mark.parametrize( + ("union", "expected"), [case[1:] for case in _FANOUT_UNIONS], ids=[case[0] for case in _FANOUT_UNIONS] +) +def test_gemini_token_counter_sends_the_folded_union(gateway: Gateway, union: JsonValue, expected: JsonValue) -> None: + with wire_server(_google) as wire, gateway.scenario() as scenario: + model: Final = _deployment(gateway, scenario, wire, "gemini") + description: Final = _description() + response: Final = gateway.request( + "POST", + "/utils/token_counter", + { + "model": model, + "contents": [ + {"role": "user", "parts": [{"text": "Record the value hello by calling the record tool."}]} + ], + "tools": [_tool("chat", description, _parameters(union))], + }, + params={"call_endpoint": "true"}, + ) + assert response.status_code == 200, response.text + assert _json(response)["total_tokens"] == _COUNT, response.text + requests: Final = tuple(request for request in wire.drain() if ":countTokens" in request.target) + assert len(requests) == 1, [request.target for request in requests] + sent: Final = object_value(_body(requests[0])["generateContentRequest"]) + assert _sent_declarations(sent) == [_declaration(description, expected)] + + +def test_response_format_on_a_model_without_json_schema_support_folds_the_union(gateway: Gateway) -> None: + with wire_server(_google) as wire, gateway.scenario() as scenario: + model: Final = _deployment(gateway, scenario, wire, "gemini", _NO_JSON_SCHEMA_BACKEND) + response: Final = gateway.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "messages": [{"role": "user", "content": "Return the value as JSON."}], + "response_format": { + "type": "json_schema", + "json_schema": {"name": "record", "schema": _parameters({"anyOf": [_NULL, _NULL, _STRING]})}, + }, + }, + ) + _assert_caller_saw_ok("chat", response, False) + requests: Final = _generate_requests(wire) + assert len(requests) == 1, [request.target for request in requests] + generation_config: Final = object_value(_body(requests[0])["generationConfig"]) + assert generation_config["response_mime_type"] == "application/json", generation_config + assert generation_config["response_schema"] == {**_parameters(_NULLABLE_STRING), "propertyOrdering": ["value"]} + + +def test_native_generate_content_response_schema_folds_the_union(gateway: Gateway) -> None: + with wire_server(_google) as wire, gateway.scenario() as scenario: + model: Final = _deployment(gateway, scenario, wire, "gemini", _NO_JSON_SCHEMA_BACKEND) + response: Final = gateway.request( + "POST", + f"/v1beta/models/{model}:generateContent", + { + "contents": [{"role": "user", "parts": [{"text": "Return the value as JSON."}]}], + "generationConfig": { + "responseMimeType": "application/json", + "responseSchema": _parameters({"anyOf": [_NULL, _NULL, _STRING]}), + }, + }, + ) + assert response.status_code == 200, response.text + requests: Final = _generate_requests(wire) + assert len(requests) == 1, [request.target for request in requests] + generation_config: Final = object_value(_body(requests[0])["generationConfig"]) + assert generation_config["responseSchema"] == {**_parameters(_NULLABLE_STRING), "propertyOrdering": ["value"]} + + +@pytest.mark.parametrize( + ("union", "expected"), [case[1:] for case in _FANOUT_UNIONS], ids=[case[0] for case in _FANOUT_UNIONS] +) +def test_image_generation_tools_send_the_folded_union(gateway: Gateway, union: JsonValue, expected: JsonValue) -> None: + with wire_server(_google) as wire, gateway.scenario() as scenario: + model: Final = _deployment(gateway, scenario, wire, "gemini", _IMAGE_BACKEND) + description: Final = _description() + response: Final = gateway.request( + "POST", + "/v1/images/generations", + { + "model": model, + "prompt": "A small red square on a white background", + "tools": [_tool("chat", description, _parameters(union))], + }, + ) + assert response.status_code == 200, response.text + assert len(_JSON_LIST.validate_python(_json(response)["data"])) == 1, response.text + requests: Final = _generate_requests(wire) + assert len(requests) == 1, [request.target for request in requests] + assert _sent_declarations(_body(requests[0])) == [_declaration(description, expected)] + + +def test_vertex_batch_row_is_stored_with_the_folded_union(gateway: Gateway) -> None: + with wire_server(_google) as wire, gateway.scenario() as scenario: + model: Final = vertex_model(gateway, scenario, wire.url, _BACKEND) + description: Final = _description() + row: Final[dict[str, JsonValue]] = { + "custom_id": "anyof-row", + "method": "POST", + "url": "/v1/chat/completions", + "body": { + "model": model, + "messages": [{"role": "user", "content": "Record the value hello by calling the record tool."}], + "tools": [_tool("chat", description, _parameters({"anyOf": [_NULL, _NULL, _STRING]}))], + }, + } + response: Final = upload(gateway, "model", model, jsonl(encoded(row))) + assert response.status_code == 200, response.text + uploads: Final = tuple(request for request in wire.drain() if is_gcs_upload(request)) + assert len(uploads) == 1, [request.target for request in uploads] + stored: Final = tuple( + object_value(JSON_OBJECT.validate_json(line)["request"]) for line in uploads[0].body.splitlines() if line + ) + assert len(stored) == 1, stored + assert _sent_declarations(stored[0]) == [_declaration(description, _NULLABLE_STRING)] + + +async def _event_types_until_done(socket: websockets.ClientConnection) -> AsyncIterator[str]: + async for message in socket: + kind: Final = string_value(JSON_OBJECT.validate_json(message)["type"]) + yield kind + if kind in {"response.done", "error"}: + return + + +async def _realtime_session_with_tools(gateway: Gateway, model: str, tools: list[JsonValue]) -> tuple[str, ...]: + ws_base: Final = str(gateway.client.base_url).rstrip("/").replace("http://", "ws://") + headers: Final = {"Authorization": f"Bearer {gateway.key}"} + async with websockets.connect(f"{ws_base}/v1/realtime?model={model}", additional_headers=headers) as socket: + first: Final = JSON_OBJECT.validate_json(await socket.recv()) + await socket.send(json.dumps({"type": "session.update", "session": {"tools": tools}})) + item: Final = {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "say the line"}]} + await socket.send(json.dumps({"type": "conversation.item.create", "item": item})) + events: Final = [kind async for kind in _event_types_until_done(socket)] + return (string_value(first["type"]), *events) + + +def _realtime_types(gateway: Gateway, model: str, tools: list[JsonValue]) -> tuple[str, ...]: + async def bounded() -> tuple[str, ...]: + try: + return await asyncio.wait_for(_realtime_session_with_tools(gateway, model, tools), 30) + except TimeoutError: + return ("timeout",) + + return asyncio.run(bounded()) + + +def _live_frames(gateway: Gateway, scenario_id: str) -> tuple[dict[str, JsonValue], ...]: + with httpx.Client(base_url=gateway.upstream_url, timeout=5, trust_env=False) as upstream: + observed: Final = tuple( + map(object_value, _JSON_LIST.validate_python(_json(upstream.get("/__observations"))["requests"])) + ) + return tuple( + object_value(request["body"]) + for request in observed + if request["api_key"] == scenario_id + and request["method"] == "WEBSOCKET_FRAME" + and request["path"] == GEMINI_LIVE_PATH + ) + + +@pytest.fixture(scope="module") +def deferred_live(tmp_path_factory: pytest.TempPathFactory) -> Iterator[OwnedProxy]: + directory: Final = tmp_path_factory.mktemp("anyof-live") + with gateway_from_environment() as gateway: + with owned_proxy_process(gateway, directory, {"LITELLM_GEMINI_LIVE_DEFER_SETUP": "true"}, workers=2) as proxy: + yield proxy + + +@pytest.mark.timeout(2 * graceful_stop_seconds() + 120) +def test_deferred_realtime_setup_carries_session_tools_with_the_folded_union(deferred_live: OwnedProxy) -> None: + gateway: Final = deferred_live.gateway + with gateway.scenario() as scenario: + handle: Final = live_scenario(scenario) + model: Final = live_deployment(scenario, handle.scenario_id, gateway.upstream_url) + description: Final = _description() + tools: Final[list[JsonValue]] = [_tool("chat", description, _parameters({"anyOf": [_NULL, _NULL, _STRING]}))] + types: Final = _realtime_types(gateway, model, tools) + assert types[0] == "session.created", types + assert types[-1] == "response.done", types + frames: Final = _live_frames(gateway, handle.scenario_id) + setups: Final = tuple(object_value(frame["setup"]) for frame in frames if "setup" in frame) + assert len(setups) == 1, [sorted(frame) for frame in frames] + assert _sent_declarations(setups[0]) == [_declaration(description, _NULLABLE_STRING)] + + +_PROVIDERS: Final[tuple[Provider, ...]] = ("vertex_ai", "gemini") +_ENDPOINTS: Final[tuple[Endpoint, ...]] = ("chat", "messages", "responses") +_BURST: Final[tuple[tuple[Provider, Endpoint, bool, bool], ...]] = tuple( + (provider, endpoint, stream, index % 4 == 0) + for index, (provider, endpoint, stream, _) in enumerate(product(_PROVIDERS, _ENDPOINTS, (False, True), range(2))) +) + + +def test_a_concurrent_mixed_burst_folds_every_union_once_and_keeps_each_status(gateway: Gateway) -> None: + with wire_server(_google) as wire, gateway.scenario() as scenario: + models: Final = MappingProxyType( + {provider: _deployment(gateway, scenario, wire, provider) for provider in _PROVIDERS} + ) + shots: Final = tuple( + (provider, endpoint, stream, f"{_FAIL_MARKER if refused else 'burst'} {uuid.uuid4().hex}") + for provider, endpoint, stream, refused in _BURST + ) + + def send(shot: tuple[Provider, Endpoint, bool, str]) -> httpx.Response: + provider, endpoint, stream, description = shot + path, body = _request_body( + endpoint, models[provider], description, _parameters({"anyOf": [_NULL, _NULL, _STRING]}), stream + ) + return gateway.request("POST", path, body) + + with ThreadPoolExecutor(max_workers=len(shots)) as pool: + responses: Final = tuple(pool.map(send, shots)) + expected_statuses: Final = [400 if _FAIL_MARKER in shot[3] else 200 for shot in shots] + assert [response.status_code for response in responses] == expected_statuses, [ + response.text[:120] for response in responses if response.status_code not in {200, 400} + ] + requests: Final = _generate_requests(wire) + assert len(requests) == len(shots), len(requests) + sent: Final = sorted((json.dumps(_sent_declarations(_body(request)), sort_keys=True) for request in requests)) + assert sent == sorted(json.dumps([_declaration(shot[3], _NULLABLE_STRING)], sort_keys=True) for shot in shots) diff --git a/tests/unit/llms/vertex_ai/test_vertex.py b/tests/unit/llms/vertex_ai/test_vertex.py index 044a07859f9..04cf391c700 100644 --- a/tests/unit/llms/vertex_ai/test_vertex.py +++ b/tests/unit/llms/vertex_ai/test_vertex.py @@ -1814,8 +1814,8 @@ def test_gemini_nullable_object_tool_schema_httpx( "type": "object", "required": ["source"], "properties": {"source": {"type": "string"}}, + "nullable": True, }, - {"type": "null"}, ] diff --git a/tests/unit/llms/vertex_ai/test_vertex_ai_common_utils.py b/tests/unit/llms/vertex_ai/test_vertex_ai_common_utils.py index 8528724a24e..d244fe6bd90 100644 --- a/tests/unit/llms/vertex_ai/test_vertex_ai_common_utils.py +++ b/tests/unit/llms/vertex_ai/test_vertex_ai_common_utils.py @@ -1,3 +1,4 @@ +import copy import time from collections.abc import Iterator from contextlib import contextmanager @@ -14,6 +15,7 @@ from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler from litellm.llms.gemini.count_tokens import handler as gemini_count_tokens_handler from litellm.llms.vertex_ai.common_utils import ( VertexAITokenCounter, + build_vertex_schema, convert_anyof_null_to_nullable, convert_vertex_datetime_to_openai_datetime, get_vertex_location_from_url, @@ -234,6 +236,147 @@ def test_nested_anyof_conversion(): assert schema == expected +def test_anyof_conversion_with_multiple_null_branches(): + """Adjacent null branches must all be removed.""" + schema: Final = { + "type": "object", + "properties": { + "example": { + "anyOf": [ + {"type": "null"}, + {"type": "null"}, + {"type": "string"}, + ] + } + }, + } + + convert_anyof_null_to_nullable(schema) + + expected: Final = { + "type": "object", + "properties": {"example": {"anyOf": [{"type": "string", "nullable": True}]}}, + } + assert schema == expected + + +def test_anyof_conversion_with_null_before_empty_object(): + """An empty-object branch following a null branch still gets a type.""" + schema: Final = { + "type": "object", + "properties": {"example": {"anyOf": [{"type": "null"}, {}]}}, + } + + convert_anyof_null_to_nullable(schema) + + expected: Final = { + "type": "object", + "properties": {"example": {"anyOf": [{"type": "object", "nullable": True}]}}, + } + assert schema == expected + + +@pytest.mark.parametrize("null_count", [1, 2, 4]) +def test_anyof_conversion_with_only_null_branches_keeps_one_nullable_null(null_count: int): + """Vertex accepts a NULL type, so a null-only union is forwarded as one nullable null branch.""" + schema: Final = { + "type": "object", + "properties": {"example": {"anyOf": [{"type": "null"} for _ in range(null_count)]}}, + } + + convert_anyof_null_to_nullable(schema) + + expected: Final = { + "type": "object", + "properties": {"example": {"anyOf": [{"type": "null", "nullable": True}]}}, + } + assert schema == expected + + +def test_anyof_conversion_leaves_empty_anyof_for_the_provider_to_judge(): + """An empty union has no null branch to strip, so it is forwarded unchanged.""" + schema: Final = {"type": "object", "properties": {"example": {"anyOf": []}}} + + convert_anyof_null_to_nullable(schema) + + assert schema == {"type": "object", "properties": {"example": {"anyOf": []}}} + + +@pytest.mark.parametrize("malformed", ["string", {"type": "string"}], ids=["string", "object"]) +def test_anyof_conversion_forwards_a_non_list_anyof_unchanged(malformed: JsonValue): + """A malformed union is the provider's to judge, as it was before null branches were stripped in one pass.""" + schema: Final = {"type": "object", "properties": {"example": {"anyOf": copy.deepcopy(malformed)}}} + + convert_anyof_null_to_nullable(schema) + + assert schema == {"type": "object", "properties": {"example": {"anyOf": malformed}}} + + +@pytest.mark.parametrize( + ("branches", "expected"), + [ + ([{"type": "null"}, 5], [5]), + ([True, {"type": "null"}], [True]), + ([{"type": "null"}, 5, {"type": "string"}], [5, {"type": "string", "nullable": True}]), + ], + ids=["null-int", "true-null", "null-int-string"], +) +def test_anyof_conversion_forwards_a_non_object_branch_for_the_provider_to_judge( + branches: list[JsonValue], expected: list[JsonValue] +): + schema: Final = {"type": "object", "properties": {"example": {"anyOf": branches}}} + + convert_anyof_null_to_nullable(schema) + + assert schema == {"type": "object", "properties": {"example": {"anyOf": expected}}} + + + +@pytest.mark.parametrize( + ("field", "expected"), + [ + ({"type": ["integer", "null"]}, {"anyOf": [{"type": "integer", "nullable": True}]}), + ( + { + "anyOf": [ + {"type": "object", "properties": {"x": {"anyOf": [{"type": "integer"}, {"type": "null"}]}}}, + {"type": "null"}, + ] + }, + { + "anyOf": [ + { + "type": "object", + "properties": {"x": {"anyOf": [{"type": "integer", "nullable": True}]}}, + "nullable": True, + } + ] + }, + ), + ( + {"type": ["object", "null"], "properties": {"x": {"type": ["string", "null"]}}}, + { + "anyOf": [ + { + "type": "object", + "properties": {"x": {"anyOf": [{"type": "string", "nullable": True}]}}, + "nullable": True, + } + ] + }, + ), + ( + {"type": "array", "items": {"type": ["number", "null"]}}, + {"type": "array", "items": {"anyOf": [{"type": "number", "nullable": True}]}}, + ), + ], + ids=["type-array", "null-inside-a-union-branch", "nullable-object-with-nullable-property", "nullable-items"], +) +def test_vertex_schema_folds_every_null_branch_into_nullable(field: JsonValue, expected: JsonValue): + schema: Final = build_vertex_schema({"type": "object", "properties": {"value": field}}) + + assert schema["properties"]["value"] == expected + def test_anyof_with_excessive_nesting(): """Test conversion with excessive nesting > max levels +1 deep.""" # generate a schema with excessive nesting