fix(moderations): stop the router image check from 500ing on a list input (#49454)

* fix(moderations): stop the router image check from 500ing on a list input

* fix(moderations): give the list cast its cast-ok reason

* fix(moderations): type the responses input guard and keep SDK compaction items

* fix(moderations): import the compaction item through litellm.types and mark the TypeIs import

* fix(moderations): sort the compaction item import

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Kerry Lu 2026-10-31 12:05:13 -07:00 • committed by GitHub
parent 941f52f26f
commit 6f86b117b5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 38 additions and 3 deletions

View file

@ -40,7 +40,11 @@ from typing import TYPE_CHECKING, Any, Final, NamedTuple, Protocol, TypeVar, Uni
from openai.types.responses.response_output_refusal import ResponseOutputRefusal
from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
from typing_extensions import ReadOnly, TypedDict
from typing_extensions import (
ReadOnly,
TypedDict,
TypeIs, # noqa: TID251 # TypeIs reaches typing only on 3.13
)
from litellm._logging import verbose_proxy_logger
from litellm.completion_extras.litellm_responses_transformation.transformation import (
@ -100,6 +104,7 @@ from litellm.types.llms.openai import (
OutputTextDeltaEvent,
OutputTextDoneEvent,
ResponseAPIUsage,
ResponseCompactionItem,
ResponseCompletedEvent,
ResponsesAPIOptionalRequestParams,
ResponsesAPIResponse,
@ -712,6 +717,15 @@ def _next_stream_sequence_number(responses_so_far: Sequence[object] | None) -> i
return next((n + 1 for n in sequence_numbers if isinstance(n, int)), 0)
def _is_responses_input(
input_data: object,
) -> TypeIs["str | ResponseInputParam"]: # guard-ok: str, or a list of Mapping items
if isinstance(input_data, str):
return True
items: Final = cast(list[object], input_data) # cast-ok: isinstance narrows to list[Unknown] only
return isinstance(input_data, list) and all(isinstance(item, (Mapping, ResponseCompactionItem)) for item in items)
class OpenAIResponsesHandler(BaseTranslation):
"""
Handler for processing OpenAI Responses API with guardrails.
@ -733,8 +747,8 @@ class OpenAIResponsesHandler(BaseTranslation):
Transforms `input` (string or ResponseInputParam) and optional
`instructions` into chat completion messages.
"""
input_data: Final = data.get("input")
if input_data is None:
input_data: Final[object] = data.get("input")
if not _is_responses_input(input_data):
return None
messages: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_input_to_messages(
input=input_data,

View file

@ -29,6 +29,7 @@ from openai.types.responses import (
ResponseOutputRefusal,
ResponseOutputText,
)
from openai.types.responses.response_compaction_item import ResponseCompactionItem
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@ -3349,6 +3350,25 @@ class TestGetStructuredMessages:
result = handler.get_structured_messages(data)
assert result is None
def test_should_return_none_for_a_moderation_style_list_of_strings(self):
handler = OpenAIResponsesHandler()
data = {"model": "omni-moderation-latest", "input": ["safe text", "more safe text"]}
assert handler.get_structured_messages(data) is None
def test_should_keep_an_image_item_replayed_with_an_sdk_compaction_item(self):
handler = OpenAIResponsesHandler()
image_url = "https://example.com/cat.png"
data = {
"model": "gpt-5.5",
"input": [
ResponseCompactionItem(type="compaction", id="cmp_1", encrypted_content="abc"),
{"role": "user", "content": [{"type": "input_image", "image_url": image_url}]},
],
}
result = handler.get_structured_messages(data)
assert result is not None
assert image_url in json.dumps(result)
class ToolAppendingGuardrail(CustomGuardrail):
"""Guardrail that appends a new function tool, mimicking a guardrail that

View file

@ -31445,6 +31445,7 @@ def test_request_carries_image_sees_chat_parts_and_responses_input_items():
assert litellm.Router._request_carries_image(None, {"input": _IMAGE_RESPONSES_INPUT}) is True
assert litellm.Router._request_carries_image([{"role": "user", "content": "Say hi"}], {}) is False
assert litellm.Router._request_carries_image(None, {"input": "Say hi"}) is False
assert litellm.Router._request_carries_image(None, {"input": ["safe text", "more safe text"]}) is False
@pytest.mark.asyncio