from __future__ import annotations import asyncio import json import re import threading from collections.abc import Mapping from concurrent.futures import wait from types import MappingProxyType from typing import Final import httpx import pytest import respx from botocore.auth import SigV4Auth from botocore.awsrequest import AWSRequest from botocore.credentials import Credentials import litellm 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 from litellm.types.decisions import ( ChoiceAnswer, DecisionsResponse, DecisionsUsage, NoulAnswer, OpenAIDecisionResponse, ScoreAnswer, ) from litellm.types.utils import LlmProviders from litellm.utils import ProviderConfigManager _QUESTIONS: Final[Mapping[str, object]] = MappingProxyType( { "is_defect": {"type": "noul", "instructions": "Is this a defect?", "provider_field": "kept"}, "sentiment": {"type": "choice", "criteria": {"positive": None, "negative": "unhappy"}}, "severity": {"type": "score", "criteria": ["none", "low", "high"]}, } ) _INPUT_TOKENS: Final[int] = 367 _OUTPUT_TOKENS: Final[int] = 3 _CACHED_TOKENS: Final[int] = 256 _CACHE_WRITE_TOKENS: Final[int] = 64 _RESPONSE: Final[Mapping[str, object]] = { "model": "jev-1.13", "answers": { "is_defect": {"type": "noul", "noul": 0.9}, "sentiment": { "type": "choice", "choice": "positive", "confidence": 0.8, "probabilities": {"positive": 0.8, "negative": 0.2}, }, "severity": { "type": "score", "score": 1, "confidence": 0.7, "legend": {"0": "none", "1": "low", "2": "high"}, "probabilities": {"0": 0.1, "1": 0.8, "2": 0.1}, }, }, "usage": {"input_tokens": _INPUT_TOKENS, "output_tokens": _OUTPUT_TOKENS}, } _STRANDS_RESPONSE: Final[Mapping[str, object]] = { "model": "strands-decider-2B-hobson-v19", "answers": { "severity": { "type": "score", "score": 1, "confidence": 0.7, "legend": {"0": "none", "1": "low", "2": "high"}, "probabilities": {"0": 0.1, "1": 0.8, "2": 0.1}, } }, "usage": {"input_tokens": 216, "output_tokens": 3}, "latency_ms": 3722.17, } _DATABRICKS_ENDPOINT: Final = "databricks-openjev-qwen35-4b" _DATABRICKS_ENDPOINT_RESPONSE: Final[Mapping[str, object]] = { "model": "/mosaicml/local_model", "answers": { "is_defect": {"type": "noul", "noul": 0.9}, "sentiment": { "type": "choice", "choice": "positive", "confidence": 0.8, "probabilities": {"positive": 0.8, "negative": 0.2}, }, }, "usage": {"input_tokens": 214, "output_tokens": 0}, } _DATABRICKS_ROUTE: Final = "https://workspace.example/api/2.0/ai-functions/ai-decide" _DATABRICKS_RESPONSE: Final[Mapping[str, object]] = { "response": { "answers": { "is_defect": {"type": "noul", "probability": 0.9}, "sentiment": { "type": "choice", "choice": "positive", "confidence": 0.8, "probabilities": {"positive": 0.8, "negative": 0.2}, }, } }, "metadata": {"version": "1.0"}, } _PROVIDERS: Final[tuple[tuple[str, str, str, str], ...]] = ( ( "perplexity", "perplexity/pplx-decider-v1-27b", "https://api.perplexity.ai/v1/decisions", "pplx-decider-v1-27b", ), ("typesafe", "typesafe/jev-1.13", "https://api.typesafe.ai/v1/systemone", "jev-1.13"), ( "openrouter", "openrouter/typesafe/jev-1.13", "https://openrouter.ai/api/alpha/decisions", "typesafe/jev-1.13", ), ) _OPENAI_RESPONSE: Final[Mapping[str, object]] = { "model": "gpt-6-luna", "answers": [ {"type": "predicate", "name": "is_defect", "probability": 0.9}, {"type": "refusal", "name": "sentiment"}, ], "usage": { "input_tokens": _INPUT_TOKENS, "input_tokens_details": {"cached_tokens": _CACHED_TOKENS, "cache_write_tokens": _CACHE_WRITE_TOKENS}, "output_tokens": _OUTPUT_TOKENS, "output_tokens_details": {"reasoning_tokens": 0}, "total_tokens": _INPUT_TOKENS + _OUTPUT_TOKENS, }, } class _RecordingLogger(CustomLogger): def __init__(self) -> None: super().__init__() self.standard_logging_object: Mapping[str, object] | None = None async def async_log_success_event( self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object, ) -> None: standard_logging_object: Final = kwargs.get("standard_logging_object") if isinstance(standard_logging_object, dict): self.standard_logging_object = standard_logging_object async def _drain_logging_worker() -> None: await asyncio.sleep(0) GLOBAL_LOGGING_WORKER.start() await asyncio.wait_for(GLOBAL_LOGGING_WORKER.flush(), timeout=10.0) @pytest.fixture(autouse=True) def _httpx_transport(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) litellm.in_memory_llm_clients_cache.flush_cache() _GPT_6_LUNA_DECISIONS_INPUT_RATE: Final = 1e-07 _MICROSOFT_DECISION_1_DECISIONS_INPUT_RATE: Final = 7e-08 _DECISIONS_PRICING_FIELDS: Final[dict[str, dict[str, object]]] = { "gpt-6-luna": { "input_cost_per_token_decisions": _GPT_6_LUNA_DECISIONS_INPUT_RATE, "cache_read_input_token_cost_decisions": _GPT_6_LUNA_DECISIONS_INPUT_RATE, "cache_creation_input_token_cost_decisions": _GPT_6_LUNA_DECISIONS_INPUT_RATE, "output_cost_per_token_decisions": 0.0, }, "azure_ai/Microsoft-Decision-1": {"input_cost_per_token_decisions": _MICROSOFT_DECISION_1_DECISIONS_INPUT_RATE}, } @pytest.fixture def decisions_pricing_fields(monkeypatch: pytest.MonkeyPatch) -> None: for key, fields in _DECISIONS_PRICING_FIELDS.items(): monkeypatch.setitem(litellm.model_cost, key, {**litellm.model_cost[key], **fields}) @pytest.mark.asyncio @pytest.mark.parametrize(("provider", "model", "url", "upstream_model"), _PROVIDERS) async def test_adecisions_sends_the_provider_wire_contract( provider: str, model: str, url: str, upstream_model: str, respx_mock: respx.MockRouter, ) -> None: route: Final = respx_mock.post(url).respond(json=_RESPONSE) response: Final = await litellm.asystemone( model=model, state={"source": "unit-test"}, questions=_QUESTIONS, api_key="caller-key", extra_headers={ "x-request-tag": "decisions-test", "AUTHORIZATION": "attacker-key", "Content-Type": "text/plain", }, internal_kwarg="must-not-leak", ) assert route.called assert len(respx_mock.calls) == 1 request: Final = respx_mock.calls[0].request assert request.headers["authorization"] == "Bearer caller-key" assert request.headers["content-type"] == "application/json" assert request.headers["x-request-tag"] == "decisions-test" assert json.loads(request.content) == { "model": upstream_model, "state": {"source": "unit-test"}, "questions": { "is_defect": { "type": "noul", "instructions": "Is this a defect?", "provider_field": "kept", }, "sentiment": {"type": "choice", "criteria": {"positive": None, "negative": "unhappy"}}, "severity": {"type": "score", "criteria": ["none", "low", "high"]}, }, } assert isinstance(response.answers["is_defect"], NoulAnswer) assert isinstance(response.answers["sentiment"], ChoiceAnswer) assert isinstance(response.answers["severity"], ScoreAnswer) assert response._hidden_params["custom_llm_provider"] == provider @pytest.mark.asyncio async def test_router_dispatches_typesafe_decisions_without_api_base( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("TYPESAFE_API_KEY", raising=False) monkeypatch.delenv("TYPESAFE_API_BASE", raising=False) provider_resolution: Final = litellm.get_llm_provider("typesafe/jev-latest") assert provider_resolution[:2] == ("jev-latest", "typesafe") router: Final = litellm.Router( model_list=[ { "model_name": "jev", "litellm_params": { "model": "typesafe/jev-latest", "api_key": "k", }, } ] ) upstream: Final = respx_mock.post("https://api.typesafe.ai/v1/systemone").respond(json=_RESPONSE) response: Final = await router.asystemone( model="jev", state="router-test", questions={ "sentiment": { "type": "choice", "criteria": {"positive": None, "negative": "unhappy"}, } }, ) assert upstream.called assert len(respx_mock.calls) == 1 assert json.loads(respx_mock.calls[0].request.content) == { "model": "jev-latest", "state": "router-test", "questions": { "sentiment": { "type": "choice", "criteria": {"positive": None, "negative": "unhappy"}, } }, } assert respx_mock.calls[0].request.headers["authorization"] == "Bearer k" assert isinstance(response.answers["sentiment"], ChoiceAnswer) assert response.answers["sentiment"].choice == "positive" def test_decisions_uses_the_same_wire_contract_for_sync_calls(respx_mock: respx.MockRouter) -> None: route: Final = respx_mock.post("https://api.perplexity.ai/v1/decisions").respond(json=_RESPONSE) response: Final = litellm.systemone( model="perplexity/pplx-decider-v1-27b", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, api_key="caller-key", ) assert route.called assert response.model == "jev-1.13" def test_openrouter_response_keeps_provider_fields(respx_mock: respx.MockRouter) -> None: payload: Final = { **_RESPONSE, "id": "decision-1", "provider": "typesafe", "usage": {**_RESPONSE["usage"], "cost": 0.25}, } respx_mock.post("https://openrouter.ai/api/alpha/decisions").respond(json=payload) response: Final = litellm.systemone( model="openrouter/typesafe/jev-1.13", state="review", questions=_QUESTIONS, api_key="caller-key", ) assert response.model_extra["id"] == "decision-1" assert response.model_extra["provider"] == "typesafe" assert response.usage is not None assert response.usage.model_extra["cost"] == 0.25 def test_decisions_cost_uses_litellm_token_pricing() -> None: response: Final = DecisionsResponse( model="pplx-decider-v1-27b", answers={}, usage=DecisionsUsage(input_tokens=_INPUT_TOKENS, output_tokens=_OUTPUT_TOKENS), ) response.set_hidden_params({"model": "perplexity/pplx-decider-v1-27b", "custom_llm_provider": "perplexity"}) cost: Final = litellm.completion_cost(completion_response=response) perplexity_cost: Final = litellm.model_cost["perplexity/pplx-decider-v1-27b"] expected_cost: Final = _INPUT_TOKENS * float(perplexity_cost["input_cost_per_token"]) + _OUTPUT_TOKENS * float( perplexity_cost["output_cost_per_token"] ) assert expected_cost > 0 assert cost == pytest.approx(expected_cost) @pytest.mark.asyncio async def test_openai_decisions_uses_evaluation_pricing_and_explicit_base_model( respx_mock: respx.MockRouter, decisions_pricing_fields: None, ) -> None: route: Final = respx_mock.post("https://api.openai.com/v1/decisions").respond(json=_OPENAI_RESPONSE) response: Final = await litellm.adecisions( model="openai/gpt-6-luna", input="The package arrived broken.", questions=[{"type": "predicate", "name": "is_defect", "instructions": "Is this a defect?"}], api_key="caller-key", api_base="https://api.openai.com/v1", ) chat_cost: Mapping[str, object] = litellm.model_cost["gpt-6-luna"] expected_chat_cost: Final = ( (_INPUT_TOKENS - _CACHED_TOKENS - _CACHE_WRITE_TOKENS) * float(chat_cost["input_cost_per_token"]) + _CACHED_TOKENS * float(chat_cost["cache_read_input_token_cost"]) + _CACHE_WRITE_TOKENS * float(chat_cost["cache_creation_input_token_cost"]) + _OUTPUT_TOKENS * float(chat_cost["output_cost_per_token"]) ) assert route.called assert response._hidden_params["model"] == "openai/gpt-6-luna" assert response._hidden_params["provider_response_model"] == "openai/gpt-6-luna" assert litellm.completion_cost(completion_response=response) == pytest.approx( _INPUT_TOKENS * _GPT_6_LUNA_DECISIONS_INPUT_RATE ) assert litellm.completion_cost(completion_response=response, call_type="acompletion") == pytest.approx( expected_chat_cost ) @pytest.mark.asyncio async def test_systemone_uses_evaluation_pricing(respx_mock: respx.MockRouter, decisions_pricing_fields: None) -> None: route: Final = respx_mock.post("https://api.openai.com/v1/decisions").respond(json=_OPENAI_RESPONSE) response: Final = await litellm.asystemone( model="openai/gpt-6-luna", state="The package arrived broken.", questions=_QUESTIONS, api_key="caller-key", api_base="https://api.openai.com/v1", ) assert route.called assert response._hidden_params["model"] == "openai/gpt-6-luna" assert response._hidden_params["provider_response_model"] == "openai/gpt-6-luna" assert litellm.completion_cost(completion_response=response) == pytest.approx( _INPUT_TOKENS * _GPT_6_LUNA_DECISIONS_INPUT_RATE ) @pytest.mark.asyncio async def test_systemone_falls_back_to_plain_pricing_without_decisions_row( respx_mock: respx.MockRouter, ) -> None: route: Final = respx_mock.post("https://decisions.example/v1/decisions").respond(json=_RESPONSE) response: Final = await litellm.asystemone( model="perplexity/sonar", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, api_key="caller-key", api_base="https://decisions.example/v1", ) plain_cost: Mapping[str, object] = litellm.model_cost["perplexity/sonar"] assert route.called assert response._hidden_params["provider_response_model"] == "perplexity/sonar" assert litellm.completion_cost(completion_response=response) == pytest.approx( _INPUT_TOKENS * float(plain_cost["input_cost_per_token"]) + _OUTPUT_TOKENS * float(plain_cost["output_cost_per_token"]) ) @pytest.mark.parametrize( "response", ( DecisionsResponse( model="jev-latest", answers={}, usage=DecisionsUsage( input_tokens=_INPUT_TOKENS, output_tokens=_OUTPUT_TOKENS, cached_tokens=_CACHED_TOKENS, cache_write_tokens=_CACHE_WRITE_TOKENS, ), ), OpenAIDecisionResponse.model_validate(_OPENAI_RESPONSE), ), ids=("systemone", "openai"), ) def test_custom_token_pricing_bills_cached_decisions_input_tokens_once( response: DecisionsResponse | OpenAIDecisionResponse, ) -> None: cost: Final = litellm.completion_cost( completion_response=response, model="gpt-6-luna", custom_llm_provider="openai", custom_cost_per_token={ "input_cost_per_token": 1.0, "output_cost_per_token": 2.0, "cache_read_input_token_cost": 0.1, "cache_creation_input_token_cost": 1.25, }, ) assert cost == pytest.approx( (_INPUT_TOKENS - _CACHED_TOKENS - _CACHE_WRITE_TOKENS) * 1.0 + _CACHED_TOKENS * 0.1 + _CACHE_WRITE_TOKENS * 1.25 + _OUTPUT_TOKENS * 2.0 ) def test_decisions_response_hidden_params_getter_preserves_mutable_identity() -> None: response: Final = DecisionsResponse(model="decider", answers={}, usage=None) assert response.hidden_params is response._hidden_params response.hidden_params["mutation"] = "visible" assert response._hidden_params["mutation"] == "visible" @pytest.mark.asyncio async def test_decisions_cost_is_in_standard_logging_object(respx_mock: respx.MockRouter) -> None: respx_mock.post("https://api.perplexity.ai/v1/decisions").respond(json=_RESPONSE) recording_logger: Final = _RecordingLogger() original_callbacks: Final = litellm.callbacks litellm.callbacks = [recording_logger] try: await litellm.asystemone( model="perplexity/pplx-decider-v1-27b", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, api_key="caller-key", ) await _drain_logging_worker() finally: litellm.callbacks = original_callbacks assert recording_logger.standard_logging_object is not None perplexity_cost: Final = litellm.model_cost["perplexity/pplx-decider-v1-27b"] expected_cost: Final = _INPUT_TOKENS * float(perplexity_cost["input_cost_per_token"]) + _OUTPUT_TOKENS * float( perplexity_cost["output_cost_per_token"] ) assert expected_cost > 0 assert recording_logger.standard_logging_object["response_cost"] == pytest.approx(expected_cost) assert recording_logger.standard_logging_object["prompt_tokens"] == _INPUT_TOKENS assert recording_logger.standard_logging_object["completion_tokens"] == _OUTPUT_TOKENS @pytest.mark.asyncio async def test_unknown_provider_is_rejected_before_http(respx_mock: respx.MockRouter) -> None: with pytest.raises(litellm.BadRequestError, match="LLM Provider NOT provided"): await litellm.asystemone( model="unknown/jev-1.13", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, api_key="caller-key", ) assert len(respx_mock.calls) == 0 @pytest.mark.asyncio async def test_provider_without_decisions_support_is_rejected_before_http(respx_mock: respx.MockRouter) -> None: with pytest.raises(litellm.BadRequestError, match=r"Unknown Decisions provider 'anthropic'\. Supported providers"): await litellm.asystemone( model="anthropic/claude-sonnet-4-5", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, api_key="caller-key", ) assert len(respx_mock.calls) == 0 @pytest.mark.asyncio async def test_empty_custom_provider_falls_back_to_the_model_prefix(respx_mock: respx.MockRouter) -> None: route: Final = respx_mock.post("https://api.perplexity.ai/v1/decisions").respond(json=_RESPONSE) response: Final = await litellm.asystemone( model="perplexity/pplx-decider-v1-27b", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, api_key="caller-key", custom_llm_provider="", ) assert route.called assert json.loads(respx_mock.calls[0].request.content)["model"] == "pplx-decider-v1-27b" assert response._hidden_params["custom_llm_provider"] == "perplexity" @pytest.mark.asyncio async def test_upstream_reply_without_answers_is_a_server_error(respx_mock: respx.MockRouter) -> None: respx_mock.post("https://api.perplexity.ai/v1/decisions").respond( json={"model": "pplx-decider-v1-27b", "usage": {"input_tokens": 10, "output_tokens": 0}} ) with pytest.raises(litellm.InternalServerError, match="unexpected response"): await litellm.asystemone( model="perplexity/pplx-decider-v1-27b", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, api_key="caller-key", ) @pytest.mark.asyncio async def test_router_sends_the_openrouter_deployment_key_when_the_provider_is_already_resolved( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) router: Final = litellm.Router( model_list=[ { "model_name": "jev", "litellm_params": { "model": "openrouter/typesafe/jev-1.13", "api_key": "deployment-key", "api_base": "https://egress.example/openrouter", }, } ] ) upstream: Final = respx_mock.post("https://egress.example/openrouter/alpha/decisions").respond(json=_RESPONSE) await router.asystemone( model="jev", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, ) assert upstream.called assert respx_mock.calls[0].request.headers["authorization"] == "Bearer deployment-key" assert json.loads(respx_mock.calls[0].request.content)["model"] == "typesafe/jev-1.13" @pytest.mark.asyncio async def test_invalid_question_is_rejected_before_http(respx_mock: respx.MockRouter) -> None: with pytest.raises(litellm.BadRequestError, match="Invalid Decisions request"): await litellm.asystemone( model="perplexity/pplx-decider-v1-27b", state="review", questions={"sentiment": {"type": "choice"}}, api_key="caller-key", ) assert len(respx_mock.calls) == 0 def test_upstream_bad_request_maps_to_litellm_error(respx_mock: respx.MockRouter) -> None: respx_mock.post("https://api.perplexity.ai/v1/decisions").respond( status_code=400, json={"error": {"message": "invalid decision"}}, ) with pytest.raises(litellm.BadRequestError): litellm.systemone( model="perplexity/pplx-decider-v1-27b", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, api_key="caller-key", ) def test_unreachable_upstream_maps_to_a_connection_error(respx_mock: respx.MockRouter) -> None: respx_mock.post("https://api.perplexity.ai/v1/decisions").mock( side_effect=httpx.ConnectError("Cannot connect to host api.perplexity.ai:443") ) with pytest.raises(litellm.APIConnectionError, match="PerplexityException - Cannot connect to host"): litellm.systemone( model="perplexity/pplx-decider-v1-27b", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, api_key="caller-key", ) def test_server_key_is_sent_to_an_explicit_api_base( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.setenv("PERPLEXITYAI_API_KEY", "server-key") monkeypatch.delenv("PERPLEXITY_API_KEY", raising=False) route: Final = respx_mock.post("https://egress.example/perplexity/v1/decisions").respond(json=_RESPONSE) litellm.systemone( model="perplexity/pplx-decider-v1-27b", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, api_base="https://egress.example/perplexity", ) assert route.call_count == 1 assert route.calls[0].request.headers["authorization"] == "Bearer server-key" @pytest.mark.asyncio @pytest.mark.parametrize("model", ("cloudflare/clef", "cloudflare/@cf/cloudflare/clef")) @pytest.mark.parametrize("wrapped", (False, True)) async def test_cloudflare_clef_resolves_model_and_response_envelope( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, model: str, wrapped: bool, ) -> None: monkeypatch.setenv("CLOUDFLARE_ACCOUNT_ID", "acct") monkeypatch.setenv("CLOUDFLARE_API_KEY", "cloudflare-key") monkeypatch.delenv("CLOUDFLARE_API_BASE", raising=False) response_body: Final[Mapping[str, object]] = ( {"result": _RESPONSE, "success": True, "errors": [], "messages": []} if wrapped else _RESPONSE ) route: Final = respx_mock.post( "https://api.cloudflare.com/client/v4/accounts/acct/ai/run/@cf/cloudflare/clef" ).respond(json=response_body) response: Final = await litellm.asystemone( model=model, state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, ) assert route.called request: Final = respx_mock.calls[0].request assert request.headers["authorization"] == "Bearer cloudflare-key" assert json.loads(request.content) == { "model": "clef", "state": "review", "questions": {"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, } assert response.answers == {"is_defect": NoulAnswer(type="noul", noul=0.9)} assert response._hidden_params["model"] == "cloudflare/@cf/cloudflare/clef" @pytest.mark.asyncio async def test_cloudflare_clef_flash_uses_flash_endpoint_and_request_model( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.setenv("CLOUDFLARE_ACCOUNT_ID", "acct") monkeypatch.setenv("CLOUDFLARE_API_KEY", "cloudflare-key") monkeypatch.delenv("CLOUDFLARE_API_BASE", raising=False) route: Final = respx_mock.post( "https://api.cloudflare.com/client/v4/accounts/acct/ai/run/@cf/cloudflare/clef-flash" ).respond(json=_RESPONSE) await litellm.asystemone( model="cloudflare/clef-flash", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, ) assert route.called assert json.loads(respx_mock.calls[0].request.content)["model"] == "clef-flash" @pytest.mark.asyncio async def test_cloudflare_api_base_from_env_uses_workers_ai_run_path( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.setenv("CLOUDFLARE_API_BASE", "https://api.cloudflare.com/client/v4/accounts/acct/ai/v1") monkeypatch.setenv("CLOUDFLARE_API_KEY", "cloudflare-key") monkeypatch.delenv("CLOUDFLARE_ACCOUNT_ID", raising=False) route: Final = respx_mock.post( "https://api.cloudflare.com/client/v4/accounts/acct/ai/run/@cf/cloudflare/clef" ).respond(json=_RESPONSE) await litellm.asystemone( model="cloudflare/clef", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, ) assert route.called @pytest.mark.asyncio async def test_cloudflare_requires_account_id_or_api_base_before_http( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("CLOUDFLARE_ACCOUNT_ID", raising=False) monkeypatch.delenv("CLOUDFLARE_API_BASE", raising=False) monkeypatch.setenv("CLOUDFLARE_API_KEY", "cloudflare-key") with pytest.raises(litellm.BadRequestError, match="Missing CLOUDFLARE_ACCOUNT_ID - set CLOUDFLARE_ACCOUNT_ID"): await litellm.asystemone( model="cloudflare/clef", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, ) assert len(respx_mock.calls) == 0 @pytest.mark.asyncio async def test_cloudflare_clef_cost_uses_the_model_cost_map( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.setenv("CLOUDFLARE_ACCOUNT_ID", "acct") monkeypatch.setenv("CLOUDFLARE_API_KEY", "cloudflare-key") monkeypatch.delenv("CLOUDFLARE_API_BASE", raising=False) respx_mock.post("https://api.cloudflare.com/client/v4/accounts/acct/ai/run/@cf/cloudflare/clef").respond( json=_RESPONSE ) response: Final = await litellm.asystemone( model="cloudflare/clef", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, ) cost: Final = litellm.completion_cost(completion_response=response) clef_cost: Final = litellm.model_cost["cloudflare/@cf/cloudflare/clef"] expected_cost: Final = _INPUT_TOKENS * float(clef_cost["input_cost_per_token"]) + _OUTPUT_TOKENS * float( clef_cost["output_cost_per_token"] ) assert expected_cost > 0 assert cost == pytest.approx(expected_cost) @pytest.mark.asyncio async def test_strands_decider_requires_api_base_before_http( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("STRANDS_DECIDER_API_BASE", raising=False) monkeypatch.delenv("STRANDS_DECIDER_API_KEY", raising=False) with pytest.raises(litellm.BadRequestError, match="api_base is required"): await litellm.asystemone( model="strands_decider/strands-decider-2B-hobson-v19", state="review", questions={"severity": {"type": "score", "criteria": ["none", "low", "high"]}}, ) assert len(respx_mock.calls) == 0 @pytest.mark.asyncio async def test_strands_decider_without_key_preserves_response_extras( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("STRANDS_DECIDER_API_BASE", raising=False) monkeypatch.delenv("STRANDS_DECIDER_API_KEY", raising=False) route: Final = respx_mock.post("https://strands.example/v1/systemone").respond(json=_STRANDS_RESPONSE) response: Final = await litellm.asystemone( model="strands_decider/strands-decider-2B-hobson-v19", state="review", questions={"severity": {"type": "score", "criteria": ["none", "low", "high"]}}, api_base="https://strands.example", ) assert route.called assert "authorization" not in respx_mock.calls[0].request.headers assert response.model_extra["latency_ms"] == _STRANDS_RESPONSE["latency_ms"] severity: Final = response.answers["severity"] assert isinstance(severity, ScoreAnswer) assert severity.legend == {"0": "none", "1": "low", "2": "high"} @pytest.mark.asyncio async def test_strands_decider_plain_api_base_answers_while_every_aws_signing_thread_is_busy( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("STRANDS_DECIDER_API_KEY", raising=False) respx_mock.post("https://strands.example/v1/systemone").respond(json=_STRANDS_RESPONSE) release: Final = threading.Event() busy: Final = tuple(AWS_SIGNING_EXECUTOR.submit(release.wait) for _ in range(AWS_SIGNING_MAX_THREADS)) try: response: Final = await asyncio.wait_for( litellm.asystemone( model="strands_decider/strands-decider-2B-hobson-v19", state="review", questions={"severity": {"type": "score", "criteria": ["none", "low", "high"]}}, api_base="https://strands.example", ), timeout=5, ) finally: release.set() wait(busy) assert isinstance(response.answers["severity"], ScoreAnswer) @pytest.mark.asyncio async def test_strands_decider_uses_key_from_matching_environment_base( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.setenv("STRANDS_DECIDER_API_BASE", "https://strands.example") monkeypatch.setenv("STRANDS_DECIDER_API_KEY", "strands-key") route: Final = respx_mock.post("https://strands.example/v1/systemone").respond(json=_STRANDS_RESPONSE) await litellm.asystemone( model="strands_decider/strands-decider-2B-hobson-v19", state="review", questions={"severity": {"type": "score", "criteria": ["none", "low", "high"]}}, api_base="https://strands.example", ) assert route.called assert respx_mock.calls[0].request.headers["authorization"] == "Bearer strands-key" @pytest.mark.asyncio async def test_strands_decider_provider_resolution_and_router_dispatch( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("STRANDS_DECIDER_API_BASE", raising=False) monkeypatch.delenv("STRANDS_DECIDER_API_KEY", raising=False) provider_resolution: Final = litellm.get_llm_provider("strands_decider/strands-decider-2B-hobson-v19") router: Final = litellm.Router( model_list=[ { "model_name": "strands", "litellm_params": { "model": "strands_decider/strands-decider-2B-hobson-v19", "api_base": "https://strands.example", }, } ] ) route: Final = respx_mock.post("https://strands.example/v1/systemone").respond(json=_STRANDS_RESPONSE) response: Final = await router.asystemone( model="strands", state="review", questions={"severity": {"type": "score", "criteria": ["none", "low", "high"]}}, ) assert provider_resolution[:2] == ("strands-decider-2B-hobson-v19", "strands_decider") assert route.called assert response.model == _STRANDS_RESPONSE["model"] _RUNTIME_ARN: Final = "arn:aws:bedrock-agentcore:us-east-1:123456789012:runtime/qa_decider-AbC123xyz0" _RUNTIME_URL: Final = ( "https://bedrock-agentcore.us-east-1.amazonaws.com/runtimes/" "arn%3Aaws%3Abedrock-agentcore%3Aus-east-1%3A123456789012%3Aruntime%2Fqa_decider-AbC123xyz0/invocations" ) _RUNTIME_CREDENTIALS: Final = Credentials(access_key="AKIDEXAMPLE", secret_key="example-secret") _SESSION_HEADER: Final = "x-amzn-bedrock-agentcore-runtime-session-id" def _sigv4_verifies(request: httpx.Request) -> bool: authorization: Final = request.headers["authorization"] signed_names: Final = authorization.split("SignedHeaders=")[1].split(",")[0].split(";") replayed: Final = AWSRequest( method=request.method, url=str(request.url), data=request.content, headers={name: request.headers[name] for name in signed_names}, ) replayed.context["timestamp"] = request.headers["x-amz-date"] auth: Final = SigV4Auth(_RUNTIME_CREDENTIALS, "bedrock-agentcore", "us-east-1") string_to_sign: Final = auth.string_to_sign(replayed, auth.canonical_request(replayed)) return authorization.endswith(f"Signature={auth.signature(string_to_sign, replayed)}") @pytest.mark.parametrize("use_async", [True, False]) @pytest.mark.asyncio async def test_strands_decider_runtime_arn_sends_a_sigv4_signed_invoke_agent_runtime_call( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, use_async: bool, ) -> None: monkeypatch.delenv("STRANDS_DECIDER_API_KEY", raising=False) monkeypatch.delenv("AWS_BEARER_TOKEN_BEDROCK", raising=False) route: Final = respx_mock.post(_RUNTIME_URL).respond(json=_STRANDS_RESPONSE) call_args: Final[Mapping[str, object]] = { "model": "strands_decider/strands-decider-2B-hobson-v19", "state": "review", "questions": {"severity": {"type": "score", "criteria": ["none", "low", "high"]}}, "api_base": _RUNTIME_ARN, "aws_access_key_id": _RUNTIME_CREDENTIALS.access_key, "aws_secret_access_key": _RUNTIME_CREDENTIALS.secret_key, } response: Final = await litellm.asystemone(**call_args) if use_async else litellm.systemone(**call_args) assert route.call_count == 1 sent: Final = route.calls[0].request assert _sigv4_verifies(sent) assert json.loads(sent.content)["model"] == "strands-decider-2B-hobson-v19" assert sent.headers[_SESSION_HEADER].startswith("litellm-decider-") assert response.model == _STRANDS_RESPONSE["model"] @pytest.mark.asyncio async def test_strands_decider_runtime_arn_router_deployment_answers_openai_format_calls( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("STRANDS_DECIDER_API_KEY", raising=False) router: Final = litellm.Router( model_list=[ { "model_name": "strands", "litellm_params": { "model": "strands_decider/strands-decider-2B-hobson-v19", "api_base": _RUNTIME_ARN, "aws_access_key_id": _RUNTIME_CREDENTIALS.access_key, "aws_secret_access_key": _RUNTIME_CREDENTIALS.secret_key, "aws_region_name": "eu-west-1", "agentcore_runtime_session_id": "pooled-session-0123456789abcdef0123456", }, } ] ) route: Final = respx_mock.post(_RUNTIME_URL).respond(json=_STRANDS_RESPONSE) response: Final = await router.adecisions( model="strands", input="review", extra_headers={_SESSION_HEADER: "caller-session-0123456789abcdef01234567"}, questions=[ { "type": "score", "name": "severity", "instructions": "How bad?", "levels": [{"label": "none"}, {"label": "low"}, {"label": "high"}], } ], ) sent: Final = route.calls[0].request assert _sigv4_verifies(sent) assert sent.headers[_SESSION_HEADER] == "pooled-session-0123456789abcdef0123456" assert isinstance(response, OpenAIDecisionResponse) @pytest.mark.asyncio async def test_strands_decider_runtime_arn_with_api_key_sends_the_bearer_token_unsigned( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("STRANDS_DECIDER_API_KEY", raising=False) route: Final = respx_mock.post(_RUNTIME_URL).respond(json=_STRANDS_RESPONSE) await litellm.asystemone( model="strands_decider/strands-decider-2B-hobson-v19", state="review", questions={"severity": {"type": "score", "criteria": ["none", "low", "high"]}}, api_base=_RUNTIME_ARN, api_key="runtime-jwt", ) sent: Final = route.calls[0].request assert sent.headers["authorization"] == "Bearer runtime-jwt" assert "x-amz-date" not in sent.headers assert _SESSION_HEADER in sent.headers @pytest.mark.parametrize( ("code", "error_class"), [ ("bad_request", litellm.BadRequestError), ("loading", litellm.ServiceUnavailableError), ("inference_error", litellm.InternalServerError), ], ) @pytest.mark.asyncio async def test_strands_decider_runtime_error_envelope_raises_instead_of_answering( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, code: str, error_class: type[Exception], ) -> None: monkeypatch.delenv("STRANDS_DECIDER_API_KEY", raising=False) respx_mock.post(_RUNTIME_URL).respond(json={"error": {"code": code, "message": "at most 16 questions"}}) with pytest.raises(error_class, match="at most 16 questions"): await litellm.asystemone( model="strands_decider/strands-decider-2B-hobson-v19", state="review", questions={"severity": {"type": "score", "criteria": ["none", "low", "high"]}}, api_base=_RUNTIME_ARN, aws_access_key_id=_RUNTIME_CREDENTIALS.access_key, aws_secret_access_key=_RUNTIME_CREDENTIALS.secret_key, ) @pytest.mark.asyncio async def test_strands_decider_plain_api_base_error_envelope_stays_an_unexpected_response( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("STRANDS_DECIDER_API_KEY", raising=False) respx_mock.post("https://strands.example/v1/systemone").respond( json={"error": {"code": "bad_request", "message": "at most 16 questions"}} ) with pytest.raises(litellm.InternalServerError, match="returned an unexpected response"): await litellm.asystemone( model="strands_decider/strands-decider-2B-hobson-v19", state="review", questions={"severity": {"type": "score", "criteria": ["none", "low", "high"]}}, api_base="https://strands.example", ) _VLLM_RESPONSE: Final[Mapping[str, object]] = { "id": "systemone-4af3d2c1", "object": "structured_decision", "created": 1760012345, "model": "Qwen/Qwen3-0.6B", "answers": { "is_defect": { "type": "choice", "choice": "yes", "confidence": 0.91, "probabilities": {"yes": 0.91, "no": 0.09}, } }, "usage": {"input_tokens": 154, "output_tokens": 2}, "diagnostics": {"engine": "vllm", "logprobs_mode": "raw_logprobs"}, } @pytest.mark.asyncio async def test_hosted_vllm_requires_api_base_before_http( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("HOSTED_VLLM_API_BASE", raising=False) monkeypatch.delenv("HOSTED_VLLM_API_KEY", raising=False) with pytest.raises(litellm.BadRequestError, match="api_base is required"): await litellm.asystemone( model="hosted_vllm/Qwen/Qwen3-0.6B", state="review", questions={"is_defect": {"type": "choice", "criteria": {"yes": None, "no": None}}}, ) assert len(respx_mock.calls) == 0 @pytest.mark.asyncio async def test_hosted_vllm_without_key_sends_no_authorization_and_preserves_response_extras( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("HOSTED_VLLM_API_BASE", raising=False) monkeypatch.delenv("HOSTED_VLLM_API_KEY", raising=False) route: Final = respx_mock.post("http://vllm.local:8000/v1/systemone").respond(json=_VLLM_RESPONSE) response: Final = await litellm.asystemone( model="hosted_vllm/Qwen/Qwen3-0.6B", state="The package arrived broken.", questions={"is_defect": {"type": "choice", "criteria": {"yes": None, "no": None}}}, api_base="http://vllm.local:8000", ) assert route.called assert "authorization" not in respx_mock.calls[0].request.headers assert response.model_extra["id"] == _VLLM_RESPONSE["id"] assert response.model_extra["object"] == _VLLM_RESPONSE["object"] assert response.model_extra["diagnostics"] == _VLLM_RESPONSE["diagnostics"] @pytest.mark.asyncio async def test_hosted_vllm_uses_key_and_base_from_environment( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.setenv("HOSTED_VLLM_API_BASE", "http://vllm.local:8000/v1") monkeypatch.setenv("HOSTED_VLLM_API_KEY", "vllm-key") route: Final = respx_mock.post("http://vllm.local:8000/v1/systemone").respond(json=_VLLM_RESPONSE) await litellm.asystemone( model="hosted_vllm/Qwen/Qwen3-0.6B", state="The package arrived broken.", questions={"is_defect": {"type": "choice", "criteria": {"yes": None, "no": None}}}, ) assert route.called assert respx_mock.calls[0].request.headers["authorization"] == "Bearer vllm-key" @pytest.mark.asyncio async def test_hosted_vllm_api_base_with_v1_suffix_posts_to_v1_systemone( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("HOSTED_VLLM_API_BASE", raising=False) monkeypatch.delenv("HOSTED_VLLM_API_KEY", raising=False) route: Final = respx_mock.post("http://vllm.local:8000/v1/systemone").respond(json=_VLLM_RESPONSE) await litellm.asystemone( model="hosted_vllm/Qwen/Qwen3-0.6B", state="The package arrived broken.", questions={"is_defect": {"type": "choice", "criteria": {"yes": None, "no": None}}}, api_base="http://vllm.local:8000/v1", ) assert route.called @pytest.mark.asyncio async def test_hosted_vllm_choice_question_goes_out_as_a_jev_choice_with_the_served_model_name( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("HOSTED_VLLM_API_BASE", raising=False) monkeypatch.delenv("HOSTED_VLLM_API_KEY", raising=False) route: Final = respx_mock.post("http://vllm.local:8000/v1/systemone").respond(json=_VLLM_RESPONSE) await litellm.asystemone( model="hosted_vllm/Qwen/Qwen3-0.6B", state="The package arrived broken.", questions={ "is_defect": { "type": "choice", "instructions": "Is this a defect?", "criteria": {"yes": "defective", "no": "intact"}, } }, api_base="http://vllm.local:8000", ) assert route.called assert json.loads(respx_mock.calls[0].request.content) == { "model": "Qwen/Qwen3-0.6B", "state": "The package arrived broken.", "questions": { "is_defect": { "type": "choice", "instructions": "Is this a defect?", "criteria": {"yes": "defective", "no": "intact"}, } }, } @pytest.mark.asyncio async def test_databricks_requires_api_base_before_http( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("DATABRICKS_API_BASE", raising=False) monkeypatch.setenv("DATABRICKS_API_KEY", "dapi-key") with pytest.raises(litellm.BadRequestError, match="DATABRICKS_API_BASE"): await litellm.asystemone( model="databricks/ai_decide", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, ) assert len(respx_mock.calls) == 0 @pytest.mark.asyncio @pytest.mark.parametrize( "endpoint_name", ["serving-endpoints/openjev", "openjev?x=1", "openjev#frag", "..", "open jev"] ) async def test_databricks_rejects_a_serving_endpoint_name_that_would_rewrite_the_url( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, endpoint_name: str, ) -> None: monkeypatch.setenv("DATABRICKS_API_BASE", "https://workspace.example/serving-endpoints") monkeypatch.setenv("DATABRICKS_API_KEY", "dapi-key") with pytest.raises(litellm.BadRequestError, match="bare serving endpoint name"): await litellm.asystemone( model=f"databricks/{endpoint_name}", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, ) assert len(respx_mock.calls) == 0 @pytest.mark.asyncio async def test_databricks_requires_a_key_before_http( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.setenv("DATABRICKS_API_BASE", "https://workspace.example") monkeypatch.delenv("DATABRICKS_API_KEY", raising=False) monkeypatch.delenv("DATABRICKS_TOKEN", raising=False) with pytest.raises(litellm.AuthenticationError, match="Missing API key for Decisions provider 'databricks'"): await litellm.asystemone( model="databricks/ai_decide", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, ) assert len(respx_mock.calls) == 0 @pytest.mark.asyncio @pytest.mark.parametrize( "api_base", ["https://workspace.example/serving-endpoints", "https://workspace.example/serving-endpoints/"] ) async def test_databricks_posts_the_jev_body_to_the_serving_endpoint_invocations_route( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, api_base: str, ) -> None: monkeypatch.setenv("DATABRICKS_API_BASE", api_base) monkeypatch.delenv("DATABRICKS_API_KEY", raising=False) monkeypatch.setenv("DATABRICKS_TOKEN", "dapi-token") route: Final = respx_mock.post( f"https://workspace.example/serving-endpoints/{_DATABRICKS_ENDPOINT}/invocations" ).respond(json=_DATABRICKS_ENDPOINT_RESPONSE) questions: Final = { "is_defect": {"type": "noul", "instructions": "Is this a defect?"}, "sentiment": {"type": "choice", "criteria": {"positive": None, "negative": "unhappy"}}, } response: Final = await litellm.asystemone( model=f"databricks/{_DATABRICKS_ENDPOINT}", state={"ticket": "export hangs"}, questions=questions, ) assert route.called sent: Final = respx_mock.calls[0].request assert sent.headers["authorization"] == "Bearer dapi-token" assert json.loads(sent.content) == { "model": _DATABRICKS_ENDPOINT, "state": {"ticket": "export hangs"}, "questions": questions, } assert response.model == "/mosaicml/local_model" assert isinstance(response.answers["is_defect"], NoulAnswer) assert isinstance(response.answers["sentiment"], ChoiceAnswer) assert response.hidden_params["custom_llm_provider"] == "databricks" assert response.hidden_params["model"] == f"databricks/{_DATABRICKS_ENDPOINT}" @pytest.mark.asyncio @pytest.mark.parametrize( "api_base", [ "https://workspace.example", "https://workspace.example/", "https://workspace.example/serving-endpoints", "https://workspace.example/serving-endpoints/", ], ) async def test_databricks_posts_the_systemone_body_without_model_to_the_ai_decide_route( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, api_base: str, ) -> None: monkeypatch.setenv("DATABRICKS_API_BASE", api_base) monkeypatch.delenv("DATABRICKS_API_KEY", raising=False) monkeypatch.setenv("DATABRICKS_TOKEN", "dapi-token") route: Final = respx_mock.post(_DATABRICKS_ROUTE).respond(json=_DATABRICKS_RESPONSE) questions: Final = { "is_defect": {"type": "noul", "instructions": "Is this a defect?"}, "sentiment": {"type": "choice", "criteria": {"positive": None, "negative": "unhappy"}}, } response: Final = await litellm.adecisions( model="databricks/ai_decide", state={"ticket": "export hangs"}, questions=questions, ) assert route.called sent: Final = respx_mock.calls[0].request assert sent.headers["authorization"] == "Bearer dapi-token" assert json.loads(sent.content) == {"state": {"ticket": "export hangs"}, "questions": questions} assert response.model is None assert response.answers["is_defect"] == NoulAnswer(type="noul", noul=0.9) assert isinstance(response.answers["sentiment"], ChoiceAnswer) assert response.model_extra == {"metadata": {"version": "1.0"}} assert response.hidden_params["custom_llm_provider"] == "databricks" assert response.hidden_params["model"] == "databricks/ai_decide" @pytest.mark.asyncio async def test_databricks_router_deployment_sends_its_own_key_to_its_own_base( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.setenv("DATABRICKS_API_BASE", "https://other-workspace.example") monkeypatch.setenv("DATABRICKS_API_KEY", "env-key-stays-home") router: Final = litellm.Router( model_list=[ { "model_name": "decider", "litellm_params": { "model": "databricks/ai_decide", "api_base": "https://workspace.example", "api_key": "deployment-key", }, } ] ) route: Final = respx_mock.post(_DATABRICKS_ROUTE).respond(json=_DATABRICKS_RESPONSE) response: Final = await router.asystemone( model="decider", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, ) assert route.called assert respx_mock.calls[0].request.headers["authorization"] == "Bearer deployment-key" assert response.answers["is_defect"] == NoulAnswer(type="noul", noul=0.9) _PROVIDERS_WHOSE_CHAT_MODELS_ALSO_DECIDE: Final = frozenset( {LlmProviders.OPENAI.value, LlmProviders.OPENROUTER.value, LlmProviders.HOSTED_VLLM.value} ) _DEDICATED_DECIDER_PROVIDERS: Final[tuple[str, ...]] = tuple( sorted( provider.value for provider in LlmProviders if provider.value not in _PROVIDERS_WHOSE_CHAT_MODELS_ALSO_DECIDE and ProviderConfigManager.get_provider_decisions_config(model="", provider=provider) is not None ) ) @pytest.mark.parametrize("provider", _DEDICATED_DECIDER_PROVIDERS) def test_every_dedicated_decider_provider_ships_an_evaluation_mode_cost_map_entry(provider: str) -> None: evaluation_entries: Final = tuple( name for name, info in litellm.model_cost.items() if isinstance(info, Mapping) and info.get("litellm_provider") == provider and info.get("mode") == "evaluation" ) assert evaluation_entries, f"a {provider} decider would be health-checked as chat without a mode: evaluation entry" @pytest.mark.asyncio @pytest.mark.parametrize( ("api_base", "url"), ( (None, "https://api.openai.com/v1/decisions"), ("https://gateway.example/v1", "https://gateway.example/v1/decisions"), ), ) async def test_openai_decisions_translate_systemone_to_the_openai_wire_contract_and_back( api_base: str | None, url: str, monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, decisions_pricing_fields: None, ) -> None: monkeypatch.setattr(litellm, "api_base", None) monkeypatch.delenv("OPENAI_BASE_URL", raising=False) monkeypatch.delenv("OPENAI_API_BASE", raising=False) route: Final = respx_mock.post(url).respond(json=_OPENAI_RESPONSE) response: Final = await litellm.asystemone( model="openai/gpt-6-luna", state="The package arrived broken.", questions={ "is_defect": {"type": "noul", "instructions": "Is this a defect?"}, "sentiment": { "type": "choice", "instructions": "How does the customer feel?", "criteria": {"positive": None, "negative": "unhappy"}, }, }, api_key="caller-key", api_base=api_base, ) assert route.called request: Final = respx_mock.calls[0].request assert request.headers["authorization"] == "Bearer caller-key" assert json.loads(request.content) == { "model": "gpt-6-luna", "input": "The package arrived broken.", "questions": [ {"type": "predicate", "name": "is_defect", "instructions": "Is this a defect?"}, { "type": "choice", "name": "sentiment", "instructions": "How does the customer feel?", "choices": [{"value": "positive"}, {"value": "negative", "description": "unhappy"}], }, ], } assert response.answers == {"is_defect": NoulAnswer(type="noul", noul=0.9)} assert response.hidden_params["custom_llm_provider"] == "openai" expected_cost: Final = _INPUT_TOKENS * _GPT_6_LUNA_DECISIONS_INPUT_RATE assert expected_cost > 0 assert litellm.completion_cost(completion_response=response) == pytest.approx(expected_cost) @pytest.mark.parametrize( ("settings", "env", "url", "authorization"), ( ({"openai_key": "sdk-key"}, {}, "https://api.openai.com/v1/decisions", "Bearer sdk-key"), ( {"api_key": "global-key", "openai_key": "sdk-key"}, {"OPENAI_API_KEY": "env-key"}, "https://api.openai.com/v1/decisions", "Bearer global-key", ), ( {}, {"OPENAI_API_KEY": "env-key", "OPENAI_API_BASE": "https://legacy.example/v1"}, "https://legacy.example/v1/decisions", "Bearer env-key", ), ( {"api_base": "https://sdk.example"}, {"OPENAI_API_KEY": "env-key", "OPENAI_BASE_URL": "https://env.example"}, "https://sdk.example/v1/decisions", "Bearer env-key", ), ), ids=("openai_key", "api_key_before_env", "openai_api_base_env", "api_base_before_env"), ) def test_openai_decisions_use_the_same_settings_as_other_openai_calls( settings: Mapping[str, str], env: Mapping[str, str], url: str, authorization: str, monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: for name in ("api_key", "openai_key", "api_base"): monkeypatch.setattr(litellm, name, settings.get(name)) for name in ("OPENAI_API_KEY", "OPENAI_BASE_URL", "OPENAI_API_BASE"): monkeypatch.delenv(name, raising=False) for name, value in env.items(): monkeypatch.setenv(name, value) route: Final = respx_mock.post(url).respond(json=_OPENAI_RESPONSE) litellm.systemone( model="openai/gpt-6-luna", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, ) assert route.call_count == 1 assert route.calls[0].request.headers["authorization"] == authorization @pytest.mark.asyncio async def test_openai_format_calls_to_a_systemone_provider_get_openai_format_answers( respx_mock: respx.MockRouter, ) -> None: route: Final = respx_mock.post("https://api.typesafe.ai/v1/systemone").respond(json=_RESPONSE) response: Final = await litellm.adecisions( model="typesafe/jev-1.13", input="review", questions=[ {"type": "predicate", "name": "is_defect", "instructions": "Is this a defect?"}, { "type": "choice", "name": "sentiment", "instructions": "Tone?", "choices": [{"value": "positive"}, {"value": "negative"}], }, ], api_key="caller-key", ) assert route.called assert tuple(json.loads(respx_mock.calls[0].request.content)["questions"]) == ("is_defect", "sentiment") assert isinstance(response, OpenAIDecisionResponse) assert [answer.model_dump(mode="json") for answer in response.answers] == [ {"type": "predicate", "name": "is_defect", "probability": 0.9}, { "type": "choice", "name": "sentiment", "choice": "positive", "probabilities": [{"value": "positive", "probability": 0.8}, {"value": "negative", "probability": 0.2}], "confidence": 0.8, }, ] @pytest.mark.asyncio @pytest.mark.parametrize( ("model", "request_kwargs", "message"), ( ( "typesafe/jev-1.13", { "input": [ { "role": "user", "content": [{"type": "input_image", "image_url": "data:image/png;base64,AA=="}], } ], "questions": [{"type": "predicate", "instructions": "Is this a defect?"}], }, "cannot serve this request", ), ), ids=("image_to_systemone_provider",), ) async def test_requests_a_provider_cannot_serve_are_rejected_before_http( respx_mock: respx.MockRouter, model: str, request_kwargs: Mapping[str, object], message: str, ) -> None: with pytest.raises(litellm.BadRequestError, match=message): await litellm.adecisions(model=model, api_key="caller-key", **request_kwargs) assert len(respx_mock.calls) == 0 @pytest.mark.asyncio async def test_openrouter_decisions_uses_provider_reported_cost_without_cost_map( respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.delitem(litellm.model_cost, "openrouter/typesafe/jev-1.13", raising=False) cost: Final = 1.1802e-5 route: Final = respx_mock.post("https://openrouter.ai/api/alpha/decisions").respond( json={ "model": "typesafe/jev-1.13", "answers": _RESPONSE["answers"], "usage": { "input_tokens": _INPUT_TOKENS, "output_tokens": _OUTPUT_TOKENS, "cost": cost, }, } ) response: Final = await litellm.asystemone( model="openrouter/typesafe/jev-1.13", state={"source": "unit-test"}, questions=_QUESTIONS, api_key="caller-key", ) assert route.called assert get_response_cost_from_hidden_params(response.hidden_params) == cost _SAFETY_IDENTIFIER: Final = "end-user-7" _PREDICATE_QUESTIONS: Final[tuple[Mapping[str, object], ...]] = ( {"type": "predicate", "name": "is_defect", "instructions": "Is this a defect?"}, ) _PREDICATE_REQUESTS: Final[tuple[tuple[str, Mapping[str, object]], ...]] = ( ("decisions", MappingProxyType({"input": "review", "questions": _PREDICATE_QUESTIONS})), ( "systemone", MappingProxyType( {"state": "review", "questions": {"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}} ), ), ) _PREDICATE_REQUEST_IDS: Final = ("decisions", "systemone") @pytest.mark.parametrize(("api", "request_kwargs"), _PREDICATE_REQUESTS, ids=_PREDICATE_REQUEST_IDS) def test_safety_identifier_is_refused_before_http_when_the_provider_cannot_take_it( api: str, request_kwargs: 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) with pytest.raises(litellm.UnsupportedParamsError, match=r"safety_identifier.*drop_params") as caught: getattr(litellm, api)( model="perplexity/pplx-decider-v1-27b", safety_identifier=_SAFETY_IDENTIFIER, api_key="caller-key", **request_kwargs, ) assert caught.value.status_code == 400 assert not route.called @pytest.mark.asyncio @pytest.mark.parametrize("scope", ("call", "global")) async def test_safety_identifier_is_dropped_from_the_wire_under_drop_params( scope: str, respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setattr(litellm, "drop_params", scope == "global") route: Final = respx_mock.post("https://api.perplexity.ai/v1/decisions").respond(json=_RESPONSE) response: Final = await litellm.adecisions( model="perplexity/pplx-decider-v1-27b", input="review", questions=_PREDICATE_QUESTIONS, safety_identifier=_SAFETY_IDENTIFIER, api_key="caller-key", **({"drop_params": True} if scope == "call" else {}), ) assert route.called assert json.loads(respx_mock.calls[0].request.content) == { "model": "pplx-decider-v1-27b", "state": "review", "questions": {"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, } assert isinstance(response, OpenAIDecisionResponse) assert [answer.model_dump(mode="json") for answer in response.answers] == [ {"type": "predicate", "name": "is_defect", "probability": 0.9} ] @pytest.mark.asyncio @pytest.mark.parametrize(("api", "request_kwargs"), _PREDICATE_REQUESTS, ids=_PREDICATE_REQUEST_IDS) @pytest.mark.parametrize("drop_params", (False, True), ids=("strict", "drop_params")) async def test_safety_identifier_reaches_openai_whether_or_not_params_are_dropped( drop_params: bool, api: str, request_kwargs: Mapping[str, object], respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr(litellm, "drop_params", drop_params) monkeypatch.setattr(litellm, "api_base", None) monkeypatch.delenv("OPENAI_BASE_URL", raising=False) monkeypatch.delenv("OPENAI_API_BASE", raising=False) route: Final = respx_mock.post("https://api.openai.com/v1/decisions").respond(json=_OPENAI_RESPONSE) await getattr(litellm, f"a{api}")( model="openai/gpt-6-luna", safety_identifier=_SAFETY_IDENTIFIER, api_key="caller-key", **request_kwargs, ) assert route.called assert json.loads(respx_mock.calls[0].request.content) == { "model": "gpt-6-luna", "input": "review", "questions": [{"type": "predicate", "name": "is_defect", "instructions": "Is this a defect?"}], "safety_identifier": _SAFETY_IDENTIFIER, } @pytest.mark.asyncio @pytest.mark.parametrize(("api", "request_kwargs"), _PREDICATE_REQUESTS, ids=_PREDICATE_REQUEST_IDS) async def test_non_string_safety_identifier_is_rejected_before_http( api: str, request_kwargs: Mapping[str, object], respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setattr(litellm, "drop_params", False) route: Final = respx_mock.post("https://api.openai.com/v1/decisions").respond(json=_OPENAI_RESPONSE) invalid_safety_identifier: Final[Mapping[str, object]] = MappingProxyType({"safety_identifier": 7}) with pytest.raises(litellm.BadRequestError, match=r"(?s)Invalid Decisions request.*safety_identifier"): await getattr(litellm, f"a{api}")( model="openai/gpt-6-luna", api_key="caller-key", **request_kwargs, **invalid_safety_identifier ) assert not route.called @pytest.mark.asyncio @pytest.mark.parametrize(("api", "request_kwargs"), _PREDICATE_REQUESTS, ids=_PREDICATE_REQUEST_IDS) async def test_non_string_safety_identifier_is_dropped_from_the_wire_under_drop_params( api: str, request_kwargs: Mapping[str, object], respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setattr(litellm, "drop_params", False) monkeypatch.setattr(litellm, "api_base", None) monkeypatch.delenv("OPENAI_BASE_URL", raising=False) monkeypatch.delenv("OPENAI_API_BASE", raising=False) route: Final = respx_mock.post("https://api.openai.com/v1/decisions").respond(json=_OPENAI_RESPONSE) dropped_safety_identifier: Final[Mapping[str, object]] = MappingProxyType( {"safety_identifier": 7, "drop_params": True} ) await getattr(litellm, f"a{api}")( model="openai/gpt-6-luna", api_key="caller-key", **request_kwargs, **dropped_safety_identifier ) assert route.called assert json.loads(respx_mock.calls[0].request.content) == { "model": "gpt-6-luna", "input": "review", "questions": [{"type": "predicate", "name": "is_defect", "instructions": "Is this a defect?"}], } @pytest.mark.asyncio @pytest.mark.parametrize( "api_base", ( "https://res.services.ai.azure.com", "https://res.services.ai.azure.com/", "https://res.services.ai.azure.com/models", "https://res.services.ai.azure.com/openai/v1", "https://res.services.ai.azure.com/api/projects/proj", "https://res.services.ai.azure.com/api/projects/proj/openai/v1", "https://res.services.ai.azure.com/api/projects/proj/models", "https://res.services.ai.azure.com/models?api-version=2024-05-01-preview", ), ) async def test_azure_ai_decision_posts_deployment_to_foundry_systemone_route( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, api_base: str, ) -> None: monkeypatch.delenv("AZURE_AI_API_BASE", raising=False) monkeypatch.delenv("AZURE_AI_API_KEY", raising=False) route: Final = respx_mock.post("https://res.services.ai.azure.com/providers/microsoft/v1/systemone").respond( json={**_RESPONSE, "model": "microsoft-decision-1"} ) response: Final = await litellm.asystemone( model="azure_ai/decision-1", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, api_base=api_base, api_key="foundry-key", ) assert route.called request: Final = respx_mock.calls[0].request assert request.headers["authorization"] == "Bearer foundry-key" assert "api-key" not in request.headers assert json.loads(request.content) == { "model": "decision-1", "state": "review", "questions": {"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, } assert response.answers == {"is_defect": NoulAnswer(type="noul", noul=0.9)} assert response._hidden_params["model"] == "azure_ai/decision-1" @pytest.mark.asyncio async def test_azure_ai_systemone_uses_its_decisions_pricing_fields( respx_mock: respx.MockRouter, decisions_pricing_fields: None ) -> None: route: Final = respx_mock.post("https://res.services.ai.azure.com/providers/microsoft/v1/systemone").respond( json={**_RESPONSE, "model": "microsoft-decision-1"} ) response: Final = await litellm.asystemone( model="azure_ai/Microsoft-Decision-1", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, api_base="https://res.services.ai.azure.com", api_key="foundry-key", ) assert route.called assert response._hidden_params["provider_response_model"] == "azure_ai/Microsoft-Decision-1" assert litellm.completion_cost(completion_response=response) == pytest.approx( _INPUT_TOKENS * _MICROSOFT_DECISION_1_DECISIONS_INPUT_RATE ) @pytest.mark.asyncio async def test_azure_ai_decision_reads_foundry_env_and_keeps_gateway_path_prefix( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.setenv("AZURE_AI_API_BASE", "https://gateway.example.com/foundry/models") monkeypatch.setenv("AZURE_AI_API_KEY", "env-foundry-key") route: Final = respx_mock.post("https://gateway.example.com/foundry/providers/microsoft/v1/systemone").respond( json=_RESPONSE ) await litellm.asystemone( model="azure_ai/decision-1", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, ) assert route.called assert respx_mock.calls[0].request.headers["authorization"] == "Bearer env-foundry-key" @pytest.mark.asyncio async def test_azure_ai_decision_requires_api_base_before_http( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("AZURE_AI_API_BASE", raising=False) monkeypatch.setenv("AZURE_AI_API_KEY", "foundry-key") with pytest.raises(litellm.BadRequestError, match="Missing AZURE_AI_API_BASE"): await litellm.asystemone( model="azure_ai/decision-1", state="review", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, ) 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]] = { "model": "accounts/fireworks/routers/rune", "answers": {"is_defect": {"type": "noul", "noul": 0.9902915235185259}}, "usage": {"input_tokens": 300, "cached_input_tokens": 200, "output_tokens": 0}, } @pytest.mark.asyncio async def test_fireworks_ai_posts_the_jev_body_to_systemone_with_the_account_model_name( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("FIREWORKS_API_BASE", raising=False) monkeypatch.delenv("FIREWORKS_API_KEY", raising=False) monkeypatch.setenv("FIREWORKS_AI_API_KEY", "fireworks-key") route: Final = respx_mock.post(_FIREWORKS_URL).respond(json=_FIREWORKS_RESPONSE) response: Final = await litellm.asystemone( model=_FIREWORKS_MODEL, state="The package arrived broken.", questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, ) assert route.called assert respx_mock.calls[0].request.headers["authorization"] == "Bearer fireworks-key" assert json.loads(respx_mock.calls[0].request.content) == { "model": "accounts/fireworks/routers/rune", "state": "The package arrived broken.", "questions": {"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}, } assert response.model == _FIREWORKS_RESPONSE["model"] assert response.answers["is_defect"] == NoulAnswer(type="noul", noul=0.9902915235185259) @pytest.mark.asyncio async def test_fireworks_ai_bills_cached_input_tokens_at_the_cache_read_rate( monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter, ) -> None: monkeypatch.delenv("FIREWORKS_API_BASE", raising=False) respx_mock.post(_FIREWORKS_URL).respond(json=_FIREWORKS_RESPONSE) prices: Final = litellm.model_cost[_FIREWORKS_MODEL] uncached: Final = 100 * float(prices["input_cost_per_token"]) cached: Final = 200 * float(prices["cache_read_input_token_cost"]) recording_logger: Final = _RecordingLogger() monkeypatch.setattr(litellm, "callbacks", [recording_logger]) response: Final = await litellm.adecisions( model=_FIREWORKS_MODEL, input="The package arrived broken.", questions=[{"type": "predicate", "name": "is_defect", "instructions": "Is this a defect?"}], api_key="caller-key", ) await _drain_logging_worker() assert cached > 0 assert response.usage.input_tokens_details.cached_tokens == 200 assert recording_logger.standard_logging_object is not None assert recording_logger.standard_logging_object["response_cost"] == pytest.approx(uncached + cached) assert recording_logger.standard_logging_object["response_cost"] < 300 * float(prices["input_cost_per_token"])