fix(proxy): derive auto-router health from its underlying models (#38174)

An auto_router deployment is a marker, not something a probe can contact, so
`_run_model_health_check` returns `{}` for it and it lands healthy whatever is
behind it. This derives its verdict from the models it actually resolves.

Rules and owners:

- `strategy_router_dependencies` is the single answer to "what does this router
  call": tier, default, classifier and embedding names per router kind, aligned
  with what init and the request path actually use.
- `_health_check_eligible` is the single probe-eligibility gate, applied to the
  requested set and to the pool a router's dependencies are drawn from alike, so
  an opted-out deployment cannot re-enter through a router that depends on it.
- `_resolved_deployment_ids` resolves names through `get_model_list`, the same
  composition of alias, routing-group and wildcard channels a request uses.
- A dependency reds its router only when *every* deployment behind the name is
  known unhealthy. A replica this run never judged, hidden from the caller or
  opted out of health checks, can still serve what the dead one drops, so
  partial evidence leaves the verdict green. Absent information never reds.
- Verdicts settle over rounds, because a marker never fails a probe of its own
  and a parent whose tier is a red router must inherit that fault. Both sweeps
  are bounded loops, so a router cycle terminates green.
- Dependency probes are added only on the targeted `/health?model_id=` path the
  dashboard uses per deployment, and are dropped from the response.

Resolves LIT-6073

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
tin-berri 2026-08-26 16:41:54 -07:00 committed by GitHub
parent b98b2d562b
commit d8edfb69c2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 860 additions and 50 deletions

View file

@ -6,11 +6,15 @@ import random
import sys
import threading
import time
from collections.abc import Mapping
from typing import Final
from collections.abc import Mapping, Sequence
from types import MappingProxyType
from typing import TYPE_CHECKING, Final
import litellm
if TYPE_CHECKING:
from litellm.router import Router
logger: Final = logging.getLogger(__name__)
from litellm.constants import (
BACKGROUND_HEALTH_CHECK_MAX_TOKENS,
@ -18,7 +22,11 @@ from litellm.constants import (
DEFAULT_HEALTH_CHECK_PROMPT,
HEALTH_CHECK_TIMEOUT_SECONDS,
)
from litellm.router_utils.auto_router_model_naming import classify_strategy_router_model
from litellm.router_utils.auto_router_model_naming import (
StrategyRouterDependency,
classify_strategy_router_model,
strategy_router_dependencies,
)
ILLEGAL_DISPLAY_PARAMS: Final = [
"messages",
@ -160,7 +168,7 @@ def health_check_filter_kwargs_from_general_settings(
def filter_deployments_by_id(
model_list: list,
model_list: Sequence[Mapping[str, object]],
) -> list:
seen_ids: Final = set()
filtered_deployments: Final = []
@ -192,12 +200,240 @@ async def run_with_timeout(task, timeout):
return {"error": "Timeout exceeded", "exception": timeout_exception}
def _skips_health_checks(deployment: Mapping[str, object]) -> bool:
info: Final = deployment.get("model_info")
return bool(info.get("disable_background_health_check", False)) if isinstance(info, Mapping) else False
def _health_check_eligible(
model_list: Sequence[Mapping[str, object]], skip_disabled: bool
) -> tuple[Mapping[str, object], ...]:
"""Deployments this run is allowed to contact.
The one eligibility gate, applied to the requested set and to the pool a router's
dependencies are drawn from alike, so an opted-out deployment cannot re-enter through a
router that depends on it.
"""
return tuple(x for x in model_list if not (skip_disabled and _skips_health_checks(x)))
def _deployment_model(deployment: Mapping[str, object]) -> str | None:
params: Final = deployment.get("litellm_params")
return params.get("model") if isinstance(params, Mapping) else None
def _narrow_to_target(
model_list: Sequence[Mapping[str, object]], model: str | None, model_id: str | None
) -> tuple[Mapping[str, object], ...]:
"""Narrow to the requested deployment. An id matching nothing keeps the whole list."""
if model_id is not None:
by_id: Final = tuple(x for x in model_list if _deployment_id(x) == model_id)
return by_id or tuple(model_list)
if model is None:
return tuple(model_list)
by_param: Final = tuple(x for x in model_list if _deployment_model(x) == model)
return by_param or tuple(x for x in model_list if x.get("model_name") == model)
def _is_strategy_router_deployment(litellm_params: Mapping[str, object]) -> bool:
"""True for strategy-router deployments."""
model: Final[object] = litellm_params.get("model", "")
return isinstance(model, str) and classify_strategy_router_model(model) is not None
def _is_marker(deployment: Mapping[str, object]) -> bool:
params: Final = deployment.get("litellm_params")
return isinstance(params, Mapping) and _is_strategy_router_deployment(params)
def _deployment_id(deployment: Mapping[str, object]) -> str | None:
info: Final = deployment.get("model_info")
ident: Final = info.get("id") if isinstance(info, Mapping) else None
return str(ident) if ident else None
def _resolved_deployment_ids(router: "Router", model_name: str) -> frozenset[str] | None:
"""Deployment ids backing `model_name`, or None when the name resolves to nothing.
`get_model_list` composes every channel the request path itself uses (exact name,
model_group_alias, routing groups, wildcards); a mirror of any one channel would call a
working tier broken. An alias whose target is gone resolves to nothing, which fails a
request exactly like an unknown name.
"""
resolved: Final = router.get_model_list(model_name=model_name)
if not resolved:
return None
return frozenset(ident for entry in resolved if (ident := _deployment_id(entry)))
def _dependency_failure(
dependency: StrategyRouterDependency,
router: "Router",
unhealthy_ids: frozenset[str],
) -> str | None:
"""Why this dependency makes its router unable to serve, or None when it does not.
A name reds its router only when *every* deployment behind it is known unhealthy. One
replica this run never judged, hidden from the caller or opted out of health checks, can
still serve what the dead one drops, so partial evidence leaves the verdict green.
"""
resolved: Final = _resolved_deployment_ids(router, dependency.model_name)
if resolved is None:
return f"{dependency.role} model '{dependency.model_name}' matches no deployment on this proxy"
if not resolved or not resolved <= unhealthy_ids:
return None
return f"{dependency.role} model '{dependency.model_name}' has no healthy deployment"
def _strategy_router_dependency_error(
deployment: Mapping[str, object],
router: "Router",
unhealthy_ids: frozenset[str],
) -> str | None:
"""The first dependency fault that makes this router unable to serve, if any."""
params: Final = deployment.get("litellm_params")
if not isinstance(params, Mapping):
return None
return next(
(
failure
for dependency in strategy_router_dependencies(params)
if (failure := _dependency_failure(dependency, router, unhealthy_ids))
),
None,
)
def _deployments_by_id(
universe: Sequence[Mapping[str, object]], ids: frozenset[str]
) -> tuple[Mapping[str, object], ...]:
"""The deployments for `ids`, one row per id.
Reuses the requested set's own dedupe rule, so an alias that duplicates a row cannot get
it probed twice or split a single id's verdict across two disagreeing results.
"""
matched: Final = tuple(d for d in universe if (uid := _deployment_id(d)) and uid in ids)
return tuple(filter_deployments_by_id(model_list=matched))
def _dependency_deployments_to_probe(
checked: Sequence[Mapping[str, object]],
universe: Sequence[Mapping[str, object]],
router: "Router",
) -> tuple[Mapping[str, object], ...]:
"""Deployments backing the checked routers' dependencies that are not already checked.
Empty on a full-list run, which therefore gains no probe; it is the targeted
`/health?model_id=<router>` call the dashboard makes per deployment that needs them,
since a router's verdict is a statement about models the request never named. Drawn from
`universe`, the caller's access-filtered list, so no deployment is probed that the caller
was not already granted. Expansion follows routers through routers, one hop per round,
because a child router's own models must be probed for the parent to fail; stopping when
a round adds nothing is what makes a router cycle terminate.
"""
checked_ids: Final = frozenset(cid for d in checked if (cid := _deployment_id(d)))
reached = checked_ids # rebind-ok: the sweep's cursor, one hop wider per round
frontier = tuple(checked) # rebind-ok: the routers whose dependencies the next round expands
for _ in range(len(universe)):
names = frozenset(
dependency.model_name
for deployment in frontier
if isinstance(params := deployment.get("litellm_params"), Mapping)
for dependency in strategy_router_dependencies(params)
)
fresh_ids = (
frozenset(ident for name in names for ident in (_resolved_deployment_ids(router, name) or ())) - reached
)
if not fresh_ids:
break
frontier = _deployments_by_id(universe, fresh_ids)
reached = reached | fresh_ids
return _deployments_by_id(universe, reached - checked_ids)
def _strategy_router_verdicts(
healthy_endpoints: Sequence[Mapping[str, object]],
unhealthy_endpoints: Sequence[Mapping[str, object]],
checked: Sequence[Mapping[str, object]],
router: "Router",
) -> Mapping[str, str]:
"""The dependency fault, per model id, for every strategy router that cannot serve.
A marker is filed healthy by `_run_model_health_check` returning `{}`, which says only
that nothing was probed. This is where that placeholder becomes a verdict, derived from
this run's own results rather than a re-probe or a cache that is empty unless
`enable_health_check_routing` is on. A marker never fails a probe of its own, so verdicts
settle over rounds, each feeding the last round's reds back in as unhealthy; without that
the parent of a red child would stay green. Bounded by the marker count, which is what
makes a router cycle terminate green rather than spin.
"""
by_id: Final = MappingProxyType({i: d for d in checked if (i := _deployment_id(d))})
markers: Final = MappingProxyType(
{
marker_id: by_id[marker_id]
for endpoint in healthy_endpoints
if isinstance(marker_id := endpoint.get("model_id"), str) and marker_id in by_id
if _is_marker(by_id[marker_id])
}
)
probe_failures: Final = frozenset(
ident for endpoint in unhealthy_endpoints if isinstance(ident := endpoint.get("model_id"), str)
)
settled: Mapping[str, str] = MappingProxyType({}) # rebind-ok: the fixed point, a round's verdicts at a time
for _ in range(len(markers)):
fresh = MappingProxyType(
{
marker_id: error
for marker_id, deployment in markers.items()
if marker_id not in settled
if (error := _strategy_router_dependency_error(deployment, router, probe_failures | frozenset(settled)))
}
)
if not fresh:
break
settled = MappingProxyType({**settled, **fresh})
return settled
def _finalize_strategy_router_endpoints(
healthy_endpoints: Sequence[Mapping[str, object]],
unhealthy_endpoints: Sequence[Mapping[str, object]],
checked: Sequence[Mapping[str, object]],
router: "Router | None",
dependency_probes: Sequence[Mapping[str, object]],
) -> tuple[Sequence[Mapping[str, object]], Sequence[Mapping[str, object]]]:
"""Apply router verdicts, then drop the deployments probed only to reach them.
The probes exist to judge the routers that depend on them; reporting them would answer a
targeted request with deployments the caller never asked about.
"""
verdicts: Final = (
_strategy_router_verdicts(healthy_endpoints, unhealthy_endpoints, checked, router)
if router is not None
else MappingProxyType({})
)
dropped: Final = frozenset(i for d in dependency_probes if (i := _deployment_id(d)))
def keep(endpoint: Mapping[str, object]) -> bool:
model_id: Final = endpoint.get("model_id")
return not (isinstance(model_id, str) and model_id in dropped)
def verdict_for(endpoint: Mapping[str, object]) -> str | None:
model_id: Final = endpoint.get("model_id")
return verdicts.get(model_id) if isinstance(model_id, str) else None
kept_healthy: Final = tuple(e for e in healthy_endpoints if keep(e))
return (
tuple(e for e in kept_healthy if verdict_for(e) is None),
tuple(e for e in unhealthy_endpoints if keep(e))
+ tuple(
dict(e, error=error) # mutable-ok: the /health payload must stay a plain JSON-serializable dict
for e in kept_healthy
if (error := verdict_for(e)) is not None
),
)
async def _run_model_health_check(model: dict):
litellm_params = model["litellm_params"]
model_info: Final = model.get("model_info", {})
@ -540,6 +776,7 @@ async def perform_health_check(
max_concurrency: int | None = None,
instrumentation_context: dict | None = None,
health_check_skip_disabled_background_models: bool = False,
router: "Router | None" = None,
):
"""
Perform a health check on the system.
@ -576,23 +813,9 @@ async def perform_health_check(
cycle_start_time: Final = time.monotonic()
requested_model_count: Final = len(model_list)
# Filter by model_id first so a single deployment is checked when id is specified
if model_id is not None:
_by_id: Final = [x for x in model_list if (x.get("model_info") or {}).get("id") == model_id]
if _by_id:
model_list = _by_id
elif model is not None:
_new_model_list = [x for x in model_list if x["litellm_params"]["model"] == model]
if _new_model_list == []:
_new_model_list = [x for x in model_list if x["model_name"] == model]
model_list = _new_model_list
if health_check_skip_disabled_background_models:
model_list = [
x for x in model_list if not (x.get("model_info") or {}).get("disable_background_health_check", False)
]
if not model_list:
skip_disabled: Final = health_check_skip_disabled_background_models
narrowed: Final = _health_check_eligible(_narrow_to_target(model_list, model, model_id), skip_disabled)
if not narrowed:
if instrumentation_enabled:
logger.debug(
"health_check_cycle_skipped source=%s cycle_id=%s reason=no_models_after_filter",
@ -601,11 +824,16 @@ async def perform_health_check(
)
return [], [], {}
post_filter_model_count: Final = len(model_list)
model_list = filter_deployments_by_id(
model_list=model_list
) # filter duplicate deployments (e.g. when model alias'es are used)
deduped_model_count: Final = len(model_list)
post_filter_model_count: Final = len(narrowed)
requested: Final = filter_deployments_by_id(model_list=narrowed)
deduped_model_count: Final = len(requested)
dependency_probes: Final = (
_dependency_deployments_to_probe(requested, _health_check_eligible(model_list, skip_disabled), router)
if router is not None
else ()
)
checked: Final = requested + list(dependency_probes) # mutable-ok: _perform_health_check takes a list
if instrumentation_enabled:
logger.debug(
@ -622,15 +850,20 @@ async def perform_health_check(
try:
(
healthy_endpoints,
unhealthy_endpoints,
probed_healthy,
probed_unhealthy,
exceptions_by_model_id,
) = await _perform_health_check(
model_list,
checked,
details,
max_concurrency=max_concurrency,
instrumentation_context=instrumentation_context,
)
graded_healthy, graded_unhealthy = _finalize_strategy_router_endpoints(
probed_healthy, probed_unhealthy, checked, router, dependency_probes
)
healthy_endpoints: Final = list(graded_healthy)
unhealthy_endpoints: Final = list(graded_unhealthy)
except Exception:
if instrumentation_enabled:
logger.exception(

View file

@ -1,7 +1,7 @@
import asyncio
import json
import time
from typing import Any, Final
from typing import TYPE_CHECKING, Any, Final
from litellm._logging import verbose_proxy_logger
from litellm.caching.redis_cache import RedisCache
@ -12,6 +12,9 @@ from litellm.constants import (
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
from litellm.proxy.health_check import perform_health_check
if TYPE_CHECKING:
from litellm.router import Router
class SharedHealthCheckManager:
"""
@ -185,6 +188,7 @@ class SharedHealthCheckManager:
details: bool = True,
max_concurrency: int | None = None,
health_check_skip_disabled_background_models: bool = False,
router: "Router | None" = None,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]], dict[str, Any]]:
"""
Perform health check with shared state coordination.
@ -235,6 +239,7 @@ class SharedHealthCheckManager:
details=details,
max_concurrency=max_concurrency,
health_check_skip_disabled_background_models=health_check_skip_disabled_background_models,
router=router,
)
# Cache the results
@ -254,6 +259,7 @@ class SharedHealthCheckManager:
details=details,
max_concurrency=max_concurrency,
health_check_skip_disabled_background_models=health_check_skip_disabled_background_models,
router=router,
)
# Lock not acquired — poll for cached results until the lock
@ -309,6 +315,7 @@ class SharedHealthCheckManager:
details=details,
max_concurrency=max_concurrency,
health_check_skip_disabled_background_models=health_check_skip_disabled_background_models,
router=router,
)
async def is_health_check_in_progress(self) -> bool:

View file

@ -1113,6 +1113,7 @@ async def health_endpoint(
user_id=user_api_key_dict.user_id,
model_id=model_id,
max_concurrency=health_check_concurrency,
router=llm_router,
**_hc_filter,
)
return _post_process(router_result)

View file

@ -3393,9 +3393,7 @@ def _rss_mb_for_log() -> str:
return f"{rss_mb:.2f}"
def _is_unexpected_keyword_argument_type_error(exc: BaseException) -> bool:
"""True when ``exc`` is a TypeError from passing a kwarg the callee does not accept."""
return isinstance(exc, TypeError) and ("unexpected keyword argument" in str(exc).lower())
_UNEXPECTED_KWARG: Final = re.compile(r"unexpected keyword argument '(?P<name>[^']+)'")
async def _run_direct_health_check_with_instrumentation(
@ -3404,31 +3402,33 @@ async def _run_direct_health_check_with_instrumentation(
max_concurrency: int | None,
instrumentation_context: dict,
):
"""Call ``perform_health_check``, retrying with fewer kwargs on unexpected-kw TypeErrors."""
_hc_filter: Final = health_check_filter_kwargs_from_general_settings(general_settings)
last_type_error: TypeError | None = None
for extra_kwargs in (
"""Call ``perform_health_check``, dropping exactly the optional kwarg each TypeError names.
A callee that predates an argument rejects it by name, so only that one is dropped. A
hand-written ladder of combinations would drop working options alongside it, and would
need a new rung every time an argument is added.
"""
optional: Mapping[str, object] = MappingProxyType( # rebind-ok: loses the kwarg the callee rejected
{
"router": llm_router,
"instrumentation_context": instrumentation_context,
**_hc_filter,
},
{"instrumentation_context": instrumentation_context},
dict(_hc_filter),
{},
):
**health_check_filter_kwargs_from_general_settings(general_settings),
}
)
for _ in range(len(optional) + 1):
try:
return await perform_health_check(
model_list=model_list,
details=details,
max_concurrency=max_concurrency,
**extra_kwargs,
**optional,
)
except TypeError as e:
if not _is_unexpected_keyword_argument_type_error(e):
rejected = _UNEXPECTED_KWARG.search(str(e))
if rejected is None or rejected["name"] not in optional:
raise
last_type_error = e
assert last_type_error is not None
raise last_type_error
optional = MappingProxyType({k: v for k, v in optional.items() if k != rejected["name"]})
raise AssertionError("perform_health_check rejected every optional argument")
def _schedule_background_health_check_db_save(
@ -3683,6 +3683,7 @@ async def _run_background_health_check():
model_list=_llm_model_list,
details=details_bool,
max_concurrency=health_check_concurrency,
router=llm_router,
**_hc_filter,
)
except Exception as e:

View file

@ -10,13 +10,26 @@ the router silently dropping the deployment at load time under
``ignore_invalid_deployments``.
"""
from collections.abc import Mapping
from typing import Final, Literal
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from types import MappingProxyType
from typing import Final, Literal, TypeAlias
AUTO_ROUTER_MODEL_PREFIX: Final = "auto_router/"
StrategyRouterKind = Literal["semantic", "complexity", "adaptive", "quality"]
StrategyRouterDependencyRole: TypeAlias = Literal["tier", "default", "classifier", "embedding"]
@dataclass(frozen=True, slots=True)
class StrategyRouterDependency:
"""A model name a strategy router must be able to reach to do its job."""
model_name: str
role: StrategyRouterDependencyRole
STRATEGY_ROUTER_PARAM_FIELDS: Final[frozenset[str]] = frozenset(
{
"auto_router_config",
@ -63,6 +76,84 @@ def classify_strategy_router_model(model: str) -> StrategyRouterKind | None:
return "semantic"
def _named(value: object, role: StrategyRouterDependencyRole) -> tuple[StrategyRouterDependency, ...]:
"""One dependency from a scalar field, or none when it is absent or not a name."""
return (StrategyRouterDependency(value, role),) if isinstance(value, str) and value else ()
def _pool(value: object, role: StrategyRouterDependencyRole) -> tuple[StrategyRouterDependency, ...]:
"""Dependencies from a field holding either a single name or a pool of them."""
if isinstance(value, str):
return _named(value, role)
if isinstance(value, Sequence):
return tuple(dep for entry in value for dep in _named(entry, role))
return ()
_NO_CONFIG: Final[Mapping[str, object]] = MappingProxyType({})
def _mapping(value: object) -> Mapping[str, object]:
return value if isinstance(value, Mapping) else _NO_CONFIG
def strategy_router_dependencies(
litellm_params: Mapping[str, object],
) -> tuple[StrategyRouterDependency, ...]:
"""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
`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
config's `default_model`, so both are read. A complexity router ignores that field and
derives its default from the tiers instead (`fallback_tier`, then MEDIUM, then SIMPLE),
overwriting the config value at init, so only the `litellm_params` spelling is a
dependency here; the derived one is already covered as a tier.
Returns empty for a regular deployment, and for any name this module cannot reach from
the deployment dict alone: a semantic router's routes live in an `auto_router_config`
JSON string or an `auto_router_config_path` file, so only its default and embedding
models are enumerable here. Every field is read defensively, since a caller may hold a
config the router itself would refuse, and a health check must not raise on one.
"""
kind: Final = classify_strategy_router_model(str(litellm_params.get("model", "")))
if kind is None:
return ()
if kind == "semantic":
return _named(litellm_params.get("auto_router_default_model"), "default") + _named(
litellm_params.get("auto_router_embedding_model"), "embedding"
)
if kind == "adaptive":
return _pool(_mapping(litellm_params.get("adaptive_router_config")).get("available_models"), "tier")
if kind == "quality":
quality: Final = _mapping(litellm_params.get("quality_router_config"))
return tuple(
dict.fromkeys(
_pool(quality.get("available_models"), "tier")
+ _named(
litellm_params.get("quality_router_default_model") or quality.get("default_model"),
"default",
)
)
)
complexity: Final = _mapping(litellm_params.get("complexity_router_config"))
classifier: Final = _mapping(complexity.get("classifier_llm_config"))
return tuple(
dict.fromkeys(
tuple(dep for tier in _mapping(complexity.get("tiers")).values() for dep in _pool(tier, "tier"))
+ _named(litellm_params.get("complexity_router_default_model"), "default")
+ (_named(classifier.get("model"), "classifier") if complexity.get("classifier_type") == "llm" else ())
+ (
_named(complexity.get("embedding_model"), "embedding")
if complexity.get("semantic_keyword_matching")
else ()
)
)
)
def validate_complexity_router_config_write(complexity_router_config: Mapping[str, object] | None) -> str | None:
"""Reject a complexity config the router would refuse to build a deployment from.

View file

@ -2655,6 +2655,35 @@ async def test_run_direct_health_check_with_instrumentation_accepts_filter_only(
assert seen[0] is False
@pytest.mark.asyncio
async def test_run_direct_health_check_drops_only_the_rejected_kwarg(monkeypatch):
"""A callee that predates `router` must still get the skip-disabled filter: dropping the
rejected argument alongside working ones would probe deployments the operator opted out."""
import litellm.proxy.proxy_server as proxy_server
seen: list = []
async def fake_perform_health_check(
model_list,
details,
max_concurrency=None,
instrumentation_context=None,
health_check_skip_disabled_background_models=False,
):
seen.append((instrumentation_context, health_check_skip_disabled_background_models))
return ([], [], {})
monkeypatch.setattr(proxy_server, "perform_health_check", fake_perform_health_check)
monkeypatch.setattr(
proxy_server,
"general_settings",
{"health_check_skip_disabled_background_models": True},
)
await proxy_server._run_direct_health_check_with_instrumentation([], True, 1, {"cycle_id": "c3"})
assert seen == [({"cycle_id": "c3"}, True)]
@pytest.mark.asyncio
async def test_run_direct_health_check_with_instrumentation_non_kw_typeerror_reraises(
monkeypatch,

View file

@ -684,3 +684,326 @@ async def test_run_model_health_check_skips_complexity_router_deployment():
fake_ahealth_check.assert_not_called()
assert result == {}
def _router_health_fixture():
"""A real Router whose SIMPLE tier, default and classifier can each be pointed at a dead
group. That group has two replicas, so a verdict reached on only one of them is visible."""
return litellm.Router(
model_list=[
{
"model_name": "live-group",
"litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "sk-x"},
"model_info": {"id": "live-1"},
},
{
"model_name": "dead-group",
"litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "sk-x"},
"model_info": {"id": "dead-1"},
},
{
"model_name": "dead-group",
"litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "sk-x"},
"model_info": {"id": "dead-2"},
},
{
"model_name": "smart-router",
"litellm_params": {
"model": "auto_router/complexity_router",
"complexity_router_config": {"tiers": {"SIMPLE": "dead-group", "MEDIUM": "live-group"}},
"complexity_router_default_model": "live-group",
},
"model_info": {"id": "router-1"},
},
],
ignore_invalid_deployments=True,
)
def _marker_deployment(router):
return next(d for d in router.model_list if d["model_info"]["id"] == "router-1")
def test_strategy_router_reds_when_a_tier_group_has_no_healthy_deployment():
"""LIT-6073: the marker is filed healthy by the {} placeholder; the verdict must override it."""
router = _router_health_fixture()
healthy = [{"model_id": "router-1"}, {"model_id": "live-1"}]
unhealthy = [{"model_id": "dead-1", "error": "boom"}, {"model_id": "dead-2", "error": "boom"}]
new_healthy, new_unhealthy = hc_module._finalize_strategy_router_endpoints(
healthy, unhealthy, router.model_list, router, ()
)
assert [e["model_id"] for e in new_healthy] == ["live-1"]
moved = next(e for e in new_unhealthy if e["model_id"] == "router-1")
assert moved["error"] == "tier model 'dead-group' has no healthy deployment"
def test_strategy_router_stays_green_when_every_dependency_has_a_healthy_deployment():
"""The negative class: same router, same code path, nothing unhealthy behind it."""
router = _router_health_fixture()
healthy = [{"model_id": "router-1"}, {"model_id": "live-1"}, {"model_id": "dead-1"}, {"model_id": "dead-2"}]
new_healthy, new_unhealthy = hc_module._finalize_strategy_router_endpoints(
healthy, [], router.model_list, router, ()
)
assert {e["model_id"] for e in new_healthy} == {"router-1", "live-1", "dead-1", "dead-2"}
assert new_unhealthy == ()
def test_strategy_router_reds_when_a_dependency_name_matches_no_deployment():
"""An unresolvable tier name is a different fault from an unhealthy one, and says so."""
router = _router_health_fixture()
marker = _marker_deployment(router)
marker["litellm_params"]["complexity_router_config"]["tiers"]["SIMPLE"] = "typo-group"
_, new_unhealthy = hc_module._finalize_strategy_router_endpoints(
[{"model_id": "router-1"}], [], router.model_list, router, ()
)
assert new_unhealthy[0]["error"] == "tier model 'typo-group' matches no deployment on this proxy"
@pytest.mark.parametrize("judged", [("router-1", "live-1"), ("router-1", "live-1", "dead-1")])
def test_strategy_router_verdict_is_silent_when_part_of_a_group_went_unjudged(judged):
"""Absent information never reds a router, whether the whole group went unjudged (hidden
from the caller) or only a replica did (opted out of health checks). The replica this run
never contacted can still serve every request the dead one drops."""
router = _router_health_fixture()
scope = [d for d in router.model_list if d["model_info"]["id"] in judged]
new_healthy, new_unhealthy = hc_module._finalize_strategy_router_endpoints(
[{"model_id": "router-1"}], [{"model_id": "dead-1", "error": "boom"}], scope, router, ()
)
assert [e["model_id"] for e in new_healthy] == ["router-1"]
assert new_unhealthy == ({"model_id": "dead-1", "error": "boom"},)
def test_dependency_probe_expansion_is_a_no_op_when_every_dependency_is_already_checked():
"""The full-list run must gain no extra probe, or /health doubles its provider spend."""
router = _router_health_fixture()
assert hc_module._dependency_deployments_to_probe(router.model_list, router.model_list, router) == ()
def test_dependency_probe_expansion_adds_dependencies_for_a_targeted_router_check():
"""GET /health?model_id=<router> narrows to the marker, so the deps must be pulled back in."""
router = _router_health_fixture()
marker_only = [_marker_deployment(router)]
probes = hc_module._dependency_deployments_to_probe(marker_only, router.model_list, router)
assert {d["model_info"]["id"] for d in probes} == {"dead-1", "dead-2", "live-1"}
def test_dependency_probes_carry_one_row_per_id():
"""An alias can put the same deployment in the list twice, which is what
filter_deployments_by_id exists for. Probing it twice doubles the provider spend, and two
results for one id can disagree, reding the router on whichever landed in the loser."""
router = _router_health_fixture()
duplicated = tuple(router.model_list) + tuple(d for d in router.model_list if d["model_info"]["id"] == "dead-1")
probes = hc_module._dependency_deployments_to_probe([_marker_deployment(router)], duplicated, router)
assert [d["model_info"]["id"] for d in probes].count("dead-1") == 1
def test_a_dependency_alias_whose_target_is_gone_reds_the_router():
"""An alias resolving to nothing fails a request exactly like an unknown name, so the
health check must not read the empty resolution as "no information" and stay green."""
router = litellm.Router(
model_list=[
{
"model_name": "smart-router",
"litellm_params": {
"model": "auto_router/complexity_router",
"complexity_router_config": {"tiers": {"SIMPLE": "broken-alias"}},
"complexity_router_default_model": "broken-alias",
},
"model_info": {"id": "router-1"},
},
],
model_group_alias={"broken-alias": "target-that-no-longer-exists"},
ignore_invalid_deployments=True,
)
_, new_unhealthy = hc_module._finalize_strategy_router_endpoints(
[{"model_id": "router-1"}], [], router.model_list, router, ()
)
assert new_unhealthy[0]["error"] == "tier model 'broken-alias' matches no deployment on this proxy"
def test_a_dependency_that_opted_out_of_health_checks_is_never_probed():
"""skip-disabled is an operator opt-out. A router depending on that deployment must not
pull it back in and spend the proxy's provider credentials probing it."""
disabled_dep = {
"model_name": "dead-group",
"litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "sk-x"},
"model_info": {"id": "dead-1", "disable_background_health_check": True},
}
router = litellm.Router(
model_list=[
disabled_dep,
{
"model_name": "smart-router",
"litellm_params": {
"model": "auto_router/complexity_router",
"complexity_router_config": {"tiers": {"SIMPLE": "dead-group"}},
"complexity_router_default_model": "dead-group",
},
"model_info": {"id": "router-1"},
},
],
ignore_invalid_deployments=True,
)
marker = [d for d in router.model_list if d["model_info"]["id"] == "router-1"]
eligible = hc_module._health_check_eligible(router.model_list, skip_disabled=True)
probes = hc_module._dependency_deployments_to_probe(marker, eligible, router)
assert probes == ()
assert [d["model_info"]["id"] for d in eligible] == ["router-1"]
def test_narrowing_by_an_id_that_matches_nothing_keeps_the_whole_list():
"""Pinned because the disabled-dependency fix moved this filter into its own helper."""
deployments = [{"model_name": "a", "litellm_params": {"model": "openai/a"}, "model_info": {"id": "a-1"}}]
assert hc_module._narrow_to_target(deployments, None, "no-such-id") == tuple(deployments)
assert hc_module._narrow_to_target(deployments, None, "a-1") == tuple(deployments)
assert hc_module._narrow_to_target(deployments, "a", None) == tuple(deployments)
def _nested_router_fixture(parent_tier: str):
return litellm.Router(
model_list=[
{
"model_name": "dead-group",
"litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "sk-x"},
"model_info": {"id": "dead-1"},
},
{
"model_name": "child",
"litellm_params": {
"model": "auto_router/complexity_router",
"complexity_router_config": {"tiers": {"SIMPLE": "dead-group"}},
"complexity_router_default_model": "dead-group",
},
"model_info": {"id": "child-1"},
},
{
"model_name": "parent",
"litellm_params": {
"model": "auto_router/complexity_router",
"complexity_router_config": {"tiers": {"SIMPLE": parent_tier}},
"complexity_router_default_model": parent_tier,
},
"model_info": {"id": "parent-1"},
},
],
ignore_invalid_deployments=True,
)
def test_a_router_routing_to_a_red_router_is_itself_red():
"""A marker never fails a probe of its own, so a single pass sees only probe failures and
leaves the parent of a dead child green while every request through it fails."""
router = _nested_router_fixture("child")
new_healthy, new_unhealthy = hc_module._finalize_strategy_router_endpoints(
[{"model_id": "parent-1"}, {"model_id": "child-1"}],
[{"model_id": "dead-1", "error": "boom"}],
router.model_list,
router,
(),
)
errors = {e["model_id"]: e["error"] for e in new_unhealthy if e["model_id"] != "dead-1"}
assert errors["child-1"] == "tier model 'dead-group' has no healthy deployment"
assert errors["parent-1"] == "tier model 'child' has no healthy deployment"
assert new_healthy == ()
def test_a_router_routing_to_a_healthy_router_stays_green():
"""The negative class for nested propagation: the child serves, so the parent must not
inherit a red merely for depending on another router."""
router = _nested_router_fixture("child")
child = next(d for d in router.model_list if d["model_info"]["id"] == "child-1")
child["litellm_params"]["complexity_router_config"]["tiers"]["SIMPLE"] = "dead-group"
new_healthy, new_unhealthy = hc_module._finalize_strategy_router_endpoints(
[{"model_id": "parent-1"}, {"model_id": "child-1"}, {"model_id": "dead-1"}],
[],
router.model_list,
router,
(),
)
assert {e["model_id"] for e in new_healthy} == {"parent-1", "child-1", "dead-1"}
assert new_unhealthy == ()
def test_two_routers_pointing_at_each_other_terminate_instead_of_recursing():
"""The round bound is what makes a cycle finish. Neither has a failing dependency, so
neither reds, and the walk must not recurse forever proving it."""
router = litellm.Router(
model_list=[
{
"model_name": name,
"litellm_params": {
"model": "auto_router/complexity_router",
"complexity_router_config": {"tiers": {"SIMPLE": other}},
"complexity_router_default_model": other,
},
"model_info": {"id": f"{name}-1"},
}
for name, other in (("a", "b"), ("b", "a"))
],
ignore_invalid_deployments=True,
)
new_healthy, new_unhealthy = hc_module._finalize_strategy_router_endpoints(
[{"model_id": "a-1"}, {"model_id": "b-1"}], [], router.model_list, router, ()
)
assert {e["model_id"] for e in new_healthy} == {"a-1", "b-1"}
assert new_unhealthy == ()
def test_a_targeted_check_on_a_nested_router_probes_the_grandchild_models():
"""One hop is not enough. GET /health?model_id=<parent> narrows to the parent, and pulling
in only the child marker leaves the child's own models unprobed, so nothing ever fails and
both settle green on the exact path the Admin UI uses."""
router = _nested_router_fixture("child")
parent_only = [d for d in router.model_list if d["model_info"]["id"] == "parent-1"]
probes = hc_module._dependency_deployments_to_probe(parent_only, router.model_list, router)
assert {d["model_info"]["id"] for d in probes} == {"child-1", "dead-1"}
def test_transitive_probe_expansion_terminates_on_a_router_cycle():
"""Expansion follows routers through routers, so a cycle must stop rather than recurse."""
router = litellm.Router(
model_list=[
{
"model_name": name,
"litellm_params": {
"model": "auto_router/complexity_router",
"complexity_router_config": {"tiers": {"SIMPLE": other}},
"complexity_router_default_model": other,
},
"model_info": {"id": f"{name}-1"},
}
for name, other in (("a", "b"), ("b", "a"))
],
ignore_invalid_deployments=True,
)
a_only = [d for d in router.model_list if d["model_info"]["id"] == "a-1"]
probes = hc_module._dependency_deployments_to_probe(a_only, router.model_list, router)
assert {d["model_info"]["id"] for d in probes} == {"b-1"}

View file

@ -314,6 +314,7 @@ class TestSharedHealthCheckManager:
details=True,
max_concurrency=None,
health_check_skip_disabled_background_models=False,
router=None,
)
assert healthy == expected_healthy
assert unhealthy == expected_unhealthy
@ -404,6 +405,7 @@ class TestSharedHealthCheckManager:
details=True,
max_concurrency=None,
health_check_skip_disabled_background_models=False,
router=None,
)
assert healthy == expected_healthy
assert unhealthy == expected_unhealthy
@ -447,6 +449,7 @@ class TestSharedHealthCheckManager:
details=True,
max_concurrency=None,
health_check_skip_disabled_background_models=False,
router=None,
)
assert healthy == expected_healthy
assert unhealthy == expected_unhealthy
@ -519,6 +522,7 @@ class TestSharedHealthCheckManager:
details=True,
max_concurrency=None,
health_check_skip_disabled_background_models=False,
router=None,
)
assert healthy == expected_healthy
assert unhealthy == expected_unhealthy

View file

@ -2,6 +2,7 @@ import pytest
from litellm.router_utils.auto_router_model_naming import (
classify_strategy_router_model,
strategy_router_dependencies,
validate_complexity_router_config_write,
validate_strategy_router_model_write,
)
@ -179,3 +180,123 @@ def test_config_check_ignores_the_model_entirely():
)
is not None
)
@pytest.mark.parametrize(
"litellm_params, expected",
[
({"model": "openai/gpt-4o"}, ()),
(
{
"model": "auto_router/complexity_router",
"complexity_router_config": {"tiers": {"SIMPLE": "a", "MEDIUM": ["b", "c"]}},
"complexity_router_default_model": "d",
},
(("a", "tier"), ("b", "tier"), ("c", "tier"), ("d", "default")),
),
(
{
"model": "auto_router/complexity_router",
"complexity_router_config": {
"tiers": {"SIMPLE": "a"},
"classifier_type": "llm",
"classifier_llm_config": {"model": "clf"},
},
},
(("a", "tier"), ("clf", "classifier")),
),
(
{
"model": "auto_router/complexity_router",
"complexity_router_config": {
"tiers": {"SIMPLE": "a"},
"classifier_llm_config": {"model": "clf"},
},
},
(("a", "tier"),),
),
(
{"model": "auto_router/my_router", "auto_router_default_model": "d", "auto_router_embedding_model": "e"},
(("d", "default"), ("e", "embedding")),
),
(
{"model": "auto_router/adaptive_router", "adaptive_router_config": {"available_models": ["m1", "m2"]}},
(("m1", "tier"), ("m2", "tier")),
),
(
{
"model": "auto_router/quality_router",
"quality_router_config": {"available_models": ["q1"], "default_model": "qd"},
},
(("q1", "tier"), ("qd", "default")),
),
],
)
def test_strategy_router_dependencies(litellm_params, expected):
found = strategy_router_dependencies(litellm_params)
assert tuple((d.model_name, d.role) for d in found) == expected
def test_complexity_default_model_param_wins_over_the_config_field():
"""ComplexityRouter overwrites config.default_model with the litellm_params one, so the
config field is dead whenever the param is set and must not be able to red the router."""
found = strategy_router_dependencies(
{
"model": "auto_router/complexity_router",
"complexity_router_config": {"tiers": {}, "default_model": "shadowed"},
"complexity_router_default_model": "winner",
}
)
assert tuple(d.model_name for d in found) == ("winner",)
def test_complexity_ignores_its_config_default_model_and_quality_does_not():
"""Router init derives a complexity default from the tiers (fallback_tier, MEDIUM, SIMPLE)
and overwrites config.default_model, so that field names a model complexity never calls.
Quality init really does fall back to it, so the two must not be treated alike."""
complexity = strategy_router_dependencies(
{
"model": "auto_router/complexity_router",
"complexity_router_config": {"tiers": {"MEDIUM": "derived"}, "default_model": "never-called"},
}
)
quality = strategy_router_dependencies(
{
"model": "auto_router/quality_router",
"quality_router_config": {"available_models": ["q1"], "default_model": "really-used"},
}
)
assert tuple(d.model_name for d in complexity) == ("derived",)
assert tuple(d.model_name for d in quality) == ("q1", "really-used")
@pytest.mark.parametrize(
"config",
["not-a-dict", None, {"tiers": "not-a-dict"}, {"tiers": {"SIMPLE": 7}}, {"tiers": {"SIMPLE": [None, ""]}}],
)
def test_strategy_router_dependencies_never_raises_on_a_malformed_config(config):
"""A config the router itself would refuse must not take the whole /health response down."""
assert strategy_router_dependencies({"model": "auto_router/complexity_router", "complexity_router_config": config}) == ()
@pytest.mark.parametrize(
"semantic_on, expected",
[(False, ("t",)), (True, ("t", "emb"))],
)
def test_complexity_embedding_model_is_a_dependency_only_when_semantic_matching_is_on(semantic_on, expected):
"""The runtime reads embedding_model only under semantic_keyword_matching, so listing it
unconditionally would red a router that never calls it."""
found = strategy_router_dependencies(
{
"model": "auto_router/complexity_router",
"complexity_router_config": {
"tiers": {"SIMPLE": "t"},
"embedding_model": "emb",
"semantic_keyword_matching": semantic_on,
},
}
)
assert tuple(d.model_name for d in found) == expected