scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-14 04:09:09 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 97

Merge remote-tracking branch 'origin/main' into litellm_mat_140_base_count_gates

Browse source 
# Conflicts:
#	scripts/type_discipline_gate.py
#	type-discipline-budget.json
This commit is contained in:
mateo-berri 2026-10-07 22:28:11 -07:00
commit d24215c105
42 changed files with 2873 additions and 182 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
AGENTS.md
View file

@ -97,6 +97,7 @@ Follow these coding conventions for new/updated code (a three-line fix in a lega
- Annotate every variable with `: Final` (LIT010). Unpacking and walrus targets cannot carry the annotation, so they are implicitly final. Don't rebind them. Never rebind or mutate function parameters (LIT011); `self`/`cls` attribute stores are the exception. If rebinding or in-place mutation is truly unavoidable, suppress with `# rebind-ok: <reason>`
- Qualify every TypedDict field with `ReadOnly[...]` (LIT012), which nests freely with `Required` / `NotRequired` / `Annotated` in any order. If making the key writable is truly unavoidable, suppress with `# writable-ok: <reason>`
- Comprehensions take at most one `for` clause and one `if` clause (LIT014); split stacked clauses into a helper generator, a named intermediate, or a plain loop. Suppress with `# comprehension-ok: <reason>` only when unavoidable
- Every pydantic model must be frozen (LIT015), set via `model_config = ConfigDict(frozen=True)`, a dict-literal `model_config`, an inner `class Config`, or the class keywords. Subclasses inherit it unless they override it. Replace in-place field writes with `model_copy(update=...)`. If making the model mutable is truly unavoidable, suppress with `# frozen-ok: <reason>` on the `class` line
- Use dependency injection
- Fully typed; no `Any` or coarse types like `dict[str, Any]` or just `dict`. Every function parameter must be strongly typed
- Use tagged unions + match

2
gateway/routes/allowlist.py
View file

@ -62,6 +62,8 @@ GATEWAY_PATH_PREFIXES: tuple[str, ...] = (
"/rerank",
"/v1/decisions",
"/decisions",
"/v1/systemone",
"/systemone",
"/v1/ocr",
"/ocr",
"/v1/rag/",

3
litellm/__init__.py
View file

@ -1682,6 +1682,9 @@ if TYPE_CHECKING:
from .llms.strands_decider.decisions.transformation import (
StrandsDeciderDecisionsConfig as StrandsDeciderDecisionsConfig,
)
from .llms.openai.decisions.transformation import (
OpenAIDecisionsConfig as OpenAIDecisionsConfig,
)
from .llms.nvidia_nim.rerank.transformation import (
NvidiaNimRerankConfig as NvidiaNimRerankConfig,
)

2
litellm/_lazy_imports_registry.py
View file

@ -161,6 +161,7 @@ LLM_CONFIG_NAMES: Final = (
"OpenRouterDecisionsConfig",
"CloudflareDecisionsConfig",
"StrandsDeciderDecisionsConfig",
"OpenAIDecisionsConfig",
"NvidiaNimRerankConfig",
"NvidiaNimRankingConfig",
"VertexAIRerankConfig",
@ -720,6 +721,7 @@ _LLM_CONFIGS_IMPORT_MAP: Final = {
".llms.strands_decider.decisions.transformation",
"StrandsDeciderDecisionsConfig",
),
"OpenAIDecisionsConfig": (".llms.openai.decisions.transformation", "OpenAIDecisionsConfig"),
"NvidiaNimRerankConfig": (
".llms.nvidia_nim.rerank.transformation",
"NvidiaNimRerankConfig",

41
litellm/cost_calculator.py
View file

@ -9,7 +9,7 @@ from typing import TYPE_CHECKING, Any, Final, Literal, cast
from httpx import Response
from pydantic import BaseModel
from typing_extensions import ReadOnly, TypedDict
from typing_extensions import ReadOnly, TypedDict, assert_never
import litellm
import litellm._logging
@ -100,7 +100,7 @@ from litellm.llms.vertex_ai.cost_calculator import cost_router as google_cost_ro
from litellm.llms.xai.cost_calculator import cost_per_token as xai_cost_per_token
from litellm.responses.utils import ResponseAPILoggingUtils
from litellm.types.agents import LiteLLMSendMessageResponse
from litellm.types.decisions import DecisionsResponse, DecisionsUsage
from litellm.types.decisions import DecisionsResponse, DecisionsUsage, OpenAIDecisionResponse, OpenAIDecisionUsage
from litellm.types.llms.base import CachedTokensDetails, LiteLLMBaseModel
from litellm.types.llms.openai import (
HttpxBinaryResponseContent,
@ -1034,6 +1034,8 @@ def get_usage_object(
),
)
if isinstance(completion_response, (DecisionsResponse, OpenAIDecisionResponse)):
return None if completion_response.usage is None else _decisions_usage(completion_response.usage)
if usage_obj is None:
return None
if isinstance(usage_obj, Usage):
@ -1064,12 +1066,37 @@ def get_usage_object(
return None
def _decisions_prompt_tokens_details(usage: DecisionsUsage | OpenAIDecisionUsage) -> PromptTokensDetailsWrapper:
match usage:
case DecisionsUsage():
return PromptTokensDetailsWrapper(
cached_tokens=usage.cached_tokens, cache_write_tokens=usage.cache_write_tokens
)
case OpenAIDecisionUsage():
return PromptTokensDetailsWrapper(
cached_tokens=usage.input_tokens_details.cached_tokens,
cache_write_tokens=usage.input_tokens_details.cache_write_tokens,
)
case _:
assert_never(usage)
def _decisions_usage(usage: DecisionsUsage | OpenAIDecisionUsage) -> Usage:
return Usage(
prompt_tokens=usage.input_tokens,
completion_tokens=usage.output_tokens,
total_tokens=usage.input_tokens + usage.output_tokens,
prompt_tokens_details=_decisions_prompt_tokens_details(usage),
)
def _is_known_usage_objects(usage_obj):
"""Returns True if the usage obj is a known Usage type"""
return (
isinstance(usage_obj, litellm.Usage)
or isinstance(usage_obj, ResponseAPIUsage)
or isinstance(usage_obj, DecisionsUsage)
or isinstance(usage_obj, OpenAIDecisionUsage)
or TranscriptionUsageObjectTransformation.is_transcription_usage_object(usage_obj)
)
@ -1481,11 +1508,8 @@ def completion_cost(
"usage",
litellm.Usage(**_usage_for_dump.model_dump()),
)
if isinstance(usage_obj, DecisionsUsage):
_usage = {
"prompt_tokens": usage_obj.input_tokens,
"completion_tokens": usage_obj.output_tokens,
}
if isinstance(usage_obj, (DecisionsUsage, OpenAIDecisionUsage)):
_usage = _decisions_usage(usage_obj).model_dump()
elif usage_obj is None:
_usage = {}
elif isinstance(usage_obj, BaseModel):
@ -1977,7 +2001,8 @@ def response_cost_calculator(
| OpenAIModerationResponse
| Response
| SearchResponse
| DecisionsResponse,
| DecisionsResponse
| OpenAIDecisionResponse,
model: str,
custom_llm_provider: str | None,
call_type: Literal[

144
litellm/decisions/main.py
View file

@ -1,34 +1,56 @@
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from typing import Final
from typing import Final, TypeAlias
import httpx
from pydantic import TypeAdapter, ValidationError
from typing_extensions import assert_never
import litellm
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.decisions.transformation import BaseDecisionsConfig
from litellm.llms.base_llm.decisions.transformation import (
BaseDecisionsConfig,
ir_to_systemone_response,
systemone_request_to_ir,
)
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,
DecisionsIRRequest,
DecisionsIRResponse,
DecisionsJSON,
DecisionsRequest,
DecisionsRequestBody,
DecisionsResponse,
OpenAIDecisionInput,
OpenAIDecisionQuestion,
OpenAIDecisionRequestBody,
OpenAIDecisionResponse,
UnsupportedDecisionsRequest,
)
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager, client
_DECISIONS_REQUEST_ADAPTER: Final[TypeAdapter[DecisionsRequest]] = TypeAdapter(DecisionsRequest)
DecisionsQuestions: TypeAlias = (
Mapping[str, DecisionQuestion | Mapping[str, object]] | Sequence[OpenAIDecisionQuestion | Mapping[str, object]]
)
DecisionsRequestFormat: TypeAlias = DecisionsRequestBody | OpenAIDecisionRequestBody
_SYSTEMONE_REQUEST_ADAPTER: Final[TypeAdapter[DecisionsRequestBody]] = TypeAdapter(DecisionsRequestBody)
_OPENAI_REQUEST_ADAPTER: Final[TypeAdapter[OpenAIDecisionRequestBody]] = TypeAdapter(OpenAIDecisionRequestBody)
_HANDLER: Final = BaseLLMHTTPHandler()
@dataclass(frozen=True, slots=True, repr=False)
class _DecisionsCall:
model: str
requested_model: str
custom_llm_provider: str
provider_config: BaseDecisionsConfig
request: DecisionsRequest
request: DecisionsRequestFormat
ir_request: DecisionsIRRequest
body: Mapping[str, object] = field(repr=False)
api_base: str
api_key: str | None = field(repr=False)
logging_obj: LiteLLMLoggingObj | None
@ -59,11 +81,37 @@ def _provider_config(model: str, custom_llm_provider: str) -> BaseDecisionsConfi
return provider_config
def _validate_request(
*,
state: DecisionsJSON | None,
questions: DecisionsQuestions | None,
decision_input: OpenAIDecisionInput | None,
safety_identifier: str | None,
) -> DecisionsRequestFormat:
if decision_input is None:
return _SYSTEMONE_REQUEST_ADAPTER.validate_python({"state": state, "questions": questions})
return _OPENAI_REQUEST_ADAPTER.validate_python(
{"input": decision_input, "questions": questions, "safety_identifier": safety_identifier}
)
def _ir_request(request: DecisionsRequestFormat) -> DecisionsIRRequest:
match request:
case DecisionsRequestBody():
return systemone_request_to_ir(request)
case OpenAIDecisionRequestBody():
return openai_request_to_ir(request)
case _:
assert_never(request)
def _prepare_call(
*,
model: str,
state: DecisionsJSON,
questions: Mapping[str, DecisionQuestion | Mapping[str, object]],
state: DecisionsJSON | None,
questions: DecisionsQuestions | None,
decision_input: OpenAIDecisionInput | None,
safety_identifier: str | None,
api_key: str | None,
api_base: str | None,
timeout: float | httpx.Timeout | None,
@ -85,9 +133,15 @@ def _prepare_call(
model=model,
llm_provider=provider,
)
if state is not None and decision_input is not None:
raise litellm.BadRequestError(
message="Pass either state (System One format) or input (OpenAI format) to the Decisions API, not both",
model=model,
llm_provider=provider,
)
try:
request: Final = _DECISIONS_REQUEST_ADAPTER.validate_python(
{"model": canonical_model, "state": state, "questions": questions}
request: Final = _validate_request(
state=state, questions=questions, decision_input=decision_input, safety_identifier=safety_identifier
)
except ValidationError as error:
raise litellm.BadRequestError(
@ -111,6 +165,17 @@ def _prepare_call(
llm_provider=provider,
)
ir_request: Final = _ir_request(request)
body: Final = provider_config.transform_decisions_request(
model=canonical_model, request=ir_request, custom_llm_provider=provider
)
if isinstance(body, UnsupportedDecisionsRequest):
raise litellm.BadRequestError(
message=f"Decisions provider '{provider}' cannot serve this request: {body.reason}",
model=model,
llm_provider=provider,
)
logging_obj: Final = kwargs.get("litellm_logging_obj")
if isinstance(logging_obj, LiteLLMLoggingObj):
logging_obj.update_from_kwargs(
@ -124,9 +189,12 @@ def _prepare_call(
)
return _DecisionsCall(
model=canonical_model,
requested_model=model,
custom_llm_provider=provider,
provider_config=provider_config,
request=request,
ir_request=ir_request,
body=body,
api_base=resolved_api_base,
api_key=resolved_api_key,
logging_obj=logging_obj if isinstance(logging_obj, LiteLLMLoggingObj) else None,
@ -135,6 +203,30 @@ def _prepare_call(
)
def _format_response(response: DecisionsIRResponse, call: _DecisionsCall) -> DecisionsResponse | OpenAIDecisionResponse:
formatted: Final = _formatted_response(response, call)
formatted.set_hidden_params(
{
"model": f"{call.custom_llm_provider}/{call.model}",
"custom_llm_provider": call.custom_llm_provider,
"provider_response_model": f"{call.custom_llm_provider}/{call.model}",
}
)
return formatted
def _formatted_response(
response: DecisionsIRResponse, call: _DecisionsCall
) -> DecisionsResponse | OpenAIDecisionResponse:
match call.request:
case DecisionsRequestBody():
return ir_to_systemone_response(response, call.ir_request)
case OpenAIDecisionRequestBody():
return ir_to_openai_response(response, call.ir_request, call.requested_model)
case _:
assert_never(call.request)
def _map_upstream_exception(error: Exception, call: _DecisionsCall) -> Exception:
if isinstance(error, BaseLLMException) and error.status_code_is_synthesized:
provider_label: Final = f"{call.custom_llm_provider[0].upper()}{call.custom_llm_provider[1:]}Exception"
@ -153,19 +245,23 @@ def _map_upstream_exception(error: Exception, call: _DecisionsCall) -> Exception
@client
async def adecisions(
model: str,
state: DecisionsJSON,
questions: Mapping[str, DecisionQuestion | Mapping[str, object]],
state: DecisionsJSON | None = None,
questions: DecisionsQuestions | None = None,
api_key: str | None = None,
api_base: str | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
extra_headers: Mapping[str, str] | None = None,
input: OpenAIDecisionInput | None = None,
safety_identifier: str | None = None,
**kwargs: object,
) -> DecisionsResponse:
) -> DecisionsResponse | OpenAIDecisionResponse:
call: Final = _prepare_call(
model=model,
state=state,
questions=questions,
decision_input=input,
safety_identifier=safety_identifier,
api_key=api_key,
api_base=api_base,
timeout=timeout,
@ -174,12 +270,13 @@ async def adecisions(
kwargs=kwargs,
)
try:
return await _HANDLER.adecisions(
response: Final = await _HANDLER.adecisions(
model=call.model,
custom_llm_provider=call.custom_llm_provider,
logging_obj=call.logging_obj,
provider_config=call.provider_config,
request=call.request,
request=call.ir_request,
body=call.body,
api_base=call.api_base,
api_key=call.api_key,
headers=call.headers,
@ -187,24 +284,29 @@ async def adecisions(
)
except Exception as error:
raise _map_upstream_exception(error, call) from error
return _format_response(response, call)
@client
def decisions(
model: str,
state: DecisionsJSON,
questions: Mapping[str, DecisionQuestion | Mapping[str, object]],
state: DecisionsJSON | None = None,
questions: DecisionsQuestions | None = None,
api_key: str | None = None,
api_base: str | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
extra_headers: Mapping[str, str] | None = None,
input: OpenAIDecisionInput | None = None,
safety_identifier: str | None = None,
**kwargs: object,
) -> DecisionsResponse:
) -> DecisionsResponse | OpenAIDecisionResponse:
call: Final = _prepare_call(
model=model,
state=state,
questions=questions,
decision_input=input,
safety_identifier=safety_identifier,
api_key=api_key,
api_base=api_base,
timeout=timeout,
@ -213,12 +315,13 @@ def decisions(
kwargs=kwargs,
)
try:
return _HANDLER.decisions(
response: Final = _HANDLER.decisions(
model=call.model,
custom_llm_provider=call.custom_llm_provider,
logging_obj=call.logging_obj,
provider_config=call.provider_config,
request=call.request,
request=call.ir_request,
body=call.body,
api_base=call.api_base,
api_key=call.api_key,
headers=call.headers,
@ -226,6 +329,7 @@ def decisions(
)
except Exception as error:
raise _map_upstream_exception(error, call) from error
return _format_response(response, call)
__all__ = ["adecisions", "decisions"]

3
litellm/litellm_core_utils/litellm_logging.py
View file

@ -120,7 +120,7 @@ from litellm.llms.base_llm.search.transformation import SearchResponse
from litellm.responses.utils import ResponseAPILoggingUtils
from litellm.types.agents import LiteLLMSendMessageResponse
from litellm.types.containers.main import ContainerObject
from litellm.types.decisions import DecisionsResponse
from litellm.types.decisions import DecisionsResponse, OpenAIDecisionResponse
from litellm.types.integrations.s3_v2 import S3PartitionGranularity
from litellm.types.interactions import (
InteractionsAPIResponse,
@ -2650,6 +2650,7 @@ class Logging(LiteLLMLoggingBaseClass):
or isinstance(logging_result, OCRResponse) # OCR
or isinstance(logging_result, SearchResponse) # Search API
or isinstance(logging_result, DecisionsResponse)
or isinstance(logging_result, OpenAIDecisionResponse)
or (
isinstance(logging_result, InteractionsAPIResponse)
and logging_result.usage is not None

326
litellm/llms/base_llm/decisions/transformation.py
View file

@ -1,17 +1,299 @@
import json
from abc import ABC
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from types import MappingProxyType
from typing import Final
import httpx
from pydantic import TypeAdapter, ValidationError
from typing_extensions import assert_never
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.secret_managers.main import get_secret_str
from litellm.types.decisions import DecisionsRequest, DecisionsResponse
from litellm.types.decisions import (
ChoiceAnswer,
ChoiceQuestion,
DecisionAnswer,
DecisionQuestion,
DecisionsIRAnswer,
DecisionsIRChoiceAnswer,
DecisionsIRChoiceOption,
DecisionsIRChoiceProbability,
DecisionsIRChoiceQuestion,
DecisionsIRMessages,
DecisionsIRPredicateAnswer,
DecisionsIRPredicateQuestion,
DecisionsIRQuestion,
DecisionsIRRefusal,
DecisionsIRRequest,
DecisionsIRResponse,
DecisionsIRScoreAnswer,
DecisionsIRScoreLevel,
DecisionsIRScoreProbability,
DecisionsIRScoreQuestion,
DecisionsIRState,
DecisionsIRUsage,
DecisionsJSON,
DecisionsRequestBody,
DecisionsResponse,
DecisionsUsage,
NoulAnswer,
NoulQuestion,
OpenAIDecisionInputImage,
OpenAIDecisionInputMessage,
OpenAIDecisionInputText,
ScoreAnswer,
ScoreQuestion,
UnsupportedDecisionsRequest,
systemone_choice_key,
)
PAYLOAD_ADAPTER: Final[TypeAdapter[object]] = TypeAdapter(object)
_RESPONSE_ADAPTER: Final[TypeAdapter[DecisionsResponse]] = TypeAdapter(DecisionsResponse)
_PAYLOAD_ADAPTER: Final[TypeAdapter[object]] = TypeAdapter(object)
_SYSTEMONE_RESPONSE_ADAPTER: Final[TypeAdapter[DecisionsResponse]] = TypeAdapter(DecisionsResponse)
_RESERVED_HEADERS: Final[frozenset[str]] = frozenset({"authorization", "content-type"})
_TEXT_ONLY: Final = UnsupportedDecisionsRequest(
reason="input_image content parts are not supported because System One providers accept text input only"
)
def decisions_text(value: DecisionsJSON) -> str:
return value if isinstance(value, str) else json.dumps(value)
def systemone_keys(questions: Sequence[DecisionsIRQuestion]) -> tuple[str, ...]:
names: Final = tuple(question.name for question in questions if question.name is not None)
if len(frozenset(names)) == len(questions):
return names
return tuple(str(index) for index in range(len(questions)))
def _ir_question(name: str, question: DecisionQuestion) -> DecisionsIRQuestion:
extra: Final = MappingProxyType(question.model_extra or {})
match question:
case NoulQuestion():
return DecisionsIRPredicateQuestion(
name=name, instructions=question.instructions, criteria=question.criteria, extra=extra
)
case ChoiceQuestion():
return DecisionsIRChoiceQuestion(
name=name,
instructions=question.instructions,
choices=tuple(
DecisionsIRChoiceOption(value=value, description=description)
for value, description in question.criteria.items()
),
extra=extra,
)
case ScoreQuestion():
return DecisionsIRScoreQuestion(
name=name,
instructions=question.instructions,
levels=tuple(
DecisionsIRScoreLevel(label=criterion, description=None) for criterion in question.criteria
),
extra=extra,
)
case _:
assert_never(question)
def systemone_request_to_ir(request: DecisionsRequestBody) -> DecisionsIRRequest:
return DecisionsIRRequest(
input=DecisionsIRState(state=request.state),
questions=tuple(_ir_question(name, question) for name, question in request.questions.items()),
)
def _has_image(message: OpenAIDecisionInputMessage) -> bool:
return not isinstance(message.content, str) and any(
isinstance(part, OpenAIDecisionInputImage) for part in message.content
)
def _message_text(message: OpenAIDecisionInputMessage) -> str:
if isinstance(message.content, str):
return message.content
return "\n\n".join(part.text for part in message.content if isinstance(part, OpenAIDecisionInputText))
def _systemone_state(
decision_input: DecisionsIRState | DecisionsIRMessages,
) -> DecisionsJSON | UnsupportedDecisionsRequest:
match decision_input:
case DecisionsIRState():
return decision_input.state
case DecisionsIRMessages():
if any(_has_image(message) for message in decision_input.messages):
return _TEXT_ONLY
return "\n\n".join(_message_text(message) for message in decision_input.messages)
case _:
assert_never(decision_input)
def _optional(key: str, value: object) -> Mapping[str, object]:
return {} if value is None else {key: value}
def _level_criterion(level: DecisionsIRScoreLevel) -> DecisionsJSON:
return level.label if level.description is None else f"{decisions_text(level.label)}: {level.description}"
def _systemone_question(question: DecisionsIRQuestion) -> Mapping[str, object]:
match question:
case DecisionsIRPredicateQuestion():
return {
"type": "noul",
**_optional("instructions", question.instructions),
**_optional("criteria", question.criteria),
**question.extra,
}
case DecisionsIRChoiceQuestion():
return {
"type": "choice",
**_optional("instructions", question.instructions),
"criteria": {systemone_choice_key(option.value): option.description for option in question.choices},
**question.extra,
}
case DecisionsIRScoreQuestion():
return {
"type": "score",
**_optional("instructions", question.instructions),
"criteria": [_level_criterion(level) for level in question.levels],
**question.extra,
}
case _:
assert_never(question)
def ir_to_systemone_request(
model: str, request: DecisionsIRRequest
) -> Mapping[str, object] | UnsupportedDecisionsRequest:
state: Final = _systemone_state(request.input)
if isinstance(state, UnsupportedDecisionsRequest):
return state
keyed_questions: Final = zip(systemone_keys(request.questions), request.questions, strict=True)
return {
"model": model,
"state": state,
"questions": {key: _systemone_question(question) for key, question in keyed_questions},
}
def _ir_choice_answer(question: DecisionsIRChoiceQuestion, answer: ChoiceAnswer) -> DecisionsIRChoiceAnswer:
typed_values: Final = {systemone_choice_key(option.value): option.value for option in question.choices}
return DecisionsIRChoiceAnswer(
choice=typed_values.get(answer.choice, answer.choice),
confidence=answer.confidence,
probabilities=tuple(
DecisionsIRChoiceProbability(value=typed_values.get(key, key), probability=probability)
for key, probability in answer.probabilities.items()
),
extra=MappingProxyType(answer.model_extra or {}),
)
def _ir_score_answer(question: DecisionsIRScoreQuestion, answer: ScoreAnswer) -> DecisionsIRScoreAnswer:
return DecisionsIRScoreAnswer(
score=answer.score,
confidence=answer.confidence,
probabilities=tuple(
DecisionsIRScoreProbability(
value=index,
label=answer.legend.get(str(index), level.label),
probability=answer.probabilities.get(str(index), 0.0),
)
for index, level in enumerate(question.levels)
),
extra=MappingProxyType(answer.model_extra or {}),
)
def _ir_answer(question: DecisionsIRQuestion, answer: DecisionAnswer | None) -> DecisionsIRAnswer:
match question, answer:
case DecisionsIRPredicateQuestion(), NoulAnswer():
return DecisionsIRPredicateAnswer(probability=answer.noul, extra=MappingProxyType(answer.model_extra or {}))
case DecisionsIRChoiceQuestion(), ChoiceAnswer():
return _ir_choice_answer(question, answer)
case DecisionsIRScoreQuestion(), ScoreAnswer():
return _ir_score_answer(question, answer)
case _:
return DecisionsIRRefusal()
def systemone_response_to_ir(response: DecisionsResponse, request: DecisionsIRRequest) -> DecisionsIRResponse:
usage: Final = response.usage or DecisionsUsage()
keyed_questions: Final = zip(systemone_keys(request.questions), request.questions, strict=True)
return DecisionsIRResponse(
model=response.model,
answers=tuple(_ir_answer(question, response.answers.get(key)) for key, question in keyed_questions),
usage=DecisionsIRUsage(
input_tokens=usage.input_tokens,
output_tokens=usage.output_tokens,
cached_tokens=usage.cached_tokens,
cache_write_tokens=usage.cache_write_tokens,
extra=MappingProxyType(usage.model_extra or {}),
),
extra=MappingProxyType(response.model_extra or {}),
)
def parse_systemone_response(payload: object, request: DecisionsIRRequest) -> DecisionsIRResponse:
return systemone_response_to_ir(_SYSTEMONE_RESPONSE_ADAPTER.validate_python(payload), request)
def _systemone_answer(answer: DecisionsIRAnswer) -> DecisionAnswer | None:
match answer:
case DecisionsIRPredicateAnswer():
return NoulAnswer.model_validate({**answer.extra, "type": "noul", "noul": answer.probability})
case DecisionsIRChoiceAnswer():
return ChoiceAnswer.model_validate(
{
**answer.extra,
"type": "choice",
"choice": systemone_choice_key(answer.choice),
"confidence": answer.confidence,
"probabilities": {
systemone_choice_key(item.value): item.probability for item in answer.probabilities
},
}
)
case DecisionsIRScoreAnswer():
return ScoreAnswer.model_validate(
{
**answer.extra,
"type": "score",
"score": answer.score,
"confidence": answer.confidence,
"legend": {str(item.value): item.label for item in answer.probabilities},
"probabilities": {str(item.value): item.probability for item in answer.probabilities},
}
)
case DecisionsIRRefusal():
return None
case _:
assert_never(answer)
def ir_to_systemone_response(response: DecisionsIRResponse, request: DecisionsIRRequest) -> DecisionsResponse:
keyed_answers: Final = zip(
systemone_keys(request.questions), (_systemone_answer(answer) for answer in response.answers), strict=True
)
return DecisionsResponse.model_validate(
{
**response.extra,
"model": response.model,
"answers": {key: answer for key, answer in keyed_answers if answer is not None},
"usage": DecisionsUsage.model_validate(
{
**response.usage.extra,
"input_tokens": response.usage.input_tokens,
"output_tokens": response.usage.output_tokens,
"cached_tokens": response.usage.cached_tokens,
"cache_write_tokens": response.usage.cache_write_tokens,
}
),
}
)
class BaseDecisionsConfig(ABC):
@ -55,48 +337,32 @@ class BaseDecisionsConfig(ABC):
def transform_decisions_request(
self,
model: str,
request: DecisionsRequest,
request: DecisionsIRRequest,
custom_llm_provider: str,
) -> dict[str, object]:
return {
"model": self.request_model(model),
"state": request.state,
"questions": {
name: question.model_dump(mode="json", exclude_none=True)
for name, question in request.questions.items()
},
}
) -> Mapping[str, object] | UnsupportedDecisionsRequest:
return ir_to_systemone_request(self.request_model(model), request)
def unwrap_response(self, payload: object) -> object:
return payload
def parse_response(self, payload: object, request: DecisionsIRRequest) -> DecisionsIRResponse:
return parse_systemone_response(self.unwrap_response(payload), request)
def transform_decisions_response(
self,
model: str,
custom_llm_provider: str,
raw_response: httpx.Response,
request: DecisionsRequest,
) -> DecisionsResponse:
payload: Final[object] = PAYLOAD_ADAPTER.validate_json(raw_response.content)
request: DecisionsIRRequest,
) -> DecisionsIRResponse:
payload: Final[object] = _PAYLOAD_ADAPTER.validate_json(raw_response.content)
try:
response: Final = _RESPONSE_ADAPTER.validate_python(self.unwrap_response(payload))
return self.parse_response(payload, request)
except ValidationError as error:
raise BaseLLMException(
status_code=500,
message=f"Decisions provider '{custom_llm_provider}' returned an unexpected response: {error}",
) from error
self.set_hidden_params(response, model, custom_llm_provider)
return response
@staticmethod
def set_hidden_params(response: DecisionsResponse, model: str, custom_llm_provider: str) -> None:
response.set_hidden_params(
{
"model": f"{custom_llm_provider}/{model}",
"custom_llm_provider": custom_llm_provider,
"provider_response_model": f"{custom_llm_provider}/{model}",
}
)
def get_error_class(
self,

25
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -123,7 +123,7 @@ from litellm.types.containers.main import (
ContainerObject,
DeleteContainerResult,
)
from litellm.types.decisions import DecisionsRequest, DecisionsResponse
from litellm.types.decisions import DecisionsIRRequest, DecisionsIRResponse
from litellm.types.files import StreamingMediaUploadConfig, TwoStepFileUploadConfig
from litellm.types.integrations.custom_logger import (
NON_CODE_INTERPRETER_INTERCEPTION_INTERNAL_PREFIXES,
@ -1486,19 +1486,16 @@ class BaseLLMHTTPHandler:
def _prepare_decisions_request(
self,
model: str,
custom_llm_provider: str,
logging_obj: LiteLLMLoggingObj | None,
provider_config: BaseDecisionsConfig,
request: DecisionsRequest,
body: Mapping[str, object],
api_base: str,
api_key: str | None,
headers: Mapping[str, str],
) -> tuple[str, dict[str, str], dict[str, object]]:
outbound_headers: Final = provider_config.validate_environment(headers=headers, model=model, api_key=api_key)
url: Final = provider_config.get_complete_url(api_base=api_base, model=model)
data: Final = provider_config.transform_decisions_request(
model=model, request=request, custom_llm_provider=custom_llm_provider
)
data: Final = dict(body)
if logging_obj is not None:
logging_obj.pre_call(
input=data,
@ -1514,19 +1511,19 @@ class BaseLLMHTTPHandler:
custom_llm_provider: str,
logging_obj: LiteLLMLoggingObj | None,
provider_config: BaseDecisionsConfig,
request: DecisionsRequest,
request: DecisionsIRRequest,
body: Mapping[str, object],
api_base: str,
api_key: str | None,
headers: Mapping[str, str],
timeout: float | httpx.Timeout | None,
client: HTTPHandler | None = None,
) -> DecisionsResponse:
) -> DecisionsIRResponse:
url, outbound_headers, data = self._prepare_decisions_request(
model=model,
custom_llm_provider=custom_llm_provider,
logging_obj=logging_obj,
provider_config=provider_config,
request=request,
body=body,
api_base=api_base,
api_key=api_key,
headers=headers,
@ -1548,19 +1545,19 @@ class BaseLLMHTTPHandler:
custom_llm_provider: str,
logging_obj: LiteLLMLoggingObj | None,
provider_config: BaseDecisionsConfig,
request: DecisionsRequest,
request: DecisionsIRRequest,
body: Mapping[str, object],
api_base: str,
api_key: str | None,
headers: Mapping[str, str],
timeout: float | httpx.Timeout | None,
client: AsyncHTTPHandler | None = None,
) -> DecisionsResponse:
) -> DecisionsIRResponse:
url, outbound_headers, data = self._prepare_decisions_request(
model=model,
custom_llm_provider=custom_llm_provider,
logging_obj=logging_obj,
provider_config=provider_config,
request=request,
body=body,
api_base=api_base,
api_key=api_key,
headers=headers,

320
litellm/llms/openai/decisions/transformation.py Normal file
View file

@ -0,0 +1,320 @@
from collections.abc import Mapping, Sequence
from typing import Final
from pydantic import TypeAdapter
from typing_extensions import assert_never
import litellm
from litellm.llms.base_llm.decisions.transformation import BaseDecisionsConfig, decisions_text
from litellm.secret_managers.main import get_secret_str
from litellm.types.decisions import (
DecisionsIRAnswer,
DecisionsIRChoiceAnswer,
DecisionsIRChoiceOption,
DecisionsIRChoiceProbability,
DecisionsIRChoiceQuestion,
DecisionsIRMessages,
DecisionsIRPredicateAnswer,
DecisionsIRPredicateQuestion,
DecisionsIRQuestion,
DecisionsIRRefusal,
DecisionsIRRequest,
DecisionsIRResponse,
DecisionsIRScoreAnswer,
DecisionsIRScoreLevel,
DecisionsIRScoreProbability,
DecisionsIRScoreQuestion,
DecisionsIRState,
DecisionsIRUsage,
DecisionsJSON,
OpenAIChoiceAnswer,
OpenAIChoiceProbability,
OpenAIChoiceQuestion,
OpenAIDecisionAnswer,
OpenAIDecisionInput,
OpenAIDecisionInputTokensDetails,
OpenAIDecisionOutputTokensDetails,
OpenAIDecisionQuestion,
OpenAIDecisionRequestBody,
OpenAIDecisionResponse,
OpenAIDecisionUsage,
OpenAIPredicateAnswer,
OpenAIPredicateQuestion,
OpenAIRefusalAnswer,
OpenAIScoreAnswer,
OpenAIScoreProbability,
OpenAIScoreQuestion,
UnsupportedDecisionsRequest,
systemone_choice_key,
)
_OPENAI_RESPONSE_ADAPTER: Final[TypeAdapter[OpenAIDecisionResponse]] = TypeAdapter(OpenAIDecisionResponse)
_SINGLE_OPTION: Final = UnsupportedDecisionsRequest(
reason="OpenAI needs at least 2 choices or levels on every choice or score question"
)
def _ir_input(decision_input: OpenAIDecisionInput) -> DecisionsIRState | DecisionsIRMessages:
if isinstance(decision_input, str):
return DecisionsIRState(state=decision_input)
return DecisionsIRMessages(messages=tuple(decision_input))
def _ir_question(question: OpenAIDecisionQuestion) -> DecisionsIRQuestion:
match question:
case OpenAIPredicateQuestion():
return DecisionsIRPredicateQuestion(name=question.name, instructions=question.instructions)
case OpenAIChoiceQuestion():
return DecisionsIRChoiceQuestion(
name=question.name,
instructions=question.instructions,
choices=tuple(
DecisionsIRChoiceOption(value=option.value, description=option.description)
for option in question.choices
),
)
case OpenAIScoreQuestion():
return DecisionsIRScoreQuestion(
name=question.name,
instructions=question.instructions,
levels=tuple(
DecisionsIRScoreLevel(label=level.label, description=level.description) for level in question.levels
),
)
case _:
assert_never(question)
def openai_request_to_ir(request: OpenAIDecisionRequestBody) -> DecisionsIRRequest:
return DecisionsIRRequest(
input=_ir_input(request.input),
questions=tuple(_ir_question(question) for question in request.questions),
safety_identifier=request.safety_identifier,
)
def _text_field(key: str, value: DecisionsJSON | None) -> Mapping[str, str]:
return {} if value is None else {key: decisions_text(value)}
def _instructions(value: DecisionsJSON | None, default: str) -> str:
return default if value is None else decisions_text(value)
def _predicate_instructions(question: DecisionsIRPredicateQuestion) -> str:
instructions: Final = () if question.instructions is None else (decisions_text(question.instructions),)
criteria: Final = tuple(
f"Answer {answer} when: {decisions_text(rule)}"
for answer, rule in (question.criteria or {}).items()
if rule is not None
)
return "\n\n".join((*instructions, *criteria))
def _openai_question(question: DecisionsIRQuestion) -> Mapping[str, object]:
match question:
case DecisionsIRPredicateQuestion():
return {
"type": "predicate",
**_text_field("name", question.name),
"instructions": _predicate_instructions(question) or "Is this true of the input?",
}
case DecisionsIRChoiceQuestion():
return {
"type": "choice",
**_text_field("name", question.name),
"instructions": _instructions(question.instructions, "Which choice best fits the input?"),
"choices": [
{"value": option.value, **_text_field("description", option.description)}
for option in question.choices
],
}
case DecisionsIRScoreQuestion():
return {
"type": "score",
**_text_field("name", question.name),
"instructions": _instructions(question.instructions, "Which level best fits the input?"),
"levels": [
{"label": decisions_text(level.label), **_text_field("description", level.description)}
for level in question.levels
],
}
case _:
assert_never(question)
def _openai_input(decision_input: DecisionsIRState | DecisionsIRMessages) -> str | Sequence[Mapping[str, object]]:
match decision_input:
case DecisionsIRState():
return decisions_text(decision_input.state)
case DecisionsIRMessages():
return [message.model_dump(mode="json", exclude_none=True) for message in decision_input.messages]
case _:
assert_never(decision_input)
def _has_one_option(question: DecisionsIRQuestion) -> bool:
match question:
case DecisionsIRChoiceQuestion():
return len(question.choices) < 2
case DecisionsIRScoreQuestion():
return len(question.levels) < 2
case _:
return False
def ir_to_openai_request(model: str, request: DecisionsIRRequest) -> Mapping[str, object] | UnsupportedDecisionsRequest:
if any(_has_one_option(question) for question in request.questions):
return _SINGLE_OPTION
return {
"model": model,
"input": _openai_input(request.input),
"questions": [_openai_question(question) for question in request.questions],
**_text_field("safety_identifier", request.safety_identifier),
}
def _level_label(question: DecisionsIRScoreQuestion, probability: OpenAIScoreProbability) -> DecisionsJSON:
if 0 <= probability.value < len(question.levels):
return question.levels[probability.value].label
return probability.label
def _ir_answer(question: DecisionsIRQuestion, answer: OpenAIDecisionAnswer | None) -> DecisionsIRAnswer:
match question, answer:
case DecisionsIRPredicateQuestion(), OpenAIPredicateAnswer():
return DecisionsIRPredicateAnswer(probability=answer.probability)
case DecisionsIRChoiceQuestion(), OpenAIChoiceAnswer():
return DecisionsIRChoiceAnswer(
choice=answer.choice,
confidence=answer.confidence,
probabilities=tuple(
DecisionsIRChoiceProbability(value=item.value, probability=item.probability)
for item in answer.probabilities
),
)
case DecisionsIRScoreQuestion(), OpenAIScoreAnswer():
return DecisionsIRScoreAnswer(
score=answer.score,
confidence=answer.confidence,
probabilities=tuple(
DecisionsIRScoreProbability(
value=item.value, label=_level_label(question, item), probability=item.probability
)
for item in answer.probabilities
),
)
case _:
return DecisionsIRRefusal()
def openai_response_to_ir(response: OpenAIDecisionResponse, request: DecisionsIRRequest) -> DecisionsIRResponse:
answers: Final = response.answers
return DecisionsIRResponse(
model=response.model,
answers=tuple(
_ir_answer(question, answers[index] if index < len(answers) else None)
for index, question in enumerate(request.questions)
),
usage=DecisionsIRUsage(
input_tokens=response.usage.input_tokens,
output_tokens=response.usage.output_tokens,
cached_tokens=response.usage.input_tokens_details.cached_tokens,
cache_write_tokens=response.usage.input_tokens_details.cache_write_tokens,
reasoning_tokens=response.usage.output_tokens_details.reasoning_tokens,
),
)
def _openai_choice_answer(question: DecisionsIRChoiceQuestion, answer: DecisionsIRChoiceAnswer) -> OpenAIChoiceAnswer:
probabilities: Final = {systemone_choice_key(item.value): item.probability for item in answer.probabilities}
return OpenAIChoiceAnswer(
name=question.name,
choice=answer.choice,
probabilities=tuple(
OpenAIChoiceProbability(
value=option.value, probability=probabilities.get(systemone_choice_key(option.value), 0.0)
)
for option in question.choices
),
confidence=answer.confidence,
)
def _openai_score_answer(question: DecisionsIRScoreQuestion, answer: DecisionsIRScoreAnswer) -> OpenAIScoreAnswer:
probabilities: Final = {item.value: item.probability for item in answer.probabilities}
return OpenAIScoreAnswer(
name=question.name,
score=answer.score,
probabilities=tuple(
OpenAIScoreProbability(
value=index, label=decisions_text(level.label), probability=probabilities.get(index, 0.0)
)
for index, level in enumerate(question.levels)
),
confidence=answer.confidence,
)
def _openai_answer(question: DecisionsIRQuestion, answer: DecisionsIRAnswer) -> OpenAIDecisionAnswer:
match question, answer:
case DecisionsIRPredicateQuestion(), DecisionsIRPredicateAnswer():
return OpenAIPredicateAnswer(name=question.name, probability=answer.probability)
case DecisionsIRChoiceQuestion(), DecisionsIRChoiceAnswer():
return _openai_choice_answer(question, answer)
case DecisionsIRScoreQuestion(), DecisionsIRScoreAnswer():
return _openai_score_answer(question, answer)
case _:
return OpenAIRefusalAnswer(name=question.name)
def ir_to_openai_response(
response: DecisionsIRResponse, request: DecisionsIRRequest, requested_model: str
) -> OpenAIDecisionResponse:
return OpenAIDecisionResponse(
model=response.model or requested_model,
answers=tuple(
_openai_answer(question, answer)
for question, answer in zip(request.questions, response.answers, strict=True)
),
usage=OpenAIDecisionUsage(
input_tokens=response.usage.input_tokens,
input_tokens_details=OpenAIDecisionInputTokensDetails(
cached_tokens=response.usage.cached_tokens,
cache_write_tokens=response.usage.cache_write_tokens,
),
output_tokens=response.usage.output_tokens,
output_tokens_details=OpenAIDecisionOutputTokensDetails(reasoning_tokens=response.usage.reasoning_tokens),
total_tokens=response.usage.input_tokens + response.usage.output_tokens,
),
)
class OpenAIDecisionsConfig(BaseDecisionsConfig):
path = "/v1/decisions"
def get_default_api_base(self) -> str | None:
return "https://api.openai.com"
def resolve_api_base(self, api_base: str | None) -> str | None:
return (
api_base
or litellm.api_base
or get_secret_str("OPENAI_BASE_URL")
or get_secret_str("OPENAI_API_BASE")
or self.get_default_api_base()
)
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 transform_decisions_request(
self,
model: str,
request: DecisionsIRRequest,
custom_llm_provider: str,
) -> Mapping[str, object] | UnsupportedDecisionsRequest:
return ir_to_openai_request(self.request_model(model), request)
def parse_response(self, payload: object, request: DecisionsIRRequest) -> DecisionsIRResponse:
return openai_response_to_ir(_OPENAI_RESPONSE_ADAPTER.validate_python(payload), request)

3
litellm/model_prices_and_context_window_backup.json
View file

@ -34527,7 +34527,8 @@
"supported_endpoints": [
"/v1/chat/completions",
"/v1/batch",
"/v1/responses"
"/v1/responses",
"/v1/decisions"
],
"supported_modalities": [
"text",

2
litellm/proxy/_lazy_features.py
View file

@ -267,7 +267,7 @@ LAZY_FEATURES: Final[tuple[LazyFeature, ...]] = (
LazyFeature(
name="decisions",
module_path="litellm.proxy.decisions_endpoints.endpoints",
path_prefixes=("/v1/decisions", "/decisions"),
path_prefixes=("/v1/decisions", "/decisions", "/v1/systemone", "/systemone"),
),
LazyFeature(
name="claude_code_marketplace",

48
litellm/proxy/_lazy_openapi_snapshot.json
View file

@ -9521,6 +9521,30 @@
]
}
},
"/systemone": {
"post": {
"operationId": "systemone_systemone_post",
"responses": {
"200": {
"content": {
"application/json": {
"schema": {}
}
},
"description": "Successful Response"
}
},
"security": [
{
"APIKeyHeader": []
}
],
"summary": "Systemone",
"tags": [
"decisions"
]
}
},
"/v1/decisions": {
"post": {
"operationId": "decisions_v1_decisions_post",
@ -9544,6 +9568,30 @@
"decisions"
]
}
},
"/v1/systemone": {
"post": {
"operationId": "systemone_v1_systemone_post",
"responses": {
"200": {
"content": {
"application/json": {
"schema": {}
}
},
"description": "Successful Response"
}
},
"security": [
{
"APIKeyHeader": []
}
],
"summary": "Systemone",
"tags": [
"decisions"
]
}
}
}
},

2
litellm/proxy/_types.py
View file

@ -488,6 +488,8 @@ class LiteLLMRoutes(enum.Enum):
"/v1/search/{search_tool_name}",
"/decisions",
"/v1/decisions",
"/systemone",
"/v1/systemone",
# OCR
"/ocr",
"/v1/ocr",

2
litellm/proxy/agent_endpoints/auth/managed_authorization.py
View file

@ -30,6 +30,7 @@ _MANAGED_MODEL_ROUTES: Final = frozenset(
"moderations",
"rerank",
"decisions",
"systemone",
"ocr",
),
)
@ -74,6 +75,7 @@ _MODEL_ROUTE_KINDS: Final[
"/audio/speech": "speech",
"/rerank": "body",
"/decisions": "body",
"/systemone": "body",
"/messages/count_tokens": "body",
":countTokens": "path",
}

11
litellm/proxy/auth/model_checks.py
View file

@ -13,7 +13,7 @@ from litellm.router import Router
from litellm.router_utils.fallback_event_handlers import get_fallback_model_group
from litellm.types.router import CredentialLiteLLMParams, LiteLLM_Params
from litellm.types.utils import LlmProviders
from litellm.utils import get_valid_models
from litellm.utils import ProviderConfigManager, get_valid_models
_CREDENTIAL_LITELLM_PARAM_FIELDS = set(CredentialLiteLLMParams.model_fields)
@ -45,7 +45,14 @@ def get_provider_models(provider: str, litellm_params: LiteLLM_Params | None = N
if provider in litellm.models_by_provider:
provider_models: Final = get_valid_models(custom_llm_provider=provider, litellm_params=litellm_params)
return provider_models
return None
try:
llm_provider: Final = LlmProviders(provider)
except ValueError:
return None
if ProviderConfigManager.get_provider_model_info(model=None, provider=llm_provider) is None:
return None
return get_valid_models(custom_llm_provider=provider, litellm_params=litellm_params)
def _get_models_from_access_groups(

119
litellm/proxy/decisions_endpoints/endpoints.py
View file

@ -1,3 +1,4 @@
from collections.abc import Mapping
from typing import Annotated, Final
from fastapi import APIRouter, Depends, Request, Response
@ -7,33 +8,50 @@ from pydantic import TypeAdapter, ValidationError
from litellm.exceptions import BadRequestError
from litellm.proxy.auth.user_api_key_auth import UserAPIKeyAuth, user_api_key_auth
from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
from litellm.types.decisions import DecisionsRequestBody
from litellm.types.decisions import DecisionsRequestBody, OpenAIDecisionRequestBody
router: Final = APIRouter()
_REQUEST_DATA_ADAPTER: Final[TypeAdapter[dict[str, object]]] = TypeAdapter(dict[str, object])
_DECISIONS_REQUEST_BODY_ADAPTER: Final[TypeAdapter[DecisionsRequestBody]] = TypeAdapter(DecisionsRequestBody)
_OPENAI_DECISION_REQUEST_BODY_ADAPTER: Final[TypeAdapter[OpenAIDecisionRequestBody]] = TypeAdapter(
OpenAIDecisionRequestBody
)
_GENERAL_SETTINGS_ADAPTER: Final[TypeAdapter[dict[str, object]]] = TypeAdapter(dict[str, object])
_OPTIONAL_STRING_ADAPTER: Final[TypeAdapter[str | None]] = TypeAdapter(str | None)
_OPTIONAL_FLOAT_ADAPTER: Final[TypeAdapter[float | None]] = TypeAdapter(float | None)
@router.post(
"/v1/decisions",
dependencies=[Depends(user_api_key_auth)],
response_class=ORJSONResponse, # pyright: ignore[reportDeprecated] # required endpoint contract
tags=["decisions"],
)
@router.post(
"/decisions",
dependencies=[Depends(user_api_key_auth)],
response_class=ORJSONResponse, # pyright: ignore[reportDeprecated] # required endpoint contract
tags=["decisions"],
)
async def decisions(
async def _invalid_request(
raw_data: Mapping[str, object], error: ValidationError, user_api_key_dict: UserAPIKeyAuth
) -> Exception:
from litellm.proxy.proxy_server import proxy_logging_obj, version
return await ProxyBaseLLMRequestProcessing(data=dict(raw_data)).handle_llm_api_exception(
e=BadRequestError(
message=f"Invalid Decisions request: {error}",
model=str(raw_data.get("model", "")),
llm_provider="",
),
user_api_key_dict=user_api_key_dict,
proxy_logging_obj=proxy_logging_obj,
version=version,
)
async def _request_data(request: Request, user_api_key_dict: UserAPIKeyAuth) -> dict[str, object]:
body: Final = await request.body()
try:
return _REQUEST_DATA_ADAPTER.validate_json(body)
except ValidationError as error:
raise await _invalid_request(raw_data={}, error=error, user_api_key_dict=user_api_key_dict)
async def _process_decisions(
request: Request,
fastapi_response: Response,
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
):
user_api_key_dict: UserAPIKeyAuth,
body_adapter: TypeAdapter[DecisionsRequestBody] | TypeAdapter[OpenAIDecisionRequestBody],
) -> object:
from litellm.proxy.proxy_server import (
general_settings as proxy_general_settings,
)
@ -55,14 +73,17 @@ async def decisions(
user_temperature as proxy_user_temperature,
)
data: Final = _REQUEST_DATA_ADAPTER.validate_json(await request.body())
data: Final = await _request_data(request, user_api_key_dict)
try:
body_adapter.validate_python(data)
except ValidationError as error:
raise await _invalid_request(raw_data=data, error=error, user_api_key_dict=user_api_key_dict)
general_settings: Final = _GENERAL_SETTINGS_ADAPTER.validate_python(proxy_general_settings)
user_api_base: Final = _OPTIONAL_STRING_ADAPTER.validate_python(proxy_user_api_base)
user_model: Final = _OPTIONAL_STRING_ADAPTER.validate_python(proxy_user_model)
user_temperature: Final = _OPTIONAL_FLOAT_ADAPTER.validate_python(proxy_user_temperature)
processor: Final = ProxyBaseLLMRequestProcessing(data=data)
try:
_DECISIONS_REQUEST_BODY_ADAPTER.validate_python(data)
return await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
@ -81,18 +102,6 @@ async def decisions(
user_api_base=user_api_base,
version=version,
)
except ValidationError as error:
bad_request_error: Final = BadRequestError(
message=f"Invalid Decisions request: {error}",
model=str(data.get("model", "")),
llm_provider="",
)
raise await processor.handle_llm_api_exception(
e=bad_request_error,
user_api_key_dict=user_api_key_dict,
proxy_logging_obj=proxy_logging_obj,
version=version,
)
except Exception as error:
raise await processor.handle_llm_api_exception(
e=error,
@ -100,3 +109,53 @@ async def decisions(
proxy_logging_obj=proxy_logging_obj,
version=version,
)
@router.post(
"/v1/systemone",
dependencies=[Depends(user_api_key_auth)],
response_class=ORJSONResponse, # pyright: ignore[reportDeprecated] # required endpoint contract
tags=["decisions"],
)
@router.post(
"/systemone",
dependencies=[Depends(user_api_key_auth)],
response_class=ORJSONResponse, # pyright: ignore[reportDeprecated] # required endpoint contract
tags=["decisions"],
)
async def systemone(
request: Request,
fastapi_response: Response,
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
):
return await _process_decisions(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
body_adapter=_DECISIONS_REQUEST_BODY_ADAPTER,
)
@router.post(
"/v1/decisions",
dependencies=[Depends(user_api_key_auth)],
response_class=ORJSONResponse, # pyright: ignore[reportDeprecated] # required endpoint contract
tags=["decisions"],
)
@router.post(
"/decisions",
dependencies=[Depends(user_api_key_auth)],
response_class=ORJSONResponse, # pyright: ignore[reportDeprecated] # required endpoint contract
tags=["decisions"],
)
async def decisions(
request: Request,
fastapi_response: Response,
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
):
return await _process_decisions(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
body_adapter=_OPENAI_DECISION_REQUEST_BODY_ADAPTER,
)

338
litellm/types/decisions.py
View file

@ -1,5 +1,7 @@
from collections.abc import Mapping, Sequence
from typing import Annotated, Literal, TypeAlias
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 typing_extensions import ReadOnly, Required, TypedDict
@ -8,6 +10,7 @@ from litellm.types.llms.base import LiteLLMPydanticObjectBase
DecisionsJSON: TypeAlias = str | Mapping[str, object] | Sequence[object]
NoulCriteria: TypeAlias = Mapping[Literal["true", "false"], DecisionsJSON | None]
MAX_DECISION_QUESTIONS: Final = 128
class NoulQuestion(LiteLLMPydanticObjectBase):
@ -47,7 +50,7 @@ DecisionQuestion: TypeAlias = Annotated[
DecisionQuestionMap: TypeAlias = Annotated[
Mapping[Annotated[str, Field(min_length=1)], DecisionQuestion],
Field(min_length=1, max_length=128),
Field(min_length=1, max_length=MAX_DECISION_QUESTIONS),
]
@ -109,17 +112,13 @@ DecisionAnswer: TypeAlias = Annotated[
class DecisionsUsage(LiteLLMPydanticObjectBase):
input_tokens: int = 0
output_tokens: int = 0
cached_tokens: Annotated[int, Field(exclude=True)] = 0
cache_write_tokens: Annotated[int, Field(exclude=True)] = 0
model_config = ConfigDict(extra="allow", frozen=True)
class DecisionsResponse(LiteLLMPydanticObjectBase):
model: str | None = None
answers: Mapping[str, DecisionAnswer]
usage: DecisionsUsage | None = None
model_config = ConfigDict(extra="allow", frozen=True)
class _HiddenParamsResponse(LiteLLMPydanticObjectBase):
_hidden_params: dict[str, object] = PrivateAttr(default_factory=dict)
@property
@ -128,3 +127,324 @@ class DecisionsResponse(LiteLLMPydanticObjectBase):
def set_hidden_params(self, params: Mapping[str, object]) -> None:
self._hidden_params.update(params)
class DecisionsResponse(_HiddenParamsResponse):
model: str | None = None
answers: Mapping[str, DecisionAnswer]
usage: DecisionsUsage | None = None
model_config = ConfigDict(extra="allow", frozen=True)
class OpenAIDecisionInputText(LiteLLMPydanticObjectBase):
type: Literal["input_text"]
text: str
model_config = ConfigDict(extra="forbid", frozen=True)
class OpenAIDecisionInputImage(LiteLLMPydanticObjectBase):
type: Literal["input_image"]
image_url: str
detail: str | None = None
model_config = ConfigDict(extra="forbid", frozen=True)
OpenAIDecisionContentPart: TypeAlias = Annotated[
OpenAIDecisionInputText | OpenAIDecisionInputImage,
Field(discriminator="type"),
]
class OpenAIDecisionInputMessage(LiteLLMPydanticObjectBase):
role: Literal["user"] = "user"
type: Literal["message"] = "message"
content: str | Sequence[OpenAIDecisionContentPart]
model_config = ConfigDict(extra="forbid", frozen=True)
OpenAIDecisionInput: TypeAlias = str | Sequence[OpenAIDecisionInputMessage]
class OpenAIPredicateQuestion(LiteLLMPydanticObjectBase):
type: Literal["predicate"]
name: str | None = None
instructions: str
model_config = ConfigDict(extra="forbid", frozen=True)
class OpenAIChoiceOption(LiteLLMPydanticObjectBase):
value: str | bool
description: str | None = None
model_config = ConfigDict(extra="forbid", frozen=True)
def systemone_choice_key(value: str | bool) -> str:
if isinstance(value, bool):
return "true" if value else "false"
return value
class OpenAIChoiceQuestion(LiteLLMPydanticObjectBase):
type: Literal["choice"]
name: str | None = None
instructions: str
choices: Annotated[Sequence[OpenAIChoiceOption], Field(min_length=2, max_length=255)]
model_config = ConfigDict(extra="forbid", frozen=True)
@model_validator(mode="after")
def require_unique_systemone_keys(self) -> "OpenAIChoiceQuestion":
keys: Final = frozenset(systemone_choice_key(option.value) for option in self.choices)
if len(keys) != len(self.choices):
raise ValueError("Choice values must be unique, and a boolean cannot share its text with a string choice")
return self
class OpenAIScoreLevel(LiteLLMPydanticObjectBase):
label: str
description: str | None = None
model_config = ConfigDict(extra="forbid", frozen=True)
class OpenAIScoreQuestion(LiteLLMPydanticObjectBase):
type: Literal["score"]
name: str | None = None
instructions: str
levels: Annotated[Sequence[OpenAIScoreLevel], Field(min_length=2, max_length=10)]
model_config = ConfigDict(extra="forbid", frozen=True)
OpenAIDecisionQuestion: TypeAlias = Annotated[
OpenAIPredicateQuestion | OpenAIChoiceQuestion | OpenAIScoreQuestion,
Field(discriminator="type"),
]
class OpenAIDecisionRequestBody(LiteLLMPydanticObjectBase):
input: OpenAIDecisionInput
questions: Annotated[Sequence[OpenAIDecisionQuestion], Field(min_length=1, max_length=MAX_DECISION_QUESTIONS)]
safety_identifier: str | None = None
model_config = ConfigDict(extra="allow", frozen=True)
class OpenAIPredicateAnswer(LiteLLMPydanticObjectBase):
type: Literal["predicate"] = "predicate"
name: str | None
probability: float
model_config = ConfigDict(frozen=True)
class OpenAIChoiceProbability(LiteLLMPydanticObjectBase):
value: str | bool
probability: float
model_config = ConfigDict(frozen=True)
class OpenAIChoiceAnswer(LiteLLMPydanticObjectBase):
type: Literal["choice"] = "choice"
name: str | None
choice: str | bool
probabilities: tuple[OpenAIChoiceProbability, ...]
confidence: float
model_config = ConfigDict(frozen=True)
class OpenAIScoreProbability(LiteLLMPydanticObjectBase):
value: int
label: str
probability: float
model_config = ConfigDict(frozen=True)
class OpenAIScoreAnswer(LiteLLMPydanticObjectBase):
type: Literal["score"] = "score"
name: str | None
score: float
probabilities: tuple[OpenAIScoreProbability, ...]
confidence: float
model_config = ConfigDict(frozen=True)
class OpenAIRefusalAnswer(LiteLLMPydanticObjectBase):
type: Literal["refusal"] = "refusal"
name: str | None
model_config = ConfigDict(frozen=True)
OpenAIDecisionAnswer: TypeAlias = Annotated[
OpenAIPredicateAnswer | OpenAIChoiceAnswer | OpenAIScoreAnswer | OpenAIRefusalAnswer,
Field(discriminator="type"),
]
class OpenAIDecisionInputTokensDetails(LiteLLMPydanticObjectBase):
cached_tokens: int = 0
cache_write_tokens: int = 0
model_config = ConfigDict(frozen=True)
class OpenAIDecisionOutputTokensDetails(LiteLLMPydanticObjectBase):
reasoning_tokens: int = 0
model_config = ConfigDict(frozen=True)
class OpenAIDecisionUsage(LiteLLMPydanticObjectBase):
input_tokens: int
input_tokens_details: OpenAIDecisionInputTokensDetails = OpenAIDecisionInputTokensDetails()
output_tokens: int
output_tokens_details: OpenAIDecisionOutputTokensDetails = OpenAIDecisionOutputTokensDetails()
total_tokens: int
model_config = ConfigDict(frozen=True)
class OpenAIDecisionResponse(_HiddenParamsResponse):
model: str
answers: tuple[OpenAIDecisionAnswer, ...]
usage: OpenAIDecisionUsage
model_config = ConfigDict(extra="allow", frozen=True)
_NO_EXTRA: Final[Mapping[str, object]] = MappingProxyType({})
@dataclass(frozen=True, slots=True)
class DecisionsIRState:
state: DecisionsJSON
@dataclass(frozen=True, slots=True)
class DecisionsIRMessages:
messages: tuple[OpenAIDecisionInputMessage, ...]
@dataclass(frozen=True, slots=True)
class DecisionsIRPredicateQuestion:
name: str | None
instructions: DecisionsJSON | None
criteria: NoulCriteria | None = None
extra: Mapping[str, object] = field(default_factory=lambda: _NO_EXTRA)
@dataclass(frozen=True, slots=True)
class DecisionsIRChoiceOption:
value: str | bool
description: DecisionsJSON | None
@dataclass(frozen=True, slots=True)
class DecisionsIRChoiceQuestion:
name: str | None
instructions: DecisionsJSON | None
choices: tuple[DecisionsIRChoiceOption, ...]
extra: Mapping[str, object] = field(default_factory=lambda: _NO_EXTRA)
@dataclass(frozen=True, slots=True)
class DecisionsIRScoreLevel:
label: DecisionsJSON
description: str | None
@dataclass(frozen=True, slots=True)
class DecisionsIRScoreQuestion:
name: str | None
instructions: DecisionsJSON | None
levels: tuple[DecisionsIRScoreLevel, ...]
extra: Mapping[str, object] = field(default_factory=lambda: _NO_EXTRA)
DecisionsIRQuestion: TypeAlias = DecisionsIRPredicateQuestion | DecisionsIRChoiceQuestion | DecisionsIRScoreQuestion
@dataclass(frozen=True, slots=True)
class DecisionsIRRequest:
input: DecisionsIRState | DecisionsIRMessages
questions: tuple[DecisionsIRQuestion, ...]
safety_identifier: str | None = None
@dataclass(frozen=True, slots=True)
class DecisionsIRPredicateAnswer:
probability: float
extra: Mapping[str, object] = field(default_factory=lambda: _NO_EXTRA)
@dataclass(frozen=True, slots=True)
class DecisionsIRChoiceProbability:
value: str | bool
probability: float
@dataclass(frozen=True, slots=True)
class DecisionsIRChoiceAnswer:
choice: str | bool
confidence: float
probabilities: tuple[DecisionsIRChoiceProbability, ...]
extra: Mapping[str, object] = field(default_factory=lambda: _NO_EXTRA)
@dataclass(frozen=True, slots=True)
class DecisionsIRScoreProbability:
value: int
label: DecisionsJSON
probability: float
@dataclass(frozen=True, slots=True)
class DecisionsIRScoreAnswer:
score: float
confidence: float
probabilities: tuple[DecisionsIRScoreProbability, ...]
extra: Mapping[str, object] = field(default_factory=lambda: _NO_EXTRA)
@dataclass(frozen=True, slots=True)
class DecisionsIRRefusal:
pass
DecisionsIRAnswer: TypeAlias = (
DecisionsIRPredicateAnswer | DecisionsIRChoiceAnswer | DecisionsIRScoreAnswer | DecisionsIRRefusal
)
@dataclass(frozen=True, slots=True)
class DecisionsIRUsage:
input_tokens: int = 0
output_tokens: int = 0
cached_tokens: int = 0
cache_write_tokens: int = 0
reasoning_tokens: int = 0
extra: Mapping[str, object] = field(default_factory=lambda: _NO_EXTRA)
@dataclass(frozen=True, slots=True)
class DecisionsIRResponse:
model: str | None
answers: tuple[DecisionsIRAnswer, ...]
usage: DecisionsIRUsage
extra: Mapping[str, object] = field(default_factory=lambda: _NO_EXTRA)
@dataclass(frozen=True, slots=True)
class UnsupportedDecisionsRequest:
reason: str

2
litellm/types/utils.py
View file

@ -788,6 +788,8 @@ API_ROUTE_TO_CALL_TYPES: Final[Mapping[str, Sequence[CallTypes]]] = {
"/v1/search": [CallTypes.asearch, CallTypes.search],
"/decisions": [CallTypes.adecisions, CallTypes.decisions],
"/v1/decisions": [CallTypes.adecisions, CallTypes.decisions],
"/systemone": [CallTypes.adecisions, CallTypes.decisions],
"/v1/systemone": [CallTypes.adecisions, CallTypes.decisions],
# Batches
"/batches": [CallTypes.acreate_batch, CallTypes.create_batch],
"/v1/batches": [CallTypes.acreate_batch, CallTypes.create_batch],

6
litellm/utils.py
View file

@ -1219,8 +1219,8 @@ def function_setup(
else search_query
)
elif call_type in (CallTypes.decisions.value, CallTypes.adecisions.value):
decisions_state: Final = args[1] if len(args) > 1 else kwargs.get("state", "")
messages = decisions_state if isinstance(decisions_state, str) else json.dumps(decisions_state)
decisions_state: Final = args[1] if len(args) > 1 else kwargs.get("state") or kwargs.get("input") or ""
messages = decisions_state if isinstance(decisions_state, str) else json.dumps(decisions_state, default=str)
elif call_type in (CallTypes.image_edit.value, CallTypes.aimage_edit.value):
messages = args[1] if len(args) > 1 else kwargs.get("prompt")
elif call_type in (CallTypes.ocr.value, CallTypes.aocr.value):
@ -8974,6 +8974,8 @@ class ProviderConfigManager:
return litellm.CloudflareDecisionsConfig()
if provider == LlmProviders.STRANDS_DECIDER:
return litellm.StrandsDeciderDecisionsConfig()
if provider == LlmProviders.OPENAI:
return litellm.OpenAIDecisionsConfig()
return None
@staticmethod

3
model_prices_and_context_window.json
View file

@ -34527,7 +34527,8 @@
"supported_endpoints": [
"/v1/chat/completions",
"/v1/batch",
"/v1/responses"
"/v1/responses",
"/v1/decisions"
],
"supported_modalities": [
"text",

185
scripts/check_type_discipline.py
View file

@ -18,8 +18,8 @@ LIT003 noqa suppression without rule codes or without a reason.
LIT004 pyright/mypy ignore without bracketed codes or without a reason.
Required shape: `# pyright: ignore[reportArgumentType] # <reason>`
LIT005 A `# mutable-ok` / `# cast-ok` / `# guard-ok` / `# kwargs-ok` /
`# rebind-ok` / `# writable-ok` / `# comprehension-ok` suppression
without a reason.
`# rebind-ok` / `# writable-ok` / `# comprehension-ok` / `# frozen-ok`
suppression without a reason.
LIT006 `cast(...)` call. typing.cast is an unchecked assertion (the moral equivalent
of TypeScript's `as`); it lies to the type checker with zero runtime guarantee.
Validate into a concrete frozen type at the boundary instead.
@ -91,6 +91,16 @@ LIT014 Comprehension with more than one `for` clause or more than one `if` clau
`# comprehension-ok: <reason>` on any line the comprehension spans. The
marker belongs to the innermost violating comprehension spanning that
line, and also to any single-line violating comprehension on that line.
LIT015 Pydantic model class that is not frozen. Set `frozen=True` in
`model_config = ConfigDict(...)`, `SettingsConfigDict(...)`, a dict-literal
`model_config`, an inner `class Config`, or the class keywords. Classes
inherit the frozen setting from in-module model bases, unless their own
configuration overrides it. Detection is name-based: `BaseModel`,
`pydantic.BaseModel`, `LiteLLMBaseModel`, `BaseLiteLLMOpenAIResponseObject`,
`LiteLLMPydanticObjectBase`, `OpenAIObject`, `RootModel`, and `BaseSettings` identify
models, while `TypedDict` classes are exempt. Replace in-place field writes
with `model_copy(update=...)`. Suppress with `# frozen-ok: <reason>` on the
`class` line.
LIT000 Setup failure: a target file could not be read, or contains a syntax error.
Reported as a violation rather than crashing the run.
@ -110,12 +120,13 @@ import os
import re
import sys
import tokenize
from collections.abc import Iterable, Iterator, Mapping, Sequence
from dataclasses import dataclass
from itertools import groupby
from multiprocessing import Pool
from pathlib import Path
from collections.abc import Iterable, Iterator, Mapping, Sequence
from types import MappingProxyType
from typing import NamedTuple
from typing import Final, NamedTuple
# Mutable collection types, banned in *every* annotation. Name-based, so `dict`,
# `typing.Dict`, `collections.deque`, and `collections.abc.MutableMapping` all match
@ -149,6 +160,19 @@ READONLY_QUALIFIER = "ReadOnly"
# first argument is type syntax, the rest is metadata and never qualifies the field.
FIELD_QUALIFIER_WRAPPERS = frozenset(("Required", "NotRequired", "Annotated"))
TYPEDDICT_BASE = "TypedDict"
# Base names that mark a class as a pydantic model (LIT015).
PYDANTIC_BASES: Final = frozenset(
(
"BaseModel",
"LiteLLMBaseModel",
"BaseLiteLLMOpenAIResponseObject",
"LiteLLMPydanticObjectBase",
"OpenAIObject",
"RootModel",
"BaseSettings",
)
)
PYDANTIC_CONFIG_FACTORIES: Final = frozenset(("ConfigDict", "SettingsConfigDict"))
MIN_REASON_LEN = 3
NOQA_RE = re.compile(
@ -166,6 +190,8 @@ KWARGS_OK_RE = re.compile(r"#\s*kwargs-ok(?::\s*(?P<reason>.*))?")
REBIND_OK_RE = re.compile(r"#\s*rebind-ok(?::\s*(?P<reason>.*))?")
WRITABLE_OK_RE = re.compile(r"#\s*writable-ok(?::\s*(?P<reason>.*))?")
COMPREHENSION_OK_RE = re.compile(r"#\s*comprehension-ok(?::\s*(?P<reason>.*))?")
FROZEN_OK_RE: Final = re.compile(r"#\s*frozen-ok(?::\s*(?P<reason>.*))?")
@dataclass(frozen=True, slots=True)
class _OkToken:
@ -185,6 +211,7 @@ OK_SUPPRESSIONS: Final[tuple[_OkToken, ...]] = (
_OkToken("rebind-ok", REBIND_OK_RE, frozenset(("LIT010", "LIT011"))),
_OkToken("writable-ok", WRITABLE_OK_RE, frozenset(("LIT012",))),
_OkToken("comprehension-ok", COMPREHENSION_OK_RE, frozenset(("LIT014",))),
_OkToken("frozen-ok", FROZEN_OK_RE, frozenset(("LIT015",))),
)
@ -833,6 +860,147 @@ def iter_typeddict_violations(path: Path, tree: ast.AST) -> Iterator[Violation]:
)
# --------------------------------------------------------------------------- #
# Unfrozen pydantic models (LIT015)
# --------------------------------------------------------------------------- #
def _pydantic_classes(tree: ast.AST) -> tuple[ast.ClassDef, ...]:
classes: Final = tuple(node for node in ast.walk(tree) if isinstance(node, ast.ClassDef))
def expand(known: frozenset[str]) -> frozenset[str]:
grown: Final = known | frozenset(cls.name for cls in classes if _base_names(cls) & known)
return grown if grown == known else expand(grown)
model_names: Final = expand(PYDANTIC_BASES)
typeddict_names: Final = expand(frozenset((TYPEDDICT_BASE,)))
return tuple(cls for cls in classes if _base_names(cls) & model_names and not _base_names(cls) & typeddict_names)
def _bool_constant(value: ast.expr) -> bool | None:
return value.value if isinstance(value, ast.Constant) and isinstance(value.value, bool) else None
def _module_assignment_items(tree: ast.AST) -> Iterator[tuple[str, tuple[int, ast.expr]]]:
if not isinstance(tree, ast.Module):
return
for stmt in tree.body:
if isinstance(stmt, ast.Assign):
yield from (
(target.id, (stmt.lineno, stmt.value)) for target in stmt.targets if isinstance(target, ast.Name)
)
elif isinstance(stmt, ast.AnnAssign) and isinstance(stmt.target, ast.Name) and stmt.value is not None:
yield stmt.target.id, (stmt.lineno, stmt.value)
def _model_config_frozen(
value: ast.expr,
module_assignments: Mapping[str, tuple[tuple[int, ast.expr], ...]],
class_lineno: int,
) -> bool | None:
config: Final = (
next(
(
expression
for lineno, expression in reversed(module_assignments.get(value.id, ()))
if lineno < class_lineno
),
None,
)
if isinstance(value, ast.Name)
else value
)
if isinstance(config, ast.Call) and _head_name(config.func) in PYDANTIC_CONFIG_FACTORIES:
flags: Final = tuple(_bool_constant(kw.value) for kw in config.keywords if kw.arg == "frozen")
return flags[-1] if flags else None
if isinstance(config, ast.Dict):
flags: Final = tuple(
_bool_constant(item)
for key, item in zip(config.keys, config.values)
if isinstance(key, ast.Constant) and key.value == "frozen"
)
return flags[-1] if flags else None
return None
def _assigns_name(stmt: ast.stmt, name: str) -> ast.expr | None:
value: Final = stmt.value if isinstance(stmt, (ast.Assign, ast.AnnAssign)) else None
targets: Final = (
stmt.targets if isinstance(stmt, ast.Assign) else (stmt.target,) if isinstance(stmt, ast.AnnAssign) else ()
)
if value is not None and any(isinstance(target, ast.Name) and target.id == name for target in targets):
return value
return None
def _config_class_frozen(node: ast.ClassDef) -> bool | None:
values: Final = tuple(_assigns_name(stmt, "frozen") for stmt in node.body)
flags: Final = tuple(_bool_constant(value) for value in values if value is not None)
return flags[-1] if flags else None
def _stmt_frozen_flag(
stmt: ast.stmt,
module_assignments: Mapping[str, tuple[tuple[int, ast.expr], ...]],
class_lineno: int,
) -> bool | None:
config_value: Final = _assigns_name(stmt, "model_config")
if config_value is not None:
return _model_config_frozen(config_value, module_assignments, class_lineno)
if isinstance(stmt, ast.ClassDef) and stmt.name == "Config":
return _config_class_frozen(stmt)
return None
def _class_frozen_override(
cls: ast.ClassDef,
module_assignments: Mapping[str, tuple[tuple[int, ast.expr], ...]],
class_lineno: int,
) -> bool | None:
keyword_flags: Final = tuple(_bool_constant(keyword.value) for keyword in cls.keywords if keyword.arg == "frozen")
if keyword_flags:
return keyword_flags[-1]
body_flags: Final = tuple(_stmt_frozen_flag(stmt, module_assignments, class_lineno) for stmt in cls.body)
return next((flag for flag in reversed(body_flags) if flag is not None), None)
def iter_pydantic_violations(path: Path, tree: ast.AST) -> Iterator[Violation]:
module_assignments: Final = MappingProxyType(
{
name: tuple(binding for _, binding in assignments)
for name, assignments in groupby(
sorted(_module_assignment_items(tree), key=lambda item: item[0]),
key=lambda item: item[0],
)
}
)
models: Final = _pydantic_classes(tree)
bases_of: Final = {cls: _base_names(cls) for cls in models}
override_of: Final = {cls: _class_frozen_override(cls, module_assignments, cls.lineno) for cls in models}
def frozen(known: frozenset[str]) -> frozenset[str]:
grown: Final = known | frozenset(
cls.name
for cls in models
if override_of[cls] is True or (override_of[cls] is None and bases_of[cls] & known)
)
return grown if grown == known else frozen(grown)
frozen_names: Final = frozen(frozenset())
for cls in models:
if override_of[cls] is True or (override_of[cls] is None and bases_of[cls] & frozen_names):
continue
yield Violation(
path,
cls.lineno,
"LIT015",
f"pydantic model `{cls.name}` is not frozen: set `frozen=True` in "
f"`model_config`, an inner `class Config`, or the class keywords, and "
f"replace in-place field writes with `model_copy(update=...)` "
f"(suppress: `# frozen-ok: <reason>`)",
)
# --------------------------------------------------------------------------- #
# Stacked comprehension clauses (LIT014)
# --------------------------------------------------------------------------- #
@ -978,6 +1146,7 @@ def check_file(path: Path) -> tuple[Violation, ...]:
*iter_final_violations(path, tree),
*iter_param_violations(path, tree),
*iter_typeddict_violations(path, tree),
*iter_pydantic_violations(path, tree),
*(v for v, owned in comprehension_violations if owned),
),
suppressions,
@ -1009,7 +1178,7 @@ def _worker_count(path_count: int) -> int:
def scan_paths(paths: Sequence[Path]) -> tuple[Violation, ...]:
"""check_file over every path. Pure per-file work, so it fans out across
processes; callers sort, which is what keeps output order stable."""
workers = _worker_count(len(paths))
workers: Final = _worker_count(len(paths))
if workers == 1:
return tuple(v for path in paths for v in check_file(path))
with Pool(workers) as pool:
@ -1017,13 +1186,13 @@ def scan_paths(paths: Sequence[Path]) -> tuple[Violation, ...]:
def main(argv: Sequence[str]) -> int:
paths = tuple(a for a in argv if not a.startswith("-"))
paths: Final = tuple(a for a in argv if not a.startswith("-"))
if not paths:
print("usage: check_type_discipline.py <files-or-dirs>...", file=sys.stderr)
return 2
targets = tuple(collect_paths(paths))
violations = sorted(scan_paths(targets))
targets: Final = tuple(collect_paths(paths))
violations: Final = sorted(scan_paths(targets))
for v in violations:
print(v.render())

8
scripts/type_discipline_gate.py
View file

@ -16,11 +16,12 @@ syntax while enableTypeIgnoreComments is false), LIT010 (assignment without a
Final declaration; suppress deliberate rebinding with `# rebind-ok: <reason>`),
LIT011 (parameter rebinding or in-place mutation), LIT012 (TypedDict field
without a `ReadOnly[...]` qualifier; suppress with `# writable-ok: <reason>`),
LIT013 (`*-ok` suppression that suppresses nothing), and LIT014 (comprehension
LIT013 (`*-ok` suppression that suppresses nothing), LIT014 (comprehension
with more than one `for` or `if` clause; suppress with
`# comprehension-ok: <reason>` on a spanned line, which belongs to the innermost
violating comprehension spanning it and to any single-line violating
comprehension on that line).
comprehension on that line), and LIT015 (pydantic model not frozen; suppress
with `# frozen-ok: <reason>`).
The merge-base counts come from scripts/lint_base_counts.py: the disk cache,
then the CI artifact published for that commit, then a pass of the current
@ -158,7 +159,8 @@ def cmd_check(base: str) -> None:
"Remove the new violations, give each a reason (`# noqa: XXX # <reason>`, "
"`# pyright: ignore[rule] # <reason>`, `# mutable-ok: <reason>`, `# cast-ok: <reason>`, "
"`# guard-ok: <reason>`, `# kwargs-ok: <reason>`, `# rebind-ok: <reason>`, "
"`# writable-ok: <reason>`, `# comprehension-ok: <reason>`), or remove an equal number "
"`# writable-ok: <reason>`, `# comprehension-ok: <reason>`, `# frozen-ok: <reason>`), "
"or remove an equal number "
"elsewhere; the ceiling is the merge-base count."
)
raise SystemExit(1)

2
tests/integration/cost_calculation/cost_tracking_case.py
View file

@ -250,7 +250,7 @@ class CostTrackingTestCase(BaseModel):
"/v1/audio/speech",
"/v1/images/generations",
"/v1/images/edits",
"/v1/decisions",
"/v1/systemone",
]
| Annotated[str, Field(pattern=r"^/(gemini|anthropic|bedrock)/")]
) = "/v1/chat/completions"

2
tests/integration/cost_calculation/cost_tracking_cases.json
View file

@ -31207,7 +31207,7 @@
"name": "perplexity/pplx-decider-v1-27b-decisions",
"covers": "quota_management.spend_tracking.decisions_costs",
"model": "perplexity/pplx-decider-v1-27b",
"endpoint": "/v1/decisions",
"endpoint": "/v1/systemone",
"request": {
"model": "$MODEL",
"state": {

2
tests/integration/cost_calculation/test_cost_tracking.py
View file

@ -268,7 +268,7 @@ def test_case_bills_expected_cost(gateway: Gateway, case: CostTrackingTestCase)
assert row.spend == 0, f"{case.name}: failure spend was {row.spend}"
return
assert response.is_success, f"{case.name}: proxy returned {response.status_code}: {response.text[:400]}"
if case.endpoint == "/v1/decisions":
if case.endpoint == "/v1/systemone":
observed: Final = JSON_OBJECT.validate_json(
httpx.get(f"{gateway.upstream_url}/__observations", timeout=5, trust_env=False).content
)

4
tests/integration/providers/test_decisions_chaos.py
View file

@ -27,7 +27,7 @@ _API_KEY: Final = "synthetic-decisions-key"
_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
_STARTED_WORKER: Final = re.compile(r"Started server process \[(\d+)\]")
_QUESTIONS: Final[dict[str, JsonValue]] = {"fine": {"type": "noul", "instructions": "Is the state fine?"}}
_ROUTES: Final = ("/v1/decisions", "/decisions")
_ROUTES: Final = ("/v1/systemone", "/systemone")
@dataclass(frozen=True, slots=True)
@ -221,7 +221,7 @@ async def test_worker_sigkill_mid_burst_leaves_the_sibling_serving_the_default_m
release.set()
served: Final = await burst
assert len(served) == held_by[survivor_pid], (held_by, len(served))
follow_up: Final = _Call(route="/decisions", marker=f"ok-{uuid.uuid4().hex}", fail=False)
follow_up: Final = _Call(route="/systemone", marker=f"ok-{uuid.uuid4().hex}", fail=False)
(answered,) = await _burst(base_url, candidate.key, None, (follow_up,))
await asyncio.to_thread(
eventually,

14
tests/integration/providers/test_decisions_wire.py
View file

@ -158,7 +158,7 @@ def _deployment(scenario: Scenario, handle: ScenarioHandle, provider: _Provider)
def _decide(gateway: Gateway, model: str, *, key: str | None = None, **extra: JsonValue) -> httpx.Response:
return gateway.request(
"POST", "/v1/decisions", {"model": model, "state": _STATE, "questions": _QUESTIONS, **extra}, key=key
"POST", "/v1/systemone", {"model": model, "state": _STATE, "questions": _QUESTIONS, **extra}, key=key
)
@ -310,7 +310,7 @@ def test_invalid_bodies_are_refused_at_the_gateway_without_an_upstream_call(gate
handle: Final = _register(scenario, _answer_body(_PERPLEXITY))
model: Final = _deployment(scenario, handle, _PERPLEXITY)
for label, body in _INVALID_BODIES:
response: Final = gateway.request("POST", "/v1/decisions", {"model": model, **body})
response: Final = gateway.request("POST", "/v1/systemone", {"model": model, **body})
assert response.status_code == 400, (label, response.text)
assert "Invalid Decisions request" in response.text, (label, response.text)
assert _upstream_calls(gateway, handle) == []
@ -329,7 +329,7 @@ def test_key_checks_match_chat(gateway: Gateway) -> None:
handle: Final = _register(scenario, _answer_body(_PERPLEXITY))
model: Final = _deployment(scenario, handle, _PERPLEXITY)
anonymous: Final = gateway.client.post(
"/v1/decisions", json={"model": model, "state": _STATE, "questions": _QUESTIONS}
"/v1/systemone", json={"model": model, "state": _STATE, "questions": _QUESTIONS}
)
assert anonymous.status_code == 401, anonymous.text
restricted: Final = scenario.key(models=[f"other-{uuid.uuid4().hex}"])
@ -395,7 +395,7 @@ def test_a_deployment_opted_into_client_api_base_sends_decisions_and_chat_to_the
assert _calls_to(observed, configured) == []
def test_a_config_pass_through_at_v1_decisions_keeps_answering_and_the_native_api_serves_decisions(
def test_a_config_pass_through_at_v1_decisions_keeps_answering_and_the_native_api_serves_system_one(
gateway: Gateway, tmp_path: Path
) -> None:
with gateway.scenario() as scenario:
@ -407,9 +407,11 @@ def test_a_config_pass_through_at_v1_decisions_keeps_answering_and_the_native_ap
tmp_path, f"{pass_through_target.api_base()}/v1/decisions", native_target.api_base()
)
with owned_proxy_process(gateway, tmp_path, {}, config=config) as owned:
through: Final = _decide(owned.gateway, _PASS_THROUGH_MODEL)
through: Final = owned.gateway.request(
"POST", "/v1/decisions", {"model": _PASS_THROUGH_MODEL, "state": _STATE, "questions": _QUESTIONS}
)
native: Final = owned.gateway.request(
"POST", "/decisions", {"model": _PASS_THROUGH_NEIGHBOUR, "state": _STATE, "questions": _QUESTIONS}
"POST", "/systemone", {"model": _PASS_THROUGH_NEIGHBOUR, "state": _STATE, "questions": _QUESTIONS}
)
assert through.status_code == 200, through.text
assert through.json() == {"model": _PASS_THROUGH_MODEL, "answers": _ANSWERS, "usage": _USAGE}

251
tests/unit/decisions/test_main.py
View file

@ -18,6 +18,7 @@ from litellm.types.decisions import (
DecisionsResponse,
DecisionsUsage,
NoulAnswer,
OpenAIDecisionResponse,
ScoreAnswer,
)
@ -30,6 +31,8 @@ _QUESTIONS: Final[Mapping[str, object]] = MappingProxyType(
)
_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": {
@ -80,6 +83,21 @@ _PROVIDERS: Final[tuple[tuple[str, str, str, str], ...]] = (
),
)
_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:
@ -265,6 +283,46 @@ def test_decisions_cost_uses_litellm_token_pricing() -> None:
assert cost == pytest.approx(expected_cost)
@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)
@ -487,7 +545,7 @@ async def test_cloudflare_clef_resolves_model_and_response_envelope(
"state": "review",
"questions": {"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
}
assert response.answers == DecisionsResponse.model_validate(_RESPONSE).answers
assert response.answers == {"is_defect": NoulAnswer(type="noul", noul=0.9)}
assert response._hidden_params["model"] == "cloudflare/@cf/cloudflare/clef"
@ -673,3 +731,194 @@ async def test_strands_decider_provider_resolution_and_router_dispatch(
assert provider_resolution[:2] == ("strands-decider-2B-hobson-v19", "strands_decider")
assert route.called
assert response.model == _STRANDS_RESPONSE["model"]
@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,
) -> 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.adecisions(
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"
luna_cost: Final = litellm.model_cost["gpt-6-luna"]
expected_cost: Final = (
(_INPUT_TOKENS - _CACHED_TOKENS - _CACHE_WRITE_TOKENS) * float(luna_cost["input_cost_per_token"])
+ _CACHED_TOKENS * float(luna_cost["cache_read_input_token_cost"])
+ _CACHE_WRITE_TOKENS * float(luna_cost["cache_creation_input_token_cost"])
+ _OUTPUT_TOKENS * float(luna_cost["output_cost_per_token"])
)
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.decisions(
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",
),
(
"openai/gpt-6-luna",
{
"state": "review",
"input": "review",
"questions": [{"type": "predicate", "instructions": "Is this a defect?"}],
},
"not both",
),
),
ids=("image_to_systemone_provider", "state_and_input"),
)
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

180
tests/unit/llms/base_llm/decisions/test_base_decisions_transformation.py Normal file
View file

@ -0,0 +1,180 @@
from collections.abc import Mapping
from typing import Final
import pytest
from pydantic import TypeAdapter
from litellm.llms.base_llm.decisions.transformation import (
ir_to_systemone_request,
ir_to_systemone_response,
parse_systemone_response,
systemone_request_to_ir,
)
from litellm.llms.openai.decisions.transformation import openai_request_to_ir
from litellm.types.decisions import (
MAX_DECISION_QUESTIONS,
DecisionsIRRefusal,
DecisionsIRRequest,
DecisionsRequestBody,
OpenAIDecisionRequestBody,
UnsupportedDecisionsRequest,
)
_SYSTEMONE_BODY: Final[TypeAdapter[DecisionsRequestBody]] = TypeAdapter(DecisionsRequestBody)
_OPENAI_BODY: Final[TypeAdapter[OpenAIDecisionRequestBody]] = TypeAdapter(OpenAIDecisionRequestBody)
_SYSTEMONE_REQUEST: Final[Mapping[str, object]] = {
"state": {"ticket": 1234, "text": "Screen cracked"},
"questions": {
"damaged": {
"type": "noul",
"instructions": "Is the item damaged?",
"criteria": {"true": "Visible damage", "false": None},
"provider_field": "dropped",
},
"rubric_only": {"type": "noul", "criteria": {"true": {"signal": "refund"}}},
"action": {
"type": "choice",
"instructions": {"policy": "refund-v2"},
"criteria": {"refund": "Within 30 days", "escalate": None},
},
"severity": {"type": "score", "criteria": ["minor", {"label": "major"}]},
},
}
def _openai_ir(raw: Mapping[str, object]) -> DecisionsIRRequest:
return openai_request_to_ir(_OPENAI_BODY.validate_python(raw))
def test_a_systemone_request_reaches_a_systemone_provider_unchanged() -> None:
ir: Final = systemone_request_to_ir(_SYSTEMONE_BODY.validate_python(_SYSTEMONE_REQUEST))
assert ir_to_systemone_request("jev-latest", ir) == {"model": "jev-latest", **_SYSTEMONE_REQUEST}
@pytest.mark.parametrize(
("names", "keys"),
(
(("damaged", "action"), ("damaged", "action")),
(("damaged", None), ("0", "1")),
(("damaged", "damaged"), ("0", "1")),
),
ids=("all_named", "one_unnamed", "duplicate_names"),
)
def test_openai_questions_are_keyed_by_name_only_when_every_name_is_unique(
names: tuple[str | None, str | None], keys: tuple[str, str]
) -> None:
questions: Final = [
{"type": "predicate", "instructions": f"Question {index}?", **({} if name is None else {"name": name})}
for index, name in enumerate(names)
]
body: Final = ir_to_systemone_request("jev-latest", _openai_ir({"input": "review", "questions": questions}))
assert tuple(_SYSTEMONE_BODY.validate_python(body).questions) == keys
def test_openai_messages_become_systemone_state_text() -> None:
ir: Final = _openai_ir(
{
"input": [
{
"role": "user",
"content": [
{"type": "input_text", "text": "The package arrived broken."},
{"type": "input_text", "text": "I want a refund."},
],
},
{"role": "user", "content": "Order 1234."},
],
"questions": [{"type": "predicate", "instructions": "Is this a defect?"}],
}
)
body: Final = ir_to_systemone_request("jev-latest", ir)
assert isinstance(body, Mapping)
assert body["state"] == "The package arrived broken.\n\nI want a refund.\n\nOrder 1234."
def test_image_input_is_unsupported_by_systemone_providers() -> None:
ir: Final = _openai_ir(
{
"input": [
{
"role": "user",
"content": [
{"type": "input_text", "text": "Is the screen cracked?"},
{"type": "input_image", "image_url": "data:image/png;base64,AA=="},
],
}
],
"questions": [{"type": "predicate", "instructions": "Is this a defect?"}],
}
)
assert isinstance(ir_to_systemone_request("jev-latest", ir), UnsupportedDecisionsRequest)
def test_the_largest_openai_request_accepted_translates_to_a_valid_systemone_request() -> None:
ir: Final = _openai_ir(
{
"input": "review",
"questions": [
{"type": "predicate", "instructions": f"Question {index}?"} for index in range(MAX_DECISION_QUESTIONS)
],
}
)
translated: Final = _SYSTEMONE_BODY.validate_python(ir_to_systemone_request("jev-latest", ir))
assert len(translated.questions) == MAX_DECISION_QUESTIONS
def test_a_systemone_response_reaches_the_caller_unchanged_with_provider_extras() -> None:
payload: Final = {
"model": "jev-latest",
"answers": {
"damaged": {"type": "noul", "noul": 0.95, "rationale": "crack visible"},
"action": {
"type": "choice",
"choice": "refund",
"confidence": 0.8,
"probabilities": {"refund": 0.9, "escalate": 0.1},
"calibrated": True,
},
"severity": {
"type": "score",
"score": 0.4,
"confidence": 0.6,
"legend": {"0": "minor", "1": {"label": "major"}},
"probabilities": {"0": 0.6, "1": 0.4},
"raw_logits": [0.1, 0.2],
},
},
"usage": {"input_tokens": 383, "output_tokens": 2, "cost": 0.25},
"latency_ms": 3722.17,
}
ir: Final = systemone_request_to_ir(_SYSTEMONE_BODY.validate_python(_SYSTEMONE_REQUEST))
response: Final = ir_to_systemone_response(parse_systemone_response(payload, ir), ir)
assert response.model_dump(mode="json") == payload
def test_missing_or_mismatched_systemone_answers_are_refusals_left_out_of_the_response() -> None:
ir: Final = systemone_request_to_ir(_SYSTEMONE_BODY.validate_python(_SYSTEMONE_REQUEST))
payload: Final = {
"model": "jev-latest",
"answers": {
"damaged": {"type": "noul", "noul": 0.95},
"action": {"type": "noul", "noul": 0.5},
},
"usage": {"input_tokens": 10, "output_tokens": 1},
}
parsed: Final = parse_systemone_response(payload, ir)
assert parsed.answers[1:] == (DecisionsIRRefusal(), DecisionsIRRefusal(), DecisionsIRRefusal())
assert tuple(ir_to_systemone_response(parsed, ir).answers) == ("damaged",)

0
tests/unit/llms/openai/decisions/__init__.py Normal file
View file

256
tests/unit/llms/openai/decisions/test_openai_decisions_transformation.py Normal file
View file

@ -0,0 +1,256 @@
from collections.abc import Mapping
from typing import Final
import pytest
from pydantic import TypeAdapter
from litellm.llms.base_llm.decisions.transformation import ir_to_systemone_response, systemone_request_to_ir
from litellm.llms.openai.decisions.transformation import (
OpenAIDecisionsConfig,
ir_to_openai_request,
ir_to_openai_response,
openai_request_to_ir,
)
from litellm.types.decisions import (
DecisionsIRRequest,
DecisionsRequestBody,
OpenAIDecisionRequestBody,
UnsupportedDecisionsRequest,
)
_SYSTEMONE_BODY: Final[TypeAdapter[DecisionsRequestBody]] = TypeAdapter(DecisionsRequestBody)
_OPENAI_BODY: Final[TypeAdapter[OpenAIDecisionRequestBody]] = TypeAdapter(OpenAIDecisionRequestBody)
_OPENAI_REQUEST: Final[Mapping[str, object]] = {
"input": [
{
"type": "message",
"role": "user",
"content": [
{"type": "input_text", "text": "The screen is cracked."},
{"type": "input_image", "image_url": "data:image/png;base64,AA==", "detail": "high"},
],
},
{"type": "message", "role": "user", "content": "Order 1234."},
],
"questions": [
{"type": "predicate", "name": "damaged", "instructions": "Is the item damaged?"},
{
"type": "choice",
"instructions": "Should we refund?",
"choices": [{"value": True, "description": "Refund now"}, {"value": "escalate"}],
},
{
"type": "score",
"name": "severity",
"instructions": "How severe is it?",
"levels": [{"label": "minor"}, {"label": "major", "description": "Product unusable"}],
},
{"type": "predicate", "name": "fraud", "instructions": "Is this fraud?"},
],
"safety_identifier": "end-user-1",
}
_PREDICATE_ANSWER: Final[Mapping[str, object]] = {"type": "predicate", "name": "damaged", "probability": 0.95}
_CHOICE_ANSWER: Final[Mapping[str, object]] = {
"type": "choice",
"name": None,
"choice": True,
"probabilities": [{"value": True, "probability": 0.9}, {"value": "escalate", "probability": 0.1}],
"confidence": 0.8,
}
_REFUSAL_ANSWER: Final[Mapping[str, object]] = {"type": "refusal", "name": "fraud"}
_USAGE: Final[Mapping[str, object]] = {
"input_tokens": 383,
"input_tokens_details": {"cached_tokens": 256, "cache_write_tokens": 64},
"output_tokens": 2,
"output_tokens_details": {"reasoning_tokens": 1},
"total_tokens": 385,
}
_OPENAI_RESPONSE: Final[Mapping[str, object]] = {
"model": "gpt-6-luna",
"answers": [
_PREDICATE_ANSWER,
_CHOICE_ANSWER,
{
"type": "score",
"name": "severity",
"score": 0.7,
"probabilities": [
{"value": 0, "label": "minor", "probability": 0.3},
{"value": 1, "label": "major", "probability": 0.7},
],
"confidence": 0.6,
},
_REFUSAL_ANSWER,
],
"usage": _USAGE,
}
def _openai_ir(raw: Mapping[str, object]) -> DecisionsIRRequest:
return openai_request_to_ir(_OPENAI_BODY.validate_python(raw))
def test_an_openai_request_reaches_openai_unchanged() -> None:
assert ir_to_openai_request("gpt-6-luna", _openai_ir(_OPENAI_REQUEST)) == {
"model": "gpt-6-luna",
**_OPENAI_REQUEST,
}
def test_an_openai_response_reaches_the_caller_unchanged() -> None:
ir: Final = _openai_ir(_OPENAI_REQUEST)
parsed: Final = OpenAIDecisionsConfig().parse_response(_OPENAI_RESPONSE, ir)
assert ir_to_openai_response(parsed, ir, "requested").model_dump(mode="json") == _OPENAI_RESPONSE
def test_answers_openai_did_not_return_are_refusals() -> None:
ir: Final = _openai_ir(_OPENAI_REQUEST)
payload: Final = {**_OPENAI_RESPONSE, "answers": [_PREDICATE_ANSWER]}
response: Final = ir_to_openai_response(OpenAIDecisionsConfig().parse_response(payload, ir), ir, "requested")
assert [answer.type for answer in response.answers] == ["predicate", "refusal", "refusal", "refusal"]
assert [answer.name for answer in response.answers] == ["damaged", None, "severity", "fraud"]
def test_a_systemone_request_becomes_an_openai_request_with_questions_named_by_their_keys() -> None:
request: Final = _SYSTEMONE_BODY.validate_python(
{
"state": {"ticket": 1234, "text": "Screen cracked"},
"questions": {
"damaged": {
"type": "noul",
"instructions": "Is the item damaged?",
"criteria": {"true": "Visible damage", "false": None},
"provider_field": "dropped",
},
"rubric_only": {"type": "noul", "criteria": {"true": {"signal": "refund"}}},
"action": {
"type": "choice",
"instructions": {"policy": "refund-v2"},
"criteria": {"refund": "Within 30 days", "escalate": None},
},
"severity": {"type": "score", "criteria": ["minor", {"label": "major"}]},
},
}
)
assert ir_to_openai_request("gpt-6-luna", systemone_request_to_ir(request)) == {
"model": "gpt-6-luna",
"input": '{"ticket": 1234, "text": "Screen cracked"}',
"questions": [
{
"type": "predicate",
"name": "damaged",
"instructions": "Is the item damaged?\n\nAnswer true when: Visible damage",
},
{"type": "predicate", "name": "rubric_only", "instructions": 'Answer true when: {"signal": "refund"}'},
{
"type": "choice",
"name": "action",
"instructions": '{"policy": "refund-v2"}',
"choices": [{"value": "refund", "description": "Within 30 days"}, {"value": "escalate"}],
},
{
"type": "score",
"name": "severity",
"instructions": "Which level best fits the input?",
"levels": [{"label": "minor"}, {"label": '{"label": "major"}'}],
},
],
}
def test_systemone_questions_without_instructions_become_valid_openai_questions() -> None:
request: Final = _SYSTEMONE_BODY.validate_python(
{
"state": "Screen cracked",
"questions": {
"damaged": {"type": "noul", "criteria": {"false": None}},
"action": {"type": "choice", "criteria": {"refund": None, "escalate": None}},
"severity": {"type": "score", "criteria": ["minor", "major"]},
},
}
)
body: Final = ir_to_openai_request("gpt-6-luna", systemone_request_to_ir(request))
assert all(question.instructions for question in _OPENAI_BODY.validate_python(body).questions)
@pytest.mark.parametrize(
"question",
({"type": "choice", "criteria": {"refund": None}}, {"type": "score", "criteria": ["minor"]}),
ids=("choice", "score"),
)
def test_a_systemone_question_with_one_option_is_unsupported_by_openai(question: Mapping[str, object]) -> None:
request: Final = _SYSTEMONE_BODY.validate_python(
{
"state": "Screen cracked",
"questions": {"damaged": {"type": "noul", "instructions": "Damaged?"}, "q": question},
}
)
assert isinstance(ir_to_openai_request("gpt-6-luna", systemone_request_to_ir(request)), UnsupportedDecisionsRequest)
def test_an_openai_response_becomes_systemone_answers_with_the_callers_score_labels() -> None:
ir: Final = systemone_request_to_ir(
_SYSTEMONE_BODY.validate_python(
{
"state": "Screen cracked",
"questions": {
"damaged": {"type": "noul", "instructions": "Damaged?"},
"action": {"type": "choice", "criteria": {"true": None, "escalate": None}},
"severity": {"type": "score", "criteria": ["minor", {"label": "major"}]},
"fraud": {"type": "noul", "instructions": "Fraud?"},
},
}
)
)
payload: Final = {
**_OPENAI_RESPONSE,
"answers": [
_PREDICATE_ANSWER,
_CHOICE_ANSWER,
{
"type": "score",
"name": "severity",
"score": 0.7,
"probabilities": [
{"value": 0, "label": "minor", "probability": 0.3},
{"value": 1, "label": '{"label": "major"}', "probability": 0.7},
],
"confidence": 0.6,
},
_REFUSAL_ANSWER,
],
}
response: Final = ir_to_systemone_response(OpenAIDecisionsConfig().parse_response(payload, ir), ir)
assert response.model_dump(mode="json") == {
"model": "gpt-6-luna",
"answers": {
"damaged": {"type": "noul", "noul": 0.95},
"action": {
"type": "choice",
"choice": "true",
"confidence": 0.8,
"probabilities": {"true": 0.9, "escalate": 0.1},
},
"severity": {
"type": "score",
"score": 0.7,
"confidence": 0.6,
"legend": {"0": "minor", "1": {"label": "major"}},
"probabilities": {"0": 0.3, "1": 0.7},
},
},
"usage": {"input_tokens": 383, "output_tokens": 2},
}
assert response.usage is not None
assert (response.usage.cached_tokens, response.usage.cache_write_tokens) == (256, 64)

75
tests/unit/proxy/auth/test_model_checks.py
View file

@ -967,6 +967,81 @@ def test_get_complete_model_list_sentinel_only_grants_nothing():
assert result == []
def test_get_provider_models_admits_providers_without_a_static_catalog():
"""Providers without a static model list are no longer rejected up front.
With endpoint discovery off, get_valid_models falls back to the (empty)
static list, so the result is [] rather than None. Before the fix this
returned None and the wildcard was never expanded.
"""
import litellm
from litellm.proxy.auth.model_checks import get_provider_models
from litellm.types.router import LiteLLM_Params
assert "litellm_proxy" not in litellm.models_by_provider
assert "hosted_vllm" not in litellm.models_by_provider
result = get_provider_models(
"litellm_proxy",
litellm_params=LiteLLM_Params(
model="litellm_proxy/*",
api_base="http://upstream:4000",
api_key="sk-upstream",
),
)
assert result == []
def test_get_complete_model_list_discovers_litellm_proxy_wildcard_models(monkeypatch):
"""A litellm_proxy/* deployment lists the upstream proxy's models when endpoint discovery is on."""
import litellm
from litellm import Router
from litellm.llms.openai.chat.gpt_transformation import OpenAIGPTConfig
from litellm.proxy.auth.model_checks import get_complete_model_list
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
captured = {}
def fake_get_models(self, api_key=None, api_base=None):
captured["api_key"] = api_key
captured["api_base"] = api_base
return ["gpt-4o", "claude-sonnet"]
monkeypatch.setattr(OpenAIGPTConfig, "get_models", fake_get_models)
router = Router(
model_list=[
{
"model_name": "litellm_proxy/*",
"litellm_params": {
"model": "litellm_proxy/*",
"api_base": "http://upstream:4000",
"api_key": "sk-upstream",
},
}
]
)
result = get_complete_model_list(
key_models=[],
team_models=[],
proxy_model_list=["litellm_proxy/*"],
user_model=None,
infer_model_from_keys=False,
llm_router=router,
)
assert captured == {"api_key": "sk-upstream", "api_base": "http://upstream:4000"}
assert "litellm_proxy/gpt-4o" in result
assert "litellm_proxy/claude-sonnet" in result
def test_get_provider_models_returns_none_for_an_unknown_provider():
from litellm.proxy.auth.model_checks import get_provider_models
assert get_provider_models("not-a-real-provider") is None
def test_transcribe_is_a_known_provider_for_wildcard_expansion():
import litellm
from litellm.proxy.auth.model_checks import (

273
tests/unit/proxy/decisions_endpoints/test_endpoints.py
View file

@ -16,7 +16,7 @@ from starlette.routing import Match
import litellm
from litellm.proxy._lazy_features import LAZY_FEATURES, LazyFeature, attach_lazy_features
from litellm.proxy.decisions_endpoints.endpoints import decisions
from litellm.proxy.decisions_endpoints.endpoints import decisions, systemone
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import SafeRouteAdder
from litellm.proxy.proxy_server import (
app,
@ -77,7 +77,7 @@ def client(monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:
litellm.in_memory_llm_clients_cache.flush_cache()
@pytest.mark.parametrize("endpoint", ("/v1/decisions", "/decisions"))
@pytest.mark.parametrize("endpoint", ("/v1/systemone", "/systemone"))
def test_proxy_decisions_route_returns_answers_and_cost(
client: TestClient,
respx_mock: respx.MockRouter,
@ -126,7 +126,7 @@ def test_proxy_decisions_dispatches_typesafe_deployment(
upstream: Final = respx_mock.post("https://api.typesafe.ai/v1/systemone").respond(json=_RESPONSE)
response: Final = client.post(
"/v1/decisions",
"/v1/systemone",
json={
"model": "jev",
"state": {"source": "proxy-test"},
@ -165,7 +165,7 @@ def test_proxy_decisions_sends_the_env_key_to_the_deployment_api_base(
monkeypatch.setattr(litellm.proxy.proxy_server, "llm_router", router)
upstream: Final = respx_mock.post("https://egress.example/perplexity/v1/decisions").respond(json=_RESPONSE)
response: Final = client.post("/v1/decisions", json=_REQUEST)
response: Final = client.post("/v1/systemone", json=_REQUEST)
assert response.status_code == 200, response.text
assert upstream.call_count == 1
@ -177,7 +177,7 @@ def test_proxy_decisions_unknown_model_is_a_client_error(
respx_mock: respx.MockRouter,
) -> None:
response: Final = client.post(
"/v1/decisions",
"/v1/systemone",
json={
"model": "missing-model",
"state": "review",
@ -210,7 +210,7 @@ def test_proxy_decisions_missing_required_field_is_a_client_error(
) -> None:
upstream: Final = respx_mock.post("https://api.perplexity.ai/v1/decisions").respond(json=_RESPONSE)
response: Final = client.post("/v1/decisions", json=request_body)
response: Final = client.post("/v1/systemone", json=request_body)
assert response.status_code == 400, response.text
assert not upstream.called
@ -238,7 +238,7 @@ def test_proxy_decisions_dispatches_strands_decider(
upstream: Final = respx_mock.post("https://strands.example/v1/systemone").respond(json=_STRANDS_RESPONSE)
response: Final = client.post(
"/v1/decisions",
"/v1/systemone",
json={
"model": "strands",
"state": {"source": "proxy-test"},
@ -266,7 +266,7 @@ def test_proxy_decisions_without_model_uses_the_proxy_default_model(
upstream: Final = respx_mock.post("https://api.perplexity.ai/v1/decisions").respond(json=_RESPONSE)
response: Final = client.post(
"/v1/decisions", json={key: value for key, value in _REQUEST.items() if key != "model"}
"/v1/systemone", json={key: value for key, value in _REQUEST.items() if key != "model"}
)
assert response.status_code == 200, response.text
@ -275,6 +275,261 @@ def test_proxy_decisions_without_model_uses_the_proxy_default_model(
assert json.loads(upstream.calls[0].request.content)["model"] == "pplx-decider-v1-27b"
_OPENAI_FORMAT_REQUEST: Final[Mapping[str, object]] = {
"model": "decider",
"input": [
{
"role": "user",
"content": [
{"type": "input_text", "text": "The package arrived with a broken screen."},
{"type": "input_text", "text": "I want a refund."},
],
},
{"role": "user", "content": "Order 1234."},
],
"questions": [
{"type": "predicate", "name": "damaged", "instructions": "Does the customer report a damaged item?"},
{
"type": "choice",
"instructions": "Should we refund?",
"choices": [{"value": True, "description": "Refund now"}, {"value": "escalate"}],
},
{
"type": "score",
"name": "severity",
"instructions": "How severe is the issue?",
"levels": [{"label": "minor"}, {"label": "major", "description": "Product unusable"}],
},
{"type": "predicate", "name": "fraud", "instructions": "Is this fraud?"},
],
"safety_identifier": "end-user-1",
}
_SYSTEMONE_ANSWERS_FOR_OPENAI_REQUEST: Final[Mapping[str, object]] = {
"model": "pplx-decider-v1-27b",
"answers": {
"0": {"type": "noul", "noul": 0.95},
"1": {"type": "choice", "choice": "true", "confidence": 0.8, "probabilities": {"true": 0.9, "escalate": 0.1}},
"2": {
"type": "score",
"score": 0.7,
"confidence": 0.6,
"legend": {"0": "minor", "1": "major: Product unusable"},
"probabilities": {"0": 0.3, "1": 0.7},
},
},
"usage": {"input_tokens": _INPUT_TOKENS, "output_tokens": _OUTPUT_TOKENS},
}
_OPENAI_FORMAT_ANSWERS: Final = [
{"type": "predicate", "name": "damaged", "probability": 0.95},
{
"type": "choice",
"name": None,
"choice": True,
"probabilities": [{"value": True, "probability": 0.9}, {"value": "escalate", "probability": 0.1}],
"confidence": 0.8,
},
{
"type": "score",
"name": "severity",
"score": 0.7,
"probabilities": [
{"value": 0, "label": "minor", "probability": 0.3},
{"value": 1, "label": "major", "probability": 0.7},
],
"confidence": 0.6,
},
{"type": "refusal", "name": "fraud"},
]
@pytest.mark.parametrize("endpoint", ("/v1/decisions", "/decisions"))
def test_openai_format_decisions_translate_through_systemone(
client: TestClient,
respx_mock: respx.MockRouter,
endpoint: str,
) -> None:
upstream: Final = respx_mock.post("https://api.perplexity.ai/v1/decisions").respond(
json=_SYSTEMONE_ANSWERS_FOR_OPENAI_REQUEST
)
response: Final = client.post(endpoint, json=_OPENAI_FORMAT_REQUEST)
assert response.status_code == 200, response.text
assert json.loads(upstream.calls[0].request.content) == {
"model": "pplx-decider-v1-27b",
"state": "The package arrived with a broken screen.\n\nI want a refund.\n\nOrder 1234.",
"questions": {
"0": {"type": "noul", "instructions": "Does the customer report a damaged item?"},
"1": {
"type": "choice",
"instructions": "Should we refund?",
"criteria": {"true": "Refund now", "escalate": None},
},
"2": {
"type": "score",
"instructions": "How severe is the issue?",
"criteria": ["minor", "major: Product unusable"],
},
"3": {"type": "noul", "instructions": "Is this fraud?"},
},
}
body: Final = response.json()
assert body["model"] == _SYSTEMONE_ANSWERS_FOR_OPENAI_REQUEST["model"]
assert body["answers"] == _OPENAI_FORMAT_ANSWERS
assert body["usage"] == {
"input_tokens": _INPUT_TOKENS,
"input_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 0},
"output_tokens": _OUTPUT_TOKENS,
"output_tokens_details": {"reasoning_tokens": 0},
"total_tokens": _INPUT_TOKENS + _OUTPUT_TOKENS,
}
assert float(response.headers["x-litellm-response-cost"]) > 0
@pytest.mark.parametrize(
("endpoint", "request_body", "upstream_response"),
(
("/v1/systemone", _REQUEST, _RESPONSE),
("/v1/decisions", _OPENAI_FORMAT_REQUEST, _SYSTEMONE_ANSWERS_FOR_OPENAI_REQUEST),
),
ids=("systemone", "openai_format"),
)
def test_decisions_return_guardrail_information_when_requested(
client: TestClient,
respx_mock: respx.MockRouter,
endpoint: str,
request_body: Mapping[str, object],
upstream_response: Mapping[str, object],
) -> None:
respx_mock.post("https://api.perplexity.ai/v1/decisions").respond(json=upstream_response)
response: Final = client.post(endpoint, json={**request_body, "include_guardrail_response": True})
assert response.status_code == 200, response.text
assert response.json()["guardrail_information"] == []
@pytest.mark.parametrize(
"request_body",
(
_REQUEST,
{
"model": "decider",
"input": "review",
"questions": [
{
"type": "choice",
"instructions": "Pick one",
"choices": [{"value": True}, {"value": "true"}],
}
],
},
{
"model": "decider",
"input": [
{"role": "user", "content": [{"type": "input_image", "image_url": "data:image/png;base64,AA=="}]}
],
"questions": [{"type": "predicate", "instructions": "Is this a defect?"}],
},
),
ids=("systemone_body", "colliding_choice_values", "image_input"),
)
def test_openai_format_decisions_rejects_bodies_it_cannot_translate(
client: TestClient,
respx_mock: respx.MockRouter,
request_body: Mapping[str, object],
) -> None:
upstream: Final = respx_mock.post("https://api.perplexity.ai/v1/decisions").respond(json=_RESPONSE)
response: Final = client.post("/v1/decisions", json=request_body)
assert response.status_code == 400, response.text
assert not upstream.called
@pytest.mark.parametrize("endpoint", ("/v1/systemone", "/v1/decisions"))
@pytest.mark.parametrize("raw_body", (b"", b"{not json"), ids=("empty", "malformed"))
def test_a_body_that_is_not_json_is_a_client_error(
client: TestClient,
respx_mock: respx.MockRouter,
endpoint: str,
raw_body: bytes,
) -> None:
upstream: Final = respx_mock.post("https://api.perplexity.ai/v1/decisions").respond(json=_RESPONSE)
response: Final = client.post(endpoint, content=raw_body, headers={"Content-Type": "application/json"})
assert response.status_code == 400, response.text
assert not upstream.called
def test_openai_format_decisions_reach_an_openai_deployment_unchanged_including_images(
client: TestClient,
monkeypatch: pytest.MonkeyPatch,
respx_mock: respx.MockRouter,
) -> None:
monkeypatch.setattr(litellm, "api_base", None)
monkeypatch.delenv("OPENAI_BASE_URL", raising=False)
monkeypatch.delenv("OPENAI_API_BASE", raising=False)
monkeypatch.setattr(
litellm.proxy.proxy_server,
"llm_router",
litellm.Router(
model_list=[{"model_name": "decider", "litellm_params": {"model": "openai/gpt-6-luna", "api_key": "k"}}]
),
)
image_message: Final = {
"type": "message",
"role": "user",
"content": [{"type": "input_image", "image_url": "data:image/png;base64,AA==", "detail": "low"}],
}
request_body: Final = {**_OPENAI_FORMAT_REQUEST, "input": [*_OPENAI_FORMAT_REQUEST["input"], image_message]}
cached_tokens: Final = 128
upstream_usage: Final = {
"input_tokens": _INPUT_TOKENS,
"input_tokens_details": {"cached_tokens": cached_tokens, "cache_write_tokens": 0},
"output_tokens": _OUTPUT_TOKENS,
"output_tokens_details": {"reasoning_tokens": 0},
"total_tokens": _INPUT_TOKENS + _OUTPUT_TOKENS,
}
upstream: Final = respx_mock.post("https://api.openai.com/v1/decisions").respond(
json={"model": "gpt-6-luna", "answers": _OPENAI_FORMAT_ANSWERS, "usage": upstream_usage}
)
response: Final = client.post("/v1/decisions", json=request_body)
assert response.status_code == 200, response.text
assert json.loads(upstream.calls[0].request.content) == {
"model": "gpt-6-luna",
"input": [
{
"type": "message",
"role": "user",
"content": [
{"type": "input_text", "text": "The package arrived with a broken screen."},
{"type": "input_text", "text": "I want a refund."},
],
},
{"type": "message", "role": "user", "content": "Order 1234."},
image_message,
],
"questions": _OPENAI_FORMAT_REQUEST["questions"],
"safety_identifier": "end-user-1",
}
body: Final = response.json()
assert body["answers"] == _OPENAI_FORMAT_ANSWERS
assert body["usage"] == upstream_usage
luna_cost: Final = litellm.model_cost["gpt-6-luna"]
expected_cost: Final = (
(_INPUT_TOKENS - cached_tokens) * float(luna_cost["input_cost_per_token"])
+ cached_tokens * float(luna_cost["cache_read_input_token_cost"])
+ _OUTPUT_TOKENS * float(luna_cost["output_cost_per_token"])
)
assert expected_cost > 0
assert float(response.headers["x-litellm-response-cost"]) == pytest.approx(expected_cost)
def _decisions_feature() -> LazyFeature:
return next(feature for feature in LAZY_FEATURES if feature.name == "decisions")
@ -301,6 +556,8 @@ def test_a_config_pass_through_at_v1_decisions_keeps_its_route_and_the_native_ap
assert client.post("/v1/decisions", json={"model": "gpt-6-luna"}).json() == {"served_by": "pass-through"}
assert _serving_endpoint(bare, "/v1/decisions") is pass_through
assert _serving_endpoint(bare, "/decisions") is decisions
assert _serving_endpoint(bare, "/v1/systemone") is systemone
assert _serving_endpoint(bare, "/systemone") is systemone
def test_with_lazy_routes_disabled_a_config_pass_through_at_v1_decisions_still_wins(

287
tests/unit/test_check_type_discipline.py
View file

@ -11,6 +11,7 @@ import os
import subprocess
import sys
from pathlib import Path
from typing import Final
import pytest
@ -577,6 +578,277 @@ def test_writable_ok_without_reason_is_lit005_and_does_not_suppress(tmp_path):
assert "LIT012" in codes
# --------------------------------------------------------------------------- #
# Unfrozen pydantic models (LIT015)
# --------------------------------------------------------------------------- #
def test_unfrozen_basemodel_is_flagged(tmp_path):
src = "from pydantic import BaseModel\nclass P(BaseModel):\n a: int\n"
assert "LIT015" in _codes(tmp_path, src)
def test_configdict_frozen_true_is_clean(tmp_path):
src = (
"from pydantic import BaseModel, ConfigDict\n"
"class P(BaseModel):\n"
" model_config = ConfigDict(extra='allow', frozen=True)\n"
)
assert "LIT015" not in _codes(tmp_path, src)
def test_shared_frozen_configdict_constant_is_clean(tmp_path):
src: Final = (
"from typing import Final\n"
"from pydantic import BaseModel, ConfigDict\n"
"_RESPONSE_CONFIG: Final = ConfigDict(frozen=True)\n"
"class Foo(BaseModel):\n"
" model_config = _RESPONSE_CONFIG\n"
" x: int\n"
)
assert "LIT015" not in _codes(tmp_path, src)
def test_mutable_config_before_class_is_not_hidden_by_frozen_reassignment(tmp_path):
src: Final = (
"from pydantic import BaseModel, ConfigDict\n"
"_RESPONSE_CONFIG = ConfigDict(frozen=False)\n"
"class Foo(BaseModel):\n"
" model_config = _RESPONSE_CONFIG\n"
" x: int\n"
"_RESPONSE_CONFIG = ConfigDict(frozen=True)\n"
)
assert "LIT015" in _codes(tmp_path, src)
def test_frozen_config_before_class_survives_mutable_reassignment(tmp_path):
src: Final = (
"from pydantic import BaseModel, ConfigDict\n"
"_RESPONSE_CONFIG = ConfigDict(frozen=True)\n"
"class Foo(BaseModel):\n"
" model_config = _RESPONSE_CONFIG\n"
" x: int\n"
"_RESPONSE_CONFIG = ConfigDict(frozen=False)\n"
)
assert "LIT015" not in _codes(tmp_path, src)
def test_config_assigned_only_after_class_is_unresolved(tmp_path):
src: Final = (
"from pydantic import BaseModel, ConfigDict\n"
"class Foo(BaseModel):\n"
" model_config = _RESPONSE_CONFIG\n"
" x: int\n"
"_RESPONSE_CONFIG = ConfigDict(frozen=True)\n"
)
assert "LIT015" in _codes(tmp_path, src)
def test_unknown_model_config_name_is_flagged(tmp_path):
src: Final = (
"from pydantic import BaseModel\nclass Foo(BaseModel):\n model_config = UNKNOWN_CONFIG\n x: int\n"
)
assert "LIT015" in _codes(tmp_path, src)
def test_dict_literal_model_config_frozen_true_is_clean(tmp_path):
src = "from pydantic import BaseModel\nclass P(BaseModel):\n model_config = {'frozen': True, 'extra': 'allow'}\n"
assert "LIT015" not in _codes(tmp_path, src)
def test_subclass_of_in_file_frozen_model_is_clean(tmp_path):
src = (
"from pydantic import BaseModel, ConfigDict\n"
"class Base(BaseModel):\n"
" model_config = ConfigDict(frozen=True)\n"
"class Child(Base):\n"
" a: int\n"
)
assert "LIT015" not in _codes(tmp_path, src)
def test_subclass_of_in_file_unfrozen_model_flags_both(tmp_path):
src = "from pydantic import BaseModel\nclass Base(BaseModel):\n pass\nclass Child(Base):\n a: int\n"
assert _codes(tmp_path, src).count("LIT015") == 2
def test_litellm_pydantic_object_base_without_frozen_is_flagged(tmp_path):
src = "class P(LiteLLMPydanticObjectBase):\n a: int\n"
assert "LIT015" in _codes(tmp_path, src)
def test_inner_config_class_frozen_true_is_clean(tmp_path):
src = "from pydantic import BaseModel\nclass P(BaseModel):\n class Config:\n frozen = True\n"
assert "LIT015" not in _codes(tmp_path, src)
def test_frozen_false_is_flagged(tmp_path):
src = (
"from pydantic import BaseModel, ConfigDict\nclass P(BaseModel):\n model_config = ConfigDict(frozen=False)\n"
)
assert "LIT015" in _codes(tmp_path, src)
def test_later_model_config_frozen_false_overrides_earlier_frozen_true(tmp_path):
src = (
"from pydantic import BaseModel, ConfigDict\n"
"class P(BaseModel):\n"
" model_config = ConfigDict(frozen=True)\n"
" model_config = ConfigDict(frozen=False)\n"
)
assert "LIT015" in _codes(tmp_path, src)
def test_subclass_frozen_false_overrides_frozen_parent(tmp_path):
src = (
"from pydantic import BaseModel, ConfigDict\n"
"class Base(BaseModel):\n"
" model_config = ConfigDict(frozen=True)\n"
"class Writable(Base):\n"
" model_config = ConfigDict(frozen=False)\n"
"class StillFrozen(Base):\n"
" model_config = ConfigDict(extra='allow')\n"
)
assert _codes(tmp_path, src).count("LIT015") == 1
def test_root_model_without_frozen_is_flagged(tmp_path):
src = "from pydantic import RootModel\nclass P(RootModel):\n root: int\n"
assert "LIT015" in _codes(tmp_path, src)
def test_qualified_pydantic_basemodel_is_flagged(tmp_path):
src = "import pydantic\nclass P(pydantic.BaseModel):\n a: int\n"
assert "LIT015" in _codes(tmp_path, src)
def test_keyword_frozen_model_is_clean(tmp_path):
src = "from pydantic import BaseModel\nclass M(BaseModel, frozen=True):\n a: int\n"
assert "LIT015" not in _codes(tmp_path, src)
def test_subclass_of_keyword_frozen_model_is_clean(tmp_path):
src = (
"from pydantic import BaseModel\n"
"class Parent(BaseModel, frozen=True):\n"
" pass\n"
"class Child(Parent):\n"
" a: int\n"
)
assert "LIT015" not in _codes(tmp_path, src)
def test_frozen_false_keyword_overrides_body_frozen_config(tmp_path):
src = (
"from pydantic import BaseModel, ConfigDict\n"
"class M(BaseModel, frozen=False):\n"
" model_config = ConfigDict(frozen=True)\n"
)
assert "LIT015" in _codes(tmp_path, src)
def test_frozen_ok_with_reason_suppresses_and_is_not_lit013(tmp_path):
src = (
"from pydantic import BaseModel\n"
"class P(BaseModel): # frozen-ok: mutated during build before handoff\n"
" a: int\n"
)
codes = _codes(tmp_path, src)
assert "LIT015" not in codes
assert "LIT013" not in codes
def test_frozen_ok_on_frozen_model_is_lit013(tmp_path):
src = (
"from pydantic import BaseModel\n"
"class P(BaseModel, frozen=True): # frozen-ok: mutable during construction\n"
" a: int\n"
)
assert _codes(tmp_path, src) == ["LIT013"]
def test_frozen_ok_without_reason_is_lit005_and_does_not_suppress(tmp_path):
src = "from pydantic import BaseModel\nclass P(BaseModel): # frozen-ok\n a: int\n"
codes = _codes(tmp_path, src)
assert "LIT005" in codes
assert "LIT015" in codes
def test_typeddict_and_plain_classes_are_not_models(tmp_path):
src = "from typing import TypedDict\nclass T(TypedDict):\n a: int\nclass C:\n a: int\n"
assert "LIT015" not in _codes(tmp_path, src)
def test_extra_allow_does_not_exempt(tmp_path):
src = (
"from pydantic import BaseModel, ConfigDict\n"
"class P(BaseModel):\n"
" model_config = ConfigDict(extra='allow')\n"
)
assert "LIT015" in _codes(tmp_path, src)
def test_litellm_base_model_is_flagged(tmp_path):
src = "class Foo(LiteLLMBaseModel):\n x: int\n"
assert "LIT015" in _codes(tmp_path, src)
def test_litellm_openai_response_base_is_flagged(tmp_path):
src = "class R(BaseLiteLLMOpenAIResponseObject):\n x: int\n"
assert "LIT015" in _codes(tmp_path, src)
def test_base_settings_is_flagged(tmp_path):
src = "class S(BaseSettings):\n x: int\n"
assert "LIT015" in _codes(tmp_path, src)
def test_settings_config_dict_frozen_true_is_clean(tmp_path):
src = (
"from pydantic_settings import BaseSettings, SettingsConfigDict\n"
"class S(BaseSettings):\n"
" model_config = SettingsConfigDict(frozen=True)\n"
)
assert "LIT015" not in _codes(tmp_path, src)
def test_openai_object_base_is_flagged_and_can_be_frozen(tmp_path):
assert "LIT015" in _codes(tmp_path, "class Foo(OpenAIObject):\n x: int\n")
assert "LIT015" not in _codes(
tmp_path,
"from pydantic import ConfigDict\n"
"class Foo(OpenAIObject):\n"
" model_config = ConfigDict(frozen=True)\n"
" x: int\n",
)
def test_same_named_models_use_each_classes_own_override(tmp_path):
src = (
"from pydantic import BaseModel\n"
"class Dup(BaseModel, frozen=True):\n"
" x: int\n"
"class Dup(BaseModel):\n"
" x: int\n"
)
assert _codes(tmp_path, src) == ["LIT015"]
def test_pydantic_model_discovery_uses_each_class_nodes_bases(tmp_path):
src: Final = (
"from pydantic import BaseModel\n"
"class Config(BaseModel):\n"
" x: int\n"
"class Consumer(BaseModel, frozen=True):\n"
" class Config:\n"
" arbitrary_types_allowed = True\n"
)
path: Final = tmp_path / "snippet.py"
path.write_text(src, encoding="utf-8")
violations: Final = checker.check_file(path)
assert [(violation.line, violation.code) for violation in violations] == [(2, "LIT015")]
# --------------------------------------------------------------------------- #
# Stacked comprehension clauses (LIT014)
# --------------------------------------------------------------------------- #
@ -617,23 +889,12 @@ def test_comprehension_ok_with_reason_suppresses_lit014(tmp_path: Path):
def test_comprehension_ok_on_any_spanned_line_suppresses_lit014(tmp_path: Path):
src = (
"y = [\n"
" x for a in xs\n"
" for x in a\n"
"] # comprehension-ok: cartesian product is the clearest form\n"
)
src = "y = [\n x for a in xs\n for x in a\n] # comprehension-ok: cartesian product is the clearest form\n"
assert "LIT014" not in _codes(tmp_path, src)
def test_comprehension_ok_after_the_closing_line_does_not_suppress(tmp_path: Path):
src = (
"y = [\n"
" x for a in xs\n"
" for x in a\n"
"]\n"
"# comprehension-ok: cartesian product is the clearest form\n"
)
src = "y = [\n x for a in xs\n for x in a\n]\n# comprehension-ok: cartesian product is the clearest form\n"
f = tmp_path / "snippet.py"
f.write_text(src, encoding="utf-8")
violations = checker.check_file(f)

1
tests/unit/test_utils.py
View file

@ -1064,6 +1064,7 @@ def test_aaamodel_prices_and_context_window_json_is_valid():
"/vertex_ai/live",
"/v1/listen",
"/v1/systemone",
"/v1/decisions",
"/v1beta/interactions",
],
},

16
ui/litellm-dashboard/src/app/(dashboard)/playground/components/systemOneUI/SystemOneUI.integration.test.tsx
View file

@ -75,7 +75,7 @@ describe("SystemOneUI integration", () => {
render(<SystemOneUI accessToken="session-key" />);
screen.getByRole("combobox", { name: "Decision endpoint" }).focus();
await user.keyboard("{ArrowDown}");
await user.click(await screen.findByRole("option", { name: "Decisions · /v1/decisions" }));
await user.click(await screen.findByRole("option", { name: "System One · /v1/systemone" }));
expect(screen.getByRole("note", { name: "Decision endpoint notice" })).toHaveTextContent(
"omit model to use the proxy's configured default.",
@ -162,7 +162,7 @@ describe("SystemOneUI integration", () => {
render(<SystemOneUI accessToken="session-key" />);
screen.getByRole("combobox", { name: "Decision endpoint" }).focus();
await user.keyboard("{ArrowDown}");
await user.click(await screen.findByRole("option", { name: "Decisions · /v1/decisions" }));
await user.click(await screen.findByRole("option", { name: "System One · /v1/systemone" }));
const editor = screen.getByRole("textbox", { name: "System One JSON payload" });
const draft = JSON.stringify({
model: "my-decider",
@ -185,12 +185,12 @@ describe("SystemOneUI integration", () => {
screen.getByRole("combobox", { name: "Decision endpoint" }).focus();
await user.keyboard("{ArrowDown}");
await user.click(await screen.findByRole("option", { name: "Decisions · /v1/decisions" }));
await user.click(await screen.findByRole("option", { name: "System One · /v1/systemone" }));
expect(editor).toHaveValue(draft);
expect(screen.queryByText("Selected choice")).not.toBeInTheDocument();
await user.click(screen.getByRole("button", { name: "Send" }));
expect(await screen.findByText("Selected choice")).toBeInTheDocument();
expect(mockFetch.mock.calls[1]?.[0]).toMatch(/\/v1\/decisions$/);
expect(mockFetch.mock.calls[1]?.[0]).toMatch(/\/v1\/systemone$/);
expect(JSON.parse(mockFetch.mock.calls[1]?.[1]?.body as string)).toEqual(JSON.parse(draft));
});
@ -199,7 +199,7 @@ describe("SystemOneUI integration", () => {
render(<SystemOneUI accessToken="session-key" />);
screen.getByRole("combobox", { name: "Decision endpoint" }).focus();
await user.keyboard("{ArrowDown}");
await user.click(await screen.findByRole("option", { name: "Decisions · /v1/decisions" }));
await user.click(await screen.findByRole("option", { name: "System One · /v1/systemone" }));
const payload = { state: "An outage", questions: { urgent: { type: "noul", instructions: "Is this urgent?" } } };
fireEvent.change(screen.getByRole("textbox", { name: "System One JSON payload" }), {
target: { value: JSON.stringify(payload) },
@ -207,7 +207,7 @@ describe("SystemOneUI integration", () => {
expect(screen.getByRole("button", { name: "Send" })).toBeEnabled();
await user.click(screen.getByRole("button", { name: "Send" }));
expect(await screen.findByText("jev-1.13.0")).toBeInTheDocument();
expect(mockFetch.mock.calls[0]?.[0]).toMatch(/\/v1\/decisions$/);
expect(mockFetch.mock.calls[0]?.[0]).toMatch(/\/v1\/systemone$/);
const body = JSON.parse(mockFetch.mock.calls[0]?.[1]?.body as string);
expect(body).toEqual(payload);
expect(body).not.toHaveProperty("model");
@ -227,7 +227,7 @@ describe("SystemOneUI integration", () => {
await screen.findByRole("button", { name: "Cancel request" });
screen.getByRole("combobox", { name: "Decision endpoint" }).focus();
await user.keyboard("{ArrowDown}");
await user.click(await screen.findByRole("option", { name: "Decisions · /v1/decisions" }));
await user.click(await screen.findByRole("option", { name: "System One · /v1/systemone" }));
expect(mockFetch.mock.calls[0]?.[1]?.signal?.aborted).toBe(true);
expect(screen.getByRole("button", { name: "Send" })).toBeEnabled();
@ -249,7 +249,7 @@ describe("SystemOneUI integration", () => {
await user.click(screen.getByRole("button", { name: "Send" }));
expect(await screen.findByText("jev-1.13.0")).toBeInTheDocument();
expect(screen.getByText("Selected choice")).toBeInTheDocument();
expect(mockFetch.mock.calls[1]?.[0]).toMatch(/\/v1\/decisions$/);
expect(mockFetch.mock.calls[1]?.[0]).toMatch(/\/v1\/systemone$/);
expect(mockFetch).toHaveBeenCalledTimes(2);
});

16
ui/litellm-dashboard/src/app/(dashboard)/playground/components/systemOneUI/SystemOneUI.tsx
View file

@ -42,11 +42,11 @@ export default function SystemOneUI({ accessToken, disabledPersonalKeyCreation =
const [customApiKey, setCustomApiKey] = useState("");
const [endpoint, setEndpoint] = useState<DecisionEndpoint>("/typesafe/v1/systemone");
const [payloads, setPayloads] = useState<Record<DecisionEndpoint, string>>({
"/v1/decisions": DECISIONS_EXAMPLE_PAYLOAD,
"/v1/systemone": DECISIONS_EXAMPLE_PAYLOAD,
"/typesafe/v1/systemone": EXAMPLE_PAYLOAD,
});
const rawPayload = payloads[endpoint];
const examplePayload = endpoint === "/v1/decisions" ? DECISIONS_EXAMPLE_PAYLOAD : EXAMPLE_PAYLOAD;
const examplePayload = endpoint === "/v1/systemone" ? DECISIONS_EXAMPLE_PAYLOAD : EXAMPLE_PAYLOAD;
const activeController = useRef<AbortController | null>(null);
const validation = useMemo(() => validateSystemOnePayload(rawPayload, endpoint), [rawPayload, endpoint]);
const effectiveApiKey = apiKeySource === "session" ? accessToken || "" : customApiKey.trim();
@ -105,7 +105,7 @@ export default function SystemOneUI({ accessToken, disabledPersonalKeyCreation =
<Select
value={endpoint}
onValueChange={(value) => {
if (value === "/v1/decisions" || value === "/typesafe/v1/systemone") {
if (value === "/v1/systemone" || value === "/typesafe/v1/systemone") {
clearRequestState();
setEndpoint(value);
}
@ -113,11 +113,11 @@ export default function SystemOneUI({ accessToken, disabledPersonalKeyCreation =
>
<SelectTrigger className="w-80" aria-label="Decision endpoint">
<SelectValue>
{endpoint === "/v1/decisions" ? "Decisions · /v1/decisions" : "TypeSafe · /typesafe/v1/systemone"}
{endpoint === "/v1/systemone" ? "System One · /v1/systemone" : "TypeSafe · /typesafe/v1/systemone"}
</SelectValue>
</SelectTrigger>
<SelectContent>
<SelectItem value="/v1/decisions">Decisions · /v1/decisions</SelectItem>
<SelectItem value="/v1/systemone">System One · /v1/systemone</SelectItem>
<SelectItem value="/typesafe/v1/systemone">TypeSafe · /typesafe/v1/systemone</SelectItem>
</SelectContent>
</Select>
@ -182,11 +182,11 @@ export default function SystemOneUI({ accessToken, disabledPersonalKeyCreation =
<Alert role="note" aria-label="Decision endpoint notice">
<Info />
<AlertTitle>
{endpoint === "/v1/decisions" ? "Decision models · Jev format" : "TypeSafe Jev · System One"}
{endpoint === "/v1/systemone" ? "Decision models · System One" : "TypeSafe Jev · System One"}
</AlertTitle>
<AlertDescription>
{endpoint === "/v1/decisions"
? "Sends choice, noul, and score questions through /v1/decisions. Replace the example model with a decision model configured on your proxy, or omit model to use the proxy's configured default."
{endpoint === "/v1/systemone"
? "Sends choice, noul, and score questions through /v1/systemone. Replace the example model with a decision model configured on your proxy, or omit model to use the proxy's configured default."
: "Sends requests through /typesafe/v1/systemone and requires TYPESAFE_API_KEY on the proxy."}{" "}
<a href={DECISION_MODELS_DISCUSSION_URL} target="_blank" rel="noopener noreferrer" className="underline">
Give us feedback on what you want for decision models

2
ui/litellm-dashboard/src/app/(dashboard)/playground/components/systemOneUI/lib/decisions.test.ts
View file

@ -12,7 +12,7 @@ const request = {
},
provider_option: { enabled: true },
};
const validate = (value: unknown) => validateSystemOnePayload(JSON.stringify(value), "/v1/decisions");
const validate = (value: unknown) => validateSystemOnePayload(JSON.stringify(value), "/v1/systemone");
describe("native decisions validation", () => {
it("accepts structured Jev criteria, optional instructions, and provider extensions without dropping fields", () => {

2
ui/litellm-dashboard/src/app/(dashboard)/playground/components/systemOneUI/lib/schemas.ts
View file

@ -1,6 +1,6 @@
import { z } from "zod";
export type DecisionEndpoint = "/v1/decisions" | "/typesafe/v1/systemone";
export type DecisionEndpoint = "/v1/systemone" | "/typesafe/v1/systemone";
const decisionsJson = z.union([z.string(), z.record(z.string(), z.unknown()), z.array(z.unknown())]);
const decisionInstructions = decisionsJson.nullish();

2
ui/litellm-dashboard/src/app/(dashboard)/playground/components/systemOneUI/lib/validatePayload.ts
View file

@ -54,7 +54,7 @@ export function validateSystemOnePayload(
return invalid("syntax", `Invalid JSON syntax: ${json.message}`);
}
const schema = endpoint === "/v1/decisions" ? decisionsRequestSchema : systemOneRequestSchema;
const schema = endpoint === "/v1/systemone" ? decisionsRequestSchema : systemOneRequestSchema;
const result = schema.safeParse(json.value);
if (!result.success) {
return {

74
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

@ -16321,6 +16321,23 @@ export interface paths {
patch?: never;
trace?: never;
};
"/systemone": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/** Systemone */
post: operations["systemone_systemone_post"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/tag/daily/activity": {
parameters: {
query?: never;
@ -22434,6 +22451,23 @@ export interface paths {
patch?: never;
trace?: never;
};
"/v1/systemone": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/** Systemone */
post: operations["systemone_v1_systemone_post"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/v1/threads": {
parameters: {
query?: never;
@ -73321,6 +73355,26 @@ export interface operations {
};
};
};
systemone_systemone_post: {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": unknown;
};
};
};
};
get_tag_daily_activity_tag_daily_activity_get: {
parameters: {
query?: {
@ -81595,6 +81649,26 @@ export interface operations {
};
};
};
systemone_v1_systemone_post: {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": unknown;
};
};
};
};
create_threads_v1_threads_post: {
parameters: {
query?: never;