feat(router): resolve auto-router routing plugins from proxy YAML config

litellm/proxy/proxy_server.py

@@ -556,6 +556,7 @@ def generate_feedback_box():
 from litellm.types.router import ModelInfo as RouterModelInfo
 from litellm.types.router import (
     RouterGeneralSettings,
+    RoutingPlugin,
     SearchToolTypedDict,
     updateDeployment,
 )
@@ -3660,6 +3661,45 @@ def _attach_redis_usage_cache(redis_cache: RedisCache, enable_redis_auth_cache:
     litellm_config_cache.redis_cache = redis_cache
 
 
+def resolve_complexity_router_plugins(
+    model_name: str,
+    complexity_router_config: dict,
+    config_file_path: str | None,
+) -> None:
+    """
+    Resolves `complexity_router_config["plugins"]` dotted-path strings to live
+    instances via `get_instance_fn` (the same convention `litellm_settings.callbacks`
+    uses), in place. Raises at config-load time if a path resolves to something that
+    doesn't implement `RoutingPlugin`, rather than deferring to a confusing
+    `AttributeError` on the first request that reaches the plugin pipeline.
+    """
+    plugin_paths = complexity_router_config.get("plugins")
+    if not isinstance(plugin_paths, list):
+        return
+
+    resolved_plugins = [
+        get_instance_fn(value=plugin_path, config_file_path=config_file_path)
+        if isinstance(plugin_path, str)
+        else plugin_path
+        for plugin_path in plugin_paths
+    ]
+    for plugin_path, resolved_plugin in zip(plugin_paths, resolved_plugins):
+        # `@runtime_checkable` only checks that `run` exists as an attribute, not that
+        # it's a coroutine function -- a synchronous `def run(self, context)` would pass
+        # isinstance() here and only fail at request time with a confusing `TypeError:
+        # object RoutingContext can't be used in 'await' expression`.
+        if not isinstance(resolved_plugin, RoutingPlugin) or not inspect.iscoroutinefunction(
+            getattr(resolved_plugin, "run", None)
+        ):
+            raise ValueError(
+                f"complexity_router_config.plugins entry {plugin_path!r} on model {model_name!r} "
+                f"resolved to {resolved_plugin!r}, which does not implement the RoutingPlugin "
+                "interface (an async `run(context)` method). Fix the referenced module before "
+                "starting the proxy."
+            )
+    complexity_router_config["plugins"] = resolved_plugins
+
+
 class ProxyConfig:
     """
     Abstraction class on top of config loading/updating logic. Gives us one place to control all config updating logic.
@@ -4720,6 +4760,13 @@ async def load_config(self, router: Optional[litellm.Router], config_file_path:
                 for k, v in model["litellm_params"].items():
                     if isinstance(v, str) and v.startswith("os.environ/"):
                         model["litellm_params"][k] = get_secret(v)
+                complexity_router_config = model["litellm_params"].get("complexity_router_config")
+                if isinstance(complexity_router_config, dict):
+                    resolve_complexity_router_plugins(
+                        model_name=model.get("model_name", ""),
+                        complexity_router_config=complexity_router_config,
+                        config_file_path=config_file_path,
+                    )
                 print(f"\033[32m    {model.get('model_name', '')}\033[0m")  # noqa: T201
                 litellm_model_name = model["litellm_params"]["model"]
                 litellm_model_api_base = model["litellm_params"].get("api_base", None)

litellm/router.py

@@ -7284,6 +7284,14 @@ async def async_callback_filter_deployments(
                     raise e
         return returned_healthy_deployments
 
+    @staticmethod
+    def _json_default_stable_id(value: object) -> str:
+        """json.dumps default= for _generate_model_id: plain str() on an arbitrary
+        object (e.g. a RoutingPlugin instance) falls back to object.__repr__'s
+        `<module.Class object at 0x...>`, so the hash -- and deployment id -- would
+        change every restart. Use the class name instead, stable across restarts."""
+        return f"{type(value).__module__}.{type(value).__qualname__}"
+
     def _generate_model_id(self, model_group: str, litellm_params: dict):
         """
         Helper function to consistently generate the same id for a deployment
@@ -7299,14 +7307,14 @@ def _generate_model_id(self, model_group: str, litellm_params: dict):
             if isinstance(k, str):
                 parts.append(k)
             elif isinstance(k, dict):
-                parts.append(json.dumps(k))
+                parts.append(json.dumps(k, default=self._json_default_stable_id))
             else:
                 parts.append(str(k))
 
             if isinstance(v, str):
                 parts.append(v)
             elif isinstance(v, dict):
-                parts.append(json.dumps(v))
+                parts.append(json.dumps(v, default=self._json_default_stable_id))
             else:
                 parts.append(str(v))
 

litellm/router_strategy/complexity_router/complexity_router.py

@@ -468,6 +468,38 @@ def _pick_from_tier_value(model: str | list[str], tier_key: str) -> str:
     def _tier_pools(self) -> dict[str, list[str]]:
         return {tier: (models if isinstance(models, list) else [models]) for tier, models in self.config.tiers.items()}
 
+    async def _pick_model_for_tier(
+        self,
+        tier: ComplexityTier,
+        raw_messages: list[dict[str, Any]] | None,
+        resolved_messages: list[dict[str, Any]] | None,
+        request_kwargs: dict,
+    ) -> str:
+        if not self.config.plugins:
+            return self.get_model_for_tier(tier)
+
+        from litellm.types.router import RoutingContext
+
+        tier_key = tier.value
+        metadata_key = "litellm_metadata" if "litellm_metadata" in request_kwargs else "metadata"
+        context = RoutingContext(
+            raw_messages=raw_messages or [],
+            structured_messages=resolved_messages or [],
+            candidate_models=list(self._tier_pools().get(tier_key, [])),
+            metadata=request_kwargs.get(metadata_key) or {},
+        )
+        for plugin in self.config.plugins:
+            context = await plugin.run(context)
+
+        if not context.candidate_models:
+            # A plugin narrowing a tier to zero candidates is a policy decision (e.g. no
+            # model this tenant's budget allows) -- falling back to default_model here
+            # (which was never checked against the plugins) would let that policy be
+            # silently bypassed. Raise instead, matching the Router-level plugin
+            # pipeline's own fail-closed behavior for the same situation.
+            raise ValueError(f"No candidate models left for tier {tier_key} after routing-plugin filtering")
+        return self._pick_from_tier_value(context.candidate_models, tier_key)
+
     def _ensure_adaptive_router(self) -> Any | None:
         if not self.config.adaptive:
             return None
@@ -861,10 +893,16 @@ async def async_pre_routing_hook(
         When `session_affinity` is enabled and a session_id is resolvable on the request,
         pins the model chosen on the session's first turn and reuses it for every later
         turn, skipping classification entirely. Otherwise delegates to `_classify_and_route`.
+
+        Skipped entirely when `plugins` are configured: reusing a stale pin would bypass
+        the plugin pipeline on every turn after the first, since a pinned model was never
+        re-checked against a policy plugin whose decision can change between turns (e.g. a
+        budget plugin, once the session's spend crosses its cap).
         """
         from litellm.types.router import PreRoutingHookResponse
 
-        session_id = self._get_session_id_from_request_kwargs(request_kwargs) if self.config.session_affinity else None
+        use_session_affinity = self.config.session_affinity and not self.config.plugins
+        session_id = self._get_session_id_from_request_kwargs(request_kwargs) if use_session_affinity else None
         cache_key = self._get_session_affinity_cache_key(session_id, request_kwargs) if session_id is not None else None
 
         if cache_key is not None:
@@ -947,14 +985,26 @@ async def _classify_and_route(
 
         if user_message is None:
             verbose_router_logger.debug("ComplexityRouter: No user message found, routing to default model")
+            if not self.config.plugins and self.config.default_model:
+                # No plugins configured: preserve the pre-existing default_model-first
+                # priority exactly (changing it would be a silent behavior change for
+                # every non-plugin user, not just a security fix).
+                routed_model = self.config.default_model
+            else:
+                # Plugins configured: default_model must never bypass them, so it's not
+                # checked here at all -- _pick_model_for_tier -> get_model_for_tier still
+                # falls back to it (after the MEDIUM tier) once the plugin pipeline runs.
+                routed_model = await self._pick_model_for_tier(
+                    ComplexityTier.MEDIUM, messages, resolved_messages, request_kwargs
+                )
             return PreRoutingHookResponse(
-                model=self.config.default_model or self.get_model_for_tier(ComplexityTier.MEDIUM),
+                model=routed_model,
                 messages=messages if has_original_messages else None,
             )
 
         override_tier = await self._resolve_keyword_tier_override(user_message, request_kwargs)
         if override_tier is not None:
-            routed_model = self.get_model_for_tier(override_tier)
+            routed_model = await self._pick_model_for_tier(override_tier, messages, resolved_messages, request_kwargs)
             cause = "semantic_keyword_match" if self.config.semantic_keyword_matching else "literal_keyword_match"
             verbose_router_logger.info(
                 f"ComplexityRouter: routing decision cause={cause}, "
@@ -980,7 +1030,7 @@ async def _classify_and_route(
                 f"signals={signals}, routed_model={routed_model}"
             )
         else:
-            routed_model = self.get_model_for_tier(tier)
+            routed_model = await self._pick_model_for_tier(tier, messages, resolved_messages, request_kwargs)
             verbose_router_logger.info(
                 f"ComplexityRouter: routing decision cause=complexity_scorer, tier={tier.value}, "
                 f"score={score:.3f}, signals={signals}, routed_model={routed_model}"

litellm/router_strategy/complexity_router/config.py

@@ -10,7 +10,7 @@
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
-from litellm.types.router import AdaptiveRouterWeights
+from litellm.types.router import AdaptiveRouterWeights, RoutingPlugin
 
 
 class ComplexityTier(str, Enum):
@@ -375,7 +375,12 @@ class ComplexityRouterConfig(BaseModel):
         description="TTL for the session affinity pin; refreshed on every cache hit",
     )
 
-    model_config = ConfigDict(extra="allow")  # Allow additional fields
+    plugins: list[RoutingPlugin] | None = Field(
+        default=None,
+        description="RoutingPlugin instances that narrow the classified tier's candidate models before selection",
+    )
+
+    model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)  # Allow additional fields
 
     @field_validator("tiers", mode="before")
     @classmethod
@@ -421,6 +426,15 @@ def _validate_semantic_matching(self) -> "ComplexityRouterConfig":
             raise ValueError("keyword_tier_rules must be non-empty when semantic_keyword_matching is enabled")
         return self
 
+    @model_validator(mode="after")
+    def _validate_plugins_adaptive_combo(self) -> "ComplexityRouterConfig":
+        if self.plugins and self.adaptive:
+            raise ValueError(
+                "plugins and adaptive=True cannot both be set: adaptive's bandit selection doesn't yet "
+                "consume plugin-narrowed candidate pools. Disable adaptive or remove plugins."
+            )
+        return self
+
 
 # Combined default config
 DEFAULT_COMPLEXITY_CONFIG = ComplexityRouterConfig()

litellm/types/router.py

@@ -9,7 +9,7 @@
 
 import httpx
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
-from typing_extensions import Protocol, Required, TypedDict
+from typing_extensions import Protocol, Required, TypedDict, runtime_checkable
 
 from litellm._uuid import uuid
 
@@ -852,6 +852,7 @@ class RoutingContext(BaseModel):
     signals: dict[str, Any] = Field(default_factory=dict)
 
 
+@runtime_checkable
 class RoutingPlugin(Protocol):
     """Interface a custom routing plugin must implement to run in `Router(plugins=[...])`."""
 

tests/test_litellm/proxy/proxy_server/test_proxy_config.py

@@ -21,6 +21,7 @@
     _is_remote_module_url,
     _scrub_db_overlay_remote_module_loads,
     _scrub_guardrail_inner,
+    resolve_complexity_router_plugins,
 )
 
 from .conftest import normalize
@@ -112,6 +113,78 @@ def test__scrub_db_overlay_remote_module_loads_invalid_non_dict_returns_input():
     assert _scrub_db_overlay_remote_module_loads("litellm_settings", "raw") == "raw"
 
 
+# ---------------------------------------------------------------------------
+# resolve_complexity_router_plugins
+# ---------------------------------------------------------------------------
+
+
+def test_resolve_complexity_router_plugins_no_plugins_key_is_a_noop():
+    config: Dict[str, Any] = {"tiers": {"SIMPLE": "gpt-4o-mini"}}
+    resolve_complexity_router_plugins(
+        model_name="smart-router", complexity_router_config=config, config_file_path=None
+    )
+    assert config == {"tiers": {"SIMPLE": "gpt-4o-mini"}}
+
+
+def test_resolve_complexity_router_plugins_resolves_dotted_path_to_live_instance(tmp_path):
+    plugin_file = tmp_path / "my_plugin.py"
+    plugin_file.write_text(
+        "class _Plugin:\n"
+        "    async def run(self, context):\n"
+        "        return context\n"
+        "\n"
+        "my_plugin_instance = _Plugin()\n"
+    )
+    config: Dict[str, Any] = {"plugins": ["my_plugin.my_plugin_instance"]}
+
+    resolve_complexity_router_plugins(
+        model_name="smart-router",
+        complexity_router_config=config,
+        config_file_path=str(tmp_path / "config.yaml"),
+    )
+
+    assert len(config["plugins"]) == 1
+    assert hasattr(config["plugins"][0], "run")
+    assert type(config["plugins"][0]).__name__ == "_Plugin"
+
+
+def test_resolve_complexity_router_plugins_rejects_non_routing_plugin_object(tmp_path):
+    plugin_file = tmp_path / "bad_plugin.py"
+    plugin_file.write_text("not_a_plugin = object()\n")
+    config: Dict[str, Any] = {"plugins": ["bad_plugin.not_a_plugin"]}
+
+    with pytest.raises(ValueError, match="does not implement the RoutingPlugin interface"):
+        resolve_complexity_router_plugins(
+            model_name="smart-router",
+            complexity_router_config=config,
+            config_file_path=str(tmp_path / "config.yaml"),
+        )
+
+
+def test_resolve_complexity_router_plugins_rejects_synchronous_run_method(tmp_path):
+    """Regression: @runtime_checkable only checks that `run` exists as an attribute,
+    not that it's a coroutine function. A plugin with a synchronous `run` passes a bare
+    isinstance() check and would only fail at request time with a confusing
+    `TypeError: object RoutingContext can't be used in 'await' expression`. Reported
+    by Greptile on PR #33251."""
+    plugin_file = tmp_path / "sync_plugin.py"
+    plugin_file.write_text(
+        "class _SyncPlugin:\n"
+        "    def run(self, context):\n"
+        "        return context\n"
+        "\n"
+        "sync_plugin_instance = _SyncPlugin()\n"
+    )
+    config: Dict[str, Any] = {"plugins": ["sync_plugin.sync_plugin_instance"]}
+
+    with pytest.raises(ValueError, match="does not implement the RoutingPlugin interface"):
+        resolve_complexity_router_plugins(
+            model_name="smart-router",
+            complexity_router_config=config,
+            config_file_path=str(tmp_path / "config.yaml"),
+        )
+
+
 # ---------------------------------------------------------------------------
 # ProxyConfig.__init__
 # ---------------------------------------------------------------------------