diff --git a/litellm/router_strategy/complexity_router/README.md b/litellm/router_strategy/complexity_router/README.md index f55362b6c411..d9f1b7b22628 100644 --- a/litellm/router_strategy/complexity_router/README.md +++ b/litellm/router_strategy/complexity_router/README.md @@ -258,6 +258,73 @@ Spend logs record `routing_decision.cause: heuristic_v2`, the detected request type, and all four predicted probabilities. Existing `classifier_type: heuristic` configurations keep the original weighted scorer unchanged +### Custom classifier plugins + +`classifier_type: custom` hands the tier decision to a plugin of your own. The plugin is an +object with one `async classify(context) -> str | None` method, named in the config by the +dotted path to an instance. It receives a `RoutingContext` (the request messages, both raw +and normalized to chat-completions shape, the candidate models, and the request metadata +including caller identity) and returns the name of the tier to route to, or `None` to +decline. A decline, an exception, or a call that overruns `classifier_plugin_timeout_ms` +(default 3000) all fall through to `classifier_fallback`, so a classifier that is down +cannot fail a completion. + +Everything downstream of the tier is unchanged: the tier's model pool, the provider call, +your provider keys, fallbacks, and spend tracking all stay on this proxy. + +#### Nadir + +`litellm.router_strategy.complexity_router.nadir_classifier.nadir_classifier` is a bundled +plugin that asks [Nadir](https://getnadir.com)'s `/v1/bucket` endpoint which of `simple` / +`medium` / `complex` a request is. The endpoint grades the request without generating an +answer, and the three buckets map to the SIMPLE / MEDIUM / COMPLEX tiers: + +```yaml +model_list: + - model_name: smart-router + litellm_params: + model: auto_router/complexity_router + complexity_router_config: + classifier_type: custom + classifier_plugin: litellm.router_strategy.complexity_router.nadir_classifier.nadir_classifier + classifier_fallback: heuristic + tiers: + SIMPLE: gpt-5-nano + MEDIUM: gpt-5-mini + COMPLEX: gpt-5 +``` + +`NADIR_API_KEY` attributes decisions to a Nadir account and lifts the anonymous rate limit. +That account's request log then keeps the user text of each classified request, or only a +hash of it when the account has prompt storage turned off. Without a key the endpoint still +answers and keeps no prompt, but its per-IP rate limit is sized for trying the plugin out, +and requests over it route on `classifier_fallback`. + +`NADIR_API_BASE` points at a self-hosted or on-prem Nadir instead of the hosted API. +`NADIR_API_KEY` is only sent to that base, or to the hosted API when it is unset. + +Renaming the tiers with `tier_labels` needs no change here: the router still accepts the +default tier names the plugin returns. A router built on `tier_definitions` names its own +tiers, so point `classifier_plugin` at an instance of your own that maps Nadir's buckets +onto them: + +```python +# my_classifiers.py, named in the config as my_classifiers.nadir +from litellm.router_strategy.complexity_router.nadir_classifier import NadirComplexityClassifier + +nadir = NadirComplexityClassifier(tier_map={"simple": "cheap", "medium": "standard", "complex": "premium"}) +``` + +Two properties to weigh against the local scorers. The classification is a network call, so +it costs a round trip per request where `heuristic` and `heuristic_v2` cost under a +millisecond (a laptop against the hosted API measured about 160ms warm, and the first call +in a fresh process also pays connection setup; anything that overruns +`classifier_plugin_timeout_ms` routes on the fallback instead of waiting). And the messages +are sent to the configured Nadir host, which is a third party unless that host is yours, the +same disclosure `classifier_type: llm` carries when the classifier model is a hosted one. A +REASONING tier is never returned, since Nadir grades three buckets; keyword rules and the +reasoning override still place requests there. + ### Renaming the tiers `tier_labels` puts your own vocabulary on the four tiers: diff --git a/litellm/router_strategy/complexity_router/nadir_classifier.py b/litellm/router_strategy/complexity_router/nadir_classifier.py new file mode 100644 index 000000000000..cd69c0a605bb --- /dev/null +++ b/litellm/router_strategy/complexity_router/nadir_classifier.py @@ -0,0 +1,152 @@ +""" +Nadir as the Complexity Router's classifier. + +A decision-only integration: the tier comes from a call to Nadir's ``/v1/bucket`` +endpoint, and everything else stays here. The tier's model pool, the provider call, the +operator's own provider keys, fallbacks, spend tracking and the response path are +untouched, so Nadir sees the messages it is asked to classify and never the response. + +``/v1/bucket`` grades the request without generating an answer and replies with a +``routing_tier`` of ``simple`` / ``medium`` / ``complex``, which map to the router's +default SIMPLE / MEDIUM / COMPLEX tiers. Renaming the tiers with ``tier_labels`` needs +nothing here, since the router still accepts the default names. A router built on +``tier_definitions`` names its own tiers, so it constructs its own instance with a +``tier_map`` onto them. + +Configure it in the proxy by pointing ``classifier_plugin`` at the module-level +instance:: + + model_list: + - model_name: smart-router + litellm_params: + model: auto_router/complexity_router + complexity_router_config: + classifier_type: custom + classifier_plugin: litellm.router_strategy.complexity_router.nadir_classifier.nadir_classifier + tiers: + SIMPLE: gpt-5-nano + MEDIUM: gpt-5-mini + COMPLEX: gpt-5 + +``NADIR_API_KEY`` attributes decisions to a Nadir account and lifts the anonymous rate +limit. That account's request log then keeps the user text of each classified request, or +only a hash of it when the account has prompt storage turned off. Without a key the +endpoint still answers and keeps no prompt, but its per-IP rate limit is sized for trying +the plugin out, and requests over it route on ``classifier_fallback``. + +``NADIR_API_BASE`` points at a self-hosted or on-prem Nadir. ``NADIR_API_KEY`` is only +ever sent to that base, or to the hosted API when it is unset: an ``api_base`` passed in +code brings its own ``api_key``. + +Every failure mode is the router's existing one: this returns None to decline, and a +network error, an error status, a timeout past ``classifier_plugin_timeout_ms`` or an +unknown bucket all hand the request to ``classifier_fallback``. Nothing here can fail a +completion. + +Privacy note for operators: the messages are sent to the configured Nadir host, which is +a third party unless that host is your own. That is the same disclosure a remote LLM +classifier carries, and it is the reason the local heuristic scorers exist. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from types import MappingProxyType +from typing import Final + +from pydantic import TypeAdapter, ValidationError + +from litellm.constants import NADIR_DEFAULT_API_BASE +from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, get_async_httpx_client +from litellm.secret_managers.main import get_secret_str +from litellm.types.llms.custom_http import httpxSpecialProvider +from litellm.types.router import RoutingContext + +DEFAULT_TIER_MAP: Final[Mapping[str, str]] = MappingProxyType( + {"simple": "SIMPLE", "medium": "MEDIUM", "complex": "COMPLEX"} +) + +_API_PATH: Final = "/v1/bucket" +_VERDICT: Final = TypeAdapter(Mapping[str, object]) + + +def _bucket_url(api_base: str) -> str: + """Join the endpoint path to a host, tolerating a base that already ends in ``/v1``. + + Nadir's own docs advertise ``https://api.getnadir.com/v1`` as the base URL, because the + OpenAI-compatible clients that consume it append ``/chat/completions``. An operator who + copies that value into ``NADIR_API_BASE`` would otherwise send ``/v1/v1/bucket`` and get a + 404 that reads like the endpoint does not exist. + """ + base: Final = api_base.rstrip("/").removesuffix("/v1") + return f"{base}{_API_PATH}" + + +def _configured_bucket_url() -> str: + """The endpoint the environment names: ``NADIR_API_BASE``, else the hosted API.""" + return _bucket_url(get_secret_str("NADIR_API_BASE") or NADIR_DEFAULT_API_BASE) + + +class NadirComplexityClassifier: + """Classifier plugin that asks Nadir's decision API which tier a request belongs to. + + Args: + api_base: Nadir host. Defaults to ``NADIR_API_BASE``, then to the hosted API. + api_key: Nadir API key. Defaults to ``NADIR_API_KEY`` only when the request goes to the + host the environment names, so the environment key never follows an ``api_base`` + set in code; anonymous otherwise. + tier_map: Nadir bucket name -> the tier name this router routes on. Defaults to the + router's built-in tier names. + client: HTTP client the request goes out on. Defaults to litellm's shared async client. + Leave it unset on an instance a proxy config names, since the proxy deepcopies it. + """ + + def __init__( + self, + api_base: str | None = None, + api_key: str | None = None, + tier_map: Mapping[str, str] | None = None, + client: AsyncHTTPHandler | None = None, + ) -> None: + self._api_base: Final = api_base + self._api_key: Final = api_key + self._client: Final = client + # A plain dict, not a MappingProxyType: the proxy deepcopies a deployment's + # litellm_params, this instance travels inside them, and a mappingproxy cannot be + # deepcopied. Router construction would fail before the first request. + self._tier_map: Final[dict[str, str]] = dict( # mutable-ok: read-only after __init__, deepcopy-safe + DEFAULT_TIER_MAP if tier_map is None else tier_map + ) + + def _headers(self, url: str) -> dict[str, str]: + env_key: Final = get_secret_str("NADIR_API_KEY") if url == _configured_bucket_url() else None + api_key: Final = self._api_key or env_key + return {"X-API-Key": api_key} if api_key else {} + + async def classify(self, context: RoutingContext) -> str | None: + """Return the tier Nadir places this request in, or None to decline. + + Declines rather than guesses on anything the endpoint cannot grade: a request with no + messages, and a bucket name outside ``tier_map`` (which is what a ``tier_definitions`` + router looks like before ``tier_map`` is configured). Both leave the decision to + ``classifier_fallback``, where a local scorer still routes the request. + """ + messages: Final = context.structured_messages or context.raw_messages + if not messages: + return None + url: Final = _bucket_url(self._api_base) if self._api_base else _configured_bucket_url() + client: Final = self._client or get_async_httpx_client(llm_provider=httpxSpecialProvider.ComplexityClassifier) + body: Final = {"messages": list(messages), "source": "litellm"} + response: Final = await client.post(url=url, json=body, headers=self._headers(url)) + try: + verdict: Final = _VERDICT.validate_python(response.json()) + except ValidationError: + return None + # ``routing_tier`` is the tier Nadir itself routes on: the classifier's bucket after its + # deterministic confidence and gate adjustments. A server that predates it sends only ``bucket``. + tier: Final = verdict.get("routing_tier") or verdict.get("bucket") + return self._tier_map.get(tier.strip().lower()) if isinstance(tier, str) else None + + +nadir_classifier: Final = NadirComplexityClassifier() +"""Module-level instance, so the proxy config can name it by dotted path.""" diff --git a/litellm/types/llms/custom_http.py b/litellm/types/llms/custom_http.py index 47f80c528454..09a11a83d020 100644 --- a/litellm/types/llms/custom_http.py +++ b/litellm/types/llms/custom_http.py @@ -37,6 +37,7 @@ class httpxSpecialProvider(str, Enum): PasswordBreachCheck = "password_breach_check" ASGI = "asgi" AgentHarness = "agent_harness" + ComplexityClassifier = "complexity_classifier" VerifyTypes = str | bool | ssl.SSLContext diff --git a/tests/unit/router_strategy/complexity_router/test_nadir_classifier.py b/tests/unit/router_strategy/complexity_router/test_nadir_classifier.py new file mode 100644 index 000000000000..b2e6fb427f21 --- /dev/null +++ b/tests/unit/router_strategy/complexity_router/test_nadir_classifier.py @@ -0,0 +1,244 @@ +"""Tests for the Nadir classifier plugin (litellm/router_strategy/complexity_router/nadir_classifier.py).""" + +import copy + +import pytest + +from litellm.router_strategy.complexity_router import ComplexityRouter +from litellm.router_strategy.complexity_router.nadir_classifier import ( + NadirComplexityClassifier, + _bucket_url, + nadir_classifier, +) +from litellm.types.router import ClassifierPlugin, RoutingContext + +HOSTED_BUCKET_URL = "https://api.getnadir.com/v1/bucket" + + +class _FakeResponse: + def __init__(self, payload): + self._payload = payload + + def json(self): + return self._payload + + +class _FakeClient: + """Records the one request the classifier makes, and answers with a canned bucket.""" + + def __init__(self, payload=None, error=None): + self._payload = payload + self._error = error + self.calls = [] + + async def post(self, url, json=None, headers=None): + self.calls.append({"url": url, "json": json, "headers": headers}) + if self._error is not None: + raise self._error + return _FakeResponse(self._payload) + + +def _context(messages=None): + resolved = [{"role": "user", "content": "refactor the retry loop"}] if messages is None else messages + return RoutingContext( + raw_messages=resolved, + structured_messages=resolved, + candidate_models=["gpt-4o-mini", "gpt-4o"], + ) + + +class TestBucketURL: + """The endpoint URL, including the documented double-/v1 footgun.""" + + @pytest.mark.parametrize( + "api_base", + [ + "https://api.getnadir.com", + "https://api.getnadir.com/", + "https://api.getnadir.com/v1", + "https://api.getnadir.com/v1/", + ], + ) + def test_v1_is_never_doubled(self, api_base): + """Nadir's docs advertise the base URL with /v1, so operators paste it with and without.""" + assert _bucket_url(api_base) == HOSTED_BUCKET_URL + + def test_self_hosted_host_with_a_path_prefix_is_preserved(self): + assert _bucket_url("https://gateway.internal/nadir") == "https://gateway.internal/nadir/v1/bucket" + + +class TestClassify: + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("bucket", "expected_tier"), [("simple", "SIMPLE"), ("medium", "MEDIUM"), ("complex", "COMPLEX")] + ) + async def test_each_bucket_maps_to_its_default_tier(self, bucket, expected_tier): + client = _FakeClient(payload={"bucket": bucket, "confidence": 0.91}) + assert await NadirComplexityClassifier(client=client).classify(_context()) == expected_tier + assert client.calls[0]["url"] == HOSTED_BUCKET_URL + assert client.calls[0]["json"]["messages"] == [{"role": "user", "content": "refactor the retry loop"}] + assert client.calls[0]["json"]["source"] == "litellm" + + @pytest.mark.asyncio + async def test_tier_map_override_serves_custom_tier_names(self): + """A tier_definitions router names its own tiers; tier_map must answer in those names.""" + classifier = NadirComplexityClassifier( + tier_map={"simple": "cheap", "medium": "mid", "complex": "deep"}, + client=_FakeClient(payload={"bucket": "complex"}), + ) + assert await classifier.classify(_context()) == "deep" + + @pytest.mark.asyncio + async def test_api_key_is_sent_when_configured(self): + client = _FakeClient(payload={"bucket": "simple"}) + await NadirComplexityClassifier(api_key="ndr_test", client=client).classify(_context()) + assert client.calls[0]["headers"] == {"X-API-Key": "ndr_test"} + + @pytest.mark.asyncio + async def test_env_supplies_key_and_base(self, monkeypatch): + client = _FakeClient(payload={"bucket": "simple"}) + monkeypatch.setenv("NADIR_API_KEY", "ndr_from_env") + monkeypatch.setenv("NADIR_API_BASE", "https://nadir.internal") + await NadirComplexityClassifier(client=client).classify(_context()) + assert client.calls[0]["headers"] == {"X-API-Key": "ndr_from_env"} + assert client.calls[0]["url"] == "https://nadir.internal/v1/bucket" + + @pytest.mark.asyncio + async def test_env_key_never_follows_a_base_set_in_code(self, monkeypatch): + """The environment names where NADIR_API_KEY may go; an api_base set in code brings its own key.""" + client = _FakeClient(payload={"bucket": "simple"}) + monkeypatch.setenv("NADIR_API_KEY", "ndr_from_env") + monkeypatch.delenv("NADIR_API_BASE", raising=False) + await NadirComplexityClassifier(api_base="https://elsewhere.example", client=client).classify(_context()) + assert client.calls[0]["url"] == "https://elsewhere.example/v1/bucket" + assert client.calls[0]["headers"] == {} + + @pytest.mark.asyncio + async def test_env_key_goes_to_the_hosted_api_named_in_code(self, monkeypatch): + """Naming the hosted base in code, with or without /v1, is still the base the environment trusts.""" + client = _FakeClient(payload={"bucket": "simple"}) + monkeypatch.setenv("NADIR_API_KEY", "ndr_from_env") + monkeypatch.delenv("NADIR_API_BASE", raising=False) + await NadirComplexityClassifier(api_base="https://api.getnadir.com/v1", client=client).classify(_context()) + assert client.calls[0]["headers"] == {"X-API-Key": "ndr_from_env"} + + @pytest.mark.asyncio + async def test_a_base_set_in_code_sends_its_own_key(self, monkeypatch): + client = _FakeClient(payload={"bucket": "simple"}) + monkeypatch.setenv("NADIR_API_KEY", "ndr_from_env") + classifier = NadirComplexityClassifier( + api_base="https://nadir.onprem.internal", api_key="ndr_own", client=client + ) + await classifier.classify(_context()) + assert client.calls[0]["headers"] == {"X-API-Key": "ndr_own"} + + @pytest.mark.asyncio + async def test_anonymous_when_no_key_is_configured(self, monkeypatch): + client = _FakeClient(payload={"bucket": "simple"}) + monkeypatch.delenv("NADIR_API_KEY", raising=False) + await NadirComplexityClassifier(client=client).classify(_context()) + assert client.calls[0]["headers"] == {} + + @pytest.mark.asyncio + async def test_no_messages_declines_without_a_call(self): + """An empty request is a 400 from the endpoint; decline locally instead of spending the trip.""" + client = _FakeClient(payload={"bucket": "simple"}) + assert await NadirComplexityClassifier(client=client).classify(_context(messages=[])) is None + assert client.calls == [] + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "payload", [{"bucket": "reasoning"}, {"bucket": None}, {"bucket": 3}, {}, ["not", "a", "mapping"]] + ) + async def test_unusable_verdicts_decline(self, payload): + """Anything outside tier_map is declined, so classifier_fallback decides rather than this guessing.""" + assert await NadirComplexityClassifier(client=_FakeClient(payload=payload)).classify(_context()) is None + + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("payload", "expected_tier"), + [ + ({"bucket": "simple", "routing_tier": "medium"}, "MEDIUM"), + ({"bucket": "complex", "routing_tier": "medium"}, "MEDIUM"), + ({"bucket": "complex"}, "COMPLEX"), + ], + ) + async def test_routing_tier_wins_over_the_raw_bucket(self, payload, expected_tier): + """routing_tier carries Nadir's own confidence and gate adjustments; bucket is the pre-adjustment + prediction, read only from a server that predates routing_tier.""" + assert ( + await NadirComplexityClassifier(client=_FakeClient(payload=payload)).classify(_context()) == expected_tier + ) + + @pytest.mark.asyncio + async def test_unknown_routing_tier_declines_rather_than_reading_the_bucket(self): + classifier = NadirComplexityClassifier( + client=_FakeClient(payload={"bucket": "simple", "routing_tier": "reasoning"}) + ) + assert await classifier.classify(_context()) is None + + @pytest.mark.asyncio + async def test_bucket_name_is_matched_case_insensitively(self): + classifier = NadirComplexityClassifier(client=_FakeClient(payload={"bucket": " Complex "})) + assert await classifier.classify(_context()) == "COMPLEX" + + def test_module_instance_satisfies_the_plugin_protocol(self): + """The proxy resolves the dotted path to this instance and interface-checks it at startup.""" + assert isinstance(nadir_classifier, ClassifierPlugin) + + +class TestDeploymentConfigSurvivesRouterInit: + """The plugin instance travels inside a deployment's litellm_params, which get deepcopied.""" + + @pytest.mark.parametrize("tier_map", [None, {"simple": "Cheap", "medium": "Standard", "complex": "Premium"}]) + def test_classifier_is_deepcopyable(self, tier_map): + """A MappingProxyType here would raise `cannot pickle 'mappingproxy'` at Router init, i.e. at + proxy startup, long before any request reaches the classifier.""" + classifier = NadirComplexityClassifier(tier_map=tier_map) + copied = copy.deepcopy({"complexity_router_config": {"classifier_plugin": classifier}}) + assert isinstance(copied["complexity_router_config"]["classifier_plugin"], NadirComplexityClassifier) + + +class TestThroughTheComplexityRouter: + """The plugin as the router actually drives it: verdict in, tier out, failures fall back.""" + + def _router(self, client, **overrides): + return ComplexityRouter( + model_name="test-nadir-router", + litellm_router_instance=None, + complexity_router_config={ + "tiers": {"SIMPLE": "gpt-4o-mini", "MEDIUM": "gpt-4o", "COMPLEX": "o1-preview"}, + "classifier_type": "custom", + "classifier_plugin": NadirComplexityClassifier(client=client), + **overrides, + }, + ) + + @pytest.mark.asyncio + async def test_verdict_becomes_the_routed_tier(self): + router = self._router(_FakeClient(payload={"bucket": "complex"})) + outcome = await router.aclassify( + prompt="port the scheduler to the new executor", + raw_messages=[{"role": "user", "content": "port the scheduler to the new executor"}], + ) + assert outcome.tier.value == "COMPLEX" + assert outcome.cause == "classifier_plugin" + + @pytest.mark.asyncio + async def test_renamed_tier_labels_need_no_tier_map(self): + """tier_labels only renames what the dashboard shows; the router still accepts the default names.""" + router = self._router( + _FakeClient(payload={"bucket": "complex"}), + tier_labels={"SIMPLE": "Cheap", "MEDIUM": "Standard", "COMPLEX": "Premium"}, + ) + outcome = await router.aclassify(prompt="hi", raw_messages=[{"role": "user", "content": "hi"}]) + assert outcome.tier.value == "COMPLEX" + assert outcome.cause == "classifier_plugin" + + @pytest.mark.asyncio + async def test_network_failure_falls_back_to_the_local_scorer(self): + """A Nadir outage must never fail a completion: the heuristic scorer still places the request.""" + router = self._router(_FakeClient(error=ConnectionError("nadir unreachable"))) + outcome = await router.aclassify(prompt="hi", raw_messages=[{"role": "user", "content": "hi"}]) + assert outcome.cause != "classifier_plugin" + assert outcome.tier is not None