feat(router): add TypeSafe Jev as a complexity router classifier

litellm/proxy/management_helpers/auto_router_permissions.py

@@ -65,6 +65,21 @@ class _MemberRouterGenerationParams(BaseModel):
     stop: str | tuple[str, ...] | None = None
 
 
+class _MemberJevClassifierConfig(BaseModel):
+    """The Jev classifier settings a team member may set. Credentials stay the proxy's own: a member-chosen
+    api_base would receive the proxy's TYPESAFE_API_KEY, and a member-chosen api_key would be sent from the proxy."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    model: str
+    api_key: None = None
+    api_base: None = None
+    timeout_ms: int
+    instructions: str | None = None
+    circuit_breaker_enabled: bool
+    circuit_breaker_cooldown_seconds: float
+
+
 class _MemberComplexityRouterConfig(RequestComplexityRouterConfig):
     model_config = ConfigDict(extra="forbid", arbitrary_types_allowed=True)
 
@@ -113,6 +128,8 @@ def validate_member_auto_router_config(config: Mapping[str, object]) -> RequestC
         for entries in validated.tier_model_configs.values():
             for entry in entries:
                 _MemberRouterGenerationParams.model_validate(entry.litellm_params)
+        if validated.jev_classifier_config is not None:
+            _MemberJevClassifierConfig.model_validate(validated.jev_classifier_config.model_dump())
         return validated
     except ValidationError as exc:
         location: Final = ".".join(str(part) for part in exc.errors()[0]["loc"])

litellm/router_strategy/complexity_router/complexity_router.py

@@ -54,12 +54,15 @@
 from litellm.litellm_core_utils.sensitive_data_masker import mask_credentials_in_payload
 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.router_strategy.adaptive_router.classifier import classify_prompt
 from litellm.router_strategy.complexity_router.tier_predictor import (
     TierSuccessPredictor,
     resolve_tier_artifact,
 )
 from litellm.router_utils.pre_call_checks.deployment_affinity_check import DeploymentAffinityCheck
+from litellm.secret_managers.main import get_secret_str
+from litellm.types.llms.custom_http import httpxSpecialProvider
 from litellm.types.llms.openai import (
     AllMessageValues,
     ChatCompletionImageObject,
@@ -102,8 +105,17 @@
     ComplexityRouterConfig,
     ComplexityTier,
     CustomDimension,
+    JevClassifierConfig,
     TierDefinition,
 )
+from .jev_classifier import (
+    DEFAULT_JEV_INSTRUCTIONS,
+    HttpJevClassifierClient,
+    JevClassifierClient,
+    JevVerdict,
+    build_jev_request,
+    jev_classifier_cost,
+)
 from .llm_v2 import LLM_V2_PROMPT_VERSION, LLMV2Decision, LLMV2TaskContext, LLMV2Verdict, llm_v2_response_format
 from .stall_detector import detect_stalled_task
 
@@ -169,6 +181,16 @@ def _built_in_tier_or_none(tier_name: str) -> ComplexityTier | None:
     }
 )
 
+_JEV_TIER_CRITERIA: Final[Mapping[str, str]] = MappingProxyType(
+    {
+        ComplexityTier.NON_REASONING.value: "Relaying, reformatting, or extracting stated information without judgment",
+        ComplexityTier.SIMPLE.value: "Greetings, chitchat, or short factual lookups with known answers",
+        ComplexityTier.MEDIUM.value: "Everyday requests needing explanation, light reasoning, or minor technical work",
+        ComplexityTier.COMPLEX.value: "Non-trivial code, architecture, multi-step work, or specialized domain depth",
+        ComplexityTier.REASONING.value: "Open-ended analysis, proofs, tradeoffs, or tasks requiring careful thought",
+    }
+)
+
 TIER_SEVERITY_ORDER_LABELED: Final[tuple[tuple[ComplexityTier, str], ...]] = tuple(
     (tier, tier.value) for tier in TIER_SEVERITY_ORDER
 )
@@ -1006,6 +1028,7 @@ class ClassificationOutcome(NamedTuple):
         "reasoning_override",
         "llm_classifier",
         "capability_classifier",
+        "jev_classifier",
         "llm_v2_classifier",
         "llm_v2_fallback",
         "heuristic_first_short_circuit",
@@ -1019,6 +1042,7 @@ class ClassificationOutcome(NamedTuple):
     classifier_cost: float | None = None
     capability_forecast: CapabilityClassifierForecast | None = None
     llm_v2_forecast: LLMV2Decision | None = None
+    jev_verdict: JevVerdict | None = None
 
 
 def _with_signal(outcome: ClassificationOutcome, signal: str | None) -> ClassificationOutcome:
@@ -1051,6 +1075,13 @@ def _with_classifier_forecast(
     decision: StandardLoggingRoutingDecision, outcome: ClassificationOutcome
 ) -> StandardLoggingRoutingDecision:
     """Attach validated forecasts and their applied policy to the routing decision."""
+    if outcome.jev_verdict is not None:
+        forecasted_decision: Final[StandardLoggingRoutingDecision] = {
+            **decision,
+            "classifier_probabilities": outcome.jev_verdict.probabilities,
+            "classifier_confidence": outcome.jev_verdict.confidence,
+        }
+        return forecasted_decision
     if outcome.llm_v2_forecast is not None:
         return _with_llm_v2_forecast(decision, outcome.llm_v2_forecast)
     forecast: Final = outcome.capability_forecast
@@ -1235,13 +1266,26 @@ class ComplexityRouter(CustomLogger):
     - Question complexity (multiple questions)
     """
 
+    @staticmethod
+    def _build_jev_client(config: JevClassifierConfig) -> JevClassifierClient:
+        api_key: Final = config.api_key or get_secret_str("TYPESAFE_API_KEY")
+        if not api_key:
+            raise ValueError("jev_classifier_config.api_key or TYPESAFE_API_KEY is required for classifier_type 'jev'")
+        api_base: Final = config.api_base or get_secret_str("TYPESAFE_API_BASE") or "https://api.typesafe.ai"
+        return HttpJevClassifierClient(
+            api_key=api_key,
+            api_base=api_base,
+            http_client=get_async_httpx_client(httpxSpecialProvider.PassThroughEndpoint),
+        )
+
     def __init__(
         self,
         model_name: str,
         litellm_router_instance: Router,
         complexity_router_config: dict[str, Any] | None = None,
         default_model: str | None = None,
         derive_savings_baseline: bool = True,
+        jev_client: JevClassifierClient | None = None,
     ):
         """
         Initialize ComplexityRouter.
@@ -1269,6 +1313,15 @@ def __init__(
         if default_model:
             self.config.default_model = default_model
 
+        jev_config: Final = self.config.jev_classifier_config
+        self._jev_client: JevClassifierClient | None = (
+            jev_client
+            if jev_client is not None
+            else self._build_jev_client(jev_config)
+            if self.config.classifier_type == "jev" and jev_config is not None
+            else None
+        )
+
         self._tier_affinity_config = hashlib.sha256(
             self.config.model_dump_json(include=MappingProxyType({"tiers": True, "tier_model_configs": True})).encode()
         ).hexdigest()
@@ -1357,15 +1410,20 @@ def __init__(
             if llm_classifier_configured
             else None
         )
-        self._classifier_circuit_breaker: _ClassifierCircuitBreaker | None = (
-            _ClassifierCircuitBreaker(self.config.classifier_llm_config.circuit_breaker_cooldown_seconds)
+        circuit_breaker_cooldown: Final[float | None] = (
+            self.config.classifier_llm_config.circuit_breaker_cooldown_seconds
             if (
                 llm_classifier_configured
                 and self.config.classifier_llm_config is not None
                 and self.config.classifier_llm_config.circuit_breaker_enabled
             )
+            else jev_config.circuit_breaker_cooldown_seconds
+            if (self.config.classifier_type == "jev" and jev_config is not None and jev_config.circuit_breaker_enabled)
             else None
         )
+        self._classifier_circuit_breaker: _ClassifierCircuitBreaker | None = (
+            _ClassifierCircuitBreaker(circuit_breaker_cooldown) if circuit_breaker_cooldown is not None else None
+        )
         self._tier_success_predictor: TierSuccessPredictor | None = (
             TierSuccessPredictor(resolve_tier_artifact(self.config.heuristic_v2_artifact))
             if self.config.classifier_type == "heuristic_v2"
@@ -1797,6 +1855,8 @@ async def aclassify(
             return self._classify_with_heuristic_v2(prompt)
         if self.config.classifier_type == "custom":
             return await self._classify_with_plugin(prompt, system_prompt, request_kwargs, raw_messages)
+        if self.config.classifier_type == "jev":
+            return await self._jev_classifier_outcome(prompt, system_prompt)
         if self.config.classifier_type in ("heuristic_first", "hybrid") and _encrypted_classifier_task(
             request_kwargs, self._reminder_markers_for_request(request_kwargs or EMPTY_MAPPING)
         ):
@@ -2031,6 +2091,88 @@ async def _llm_classifier_outcome(
                 f"LLM classifier failed ({type(e).__name__})", prompt, system_prompt, scored
             )
 
+    async def _jev_classifier_outcome(self, prompt: str, system_prompt: str | None) -> ClassificationOutcome:
+        config: Final = self.config.jev_classifier_config
+        client: Final = self._jev_client
+        if config is None or client is None:
+            return self._classifier_failure_outcome("jev classifier is not configured", prompt, system_prompt)
+        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._classifier_failure_outcome(
+                "jev classifier circuit is open",
+                prompt,
+                system_prompt,
+                signal=_CLASSIFIER_CIRCUIT_OPEN_SIGNAL,
+            )
+        criteria: Final[Mapping[str, str]] = (
+            MappingProxyType(
+                {
+                    definition.name: definition.description
+                    or _JEV_TIER_CRITERIA.get(definition.name.upper(), definition.name)
+                    for definition in self.config.tier_definitions
+                }
+            )
+            if self.config.tier_definitions is not None
+            else MappingProxyType(
+                {label: _JEV_TIER_CRITERIA[tier.value] for tier, label in self.config.labeled_tiers()}
+            )
+        )
+        timeout_s: Final = config.timeout_ms / 1000
+        request: Final = build_jev_request(
+            prompt=prompt,
+            system_prompt=system_prompt,
+            model=config.model,
+            instructions=config.instructions or DEFAULT_JEV_INSTRUCTIONS,
+            criteria=criteria,
+        )
+        try:
+            response: Final = await asyncio.wait_for(client.evaluate(request, timeout_s), timeout_s)
+            answer: Final = response.answers.get("tier")
+            if answer is None:
+                raise ValueError("Jev response is missing the 'tier' answer")
+            tier: Final = self.config.resolve_classified_tier(answer.choice)
+            if tier is None:
+                raise ValueError(f"Jev classifier returned unknown tier {answer.choice!r}")
+            tier_name: Final = _tier_name(tier)
+            if not self._tier_pools().get(tier_name):
+                raise ValueError(f"Jev classifier returned tier {tier_name!r}, which has no models configured")
+            model: Final = response.model or config.model
+            verdict: Final = JevVerdict(
+                label=answer.choice,
+                probabilities=answer.probabilities,
+                confidence=answer.confidence,
+                model=model,
+                cost=jev_classifier_cost(response, config.model),
+            )
+            if breaker is not None and permit is not None:
+                breaker.record_success(permit)
+            return ClassificationOutcome(
+                tier=tier,
+                score=None,
+                signals=(
+                    f"jev-classifier:{tier_name}",
+                    f"jev-confidence={answer.confidence:.6f}",
+                    *(
+                        f"tier-probability:{label}={probability:.6f}"
+                        for label, probability in answer.probabilities.items()
+                    ),
+                ),
+                cause="jev_classifier",
+                classifier_cost=verdict.cost,
+                jev_verdict=verdict,
+            )
+        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 -- external Jev call can fail in many distinct ways
+            if breaker is not None and permit is not None:
+                breaker.record_failure(permit, is_timeout=_is_classifier_timeout(e))
+            return self._classifier_failure_outcome(
+                f"jev classifier failed ({type(e).__name__})", prompt, system_prompt
+            )
+
     def _classifier_failure_outcome(
         self,
         reason: str,
@@ -4467,7 +4609,9 @@ async def _classify_and_route(
 
         tier_litellm_params: Final = self._litellm_params_for_model(tier, routed_model)
         classifier_model: Final = (
-            self.config.classifier_llm_config.model
+            f"typesafe/{outcome.jev_verdict.model}"
+            if outcome.cause == "jev_classifier" and outcome.jev_verdict is not None
+            else self.config.classifier_llm_config.model
             if outcome.cause in ("llm_classifier", "capability_classifier", "llm_v2_classifier", "llm_v2_fallback")
             and self.config.classifier_llm_config is not None
             else None