diff --git a/litellm/__init__.py b/litellm/__init__.py index a33a78fb325..a48cedcff3a 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -2277,6 +2277,7 @@ if TYPE_CHECKING: supports_parallel_function_calling: Callable[..., bool] supports_vision: Callable[..., bool] supports_audio_input: Callable[..., bool] + supports_video_input: Callable[..., bool] supports_audio_output: Callable[..., bool] supports_system_messages: Callable[..., bool] supports_reasoning: Callable[..., bool] diff --git a/litellm/_lazy_imports_registry.py b/litellm/_lazy_imports_registry.py index ce69de3b240..0e7cdd115b3 100644 --- a/litellm/_lazy_imports_registry.py +++ b/litellm/_lazy_imports_registry.py @@ -36,6 +36,7 @@ UTILS_NAMES: Final = ( "supports_parallel_function_calling", "supports_vision", "supports_audio_input", + "supports_video_input", "supports_audio_output", "supports_system_messages", "supports_reasoning", @@ -481,6 +482,7 @@ _UTILS_IMPORT_MAP: Final = { ), "supports_vision": (".utils", "supports_vision"), "supports_audio_input": (".utils", "supports_audio_input"), + "supports_video_input": (".utils", "supports_video_input"), "supports_audio_output": (".utils", "supports_audio_output"), "supports_system_messages": (".utils", "supports_system_messages"), "supports_reasoning": (".utils", "supports_reasoning"), diff --git a/litellm/decisions/call.py b/litellm/decisions/call.py index 1ad2f006736..d2be14e5529 100644 --- a/litellm/decisions/call.py +++ b/litellm/decisions/call.py @@ -19,11 +19,13 @@ from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler from litellm.llms.openai.decisions.transformation import ir_to_openai_response, openai_request_to_ir from litellm.types.decisions import ( DecisionQuestion, + DecisionsAudio, DecisionsIRRequest, DecisionsIRResponse, DecisionsJSON, DecisionsRequestBody, DecisionsResponse, + DecisionsVideo, OpenAIDecisionInput, OpenAIDecisionQuestion, OpenAIDecisionRequestBody, @@ -35,6 +37,8 @@ from litellm.utils import ProviderConfigManager SystemOneQuestions: TypeAlias = Mapping[str, DecisionQuestion | Mapping[str, object]] OpenAIDecisionQuestions: TypeAlias = Sequence[OpenAIDecisionQuestion | Mapping[str, object]] +DecisionsAudioClips: TypeAlias = Sequence[DecisionsAudio | Mapping[str, object]] +DecisionsVideoClips: TypeAlias = Sequence[DecisionsVideo | Mapping[str, object]] DecisionsRequestFormat: TypeAlias = DecisionsRequestBody | OpenAIDecisionRequestBody _SYSTEMONE_REQUEST_ADAPTER: Final[TypeAdapter[DecisionsRequestBody]] = TypeAdapter(DecisionsRequestBody) @@ -49,19 +53,31 @@ _HANDLER: Final = BaseLLMHTTPHandler() class SystemOneRequestFields: state: DecisionsJSON | None questions: SystemOneQuestions | None + audio: DecisionsAudioClips | None = None + videos: DecisionsVideoClips | None = None def validate(self, safety_identifier: str | None) -> DecisionsRequestBody: - return _SYSTEMONE_REQUEST_ADAPTER.validate_python({"state": self.state, "questions": self.questions}) + return _SYSTEMONE_REQUEST_ADAPTER.validate_python( + {"state": self.state, "questions": self.questions, "audio": self.audio, "videos": self.videos} + ) @dataclass(frozen=True, slots=True) class OpenAIDecisionRequestFields: input: OpenAIDecisionInput | None questions: OpenAIDecisionQuestions | None + audio: DecisionsAudioClips | None = None + videos: DecisionsVideoClips | None = None def validate(self, safety_identifier: str | None) -> OpenAIDecisionRequestBody: return _OPENAI_REQUEST_ADAPTER.validate_python( - {"input": self.input, "questions": self.questions, "safety_identifier": safety_identifier} + { + "input": self.input, + "questions": self.questions, + "safety_identifier": safety_identifier, + "audio": self.audio, + "videos": self.videos, + } ) @@ -122,15 +138,49 @@ def _drops_params(kwargs: Mapping[str, object]) -> bool: return litellm.drop_params is True or normalize_drop_params(kwargs.get("drop_params")) is True -def _request_safety_identifier(safety_identifier: object, kwargs: Mapping[str, object]) -> str | None: - if isinstance(safety_identifier, str) or not _drops_params(kwargs): +def _request_safety_identifier( + safety_identifier: object, provider_config: BaseDecisionsConfig, kwargs: Mapping[str, object] +) -> str | None: + if not _drops_params(kwargs): return _SAFETY_IDENTIFIER_ADAPTER.validate_python(safety_identifier) + if isinstance(safety_identifier, str) and provider_config.supports_safety_identifier: + return safety_identifier return None +def _fields_without_unsupported_media( + fields: RequestFields, + provider_config: BaseDecisionsConfig, + model: str, + custom_llm_provider: str, + kwargs: Mapping[str, object], +) -> RequestFields: + if not _drops_params(kwargs): + return fields + return replace( + fields, + audio=fields.audio if provider_config.supports_audio_input(model, custom_llm_provider) else None, + videos=fields.videos if provider_config.supports_video_input(model, custom_llm_provider) else None, + ) + + @dataclass(frozen=True, slots=True) -class _UnsupportedSafetyIdentifier: - pass +class _UnsupportedParams: + names: tuple[str, ...] + + +def _unsupported_param_names( + ir_request: DecisionsIRRequest, provider_config: BaseDecisionsConfig, model: str, custom_llm_provider: str +) -> tuple[str, ...]: + checks: Final = ( + ( + "safety_identifier", + ir_request.safety_identifier is not None and not provider_config.supports_safety_identifier, + ), + ("audio", bool(ir_request.audio) and not provider_config.supports_audio_input(model, custom_llm_provider)), + ("videos", bool(ir_request.videos) and not provider_config.supports_video_input(model, custom_llm_provider)), + ) + return tuple(name for name, unsupported in checks if unsupported) def _provider_ir_request( @@ -138,14 +188,14 @@ def _provider_ir_request( *, safety_identifier: str | None, provider_config: BaseDecisionsConfig, - kwargs: Mapping[str, object], -) -> DecisionsIRRequest | _UnsupportedSafetyIdentifier: + model: str, + custom_llm_provider: str, +) -> DecisionsIRRequest | _UnsupportedParams: ir_request: Final = _ir_request(request, safety_identifier) - if ir_request.safety_identifier is None or provider_config.supports_safety_identifier: - return ir_request - if _drops_params(kwargs): - return replace(ir_request, safety_identifier=None) - return _UnsupportedSafetyIdentifier() + unsupported: Final = _unsupported_param_names(ir_request, provider_config, model, custom_llm_provider) + if unsupported: + return _UnsupportedParams(unsupported) + return ir_request def prepare_call( @@ -178,8 +228,10 @@ def prepare_call( except ValueError as error: raise litellm.BadRequestError(message=str(error), model=model, llm_provider=provider) from error try: - request_safety_identifier: Final = _request_safety_identifier(safety_identifier, kwargs) - request: Final = fields.validate(request_safety_identifier) + request_safety_identifier: Final = _request_safety_identifier(safety_identifier, provider_config, kwargs) + request: Final = _fields_without_unsupported_media( + fields, provider_config, canonical_model, provider, kwargs + ).validate(request_safety_identifier) except ValidationError as error: raise litellm.BadRequestError( message=f"Invalid Decisions request: {error}", @@ -206,12 +258,13 @@ def prepare_call( request, safety_identifier=request_safety_identifier, provider_config=provider_config, - kwargs=kwargs, + model=canonical_model, + custom_llm_provider=provider, ) - if isinstance(ir_request, _UnsupportedSafetyIdentifier): + if isinstance(ir_request, _UnsupportedParams): raise litellm.UnsupportedParamsError( message=( - f"{provider} does not support parameters: ['safety_identifier'], for model={model}. " + f"{provider} does not support parameters: {list(ir_request.names)}, for model={model}. " "To drop these, set `litellm.drop_params=True` or for proxy:\n\n" "`litellm_settings:\n drop_params: true`\n" ), diff --git a/litellm/decisions/main.py b/litellm/decisions/main.py index 8b13d04770a..760294d1c2b 100644 --- a/litellm/decisions/main.py +++ b/litellm/decisions/main.py @@ -6,7 +6,9 @@ import httpx from pydantic import TypeAdapter from litellm.decisions.call import ( + DecisionsAudioClips, DecisionsCall, + DecisionsVideoClips, OpenAIDecisionQuestions, OpenAIDecisionRequestFields, RequestFields, @@ -29,11 +31,16 @@ _OPENAI_QUESTIONS_ADAPTER: Final[TypeAdapter[OpenAIDecisionQuestions | None]] = def _request_fields( decision_input: OpenAIDecisionInput | None, questions: OpenAIDecisionQuestions | SystemOneQuestions | None, + audio: DecisionsAudioClips | None, + videos: DecisionsVideoClips | None, kwargs: Mapping[str, object], ) -> RequestFields: if "state" not in kwargs: return OpenAIDecisionRequestFields( - input=decision_input, questions=_OPENAI_QUESTIONS_ADAPTER.validate_python(questions) + input=decision_input, + questions=_OPENAI_QUESTIONS_ADAPTER.validate_python(questions), + audio=audio, + videos=videos, ) warnings.warn( "litellm.decisions(state=...) is deprecated, call litellm.systemone for the System One request format", @@ -43,6 +50,8 @@ def _request_fields( return SystemOneRequestFields( state=_STATE_ADAPTER.validate_python(kwargs["state"]), questions=_SYSTEMONE_QUESTIONS_ADAPTER.validate_python(questions), + audio=audio, + videos=videos, ) @@ -52,6 +61,8 @@ def _prepare_decisions_call( decision_input: OpenAIDecisionInput | None, questions: OpenAIDecisionQuestions | SystemOneQuestions | None, safety_identifier: str | None, + audio: DecisionsAudioClips | None, + videos: DecisionsVideoClips | None, api_key: str | None, api_base: str | None, timeout: float | httpx.Timeout | None, @@ -61,7 +72,7 @@ def _prepare_decisions_call( ) -> DecisionsCall: return prepare_call( model=model, - fields=_request_fields(decision_input, questions, kwargs), + fields=_request_fields(decision_input, questions, audio, videos, kwargs), safety_identifier=safety_identifier, api_key=api_key, api_base=api_base, @@ -78,6 +89,8 @@ async def adecisions( input: OpenAIDecisionInput | None = None, questions: OpenAIDecisionQuestions | SystemOneQuestions | None = None, safety_identifier: str | None = None, + audio: DecisionsAudioClips | None = None, + videos: DecisionsVideoClips | None = None, api_key: str | None = None, api_base: str | None = None, timeout: float | httpx.Timeout | None = None, @@ -90,6 +103,8 @@ async def adecisions( decision_input=input, questions=questions, safety_identifier=safety_identifier, + audio=audio, + videos=videos, api_key=api_key, api_base=api_base, timeout=timeout, @@ -106,6 +121,8 @@ def decisions( input: OpenAIDecisionInput | None = None, questions: OpenAIDecisionQuestions | SystemOneQuestions | None = None, safety_identifier: str | None = None, + audio: DecisionsAudioClips | None = None, + videos: DecisionsVideoClips | None = None, api_key: str | None = None, api_base: str | None = None, timeout: float | httpx.Timeout | None = None, @@ -118,6 +135,8 @@ def decisions( decision_input=input, questions=questions, safety_identifier=safety_identifier, + audio=audio, + videos=videos, api_key=api_key, api_base=api_base, timeout=timeout, diff --git a/litellm/llms/base_llm/decisions/transformation.py b/litellm/llms/base_llm/decisions/transformation.py index 724ca23c2ab..26d880c0d8a 100644 --- a/litellm/llms/base_llm/decisions/transformation.py +++ b/litellm/llms/base_llm/decisions/transformation.py @@ -8,6 +8,7 @@ import httpx from pydantic import TypeAdapter, ValidationError from typing_extensions import assert_never +import litellm from litellm.llms.base_llm.chat.transformation import BaseLLMException from litellm.secret_managers.main import get_secret_str from litellm.types.decisions import ( @@ -100,6 +101,8 @@ def _ir_question(name: str, question: DecisionQuestion) -> DecisionsIRQuestion: def systemone_request_to_ir(request: DecisionsRequestBody) -> DecisionsIRRequest: return DecisionsIRRequest( input=DecisionsIRState(state=request.state), + audio=request.audio, + videos=request.videos, questions=tuple(_ir_question(name, question) for name, question in request.questions.items()), ) @@ -166,8 +169,12 @@ def _systemone_question(question: DecisionsIRQuestion) -> Mapping[str, object]: def ir_to_systemone_request( - model: str, request: DecisionsIRRequest + model: str, request: DecisionsIRRequest, *, supports_audio: bool = False, supports_video: bool = False ) -> Mapping[str, object] | UnsupportedDecisionsRequest: + if request.audio and not supports_audio: + return UnsupportedDecisionsRequest(reason="audio input is not supported by this Decisions model") + if request.videos and not supports_video: + return UnsupportedDecisionsRequest(reason="video input is not supported by this Decisions model") state: Final = _systemone_state(request.input) if isinstance(state, UnsupportedDecisionsRequest): return state @@ -175,6 +182,16 @@ def ir_to_systemone_request( return { "model": model, "state": state, + **( + {"audio": [clip if isinstance(clip, str) else clip.model_dump() for clip in request.audio]} + if request.audio + else {} + ), + **( + {"videos": [clip if isinstance(clip, str) else clip.model_dump() for clip in request.videos]} + if request.videos + else {} + ), "questions": {key: _systemone_question(question) for key, question in keyed_questions}, } @@ -355,13 +372,24 @@ class BaseDecisionsConfig(ABC): ) -> tuple[Mapping[str, str], bytes | None]: return headers, None + def supports_audio_input(self, model: str, custom_llm_provider: str) -> bool: + return litellm.supports_audio_input(model=model, custom_llm_provider=custom_llm_provider) + + def supports_video_input(self, model: str, custom_llm_provider: str) -> bool: + return litellm.supports_video_input(model=model, custom_llm_provider=custom_llm_provider) + def transform_decisions_request( self, model: str, request: DecisionsIRRequest, custom_llm_provider: str, ) -> Mapping[str, object] | UnsupportedDecisionsRequest: - return ir_to_systemone_request(self.request_model(model), request) + return ir_to_systemone_request( + self.request_model(model), + request, + supports_audio=self.supports_audio_input(model, custom_llm_provider), + supports_video=self.supports_video_input(model, custom_llm_provider), + ) def unwrap_response(self, payload: object) -> object: return payload diff --git a/litellm/llms/openai/decisions/transformation.py b/litellm/llms/openai/decisions/transformation.py index 317e5abeeb5..a8cd50faafe 100644 --- a/litellm/llms/openai/decisions/transformation.py +++ b/litellm/llms/openai/decisions/transformation.py @@ -88,6 +88,8 @@ def _ir_question(question: OpenAIDecisionQuestion) -> DecisionsIRQuestion: def openai_request_to_ir(request: OpenAIDecisionRequestBody) -> DecisionsIRRequest: return DecisionsIRRequest( input=_ir_input(request.input), + audio=request.audio, + videos=request.videos, questions=tuple(_ir_question(question) for question in request.questions), safety_identifier=request.safety_identifier, ) @@ -164,6 +166,8 @@ def _has_one_option(question: DecisionsIRQuestion) -> bool: def ir_to_openai_request(model: str, request: DecisionsIRRequest) -> Mapping[str, object] | UnsupportedDecisionsRequest: + if request.audio or request.videos: + return UnsupportedDecisionsRequest(reason="audio/video input is not supported by this Decisions provider") if any(_has_one_option(question) for question in request.questions): return _SINGLE_OPTION return { diff --git a/litellm/model_prices_and_context_window_backup.json b/litellm/model_prices_and_context_window_backup.json index 780adab42ab..2427758877f 100644 --- a/litellm/model_prices_and_context_window_backup.json +++ b/litellm/model_prices_and_context_window_backup.json @@ -15982,6 +15982,26 @@ "prompt_cache_min_tokens": 1024, "source": "https://platform.claude.com/docs/en/about-claude/pricing" }, + "cloudflare/clef-omni": { + "input_cost_per_token": 1.5e-07, + "litellm_provider": "cloudflare", + "max_input_tokens": 64000, + "mode": "evaluation", + "output_cost_per_token": 0.0, + "source": "https://developers.cloudflare.com/workers-ai/models/clef-omni/", + "supports_audio_input": true, + "supports_video_input": true + }, + "cloudflare/@cf/cloudflare/clef-omni": { + "input_cost_per_token": 1.5e-07, + "litellm_provider": "cloudflare", + "max_input_tokens": 64000, + "mode": "evaluation", + "output_cost_per_token": 0.0, + "source": "https://developers.cloudflare.com/workers-ai/models/clef-omni/", + "supports_audio_input": true, + "supports_video_input": true + }, "cloudflare/clef": { "input_cost_per_token": 2.4e-07, "litellm_provider": "cloudflare", diff --git a/litellm/proxy/_lazy_openapi_snapshot.json b/litellm/proxy/_lazy_openapi_snapshot.json index b04a958cd03..4d428361264 100644 --- a/litellm/proxy/_lazy_openapi_snapshot.json +++ b/litellm/proxy/_lazy_openapi_snapshot.json @@ -10040,6 +10040,44 @@ "schema": { "additionalProperties": true, "properties": { + "audio": { + "default": [], + "description": "Clef-omni only. Up to 4 embedded audio clips, each a base64 data URL or {content_type, base64}", + "items": { + "anyOf": [ + { + "pattern": "^data:audio/[^;,\\s]+;base64,.+$", + "type": "string" + }, + { + "additionalProperties": false, + "properties": { + "base64": { + "description": "Base64-encoded audio bytes", + "minLength": 1, + "title": "Base64", + "type": "string" + }, + "content_type": { + "description": "An audio/* media type, such as audio/wav", + "pattern": "^audio/\\S+$", + "title": "Content Type", + "type": "string" + } + }, + "required": [ + "content_type", + "base64" + ], + "title": "DecisionsBase64Audio", + "type": "object" + } + ] + }, + "maxItems": 4, + "title": "Audio", + "type": "array" + }, "input": { "anyOf": [ { @@ -10345,6 +10383,44 @@ ], "description": "A stable id for your end user. Sent to OpenAI and dropped for other providers", "title": "Safety Identifier" + }, + "videos": { + "default": [], + "description": "Clef-omni only. Up to 2 embedded videos, each a base64 data URL or {content_type, base64}", + "items": { + "anyOf": [ + { + "pattern": "^data:video/[^;,\\s]+;base64,.+$", + "type": "string" + }, + { + "additionalProperties": false, + "properties": { + "base64": { + "description": "Base64-encoded video bytes", + "minLength": 1, + "title": "Base64", + "type": "string" + }, + "content_type": { + "description": "A video/* media type, such as video/mp4", + "pattern": "^video/\\S+$", + "title": "Content Type", + "type": "string" + } + }, + "required": [ + "content_type", + "base64" + ], + "title": "DecisionsBase64Video", + "type": "object" + } + ] + }, + "maxItems": 2, + "title": "Videos", + "type": "array" } }, "required": [ @@ -10419,6 +10495,44 @@ "schema": { "additionalProperties": true, "properties": { + "audio": { + "default": [], + "description": "Clef-omni only. Up to 4 embedded audio clips, each a base64 data URL or {content_type, base64}", + "items": { + "anyOf": [ + { + "pattern": "^data:audio/[^;,\\s]+;base64,.+$", + "type": "string" + }, + { + "additionalProperties": false, + "properties": { + "base64": { + "description": "Base64-encoded audio bytes", + "minLength": 1, + "title": "Base64", + "type": "string" + }, + "content_type": { + "description": "An audio/* media type, such as audio/wav", + "pattern": "^audio/\\S+$", + "title": "Content Type", + "type": "string" + } + }, + "required": [ + "content_type", + "base64" + ], + "title": "DecisionsBase64Audio", + "type": "object" + } + ] + }, + "maxItems": 4, + "title": "Audio", + "type": "array" + }, "model": { "description": "A decision model from the proxy model_list", "title": "Model", @@ -10647,6 +10761,44 @@ ], "description": "The thing being judged, such as a support ticket, a document or a chat transcript", "title": "State" + }, + "videos": { + "default": [], + "description": "Clef-omni only. Up to 2 embedded videos, each a base64 data URL or {content_type, base64}", + "items": { + "anyOf": [ + { + "pattern": "^data:video/[^;,\\s]+;base64,.+$", + "type": "string" + }, + { + "additionalProperties": false, + "properties": { + "base64": { + "description": "Base64-encoded video bytes", + "minLength": 1, + "title": "Base64", + "type": "string" + }, + "content_type": { + "description": "A video/* media type, such as video/mp4", + "pattern": "^video/\\S+$", + "title": "Content Type", + "type": "string" + } + }, + "required": [ + "content_type", + "base64" + ], + "title": "DecisionsBase64Video", + "type": "object" + } + ] + }, + "maxItems": 2, + "title": "Videos", + "type": "array" } }, "required": [ @@ -10736,6 +10888,44 @@ "schema": { "additionalProperties": true, "properties": { + "audio": { + "default": [], + "description": "Clef-omni only. Up to 4 embedded audio clips, each a base64 data URL or {content_type, base64}", + "items": { + "anyOf": [ + { + "pattern": "^data:audio/[^;,\\s]+;base64,.+$", + "type": "string" + }, + { + "additionalProperties": false, + "properties": { + "base64": { + "description": "Base64-encoded audio bytes", + "minLength": 1, + "title": "Base64", + "type": "string" + }, + "content_type": { + "description": "An audio/* media type, such as audio/wav", + "pattern": "^audio/\\S+$", + "title": "Content Type", + "type": "string" + } + }, + "required": [ + "content_type", + "base64" + ], + "title": "DecisionsBase64Audio", + "type": "object" + } + ] + }, + "maxItems": 4, + "title": "Audio", + "type": "array" + }, "input": { "anyOf": [ { @@ -11041,6 +11231,44 @@ ], "description": "A stable id for your end user. Sent to OpenAI and dropped for other providers", "title": "Safety Identifier" + }, + "videos": { + "default": [], + "description": "Clef-omni only. Up to 2 embedded videos, each a base64 data URL or {content_type, base64}", + "items": { + "anyOf": [ + { + "pattern": "^data:video/[^;,\\s]+;base64,.+$", + "type": "string" + }, + { + "additionalProperties": false, + "properties": { + "base64": { + "description": "Base64-encoded video bytes", + "minLength": 1, + "title": "Base64", + "type": "string" + }, + "content_type": { + "description": "A video/* media type, such as video/mp4", + "pattern": "^video/\\S+$", + "title": "Content Type", + "type": "string" + } + }, + "required": [ + "content_type", + "base64" + ], + "title": "DecisionsBase64Video", + "type": "object" + } + ] + }, + "maxItems": 2, + "title": "Videos", + "type": "array" } }, "required": [ @@ -11115,6 +11343,44 @@ "schema": { "additionalProperties": true, "properties": { + "audio": { + "default": [], + "description": "Clef-omni only. Up to 4 embedded audio clips, each a base64 data URL or {content_type, base64}", + "items": { + "anyOf": [ + { + "pattern": "^data:audio/[^;,\\s]+;base64,.+$", + "type": "string" + }, + { + "additionalProperties": false, + "properties": { + "base64": { + "description": "Base64-encoded audio bytes", + "minLength": 1, + "title": "Base64", + "type": "string" + }, + "content_type": { + "description": "An audio/* media type, such as audio/wav", + "pattern": "^audio/\\S+$", + "title": "Content Type", + "type": "string" + } + }, + "required": [ + "content_type", + "base64" + ], + "title": "DecisionsBase64Audio", + "type": "object" + } + ] + }, + "maxItems": 4, + "title": "Audio", + "type": "array" + }, "model": { "description": "A decision model from the proxy model_list", "title": "Model", @@ -11343,6 +11609,44 @@ ], "description": "The thing being judged, such as a support ticket, a document or a chat transcript", "title": "State" + }, + "videos": { + "default": [], + "description": "Clef-omni only. Up to 2 embedded videos, each a base64 data URL or {content_type, base64}", + "items": { + "anyOf": [ + { + "pattern": "^data:video/[^;,\\s]+;base64,.+$", + "type": "string" + }, + { + "additionalProperties": false, + "properties": { + "base64": { + "description": "Base64-encoded video bytes", + "minLength": 1, + "title": "Base64", + "type": "string" + }, + "content_type": { + "description": "A video/* media type, such as video/mp4", + "pattern": "^video/\\S+$", + "title": "Content Type", + "type": "string" + } + }, + "required": [ + "content_type", + "base64" + ], + "title": "DecisionsBase64Video", + "type": "object" + } + ] + }, + "maxItems": 2, + "title": "Videos", + "type": "array" } }, "required": [ diff --git a/litellm/proxy/decisions_endpoints/endpoints.py b/litellm/proxy/decisions_endpoints/endpoints.py index dbb7b69bd8b..6b6b7655719 100644 --- a/litellm/proxy/decisions_endpoints/endpoints.py +++ b/litellm/proxy/decisions_endpoints/endpoints.py @@ -92,8 +92,11 @@ async def _invalid_request( ) +_FIELDS_CHECKED_AFTER_ROUTING: Final = frozenset({"safety_identifier", "audio", "videos"}) + + def _fields_checked_before_routing(data: Mapping[str, object]) -> Mapping[str, object]: - return MappingProxyType({key: value for key, value in data.items() if key != "safety_identifier"}) + return MappingProxyType({key: value for key, value in data.items() if key not in _FIELDS_CHECKED_AFTER_ROUTING}) async def _request_data(request: Request, user_api_key_dict: UserAPIKeyAuth) -> dict[str, object]: diff --git a/litellm/systemone/main.py b/litellm/systemone/main.py index 754b69142f9..4b17f06b562 100644 --- a/litellm/systemone/main.py +++ b/litellm/systemone/main.py @@ -4,7 +4,9 @@ from typing import Final import httpx from litellm.decisions.call import ( + DecisionsAudioClips, DecisionsCall, + DecisionsVideoClips, SystemOneQuestions, SystemOneRequestFields, asend, @@ -22,6 +24,8 @@ def _prepare_systemone_call( state: DecisionsJSON | None, questions: SystemOneQuestions | None, safety_identifier: str | None, + audio: DecisionsAudioClips | None, + videos: DecisionsVideoClips | None, api_key: str | None, api_base: str | None, timeout: float | httpx.Timeout | None, @@ -32,7 +36,7 @@ def _prepare_systemone_call( reject_other_format(model=model, kwargs=kwargs, field_name="input", other_function="decisions") return prepare_call( model=model, - fields=SystemOneRequestFields(state=state, questions=questions), + fields=SystemOneRequestFields(state=state, questions=questions, audio=audio, videos=videos), safety_identifier=safety_identifier, api_key=api_key, api_base=api_base, @@ -59,6 +63,8 @@ async def asystemone( custom_llm_provider: str | None = None, extra_headers: Mapping[str, str] | None = None, safety_identifier: str | None = None, + audio: DecisionsAudioClips | None = None, + videos: DecisionsVideoClips | None = None, **kwargs: object, # kwargs-ok: litellm_params forwarded by @client ) -> DecisionsResponse: call: Final = _prepare_systemone_call( @@ -66,6 +72,8 @@ async def asystemone( state=state, questions=questions, safety_identifier=safety_identifier, + audio=audio, + videos=videos, api_key=api_key, api_base=api_base, timeout=timeout, @@ -87,6 +95,8 @@ def systemone( custom_llm_provider: str | None = None, extra_headers: Mapping[str, str] | None = None, safety_identifier: str | None = None, + audio: DecisionsAudioClips | None = None, + videos: DecisionsVideoClips | None = None, **kwargs: object, # kwargs-ok: litellm_params forwarded by @client ) -> DecisionsResponse: call: Final = _prepare_systemone_call( @@ -94,6 +104,8 @@ def systemone( state=state, questions=questions, safety_identifier=safety_identifier, + audio=audio, + videos=videos, api_key=api_key, api_base=api_base, timeout=timeout, diff --git a/litellm/types/decisions.py b/litellm/types/decisions.py index 3bc90a5109a..bef61bee40d 100644 --- a/litellm/types/decisions.py +++ b/litellm/types/decisions.py @@ -3,7 +3,7 @@ from dataclasses import dataclass, field from types import MappingProxyType from typing import Annotated, Final, Literal, TypeAlias -from pydantic import ConfigDict, Field, PrivateAttr, model_validator, with_config +from pydantic import ConfigDict, Field, PrivateAttr, field_validator, model_validator, with_config from typing_extensions import ReadOnly, Required, TypedDict from litellm.types.llms.base import LiteLLMPydanticObjectBase @@ -67,7 +67,49 @@ DecisionQuestionMap: TypeAlias = Annotated[ ] -class DecisionsRequestBody(LiteLLMPydanticObjectBase): +class DecisionsBase64Audio(LiteLLMPydanticObjectBase): + content_type: Annotated[str, Field(pattern=r"^audio/\S+$", description="An audio/* media type, such as audio/wav")] + base64: Annotated[str, Field(min_length=1, description="Base64-encoded audio bytes")] + + model_config = ConfigDict(extra="forbid", frozen=True) + + +class DecisionsBase64Video(LiteLLMPydanticObjectBase): + content_type: Annotated[str, Field(pattern=r"^video/\S+$", description="A video/* media type, such as video/mp4")] + base64: Annotated[str, Field(min_length=1, description="Base64-encoded video bytes")] + + model_config = ConfigDict(extra="forbid", frozen=True) + + +DecisionsAudio: TypeAlias = Annotated[str, Field(pattern=r"^data:audio/[^;,\s]+;base64,.+$")] | DecisionsBase64Audio +DecisionsVideo: TypeAlias = Annotated[str, Field(pattern=r"^data:video/[^;,\s]+;base64,.+$")] | DecisionsBase64Video + + +class DecisionsMediaBody(LiteLLMPydanticObjectBase): + audio: Annotated[ + tuple[DecisionsAudio, ...], + Field( + max_length=4, + description="Clef-omni only. Up to 4 embedded audio clips, each a base64 data URL or {content_type, base64}", + ), + ] = () + videos: Annotated[ + tuple[DecisionsVideo, ...], + Field( + max_length=2, + description="Clef-omni only. Up to 2 embedded videos, each a base64 data URL or {content_type, base64}", + ), + ] = () + + model_config = ConfigDict(extra="allow", frozen=True) + + @field_validator("audio", "videos", mode="before") + @classmethod + def _null_means_no_clips(cls, value: object) -> object: + return () if value is None else value + + +class DecisionsRequestBody(DecisionsMediaBody): state: Annotated[ DecisionsJSON, Field(description="The thing being judged, such as a support ticket, a document or a chat transcript"), @@ -89,6 +131,8 @@ class DecisionsCallParams(TypedDict, total=False): model: Required[ReadOnly[str]] state: Required[ReadOnly[DecisionsJSON]] questions: Required[ReadOnly[DecisionQuestionMap]] + audio: ReadOnly[Sequence[DecisionsAudio]] + videos: ReadOnly[Sequence[DecisionsVideo]] api_key: ReadOnly[str | None] api_base: ReadOnly[str | None] timeout: ReadOnly[float | None] @@ -255,7 +299,7 @@ OpenAIDecisionQuestion: TypeAlias = Annotated[ ] -class OpenAIDecisionRequestBody(LiteLLMPydanticObjectBase): +class OpenAIDecisionRequestBody(DecisionsMediaBody): input: Annotated[ OpenAIDecisionInput, Field( @@ -426,6 +470,8 @@ class DecisionsIRRequest: input: DecisionsIRState | DecisionsIRMessages questions: tuple[DecisionsIRQuestion, ...] safety_identifier: str | None = None + audio: tuple[DecisionsAudio, ...] = () + videos: tuple[DecisionsVideo, ...] = () @dataclass(frozen=True, slots=True) diff --git a/litellm/types/utils.py b/litellm/types/utils.py index f7385a6d92d..c486fc4f4d0 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -195,6 +195,7 @@ class ProviderSpecificModelInfo(TypedDict, total=False): supports_thinking_cache_preservation: ReadOnly[bool | None] supports_computer_use: bool | None supports_audio_input: bool | None + supports_video_input: ReadOnly[bool | None] supports_embedding_image_input: bool | None supports_audio_output: bool | None supports_pdf_input: bool | None diff --git a/litellm/utils.py b/litellm/utils.py index 44dc75f45d7..a37747cde2e 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -2975,6 +2975,10 @@ def supports_audio_input(model: str, custom_llm_provider: str | None = None) -> return supports_factory(model=model, custom_llm_provider=custom_llm_provider, key="supports_audio_input") +def supports_video_input(model: str, custom_llm_provider: str | None = None) -> bool: + return supports_factory(model=model, custom_llm_provider=custom_llm_provider, key="supports_video_input") + + def supports_pdf_input(model: str, custom_llm_provider: str | None = None) -> bool: """Check if a given model supports pdf input in a chat completion call""" return supports_factory(model=model, custom_llm_provider=custom_llm_provider, key="supports_pdf_input") @@ -6385,6 +6389,7 @@ def get_model_info_helper( supports_prompt_cache_breakpoint=_model_info.get("supports_prompt_cache_breakpoint", None), supports_thinking_cache_preservation=_model_info.get("supports_thinking_cache_preservation", None), supports_audio_input=_model_info.get("supports_audio_input", None), + supports_video_input=_model_info.get("supports_video_input", None), supports_audio_output=_model_info.get("supports_audio_output", None), supports_pdf_input=_model_info.get("supports_pdf_input", None), supports_embedding_image_input=_model_info.get("supports_embedding_image_input", None), @@ -6535,6 +6540,7 @@ def get_model_info( supports_prompt_caching: Optional[bool] supports_prompt_cache_breakpoint: Optional[bool] supports_audio_input: Optional[bool] + supports_video_input: Optional[bool] supports_audio_output: Optional[bool] supports_pdf_input: Optional[bool] supports_web_search: Optional[bool] diff --git a/model_prices_and_context_window.json b/model_prices_and_context_window.json index 780adab42ab..2427758877f 100644 --- a/model_prices_and_context_window.json +++ b/model_prices_and_context_window.json @@ -15982,6 +15982,26 @@ "prompt_cache_min_tokens": 1024, "source": "https://platform.claude.com/docs/en/about-claude/pricing" }, + "cloudflare/clef-omni": { + "input_cost_per_token": 1.5e-07, + "litellm_provider": "cloudflare", + "max_input_tokens": 64000, + "mode": "evaluation", + "output_cost_per_token": 0.0, + "source": "https://developers.cloudflare.com/workers-ai/models/clef-omni/", + "supports_audio_input": true, + "supports_video_input": true + }, + "cloudflare/@cf/cloudflare/clef-omni": { + "input_cost_per_token": 1.5e-07, + "litellm_provider": "cloudflare", + "max_input_tokens": 64000, + "mode": "evaluation", + "output_cost_per_token": 0.0, + "source": "https://developers.cloudflare.com/workers-ai/models/clef-omni/", + "supports_audio_input": true, + "supports_video_input": true + }, "cloudflare/clef": { "input_cost_per_token": 2.4e-07, "litellm_provider": "cloudflare", diff --git a/tests/integration/providers/test_cloudflare_clef_omni_media_chaos.py b/tests/integration/providers/test_cloudflare_clef_omni_media_chaos.py new file mode 100644 index 00000000000..1c736f59726 --- /dev/null +++ b/tests/integration/providers/test_cloudflare_clef_omni_media_chaos.py @@ -0,0 +1,294 @@ +import asyncio +import json +import re +import signal +import socket +import threading +import uuid +from dataclasses import dataclass +from pathlib import Path +from queue import SimpleQueue +from types import MappingProxyType +from typing import Final +from urllib.parse import unquote, urlsplit + +import httpx +import psutil +import pytest +import yaml +from integration._support.client import Gateway, eventually, object_value, string_value +from integration._support.database import read_rows +from integration._support.process import graceful_stop_seconds, owned_proxy_process +from integration._support.wire import Reply, Request, Wire, wire_server +from pydantic import JsonValue, TypeAdapter + +_API_KEY: Final = "synthetic-cloudflare-key" +_CANONICAL: Final = "cloudflare/@cf/cloudflare/clef-omni" +_CONFIG_MODEL: Final = "clef-omni-media-chaos" +_UPSTREAM_PATH: Final = "/ai/run/@cf/cloudflare/clef-omni" +_WAV: Final = "data:audio/wav;base64,UklGRiQAAABXQVZFZm10IBAAAAABAAEARKwAAIhYAQACABAAZGF0YQAAAAA=" +_MP4: Final = "data:video/mp4;base64,AAAAHGZ0eXBpc29tAAACAGlzb21pc28ybXA0MQ==" +_INPUT_TOKENS: Final = 537 +_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue]) +_MARKER: Final = re.compile(r"marker-([0-9a-f]{32})") +_STARTED_WORKER: Final = re.compile(r"Started server process \[(\d+)\]") +_QUESTIONS: Final[dict[str, JsonValue]] = { + "pineapple": {"type": "noul", "instructions": "Does the audio say pineapple?"}, + "red": {"type": "noul", "instructions": "Is the video red?"}, +} +_RED: Final = 0.97 +_CHAOS_TIMEOUT: Final = int(2 * graceful_stop_seconds() + 120) + + +@dataclass(frozen=True, slots=True) +class _Served: + marker: str + call_id: str + status: int + text: str + + +def _marker_of(request: Request) -> str: + found: Final = _MARKER.search(request.body.decode()) + assert found is not None, request.body[:200] + return found.group(1) + + +def _state(marker: str) -> str: + return f"Does the clip say pineapple? marker-{marker}" + + +def _probability(marker: str) -> float: + return round(int(marker[:6], 16) / 0x1000000, 7) + + +def _echo(request: Request) -> Reply: + marker: Final = _marker_of(request) + reply: Final[dict[str, JsonValue]] = { + "result": { + "model": "clef-omni", + "answers": { + "pineapple": {"type": "noul", "noul": _probability(marker)}, + "red": {"type": "noul", "noul": _RED}, + }, + "usage": {"input_tokens": _INPUT_TOKENS, "output_tokens": 0}, + }, + "success": True, + "errors": [], + "messages": [], + } + return Reply(body=json.dumps(reply).encode()) + + +def _body(model: str, marker: str) -> dict[str, JsonValue]: + return { + "model": model, + "state": _state(marker), + "questions": _QUESTIONS, + "audio": [_WAV], + "videos": [_MP4], + "num_retries": 0, + } + + +def _expected_upstream_body(marker: str) -> dict[str, JsonValue]: + return {"model": "clef-omni", "state": _state(marker), "audio": [_WAV], "videos": [_MP4], "questions": _QUESTIONS} + + +def _upstream_body(request: Request) -> dict[str, JsonValue]: + assert request.method == "POST", request.method + assert unquote(request.target) == _UPSTREAM_PATH, request.target + assert request.headers["authorization"] == f"Bearer {_API_KEY}", request.headers + return _JSON_OBJECT.validate_json(request.body) + + +def _assert_each_landed_once_with_media(received: tuple[Request, ...], markers: frozenset[str]) -> None: + bodies: Final = {_marker_of(request): _upstream_body(request) for request in received} + assert len(bodies) == len(received) == len(markers), (len(bodies), len(received), len(markers)) + for marker in markers: + assert bodies[marker] == _expected_upstream_body(marker), marker + + +def _assert_answered_with_its_own_probability(served: _Served) -> None: + assert served.status == 200, served.text + answers: Final = object_value(_JSON_OBJECT.validate_json(served.text)["answers"]) + assert object_value(answers["pineapple"])["noul"] == _probability(served.marker), served.text + + +async def _send(client: httpx.AsyncClient, key: str, model: str, marker: str) -> _Served: + response: Final = await client.post( + "/v1/systemone", json=_body(model, marker), headers={"Authorization": f"Bearer {key}"} + ) + return _Served( + marker=marker, + call_id=response.headers.get("x-litellm-call-id", ""), + status=response.status_code, + text=response.text, + ) + + +async def _burst( + base_url: str, key: str, model: str, markers: tuple[str, ...], *, tolerate_transport_errors: bool = False +) -> tuple[_Served, ...]: + async with httpx.AsyncClient(base_url=base_url, timeout=60, trust_env=False) as client: + results: Final = await asyncio.gather( + *(_send(client, key, model, marker) for marker in markers), return_exceptions=tolerate_transport_errors + ) + for result in results: + assert not isinstance(result, BaseException) or isinstance(result, httpx.TransportError), repr(result) + return tuple(result for result in results if isinstance(result, _Served)) + + +def _markers(count: int) -> tuple[str, ...]: + return tuple(uuid.uuid4().hex for _ in range(count)) + + +def _spend_rows(model: str, expected: int) -> list[dict[str, JsonValue]]: + rows: Final = eventually( + lambda: read_rows('SELECT litellm_call_id, status FROM "LiteLLM_SpendLogs" WHERE model_group=%s', (model,)), + lambda found: len(found) >= expected, + seconds=70, + ) + assert len({row["litellm_call_id"] for row in rows}) == len(rows) == expected, rows + return rows + + +def _assert_each_billed_once( + rows: list[dict[str, JsonValue]], failed: tuple[_Served, ...], served: tuple[_Served, ...] +) -> None: + by_call: Final = {string_value(row["litellm_call_id"]): row for row in rows} + assert len(by_call) == len(failed) + len(served), (len(by_call), len(failed), len(served)) + for item in failed: + assert by_call[item.call_id]["status"] == "failure", (item.call_id, by_call[item.call_id]) + for item in served: + assert by_call[item.call_id]["status"] == "success", (item.call_id, by_call[item.call_id]) + + +def _free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: + probe.bind(("127.0.0.1", 0)) + return int(probe.getsockname()[1]) + + +def _assert_failed_cleanly(served: _Served) -> None: + assert served.status >= 500, (served.status, served.text) + assert "answers" not in served.text, served.text + assert "error" in served.text.lower(), served.text + assert served.call_id, served + + +@pytest.mark.timeout(_CHAOS_TIMEOUT) +async def test_an_upstream_dropping_connections_mid_burst_then_stopped_and_restarted_lands_each_success_once( + gateway: Gateway, +) -> None: + port: Final = _free_port() + first_wave: Final = _markers(12) + dropped: Final = frozenset(marker for index, marker in enumerate(first_wave) if index % 2 == 1) + while_down: Final = _markers(6) + after_restart: Final = _markers(12) + + def flaky(request: Request) -> Reply: + if _marker_of(request) in dropped: + return Reply(drop_connection=True) + return _echo(request) + + with gateway.scenario() as scenario: + model: Final = scenario.model(model=_CANONICAL, api_base=f"http://127.0.0.1:{port}", api_key=_API_KEY) + base_url: Final = str(gateway.client.base_url) + with wire_server(flaky, port=port) as first: + first_served: Final = await _burst(base_url, gateway.key, model, first_wave) + assert len(first_served) == 12 + for item in first_served: + if item.marker in dropped: + _assert_failed_cleanly(item) + else: + _assert_answered_with_its_own_probability(item) + _assert_each_landed_once_with_media(first.drain(), frozenset(first_wave)) + down: Final = await _burst(base_url, gateway.key, model, while_down) + assert len(down) == 6 + for item in down: + _assert_failed_cleanly(item) + with wire_server(_echo, port=port) as second: + recovered: Final = await _burst(base_url, gateway.key, model, after_restart) + assert len(recovered) == 12 + for item in recovered: + _assert_answered_with_its_own_probability(item) + _assert_each_landed_once_with_media(second.drain(), frozenset(after_restart)) + failed: Final = (*(item for item in first_served if item.marker in dropped), *down) + served: Final = (*(item for item in first_served if item.marker not in dropped), *recovered) + _assert_each_billed_once(_spend_rows(model, 30), failed, served) + + +def _chaos_config(wire: Wire, tmp_path: Path) -> Path: + config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + config["model_list"] = [ + { + "model_name": _CONFIG_MODEL, + "litellm_params": {"model": _CANONICAL, "api_base": wire.url, "api_key": _API_KEY}, + } + ] + path: Final = tmp_path / "clef-omni-media-chaos.yaml" + path.write_text(yaml.safe_dump(config)) + return path + + +def _open_upstream_connections(pid: int, upstream: str) -> int: + port: Final = urlsplit(upstream).port + return sum( + 1 + for connection in psutil.Process(pid).net_connections(kind="tcp") + if connection.status == psutil.CONN_ESTABLISHED and connection.raddr and connection.raddr.port == port + ) + + +@pytest.mark.timeout(_CHAOS_TIMEOUT) +async def test_worker_sigkill_mid_burst_leaves_the_sibling_forwarding_media(gateway: Gateway, tmp_path: Path) -> None: + markers: Final = _markers(20) + release: Final = threading.Event() + held_markers: Final[SimpleQueue[str]] = SimpleQueue() + + def held(request: Request) -> Reply: + if request.method != "POST": + return Reply(body=b"{}") + held_markers.put(_marker_of(request)) + assert release.wait(timeout=60), "The burst was never released" + return _echo(request) + + with wire_server(held) as wire: + path: Final = _chaos_config(wire, tmp_path) + with owned_proxy_process(gateway, tmp_path, {}, config=path, workers=2) as owned: + candidate: Final = owned.gateway + workers: Final = eventually( + lambda: tuple(int(pid) for pid in _STARTED_WORKER.findall(owned.log.read_text())), + lambda pids: len(pids) == 2, + seconds=30, + ) + burst: Final = asyncio.create_task( + _burst( + str(candidate.client.base_url), + candidate.key, + _CONFIG_MODEL, + markers, + tolerate_transport_errors=True, + ) + ) + await asyncio.to_thread(eventually, held_markers.qsize, lambda size: size == 20, 60) + held_by: Final = MappingProxyType({pid: _open_upstream_connections(pid, wire.url) for pid in workers}) + assert sum(held_by.values()) == 20, held_by + victim_pid, survivor_pid = sorted(workers, key=held_by.__getitem__) + victim: Final = psutil.Process(victim_pid) + victim.suspend() + victim.send_signal(signal.SIGKILL) + release.set() + served: Final = await burst + assert held_by[survivor_pid] >= 10, held_by + assert len(served) == held_by[survivor_pid], (held_by, len(served)) + for item in served: + _assert_answered_with_its_own_probability(item) + follow_up: Final = uuid.uuid4().hex + (answered,) = await _burst(str(candidate.client.base_url), candidate.key, _CONFIG_MODEL, (follow_up,)) + _assert_answered_with_its_own_probability(answered) + received: Final = wire.drain() + posts: Final = tuple(request for request in received if request.method == "POST") + assert len(posts) == len(received), [(request.method, request.target) for request in received] + _assert_each_landed_once_with_media(posts, frozenset((*markers, follow_up))) diff --git a/tests/integration/providers/test_cloudflare_clef_omni_media_wire.py b/tests/integration/providers/test_cloudflare_clef_omni_media_wire.py new file mode 100644 index 00000000000..b423b407422 --- /dev/null +++ b/tests/integration/providers/test_cloudflare_clef_omni_media_wire.py @@ -0,0 +1,628 @@ +import asyncio +import json +import math +import re +import threading +import time +import uuid +from collections.abc import Sequence +from dataclasses import dataclass +from pathlib import Path +from queue import SimpleQueue +from typing import Final, Literal +from urllib.parse import unquote + +import httpx +import pytest +from integration._support.client import Gateway, eventually, list_value, object_value +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server +from pydantic import JsonValue, TypeAdapter + +Route = Literal["/v1/systemone", "/systemone", "/v1/decisions", "/decisions"] + +_ROUTES: Final[tuple[Route, ...]] = ("/v1/systemone", "/systemone", "/v1/decisions", "/decisions") +_API_KEY: Final = "synthetic-cloudflare-key" +_CANONICAL: Final = "cloudflare/@cf/cloudflare/clef-omni" +_ALIAS: Final = "cloudflare/clef-omni" +_COST_MAP_KEY: Final = "cloudflare/@cf/cloudflare/clef-omni" +_UPSTREAM_PATH: Final = "/ai/run/@cf/cloudflare/clef-omni" +_WAV: Final = "data:audio/wav;base64,UklGRiQAAABXQVZFZm10IBAAAAABAAEARKwAAIhYAQACABAAZGF0YQAAAAA=" +_MP4: Final = "data:video/mp4;base64,AAAAHGZ0eXBpc29tAAACAGlzb21pc28ybXA0MQ==" +_WAV_OBJECT: Final[dict[str, JsonValue]] = {"content_type": "audio/wav", "base64": "UklGRiQAAABXQVZF"} +_MP4_OBJECT: Final[dict[str, JsonValue]] = {"content_type": "video/mp4", "base64": "AAAAHGZ0eXBpc29t"} +_PARAMETERIZED_WAV_OBJECT: Final[dict[str, JsonValue]] = {"content_type": "audio/wav;rate=16000", "base64": "UklGRg=="} +_INPUT_TOKENS: Final = 537 +_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue]) +_JSON_ROWS: Final = TypeAdapter(list[dict[str, JsonValue]]) +_MARKER: Final = re.compile(r"marker-([0-9a-f]{32})") +_SYSTEM_ONE_QUESTIONS: Final[dict[str, JsonValue]] = { + "pineapple": {"type": "noul", "instructions": "Does the audio say pineapple?"}, + "red": {"type": "noul", "instructions": "Is the video red?"}, +} +_OPENAI_QUESTIONS: Final[list[JsonValue]] = [ + {"type": "predicate", "name": "pineapple", "instructions": "Does the audio say pineapple?"}, + {"type": "predicate", "name": "red", "instructions": "Is the video red?"}, +] +_RED: Final = 0.97 +_SPEND_QUERY: Final = ( + 'SELECT spend, status, call_type, model_group, prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" ' + "WHERE request_id = %s" +) +_WORST_CASE_LENGTH: Final = 1_048_576 + + +@dataclass(frozen=True, slots=True) +class _Call: + route: Route + marker: str + + +@dataclass(frozen=True, slots=True) +class _Served: + call: _Call + status: int + text: str + + +def _marker_of(request: Request) -> str: + found: Final = _MARKER.search(request.body.decode()) + assert found is not None, request.body[:200] + return found.group(1) + + +def _state(marker: str) -> str: + return f"Does the clip say pineapple? marker-{marker}" + + +def _probability(marker: str) -> float: + return round(int(marker[:6], 16) / 0x1000000, 7) + + +def _upstream_reply(marker: str) -> dict[str, JsonValue]: + return { + "result": { + "model": "clef-omni", + "answers": { + "pineapple": {"type": "noul", "noul": _probability(marker)}, + "red": {"type": "noul", "noul": _RED}, + }, + "usage": {"input_tokens": _INPUT_TOKENS, "output_tokens": 0}, + }, + "success": True, + "errors": [], + "messages": [], + } + + +def _openai_decider_reply(marker: str) -> dict[str, JsonValue]: + return { + "model": "gpt-6-luna", + "answers": [ + {"type": "predicate", "name": "pineapple", "probability": _probability(marker)}, + {"type": "predicate", "name": "red", "probability": _RED}, + ], + "usage": {"input_tokens": _INPUT_TOKENS, "output_tokens": 0, "total_tokens": _INPUT_TOKENS}, + } + + +def _echo(request: Request) -> Reply: + marker: Final = _marker_of(request) + if request.target.endswith("/v1/decisions"): + return Reply(body=json.dumps(_openai_decider_reply(marker)).encode()) + return Reply(body=json.dumps(_upstream_reply(marker)).encode()) + + +def _speaks_system_one(route: Route) -> bool: + return route.endswith("systemone") + + +def _body(route: Route, model: str, marker: str, **extra: JsonValue) -> dict[str, JsonValue]: + if _speaks_system_one(route): + return {"model": model, "state": _state(marker), "questions": _SYSTEM_ONE_QUESTIONS, **extra} + return {"model": model, "input": _state(marker), "questions": _OPENAI_QUESTIONS, **extra} + + +def _expected_response(route: Route, marker: str) -> dict[str, JsonValue]: + if _speaks_system_one(route): + return { + "model": "clef-omni", + "answers": { + "pineapple": {"type": "noul", "noul": _probability(marker)}, + "red": {"type": "noul", "noul": _RED}, + }, + "usage": {"input_tokens": _INPUT_TOKENS, "output_tokens": 0}, + } + return { + "model": "clef-omni", + "answers": [ + {"type": "predicate", "name": "pineapple", "probability": _probability(marker)}, + {"type": "predicate", "name": "red", "probability": _RED}, + ], + "usage": { + "input_tokens": _INPUT_TOKENS, + "input_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 0}, + "output_tokens": 0, + "output_tokens_details": {"reasoning_tokens": 0}, + "total_tokens": _INPUT_TOKENS, + }, + } + + +def _expected_upstream_body(marker: str, **media: JsonValue) -> dict[str, JsonValue]: + return {"model": "clef-omni", "state": _state(marker), **media, "questions": _SYSTEM_ONE_QUESTIONS} + + +def _upstream_body(request: Request) -> dict[str, JsonValue]: + assert request.method == "POST", request.method + assert unquote(request.target) == _UPSTREAM_PATH, request.target + assert request.headers["authorization"] == f"Bearer {_API_KEY}", request.headers + return _JSON_OBJECT.validate_json(request.body) + + +def _decide(gateway: Gateway, route: Route, model: str, marker: str, **extra: JsonValue) -> httpx.Response: + return gateway.request("POST", route, _body(route, model, marker, **extra)) + + +def _answered_probability(route: Route, text: str) -> JsonValue: + answers: Final = _JSON_OBJECT.validate_json(text)["answers"] + if _speaks_system_one(route): + return object_value(object_value(answers)["pineapple"])["noul"] + return object_value(list_value(answers)[0])["probability"] + + +def _price() -> dict[str, JsonValue]: + cost_map: Final = _JSON_OBJECT.validate_json(Path("model_prices_and_context_window.json").read_bytes()) + priced: Final = cost_map.get(_COST_MAP_KEY) + assert priced is not None, f"{_COST_MAP_KEY} has no cost map entry" + return object_value(priced) + + +def _number(value: JsonValue) -> float: + assert isinstance(value, (int, float)) and not isinstance(value, bool), value + return float(value) + + +def _expected_spend() -> float: + return _INPUT_TOKENS * _number(_price()["input_cost_per_token"]) + + +def _spend_row(call_id: str) -> dict[str, JsonValue]: + rows: Final = eventually(lambda: read_rows(_SPEND_QUERY, (call_id,)), lambda found: len(found) == 1, seconds=70) + return rows[0] + + +def _spend_rows_for(model: str, expected: int) -> list[dict[str, JsonValue]]: + rows: Final = eventually( + lambda: read_rows('SELECT litellm_call_id, status FROM "LiteLLM_SpendLogs" WHERE model_group=%s', (model,)), + lambda found: len(found) >= expected, + seconds=70, + ) + assert len({row["litellm_call_id"] for row in rows}) == len(rows) == expected, rows + return rows + + +def _call_type(route: Route) -> str: + return "asystemone" if route.endswith("systemone") else "adecisions" + + +def _assert_billed(gateway: Gateway, response: httpx.Response, model: str, route: Route) -> None: + expected: Final = _expected_spend() + assert math.isclose(float(response.headers["x-litellm-response-cost"]), expected, rel_tol=1e-9), response.headers + call_id: Final = response.headers["x-litellm-call-id"] + row: Final = _spend_row(call_id) + assert (row["status"], row["call_type"], row["model_group"], row["prompt_tokens"], row["completion_tokens"]) == ( + "success", + _call_type(route), + model, + _INPUT_TOKENS, + 0, + ), row + assert math.isclose(_number(row["spend"]), expected, rel_tol=1e-9), row + listed: Final = gateway.request("GET", "/spend/logs", params={"request_id": call_id}) + assert listed.status_code == 200, listed.text + (reported,) = _JSON_ROWS.validate_json(listed.content) + assert reported["request_id"] == call_id, reported + assert (reported["model_group"], reported["prompt_tokens"], reported["completion_tokens"]) == ( + model, + _INPUT_TOKENS, + 0, + ), reported + assert math.isclose(_number(reported["spend"]), expected, rel_tol=1e-9), reported + + +def _assert_served_with_media(response: httpx.Response, route: Route, marker: str, request: Request) -> None: + assert response.status_code == 200, response.text + assert response.json() == _expected_response(route, marker) + assert _upstream_body(request) == _expected_upstream_body(marker, audio=[_WAV], videos=[_MP4]) + + +@pytest.mark.parametrize("route", _ROUTES) +def test_data_url_clips_reach_clef_omni_on_every_route_and_are_billed(gateway: Gateway, route: Route) -> None: + marker: Final = uuid.uuid4().hex + with wire_server(_echo) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_CANONICAL, api_base=wire.url, api_key=_API_KEY) + response: Final = _decide(gateway, route, model, marker, audio=[_WAV], videos=[_MP4]) + (request,) = wire.drain() + _assert_served_with_media(response, route, marker, request) + _assert_billed(gateway, response, model, route) + + +async def test_an_async_caller_on_the_unversioned_decisions_route_gets_its_media_forwarded(gateway: Gateway) -> None: + marker: Final = uuid.uuid4().hex + with wire_server(_echo) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_CANONICAL, api_base=wire.url, api_key=_API_KEY) + async with httpx.AsyncClient(base_url=str(gateway.client.base_url), timeout=60, trust_env=False) as client: + response: Final = await client.post( + "/decisions", + json=_body("/decisions", model, marker, audio=[_WAV], videos=[_MP4]), + headers={"Authorization": f"Bearer {gateway.key}"}, + ) + (request,) = wire.drain() + _assert_served_with_media(response, "/decisions", marker, request) + _assert_billed(gateway, response, model, "/decisions") + + +@pytest.mark.parametrize( + "audio_clip", (_WAV_OBJECT, _PARAMETERIZED_WAV_OBJECT), ids=("plain_content_type", "parameterized_content_type") +) +async def test_object_form_clips_are_forwarded_verbatim(gateway: Gateway, audio_clip: dict[str, JsonValue]) -> None: + marker: Final = uuid.uuid4().hex + with wire_server(_echo) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_CANONICAL, api_base=wire.url, api_key=_API_KEY) + async with httpx.AsyncClient(base_url=str(gateway.client.base_url), timeout=60, trust_env=False) as client: + response: Final = await client.post( + "/v1/systemone", + json=_body("/v1/systemone", model, marker, audio=[audio_clip], videos=[_MP4_OBJECT]), + headers={"Authorization": f"Bearer {gateway.key}"}, + ) + assert response.status_code == 200, response.text + assert response.json() == _expected_response("/v1/systemone", marker) + (request,) = wire.drain() + assert _upstream_body(request) == _expected_upstream_body(marker, audio=[audio_clip], videos=[_MP4_OBJECT]) + + +def test_both_clef_omni_spellings_forward_media_and_carry_the_cost_map_price(gateway: Gateway) -> None: + price: Final = _price() + with wire_server(_echo) as wire, gateway.scenario() as scenario: + deployments: Final = tuple( + scenario.model(model=spelling, api_base=wire.url, api_key=_API_KEY) for spelling in (_ALIAS, _CANONICAL) + ) + for model in deployments: + marker: Final = uuid.uuid4().hex + response: Final = _decide(gateway, "/v1/systemone", model, marker, audio=[_WAV], videos=[_MP4]) + (request,) = wire.drain() + _assert_served_with_media(response, "/v1/systemone", marker, request) + _assert_billed(gateway, response, model, "/v1/systemone") + listed: Final = list_value(gateway.get("/model/info")["data"]) + for model in deployments: + entry: Final = next(object_value(item) for item in listed if object_value(item)["model_name"] == model) + info: Final = object_value(entry["model_info"]) + assert ( + info.get("input_cost_per_token"), + info.get("output_cost_per_token"), + info.get("mode"), + info.get("max_input_tokens"), + info.get("supports_audio_input"), + info.get("supports_video_input"), + ) == ( + price["input_cost_per_token"], + price["output_cost_per_token"], + price["mode"], + price["max_input_tokens"], + price["supports_audio_input"], + price["supports_video_input"], + ), (model, info) + + +def test_the_maxima_of_four_audio_and_two_video_clips_are_forwarded(gateway: Gateway) -> None: + marker: Final = uuid.uuid4().hex + audio: Final[list[JsonValue]] = [_WAV, _WAV_OBJECT, _WAV, _WAV_OBJECT] + videos: Final[list[JsonValue]] = [_MP4, _MP4_OBJECT] + with wire_server(_echo) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_CANONICAL, api_base=wire.url, api_key=_API_KEY) + response: Final = _decide(gateway, "/v1/systemone", model, marker, audio=audio, videos=videos) + assert response.status_code == 200, response.text + (request,) = wire.drain() + assert _upstream_body(request) == _expected_upstream_body(marker, audio=audio, videos=videos) + + +def test_missing_empty_and_null_media_send_the_same_text_only_body(gateway: Gateway) -> None: + marker: Final = uuid.uuid4().hex + with wire_server(_echo) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_CANONICAL, api_base=wire.url, api_key=_API_KEY) + missing: Final = _decide(gateway, "/v1/systemone", model, marker) + empty: Final = _decide(gateway, "/v1/systemone", model, marker, audio=[], videos=[]) + null: Final = _decide(gateway, "/v1/systemone", model, marker, audio=None, videos=None) + for response in (missing, empty, null): + assert response.status_code == 200, response.text + assert response.json() == _expected_response("/v1/systemone", marker) + bodies: Final = tuple(_upstream_body(request) for request in wire.drain()) + assert bodies == (_expected_upstream_body(marker),) * 3, bodies + + +_MALFORMED: Final[tuple[tuple[str, dict[str, JsonValue], str], ...]] = ( + ("audio_int", {"audio": 5}, "audio"), + ("audio_list_of_ints", {"audio": [1, 2]}, "audio"), + ("audio_empty_string", {"audio": ""}, "audio"), + ("audio_5kb_garbage", {"audio": ["a" * 5000]}, "pattern"), + ("videos_object_missing_base64", {"videos": [{"content_type": "video/mp4"}]}, "base64"), + ("videos_object_extra_key", {"videos": [{**_MP4_OBJECT, "url": "https://example.invalid/clip.mp4"}]}, "url"), + ("image_data_url_in_audio", {"audio": ["data:image/png;base64,iVBORw0KGgo="]}, "pattern"), + ( + "parameterized_audio_data_url", + {"audio": ["data:audio/webm;codecs=opus;base64,GkXfo0AgQoaBAUL3gQFC8oEEQvOBCA=="]}, + "pattern", + ), + ("five_audio_clips", {"audio": [_WAV] * 5}, "at most 4"), + ("three_videos", {"videos": [_MP4] * 3}, "at most 2"), + ( + "audio_object_with_image_content_type", + {"audio": [{"content_type": "image/png", "base64": "iVBORw0KGgo="}]}, + "pattern", + ), + ("video_object_with_empty_base64", {"videos": [{"content_type": "video/mp4", "base64": ""}]}, "at least 1"), + ( + "audio_object_content_type_with_a_space", + {"audio": [{"content_type": "audio/wav x", "base64": "UklGRg=="}]}, + "pattern", + ), +) + + +@pytest.mark.parametrize(("label", "extra", "fragment"), _MALFORMED, ids=[label for label, _, _ in _MALFORMED]) +def test_malformed_clips_are_refused_before_any_upstream_call( + gateway: Gateway, label: str, extra: dict[str, JsonValue], fragment: str +) -> None: + marker: Final = uuid.uuid4().hex + with wire_server(_echo) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_CANONICAL, api_base=wire.url, api_key=_API_KEY) + response: Final = _decide(gateway, "/v1/systemone", model, marker, **extra) + assert response.status_code == 400, (label, response.text) + assert "Invalid Decisions request" in response.text and fragment in response.text, (label, response.text) + assert wire.drain() == (), label + + +def _raw_system_one(gateway: Gateway, model: str, marker: str, audio_fragment: str) -> httpx.Response: + raw: Final = ( + f'{{"model": {json.dumps(model)}, "state": {json.dumps(_state(marker))}, ' + f'"questions": {json.dumps(_SYSTEM_ONE_QUESTIONS)}, {audio_fragment}}}' + ) + return gateway.client.post( + "/v1/systemone", + content=raw.encode(), + headers={"Authorization": f"Bearer {gateway.key}", "Content-Type": "application/json"}, + ) + + +def test_a_duplicated_audio_key_is_judged_by_its_last_value(gateway: Gateway) -> None: + marker: Final = uuid.uuid4().hex + image: Final = '"data:image/png;base64,iVBORw0KGgo="' + with wire_server(_echo) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_CANONICAL, api_base=wire.url, api_key=_API_KEY) + refused: Final = _raw_system_one(gateway, model, marker, f'"audio": [{json.dumps(_WAV)}], "audio": [{image}]') + assert refused.status_code == 400, refused.text + assert "Invalid Decisions request" in refused.text and "pattern" in refused.text, refused.text + assert wire.drain() == () + accepted: Final = _raw_system_one(gateway, model, marker, f'"audio": [{image}], "audio": [{json.dumps(_WAV)}]') + assert accepted.status_code == 200, accepted.text + (request,) = wire.drain() + assert _upstream_body(request) == _expected_upstream_body(marker, audio=[_WAV]) + + +_UNSUPPORTED: Final[tuple[tuple[str, Route], ...]] = ( + ("cloudflare/clef-flash", "/v1/systemone"), + ("cloudflare/clef", "/v1/systemone"), + ("cloudflare/@cf/cloudflare/clef-flash", "/v1/decisions"), + ("openai/gpt-6-luna", "/v1/systemone"), + ("openai/gpt-6-luna", "/v1/decisions"), +) + + +@pytest.mark.parametrize(("spelling", "route"), _UNSUPPORTED, ids=[f"{m}@{r}" for m, r in _UNSUPPORTED]) +def test_media_on_a_model_without_support_is_refused_before_any_upstream_call( + gateway: Gateway, spelling: str, route: Route +) -> None: + marker: Final = uuid.uuid4().hex + with wire_server(_echo) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=spelling, api_base=wire.url, api_key=_API_KEY) + response: Final = _decide(gateway, route, model, marker, audio=[_WAV]) + assert response.status_code == 400, response.text + assert "does not support parameters: ['audio']" in response.text, response.text + assert "drop_params" in response.text, response.text + assert wire.drain() == () + text_only: Final = _decide(gateway, route, model, marker) + assert text_only.status_code == 200, text_only.text + assert len(wire.drain()) == 1 + + +@pytest.mark.parametrize("route", ("/v1/systemone", "/v1/decisions")) +def test_media_on_a_model_without_support_is_dropped_when_the_deployment_drops_params( + gateway: Gateway, route: Route +) -> None: + marker: Final = uuid.uuid4().hex + with wire_server(_echo) as wire, gateway.scenario() as scenario: + model: Final = scenario.model( + model="cloudflare/clef-flash", api_base=wire.url, api_key=_API_KEY, drop_params=True + ) + text_only: Final = _decide(gateway, route, model, marker) + assert text_only.status_code == 200, text_only.text + (text_only_request,) = wire.drain() + with_clips: Final = _decide(gateway, route, model, marker, audio=[_WAV], videos=[_MP4]) + assert with_clips.status_code == 200, with_clips.text + (dropped_request,) = wire.drain() + text_only_body: Final = _JSON_OBJECT.validate_json(text_only_request.body) + assert "audio" not in text_only_body and "videos" not in text_only_body, text_only_body + assert _JSON_OBJECT.validate_json(dropped_request.body) == text_only_body + assert with_clips.json() == text_only.json() + + +_UPSTREAM_REJECTIONS: Final[tuple[tuple[int, str], ...]] = ( + (422, "Each audio entry must be a data URL or an object with content_type and base64"), + (413, "Request entity too large"), +) + + +@pytest.mark.parametrize(("status", "message"), _UPSTREAM_REJECTIONS, ids=("422", "413")) +def test_an_upstream_rejection_of_a_media_request_reaches_the_caller_like_a_text_only_one( + gateway: Gateway, status: int, message: str +) -> None: + marker: Final = uuid.uuid4().hex + rejection: Final = json.dumps( + {"result": {}, "success": False, "errors": [{"code": 5006, "message": message}], "messages": []} + ).encode() + with wire_server(lambda _: Reply(status=status, body=rejection)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_CANONICAL, api_base=wire.url, api_key=_API_KEY) + text_only: Final = _decide(gateway, "/v1/systemone", model, marker, num_retries=0) + with_media: Final = _decide(gateway, "/v1/systemone", model, marker, num_retries=0, audio=[_WAV], videos=[_MP4]) + for response in (text_only, with_media): + assert response.status_code >= 400, response.text + assert message in response.text, response.text + assert "answers" not in response.text, response.text + assert with_media.status_code == text_only.status_code, (text_only.status_code, with_media.status_code) + first, second = (_upstream_body(request) for request in wire.drain()) + assert first == _expected_upstream_body(marker) + assert second == _expected_upstream_body(marker, audio=[_WAV], videos=[_MP4]) + + +def test_an_unauthenticated_media_request_is_refused_before_validation(gateway: Gateway) -> None: + marker: Final = uuid.uuid4().hex + with wire_server(_echo) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_CANONICAL, api_base=wire.url, api_key=_API_KEY) + for route in _ROUTES: + response: Final = gateway.client.post(route, json=_body(route, model, marker, audio=[_WAV], videos=[_MP4])) + assert response.status_code == 401, (route, response.text) + assert "Authentication Error" in response.text, response.text + assert wire.drain() == () + + +def test_unrelated_chat_keeps_working_after_a_run_of_refused_media_requests(gateway: Gateway) -> None: + marker: Final = uuid.uuid4().hex + with wire_server(_echo) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_CANONICAL, api_base=wire.url, api_key=_API_KEY) + control: Final = scenario.model() + for _, extra, _ in _MALFORMED: + assert _decide(gateway, "/v1/systemone", model, marker, **extra).status_code == 400 + assert wire.drain() == () + assert gateway.chat(control)["choices"], "the control deployment stopped answering" + accepted: Final = _decide(gateway, "/v1/systemone", model, marker, audio=[_WAV]) + assert accepted.status_code == 200, accepted.text + (request,) = wire.drain() + assert _upstream_body(request) == _expected_upstream_body(marker, audio=[_WAV]) + + +def test_repeated_and_clip_changed_requests_each_reach_the_upstream_and_are_each_billed(gateway: Gateway) -> None: + marker: Final = uuid.uuid4().hex + with wire_server(_echo) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_CANONICAL, api_base=wire.url, api_key=_API_KEY) + first: Final = _decide(gateway, "/v1/systemone", model, marker, audio=[_WAV], videos=[_MP4]) + second: Final = _decide(gateway, "/v1/systemone", model, marker, audio=[_WAV], videos=[_MP4]) + swapped: Final = _decide(gateway, "/v1/systemone", model, marker, audio=[_WAV_OBJECT], videos=[_MP4]) + responses: Final = (first, second, swapped) + for response in responses: + assert response.status_code == 200, response.text + assert response.json() == _expected_response("/v1/systemone", marker) + call_ids: Final = tuple(response.headers["x-litellm-call-id"] for response in responses) + assert len(set(call_ids)) == 3, call_ids + bodies: Final = tuple(_upstream_body(request) for request in wire.drain()) + assert bodies == ( + _expected_upstream_body(marker, audio=[_WAV], videos=[_MP4]), + _expected_upstream_body(marker, audio=[_WAV], videos=[_MP4]), + _expected_upstream_body(marker, audio=[_WAV_OBJECT], videos=[_MP4]), + ), bodies + for call_id in call_ids: + assert _spend_row(call_id)["status"] == "success", call_id + + +async def _send(client: httpx.AsyncClient, key: str, model: str, call: _Call) -> _Served: + response: Final = await client.post( + call.route, + json=_body(call.route, model, call.marker, audio=[_WAV], videos=[_MP4]), + headers={"Authorization": f"Bearer {key}"}, + ) + return _Served(call=call, status=response.status_code, text=response.text) + + +async def _burst(base_url: str, key: str, model: str, calls: Sequence[_Call]) -> tuple[_Served, ...]: + async with httpx.AsyncClient(base_url=base_url, timeout=60, trust_env=False) as client: + return tuple(await asyncio.gather(*(_send(client, key, model, call) for call in calls))) + + +def _assert_answered_with_its_own_probability(served: _Served) -> None: + assert served.status == 200, served.text + assert _answered_probability(served.call.route, served.text) == _probability(served.call.marker), served.text + + +def _assert_each_landed_once_with_media(received: tuple[Request, ...], calls: Sequence[_Call]) -> None: + bodies: Final = {_marker_of(request): _upstream_body(request) for request in received} + assert len(bodies) == len(received) == len(calls), (len(bodies), len(received), len(calls)) + for call in calls: + assert bodies[call.marker] == _expected_upstream_body(call.marker, audio=[_WAV], videos=[_MP4]), call + + +async def test_a_concurrent_burst_across_the_four_routes_lands_each_request_once_with_its_media( + gateway: Gateway, +) -> None: + calls: Final = tuple(_Call(route=_ROUTES[index % len(_ROUTES)], marker=uuid.uuid4().hex) for index in range(20)) + with wire_server(_echo) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_CANONICAL, api_base=wire.url, api_key=_API_KEY) + served: Final = await _burst(str(gateway.client.base_url), gateway.key, model, calls) + assert len(served) == 20 + for item in served: + _assert_answered_with_its_own_probability(item) + _assert_each_landed_once_with_media(wire.drain(), calls) + rows: Final = _spend_rows_for(model, 20) + assert [row["status"] for row in rows] == ["success"] * 20, rows + + +def test_openapi_documents_the_media_fields_on_every_decisions_route(gateway: Gateway) -> None: + paths: Final = object_value(gateway.get("/openapi.json")["paths"]) + for route in _ROUTES: + post: Final = object_value(object_value(paths[route])["post"]) + content: Final = object_value(object_value(post["requestBody"])["content"]) + schema: Final = object_value(object_value(content["application/json"])["schema"]) + properties: Final = object_value(schema["properties"]) + assert "audio" in properties and "videos" in properties, (route, sorted(properties)) + assert object_value(properties["audio"])["maxItems"] == 4, (route, properties["audio"]) + assert object_value(properties["videos"])["maxItems"] == 2, (route, properties["videos"]) + + +def _probe_liveliness(base_url: str, stop: threading.Event, latencies: SimpleQueue[float]) -> None: + with httpx.Client(base_url=base_url, timeout=10, trust_env=False) as client: + while not stop.is_set(): + started: Final = time.perf_counter() + response: Final = client.get("/health/liveliness") + latencies.put(time.perf_counter() - started) + assert response.status_code == 200, response.text + + +def test_the_data_url_pattern_worst_case_is_refused_in_bounded_time_while_liveliness_answers( + gateway: Gateway, +) -> None: + marker: Final = uuid.uuid4().hex + worst_case: Final = "data:audio/" + "a" * _WORST_CASE_LENGTH + stop: Final = threading.Event() + latencies: Final[SimpleQueue[float]] = SimpleQueue() + with wire_server(_echo) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_CANONICAL, api_base=wire.url, api_key=_API_KEY) + probe: Final = threading.Thread( + target=_probe_liveliness, args=(str(gateway.client.base_url), stop, latencies), daemon=True + ) + probe.start() + started: Final = time.perf_counter() + with httpx.Client(base_url=str(gateway.client.base_url), timeout=60, trust_env=False) as client: + response: Final = client.post( + "/v1/systemone", + json=_body("/v1/systemone", model, marker, audio=[worst_case]), + headers={"Authorization": f"Bearer {gateway.key}"}, + ) + elapsed: Final = time.perf_counter() - started + stop.set() + probe.join(timeout=15) + observed: Final = tuple(latencies.get_nowait() for _ in range(latencies.qsize())) + timing: Final = {"classifier_seconds": round(elapsed, 3), "slowest_liveliness_seconds": round(max(observed), 3)} + assert elapsed < 5.0, timing + assert max(observed) < 1.0, timing + assert response.status_code == 400, (timing, response.text[:300]) + assert "pattern" in response.text, response.text[:300] + assert wire.drain() == (), timing diff --git a/tests/unit/decisions/test_main.py b/tests/unit/decisions/test_main.py index e1b2028fa30..dbdf84534a4 100644 --- a/tests/unit/decisions/test_main.py +++ b/tests/unit/decisions/test_main.py @@ -2,9 +2,10 @@ from __future__ import annotations import asyncio import json +import re import threading -from concurrent.futures import wait from collections.abc import Mapping +from concurrent.futures import wait from types import MappingProxyType from typing import Final @@ -16,8 +17,8 @@ from botocore.awsrequest import AWSRequest from botocore.credentials import Credentials import litellm -from litellm.cost_calculator import get_response_cost_from_hidden_params from litellm.constants import AWS_SIGNING_MAX_THREADS +from litellm.cost_calculator import get_response_cost_from_hidden_params from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER from litellm.llms.bedrock.base_aws_llm import AWS_SIGNING_EXECUTOR @@ -1730,3 +1731,245 @@ async def test_azure_ai_decision_requires_api_base_before_http( ) assert len(respx_mock.calls) == 0 + + +_CLEF_OMNI_UPSTREAM: Final = "https://example.com/ai/run/@cf/cloudflare/clef-omni" +_CLEF_OMNI_RESPONSE: Final[Mapping[str, object]] = { + "success": True, + "result": { + "model": "clef-omni", + "answers": {"heard": {"type": "noul", "noul": 0.9}}, + "usage": {"input_tokens": 100, "output_tokens": 0}, + }, +} +_MEDIA_REQUESTS: Final[tuple[tuple[str, Mapping[str, object]], ...]] = ( + ( + "decisions", + MappingProxyType( + { + "input": "Evaluate the clips", + "questions": [{"type": "predicate", "name": "heard", "instructions": "Is there speech?"}], + } + ), + ), + ( + "systemone", + MappingProxyType( + { + "state": "Evaluate the clips", + "questions": {"heard": {"type": "noul", "instructions": "Is there speech?"}}, + } + ), + ), +) + + +@pytest.mark.parametrize("async_mode", (False, True), ids=("sync", "async")) +@pytest.mark.parametrize(("api", "request_kwargs"), _MEDIA_REQUESTS, ids=_PREDICATE_REQUEST_IDS) +@pytest.mark.parametrize("model", ("cloudflare/clef-omni", "cloudflare/@cf/cloudflare/clef-omni")) +@pytest.mark.asyncio +async def test_clef_omni_media_reaches_upstream_in_both_formats( + respx_mock: respx.MockRouter, async_mode: bool, api: str, request_kwargs: Mapping[str, object], model: str +) -> None: + audio: Final = [ + "data:audio/wav;base64,YQ==", + {"content_type": "audio/mpeg", "base64": "Yg=="}, + {"content_type": "audio/webm;codecs=opus", "base64": "ZQ=="}, + ] + videos: Final = ["data:video/mp4;base64,Yw==", {"content_type": "video/webm;codecs=vp9", "base64": "ZA=="}] + route: Final = respx_mock.post(_CLEF_OMNI_UPSTREAM).respond(200, json=dict(_CLEF_OMNI_RESPONSE)) + call: Final = getattr(litellm, f"a{api}" if async_mode else api) + + response: Final = ( + await call( + model=model, api_base="https://example.com", api_key="test", audio=audio, videos=videos, **request_kwargs + ) + if async_mode + else call( + model=model, api_base="https://example.com", api_key="test", audio=audio, videos=videos, **request_kwargs + ) + ) + + assert json.loads(route.calls[0].request.content) == { + "model": "clef-omni", + "state": "Evaluate the clips", + "questions": {"heard": {"type": "noul", "instructions": "Is there speech?"}}, + "audio": audio, + "videos": videos, + } + assert response.usage is not None + assert response.usage.input_tokens == 100 + assert len(route.calls) == 1 + + +@pytest.mark.parametrize( + "model", ("cloudflare/clef", "cloudflare/clef-flash", "typesafe/jev-latest", "openai/decision") +) +@pytest.mark.parametrize( + "media", + ( + {"audio": ["data:audio/wav;base64,YQ=="]}, + {"videos": ["data:video/mp4;base64,YQ=="]}, + {"audio": ["data:audio/wav;base64,YQ=="], "videos": ["data:video/mp4;base64,YQ=="]}, + ), + ids=("audio", "videos", "both"), +) +@pytest.mark.asyncio +async def test_clips_for_a_model_without_media_support_are_refused_before_http( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, model: str, media: Mapping[str, object] +) -> None: + monkeypatch.setattr(litellm, "drop_params", False) + named_params: Final = str(list(media)) + + with pytest.raises( + litellm.UnsupportedParamsError, match=rf"does not support parameters: {re.escape(named_params)}, .*drop_params" + ): + await litellm.asystemone( + model=model, + state="test", + questions=_QUESTIONS, + api_base="https://example.com", + api_key="test", + **media, + ) + assert len(respx_mock.calls) == 0 + + +@pytest.mark.parametrize( + "clips", + ( + {"audio": ["data:audio/wav;base64,YQ=="], "videos": ["data:video/mp4;base64,YQ=="]}, + {"audio": ["https://example.com/clip.wav"], "videos": [{"content_type": "video/mp4"}]}, + ), + ids=("well_formed", "malformed"), +) +@pytest.mark.parametrize(("api", "request_kwargs"), _PREDICATE_REQUESTS, ids=_PREDICATE_REQUEST_IDS) +@pytest.mark.asyncio +async def test_clips_are_dropped_from_the_wire_for_a_model_without_media_support_under_drop_params( + api: str, + request_kwargs: Mapping[str, object], + clips: Mapping[str, object], + respx_mock: respx.MockRouter, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(litellm, "drop_params", False) + route: Final = respx_mock.post("https://api.perplexity.ai/v1/decisions").respond(json=_RESPONSE) + call: Final = getattr(litellm, f"a{api}") + + await call(model="perplexity/pplx-decider-v1-27b", api_key="caller-key", **request_kwargs) + await call(model="perplexity/pplx-decider-v1-27b", api_key="caller-key", drop_params=True, **request_kwargs, **clips) + + assert len(route.calls) == 2 + text_only_body: Final = json.loads(route.calls[0].request.content) + assert "audio" not in text_only_body and "videos" not in text_only_body + assert json.loads(route.calls[1].request.content) == text_only_body + + +@pytest.mark.parametrize( + "media", + ( + {"audio": ["https://example.com/clip.wav"]}, + {"videos": ["https://example.com/clip.mp4"]}, + {"audio": [{"content_type": "audio/wav"}]}, + {"videos": [{"content_type": "video/mp4"}]}, + {"audio": [{"content_type": "video/mp4", "base64": "YQ=="}]}, + {"videos": ["data:audio/wav;base64,YQ=="]}, + {"audio": ["data:audio/webm;codecs=opus;base64,YQ=="]}, + {"videos": ["data:video/webm;codecs=vp9;base64,YQ=="]}, + {"audio": ["data:audio/wav;base64,YQ=="] * 5}, + {"videos": ["data:video/mp4;base64,YQ=="] * 3}, + ), +) +@pytest.mark.asyncio +async def test_systemone_validates_media_before_http(respx_mock: respx.MockRouter, media: Mapping[str, object]) -> None: + with pytest.raises(litellm.BadRequestError, match="Invalid Decisions request"): + await litellm.asystemone( + model="cloudflare/clef-omni", + state="test", + questions=_QUESTIONS, + api_base="https://example.com", + api_key="test", + **media, + ) + assert len(respx_mock.calls) == 0 + + +@pytest.mark.parametrize( + "media", ({"audio": [], "videos": []}, {"audio": None, "videos": None}, {"audio": None, "videos": []}), ids=repr +) +@pytest.mark.asyncio +async def test_empty_or_null_media_keeps_text_only_request_unchanged( + respx_mock: respx.MockRouter, media: Mapping[str, object] +) -> None: + route: Final = respx_mock.post("https://example.com/v1/systemone").respond(200, json=dict(_RESPONSE)) + await litellm.asystemone( + model="typesafe/jev-latest", + state="test", + questions=_QUESTIONS, + api_base="https://example.com", + api_key="test", + **media, + ) + assert json.loads(route.calls[0].request.content) == { + "model": "jev-latest", + "state": "test", + "questions": _QUESTIONS, + } + + +@pytest.mark.asyncio +async def test_a_malformed_clip_for_a_model_that_takes_media_is_refused_even_under_drop_params( + respx_mock: respx.MockRouter, +) -> None: + with pytest.raises(litellm.BadRequestError, match="Invalid Decisions request"): + await litellm.asystemone( + model="cloudflare/clef-omni", + state="test", + questions=_QUESTIONS, + api_base="https://example.com", + api_key="test", + audio=["https://example.com/clip.wav"], + drop_params=True, + ) + assert len(respx_mock.calls) == 0 + + +@pytest.mark.asyncio +async def test_media_support_is_read_per_kind_from_the_cost_map( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, local_model_cost_map: None +) -> None: + monkeypatch.setattr(litellm, "drop_params", False) + monkeypatch.setitem( + litellm.model_cost, + "typesafe/jev-hears", + {"litellm_provider": "typesafe", "mode": "evaluation", "supports_audio_input": True}, + ) + route: Final = respx_mock.post("https://example.com/v1/systemone").respond(200, json=dict(_RESPONSE)) + audio: Final = ["data:audio/wav;base64,YQ=="] + + await litellm.asystemone( + model="typesafe/jev-hears", + state="test", + questions=_QUESTIONS, + api_base="https://example.com", + api_key="test", + audio=audio, + ) + with pytest.raises(litellm.UnsupportedParamsError, match=re.escape("does not support parameters: ['videos']")): + await litellm.asystemone( + model="typesafe/jev-hears", + state="test", + questions=_QUESTIONS, + api_base="https://example.com", + api_key="test", + audio=audio, + videos=["data:video/mp4;base64,YQ=="], + ) + + assert len(route.calls) == 1 + assert json.loads(route.calls[0].request.content) == { + "model": "jev-hears", + "state": "test", + "questions": _QUESTIONS, + "audio": audio, + } diff --git a/tests/unit/llms/base_llm/decisions/test_base_decisions_transformation.py b/tests/unit/llms/base_llm/decisions/test_base_decisions_transformation.py index a452a621d2e..454fa393ea8 100644 --- a/tests/unit/llms/base_llm/decisions/test_base_decisions_transformation.py +++ b/tests/unit/llms/base_llm/decisions/test_base_decisions_transformation.py @@ -178,3 +178,23 @@ def test_missing_or_mismatched_systemone_answers_are_refusals_left_out_of_the_re assert parsed.answers[1:] == (DecisionsIRRefusal(), DecisionsIRRefusal(), DecisionsIRRefusal()) assert tuple(ir_to_systemone_response(parsed, ir).answers) == ("damaged",) + + +def test_each_media_kind_is_gated_by_its_own_support_flag() -> None: + audio: Final = ["data:audio/wav;base64,YQ=="] + videos: Final = ["data:video/mp4;base64,YQ=="] + audio_only: Final = systemone_request_to_ir(_SYSTEMONE_BODY.validate_python({**_SYSTEMONE_REQUEST, "audio": audio})) + video_only: Final = systemone_request_to_ir(_SYSTEMONE_BODY.validate_python({**_SYSTEMONE_REQUEST, "videos": videos})) + + assert ir_to_systemone_request("clef-omni", audio_only, supports_audio=True) == { + "model": "clef-omni", + **_SYSTEMONE_REQUEST, + "audio": audio, + } + assert ir_to_systemone_request("clef-omni", video_only, supports_video=True) == { + "model": "clef-omni", + **_SYSTEMONE_REQUEST, + "videos": videos, + } + assert isinstance(ir_to_systemone_request("clef-omni", audio_only, supports_video=True), UnsupportedDecisionsRequest) + assert isinstance(ir_to_systemone_request("clef-omni", video_only, supports_audio=True), UnsupportedDecisionsRequest) diff --git a/tests/unit/proxy/decisions_endpoints/test_endpoints.py b/tests/unit/proxy/decisions_endpoints/test_endpoints.py index 81d7f8eb7ac..2f7cbbc8071 100644 --- a/tests/unit/proxy/decisions_endpoints/test_endpoints.py +++ b/tests/unit/proxy/decisions_endpoints/test_endpoints.py @@ -624,6 +624,115 @@ def test_openai_format_decisions_reach_an_openai_deployment_unchanged_including_ assert float(response.headers["x-litellm-response-cost"]) == pytest.approx(expected_cost) +_CLEF_OMNI_MEDIA: Final[Mapping[str, object]] = { + "audio": ["data:audio/wav;base64,YQ==", {"content_type": "audio/webm;codecs=opus", "base64": "Yg=="}], + "videos": ["data:video/mp4;base64,Yw=="], +} +_CLEF_OMNI_RESPONSE: Final[Mapping[str, object]] = { + "success": True, + "result": { + "model": "clef-omni", + "answers": {"heard": {"type": "noul", "noul": 0.9}}, + "usage": {"input_tokens": 100, "output_tokens": 0}, + }, +} +_CLEF_OMNI_UPSTREAM: Final = "https://cf.example/ai/run/@cf/cloudflare/clef-omni" + + +def _route_to_clef_omni(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + litellm.proxy.proxy_server, + "llm_router", + litellm.Router( + model_list=[ + { + "model_name": "clef-omni", + "litellm_params": { + "model": "cloudflare/clef-omni", + "api_base": "https://cf.example", + "api_key": "cf-key", + }, + }, + { + "model_name": "decider", + "litellm_params": {"model": "perplexity/pplx-decider-v1-27b", "api_key": "k"}, + }, + ] + ), + ) + + +def _clef_omni_request(endpoint: str) -> Mapping[str, object]: + if endpoint.endswith("systemone"): + return { + "model": "clef-omni", + "state": "Evaluate the clips", + "questions": {"heard": {"type": "noul", "instructions": "Is there speech?"}}, + **_CLEF_OMNI_MEDIA, + } + return { + "model": "clef-omni", + "input": "Evaluate the clips", + "questions": [{"type": "predicate", "name": "heard", "instructions": "Is there speech?"}], + **_CLEF_OMNI_MEDIA, + } + + +@pytest.mark.parametrize("endpoint", ("/v1/systemone", "/systemone", "/v1/decisions", "/decisions")) +def test_proxy_forwards_clef_omni_media_on_every_decisions_route( + client: TestClient, + monkeypatch: pytest.MonkeyPatch, + respx_mock: respx.MockRouter, + endpoint: str, +) -> None: + _route_to_clef_omni(monkeypatch) + upstream: Final = respx_mock.post(_CLEF_OMNI_UPSTREAM).respond(json=_CLEF_OMNI_RESPONSE) + + response: Final = client.post(endpoint, json=_clef_omni_request(endpoint)) + + assert response.status_code == 200, response.text + assert response.json()["answers"] == ( + {"heard": {"type": "noul", "noul": 0.9}} + if endpoint.endswith("systemone") + else [{"type": "predicate", "name": "heard", "probability": 0.9}] + ) + assert json.loads(upstream.calls[0].request.content) == { + "model": "clef-omni", + "state": "Evaluate the clips", + "questions": {"heard": {"type": "noul", "instructions": "Is there speech?"}}, + **_CLEF_OMNI_MEDIA, + } + + +@pytest.mark.parametrize( + ("endpoint", "request_body"), + ( + ("/v1/systemone", {**_REQUEST, **_CLEF_OMNI_MEDIA}), + ("/v1/decisions", {**_clef_omni_request("/v1/decisions"), "audio": [{"content_type": "audio/wav"}]}), + ("/systemone", {**_clef_omni_request("/systemone"), "audio": ["data:image/png;base64,AA=="]}), + ("/decisions", {**_clef_omni_request("/decisions"), "videos": ["data:video/mp4;base64,Yw=="] * 3}), + ), + ids=("model_without_media_support", "clip_missing_base64", "audio_slot_holding_an_image", "too_many_videos"), +) +def test_proxy_refuses_decisions_media_it_cannot_send( + client: TestClient, + monkeypatch: pytest.MonkeyPatch, + respx_mock: respx.MockRouter, + endpoint: str, + request_body: Mapping[str, object], +) -> None: + monkeypatch.setattr(litellm, "drop_params", False) + _route_to_clef_omni(monkeypatch) + clef_upstream: Final = respx_mock.post(_CLEF_OMNI_UPSTREAM).respond(json=_CLEF_OMNI_RESPONSE) + perplexity_upstream: Final = respx_mock.post("https://api.perplexity.ai/v1/decisions").respond(json=_RESPONSE) + + response: Final = client.post(endpoint, json=request_body) + + assert response.status_code == 400, response.text + assert not clef_upstream.called + assert not perplexity_upstream.called + + @pytest.mark.parametrize("safety_identifier", (7, ["end-user-1"]), ids=("numeric", "list")) @pytest.mark.parametrize("deployment_drops_params", (False, True), ids=("strict", "drop_params")) def test_a_non_string_safety_identifier_is_refused_unless_the_deployment_drops_params( @@ -686,6 +795,119 @@ def test_a_non_string_safety_identifier_is_refused_unless_the_deployment_drops_p } +@pytest.mark.parametrize("endpoint", ("/v1/systemone", "/v1/decisions")) +@pytest.mark.parametrize("deployment_drops_params", (False, True), ids=("strict", "drop_params")) +def test_clips_for_a_model_without_media_support_are_refused_unless_the_deployment_drops_params( + client: TestClient, + monkeypatch: pytest.MonkeyPatch, + respx_mock: respx.MockRouter, + endpoint: str, + deployment_drops_params: bool, +) -> None: + monkeypatch.setattr(litellm, "drop_params", False) + monkeypatch.setattr( + litellm.proxy.proxy_server, + "llm_router", + litellm.Router( + model_list=[ + { + "model_name": "clef-flash", + "litellm_params": { + "model": "cloudflare/clef-flash", + "api_base": "https://cf.example", + "api_key": "cf-key", + "drop_params": deployment_drops_params, + }, + } + ] + ), + ) + upstream: Final = respx_mock.post("https://cf.example/ai/run/@cf/cloudflare/clef-flash").respond( + json=_CLEF_OMNI_RESPONSE + ) + + response: Final = client.post(endpoint, json={**_clef_omni_request(endpoint), "model": "clef-flash"}) + + if not deployment_drops_params: + assert response.status_code == 400, response.text + assert "['audio', 'videos']" in response.json()["error"]["message"] + assert not upstream.called + return + assert response.status_code == 200, response.text + assert json.loads(upstream.calls[0].request.content) == { + "model": "clef-flash", + "state": "Evaluate the clips", + "questions": {"heard": {"type": "noul", "instructions": "Is there speech?"}}, + } + + +@pytest.mark.parametrize("endpoint", ("/v1/systemone", "/systemone", "/v1/decisions", "/decisions")) +def test_null_media_on_every_decisions_route_means_no_clips( + client: TestClient, + monkeypatch: pytest.MonkeyPatch, + respx_mock: respx.MockRouter, + endpoint: str, +) -> None: + _route_to_clef_omni(monkeypatch) + upstream: Final = respx_mock.post(_CLEF_OMNI_UPSTREAM).respond(json=_CLEF_OMNI_RESPONSE) + + response: Final = client.post(endpoint, json={**_clef_omni_request(endpoint), "audio": None, "videos": None}) + + assert response.status_code == 200, response.text + assert json.loads(upstream.calls[0].request.content) == { + "model": "clef-omni", + "state": "Evaluate the clips", + "questions": {"heard": {"type": "noul", "instructions": "Is there speech?"}}, + } + + +@pytest.mark.parametrize("endpoint", ("/v1/systemone", "/v1/decisions")) +@pytest.mark.parametrize("deployment_drops_params", (False, True), ids=("strict", "drop_params")) +def test_a_malformed_clip_for_a_model_without_media_support_is_dropped_only_when_the_deployment_drops_params( + client: TestClient, + monkeypatch: pytest.MonkeyPatch, + respx_mock: respx.MockRouter, + endpoint: str, + deployment_drops_params: bool, +) -> None: + monkeypatch.setattr(litellm, "drop_params", False) + monkeypatch.setattr( + litellm.proxy.proxy_server, + "llm_router", + litellm.Router( + model_list=[ + { + "model_name": "clef-flash", + "litellm_params": { + "model": "cloudflare/clef-flash", + "api_base": "https://cf.example", + "api_key": "cf-key", + "drop_params": deployment_drops_params, + }, + } + ] + ), + ) + upstream: Final = respx_mock.post("https://cf.example/ai/run/@cf/cloudflare/clef-flash").respond( + json=_CLEF_OMNI_RESPONSE + ) + malformed: Final = {"audio": ["https://example.com/clip.wav"], "videos": [{"content_type": "video/mp4"}]} + + response: Final = client.post(endpoint, json={**_clef_omni_request(endpoint), "model": "clef-flash", **malformed}) + + if not deployment_drops_params: + assert response.status_code == 400, response.text + assert "Invalid Decisions request" in response.json()["error"]["message"] + assert not upstream.called + return + assert response.status_code == 200, response.text + assert json.loads(upstream.calls[0].request.content) == { + "model": "clef-flash", + "state": "Evaluate the clips", + "questions": {"heard": {"type": "noul", "instructions": "Is there speech?"}}, + } + + def _decisions_feature() -> LazyFeature: return next(feature for feature in LAZY_FEATURES if feature.name == "decisions") diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index 466f3c988e4..33a3328e0b5 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -438,6 +438,26 @@ def test_supports_audio_output_reads_its_own_cost_map_flag( assert litellm.supports_audio_output(model) is audio_output +@pytest.mark.parametrize(("audio_input", "video_input"), [(True, False), (False, True)]) +def test_supports_video_input_reads_its_own_cost_map_flag( + local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch, audio_input: bool, video_input: bool +) -> None: + model: Final = "openai/video-flags-disagree-model" + monkeypatch.setitem( + litellm.model_cost, + model, + { + "litellm_provider": "openai", + "mode": "chat", + "supports_audio_input": audio_input, + "supports_video_input": video_input, + }, + ) + assert litellm.supports_video_input(model) is video_input + assert litellm.supports_audio_input(model) is audio_input + assert litellm.get_model_info(model)["supports_video_input"] is video_input + + def test_get_optional_params_image_gen(): from litellm.llms.azure.image_generation import AzureGPTImageGenerationConfig diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index ada0bb2b8a2..9e83e478131 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -56955,6 +56955,23 @@ export interface operations { * } */ "application/json": { + /** + * Audio + * @description Clef-omni only. Up to 4 embedded audio clips, each a base64 data URL or {content_type, base64} + * @default [] + */ + audio?: (string | { + /** + * Base64 + * @description Base64-encoded audio bytes + */ + base64: string; + /** + * Content Type + * @description An audio/* media type, such as audio/wav + */ + content_type: string; + })[]; /** * Input * @description The text being judged, or user messages whose content mixes input_text and input_image parts @@ -57076,6 +57093,23 @@ export interface operations { * @description A stable id for your end user. Sent to OpenAI and dropped for other providers */ safety_identifier?: string | null; + /** + * Videos + * @description Clef-omni only. Up to 2 embedded videos, each a base64 data URL or {content_type, base64} + * @default [] + */ + videos?: (string | { + /** + * Base64 + * @description Base64-encoded video bytes + */ + base64: string; + /** + * Content Type + * @description A video/* media type, such as video/mp4 + */ + content_type: string; + })[]; } & { [key: string]: unknown; }; @@ -71249,6 +71283,23 @@ export interface operations { * } */ "application/json": { + /** + * Audio + * @description Clef-omni only. Up to 4 embedded audio clips, each a base64 data URL or {content_type, base64} + * @default [] + */ + audio?: (string | { + /** + * Base64 + * @description Base64-encoded audio bytes + */ + base64: string; + /** + * Content Type + * @description An audio/* media type, such as audio/wav + */ + content_type: string; + })[]; /** * Model * @description A decision model from the proxy model_list @@ -71338,6 +71389,23 @@ export interface operations { state: string | { [key: string]: unknown; } | unknown[]; + /** + * Videos + * @description Clef-omni only. Up to 2 embedded videos, each a base64 data URL or {content_type, base64} + * @default [] + */ + videos?: (string | { + /** + * Base64 + * @description Base64-encoded video bytes + */ + base64: string; + /** + * Content Type + * @description A video/* media type, such as video/mp4 + */ + content_type: string; + })[]; } & { [key: string]: unknown; }; @@ -76556,6 +76624,23 @@ export interface operations { * } */ "application/json": { + /** + * Audio + * @description Clef-omni only. Up to 4 embedded audio clips, each a base64 data URL or {content_type, base64} + * @default [] + */ + audio?: (string | { + /** + * Base64 + * @description Base64-encoded audio bytes + */ + base64: string; + /** + * Content Type + * @description An audio/* media type, such as audio/wav + */ + content_type: string; + })[]; /** * Input * @description The text being judged, or user messages whose content mixes input_text and input_image parts @@ -76677,6 +76762,23 @@ export interface operations { * @description A stable id for your end user. Sent to OpenAI and dropped for other providers */ safety_identifier?: string | null; + /** + * Videos + * @description Clef-omni only. Up to 2 embedded videos, each a base64 data URL or {content_type, base64} + * @default [] + */ + videos?: (string | { + /** + * Base64 + * @description Base64-encoded video bytes + */ + base64: string; + /** + * Content Type + * @description A video/* media type, such as video/mp4 + */ + content_type: string; + })[]; } & { [key: string]: unknown; }; @@ -79886,6 +79988,23 @@ export interface operations { * } */ "application/json": { + /** + * Audio + * @description Clef-omni only. Up to 4 embedded audio clips, each a base64 data URL or {content_type, base64} + * @default [] + */ + audio?: (string | { + /** + * Base64 + * @description Base64-encoded audio bytes + */ + base64: string; + /** + * Content Type + * @description An audio/* media type, such as audio/wav + */ + content_type: string; + })[]; /** * Model * @description A decision model from the proxy model_list @@ -79975,6 +80094,23 @@ export interface operations { state: string | { [key: string]: unknown; } | unknown[]; + /** + * Videos + * @description Clef-omni only. Up to 2 embedded videos, each a base64 data URL or {content_type, base64} + * @default [] + */ + videos?: (string | { + /** + * Base64 + * @description Base64-encoded video bytes + */ + base64: string; + /** + * Content Type + * @description A video/* media type, such as video/mp4 + */ + content_type: string; + })[]; } & { [key: string]: unknown; };