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

feat(router): add Switchyard capability classifier

Browse source
This commit is contained in:
Tin 2026-09-05 15:31:14 -07:00
parent bf51dea36b
commit a6a58b3e5d
11 changed files with 1003 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

60
litellm/router_strategy/complexity_router/README.md
View file

@ -68,6 +68,66 @@ still resolve to a deployment in `model_list`; this configuration does not creat
- abc
```
### Capability forecasting
Set `classifier_type: capability` to use
[NVIDIA NeMo Switchyard's packaged capability classifier](https://github.com/NVIDIA-NeMo/Switchyard/blob/main/crates/libsy/src/prompts/capability-classifier/prompt.md).
The classifier forecasts the probability that an efficient model completes
the whole task, identifies the capability-card boundary that applies, and leaves the
route choice to a deterministic threshold policy
```yaml
model_list:
- model_name: smart-router
litellm_params:
model: auto_router/complexity_router
complexity_router_config:
classifier_type: capability
classifier_llm_config:
model: classifier-model
capability_classifier_config:
efficient_tier: SIMPLE
capable_tier: REASONING
base_threshold: 0.5
threshold_step: 0.1
tiers:
SIMPLE:
- efficient-model-a
- efficient-model-b
REASONING: capable-model
```
The structured classifier verdict contains `crux`, `primary_rule`,
`capability_boundary`, and `p_solve`. The policy computes the required solve
probability as follows
- `supported`: `base_threshold`
- `uncertain` or `unmatched`: `base_threshold + threshold_step`
- `unsupported`: `base_threshold + 2 * threshold_step`
The efficient tier is selected when `p_solve` is greater than or equal to the
adjusted threshold. Otherwise the capable tier is selected. A malformed,
inconsistent, empty, or unavailable verdict always fails closed to the capable
tier. `base_threshold` is required, `threshold_step` defaults to `0`, and their
maximum adjusted threshold must not exceed `1`
The classifier receives the packaged Switchyard system prompt, the opening user
task, and the latest user follow-up when present. Caller system messages,
assistant turns, and intermediate tool results are not sent. The classifier call
uses strict JSON Schema output and the existing classifier timeout, circuit
breaker, attribution, redaction, reasoning-effort, and optional vision settings
`efficient_tier` and `capable_tier` name built-in complexity tiers with configured
model pools. The forecast still makes one binary quality decision, while the
ordinary tier pool may contain multiple equivalent deployments. Session affinity,
keyword overrides, plan-mode floors, modality checks, and other post-classification
complexity-router controls continue to apply
Routing decisions record the adjusted threshold and the complete valid forecast:
`classifier_p_solve`, `classifier_capability_boundary`, `classifier_primary_rule`,
and `classifier_crux`. Prompt redaction removes `classifier_crux` while retaining
the derived fields needed to audit the decision
### Heuristic v2
Set `classifier_type: heuristic_v2` to classify with the bundled calibrated

2
litellm/router_strategy/complexity_router/__init__.py
View file

@ -16,6 +16,7 @@ from litellm.router_strategy.complexity_router.complexity_router import (
from litellm.router_strategy.complexity_router.config import (
DEFAULT_CLASSIFIER_CONTEXT_WINDOW_SIZE,
DEFAULT_COMPLEXITY_CONFIG,
CapabilityClassifierConfig,
ClassificationRubric,
ComplexityRouterConfig,
ComplexityTier,
@ -28,6 +29,7 @@ from litellm.router_strategy.complexity_router.config import (
__all__ = [
"DEFAULT_CLASSIFIER_CONTEXT_WINDOW_SIZE",
"DEFAULT_COMPLEXITY_CONFIG",
"CapabilityClassifierConfig",
"ClassificationRubric",
"ComplexityRouter",
"ComplexityRouterConfig",

183
litellm/router_strategy/complexity_router/capability_classifier.py Normal file
View file

@ -0,0 +1,183 @@
# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
"""Capability forecast contract and routing policy adapted from NVIDIA NeMo Switchyard."""
from collections.abc import Mapping
from sys import float_info
from types import MappingProxyType
from typing import Final, Literal, TypeAlias
from pydantic import BaseModel, ConfigDict, Field, StrictFloat, TypeAdapter, model_validator
CapabilityBoundary: TypeAlias = Literal["supported", "uncertain", "unsupported", "unmatched"]
CapabilityRule: TypeAlias = Literal[
"SUP-1",
"SUP-2",
"SUP-3",
"SUP-4",
"SUP-5",
"UNC-1",
"UNC-2",
"LIM-1",
"LIM-2",
"none",
]
CAPABILITY_CLASSIFIER_SYSTEM_PROMPT: Final = """You are a task-level probability forecaster for a model router. You receive the
task's opening instruction and, when present, its latest user follow-up, plus
the qualitative capability card below.
Forecast one binary event:
SUCCESS means that the efficient agent completes the whole task correctly on
one fresh run under the actual harness, tools, and budget, as judged by the
final verifier. FAILURE means any other outcome. The two outcomes are
exhaustive.
Use only evidence in the instruction and the capability card. Do not assume
hidden repository state, unmentioned tools, validators, documentation, access,
or future work habits. Do not invent empirical counts, success rates, or base
rates. The capability card is qualitative evidence, not a measured prior.
# Assessment procedure
1. State the crux: the hardest material requirement for whole-task success.
2. Select the one capability rule that best describes the crux. Use
primary_rule=none and capability_boundary=unmatched when no rule applies.
Rule ids are opaque labels. Do not infer a boundary from an id's spelling.
3. Privately identify the strongest instruction-visible reasons for SUCCESS
and FAILURE, then imagine the most likely concrete failure.
4. Privately consider material unknowns. Missing information should limit
extreme estimates, but it is not evidence that p_solve must equal 0.50.
5. Estimate p_solve last. It is the probability of whole-task SUCCESS, not
confidence in this assessment, a route recommendation, or a cost judgment.
Interpret probabilities as natural frequencies. If p_solve is 0.70 for 100
comparable fresh runs, about 70 should succeed and 30 should fail. Use the full
range when justified. Reserve 0.00 and 1.00 for outcomes that are logically
impossible or certain under the visible contract. Supported does not mean 1.00,
and unsupported does not mean 0.00. The downstream routing threshold is not
part of this forecast.
# Efficient-agent capability card
The route verbs in this source card are inherited qualitative descriptions.
They do not ask you to output a route and do not assign a fixed probability to
any boundary.
- SUP-1 [supported]: Route to the Efficient model when the task provides a complete output contract and a deterministic local validator that covers the material requirements.
- SUP-2 [supported]: Route to the Efficient model when all required inputs are available, the target environment can be inspected, and correctness can be verified end-to-end without inaccessible external state.
- SUP-3 [supported]: Route to the Efficient model when mathematical behavior, interfaces, shapes, data types, tolerances, and performance requirements are explicit and exercised by a representative harness.
- SUP-4 [supported]: Route to the Efficient model when the required mechanism is identified, the relevant search space is bounded, and the success condition is executable. Do not infer this rule merely from the task's technical domain.
- SUP-5 [supported]: Route to the Efficient model when reconstruction or behavioral reproduction is constrained by an executable reference, parser, format specification, or checker strong enough to distinguish correct from merely plausible output.
- UNC-1 [uncertain]: Treat the route as uncertain when multiple reasonable interpretations of preprocessing, representation, indexing, naming, or output placement would produce different results and neither the instructions nor a validator resolve the choice.
- UNC-2 [uncertain]: Treat the route as uncertain when success requires finding every relevant item across heterogeneous inputs or environment state, but the task does not define the search boundary or provide a completeness check.
- LIM-1 [unsupported]: Prefer the Capable model when correctness depends primarily on extracting precise information from noisy visual, temporal, or rendered media and no machine-checkable extraction or replay mechanism is available.
- LIM-2 [unsupported]: Prefer the Capable model when success depends on reproducing undocumented reference behavior, hidden intermediate state, or an unknown configuration, and small deviations fail despite satisfying the visible specification.
# Output
Return exactly one JSON object matching the response schema supplied with the
request. Do not include markdown or commentary.
p_solve must be between 0.00 and 1.00. p_fail is exactly 1.00 - p_solve and
must not be emitted separately. Do not output recommended_route, confidence,
abstain, counts, task totals, empirical rates, or any other field."""
_BOUNDARY_STEPS: Final = MappingProxyType(
{
"supported": 0,
"uncertain": 1,
"unmatched": 1,
"unsupported": 2,
}
)
_RULE_BOUNDARIES: Final = MappingProxyType(
{
"SUP-1": "supported",
"SUP-2": "supported",
"SUP-3": "supported",
"SUP-4": "supported",
"SUP-5": "supported",
"UNC-1": "uncertain",
"UNC-2": "uncertain",
"LIM-1": "unsupported",
"LIM-2": "unsupported",
"none": "unmatched",
}
)
class CapabilityClassifierVerdict(BaseModel):
"""Strict structured verdict returned by the capability forecaster."""
model_config = ConfigDict(extra="forbid", frozen=True)
crux: str = Field(min_length=1)
primary_rule: CapabilityRule
capability_boundary: CapabilityBoundary
p_solve: StrictFloat = Field(ge=0.0, le=1.0)
@model_validator(mode="after")
def _validate_rule_boundary_pair(self) -> "CapabilityClassifierVerdict":
if not self.crux.strip():
raise ValueError("crux must contain non-whitespace text")
expected: Final = _RULE_BOUNDARIES[self.primary_rule]
if self.capability_boundary != expected:
raise ValueError(
f"primary_rule {self.primary_rule!r} requires capability_boundary {expected!r}, "
f"got {self.capability_boundary!r}"
)
return self
def routing_threshold(self, base_threshold: float, threshold_step: float) -> float:
"""Required efficient-model solve probability for this boundary."""
return base_threshold + _BOUNDARY_STEPS[self.capability_boundary] * threshold_step
def meets_routing_threshold(self, threshold: float) -> bool:
"""Inclusive comparison with Switchyard's one-epsilon rounding guard."""
return self.p_solve >= threshold or abs(threshold - self.p_solve) <= float_info.epsilon
_CAPABILITY_CLASSIFIER_RESPONSE_FORMAT_JSON: Final = """{
"type": "json_schema",
"json_schema": {
"name": "CapabilityClassifierDecision",
"strict": true,
"schema": {
"type": "object",
"additionalProperties": false,
"required": ["crux", "primary_rule", "capability_boundary", "p_solve"],
"properties": {
"crux": {"type": "string", "minLength": 1},
"primary_rule": {
"type": "string",
"enum": ["SUP-1", "SUP-2", "SUP-3", "SUP-4", "SUP-5", "UNC-1", "UNC-2", "LIM-1", "LIM-2", "none"]
},
"capability_boundary": {
"type": "string",
"enum": ["supported", "uncertain", "unsupported", "unmatched"]
},
"p_solve": {"type": "number", "minimum": 0.0, "maximum": 1.0}
}
}
}
}"""
_RESPONSE_FORMAT_ADAPTER: Final = TypeAdapter(Mapping[str, object])
def capability_classifier_response_format() -> Mapping[str, object]:
"""Fresh copy of Switchyard's packaged strict JSON Schema wrapper."""
return _RESPONSE_FORMAT_ADAPTER.validate_json(_CAPABILITY_CLASSIFIER_RESPONSE_FORMAT_JSON)
def parse_capability_classifier_verdict(content: str) -> CapabilityClassifierVerdict:
"""Parse raw JSON or the fenced JSON shape tolerated by Switchyard."""
text: Final = content.strip()
if not text.startswith("```"):
return CapabilityClassifierVerdict.model_validate_json(text)
unfenced: Final = text.removeprefix("```").removeprefix("json").lstrip("\n\r")
return CapabilityClassifierVerdict.model_validate_json(unfenced.removesuffix("```").strip())

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

@ -5,8 +5,9 @@ A rule-based routing strategy that uses weighted scoring across multiple dimensi
to classify requests by complexity and route them to appropriate models.
By default, scoring is local (regex/keyword-based) with no external API calls and <1ms
latency. Optionally, classifier_type="llm" routes classification through a configured
model instead, trading that latency/cost guarantee for potentially better accuracy.
latency. Optionally, classifier_type="llm" selects a tier through a configured model,
while classifier_type="capability" forecasts efficient-model success and applies a
Switchyard-compatible threshold policy.
keyword_tier_rules (lexical or, with semantic_keyword_matching, embedding-based) are
evaluated before either classification strategy and force a tier outright when matched.
@ -64,6 +65,12 @@ from litellm.types.utils import (
StandardLoggingRoutingDecisionTierBoundaries,
)
from .capability_classifier import (
CAPABILITY_CLASSIFIER_SYSTEM_PROMPT,
CapabilityClassifierVerdict,
capability_classifier_response_format,
parse_capability_classifier_verdict,
)
from .classification_rubrics import BUSINESS_TIER_CRITERIA, calibration_examples_section
from .config import (
CALIBRATION_EXAMPLES_HEADING,
@ -904,20 +911,43 @@ class ClassificationOutcome(NamedTuple):
"heuristic_v2",
"reasoning_override",
"llm_classifier",
"capability_classifier",
"heuristic_first_short_circuit",
"hybrid_short_circuit",
"housekeeping",
"classifier_plugin",
"classifier_fallback",
"capability_classifier_fallback",
"default_model_fallback",
]
classifier_cost: float | None = None
capability_verdict: CapabilityClassifierVerdict | None = None
capability_threshold: float | None = None
def _with_signal(outcome: ClassificationOutcome, signal: str | None) -> ClassificationOutcome:
return outcome if signal is None else outcome._replace(signals=(*outcome.signals, signal))
def _with_capability_forecast(
decision: StandardLoggingRoutingDecision, outcome: ClassificationOutcome
) -> StandardLoggingRoutingDecision:
"""Attach the validated capability verdict and applied threshold to its decision record."""
verdict: Final = outcome.capability_verdict
threshold: Final = outcome.capability_threshold
if verdict is None or threshold is None:
return decision
enriched: Final[StandardLoggingRoutingDecision] = { # mutable-ok: routing decisions are JSON TypedDict records
**decision,
"classifier_crux": verdict.crux,
"classifier_primary_rule": verdict.primary_rule,
"classifier_capability_boundary": verdict.capability_boundary,
"classifier_p_solve": verdict.p_solve,
"classifier_threshold": threshold,
}
return enriched
class _ClassifierCircuitBreaker:
"""Process-local timeout breaker for one complexity-router classifier.
@ -1162,7 +1192,11 @@ class ComplexityRouter(CustomLogger):
self._build_classifier_system_prompt() if llm_classifier_configured else None
)
self._classifier_response_format: Mapping[str, object] | None = (
type_to_response_format_param(_tier_classification_model(self.config.classifier_wire_labels()))
(
capability_classifier_response_format()
if self.config.classifier_type == "capability"
else type_to_response_format_param(_tier_classification_model(self.config.classifier_wire_labels()))
)
if llm_classifier_configured
else None
)
@ -1188,6 +1222,8 @@ class ComplexityRouter(CustomLogger):
llm_config: Final = self.config.classifier_llm_config
if llm_config is None:
raise ValueError("classifier_llm_config is not set")
if self.config.classifier_type == "capability":
return CAPABILITY_CLASSIFIER_SYSTEM_PROMPT
definitions: Final = self.config.tier_definitions
if definitions is not None:
return custom_tier_classification_prompt(
@ -1578,6 +1614,8 @@ class ComplexityRouter(CustomLogger):
return await self._classify_heuristic_first(prompt, system_prompt, request_kwargs, messages)
if self.config.classifier_type == "hybrid" and self.config.classifier_llm_config is not None:
return await self._classify_hybrid(prompt, system_prompt, request_kwargs, messages)
if self.config.classifier_type == "capability" and self.config.classifier_llm_config is not None:
return await self._capability_classifier_outcome(prompt, request_kwargs, messages)
if self.config.classifier_type != "llm" or self.config.classifier_llm_config is None:
tier, score, signals, cause = self._score_and_classify(prompt, system_prompt)
return ClassificationOutcome(tier=tier, score=score, signals=signals, cause=cause)
@ -1689,6 +1727,69 @@ class ComplexityRouter(CustomLogger):
)
)
async def _capability_classifier_outcome(
self,
prompt: str,
request_kwargs: Mapping[str, object] | None,
messages: Sequence[Mapping[str, object]] | None,
) -> ClassificationOutcome:
"""Forecast efficient-tier success, then apply the deterministic boundary policy."""
breaker: Final = self._classifier_circuit_breaker
permit: Final = breaker.acquire_permit() if breaker is not None else None
if breaker is not None and permit is None:
return self._capability_classifier_failure_outcome(
"capability classifier circuit is open", signal=_CLASSIFIER_CIRCUIT_OPEN_SIGNAL
)
try:
tier, classifier_cost, verdict, threshold = await self._classify_with_capability_llm(
prompt, request_kwargs, messages
)
if breaker is not None and permit is not None:
breaker.record_success(permit)
return ClassificationOutcome(
tier=tier,
score=None,
signals=(
f"capability-boundary:{verdict.capability_boundary}",
f"capability-rule:{verdict.primary_rule}",
),
cause="capability_classifier",
classifier_cost=classifier_cost,
capability_verdict=verdict,
capability_threshold=threshold,
)
except asyncio.CancelledError:
if breaker is not None and permit is not None:
breaker.record_failure(permit, is_timeout=False)
raise
except Exception as e: # noqa: BLE001 -- every unavailable or invalid judge verdict must fail closed
if breaker is not None and permit is not None:
breaker.record_failure(permit, is_timeout=_is_classifier_timeout(e))
return self._capability_classifier_failure_outcome(f"capability classifier failed ({e})")
def _capability_classifier_failure_outcome(self, reason: str, signal: str | None = None) -> ClassificationOutcome:
"""Fail closed to the configured capable tier without consulting another taxonomy."""
capability: Final = self.config.capability_classifier_config
if capability is None:
raise ValueError("capability_classifier_config is not set")
verbose_router_logger.warning(
"ComplexityRouter: %s, routing to capable_tier %s", reason, capability.capable_tier
)
signals: Final = (
("capability-classifier-fallback",)
if signal is None
else (
"capability-classifier-fallback",
signal,
)
)
return ClassificationOutcome(
tier=ComplexityTier(capability.capable_tier),
score=None,
signals=signals,
cause="capability_classifier_fallback",
)
async def _llm_classifier_outcome(
self,
prompt: str,
@ -1919,13 +2020,6 @@ class ComplexityRouter(CustomLogger):
label_roles=include_assistant,
)
request_metadata = (request_kwargs or {}).get("litellm_metadata") or (request_kwargs or {}).get("metadata")
metadata: Final = { # mutable-ok: SDK metadata kwarg is enriched by the request pipeline
**forwarded_internal_call_metadata(request_metadata, AUTOROUTER_CLASSIFIER_CALL_ORIGIN),
INTERNAL_CALL_ORIGIN_METADATA_KEY: AUTOROUTER_CLASSIFIER_CALL_ORIGIN,
}
turn_off_message_logging: Final = _effective_turn_off_message_logging(request_kwargs)
image_parts: Final = self._classifier_image_parts(messages)
user_content: Final[str | Sequence[ChatCompletionTextObject | ChatCompletionImageObject]] = (
[ # mutable-ok: SDK request payload content list is built once
@ -1939,11 +2033,85 @@ class ComplexityRouter(CustomLogger):
{"role": "system", "content": classifier_system_prompt},
{"role": "user", "content": user_content},
]
response_format: Final = classifier_response_format
classifier_call_params: Mapping[str, str] = EMPTY_MAPPING
if llm_config.reasoning_effort is not None:
classifier_call_params = MappingProxyType({"reasoning_effort": llm_config.reasoning_effort})
content, classifier_cost = await self._call_classifier_model(messages_for_call, request_kwargs)
raw_tier: Final = _LabeledTierClassification.model_validate_json(content).tier
tier: Final = self.config.resolve_classified_tier(raw_tier)
if tier is None:
raise ValueError(f"LLM classifier returned an unrecognized tier: {raw_tier!r}")
return tier, classifier_cost
async def _classify_with_capability_llm(
self,
prompt: str,
request_kwargs: Mapping[str, object] | None,
messages: Sequence[Mapping[str, object]] | None,
) -> tuple[ComplexityTier, float | None, CapabilityClassifierVerdict, float]:
"""Call the packaged capability forecaster and apply its two-tier policy."""
capability: Final = self.config.capability_classifier_config
classifier_system_prompt: Final = self._classifier_system_prompt
if capability is None or classifier_system_prompt is None:
raise ValueError("capability classifier is not configured")
asks_newest_first: Final = tuple(_iter_human_asks_newest_first(messages or (), self._reminder_markers))
opening_task: Final = asks_newest_first[-1] if asks_newest_first else prompt
latest_follow_up: Final = asks_newest_first[0] if len(asks_newest_first) > 1 else None
task_messages: list[AllMessageValues] = [ # mutable-ok: the latest message gains optional image parts below
{"role": "user", "content": opening_task}, # mutable-ok: SDK messages are dict-shaped
]
if latest_follow_up is not None:
task_messages.append( # mutable-ok: the provider SDK requires a concrete message list
{"role": "user", "content": latest_follow_up} # mutable-ok: SDK messages are dict-shaped
)
image_parts: Final = self._classifier_image_parts(messages)
if image_parts:
latest_text: Final = latest_follow_up or opening_task
task_messages[-1] = { # mutable-ok: SDK messages are dict-shaped
"role": "user",
"content": [ # mutable-ok: multimodal SDK content is a JSON array
{"type": "text", "text": latest_text}, # mutable-ok: SDK content parts are dict-shaped
*image_parts,
],
}
messages_for_call: Final[list[AllMessageValues]] = [ # mutable-ok: provider SDK requires a concrete list
{"role": "system", "content": classifier_system_prompt}, # mutable-ok: SDK messages are dict-shaped
*task_messages,
]
content, classifier_cost = await self._call_classifier_model(
messages_for_call,
request_kwargs,
max_output_tokens=capability.max_output_tokens,
)
verdict: Final = parse_capability_classifier_verdict(content)
threshold: Final = verdict.routing_threshold(capability.base_threshold, capability.threshold_step)
selected_tier: Final = (
capability.efficient_tier if verdict.meets_routing_threshold(threshold) else capability.capable_tier
)
return ComplexityTier(selected_tier), classifier_cost, verdict, threshold
async def _call_classifier_model(
self,
messages_for_call: list[AllMessageValues], # mutable-ok: provider SDK requires a concrete message list
request_kwargs: Mapping[str, object] | None,
max_output_tokens: int | None = None,
) -> tuple[str, float | None]:
"""Execute one structured classifier call with the router's shared safeguards."""
llm_config: Final = self.config.classifier_llm_config
response_format: Final = self._classifier_response_format
if llm_config is None or response_format is None:
raise ValueError("classifier_llm_config is not set")
request_values: Final = request_kwargs or EMPTY_MAPPING
request_metadata = request_values.get("litellm_metadata") or request_values.get("metadata")
metadata: Final = { # mutable-ok: SDK metadata kwarg is enriched by the request pipeline
**forwarded_internal_call_metadata(request_metadata, AUTOROUTER_CLASSIFIER_CALL_ORIGIN),
INTERNAL_CALL_ORIGIN_METADATA_KEY: AUTOROUTER_CLASSIFIER_CALL_ORIGIN,
}
classifier_call_params: dict[str, object] = {} # mutable-ok: optional SDK kwargs are assembled conditionally
if llm_config.reasoning_effort is not None:
classifier_call_params["reasoning_effort"] = llm_config.reasoning_effort
if max_output_tokens is not None:
classifier_call_params["max_tokens"] = max_output_tokens
proxy_server_request: Final = {
"body": {
"model": llm_config.model,
@ -1965,7 +2133,7 @@ class ComplexityRouter(CustomLogger):
disable_fallbacks=True,
metadata=metadata,
proxy_server_request=proxy_server_request,
turn_off_message_logging=turn_off_message_logging,
turn_off_message_logging=_effective_turn_off_message_logging(request_kwargs),
**classifier_call_params,
**_parent_session_kwargs(request_kwargs),
),
@ -1974,11 +2142,7 @@ class ComplexityRouter(CustomLogger):
content: Final = response.choices[0].message.content
if not content:
raise ValueError("LLM classifier returned empty content")
raw_tier: Final = _LabeledTierClassification.model_validate_json(content).tier
tier: Final = self.config.resolve_classified_tier(raw_tier)
if tier is None:
raise ValueError(f"LLM classifier returned an unrecognized tier: {raw_tier!r}")
return tier, _response_cost_or_none(response)
return content, _response_cost_or_none(response)
@staticmethod
def _build_classifier_user_payload(
@ -3690,7 +3854,8 @@ class ComplexityRouter(CustomLogger):
tier_litellm_params: Final = self._litellm_params_for_model(tier, routed_model)
classifier_model: Final = (
self.config.classifier_llm_config.model
if outcome.cause == "llm_classifier" and self.config.classifier_llm_config is not None
if outcome.cause in ("llm_classifier", "capability_classifier")
and self.config.classifier_llm_config is not None
else None
)
# cause=default_model_fallback means no tier was decided: the classifier failed and the
@ -3713,23 +3878,24 @@ class ComplexityRouter(CustomLogger):
decision_keyword: Final = (
plan_mode_sentinel if plan_floored else (housekeeping_sentinel if outcome.cause == "housekeeping" else None)
)
routing_decision: Final = self._build_routing_decision(
routed_model=routed_model,
conversation_continuing=conversation_continuing,
cause=decision_cause,
tier=classified_pool_tier,
score=score,
signals=decision_signals,
matched_keyword=decision_keyword,
escalation_keyword=escalation_keyword,
escalated=escalated,
classifier_model=classifier_model,
classifier_cost=outcome.classifier_cost,
tier_litellm_params=tier_litellm_params,
context_escalation_original_tier=context_original_tier,
)
return PreRoutingHookResponse(
model=routed_model,
messages=messages if has_original_messages else None,
litellm_params=tier_litellm_params,
routing_decision=self._build_routing_decision(
routed_model=routed_model,
conversation_continuing=conversation_continuing,
cause=decision_cause,
tier=classified_pool_tier,
score=score,
signals=decision_signals,
matched_keyword=decision_keyword,
escalation_keyword=escalation_keyword,
escalated=escalated,
classifier_model=classifier_model,
classifier_cost=outcome.classifier_cost,
tier_litellm_params=tier_litellm_params,
context_escalation_original_tier=context_original_tier,
),
routing_decision=_with_capability_forecast(routing_decision, outcome),
)

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

@ -10,7 +10,16 @@ from enum import Enum
from types import MappingProxyType
from typing import Annotated, Final, Literal
from pydantic import BaseModel, ConfigDict, Field, SkipValidation, field_serializer, field_validator, model_validator
from pydantic import (
BaseModel,
ConfigDict,
Field,
SkipValidation,
StrictFloat,
field_serializer,
field_validator,
model_validator,
)
from litellm.types.llms.openai import REASONING_EFFORT
from litellm.types.router import AdaptiveRouterWeights, ClassifierPlugin, RoutingPlugin
@ -44,7 +53,7 @@ DEFAULT_CLASSIFICATION_RUBRIC: Final[ClassificationRubric] = ClassificationRubri
# The classifier_type values that can call classifier_llm_config.model. Every consumer asking
# "is the classifier model a real dependency of this router" resolves it here, including the ones
# that only hold the raw config mapping and cannot reach ComplexityRouterConfig.uses_llm_classifier.
LLM_CLASSIFIER_TYPES: Final[frozenset[str]] = frozenset({"llm", "heuristic_first", "hybrid"})
LLM_CLASSIFIER_TYPES: Final[frozenset[str]] = frozenset({"llm", "capability", "heuristic_first", "hybrid"})
TIER_SEVERITY_ORDER: Final[tuple[ComplexityTier, ...]] = (
@ -569,6 +578,50 @@ class ClassifierLLMConfig(BaseModel):
return self
class CapabilityClassifierConfig(BaseModel):
"""Switchyard-compatible probability threshold policy for two model tiers."""
model_config = ConfigDict(frozen=True)
efficient_tier: str = Field(
description="Tier used when the efficient model's forecasted solve probability meets the adjusted threshold",
)
capable_tier: str = Field(
description=(
"Higher, fail-closed tier used below the adjusted threshold or when the classifier verdict is unavailable"
),
)
base_threshold: StrictFloat = Field(
ge=0.0,
le=1.0,
description="Lowest p_solve that routes a supported task to efficient_tier",
)
threshold_step: StrictFloat = Field(
default=0.0,
ge=0.0,
description=("Amount added once for uncertain or unmatched verdicts and twice for unsupported verdicts"),
)
max_output_tokens: int = Field(
default=4096,
ge=1,
description="Maximum completion tokens available to the capability classifier verdict",
)
@field_validator("efficient_tier", "capable_tier")
@classmethod
def _normalize_tier(cls, value: str) -> str:
normalized: Final = value.strip()
if not normalized:
raise ValueError("tier must be non-empty")
return normalized
@model_validator(mode="after")
def _validate_threshold_range(self) -> "CapabilityClassifierConfig":
if self.base_threshold + 2 * self.threshold_step > 1.0:
raise ValueError("base_threshold + 2 * threshold_step must be at most 1")
return self
class ComplexityRouterConfig(BaseModel):
"""Configuration for the ComplexityRouter."""
@ -713,13 +766,16 @@ class ComplexityRouterConfig(BaseModel):
)
# Classifier strategy
classifier_type: Literal["heuristic", "heuristic_v2", "llm", "custom", "heuristic_first", "hybrid"] = Field(
classifier_type: Literal[
"heuristic", "heuristic_v2", "llm", "capability", "custom", "heuristic_first", "hybrid"
] = Field(
default="heuristic",
description=(
"Classification strategy: local regex/keyword scoring, the bundled trained four-tier heuristic, "
"an LLM call, a custom classifier plugin, 'heuristic_first', which scores locally and only pays "
"for the LLM classifier when the local scorer does not confidently land a cheap tier, or 'hybrid', "
"which trusts the local scorer everywhere except when its score lands near a tier boundary"
"an LLM tier-selection call, a Switchyard-compatible capability forecast, a custom classifier "
"plugin, 'heuristic_first', which scores locally and only pays for the LLM classifier when the "
"local scorer does not confidently land a cheap tier, or 'hybrid', which trusts the local scorer "
"everywhere except when its score lands near a tier boundary"
),
)
heuristic_v2_artifact: TrainedTierArtifact | Literal["ultrafeedback"] = Field(
@ -733,7 +789,15 @@ class ComplexityRouterConfig(BaseModel):
default=None,
description=(
"Configuration for the LLM classifier; required when classifier_type is 'llm', "
"'heuristic_first' or 'hybrid'"
"'capability', 'heuristic_first' or 'hybrid'"
),
)
capability_classifier_config: CapabilityClassifierConfig | None = Field(
default=None,
description=(
"Probability threshold policy required when classifier_type is 'capability'. The classifier "
"forecasts p_solve for efficient_tier, adjusts base_threshold using the capability-card boundary, "
"and otherwise routes to capable_tier"
),
)
heuristic_first_max_tier: str | None = Field(
@ -1245,6 +1309,66 @@ class ComplexityRouterConfig(BaseModel):
)
return self
@model_validator(mode="after")
def _validate_capability_classifier_config(self) -> "ComplexityRouterConfig":
capability: Final = self.capability_classifier_config
if self.classifier_type != "capability":
if capability is not None:
raise ValueError(
"capability_classifier_config requires classifier_type 'capability'; otherwise it has no effect"
)
return self
if capability is None:
raise ValueError("capability_classifier_config is required when classifier_type is 'capability'")
return self
@model_validator(mode="after")
def _validate_capability_classifier_tiers(self) -> "ComplexityRouterConfig":
capability: Final = self.capability_classifier_config
if self.classifier_type != "capability" or capability is None:
return self
if self.tier_definitions is not None:
raise ValueError(
"classifier_type 'capability' uses the built-in tier map and cannot be combined with tier_definitions"
)
for field, tier in (
("efficient_tier", capability.efficient_tier),
("capable_tier", capability.capable_tier),
):
if tier not in self.tier_names():
raise ValueError(
f"{field} {tier!r} is not an active tier: it must name one of {', '.join(self.tier_names())}"
)
if not self.tiers.get(tier):
raise ValueError(f"{field} {tier!r} has no model configured in tiers")
names: Final = self.tier_names()
if names.index(capability.capable_tier) <= names.index(capability.efficient_tier):
raise ValueError("capable_tier must be a higher tier than efficient_tier")
return self
@model_validator(mode="after")
def _validate_capability_classifier_prompt_policy(self) -> "ComplexityRouterConfig":
if self.classifier_type != "capability":
return self
llm_config: Final = self.classifier_llm_config
if llm_config is not None and (
llm_config.system_prompt is not None or llm_config.classification_rubric is not None
):
raise ValueError(
"classifier_type 'capability' uses the packaged capability card; classifier_llm_config.system_prompt "
"and classification_rubric are not supported"
)
if self.classification_prompt is not None or self.classification_examples is not None:
raise ValueError(
"classifier_type 'capability' uses the packaged capability card; classification_prompt and "
"classification_examples are not supported"
)
if self.classifier_fallback != "heuristic":
raise ValueError(
"classifier_type 'capability' always fails closed to capable_tier; classifier_fallback cannot override it"
)
return self
@field_validator("heuristic_first_max_tier", mode="before")
@classmethod
def _coerce_heuristic_first_max_tier(cls, value: object) -> object:
@ -1453,7 +1577,7 @@ class ComplexityRouterConfig(BaseModel):
)
if duplicated:
raise ValueError(f"tier_definitions names must be unique (case-insensitive): {', '.join(duplicated)}")
if self.classifier_type in ("heuristic", "heuristic_v2", "heuristic_first", "hybrid"):
if self.classifier_type in ("heuristic", "heuristic_v2", "capability", "heuristic_first", "hybrid"):
raise ValueError(
"tier_definitions requires classifier_type 'llm' or 'custom': the heuristic scorer only "
"produces the four built-in tiers, as does heuristic_v2"

2
litellm/router_utils/auto_router_model_naming.py
View file

@ -113,7 +113,7 @@ def strategy_router_dependencies(
"""The model names a strategy-router deployment must reach, in no particular order.
A field is a dependency only under the condition the runtime itself reads it: the
classifier model needs `classifier_type: llm`, and the complexity embedding model needs
classifier model needs an LLM-backed classifier type, and the complexity embedding model needs
`semantic_keyword_matching`. Listing one the router never calls reds a working deployment.
The two default-model spellings are not symmetric. A quality router falls back to its

17
litellm/types/utils.py
View file

@ -2849,6 +2849,7 @@ RoutingDecisionCause = Literal[
# meant anything that filtered `signals` silently changed what the row claimed.
"reasoning_override",
"llm_classifier",
"capability_classifier",
# classifier_type 'heuristic_first': the local scorer produced at least one signal and landed at
# or below heuristic_first_max_tier, so it decided the tier and the LLM classifier was never
# called. Distinct from "heuristic_scorer", which is a router whose only classifier IS the
@ -2861,6 +2862,9 @@ RoutingDecisionCause = Literal[
# The LLM classifier or classifier plugin failed on a router with an operator-defined
# tier set, so the request routed to the configured fallback_tier without being classified.
"classifier_fallback",
# The capability judge failed or returned an invalid verdict, so its fail-closed policy
# routed to capable_tier without consulting the unrelated complexity heuristic.
"capability_classifier_fallback",
# The LLM classifier or classifier plugin failed and classifier_fallback is
# 'default_model', so the request went to default_model without being classified.
# Distinct from "default_fallback",
@ -2935,6 +2939,11 @@ class StandardLoggingRoutingDecision(TypedDict, total=False):
escalation_keyword: str
classifier_model: str
classifier_cost: float
classifier_crux: str # writable-ok: added only when a capability verdict is available
classifier_primary_rule: str # writable-ok: added only when a capability verdict is available
classifier_capability_boundary: str # writable-ok: added only when a capability verdict is available
classifier_p_solve: float # writable-ok: added only when a capability verdict is available
classifier_threshold: float # writable-ok: added only when a capability verdict is available
escalated: bool
context_escalated: bool # writable-ok: Pydantic warns on ReadOnly TypedDict fields
context_escalation_original_tier: str # writable-ok: Pydantic warns on ReadOnly TypedDict fields
@ -2950,7 +2959,9 @@ class StandardLoggingRoutingDecision(TypedDict, total=False):
# logging off. Every other field aggregates the prompt without reproducing it and is kept,
# so a redacted row stays explainable. `test_every_routing_decision_field_is_classified`
# fails if a field is added to the record without being placed in one set or the other.
PROMPT_QUOTING_ROUTING_DECISION_FIELDS: frozenset[str] = frozenset({"signals", "matched_keyword", "escalation_keyword"})
PROMPT_QUOTING_ROUTING_DECISION_FIELDS: frozenset[str] = frozenset(
{"signals", "matched_keyword", "escalation_keyword", "classifier_crux"}
)
DERIVED_ROUTING_DECISION_FIELDS: Final[frozenset[str]] = frozenset(
{
"router_model_name",
@ -2963,6 +2974,10 @@ DERIVED_ROUTING_DECISION_FIELDS: Final[frozenset[str]] = frozenset(
"score",
"classifier_model",
"classifier_cost",
"classifier_primary_rule",
"classifier_capability_boundary",
"classifier_p_solve",
"classifier_threshold",
"escalated",
"context_escalated",
"context_escalation_original_tier",

9
tests/test_litellm/proxy/management_endpoints/test_auto_router_endpoints.py
View file

@ -334,6 +334,15 @@ def test_a_request_must_carry_exactly_one_usable_conversation(body: dict):
"config_overrides",
[
{"classifier_type": "llm", "classifier_llm_config": {"model": "classifier-model"}},
{
"classifier_type": "capability",
"classifier_llm_config": {"model": "classifier-model"},
"capability_classifier_config": {
"efficient_tier": "SIMPLE",
"capable_tier": "REASONING",
"base_threshold": 0.5,
},
},
{
"semantic_keyword_matching": True,
"embedding_model": "classifier-model",

334
tests/test_litellm/router_strategy/test_complexity_router.py
View file

@ -5,6 +5,7 @@ Tests the rule-based complexity scoring and tier assignment logic.
"""
import asyncio
import json
import logging
import sys
from typing import Dict, List
@ -38,7 +39,12 @@ from litellm.router_strategy.complexity_router.complexity_router import (
classification_system_prompt,
custom_tier_classification_prompt,
)
from litellm.router_strategy.complexity_router.capability_classifier import (
CAPABILITY_CLASSIFIER_SYSTEM_PROMPT,
CapabilityClassifierVerdict,
)
from litellm.router_strategy.complexity_router.config import (
CapabilityClassifierConfig,
DEFAULT_CLASSIFICATION_RUBRIC,
DEFAULT_CLASSIFIER_CONTEXT_WINDOW_SIZE,
DEFAULT_COMPLEXITY_CONFIG,
@ -1947,6 +1953,321 @@ class TestLLMClassifierConfig:
)
CAPABILITY_TIERS: Dict[str, str] = {
"SIMPLE": "efficient-model",
"REASONING": "capable-model",
}
def _capability_router_config(**overrides):
return {
"tiers": dict(CAPABILITY_TIERS),
"classifier_type": "capability",
"classifier_llm_config": {"model": "judge-model", "timeout_ms": 400},
"capability_classifier_config": {
"efficient_tier": "SIMPLE",
"capable_tier": "REASONING",
"base_threshold": 0.5,
"threshold_step": 0.1,
},
**overrides,
}
def _capability_reply(
*,
p_solve: float,
primary_rule: str = "SUP-1",
capability_boundary: str = "supported",
crux: str = "complete the requested change",
) -> str:
return json.dumps(
{
"crux": crux,
"primary_rule": primary_rule,
"capability_boundary": capability_boundary,
"p_solve": p_solve,
}
)
class TestCapabilityClassifierConfig:
@pytest.mark.parametrize(
"patch,error_match",
[
({"classifier_llm_config": None}, "classifier_llm_config is required"),
({"capability_classifier_config": None}, "capability_classifier_config is required"),
(
{
"capability_classifier_config": {
"efficient_tier": "SIMPLE",
"capable_tier": "SIMPLE",
"base_threshold": 0.5,
}
},
"must be a higher tier",
),
(
{
"capability_classifier_config": {
"efficient_tier": "REASONING",
"capable_tier": "SIMPLE",
"base_threshold": 0.5,
}
},
"must be a higher tier",
),
(
{
"capability_classifier_config": {
"efficient_tier": "MEDIUM",
"capable_tier": "REASONING",
"base_threshold": 0.5,
}
},
"has no model configured",
),
(
{
"capability_classifier_config": {
"efficient_tier": "SIMPLE",
"capable_tier": "REASONING",
"base_threshold": 0.9,
"threshold_step": 0.1,
}
},
r"base_threshold \+ 2 \* threshold_step must be at most 1",
),
({"classifier_fallback": "default_model", "default_model": "fallback"}, "always fails closed"),
(
{"classifier_llm_config": {"model": "judge-model", "system_prompt": "pick one"}},
"uses the packaged capability card",
),
({"classification_examples": "example"}, "uses the packaged capability card"),
],
)
def test_rejects_incoherent_configuration(self, patch, error_match):
with pytest.raises(ValidationError, match=error_match):
ComplexityRouterConfig(**{**_capability_router_config(), **patch})
def test_capability_config_is_rejected_on_other_classifier_types(self):
config = _capability_router_config(classifier_type="llm")
with pytest.raises(ValidationError, match="requires classifier_type 'capability'"):
ComplexityRouterConfig(**config)
def test_threshold_defaults_match_switchyard(self):
config = CapabilityClassifierConfig(efficient_tier=" SIMPLE ", capable_tier=" REASONING ", base_threshold=0.5)
assert config.efficient_tier == "SIMPLE"
assert config.capable_tier == "REASONING"
assert config.threshold_step == 0.0
assert config.max_output_tokens == 4096
def test_classifier_model_is_registered_as_a_dependency(self):
assert ComplexityRouterConfig(**_capability_router_config()).uses_llm_classifier is True
class TestCapabilityClassifierVerdict:
@pytest.mark.parametrize(
"primary_rule,capability_boundary",
[
*((f"SUP-{index}", "supported") for index in range(1, 6)),
*((f"UNC-{index}", "uncertain") for index in range(1, 3)),
*((f"LIM-{index}", "unsupported") for index in range(1, 3)),
("none", "unmatched"),
],
)
def test_accepts_every_valid_rule_boundary_pair(self, primary_rule, capability_boundary):
verdict = CapabilityClassifierVerdict(
crux="the hard part",
primary_rule=primary_rule,
capability_boundary=capability_boundary,
p_solve=0.5,
)
assert verdict.primary_rule == primary_rule
assert verdict.capability_boundary == capability_boundary
@pytest.mark.parametrize(
"payload,error_match",
[
(
{
"crux": "x",
"primary_rule": "SUP-1",
"capability_boundary": "unsupported",
"p_solve": 0.5,
},
"requires capability_boundary",
),
(
{"crux": " ", "primary_rule": "none", "capability_boundary": "unmatched", "p_solve": 0.5},
"non-whitespace",
),
(
{
"crux": "x",
"primary_rule": "none",
"capability_boundary": "unmatched",
"p_solve": 0.5,
"recommended_route": "efficient",
},
"Extra inputs are not permitted",
),
(
{"crux": "x", "primary_rule": "none", "capability_boundary": "unmatched", "p_solve": True},
"valid number",
),
],
)
def test_rejects_invalid_or_inconsistent_verdicts(self, payload, error_match):
with pytest.raises(ValidationError, match=error_match):
CapabilityClassifierVerdict.model_validate(payload)
class TestCapabilityClassifier:
@staticmethod
def _router(mock_router_instance, **overrides):
return ComplexityRouter(
model_name="capability-router",
litellm_router_instance=mock_router_instance,
complexity_router_config=_capability_router_config(**overrides),
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"p_solve,primary_rule,boundary,expected_tier,expected_threshold",
[
(0.5, "SUP-1", "supported", ComplexityTier.SIMPLE, 0.5),
(0.59, "UNC-1", "uncertain", ComplexityTier.REASONING, 0.6),
(0.6, "UNC-1", "uncertain", ComplexityTier.SIMPLE, 0.6),
(0.59, "none", "unmatched", ComplexityTier.REASONING, 0.6),
(0.69, "LIM-1", "unsupported", ComplexityTier.REASONING, 0.7),
(0.7, "LIM-1", "unsupported", ComplexityTier.SIMPLE, 0.7),
],
)
async def test_boundary_adjusted_threshold_is_inclusive(
self, mock_router_instance, p_solve, primary_rule, boundary, expected_tier, expected_threshold
):
mock_router_instance.acompletion = AsyncMock(
return_value=_llm_response(
_capability_reply(p_solve=p_solve, primary_rule=primary_rule, capability_boundary=boundary)
)
)
outcome = await self._router(mock_router_instance).aclassify("do the task")
assert outcome.tier == expected_tier
assert outcome.cause == "capability_classifier"
assert outcome.capability_threshold == pytest.approx(expected_threshold)
@pytest.mark.asyncio
async def test_fenced_json_verdict_is_accepted(self, mock_router_instance):
reply = _capability_reply(p_solve=0.8)
mock_router_instance.acompletion = AsyncMock(return_value=_llm_response(f"```json\n{reply}\n```"))
outcome = await self._router(mock_router_instance).aclassify("do the task")
assert outcome.tier == ComplexityTier.SIMPLE
assert outcome.cause == "capability_classifier"
@pytest.mark.asyncio
async def test_decimal_rounding_does_not_break_inclusive_threshold(self, mock_router_instance):
config = _capability_router_config(
capability_classifier_config={
"efficient_tier": "SIMPLE",
"capable_tier": "REASONING",
"base_threshold": 0.1,
"threshold_step": 0.1,
}
)
mock_router_instance.acompletion = AsyncMock(
return_value=_llm_response(
_capability_reply(p_solve=0.3, primary_rule="LIM-1", capability_boundary="unsupported")
)
)
router = ComplexityRouter(
model_name="capability-router",
litellm_router_instance=mock_router_instance,
complexity_router_config=config,
)
outcome = await router.aclassify("do the task")
assert outcome.capability_threshold == 0.30000000000000004
assert outcome.tier == ComplexityTier.SIMPLE
@pytest.mark.asyncio
async def test_call_uses_packaged_prompt_schema_and_opening_plus_latest_user_task(self, mock_router_instance):
mock_router_instance.acompletion = AsyncMock(
return_value=_llm_response(_capability_reply(p_solve=0.8), response_cost=0.002)
)
router = self._router(mock_router_instance)
messages = [
{"role": "system", "content": "Never expose this caller instruction to the judge"},
{"role": "user", "content": "Build the feature"},
{"role": "assistant", "content": "I need more information"},
{"role": "user", "content": "Use the existing API"},
]
response = await router.async_pre_routing_hook(model="capability-router", request_kwargs={}, messages=messages)
assert response.model == "efficient-model"
call = mock_router_instance.acompletion.call_args.kwargs
assert call["messages"] == [
{"role": "system", "content": CAPABILITY_CLASSIFIER_SYSTEM_PROMPT},
{"role": "user", "content": "Build the feature"},
{"role": "user", "content": "Use the existing API"},
]
schema = call["response_format"]["json_schema"]["schema"]
assert call["response_format"]["json_schema"]["name"] == "CapabilityClassifierDecision"
assert call["response_format"]["json_schema"]["strict"] is True
assert schema["additionalProperties"] is False
assert set(schema["required"]) == {"crux", "primary_rule", "capability_boundary", "p_solve"}
assert schema["properties"]["primary_rule"]["enum"] == [
"SUP-1",
"SUP-2",
"SUP-3",
"SUP-4",
"SUP-5",
"UNC-1",
"UNC-2",
"LIM-1",
"LIM-2",
"none",
]
assert call["max_tokens"] == 4096
decision = response.routing_decision
assert decision["cause"] == "capability_classifier"
assert decision["classifier_model"] == "judge-model"
assert decision["classifier_cost"] == 0.002
assert decision["classifier_crux"] == "complete the requested change"
assert decision["classifier_primary_rule"] == "SUP-1"
assert decision["classifier_capability_boundary"] == "supported"
assert decision["classifier_p_solve"] == 0.8
assert decision["classifier_threshold"] == 0.5
@pytest.mark.asyncio
@pytest.mark.parametrize(
"reply",
[
"not json",
_capability_reply(p_solve=0.9, primary_rule="SUP-1", capability_boundary="unsupported"),
'{"crux":"x","primary_rule":"SUP-1","capability_boundary":"supported","p_solve":0.9,"route":"efficient"}',
],
ids=["malformed", "inconsistent-pair", "extra-field"],
)
async def test_invalid_verdict_fails_closed_to_capable_tier(self, mock_router_instance, reply):
mock_router_instance.acompletion = AsyncMock(return_value=_llm_response(reply))
outcome = await self._router(mock_router_instance).aclassify("do the task")
assert outcome.tier == ComplexityTier.REASONING
assert outcome.cause == "capability_classifier_fallback"
assert outcome.signals == ("capability-classifier-fallback",)
@pytest.mark.asyncio
async def test_classifier_call_failure_fails_closed_to_capable_model(self, mock_router_instance):
mock_router_instance.acompletion = AsyncMock(side_effect=TimeoutError("judge unavailable"))
response = await self._router(mock_router_instance).async_pre_routing_hook(
model="capability-router",
request_kwargs={},
messages=[{"role": "user", "content": "do the task"}],
)
assert response.model == "capable-model"
assert response.routing_decision["cause"] == "capability_classifier_fallback"
CUSTOM_TIER_LABELS: Dict[str, str] = {
"SIMPLE": "Cheap",
"MEDIUM": "Standard",
@ -7164,6 +7485,11 @@ class TestRedactedLoggingDropsPromptText:
"score": 0.8,
"tier_boundaries": {"simple_medium": 0.15, "medium_complex": 0.35, "complex_reasoning": 0.6},
"classifier_model": "claude-haiku",
"classifier_crux": "deploy the requested service to k8s",
"classifier_primary_rule": "SUP-2",
"classifier_capability_boundary": "supported",
"classifier_p_solve": 0.8,
"classifier_threshold": 0.5,
"escalated": True,
"tier_litellm_params": {"reasoning_effort": "xhigh"},
"signals": ["code (python)"],
@ -7171,7 +7497,13 @@ class TestRedactedLoggingDropsPromptText:
"escalation_keyword": "LITELLM ESCALATE",
}
kept = Router._redact_prompt_text_if_needed(request_kwargs={}, routing_decision=full)
assert set(full) - set(kept) == {"signals", "matched_keyword", "escalation_keyword"}
assert set(full) - set(kept) == {
"signals",
"matched_keyword",
"escalation_keyword",
"classifier_crux",
}
assert kept["classifier_p_solve"] == 0.8
assert kept["tier_litellm_params"] == {"reasoning_effort": "xhigh"}
@pytest.mark.asyncio

16
tests/test_litellm/router_utils/test_auto_router_model_naming.py
View file

@ -214,6 +214,22 @@ def test_config_check_ignores_the_model_entirely():
},
(("a", "tier"), ("clf", "classifier")),
),
(
{
"model": "auto_router/complexity_router",
"complexity_router_config": {
"tiers": {"SIMPLE": "a", "REASONING": "b"},
"classifier_type": "capability",
"classifier_llm_config": {"model": "clf"},
"capability_classifier_config": {
"efficient_tier": "SIMPLE",
"capable_tier": "REASONING",
"base_threshold": 0.5,
},
},
},
(("a", "tier"), ("b", "tier"), ("clf", "classifier")),
),
(
{
"model": "auto_router/complexity_router",

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

@ -24763,6 +24763,39 @@ export interface components {
*/
status: "cancelled";
};
/**
* CapabilityClassifierConfig
* @description Switchyard-compatible probability threshold policy for two model tiers.
*/
CapabilityClassifierConfig: {
/**
* Base Threshold
* @description Lowest p_solve that routes a supported task to efficient_tier
*/
base_threshold: number;
/**
* Capable Tier
* @description Higher, fail-closed tier used below the adjusted threshold or when the classifier verdict is unavailable
*/
capable_tier: string;
/**
* Efficient Tier
* @description Tier used when the efficient model's forecasted solve probability meets the adjusted threshold
*/
efficient_tier: string;
/**
* Max Output Tokens
* @description Maximum completion tokens available to the capability classifier verdict
* @default 4096
*/
max_output_tokens: number;
/**
* Threshold Step
* @description Amount added once for uncertain or unmatched verdicts and twice for unsupported verdicts
* @default 0
*/
threshold_step: number;
};
/** ChatCompletionAnnotation */
ChatCompletionAnnotation: {
/**
@ -34748,6 +34781,8 @@ export interface components {
adaptive_eligible: "all" | "classified_tier";
/** @description Quality vs cost weights for adaptive selection (used when adaptive=True) */
adaptive_weights?: components["schemas"]["AdaptiveRouterWeights"];
/** @description Probability threshold policy required when classifier_type is 'capability'. The classifier forecasts p_solve for efficient_tier, adjusts base_threshold using the capability-card boundary, and otherwise routes to capable_tier */
capability_classifier_config?: components["schemas"]["CapabilityClassifierConfig"] | null;
/**
* Classification Examples
* @description Replaces the calibration examples of the LLM classifier rubric, and nothing else. Written as example lines only: the router renders the 'Calibration examples:' heading above them, after the per-tier bullets. Requires an LLM classifier and cannot be combined with classifier_llm_config.system_prompt. With built-in tiers the rubric preset still supplies the tier criteria and, unless classification_prompt replaces them, the classification instructions; a custom tier set ships no examples of its own, so the section renders only when this is set.
@ -34795,7 +34830,7 @@ export interface components {
* @enum {string}
*/
classifier_fallback: "heuristic" | "default_model";
/** @description Configuration for the LLM classifier; required when classifier_type is 'llm', 'heuristic_first' or 'hybrid' */
/** @description Configuration for the LLM classifier; required when classifier_type is 'llm', 'capability', 'heuristic_first' or 'hybrid' */
classifier_llm_config?: components["schemas"]["ClassifierLLMConfig"] | null;
/**
* Classifier Plugin
@ -34810,11 +34845,11 @@ export interface components {
classifier_plugin_timeout_ms: number;
/**
* Classifier Type
* @description Classification strategy: local regex/keyword scoring, the bundled trained four-tier heuristic, an LLM call, a custom classifier plugin, 'heuristic_first', which scores locally and only pays for the LLM classifier when the local scorer does not confidently land a cheap tier, or 'hybrid', which trusts the local scorer everywhere except when its score lands near a tier boundary
* @description Classification strategy: local regex/keyword scoring, the bundled trained four-tier heuristic, an LLM tier-selection call, a Switchyard-compatible capability forecast, a custom classifier plugin, 'heuristic_first', which scores locally and only pays for the LLM classifier when the local scorer does not confidently land a cheap tier, or 'hybrid', which trusts the local scorer everywhere except when its score lands near a tier boundary
* @default heuristic
* @enum {string}
*/
classifier_type: "heuristic" | "heuristic_v2" | "llm" | "custom" | "heuristic_first" | "hybrid";
classifier_type: "heuristic" | "heuristic_v2" | "llm" | "capability" | "custom" | "heuristic_first" | "hybrid";
/**
* Code Keywords
* @description Keywords indicating code-related content
@ -36141,11 +36176,21 @@ export interface components {
* Cause
* @enum {string}
*/
cause?: "heuristic_scorer" | "heuristic_v2" | "reasoning_override" | "llm_classifier" | "heuristic_first_short_circuit" | "hybrid_short_circuit" | "classifier_plugin" | "classifier_fallback" | "default_model_fallback" | "literal_keyword_match" | "semantic_keyword_match" | "plan_mode" | "housekeeping" | "modality_escalation" | "modality_pin_override" | "health_failover" | "session_affinity_pin" | "session_affinity_escalation" | "user_turn_continuation" | "default_fallback" | "keyword" | "quality_tier" | "bandit";
cause?: "heuristic_scorer" | "heuristic_v2" | "reasoning_override" | "llm_classifier" | "capability_classifier" | "heuristic_first_short_circuit" | "hybrid_short_circuit" | "classifier_plugin" | "classifier_fallback" | "capability_classifier_fallback" | "default_model_fallback" | "literal_keyword_match" | "semantic_keyword_match" | "plan_mode" | "housekeeping" | "modality_escalation" | "modality_pin_override" | "health_failover" | "session_affinity_pin" | "session_affinity_escalation" | "user_turn_continuation" | "default_fallback" | "keyword" | "quality_tier" | "bandit";
/** Classifier Capability Boundary */
classifier_capability_boundary?: string;
/** Classifier Cost */
classifier_cost?: number;
/** Classifier Crux */
classifier_crux?: string;
/** Classifier Model */
classifier_model?: string;
/** Classifier P Solve */
classifier_p_solve?: number;
/** Classifier Primary Rule */
classifier_primary_rule?: string;
/** Classifier Threshold */
classifier_threshold?: number;
/** Context Escalated */
context_escalated?: boolean;
/** Context Escalation Original Tier */