scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-29 06:33:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 267

feat(decisions): add Nadir as a Decisions provider for the auto router

Browse source 
Since the Add Auto Router form picks a deployed decision model (#45913),
a provider has to be a Decisions provider to appear there. Nadir now is:
`nadir/bucket` is a cost-map decision model, NadirDecisionsConfig serves
it from POST /v1/bucket, and a Nadir deployment answers /v1/decisions,
/v1/systemone, the health check and the auto router's Decision model.

Nadir is a tier classifier, not a general judge, so it answers exactly
one choice question whose choices include simple, medium and complex
(matched case-insensitively, answered on the question's own keys) and
refuses any other question set with a 400 before a call. A state shaped
{"messages": [...]} or OpenAI-format input goes out as the conversation,
any other state as the prompt.

When the auto router's decision deployment is Nadir it sends the
budgeted, role-tagged conversation as that state rather than the
LLM-judge payload, the same input `provider: nadir` already gets.

The `provider: nadir` path now uses the same config for its URL, key
scoping and verdict parsing, so litellm/llms/nadir/common_utils.py is
gone. NADIR_API_KEY still only goes to NADIR_API_BASE or the hosted API.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
dor 2026-10-21 11:25:49 -04:00
parent 78c9a8bc31
commit db6c70e9ee
15 changed files with 550 additions and 199 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
litellm/__init__.py
View file

@ -1730,6 +1730,9 @@ if TYPE_CHECKING:
from .llms.bespoke.decisions.transformation import (
BespokeDecisionsConfig as BespokeDecisionsConfig,
)
from .llms.nadir.decisions.transformation import (
NadirDecisionsConfig as NadirDecisionsConfig,
)
from .llms.openai.decisions.transformation import (
OpenAIDecisionsConfig as OpenAIDecisionsConfig,
)

2
litellm/_lazy_imports_registry.py
View file

@ -172,6 +172,7 @@ LLM_CONFIG_NAMES: Final = (
"HostedVLLMDecisionsConfig",
"LayaDecisionsConfig",
"BespokeDecisionsConfig",
"NadirDecisionsConfig",
"OpenAIDecisionsConfig",
"AzureAIDecisionsConfig",
"NvidiaNimRerankConfig",
@ -748,6 +749,7 @@ _LLM_CONFIGS_IMPORT_MAP: Final = {
),
"LayaDecisionsConfig": (".llms.laya.decisions.transformation", "LayaDecisionsConfig"),
"BespokeDecisionsConfig": (".llms.bespoke.decisions.transformation", "BespokeDecisionsConfig"),
"NadirDecisionsConfig": (".llms.nadir.decisions.transformation", "NadirDecisionsConfig"),
"OpenAIDecisionsConfig": (".llms.openai.decisions.transformation", "OpenAIDecisionsConfig"),
"AzureAIDecisionsConfig": (".llms.azure_ai.decisions.transformation", "AzureAIDecisionsConfig"),
"NvidiaNimRerankConfig": (

87
litellm/llms/nadir/common_utils.py
View file

@ -1,87 +0,0 @@
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from types import MappingProxyType
from typing import Annotated, Final, Literal, TypeAlias
from pydantic import ConfigDict, Field, TypeAdapter
from litellm.constants import NADIR_DEFAULT_API_BASE
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.base import LiteLLMBaseModel
NADIR_CLASSIFIER_MODEL: Final = "bucket"
_BUCKET_PATH: Final = "/v1/bucket"
_REQUEST_SOURCE: Final = "litellm"
NadirTier: TypeAlias = Literal["simple", "medium", "complex"]
_Probability: TypeAlias = Annotated[float, Field(ge=0.0, le=1.0)]
class _NadirBucketVerdict(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True, allow_inf_nan=False)
routing_tier: NadirTier | None = None
bucket: NadirTier | None = None
confidence: _Probability | None = None
probabilities: Mapping[str, _Probability] = Field(default_factory=lambda: MappingProxyType({}))
_VERDICT_ADAPTER: Final = TypeAdapter(_NadirBucketVerdict)
@dataclass(frozen=True, slots=True)
class NadirClassifierConnection:
api_base: str
api_key: str | None = field(repr=False)
def nadir_classifier_connection(api_base: str | None, api_key: str | None) -> NadirClassifierConnection:
"""Resolve where the classifier calls go, with the chat provider's key scoping.
``NADIR_API_KEY`` is only sent to ``NADIR_API_BASE`` or the hosted API. With no key at all the
call is anonymous, which Nadir serves on a per-IP rate-limited tier.
"""
default_base: Final = get_secret_str("NADIR_API_BASE") or NADIR_DEFAULT_API_BASE
trusted_base: Final = api_base is None or api_base.rstrip("/") == default_base.rstrip("/")
return NadirClassifierConnection(
api_base=(api_base or default_base).rstrip("/"),
api_key=api_key or (get_secret_str("NADIR_API_KEY") if trusted_base else None),
)
def nadir_bucket_url(api_base: str) -> str:
# Nadir documents its base with the /v1 suffix for OpenAI-compatible clients.
return f"{api_base.rstrip('/').removesuffix('/v1')}{_BUCKET_PATH}"
def nadir_bucket_request(state: str, messages: Sequence[Mapping[str, str]] | None = None) -> Mapping[str, object]:
"""Nadir's classifier reads the conversation itself, so it gets the messages rather than a judge prompt.
Plain dicts, because httpx serializes ``json=`` with json.dumps, which rejects MappingProxyType.
"""
if messages:
return {"messages": [dict(message) for message in messages], "source": _REQUEST_SOURCE}
return {"prompt": state, "source": _REQUEST_SOURCE}
def nadir_classifier_response(body: Mapping[str, object], requested_model: str) -> Mapping[str, object]:
"""Translate a ``/v1/bucket`` verdict into the System One ``tier`` choice answer the router reads.
``routing_tier`` is the tier Nadir itself routes on (the bucket after its confidence and gate
adjustments), so it wins over ``bucket``. Labels are the canonical tier names, which the router
resolves whatever display labels it uses.
"""
verdict: Final = _VERDICT_ADAPTER.validate_python(body)
tier: Final = verdict.routing_tier or verdict.bucket
if tier is None:
raise ValueError("Nadir response names neither routing_tier nor bucket")
confidence: Final = verdict.confidence if verdict.confidence is not None else verdict.probabilities.get(tier, 0.0)
answer: Final = MappingProxyType(
{
"type": "choice",
"choice": tier.upper(),
"probabilities": MappingProxyType({name.upper(): p for name, p in verdict.probabilities.items()}),
"confidence": confidence,
}
)
return MappingProxyType({"model": requested_model, "answers": MappingProxyType({"tier": answer})})

205
litellm/llms/nadir/decisions/transformation.py Normal file
View file

@ -0,0 +1,205 @@
import json
from collections.abc import Mapping, Sequence
from types import MappingProxyType
from typing import Annotated, Final, Literal, TypeAlias
from pydantic import ConfigDict, Field, TypeAdapter, ValidationError, model_validator
from pydantic_core import to_jsonable_python
from typing_extensions import assert_never
from litellm.constants import NADIR_DEFAULT_API_BASE
from litellm.llms.base_llm.decisions.transformation import (
BaseDecisionsConfig,
parse_systemone_response,
systemone_keys,
)
from litellm.secret_managers.main import get_secret_str
from litellm.types.decisions import (
DecisionsIRChoiceQuestion,
DecisionsIRMessages,
DecisionsIRRequest,
DecisionsIRResponse,
DecisionsIRState,
DecisionsJSON,
OpenAIDecisionContentPart,
OpenAIDecisionInputText,
UnsupportedDecisionsRequest,
)
from litellm.types.llms.base import LiteLLMBaseModel
NADIR_CLASSIFIER_MODEL: Final = "bucket"
_REQUEST_SOURCE: Final = "litellm"
NadirTier: TypeAlias = Literal["simple", "medium", "complex"]
_TIERS: Final[tuple[NadirTier, ...]] = ("simple", "medium", "complex")
_CANONICAL_CHOICES: Final[Mapping[NadirTier, str]] = MappingProxyType({tier: tier.upper() for tier in _TIERS})
_Probability: TypeAlias = Annotated[float, Field(ge=0.0, le=1.0)]
_MAPPING_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object])
_UNSUPPORTED_QUESTIONS: Final = UnsupportedDecisionsRequest(
reason="Nadir answers exactly one choice question whose choices include simple, medium and complex"
)
class _NadirBucketVerdict(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True, allow_inf_nan=False)
routing_tier: NadirTier
confidence: _Probability | None = None
probabilities: Mapping[str, _Probability] = Field(default_factory=lambda: MappingProxyType({}))
@model_validator(mode="before")
@classmethod
def _routing_tier_falls_back_to_bucket(cls, data: object) -> object:
"""``routing_tier`` is the tier Nadir itself routes on (the bucket after its confidence and gate
adjustments), so it wins; an older server that sends only ``bucket`` still answers."""
try:
fields: Final = _MAPPING_ADAPTER.validate_python(data)
except ValidationError:
return data
return {**fields, "routing_tier": fields.get("routing_tier") or fields.get("bucket")}
_VERDICT_ADAPTER: Final = TypeAdapter(_NadirBucketVerdict)
class _ConversationMessage(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
role: Literal["system", "user", "assistant"]
content: str
class _ConversationState(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
messages: tuple[_ConversationMessage, ...] = Field(min_length=1)
_CONVERSATION_ADAPTER: Final = TypeAdapter(_ConversationState)
def _part_text(content: str | Sequence[OpenAIDecisionContentPart]) -> str:
if isinstance(content, str):
return content
return "\n".join(part.text for part in content if isinstance(part, OpenAIDecisionInputText))
def _state_body(state: DecisionsJSON) -> Mapping[str, object]:
if isinstance(state, str):
return {"prompt": state}
try:
conversation: Final = _CONVERSATION_ADAPTER.validate_python(state)
except ValidationError:
return {"prompt": json.dumps(to_jsonable_python(state))}
return {"messages": [message.model_dump() for message in conversation.messages]}
def _input_body(decision_input: DecisionsIRState | DecisionsIRMessages) -> Mapping[str, object]:
match decision_input:
case DecisionsIRMessages(messages=messages):
return {
"messages": [{"role": message.role, "content": _part_text(message.content)} for message in messages]
}
case DecisionsIRState(state=state):
return _state_body(state)
case _:
assert_never(decision_input)
def nadir_bucket_body(decision_input: DecisionsIRState | DecisionsIRMessages) -> Mapping[str, object]:
"""The ``/v1/bucket`` body for a decision input.
Nadir's classifier is trained on raw conversations, so a state shaped ``{"messages": [{role, content}]}``
goes out as those messages, as do OpenAI-format input messages; any other state is the prompt. Plain
dicts and lists, because httpx serializes ``json=`` with json.dumps, which rejects MappingProxyType.
"""
return {**_input_body(decision_input), "source": _REQUEST_SOURCE}
def _tier_choices(request: DecisionsIRRequest) -> Mapping[NadirTier, str] | None:
"""The question's own choice key for each Nadir tier, matched case-insensitively, when the request is
one choice question naming all three tiers."""
match request.questions:
case (DecisionsIRChoiceQuestion(choices=choices),):
by_name: Final = {option.value.lower(): option.value for option in choices if isinstance(option.value, str)}
by_tier: Final[Mapping[NadirTier, str]] = {tier: by_name[tier] for tier in _TIERS if tier in by_name}
return MappingProxyType(by_tier) if len(by_tier) == len(_TIERS) else None
case _:
return None
def _choice_answer(verdict: _NadirBucketVerdict, choices: Mapping[NadirTier, str]) -> Mapping[str, object]:
tier: Final = verdict.routing_tier
return {
"type": "choice",
"choice": choices[tier],
"probabilities": {choices[name]: verdict.probabilities.get(name, 0.0) for name in _TIERS},
"confidence": verdict.confidence if verdict.confidence is not None else verdict.probabilities.get(tier, 0.0),
}
class NadirDecisionsConfig(BaseDecisionsConfig):
"""Nadir as a Decisions provider: ``/v1/bucket`` answers a single simple/medium/complex choice question.
Nadir is a tier classifier, not a general judge, so any other question set is rejected before a call.
``NADIR_API_KEY`` is only sent to ``NADIR_API_BASE`` or the hosted API, the same scoping as the ``nadir``
chat provider; with no key the call is anonymous, which Nadir serves on a per-IP rate-limited tier.
"""
path = "/v1/bucket"
api_key_env = ("NADIR_API_KEY",)
api_base_env = ("NADIR_API_BASE",)
api_key_required = False
health_check_questions = MappingProxyType(
{
"tier": MappingProxyType(
{
"type": "choice",
"instructions": "Which tier should serve this request?",
"criteria": {tier: None for tier in _TIERS},
}
)
}
)
def get_default_api_base(self) -> str:
return NADIR_DEFAULT_API_BASE
def resolve_connection(self, api_base: str | None, api_key: str | None) -> tuple[str, str | None]:
default_base: Final = get_secret_str("NADIR_API_BASE") or NADIR_DEFAULT_API_BASE
trusted_base: Final = api_base is None or api_base.rstrip("/") == default_base.rstrip("/")
return (
(api_base or default_base).rstrip("/"),
api_key or (get_secret_str("NADIR_API_KEY") if trusted_base else None),
)
def transform_decisions_request(
self,
model: str,
request: DecisionsIRRequest,
custom_llm_provider: str,
) -> Mapping[str, object] | UnsupportedDecisionsRequest:
if _tier_choices(request) is None:
return _UNSUPPORTED_QUESTIONS
return nadir_bucket_body(request.input)
def parse_response(self, payload: object, request: DecisionsIRRequest) -> DecisionsIRResponse:
verdict: Final = _VERDICT_ADAPTER.validate_python(payload)
choices: Final = _tier_choices(request)
answers: Final = (
{}
if choices is None
else {key: _choice_answer(verdict, choices) for key in systemone_keys(request.questions)}
)
return parse_systemone_response({"answers": answers}, request)
def classifier_response(self, body: Mapping[str, object], requested_model: str) -> Mapping[str, object]:
"""The verdict as the System One ``tier`` answer the auto router reads, on canonical tier labels the
router resolves whatever display labels it uses."""
verdict: Final = _VERDICT_ADAPTER.validate_python(body)
return MappingProxyType(
{"model": requested_model, "answers": {"tier": _choice_answer(verdict, _CANONICAL_CHOICES)}}
)
NADIR_DECISIONS_CONFIG: Final[NadirDecisionsConfig] = NadirDecisionsConfig()

10
litellm/model_prices_and_context_window_backup.json
View file

@ -72192,6 +72192,16 @@
"notes": "Self-hosted decision model; infrastructure costs are paid separately"
}
},
"nadir/bucket": {
"input_cost_per_token": 0.0,
"litellm_provider": "nadir",
"mode": "evaluation",
"output_cost_per_token": 0.0,
"source": "https://getnadir.com/docs/decision-api/",
"metadata": {
"notes": "Nadir's simple, medium or complex tier decision from POST /v1/bucket, which Nadir does not bill"
}
},
"strands_decider/strands-decider-2B-hobson-v19": {
"input_cost_per_token": 0.0,
"litellm_provider": "strands_decider",

25
litellm/router_strategy/complexity_router/complexity_router.py
View file

@ -61,7 +61,7 @@ from litellm.llms.anthropic.common_utils import is_claude_code_user_agent
from litellm.llms.base_llm.base_utils import type_to_response_format_param
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
from litellm.llms.databricks.decisions.transformation import DATABRICKS_DECISIONS_CONFIG
from litellm.llms.nadir.common_utils import nadir_classifier_connection
from litellm.llms.nadir.decisions.transformation import NADIR_DECISIONS_CONFIG
from litellm.router_strategy.adaptive_router.classifier import classify_prompt
from litellm.router_strategy.complexity_router.context_compaction import compaction_pending
from litellm.router_strategy.complexity_router.tier_predictor import (
@ -888,6 +888,13 @@ def _truncate(text: str, limit: int) -> str:
return f"{text[:head_chars]}{_TRUNCATION_MARKER}{text[len(text) - tail_chars :]}"
def _is_nadir_deployment(deployment: Mapping[str, object]) -> bool:
litellm_params: Final = deployment.get("litellm_params")
params: Final = litellm_params if isinstance(litellm_params, Mapping) else EMPTY_MAPPING
provider: Final = params.get("custom_llm_provider") or str(params.get("model") or "").split("/", 1)[0]
return provider == "nadir"
def _iter_context_turns_newest_first(
messages: Sequence[Mapping[str, object]],
include_assistant: bool,
@ -1366,10 +1373,10 @@ class ComplexityRouter(CustomLogger):
provider="databricks",
)
if config.provider == "nadir":
nadir: Final = nadir_classifier_connection(config.api_base, config.api_key)
nadir_base, nadir_key = NADIR_DECISIONS_CONFIG.resolve_connection(config.api_base, config.api_key)
return HttpJevClassifierClient(
api_key=nadir.api_key,
api_base=nadir.api_base,
api_key=nadir_key,
api_base=nadir_base,
http_client=get_async_httpx_client(httpxSpecialProvider.PassThroughEndpoint),
provider="nadir",
)
@ -2328,7 +2335,7 @@ class ComplexityRouter(CustomLogger):
instructions=config.instructions or DEFAULT_JEV_INSTRUCTIONS,
criteria=criteria,
messages=self._classifier_context_messages(prompt, system_prompt, request_kwargs, messages)
if config.provider == "nadir"
if self._classifier_reads_conversation(config)
else None,
)
try:
@ -2577,6 +2584,14 @@ class ComplexityRouter(CustomLogger):
else system_prompt
)
def _classifier_reads_conversation(self, config: OpenSourceClassifierConfig) -> bool:
"""Nadir classifies the conversation itself, whether it is named as the provider or serves the
decision deployment, so it gets role-tagged messages in place of the LLM-judge payload."""
if config.deployment_name is None:
return config.provider == "nadir"
deployments: Final = self._group_deployments(config.deployment_name)
return bool(deployments) and all(_is_nadir_deployment(deployment) for deployment in deployments)
def _classifier_prior_turns(
self,
prompt: str,

4
litellm/router_strategy/complexity_router/config.py
View file

@ -745,7 +745,7 @@ class OpenSourceClassifierConfig(LiteLLMBaseModel):
fields: Final = TypeAdapter(Mapping[str, object]).validate_python(data)
if fields.get("provider") != "nadir" or "model" in fields:
return fields
from litellm.llms.nadir.common_utils import NADIR_CLASSIFIER_MODEL
from litellm.llms.nadir.decisions.transformation import NADIR_CLASSIFIER_MODEL
return {**fields, "model": NADIR_CLASSIFIER_MODEL}
@ -798,7 +798,7 @@ class OpenSourceClassifierConfig(LiteLLMBaseModel):
"opensource_classifier_config.instructions has no effect for provider 'nadir', "
"which classifies with its own model"
)
# nadir_classifier_connection keeps NADIR_API_KEY on NADIR_API_BASE; a keyless call is anonymous.
# NADIR_DECISIONS_CONFIG keeps NADIR_API_KEY on NADIR_API_BASE; a keyless call is anonymous.
return self
if self.api_base is not None and self.api_key is None:
raise ValueError(

23
litellm/router_strategy/complexity_router/jev_classifier.py
View file

@ -21,10 +21,10 @@ from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
from litellm.llms.databricks.decisions.transformation import DATABRICKS_DECISIONS_CONFIG, is_ai_decide_model
from litellm.llms.laya.common_utils import laya_response_model
from litellm.llms.nadir.common_utils import nadir_bucket_request, nadir_bucket_url, nadir_classifier_response
from litellm.llms.nadir.decisions.transformation import NADIR_DECISIONS_CONFIG, nadir_bucket_body
from litellm.llms.pass_through.typesafe_logging_handler import TypeSafePassthroughLoggingHandler
from litellm.router_strategy.complexity_router.config import DEFAULT_JEV_INSTRUCTIONS as _DEFAULT_JEV_INSTRUCTIONS
from litellm.types.decisions import DecisionsResponse
from litellm.types.decisions import DecisionsIRState, DecisionsJSON, DecisionsResponse
from litellm.types.llms.base import LiteLLMBaseModel
from litellm.types.utils import AUTOROUTER_CLASSIFIER_CALL_ORIGIN
@ -51,9 +51,13 @@ class JevSystemOneRequest(LiteLLMBaseModel):
state: str
model: str
questions: Mapping[str, JevChoiceQuestion]
# Native input for a classifier that reads the conversation itself (nadir); never sent to a System One server.
# The conversation, for a classifier that reads it itself (nadir); excluded from the System One body.
messages: tuple[Mapping[str, str], ...] | None = Field(default=None, exclude=True)
def decision_state(self) -> DecisionsJSON:
"""The state a decision call carries: the conversation when the classifier reads it, else the payload."""
return self.state if self.messages is None else {"messages": [dict(message) for message in self.messages]}
class JevChoiceAnswer(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True, allow_inf_nan=False)
@ -95,7 +99,7 @@ class _DecisionsRouter(Protocol):
self,
*,
model: str,
state: str,
state: DecisionsJSON,
questions: Mapping[str, object],
timeout: float,
num_retries: int,
@ -130,14 +134,15 @@ class RouterJevClassifierClient:
),
INTERNAL_CALL_ORIGIN_METADATA_KEY: AUTOROUTER_CLASSIFIER_CALL_ORIGIN,
}
state: Final = request.decision_state()
payload: Final = {
"state": request.state,
"state": state,
"questions": {key: value.model_dump() for key, value in request.questions.items()},
}
response: Final = DecisionsResponse.model_validate(
await self._router.adecisions(
model=self._deployment_name,
state=request.state,
state=state,
questions={key: value.model_dump() for key, value in request.questions.items()},
timeout=timeout_s,
num_retries=0,
@ -200,7 +205,7 @@ class HttpJevClassifierClient:
def _request_body(self, request: JevSystemOneRequest) -> Mapping[str, object]:
if self._provider == "nadir":
return nadir_bucket_request(request.state, request.messages)
return nadir_bucket_body(DecisionsIRState(state=request.decision_state()))
return request.model_dump(mode="json", exclude=self._excluded_fields(request.model))
def _excluded_fields(self, model: str) -> Mapping[str, bool] | None:
@ -210,7 +215,7 @@ class HttpJevClassifierClient:
if self._provider == "databricks":
return DATABRICKS_DECISIONS_CONFIG.get_complete_url(self._api_base, model)
if self._provider == "nadir":
return nadir_bucket_url(self._api_base)
return NADIR_DECISIONS_CONFIG.get_complete_url(self._api_base, model)
return f"{self._api_base}/v1/systemone"
def _normalized_body(self, body: Mapping[str, object], requested_model: str) -> Mapping[str, object]:
@ -220,7 +225,7 @@ class HttpJevClassifierClient:
case "databricks":
return DATABRICKS_DECISIONS_CONFIG.classifier_response(body, requested_model)
case "nadir":
return nadir_classifier_response(body, requested_model)
return NADIR_DECISIONS_CONFIG.classifier_response(body, requested_model)
case "typesafe" | "bespoke":
return body

2
litellm/utils.py
View file

@ -9638,6 +9638,8 @@ class ProviderConfigManager:
return litellm.LayaDecisionsConfig()
if provider == LlmProviders.BESPOKE:
return litellm.BespokeDecisionsConfig()
if provider == LlmProviders.NADIR:
return litellm.NadirDecisionsConfig()
if provider == LlmProviders.OPENAI:
return litellm.OpenAIDecisionsConfig()
if provider == LlmProviders.AZURE_AI:

10
model_prices_and_context_window.json
View file

@ -72192,6 +72192,16 @@
"notes": "Self-hosted decision model; infrastructure costs are paid separately"
}
},
"nadir/bucket": {
"input_cost_per_token": 0.0,
"litellm_provider": "nadir",
"mode": "evaluation",
"output_cost_per_token": 0.0,
"source": "https://getnadir.com/docs/decision-api/",
"metadata": {
"notes": "Nadir's simple, medium or complex tier decision from POST /v1/bucket, which Nadir does not bill"
}
},
"strands_decider/strands-decider-2B-hobson-v19": {
"input_cost_per_token": 0.0,
"litellm_provider": "strands_decider",

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

@ -584,6 +584,45 @@ async def test_oss_classifier_env_key_goes_only_to_its_env_api_base(
assert "authorization" not in other_upstream.calls[0].request.headers
@pytest.mark.asyncio
async def test_nadir_answers_a_tier_question_from_its_bucket_verdict_and_keeps_its_key_home(
monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter
) -> None:
monkeypatch.delenv("NADIR_API_BASE", raising=False)
monkeypatch.setenv("NADIR_API_KEY", "ndr-env")
verdict: Final = {"routing_tier": "complex", "confidence": 0.9, "probabilities": {"simple": 0.02, "complex": 0.9}}
hosted: Final = respx_mock.post("https://api.getnadir.com/v1/bucket").respond(json=verdict)
elsewhere: Final = respx_mock.post("https://collector.example/v1/bucket").respond(json=verdict)
questions: Final = {"tier": {"type": "choice", "criteria": {"Simple": None, "Medium": None, "Complex": None}}}
response: Final = await litellm.asystemone(
model="nadir/bucket", state="Design a consensus protocol", questions=questions
)
_ = await litellm.asystemone(
model="nadir/bucket", state="hi", questions=questions, api_base="https://collector.example/v1"
)
assert response.answers["tier"] == ChoiceAnswer(
type="choice", choice="Complex", confidence=0.9, probabilities={"Simple": 0.02, "Medium": 0.0, "Complex": 0.9}
)
assert json.loads(hosted.calls.last.request.content) == {
"prompt": "Design a consensus protocol",
"source": "litellm",
}
assert hosted.calls.last.request.headers["authorization"] == "Bearer ndr-env"
assert "authorization" not in elsewhere.calls.last.request.headers
@pytest.mark.asyncio
async def test_nadir_refuses_a_question_it_cannot_answer_before_any_call(respx_mock: respx.MockRouter) -> None:
upstream: Final = respx_mock.post("https://api.getnadir.com/v1/bucket")
with pytest.raises(litellm.BadRequestError, match="cannot serve this request"):
_ = await litellm.asystemone(
model="nadir/bucket", state="review", questions={"is_defect": {"type": "noul", "instructions": "Defect?"}}
)
assert not upstream.called
@pytest.mark.asyncio
@pytest.mark.parametrize(
("model", "api_base", "error"),
@ -620,7 +659,9 @@ _PPLX_USAGE: Final = {"model": "pplx-decider-v1-27b", "usage": {"input_tokens":
("reply", "detail"),
(
pytest.param(_PPLX_USAGE, "DecisionsResponse answers: Field required", id="usage"),
pytest.param({"error": "Rate limit reached"}, "DecisionsResponse answers: Field required", id="rate_limit_text"),
pytest.param(
{"error": "Rate limit reached"}, "DecisionsResponse answers: Field required", id="rate_limit_text"
),
pytest.param({"error": "Request timed out"}, "DecisionsResponse answers: Field required", id="timeout_text"),
pytest.param(
{**_PPLX_USAGE, "answers": {"rate_limit": None}},
@ -632,7 +673,9 @@ _PPLX_USAGE: Final = {"model": "pplx-decider-v1-27b", "usage": {"input_tokens":
"DecisionsResponse answers.is_defect",
id="answer_type_timeout_text",
),
pytest.param(["Rate limit reached"], "DecisionsResponse Input should be a valid dictionary", id="not_an_object"),
pytest.param(
["Rate limit reached"], "DecisionsResponse Input should be a valid dictionary", id="not_an_object"
),
),
)
async def test_a_malformed_upstream_reply_is_a_server_error_whatever_its_text(
@ -3729,7 +3772,12 @@ async def test_a_returned_model_registered_without_prices_is_billed_at_the_reque
monkeypatch.setitem(
litellm.model_cost,
"typesafe/jev-test-alias",
{"litellm_provider": "typesafe", "mode": "evaluation", "input_cost_per_token": 0.001, "output_cost_per_token": 0},
{
"litellm_provider": "typesafe",
"mode": "evaluation",
"input_cost_per_token": 0.001,
"output_cost_per_token": 0,
},
)
monkeypatch.setitem(litellm.model_cost, "typesafe/jev-test-unpriced", unpriced_row)
respx_mock.post("https://api.typesafe.ai/v1/systemone").respond(json={**_RESPONSE, "model": "jev-test-unpriced"})

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

185
tests/unit/llms/nadir/decisions/test_transformation.py Normal file
View file

@ -0,0 +1,185 @@
from typing import Final
import pytest
from pydantic import ValidationError
from litellm.llms.base_llm.decisions.transformation import systemone_request_to_ir
from litellm.llms.nadir.decisions.transformation import NADIR_DECISIONS_CONFIG, nadir_bucket_body
from litellm.types.decisions import (
DecisionsIRChoiceAnswer,
DecisionsIRChoiceProbability,
DecisionsIRMessages,
DecisionsIRRequest,
DecisionsIRState,
DecisionsRequestBody,
OpenAIDecisionInputMessage,
UnsupportedDecisionsRequest,
)
_TIER_QUESTION: Final = {
"tier": {
"type": "choice",
"instructions": "Which tier?",
"criteria": {"SIMPLE": "facts", "MEDIUM": "edits", "COMPLEX": "design", "REASONING": "proofs"},
}
}
_VERDICT: Final = {
"bucket": "medium",
"routing_tier": "complex",
"confidence": 0.72,
"probabilities": {"simple": 0.1, "medium": 0.6, "complex": 0.3},
"selected_model": None,
"decode": "complex_gate",
}
def _request(state: object = "What is 2+2?", questions: object = _TIER_QUESTION) -> DecisionsIRRequest:
return systemone_request_to_ir(DecisionsRequestBody.model_validate({"state": state, "questions": questions}))
def test_routing_tier_wins_over_bucket_and_answers_on_the_question_s_own_choice_keys() -> None:
parsed: Final = NADIR_DECISIONS_CONFIG.parse_response(_VERDICT, _request())
assert parsed.answers == (
DecisionsIRChoiceAnswer(
choice="COMPLEX",
confidence=0.72,
probabilities=(
DecisionsIRChoiceProbability(value="SIMPLE", probability=0.1),
DecisionsIRChoiceProbability(value="MEDIUM", probability=0.6),
DecisionsIRChoiceProbability(value="COMPLEX", probability=0.3),
),
),
)
def test_bucket_is_used_without_routing_tier_and_confidence_falls_back_to_its_probability() -> None:
lowercase: Final = _request(
questions={"t": {"type": "choice", "criteria": {"simple": None, "medium": None, "complex": None}}}
)
answer: Final = NADIR_DECISIONS_CONFIG.parse_response(
{"bucket": "simple", "confidence": None, "probabilities": {"simple": 0.8, "medium": 0.2}}, lowercase
).answers[0]
assert isinstance(answer, DecisionsIRChoiceAnswer)
assert (answer.choice, answer.confidence) == ("simple", 0.8)
def test_the_router_s_tier_answer_uses_canonical_labels() -> None:
translated: Final = NADIR_DECISIONS_CONFIG.classifier_response(_VERDICT, "bucket")
assert translated["model"] == "bucket"
assert translated["answers"] == {
"tier": {
"type": "choice",
"choice": "COMPLEX",
"probabilities": {"SIMPLE": 0.1, "MEDIUM": 0.6, "COMPLEX": 0.3},
"confidence": 0.72,
}
}
assert (
NADIR_DECISIONS_CONFIG.classifier_response({"routing_tier": "medium"}, "bucket")["answers"]["tier"][
"confidence"
]
== 0.0
)
@pytest.mark.parametrize(
"body",
[
{},
{"routing_tier": "reasoning"},
{"routing_tier": "simple", "confidence": 1.5},
{"routing_tier": "simple", "probabilities": {"simple": -0.1}},
],
)
def test_a_verdict_the_router_cannot_use_is_rejected(body: dict[str, object]) -> None:
with pytest.raises(ValidationError):
_ = NADIR_DECISIONS_CONFIG.classifier_response(body, "bucket")
with pytest.raises(ValidationError):
_ = NADIR_DECISIONS_CONFIG.parse_response(body, _request())
@pytest.mark.parametrize(
"questions",
[
{"q": {"type": "noul", "instructions": "Is this hard?"}},
{"q": {"type": "choice", "criteria": {"simple": None, "complex": None}}},
{"q": {"type": "choice", "criteria": {"easy": None, "hard": None, "harder": None}}},
{**_TIER_QUESTION, "second": {"type": "choice", "criteria": {"simple": None, "medium": None, "complex": None}}},
],
)
def test_anything_but_one_tier_choice_question_is_refused_before_a_call(questions: object) -> None:
refused: Final = NADIR_DECISIONS_CONFIG.transform_decisions_request(
"bucket", _request(questions=questions), "nadir"
)
assert isinstance(refused, UnsupportedDecisionsRequest)
assert "simple, medium and complex" in refused.reason
def test_the_health_check_asks_a_question_nadir_can_answer() -> None:
request: Final = _request(questions=NADIR_DECISIONS_CONFIG.health_check_questions)
assert not isinstance(
NADIR_DECISIONS_CONFIG.transform_decisions_request("bucket", request, "nadir"), UnsupportedDecisionsRequest
)
@pytest.mark.parametrize(
("decision_input", "expected"),
[
(DecisionsIRState(state="hi"), {"prompt": "hi"}),
(
DecisionsIRState(
state={"messages": [{"role": "system", "content": "Be brief"}, {"role": "user", "content": "hi"}]}
),
{"messages": [{"role": "system", "content": "Be brief"}, {"role": "user", "content": "hi"}]},
),
(DecisionsIRState(state={"messages": []}), {"prompt": '{"messages": []}'}),
(DecisionsIRState(state={"ticket": 7}), {"prompt": '{"ticket": 7}'}),
(
DecisionsIRMessages(
messages=(
OpenAIDecisionInputMessage(content="first"),
OpenAIDecisionInputMessage.model_validate(
{"content": [{"type": "input_text", "text": "a"}, {"type": "input_text", "text": "b"}]}
),
)
),
{"messages": [{"role": "user", "content": "first"}, {"role": "user", "content": "a\nb"}]},
),
],
)
def test_the_bucket_body_carries_the_conversation_when_there_is_one(
decision_input: DecisionsIRState | DecisionsIRMessages, expected: dict[str, object]
) -> None:
assert nadir_bucket_body(decision_input) == {**expected, "source": "litellm"}
@pytest.mark.parametrize(
"base",
["https://api.getnadir.com/v1", "https://api.getnadir.com/v1/", "https://api.getnadir.com", "https://n.test/"],
)
def test_bucket_url_accepts_the_documented_base_with_or_without_v1(base: str) -> None:
url: Final = NADIR_DECISIONS_CONFIG.get_complete_url(base, "bucket")
assert url.endswith("/v1/bucket")
assert "/v1/v1/" not in url
def test_environment_key_only_goes_to_the_environment_base(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("NADIR_API_BASE", raising=False)
monkeypatch.setenv("NADIR_API_KEY", "ndr-env")
connect: Final = NADIR_DECISIONS_CONFIG.resolve_connection
assert connect(None, None) == ("https://api.getnadir.com/v1", "ndr-env")
assert connect("https://api.getnadir.com/v1/", None) == ("https://api.getnadir.com/v1", "ndr-env")
assert connect("https://collector.invalid", None) == ("https://collector.invalid", None)
assert connect("https://nadir.internal", "ndr-own") == ("https://nadir.internal", "ndr-own")
monkeypatch.setenv("NADIR_API_BASE", "https://nadir.internal/v1")
assert connect(None, None) == ("https://nadir.internal/v1", "ndr-env")
assert connect("https://api.getnadir.com/v1", None) == ("https://api.getnadir.com/v1", None)
def test_no_key_anywhere_is_an_anonymous_call(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("NADIR_API_BASE", raising=False)
monkeypatch.delenv("NADIR_API_KEY", raising=False)
assert NADIR_DECISIONS_CONFIG.resolve_connection(None, None) == ("https://api.getnadir.com/v1", None)
assert not NADIR_DECISIONS_CONFIG.api_key_required

93
tests/unit/llms/nadir/test_common_utils.py
View file

@ -1,93 +0,0 @@
from typing import Final
import pytest
from pydantic import ValidationError
from litellm.llms.nadir.common_utils import (
nadir_bucket_request,
nadir_bucket_url,
nadir_classifier_connection,
nadir_classifier_response,
)
def test_routing_tier_wins_over_bucket_and_labels_are_canonical() -> None:
translated: Final = nadir_classifier_response(
{
"bucket": "medium",
"routing_tier": "complex",
"confidence": 0.72,
"probabilities": {"simple": 0.1, "medium": 0.6, "complex": 0.3},
"selected_model": None,
"decode": "complex_gate",
},
"bucket",
)
assert translated["model"] == "bucket"
assert dict(translated["answers"]["tier"]) == {
"type": "choice",
"choice": "COMPLEX",
"probabilities": {"SIMPLE": 0.1, "MEDIUM": 0.6, "COMPLEX": 0.3},
"confidence": 0.72,
}
def test_bucket_is_used_without_routing_tier_and_confidence_falls_back_to_its_probability() -> None:
answer: Final = nadir_classifier_response(
{"bucket": "simple", "confidence": None, "probabilities": {"simple": 0.8, "medium": 0.2}}, "bucket"
)["answers"]["tier"]
assert (answer["choice"], answer["confidence"]) == ("SIMPLE", 0.8)
assert nadir_classifier_response({"routing_tier": "medium"}, "bucket")["answers"]["tier"]["confidence"] == 0.0
@pytest.mark.parametrize(
"body",
[
{},
{"routing_tier": "reasoning"},
{"routing_tier": "simple", "confidence": 1.5},
{"routing_tier": "simple", "probabilities": {"simple": -0.1}},
],
)
def test_a_verdict_the_router_cannot_use_is_rejected(body: dict[str, object]) -> None:
with pytest.raises((ValueError, ValidationError)):
_ = nadir_classifier_response(body, "bucket")
@pytest.mark.parametrize(
"base",
["https://api.getnadir.com/v1", "https://api.getnadir.com/v1/", "https://api.getnadir.com", "https://n.test/"],
)
def test_bucket_url_accepts_the_documented_base_with_or_without_v1(base: str) -> None:
assert nadir_bucket_url(base).endswith("/v1/bucket")
assert "/v1/v1/" not in nadir_bucket_url(base)
def test_bucket_request_prefers_the_conversation_and_falls_back_to_the_state() -> None:
messages: Final = ({"role": "system", "content": "Be brief"}, {"role": "user", "content": "hi"})
assert dict(nadir_bucket_request("judge prompt", messages)) == {"messages": list(messages), "source": "litellm"}
assert dict(nadir_bucket_request("hi")) == {"prompt": "hi", "source": "litellm"}
def test_environment_key_only_goes_to_the_environment_base(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("NADIR_API_BASE", raising=False)
monkeypatch.setenv("NADIR_API_KEY", "ndr-env")
hosted: Final = nadir_classifier_connection(None, None)
assert (hosted.api_base, hosted.api_key) == ("https://api.getnadir.com/v1", "ndr-env")
assert nadir_classifier_connection("https://api.getnadir.com/v1/", None).api_key == "ndr-env"
elsewhere: Final = nadir_classifier_connection("https://collector.invalid", None)
assert (elsewhere.api_base, elsewhere.api_key) == ("https://collector.invalid", None)
assert nadir_classifier_connection("https://nadir.internal", "ndr-own").api_key == "ndr-own"
assert "ndr-env" not in repr(hosted)
monkeypatch.setenv("NADIR_API_BASE", "https://nadir.internal/v1")
assert nadir_classifier_connection(None, None).api_base == "https://nadir.internal/v1"
assert nadir_classifier_connection("https://api.getnadir.com/v1", None).api_key is None
def test_no_key_anywhere_is_an_anonymous_call(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("NADIR_API_BASE", raising=False)
monkeypatch.delenv("NADIR_API_KEY", raising=False)
assert nadir_classifier_connection(None, None).api_key is None

46
tests/unit/router_strategy/complexity_router/test_jev_classifier.py
View file

@ -883,6 +883,51 @@ async def test_nadir_gets_the_budgeted_conversation_without_the_judge_framing(
assert "Classify this message" not in route.calls.last.request.content.decode()
@pytest.mark.asyncio
async def test_a_nadir_decision_deployment_gets_the_conversation_and_routes_on_its_verdict(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("NADIR_API_BASE", raising=False)
monkeypatch.delenv("NADIR_API_KEY", raising=False)
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
router: Final = litellm.Router(
model_list=[
{"model_name": "nadir-decider", "litellm_params": {"model": "nadir/bucket", "api_key": "ndr-deployment"}}
],
num_retries=0,
)
classifier: Final = ComplexityRouter(
"decision-router",
router,
{
"classifier_type": "oss_classifier",
"opensource_classifier_config": {"deployment_name": "nadir-decider"},
"tiers": {"SIMPLE": "cheap", "MEDIUM": "mid", "COMPLEX": "strong", "REASONING": "top"},
},
derive_savings_baseline=False,
)
conversation: Final = [
{"role": "user", "content": "Refactor the tokenizer into its own module"},
{"role": "assistant", "content": "Done, moved it to tokenizer.py"},
{"role": "user", "content": "now add tests"},
]
with respx.mock(assert_all_called=True) as upstream:
route: Final = upstream.post("https://api.getnadir.com/v1/bucket").respond(200, json=_NADIR_VERDICT)
outcome: Final = await classifier.aclassify("now add tests", messages=conversation)
assert json.loads(route.calls.last.request.content) == {
"messages": [
{"role": "user", "content": "Refactor the tokenizer into its own module"},
{"role": "user", "content": "now add tests"},
],
"source": "litellm",
}
assert route.calls.last.request.headers["authorization"] == "Bearer ndr-deployment"
assert outcome.cause == "jev_classifier"
assert outcome.jev_verdict is not None
assert (outcome.jev_verdict.label, outcome.jev_verdict.provider) == ("SIMPLE", "nadir")
def test_nadir_messages_never_reach_a_system_one_server() -> None:
request: Final = build_jev_request(
"Classify this message:\nhi",
@ -1053,6 +1098,7 @@ async def test_deployed_decision_classifier_uses_registered_connection(
assert route.call_count == 1
assert route.calls[0].request.headers["authorization"] == "Bearer deployment-key"
assert json.loads(route.calls[0].request.content)["model"] == model
assert "Classify this message" in json.loads(route.calls[0].request.content)["state"]
assert (outcome.cause == "jev_classifier") is (status_code == 200)
if status_code == 200:
assert outcome.jev_verdict is not None