mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-07 08:26:10 +00:00
fix(complexity_router): fall back to a live peer when the decided tier model is fully cooled down (#39675)
A complexity tier can name several model groups, but the pool pick and the session-pin replay both returned a group without consulting deployment health, so a group whose every deployment was in cooldown was still routed to and the request died at the router's zero-deployment check while a healthy peer sat in the same tier. Gate the decided response at the pre-routing hook's exits, the seam the modality gate already occupies, so every arm that can place a request is covered by one owner: a fresh classification, a replayed or escalated pin, a plan-mode floor, a context-window escalation, an adaptive pick, and whatever arm is added next. Peers come from the decided tier only. Climbing to a higher tier costs more than the classifier asked for and is left to a follow-up. The gate fails open on every uncertainty: an unreadable cooldown view, a decision carrying no tier, a group the router knows no deployments for, or a tier whose peers are all cooling.
This commit is contained in:
tin-berri
GitHub
parent
df68edca76
commit
6dff3a5f72
4 changed files with 762 additions and 18 deletions
|
|
@ -35,7 +35,10 @@ from litellm.constants import (
|
|||
SESSION_ID_GENERATED_METADATA_KEY,
|
||||
)
|
||||
from litellm.integrations.custom_logger import CustomLogger
|
||||
from litellm.litellm_core_utils.core_helpers import get_metadata_variable_name_from_kwargs
|
||||
from litellm.litellm_core_utils.core_helpers import (
|
||||
_get_parent_otel_span_from_kwargs,
|
||||
get_metadata_variable_name_from_kwargs,
|
||||
)
|
||||
from litellm.litellm_core_utils.internal_call_metadata import forwarded_internal_call_metadata
|
||||
from litellm.litellm_core_utils.prompt_templates.common_utils import request_contains_image_content
|
||||
from litellm.litellm_core_utils.sensitive_data_masker import mask_credentials_in_payload
|
||||
|
|
@ -765,6 +768,11 @@ def _decision_is_pinnable(decision: StandardLoggingRoutingDecision | None) -> bo
|
|||
seconds against a TTL of an hour that every later turn refreshes. Its cause is whatever the
|
||||
fallback path reports, so the circuit signal is what marks the decision, and leaving it
|
||||
unpinned lets the session classify again as soon as the breaker closes.
|
||||
|
||||
A health failover describes the fleet's state right now, not the session's traffic, and it can
|
||||
displace decisions that were themselves unpinnable (a housekeeping call, a modality escalation).
|
||||
Pinning it would hold the session on the substitute long after the displaced group recovers; the
|
||||
gate re-fires per request, so leaving it unpinned costs nothing but the classifier call.
|
||||
"""
|
||||
return decision is None or (
|
||||
decision.get("cause")
|
||||
|
|
@ -774,6 +782,7 @@ def _decision_is_pinnable(decision: StandardLoggingRoutingDecision | None) -> bo
|
|||
"housekeeping",
|
||||
"modality_escalation",
|
||||
"modality_pin_override",
|
||||
"health_failover",
|
||||
)
|
||||
and not decision.get("context_escalated")
|
||||
and _CLASSIFIER_CIRCUIT_OPEN_SIGNAL not in (decision.get("signals") or ())
|
||||
|
|
@ -2650,6 +2659,150 @@ class ComplexityRouter(CustomLogger):
|
|||
and self._matched_plan_mode_signal(request_kwargs, resolved_messages) is None
|
||||
)
|
||||
|
||||
async def _model_group_can_serve(
|
||||
self,
|
||||
model_name: str,
|
||||
messages: list[dict[str, Any]] | None, # mutable-ok: forwarded verbatim to the router's own probe
|
||||
input: str | list | None, # mutable-ok: mirrors the owner's own input parameter, which this forwards verbatim
|
||||
request_kwargs: dict, # mutable-ok: same shape the hook receives
|
||||
) -> bool:
|
||||
"""Whether the router would find a deployment for this group ON THIS REQUEST.
|
||||
|
||||
Asks the same owner the routing path itself will ask, with the same prompt arguments it
|
||||
will pass, so every filter that decides a deployment's eligibility applies here exactly
|
||||
as it applies downstream: cooldowns, admin pause, team scoping, model access groups, tag
|
||||
routing, routing plugins, RPM limits, and the context-window pre-call check. Re-deriving
|
||||
any subset of that list is how a substitute gets chosen that the pipeline then rejects,
|
||||
and dropping `input` would silently skip the window check on the Responses API surface,
|
||||
where the prompt never arrives as messages.
|
||||
|
||||
Probed on a COPY of request_kwargs because the owner pops routing bookkeeping off the
|
||||
dict it is handed (`_target_order`, `_excluded_deployment_ids`), and this is a
|
||||
speculative question about a model that may never be picked.
|
||||
|
||||
Every way the owner says "nothing here can serve this" is a negative verdict: no healthy
|
||||
deployment for the group at all (BadRequestError, which ContextWindowExceededError
|
||||
subclasses), every deployment filtered out (RouterRateLimitError), and every deployment
|
||||
over its RPM (RouterRateLimitErrorBasic). Anything else is unknown rather than negative,
|
||||
so it reads as capacity: absent information must never decide the verdict.
|
||||
"""
|
||||
from litellm.exceptions import BadRequestError
|
||||
from litellm.types.router import RouterRateLimitError, RouterRateLimitErrorBasic
|
||||
|
||||
probe_kwargs: Final = dict(request_kwargs) # mutable-ok: the owner pops routing keys off the dict it is handed
|
||||
try:
|
||||
deployments: Final = await self.litellm_router_instance.async_get_healthy_deployments(
|
||||
model=model_name,
|
||||
request_kwargs=probe_kwargs,
|
||||
messages=messages,
|
||||
input=input,
|
||||
parent_otel_span=_get_parent_otel_span_from_kwargs(request_kwargs),
|
||||
)
|
||||
except (RouterRateLimitError, RouterRateLimitErrorBasic, BadRequestError):
|
||||
return False
|
||||
except Exception as exc: # noqa: BLE001 # a speculative eligibility read must fail open on unknown faults
|
||||
verbose_router_logger.debug(
|
||||
"ComplexityRouter: eligibility probe for %s failed, treating the group as live: %s", model_name, exc
|
||||
)
|
||||
return True
|
||||
return bool(deployments)
|
||||
|
||||
async def _gate_response_health(
|
||||
self,
|
||||
response: PreRoutingHookResponse,
|
||||
messages: list[dict[str, Any]] | None, # mutable-ok: forwarded verbatim to the list-typed re-pick
|
||||
input: str | list | None, # mutable-ok: mirrors the owner's own input parameter, which this forwards verbatim
|
||||
resolved_messages: Sequence[Mapping[str, object]] | None,
|
||||
request_kwargs: dict, # mutable-ok: same shape the hook receives
|
||||
) -> PreRoutingHookResponse:
|
||||
"""Replace a decided model group that has no serving capacity with a live peer in the same tier.
|
||||
|
||||
Applied to the decided response at the hook's exits, so every arm that can place a request
|
||||
is covered by one owner: a fresh classification, a replayed or escalated session pin, a
|
||||
plan-mode floor, a context-window escalation, an adaptive pick, and whatever arm is added
|
||||
next. Peers come from the DECIDED tier only; climbing to another tier is deliberately not
|
||||
done here, since a higher tier costs more than the classifier asked for.
|
||||
|
||||
Serving capacity is one question asked of one owner (`_model_group_can_serve`), so the
|
||||
substitute is only ever a group the pipeline would actually accept for this request. The
|
||||
pick then runs through `_pick_model_for_tier`, so routing plugins decide the substitute
|
||||
exactly as they decided the original.
|
||||
|
||||
Fails open everywhere it cannot be sure: an unreadable eligibility view, a decision
|
||||
carrying no tier (default_model), or a tier whose every peer is unusable too. It fails
|
||||
CLOSED on a plugin that empties the pool, leaving the original decision to fail rather
|
||||
than serving a model the plugin excluded.
|
||||
"""
|
||||
decision: Final = response.routing_decision
|
||||
decided_tier: Final = decision.get("tier") if decision is not None else None
|
||||
if decision is None or not isinstance(decided_tier, str):
|
||||
return response
|
||||
peers: Final = tuple(self._tier_pools().get(decided_tier, ()))
|
||||
if len(peers) < 2:
|
||||
return response
|
||||
if await self._model_group_can_serve(response.model, messages, input, request_kwargs):
|
||||
return response
|
||||
eligible: Final = (
|
||||
self._modality_eligible_models()
|
||||
if self.config.modality_routing and resolved_messages and request_contains_image_content(resolved_messages)
|
||||
else None
|
||||
)
|
||||
candidates: Final = tuple(
|
||||
peer for peer in peers if peer != response.model and (eligible is None or peer in eligible)
|
||||
)
|
||||
if not candidates:
|
||||
return response
|
||||
servable: Final = await asyncio.gather(
|
||||
*(self._model_group_can_serve(peer, messages, input, request_kwargs) for peer in candidates)
|
||||
)
|
||||
live: Final = tuple(peer for peer, can_serve in zip(candidates, servable) if can_serve)
|
||||
if not live:
|
||||
return response
|
||||
repick_messages: Final = (
|
||||
list(resolved_messages) if resolved_messages else None # mutable-ok: the pick's param is list-typed
|
||||
)
|
||||
try:
|
||||
new_model: Final = await self._pick_model_for_tier(
|
||||
decided_tier if self.config.has_custom_tiers else ComplexityTier(decided_tier),
|
||||
messages,
|
||||
repick_messages, # pyright: ignore[reportArgumentType] # hook-resolved message dicts; the pick only reads them
|
||||
request_kwargs,
|
||||
allowed_models=live,
|
||||
)
|
||||
except ValueError as exc:
|
||||
verbose_router_logger.debug(
|
||||
"ComplexityRouter: health failover found no candidate the routing plugins allow: %s", exc
|
||||
)
|
||||
return response
|
||||
self._restamp_adaptive_choice(request_kwargs, response.model, new_model)
|
||||
verbose_router_logger.info(
|
||||
"ComplexityRouter: routing decision cause=health_failover, routed_model=%s, displaced=%s",
|
||||
new_model,
|
||||
response.model,
|
||||
)
|
||||
new_decision: Final = self._build_routing_decision(
|
||||
routed_model=new_model,
|
||||
cause="health_failover",
|
||||
tier=decision.get("tier"),
|
||||
score=decision.get("score"),
|
||||
signals=(*(decision.get("signals") or ()), f"health_displaced:{response.model}"),
|
||||
matched_keyword=decision.get("matched_keyword"),
|
||||
escalation_keyword=decision.get("escalation_keyword"),
|
||||
escalated=bool(decision.get("escalated", False)),
|
||||
classifier_model=decision.get("classifier_model"),
|
||||
classifier_cost=decision.get("classifier_cost"),
|
||||
conversation_continuing=bool(decision.get("conversation_continuing", True)),
|
||||
tier_litellm_params=self._litellm_params_for_model(decided_tier, new_model),
|
||||
context_escalation_original_tier=decision.get("context_escalation_original_tier"),
|
||||
)
|
||||
return response.model_copy(
|
||||
update={ # mutable-ok: model_copy types update as a plain dict
|
||||
"model": new_model,
|
||||
"litellm_params": self._litellm_params_for_model(decided_tier, new_model),
|
||||
"routing_decision": new_decision,
|
||||
}
|
||||
)
|
||||
|
||||
def _placed_default_model(self) -> str:
|
||||
"""The default_model behind a usable-default verdict; the raise is the type-level
|
||||
proof, not a reachable path."""
|
||||
|
|
@ -3047,24 +3200,30 @@ class ComplexityRouter(CustomLogger):
|
|||
session_tier_litellm_params: Final = self._litellm_params_for_model(routed_pin_tier, routed_model)
|
||||
has_original_messages: Final = messages is not None and len(messages) > 0
|
||||
return self._with_session_deployment_affinity(
|
||||
await self._gate_response_modality(
|
||||
PreRoutingHookResponse(
|
||||
model=routed_model,
|
||||
messages=messages if has_original_messages else None,
|
||||
litellm_params=session_tier_litellm_params,
|
||||
routing_decision=self._build_routing_decision(
|
||||
routed_model=routed_model,
|
||||
cause=cause,
|
||||
tier=routed_pin_tier,
|
||||
matched_keyword=pin_plan_sentinel if plan_floored else None,
|
||||
escalation_keyword=pin_escalation_keyword,
|
||||
escalated=escalated,
|
||||
conversation_continuing=conversation_continuing,
|
||||
tier_litellm_params=session_tier_litellm_params,
|
||||
context_escalation_original_tier=pin_context_original_tier,
|
||||
await self._gate_response_health(
|
||||
await self._gate_response_modality(
|
||||
PreRoutingHookResponse(
|
||||
model=routed_model,
|
||||
messages=messages if has_original_messages else None,
|
||||
litellm_params=session_tier_litellm_params,
|
||||
routing_decision=self._build_routing_decision(
|
||||
routed_model=routed_model,
|
||||
cause=cause,
|
||||
tier=routed_pin_tier,
|
||||
matched_keyword=pin_plan_sentinel if plan_floored else None,
|
||||
escalation_keyword=pin_escalation_keyword,
|
||||
escalated=escalated,
|
||||
conversation_continuing=conversation_continuing,
|
||||
tier_litellm_params=session_tier_litellm_params,
|
||||
context_escalation_original_tier=pin_context_original_tier,
|
||||
),
|
||||
),
|
||||
messages,
|
||||
resolved_messages,
|
||||
request_kwargs,
|
||||
),
|
||||
messages,
|
||||
input,
|
||||
resolved_messages,
|
||||
request_kwargs,
|
||||
)
|
||||
|
|
@ -3080,7 +3239,13 @@ class ComplexityRouter(CustomLogger):
|
|||
resolved_messages=resolved_messages,
|
||||
)
|
||||
response: Final = (
|
||||
await self._gate_response_modality(routed_response, messages, resolved_messages, request_kwargs)
|
||||
await self._gate_response_health(
|
||||
await self._gate_response_modality(routed_response, messages, resolved_messages, request_kwargs),
|
||||
messages,
|
||||
input,
|
||||
resolved_messages,
|
||||
request_kwargs,
|
||||
)
|
||||
if routed_response is not None
|
||||
else None
|
||||
)
|
||||
|
|
|
|||
|
|
@ -2886,6 +2886,10 @@ RoutingDecisionCause = Literal[
|
|||
# carries an image the pinned model cannot accept. The stored pin is untouched, so the next
|
||||
# text turn replays it. Distinct from "modality_escalation", which never displaces a pin.
|
||||
"modality_pin_override",
|
||||
# Every deployment behind the decided model group was in cooldown, so a healthy peer in the
|
||||
# same tier served instead. The displaced group rides in signals. Reported even on a kept
|
||||
# session pin, since the pinned model did not serve the request.
|
||||
"health_failover",
|
||||
"session_affinity_pin",
|
||||
"session_affinity_escalation",
|
||||
# classification_mode 'user_turn': the request is an agent loop's continuation turn (no new
|
||||
|
|
|
|||
|
|
@ -11571,3 +11571,578 @@ class TestModalityRouting:
|
|||
model="m", request_kwargs={"metadata": {"session_id": "s1"}}, messages=self.IMAGE_MESSAGE
|
||||
)
|
||||
assert cache.async_set_cache.await_args.kwargs["value"] == {"model": "text-cheap", "tier": "SIMPLE"}
|
||||
|
||||
|
||||
class TestTierHealthFailover:
|
||||
"""A tier whose decided model group is entirely in cooldown falls back to a live peer."""
|
||||
|
||||
SIMPLE_MESSAGE = [{"role": "user", "content": "Hello!"}]
|
||||
TIERS = {"SIMPLE": ["dead-a", "live-b"], "MEDIUM": "mid", "COMPLEX": "big", "REASONING": "top"}
|
||||
|
||||
@staticmethod
|
||||
def _router(
|
||||
mock_router_instance,
|
||||
config,
|
||||
ids_by_model,
|
||||
cooling=(),
|
||||
blocked=(),
|
||||
excluded=(),
|
||||
raises_for=None,
|
||||
health_error=None,
|
||||
):
|
||||
"""ids_by_model: model group -> deployment ids the router knows.
|
||||
|
||||
The fake mirrors the real async_get_healthy_deployments contract, including how it says
|
||||
no: BadRequestError for a group with no deployment at all, RouterRateLimitError when every
|
||||
deployment is filtered out (cooling, admin-paused, or excluded by a request-scoped policy
|
||||
such as tags, team scoping or access groups), a per-model exception via raises_for (the
|
||||
RPM verdict), and an unrelated failure via health_error. It records what it was handed so
|
||||
tests can prove the probe passes a kwargs copy and forwards the prompt arguments.
|
||||
"""
|
||||
import litellm as litellm_module
|
||||
|
||||
from litellm.types.router import RouterRateLimitError
|
||||
|
||||
probed_kwargs = []
|
||||
probed_prompts = []
|
||||
|
||||
async def get_healthy_deployments(
|
||||
model, request_kwargs, messages=None, input=None, parent_otel_span=None, **kwargs
|
||||
):
|
||||
probed_kwargs.append(request_kwargs)
|
||||
probed_prompts.append((messages, input))
|
||||
if health_error is not None:
|
||||
raise health_error
|
||||
if raises_for and model in raises_for:
|
||||
raise raises_for[model]
|
||||
if not ids_by_model.get(model):
|
||||
raise litellm_module.BadRequestError(
|
||||
message=f"You passed in model={model}. There are no healthy deployments.",
|
||||
model=model,
|
||||
llm_provider="",
|
||||
)
|
||||
filtered = (*cooling, *blocked, *excluded)
|
||||
healthy = [
|
||||
{"model_name": model, "model_info": {"id": i}} for i in ids_by_model[model] if i not in filtered
|
||||
]
|
||||
if not healthy:
|
||||
raise RouterRateLimitError(
|
||||
model=model, cooldown_time=60.0, enable_pre_call_checks=False, cooldown_list=[]
|
||||
)
|
||||
return healthy
|
||||
|
||||
mock_router_instance.async_get_healthy_deployments = get_healthy_deployments
|
||||
mock_router_instance.probed_kwargs = probed_kwargs
|
||||
mock_router_instance.probed_prompts = probed_prompts
|
||||
mock_router_instance.cache = DualCache()
|
||||
return ComplexityRouter(
|
||||
model_name="health-test-router",
|
||||
litellm_router_instance=mock_router_instance,
|
||||
complexity_router_config=config,
|
||||
)
|
||||
|
||||
async def _pinned_hook(self, router, session_id="sess-1", messages=None):
|
||||
"""Drive the hook twice so the second call replays a pin, which makes the decided
|
||||
model deterministic instead of a coin flip over the tier pool."""
|
||||
kwargs = {"metadata": {"session_id": session_id}}
|
||||
await router.async_pre_routing_hook(model="m", request_kwargs=kwargs, messages=messages or self.SIMPLE_MESSAGE)
|
||||
return await router.async_pre_routing_hook(
|
||||
model="m", request_kwargs=kwargs, messages=messages or self.SIMPLE_MESSAGE
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dead_pinned_group_fails_over_to_live_peer_and_reports_the_displacement(self, mock_router_instance):
|
||||
"""The core regression: a session pinned to a group whose every deployment is cooling
|
||||
serves from the live peer, and the row says so rather than naming the pinned model."""
|
||||
router = self._router(
|
||||
mock_router_instance,
|
||||
{"tiers": dict(self.TIERS), "session_affinity": True},
|
||||
{"dead-a": ["id-a1", "id-a2"], "live-b": ["id-b1"]},
|
||||
cooling=("id-a1", "id-a2"),
|
||||
)
|
||||
# Seed the pin onto the dead group directly so the replay path is exercised.
|
||||
key = router._get_session_affinity_cache_key("sess-dead", {})
|
||||
await router.litellm_router_instance.cache.async_set_cache(
|
||||
key=key, value={"model": "dead-a", "tier": "SIMPLE"}, ttl=600
|
||||
)
|
||||
result = await router.async_pre_routing_hook(
|
||||
model="m", request_kwargs={"metadata": {"session_id": "sess-dead"}}, messages=self.SIMPLE_MESSAGE
|
||||
)
|
||||
assert result.model == "live-b"
|
||||
assert result.routing_decision["cause"] == "health_failover"
|
||||
assert "health_displaced:dead-a" in result.routing_decision["signals"]
|
||||
assert result.routing_decision["tier"] == "SIMPLE"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fresh_classification_never_serves_a_fully_cooled_group(self, mock_router_instance):
|
||||
"""The pool pick is a uniform draw, so the invariant is asserted over repeated turns:
|
||||
no turn may land on the dead group while a live peer sits in the same tier."""
|
||||
router = self._router(
|
||||
mock_router_instance,
|
||||
{"tiers": dict(self.TIERS)},
|
||||
{"dead-a": ["id-a1"], "live-b": ["id-b1"]},
|
||||
cooling=("id-a1",),
|
||||
)
|
||||
results = [
|
||||
await router.async_pre_routing_hook(model="m", request_kwargs={}, messages=self.SIMPLE_MESSAGE)
|
||||
for _ in range(20)
|
||||
]
|
||||
assert {r.model for r in results} == {"live-b"}
|
||||
assert all(r.routing_decision["cause"] in ("heuristic_scorer", "health_failover") for r in results)
|
||||
assert any(r.routing_decision["cause"] == "health_failover" for r in results)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"ids_by_model, cooling, health_error, tiers, reason",
|
||||
[
|
||||
({"dead-a": ["id-a1"], "live-b": ["id-b1"]}, (), None, None, "nothing_cooling"),
|
||||
({"dead-a": ["id-a1"], "live-b": ["id-b1"]}, ("id-a1", "id-b1"), None, None, "every_peer_dead"),
|
||||
(
|
||||
{"dead-a": ["id-a1"], "live-b": ["id-b1"]},
|
||||
("id-a1",),
|
||||
RuntimeError("redis down"),
|
||||
None,
|
||||
"health_view_unreadable",
|
||||
),
|
||||
(
|
||||
{"only": ["id-1"]},
|
||||
("id-1",),
|
||||
None,
|
||||
{"SIMPLE": "only", "MEDIUM": "mid", "COMPLEX": "big", "REASONING": "top"},
|
||||
"single_model_tier_has_no_peer",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_gate_fails_open_and_leaves_the_decision_untouched(
|
||||
self, mock_router_instance, ids_by_model, cooling, health_error, tiers, reason
|
||||
):
|
||||
"""Every uncertainty leaves the decided model in place, so the request fails exactly
|
||||
as it does today rather than being rerouted on a guess."""
|
||||
router = self._router(
|
||||
mock_router_instance,
|
||||
{"tiers": dict(tiers or self.TIERS), "session_affinity": True},
|
||||
ids_by_model,
|
||||
cooling=cooling,
|
||||
health_error=health_error,
|
||||
)
|
||||
pinned = "only" if tiers else "dead-a"
|
||||
key = router._get_session_affinity_cache_key("sess-open", {})
|
||||
await router.litellm_router_instance.cache.async_set_cache(
|
||||
key=key, value={"model": pinned, "tier": "SIMPLE"}, ttl=600
|
||||
)
|
||||
result = await router.async_pre_routing_hook(
|
||||
model="m", request_kwargs={"metadata": {"session_id": "sess-open"}}, messages=self.SIMPLE_MESSAGE
|
||||
)
|
||||
assert result.model == pinned, reason
|
||||
assert result.routing_decision["cause"] == "session_affinity_pin", reason
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_failed_over_turn_is_never_pinned(self, mock_router_instance):
|
||||
"""A failover describes the fleet's state, not the session's traffic, so it must not
|
||||
become the pin: the substitute would outlive the outage that caused it.
|
||||
|
||||
Asserted over many sessions because the underlying pool pick is a uniform draw.
|
||||
"""
|
||||
router = self._router(
|
||||
mock_router_instance,
|
||||
{"tiers": dict(self.TIERS), "session_affinity": True},
|
||||
{"dead-a": ["id-a1"], "live-b": ["id-b1"]},
|
||||
cooling=("id-a1",),
|
||||
)
|
||||
|
||||
async def pin_after_session(turn: int):
|
||||
session_id = f"sess-write-{turn}"
|
||||
await router.async_pre_routing_hook(
|
||||
model="m",
|
||||
request_kwargs={"metadata": {"session_id": session_id}},
|
||||
messages=self.SIMPLE_MESSAGE,
|
||||
)
|
||||
return await router.litellm_router_instance.cache.async_get_cache(
|
||||
key=router._get_session_affinity_cache_key(session_id, {})
|
||||
)
|
||||
|
||||
stored = [await pin_after_session(turn) for turn in range(20)]
|
||||
assert all(entry in (None, {"model": "live-b", "tier": "SIMPLE"}) for entry in stored)
|
||||
assert any(entry is None for entry in stored), "a failed-over turn must leave the pin unwritten"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_unpinnable_displaced_cause_stays_unpinnable_after_failover(self, mock_router_instance):
|
||||
"""A housekeeping turn is deliberately never pinned. Rewriting its cause to health_failover
|
||||
must not smuggle it past that guard and lock the session onto the cheapest tier."""
|
||||
router = self._router(
|
||||
mock_router_instance,
|
||||
{"tiers": dict(self.TIERS), "session_affinity": True},
|
||||
{"dead-a": ["id-a1"], "live-b": ["id-b1"]},
|
||||
cooling=("id-a1",),
|
||||
)
|
||||
session_id = "sess-housekeeping"
|
||||
result = await router.async_pre_routing_hook(
|
||||
model="m",
|
||||
request_kwargs={"metadata": {"session_id": session_id}},
|
||||
messages=[{"role": "user", "content": TITLE_ASK}],
|
||||
)
|
||||
assert result.routing_decision["cause"] in ("housekeeping", "health_failover")
|
||||
stored = await router.litellm_router_instance.cache.async_get_cache(
|
||||
key=router._get_session_affinity_cache_key(session_id, {})
|
||||
)
|
||||
assert stored is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_peer_whose_deployments_are_admin_paused_is_not_a_failover_target(self, mock_router_instance):
|
||||
"""Capacity is the router's own verdict, not just cooldown: a paused peer would be
|
||||
rejected downstream and the request would fail with a live third peer available."""
|
||||
router = self._router(
|
||||
mock_router_instance,
|
||||
{
|
||||
"tiers": {
|
||||
"SIMPLE": ["dead-a", "paused-b", "live-c"],
|
||||
"MEDIUM": "mid",
|
||||
"COMPLEX": "big",
|
||||
"REASONING": "top",
|
||||
},
|
||||
"session_affinity": True,
|
||||
},
|
||||
{"dead-a": ["id-a1"], "paused-b": ["id-b1"], "live-c": ["id-c1"]},
|
||||
cooling=("id-a1",),
|
||||
blocked=("id-b1",),
|
||||
)
|
||||
key = router._get_session_affinity_cache_key("sess-paused", {})
|
||||
await router.litellm_router_instance.cache.async_set_cache(
|
||||
key=key, value={"model": "dead-a", "tier": "SIMPLE"}, ttl=600
|
||||
)
|
||||
results = [
|
||||
await router.async_pre_routing_hook(
|
||||
model="m", request_kwargs={"metadata": {"session_id": "sess-paused"}}, messages=self.SIMPLE_MESSAGE
|
||||
)
|
||||
for _ in range(20)
|
||||
]
|
||||
assert {r.model for r in results} == {"live-c"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_failover_fails_closed_when_a_routing_plugin_excludes_every_peer(self, mock_router_instance):
|
||||
"""A plugin's exclusion is policy, so a peer it removed must not be served just because
|
||||
the plugin's own choice went into cooldown."""
|
||||
|
||||
class ExcludeEverythingButDead:
|
||||
async def run(self, context):
|
||||
context.candidate_models = [m for m in context.candidate_models if m == "dead-a"]
|
||||
return context
|
||||
|
||||
router = self._router(
|
||||
mock_router_instance,
|
||||
{"tiers": dict(self.TIERS), "plugins": [ExcludeEverythingButDead()]},
|
||||
{"dead-a": ["id-a1"], "live-b": ["id-b1"]},
|
||||
cooling=("id-a1",),
|
||||
)
|
||||
result = await router.async_pre_routing_hook(model="m", request_kwargs={}, messages=self.SIMPLE_MESSAGE)
|
||||
assert result.model == "dead-a"
|
||||
assert result.routing_decision["cause"] != "health_failover"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_failover_moves_the_adaptive_chosen_model_marker(self, mock_router_instance):
|
||||
"""The adaptive feedback loop scores the marker, so leaving it on the displaced group
|
||||
would credit a model that never ran."""
|
||||
router = self._router(
|
||||
mock_router_instance,
|
||||
{"tiers": dict(self.TIERS), "session_affinity": True},
|
||||
{"dead-a": ["id-a1"], "live-b": ["id-b1"]},
|
||||
cooling=("id-a1",),
|
||||
)
|
||||
key = router._get_session_affinity_cache_key("sess-adaptive", {})
|
||||
await router.litellm_router_instance.cache.async_set_cache(
|
||||
key=key, value={"model": "dead-a", "tier": "SIMPLE"}, ttl=600
|
||||
)
|
||||
request_kwargs = {"metadata": {"session_id": "sess-adaptive", "adaptive_router_chosen_model": "dead-a"}}
|
||||
result = await router.async_pre_routing_hook(
|
||||
model="m", request_kwargs=request_kwargs, messages=self.SIMPLE_MESSAGE
|
||||
)
|
||||
assert result.model == "live-b"
|
||||
assert request_kwargs["metadata"]["adaptive_router_chosen_model"] == "live-b"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_health_failover_never_undoes_the_modality_gate(self, mock_router_instance):
|
||||
"""An image turn whose only live peer cannot take images keeps the vision model the
|
||||
modality gate chose: serving a cooling vision model beats a hard 400."""
|
||||
vision_by_model = {"dead-vision": True, "live-text": False}
|
||||
|
||||
def get_model_list(model_name=None):
|
||||
if model_name not in vision_by_model:
|
||||
return []
|
||||
return [
|
||||
{
|
||||
"model_name": model_name,
|
||||
"litellm_params": {"model": f"openai/unmapped-{model_name}"},
|
||||
"model_info": {"supports_vision": vision_by_model[model_name]},
|
||||
}
|
||||
]
|
||||
|
||||
mock_router_instance.get_model_list = get_model_list
|
||||
router = self._router(
|
||||
mock_router_instance,
|
||||
{
|
||||
"tiers": {
|
||||
"SIMPLE": ["dead-vision", "live-text"],
|
||||
"MEDIUM": "mid",
|
||||
"COMPLEX": "big",
|
||||
"REASONING": "top",
|
||||
},
|
||||
"session_affinity": True,
|
||||
"modality_routing": True,
|
||||
},
|
||||
{"dead-vision": ["id-v1"], "live-text": ["id-t1"]},
|
||||
cooling=("id-v1",),
|
||||
)
|
||||
key = router._get_session_affinity_cache_key("sess-image", {})
|
||||
await router.litellm_router_instance.cache.async_set_cache(
|
||||
key=key, value={"model": "dead-vision", "tier": "SIMPLE"}, ttl=600
|
||||
)
|
||||
image_message = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": "What color is this?"},
|
||||
{"type": "image_url", "image_url": {"url": "data:image/png;base64,aGk="}},
|
||||
],
|
||||
}
|
||||
]
|
||||
result = await router.async_pre_routing_hook(
|
||||
model="m", request_kwargs={"metadata": {"session_id": "sess-image"}}, messages=image_message
|
||||
)
|
||||
assert result.model == "dead-vision"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_failover_will_not_pick_a_peer_that_cannot_hold_the_prompt(self):
|
||||
"""The context-window filter is a pre-call check inside the eligibility owner, so this
|
||||
drives the REAL owner on a real Router and injects only the cooldown. A substitute the
|
||||
prompt overflows must never be chosen while a peer that holds it exists."""
|
||||
pool = ["dead-big", "live-small", "live-big"]
|
||||
router_instance = _windowed_router(
|
||||
("dead-big", "openai/gpt-4o-mini", 200000),
|
||||
("live-small", "openai/gpt-3.5-turbo", 16385),
|
||||
("live-big", "openai/gpt-4o-mini", 200000),
|
||||
)
|
||||
router_instance.enable_pre_call_checks = True
|
||||
dead_ids = {d["model_info"]["id"] for d in router_instance.model_list if d["model_name"] == "dead-big"}
|
||||
|
||||
async def active_cooldowns(model_ids, parent_otel_span):
|
||||
return [(i, {"exception_received": "boom"}) for i in model_ids if i in dead_ids]
|
||||
|
||||
router_instance.cooldown_cache.async_get_active_cooldowns = active_cooldowns
|
||||
router_instance.cache = DualCache()
|
||||
router = ComplexityRouter(
|
||||
model_name="health-window-router",
|
||||
litellm_router_instance=router_instance,
|
||||
complexity_router_config={
|
||||
"tiers": {name: list(pool) for name in ("SIMPLE", "MEDIUM", "COMPLEX", "REASONING")},
|
||||
"session_affinity": True,
|
||||
"enable_context_window_escalation": True,
|
||||
},
|
||||
)
|
||||
key = router._get_session_affinity_cache_key("sess-window", {})
|
||||
await router.litellm_router_instance.cache.async_set_cache(
|
||||
key=key, value={"model": "dead-big", "tier": "SIMPLE"}, ttl=600
|
||||
)
|
||||
results = [
|
||||
await router.async_pre_routing_hook(
|
||||
model="m",
|
||||
request_kwargs={"metadata": {"session_id": "sess-window"}},
|
||||
messages=list(_OVERSIZED_TURNS),
|
||||
)
|
||||
for _ in range(20)
|
||||
]
|
||||
assert "live-small" not in {r.model for r in results}
|
||||
assert {r.model for r in results} == {"live-big"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_decision_with_no_tier_is_left_alone(self, mock_router_instance):
|
||||
"""default_model placements carry no tier, so there is no pool to draw a peer from.
|
||||
The gate leaves them exactly as they are rather than inventing a tier."""
|
||||
router = self._router(
|
||||
mock_router_instance,
|
||||
{
|
||||
"tiers": dict(self.TIERS),
|
||||
"default_model": "fallback-model",
|
||||
"classifier_type": "llm",
|
||||
"classifier_llm_config": {"model": "gpt-4o-mini"},
|
||||
"classifier_fallback": "default_model",
|
||||
},
|
||||
{"fallback-model": ["id-f1"], "dead-a": ["id-a1"], "live-b": ["id-b1"]},
|
||||
cooling=("id-f1", "id-a1"),
|
||||
)
|
||||
mock_router_instance.acompletion = AsyncMock(side_effect=RuntimeError("classifier down"))
|
||||
result = await router.async_pre_routing_hook(model="m", request_kwargs={}, messages=self.SIMPLE_MESSAGE)
|
||||
assert result.model == "fallback-model"
|
||||
assert result.routing_decision.get("tier") is None
|
||||
assert result.routing_decision["cause"] != "health_failover"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_tier_entry_the_router_cannot_serve_fails_over_instead_of_erroring(self, mock_router_instance):
|
||||
"""A tier naming a model this proxy has no deployment for is unservable, and the
|
||||
eligibility owner says so, so the peer serves rather than the request 429ing."""
|
||||
router = self._router(
|
||||
mock_router_instance,
|
||||
{"tiers": dict(self.TIERS), "session_affinity": True},
|
||||
{"live-b": ["id-b1"]},
|
||||
)
|
||||
key = router._get_session_affinity_cache_key("sess-unknown", {})
|
||||
await router.litellm_router_instance.cache.async_set_cache(
|
||||
key=key, value={"model": "dead-a", "tier": "SIMPLE"}, ttl=600
|
||||
)
|
||||
result = await router.async_pre_routing_hook(
|
||||
model="m", request_kwargs={"metadata": {"session_id": "sess-unknown"}}, messages=self.SIMPLE_MESSAGE
|
||||
)
|
||||
assert result.model == "live-b"
|
||||
assert result.routing_decision["cause"] == "health_failover"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_peer_excluded_by_a_request_scoped_policy_is_not_a_failover_target(self, mock_router_instance):
|
||||
"""Tag, team and access-group filters are request-scoped and live inside the eligibility
|
||||
owner. A peer they exclude would be rejected downstream, so it must not be chosen."""
|
||||
router = self._router(
|
||||
mock_router_instance,
|
||||
{
|
||||
"tiers": {
|
||||
"SIMPLE": ["dead-a", "tagged-out-b", "live-c"],
|
||||
"MEDIUM": "mid",
|
||||
"COMPLEX": "big",
|
||||
"REASONING": "top",
|
||||
},
|
||||
"session_affinity": True,
|
||||
},
|
||||
{"dead-a": ["id-a1"], "tagged-out-b": ["id-b1"], "live-c": ["id-c1"]},
|
||||
cooling=("id-a1",),
|
||||
excluded=("id-b1",),
|
||||
)
|
||||
key = router._get_session_affinity_cache_key("sess-tagged", {})
|
||||
await router.litellm_router_instance.cache.async_set_cache(
|
||||
key=key, value={"model": "dead-a", "tier": "SIMPLE"}, ttl=600
|
||||
)
|
||||
results = [
|
||||
await router.async_pre_routing_hook(
|
||||
model="m", request_kwargs={"metadata": {"session_id": "sess-tagged"}}, messages=self.SIMPLE_MESSAGE
|
||||
)
|
||||
for _ in range(20)
|
||||
]
|
||||
assert {r.model for r in results} == {"live-c"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_eligibility_probe_never_mutates_the_caller_request_kwargs(self, mock_router_instance):
|
||||
"""The owner pops routing bookkeeping off the dict it is handed, so a probe that passed
|
||||
the real kwargs would strip them before the request is ever placed."""
|
||||
router = self._router(
|
||||
mock_router_instance,
|
||||
{"tiers": dict(self.TIERS), "session_affinity": True},
|
||||
{"dead-a": ["id-a1"], "live-b": ["id-b1"]},
|
||||
cooling=("id-a1",),
|
||||
)
|
||||
key = router._get_session_affinity_cache_key("sess-kwargs", {})
|
||||
await router.litellm_router_instance.cache.async_set_cache(
|
||||
key=key, value={"model": "dead-a", "tier": "SIMPLE"}, ttl=600
|
||||
)
|
||||
request_kwargs = {
|
||||
"metadata": {"session_id": "sess-kwargs"},
|
||||
"_target_order": 1,
|
||||
"_excluded_deployment_ids": ["id-x"],
|
||||
}
|
||||
result = await router.async_pre_routing_hook(
|
||||
model="m", request_kwargs=request_kwargs, messages=self.SIMPLE_MESSAGE
|
||||
)
|
||||
assert result.model == "live-b"
|
||||
assert request_kwargs["_target_order"] == 1
|
||||
assert request_kwargs["_excluded_deployment_ids"] == ["id-x"]
|
||||
assert all(probed is not request_kwargs for probed in router.litellm_router_instance.probed_kwargs)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_peer_whose_every_deployment_is_over_its_rpm_is_not_a_failover_target(
|
||||
self, mock_router_instance
|
||||
):
|
||||
"""RPM exhaustion is its own verdict from the owner (RouterRateLimitErrorBasic). A peer
|
||||
in that state would be rejected downstream, so it cannot be the substitute."""
|
||||
from litellm.types.router import RouterRateLimitErrorBasic
|
||||
|
||||
router = self._router(
|
||||
mock_router_instance,
|
||||
{
|
||||
"tiers": {
|
||||
"SIMPLE": ["dead-a", "rpm-full-b", "live-c"],
|
||||
"MEDIUM": "mid",
|
||||
"COMPLEX": "big",
|
||||
"REASONING": "top",
|
||||
},
|
||||
"session_affinity": True,
|
||||
},
|
||||
{"dead-a": ["id-a1"], "rpm-full-b": ["id-b1"], "live-c": ["id-c1"]},
|
||||
cooling=("id-a1",),
|
||||
raises_for={"rpm-full-b": RouterRateLimitErrorBasic(model="rpm-full-b")},
|
||||
)
|
||||
key = router._get_session_affinity_cache_key("sess-rpm", {})
|
||||
await router.litellm_router_instance.cache.async_set_cache(
|
||||
key=key, value={"model": "dead-a", "tier": "SIMPLE"}, ttl=600
|
||||
)
|
||||
results = [
|
||||
await router.async_pre_routing_hook(
|
||||
model="m", request_kwargs={"metadata": {"session_id": "sess-rpm"}}, messages=self.SIMPLE_MESSAGE
|
||||
)
|
||||
for _ in range(20)
|
||||
]
|
||||
assert {r.model for r in results} == {"live-c"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_probe_forwards_input_so_window_checks_run_on_input_only_surfaces(
|
||||
self, mock_router_instance
|
||||
):
|
||||
"""The Responses API carries its prompt as `input`, never as messages. The owner only
|
||||
runs its context-window pre-call check when one of them is present, so dropping `input`
|
||||
would silently skip window filtering on that whole surface."""
|
||||
router = self._router(
|
||||
mock_router_instance,
|
||||
{"tiers": dict(self.TIERS), "session_affinity": True},
|
||||
{"dead-a": ["id-a1"], "live-b": ["id-b1"]},
|
||||
cooling=("id-a1",),
|
||||
)
|
||||
key = router._get_session_affinity_cache_key("sess-input", {})
|
||||
await router.litellm_router_instance.cache.async_set_cache(
|
||||
key=key, value={"model": "dead-a", "tier": "SIMPLE"}, ttl=600
|
||||
)
|
||||
result = await router.async_pre_routing_hook(
|
||||
model="m",
|
||||
request_kwargs={"metadata": {"session_id": "sess-input"}},
|
||||
input="summarize this document for me",
|
||||
)
|
||||
assert result.model == "live-b"
|
||||
assert any(
|
||||
probed_input == "summarize this document for me"
|
||||
for _, probed_input in router.litellm_router_instance.probed_prompts
|
||||
), "the eligibility probe must forward `input` to the owner"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_group_the_router_has_no_deployment_for_is_not_a_failover_target(
|
||||
self, mock_router_instance
|
||||
):
|
||||
"""The owner answers an unconfigured group with BadRequestError. Reading that as live
|
||||
would both skip failover off it and let it be chosen as a substitute."""
|
||||
router = self._router(
|
||||
mock_router_instance,
|
||||
{
|
||||
"tiers": {
|
||||
"SIMPLE": ["dead-a", "unconfigured-b", "live-c"],
|
||||
"MEDIUM": "mid",
|
||||
"COMPLEX": "big",
|
||||
"REASONING": "top",
|
||||
},
|
||||
"session_affinity": True,
|
||||
},
|
||||
{"dead-a": ["id-a1"], "live-c": ["id-c1"]},
|
||||
cooling=("id-a1",),
|
||||
)
|
||||
key = router._get_session_affinity_cache_key("sess-missing", {})
|
||||
await router.litellm_router_instance.cache.async_set_cache(
|
||||
key=key, value={"model": "dead-a", "tier": "SIMPLE"}, ttl=600
|
||||
)
|
||||
results = [
|
||||
await router.async_pre_routing_hook(
|
||||
model="m", request_kwargs={"metadata": {"session_id": "sess-missing"}}, messages=self.SIMPLE_MESSAGE
|
||||
)
|
||||
for _ in range(20)
|
||||
]
|
||||
assert {r.model for r in results} == {"live-c"}
|
||||
|
|
|
|||
2
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
2
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
|
|
@ -36045,7 +36045,7 @@ 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" | "session_affinity_pin" | "session_affinity_escalation" | "user_turn_continuation" | "default_fallback" | "keyword" | "quality_tier" | "bandit";
|
||||
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";
|
||||
/** Classifier Cost */
|
||||
classifier_cost?: number;
|
||||
/** Classifier Model */
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue