feat(decisions): support Clef-omni audio and video (#45811)

* feat(decisions): support Clef-omni audio and video

* fix(decisions): accept Clef-omni media types with parameters and cover the proxy routes

The clip patterns now mirror Cloudflare's input schema, so a data URL or
content_type carrying parameters like audio/webm;codecs=opus is accepted.
The four proxy routes get media tests of their own, the backup cost map
is synced with the new Clef-omni rows, and the docs page moves to the
docs repo as a follow-up.

* fix(decisions): refuse parameterized media-type data URLs and refresh the OpenAPI snapshot

Cloudflare answers 422 to a data URL whose media type carries parameters even though its schema only requires the data:audio/ prefix, so LiteLLM now refuses those with a 400 before the call. The object form keeps passing parameters through, which Cloudflare accepts. The lazy OpenAPI snapshot and the dashboard API types are regenerated for the new audio and videos fields

* fix(decisions): gate media by cost-map flags, treat null clips as none, drop unsupported clips before validation

* fix(decisions): keep the SDK's positional order and gate media flags on evaluation-mode models

* fix(decisions): gate media on both decisions model modes and move the audit cells onto the harness helpers

* test(decisions): cover the Fireworks System One router in the media refusal cells

---------

Co-authored-by: moyai-devin-berriai[bot] <336287033+moyai-devin-berriai[bot]@users.noreply.github.com>
Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
moyai-devin-berriai[bot] 2026-10-12 20:29:49 +00:00 • committed by GitHub
parent 87478c7b6d
commit a86ad9ebf8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 2437 additions and 29 deletions

View file

@ -2280,6 +2280,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]

View file

@ -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",
@ -482,6 +483,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"),

View file

@ -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"
),

View file

@ -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,
@ -83,6 +94,8 @@ async def adecisions(
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
extra_headers: Mapping[str, str] | None = None,
audio: DecisionsAudioClips | None = None,
videos: DecisionsVideoClips | None = None,
**kwargs: object,
) -> OpenAIDecisionResponse | DecisionsResponse:
call: Final = _prepare_decisions_call(
@ -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,
@ -111,6 +126,8 @@ def decisions(
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
extra_headers: Mapping[str, str] | None = None,
audio: DecisionsAudioClips | None = None,
videos: DecisionsVideoClips | None = None,
**kwargs: object,
) -> OpenAIDecisionResponse | DecisionsResponse:
call: Final = _prepare_decisions_call(
@ -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,

View file

@ -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 (
@ -47,6 +48,7 @@ from litellm.types.decisions import (
UnsupportedDecisionsRequest,
systemone_choice_key,
)
from litellm.types.utils import is_decisions_model_mode
_PAYLOAD_ADAPTER: Final[TypeAdapter[object]] = TypeAdapter(object)
_SYSTEMONE_RESPONSE_ADAPTER: Final[TypeAdapter[DecisionsResponse]] = TypeAdapter(DecisionsResponse)
@ -100,6 +102,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 +170,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 +183,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},
}
@ -296,6 +314,14 @@ def ir_to_systemone_response(response: DecisionsIRResponse, request: DecisionsIR
)
def _is_decisions_model(model: str, custom_llm_provider: str) -> bool:
try:
model_mode: Final = litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider).get("mode")
except Exception: # noqa: BLE001 # get_model_info raises bare Exception for an unmapped model, which has no mode
return False
return is_decisions_model_mode(model_mode)
class BaseDecisionsConfig(ABC):
path: str = "/v1/systemone"
api_key_env: tuple[str, ...] = ()
@ -355,13 +381,28 @@ 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 _is_decisions_model(model, custom_llm_provider) and 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 _is_decisions_model(model, custom_llm_provider) and 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

View file

@ -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 {
@ -309,6 +313,12 @@ class OpenAIDecisionsConfig(BaseDecisionsConfig):
def resolve_api_key(self, api_key: str | None) -> str | None:
return api_key or litellm.api_key or litellm.openai_key or get_secret_str("OPENAI_API_KEY")
def supports_audio_input(self, model: str, custom_llm_provider: str) -> bool:
return False
def supports_video_input(self, model: str, custom_llm_provider: str) -> bool:
return False
def transform_decisions_request(
self,
model: str,

View file

@ -16022,6 +16022,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",

View file

@ -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": [

View file

@ -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]:

View file

@ -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,

View file

@ -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)

View file

@ -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

View file

@ -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]

View file

@ -16022,6 +16022,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",

View file

@ -0,0 +1,293 @@
import asyncio
import json
import re
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__)
with owned.replacing_worker(victim_pid) as kill:
kill.freeze()
await asyncio.to_thread(kill)
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)))

View file

@ -0,0 +1,856 @@
import asyncio
import json
import math
import re
import threading
import time
import uuid
from collections.abc import Mapping, 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, string_value
from integration._support.database import read_rows
from integration._support.timing import assert_within_factor_of_control
from integration._support.wire import Reply, Request, wire_server
from pydantic import JsonValue, TypeAdapter
import litellm
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())
if request.target.endswith("/systemone"):
return Reply(body=json.dumps(_upstream_reply(marker)["result"]).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
@pytest.mark.parametrize("route", _ROUTES)
def test_null_clips_mean_no_clips_of_that_kind_on_every_route(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)
both_null: Final = _decide(gateway, route, model, marker, audio=None, videos=None)
audio_only: Final = _decide(gateway, route, model, marker, audio=[_WAV], videos=None)
video_only: Final = _decide(gateway, route, model, marker, audio=None, videos=[_MP4])
for response in (both_null, audio_only, video_only):
assert response.status_code == 200, response.text
assert response.json() == _expected_response(route, marker)
bodies: Final = tuple(_upstream_body(request) for request in wire.drain())
assert bodies == (
_expected_upstream_body(marker),
_expected_upstream_body(marker, audio=[_WAV]),
_expected_upstream_body(marker, videos=[_MP4]),
), bodies
_SDK_QUESTIONS: Final = TypeAdapter(dict[str, dict[str, object]]).validate_python(_SYSTEM_ONE_QUESTIONS)
async def test_positional_sdk_callers_keep_the_api_key_fifth_and_the_api_base_sixth() -> None:
marker: Final = uuid.uuid4().hex
with wire_server(_echo) as wire:
synchronous: Final = litellm.decisions(
_CANONICAL, None, _SDK_QUESTIONS, None, _API_KEY, wire.url, state=_state(marker), audio=[_WAV]
)
asynchronous: Final = await litellm.adecisions(
_CANONICAL, None, _SDK_QUESTIONS, None, _API_KEY, wire.url, state=_state(marker), videos=[_MP4]
)
for response in (synchronous, asynchronous):
assert response.model_dump(mode="json") == _expected_response("/v1/systemone", marker)
bodies: Final = tuple(_upstream_body(request) for request in wire.drain())
assert bodies == (
_expected_upstream_body(marker, audio=[_WAV]),
_expected_upstream_body(marker, videos=[_MP4]),
), 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"),
("fireworks_ai/accounts/fireworks/routers/rune", "/v1/systemone"),
("fireworks_ai/accounts/fireworks/routers/rune", "/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()
_MALFORMED_CLIPS: Final[dict[str, JsonValue]] = {
"audio": ["https://example.invalid/clip.wav"],
"videos": [{"content_type": "video/mp4"}],
}
_NO_MEDIA_SPELLINGS: Final = ("cloudflare/clef-flash", "cloudflare/clef")
def _assert_refused_before_any_upstream_call(response: httpx.Response, fragment: str, wire_received: int) -> None:
assert response.status_code == 400, response.text
assert fragment in response.text, response.text
assert wire_received == 0
@pytest.mark.parametrize("route", ("/v1/systemone", "/v1/decisions"))
def test_malformed_clips_on_a_model_without_support_are_refused_unless_the_deployment_drops_params(
gateway: Gateway, route: Route
) -> None:
marker: Final = uuid.uuid4().hex
with wire_server(_echo) as wire, gateway.scenario() as scenario:
strict: Final = scenario.model(model="cloudflare/clef-flash", api_base=wire.url, api_key=_API_KEY)
dropping: Final = scenario.model(
model="cloudflare/clef-flash", api_base=wire.url, api_key=_API_KEY, drop_params=True
)
refused: Final = _decide(gateway, route, strict, marker, **_MALFORMED_CLIPS)
_assert_refused_before_any_upstream_call(refused, "Invalid Decisions request", len(wire.drain()))
text_only: Final = _decide(gateway, route, dropping, marker)
assert text_only.status_code == 200, text_only.text
(text_only_request,) = wire.drain()
dropped: Final = _decide(gateway, route, dropping, marker, **_MALFORMED_CLIPS)
assert dropped.status_code == 200, dropped.text
(dropped_request,) = wire.drain()
assert _JSON_OBJECT.validate_json(dropped_request.body) == _JSON_OBJECT.validate_json(text_only_request.body)
assert dropped.json() == text_only.json()
def test_a_malformed_clip_on_a_media_model_is_refused_even_when_the_deployment_drops_params(
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, drop_params=True)
refused: Final = _decide(gateway, "/v1/systemone", model, marker, **_MALFORMED_CLIPS)
_assert_refused_before_any_upstream_call(refused, "Invalid Decisions request", len(wire.drain()))
served: Final = _decide(gateway, "/v1/systemone", model, marker, audio=[_WAV], videos=[_MP4])
(request,) = wire.drain()
_assert_served_with_media(served, "/v1/systemone", marker, request)
def _advertised_media_flags(gateway: Gateway, deployments: Mapping[str, str]) -> dict[str, tuple[JsonValue, JsonValue]]:
listed: Final = list_value(gateway.get("/model/info")["data"])
entries: Final = {
string_value(object_value(item)["model_name"]): object_value(object_value(item)["model_info"])
for item in listed
}
return {
spelling: (entries[name].get("supports_audio_input"), entries[name].get("supports_video_input"))
for spelling, name in deployments.items()
}
def test_model_info_advertises_media_support_only_on_clef_omni_deployments(gateway: Gateway) -> None:
with wire_server(_echo) as wire, gateway.scenario() as scenario:
deployments: Final = {
spelling: scenario.model(model=spelling, api_base=wire.url, api_key=_API_KEY)
for spelling in (_ALIAS, _CANONICAL, *_NO_MEDIA_SPELLINGS)
}
flags: Final = _advertised_media_flags(gateway, deployments)
assert flags[_ALIAS] == (True, True), flags
assert flags[_CANONICAL] == (True, True), flags
advertised_without_media: Final = tuple(any(flags[spelling]) for spelling in _NO_MEDIA_SPELLINGS)
assert advertised_without_media == (False, False), flags
def _unique_backend(kind: str) -> str:
return f"@cf/cloudflare/clef-{kind}-{uuid.uuid4().hex[:12]}"
def _body_sent_to(request: Request, backend: str) -> dict[str, JsonValue]:
assert request.method == "POST", request.method
assert unquote(request.target) == f"/ai/run/{backend}", request.target
assert request.headers["authorization"] == f"Bearer {_API_KEY}", request.headers
return _JSON_OBJECT.validate_json(request.body)
def _expected_backend_body(backend: str, marker: str, **media: JsonValue) -> dict[str, JsonValue]:
return {**_expected_upstream_body(marker, **media), "model": backend.rsplit("/", 1)[-1]}
def test_a_deployment_advertising_only_audio_forwards_audio_and_refuses_or_drops_video(gateway: Gateway) -> None:
marker: Final = uuid.uuid4().hex
backend: Final = _unique_backend("audio")
advertised: Final[dict[str, JsonValue]] = {"mode": "evaluation", "supports_audio_input": True}
with wire_server(_echo) as wire, gateway.scenario() as scenario:
strict: Final = scenario.model(
model=f"cloudflare/{backend}", api_base=wire.url, api_key=_API_KEY, model_info=advertised
)
dropping: Final = scenario.model(
model=f"cloudflare/{backend}", api_base=wire.url, api_key=_API_KEY, drop_params=True, model_info=advertised
)
with_audio: Final = _decide(gateway, "/v1/systemone", strict, marker, audio=[_WAV])
assert with_audio.status_code == 200, with_audio.text
(audio_request,) = wire.drain()
assert _body_sent_to(audio_request, backend) == _expected_backend_body(backend, marker, audio=[_WAV])
with_video: Final = _decide(gateway, "/v1/systemone", strict, marker, videos=[_MP4])
_assert_refused_before_any_upstream_call(
with_video, "does not support parameters: ['videos']", len(wire.drain())
)
assert "drop_params" in with_video.text, with_video.text
dropped: Final = _decide(gateway, "/v1/systemone", dropping, marker, audio=[_WAV], videos=[_MP4])
assert dropped.status_code == 200, dropped.text
(dropped_request,) = wire.drain()
assert _body_sent_to(dropped_request, backend) == _expected_backend_body(backend, marker, audio=[_WAV])
def test_a_chat_mode_deployment_advertising_media_is_not_a_decisions_media_model(gateway: Gateway) -> None:
marker: Final = uuid.uuid4().hex
backend: Final = _unique_backend("chat")
advertised: Final[dict[str, JsonValue]] = {
"mode": "chat",
"supports_audio_input": True,
"supports_video_input": True,
}
with wire_server(_echo) as wire, gateway.scenario() as scenario:
strict: Final = scenario.model(
model=f"cloudflare/{backend}", api_base=wire.url, api_key=_API_KEY, model_info=advertised
)
dropping: Final = scenario.model(
model=f"cloudflare/{backend}", api_base=wire.url, api_key=_API_KEY, drop_params=True, model_info=advertised
)
refused: Final = _decide(gateway, "/v1/systemone", strict, marker, audio=[_WAV], videos=[_MP4])
_assert_refused_before_any_upstream_call(refused, "does not support parameters", len(wire.drain()))
assert "'audio'" in refused.text and "'videos'" in refused.text, refused.text
assert "drop_params" in refused.text, refused.text
text_only: Final = _decide(gateway, "/v1/systemone", dropping, marker)
assert text_only.status_code == 200, text_only.text
(text_only_request,) = wire.drain()
assert _body_sent_to(text_only_request, backend) == _expected_backend_body(backend, marker)
dropped: Final = _decide(gateway, "/v1/systemone", dropping, marker, audio=[_WAV], videos=[_MP4])
assert dropped.status_code == 200, dropped.text
(dropped_request,) = wire.drain()
assert _body_sent_to(dropped_request, backend) == _expected_backend_body(backend, marker)
assert dropped.json() == text_only.json()
def test_clips_on_an_openai_chat_model_are_refused_or_dropped_never_forwarded(gateway: Gateway) -> None:
marker: Final = uuid.uuid4().hex
with wire_server(_echo) as wire, gateway.scenario() as scenario:
strict: Final = scenario.model(model="openai/gpt-audio", api_base=wire.url, api_key=_API_KEY)
dropping: Final = scenario.model(
model="openai/gpt-audio", api_base=wire.url, api_key=_API_KEY, drop_params=True
)
refused: Final = _decide(gateway, "/v1/decisions", strict, marker, audio=[_WAV])
_assert_refused_before_any_upstream_call(refused, "does not support parameters: ['audio']", len(wire.drain()))
assert "drop_params" in refused.text, refused.text
text_only: Final = _decide(gateway, "/v1/decisions", dropping, marker)
assert text_only.status_code == 200, text_only.text
(text_only_request,) = wire.drain()
assert unquote(text_only_request.target) == "/v1/decisions", text_only_request.target
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
dropped: Final = _decide(gateway, "/v1/decisions", dropping, marker, audio=[_WAV], videos=[_MP4])
assert dropped.status_code == 200, dropped.text
(dropped_request,) = wire.drain()
assert _JSON_OBJECT.validate_json(dropped_request.body) == text_only_body
assert dropped.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 _timed_refusal_under_liveliness_probes(
gateway: Gateway, model: str, clip: str
) -> tuple[httpx.Response, float, tuple[float, ...]]:
stop: Final = threading.Event()
latencies: Final[SimpleQueue[float]] = SimpleQueue()
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, uuid.uuid4().hex, audio=[clip]),
headers={"Authorization": f"Bearer {gateway.key}"},
)
elapsed: Final = time.perf_counter() - started
stop.set()
probe.join(timeout=15)
return response, elapsed, tuple(latencies.get_nowait() for _ in range(latencies.qsize()))
def test_the_data_url_pattern_worst_case_is_refused_in_bounded_time_while_liveliness_answers(
gateway: Gateway,
) -> None:
with wire_server(_echo) as wire, gateway.scenario() as scenario:
model: Final = scenario.model(model=_CANONICAL, api_base=wire.url, api_key=_API_KEY)
tenth_response, tenth_elapsed, _ = _timed_refusal_under_liveliness_probes(
gateway, model, "data:audio/" + "a" * (_WORST_CASE_LENGTH // 10)
)
response, elapsed, observed = _timed_refusal_under_liveliness_probes(
gateway, model, "data:audio/" + "a" * _WORST_CASE_LENGTH
)
timing: Final = {
"tenth_seconds": round(tenth_elapsed, 3),
"classifier_seconds": round(elapsed, 3),
"slowest_liveliness_seconds": round(max(observed), 3),
}
assert tenth_response.status_code == 400, tenth_response.text[:300]
assert_within_factor_of_control(elapsed, control=tenth_elapsed, factor=25)
assert observed and max(observed) < max(1.0, elapsed / 2), timing
assert response.status_code == 400, (timing, response.text[:300])
assert "pattern" in response.text, response.text[:300]
assert wire.drain() == (), timing

View file

@ -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
@ -1732,6 +1733,328 @@ 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",
"fireworks_ai/accounts/fireworks/routers/rune",
),
)
@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
@pytest.mark.parametrize("mode", ("evaluation", "decisions"))
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, mode: str
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
monkeypatch.setitem(
litellm.model_cost,
"typesafe/jev-hears",
{"litellm_provider": "typesafe", "mode": mode, "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,
}
@pytest.mark.asyncio
@pytest.mark.parametrize(
("model", "url", "cost_map_entry"),
(
(
"typesafe/jev-chatty",
"https://example.com/v1/systemone",
{"litellm_provider": "typesafe", "mode": "chat", "supports_audio_input": True, "supports_video_input": True},
),
("openai/gpt-audio", "https://example.com/v1/decisions", None),
),
)
async def test_a_chat_models_media_flags_do_not_count_for_decisions(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
local_model_cost_map: None,
model: str,
url: str,
cost_map_entry: Mapping[str, object] | None,
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
if cost_map_entry is not None:
monkeypatch.setitem(litellm.model_cost, model, cost_map_entry)
assert litellm.supports_audio_input(model) is True
route: Final = respx_mock.post(url).respond(200, json=dict(_RESPONSE if "systemone" in url else _OPENAI_RESPONSE))
questions: Final = {"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}
audio: Final = ["data:audio/wav;base64,YQ=="]
videos: Final = ["data:video/mp4;base64,YQ=="]
with pytest.raises(litellm.UnsupportedParamsError, match=re.escape("does not support parameters: ['audio', 'videos']")):
await litellm.asystemone(
model=model,
state="test",
questions=questions,
api_base="https://example.com",
api_key="test",
audio=audio,
videos=videos,
)
assert not route.calls
await litellm.asystemone(
model=model,
state="test",
questions=questions,
api_base="https://example.com",
api_key="test",
audio=audio,
videos=videos,
drop_params=True,
)
sent: Final = json.loads(route.calls[0].request.content)
assert "audio" not in sent
assert "videos" not in sent
assert sent["model"] == model.split("/", 1)[1]
def test_positional_arguments_keep_the_order_the_sdk_had_before_media(respx_mock: respx.MockRouter) -> None:
decisions_route: Final = respx_mock.post("https://example.com/v1/decisions").respond(json=_OPENAI_RESPONSE)
systemone_route: Final = respx_mock.post("https://example.com/v1/systemone").respond(json=_RESPONSE)
openai_questions: Final = [{"type": "predicate", "name": "is_defect", "instructions": "Is this a defect?"}]
systemone_questions: Final = {"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}
litellm.decisions("openai/gpt-6-luna", "review", openai_questions, None, "positional-key", "https://example.com")
litellm.systemone("typesafe/jev-1.13", "review", systemone_questions, "positional-key", "https://example.com")
assert decisions_route.calls[0].request.headers["authorization"] == "Bearer positional-key"
assert systemone_route.calls[0].request.headers["authorization"] == "Bearer positional-key"
for sent in (decisions_route.calls[0].request.content, systemone_route.calls[0].request.content):
assert "audio" not in json.loads(sent)
_FIREWORKS_URL: Final = "https://api.fireworks.ai/inference/v1/systemone"
_FIREWORKS_MODEL: Final = "fireworks_ai/accounts/fireworks/routers/rune"
_FIREWORKS_RESPONSE: Final[Mapping[str, object]] = {

View file

@ -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)

View file

@ -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")

View file

@ -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

View file

@ -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;
};