scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-07 08:26:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

feat(router): fall back on anthropic safeguard refusals on /v1/messages (#39157)

Browse source 
* fix(router): resolve fallbacks against the tier a pre-routing hook selected

A complexity or auto router picks a tier behind the router group name, but
fallback lookup kept using kwargs["model"], which is still the router name. The
tier's configured chain never ran, so a provider failure on its first hop went
straight back to the client with "No fallback model group found for original
model_group=smart-router".

The hook assigns the selected model to a local only, and fallback resolution runs
on an outer kwargs dict that **kwargs already copied, so writing it there is not
visible. Record the selection in the metadata bucket instead, which is a nested
dict shared by reference across those copies and is how the router already
carries values back up, then key fallback lookup off it when present.

Applies to the generic, context-window, content-policy and weighted-failover
lookups. Reporting keeps using the router name, since that is what the caller
asked for.

Fixes #38832

* fix(router): annotate the recorded-selection helper with a read-only mapping

record_pre_routing_selection only reads the request kwargs, writing into the
nested metadata bucket it finds there, so Mapping states what it actually needs
and clears the LIT001 mutable-annotation budget without a suppression.

* test(router): assert the no-kwargs path leaks nothing

The tolerated-None case called the helper without checking anything, which the
test-quality gate counts as a test with no assertion. Assert that a fresh mapping
still reads back empty, so the case proves the call is a no-op rather than only
that it does not raise.

* fix(router): stop declaring loop-assigned locals Final in the selection helpers

Both helpers annotated a loop-assigned local as Final, which reassigns a Final on
every iteration and cost three basedpyright errors. Read the buckets through a
generator instead, so the write path iterates a for-target and the read path
resolves in one shot with next(), which also matches the functional style the
type-discipline rules ask for.

* style(router): apply ruff format to the selection helpers

* fix(router): derive the pre-routing tier fresh on every fallback hop

The metadata buckets also carry whatever the caller sent, so an inbound
pre_routing_selected_model let a client pick which fallback chain its
request fell into. A fallback hop also inherited the previous hop's tier,
so the second hop keyed its own failure off the tier that already failed
and never ran its own chain.

Clear the key at the top of async_function_with_fallbacks. Every hop
re-enters there, so only the hook that routed that hop can set it.

* fix(router): drop the cast at the fallback-hop clear call site

* feat(router): fall back on anthropic safeguard refusals on /v1/messages

---------

Co-authored-by: Priyansh Nandwana <nandwana.priyansh103@gmail.com>
This commit is contained in:
tin-berri 2026-09-01 16:50:12 -07:00 • committed by GitHub
parent 8f56dbe7a3
commit 59da6e75a5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 970 additions and 50 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py
View file

@ -86,22 +86,41 @@ def _decoded_sse_data_line(line: bytes) -> object | None:
return None
def _anthropic_error_event_payload(chunk: object) -> Mapping[str, object] | None:
def _anthropic_event_payload(chunk: object, event_type: str) -> Mapping[str, object] | None:
if isinstance(chunk, dict):
return chunk if chunk.get("type") == "error" else None
return chunk if chunk.get("type") == event_type else None
if isinstance(chunk, (bytes, bytearray)):
decoded_lines: Final = (_decoded_sse_data_line(line) for line in chunk.splitlines())
return next(
(
candidate
for candidate in decoded_lines
if isinstance(candidate, dict) and candidate.get("type") == "error"
if isinstance(candidate, dict) and candidate.get("type") == event_type
),
None,
)
return None
def _anthropic_error_event_payload(chunk: object) -> Mapping[str, object] | None:
return _anthropic_event_payload(chunk, "error")
def parse_anthropic_refusal_stop_details(chunk: object) -> Mapping[str, object] | None:
"""
Return the ``stop_details`` object of an Anthropic SSE ``message_delta``
chunk whose delta carries ``stop_reason: "refusal"`` (a safeguard refusal:
https://platform.claude.com/docs/en/build-with-claude/refusals-and-fallback),
or None for any other chunk, a plain refusal without ``stop_details`` included.
"""
payload: Final = _anthropic_event_payload(chunk, "message_delta")
delta: Final = payload.get("delta") if payload is not None else None
if not isinstance(delta, dict) or delta.get("stop_reason") != "refusal":
return None
stop_details: Final = delta.get("stop_details")
return stop_details if isinstance(stop_details, dict) else None
def _anthropic_error_body(chunk: object) -> Mapping[str, object] | None:
"""Return the ``error`` object of an Anthropic SSE ``event: error`` chunk, or None."""
payload: Final = _anthropic_error_event_payload(chunk)

49
litellm/llms/anthropic/experimental_pass_through/messages/utils.py
View file

@ -1,11 +1,40 @@
from collections.abc import Mapping
from functools import lru_cache
from typing import Any, Final, cast, get_type_hints
from typing import TYPE_CHECKING, Any, Final, cast, get_type_hints
from litellm.types.llms.anthropic import AnthropicMessagesRequestOptionalParams
from litellm.types.llms.anthropic_messages.anthropic_response import (
AnthropicMessagesResponse,
)
if TYPE_CHECKING:
from litellm.exceptions import ContentPolicyViolationError
def get_safeguard_refusal_stop_details(response: object) -> Mapping[str, Any] | None:
"""
Return the ``stop_details`` of an Anthropic Messages response refused by a
safeguard (``stop_reason: "refusal"`` carrying ``stop_details``:
https://platform.claude.com/docs/en/build-with-claude/refusals-and-fallback),
or None for any other response, a plain refusal without ``stop_details`` included.
"""
if not isinstance(response, dict) or response.get("stop_reason") != "refusal":
return None
stop_details: Final = response.get("stop_details")
return stop_details if isinstance(stop_details, dict) else None
def safeguard_refusal_error(model: str, stop_details: Mapping[str, object]) -> "ContentPolicyViolationError":
"""The exception a safeguard-refused Anthropic response converts into so the
content-policy fallback chain can re-dispatch it."""
from litellm.exceptions import ContentPolicyViolationError
return ContentPolicyViolationError(
message=f"Anthropic safeguard refusal (category: {stop_details.get('category')}).",
model=model,
llm_provider="anthropic",
)
@lru_cache(maxsize=1)
def _anthropic_messages_optional_param_keys() -> frozenset[str]:
@ -100,14 +129,12 @@ def mock_response(
model=model,
)
return AnthropicMessagesResponse(
**{
"content": [{"text": mock_response, "type": "text"}],
"id": "msg_013Zva2CMHLNnXjNJJKqJ2EF",
"model": "claude-sonnet-4-20250514",
"role": "assistant",
"stop_reason": "end_turn",
"stop_sequence": None,
"type": "message",
"usage": {"input_tokens": 2095, "output_tokens": 503},
}
content=[{"text": mock_response, "type": "text"}],
id="msg_013Zva2CMHLNnXjNJJKqJ2EF",
model="claude-sonnet-4-20250514",
role="assistant",
stop_reason="end_turn",
stop_sequence=None,
type="message",
usage={"input_tokens": 2095, "output_tokens": 503},
)

192
litellm/router.py
View file

@ -143,7 +143,11 @@ from litellm.router_utils.cooldown_handlers import (
from litellm.router_utils.fallback_event_handlers import (
AttemptedFallbackTargets,
_check_non_standard_fallback_format,
get_fallback_model_group,
clear_pre_routing_selection,
fallback_lookup_groups,
get_fallback_model_group_for_lookup_groups,
get_pre_routing_selection,
record_pre_routing_selection,
run_async_fallback,
)
from litellm.router_utils.get_retry_from_policy import (
@ -4918,6 +4922,19 @@ class Router:
)
response = await response
if self._should_raise_anthropic_refusal_error(
model=model,
original_generic_function=original_generic_function,
response=response,
kwargs=kwargs,
):
from litellm.llms.anthropic.experimental_pass_through.messages.utils import (
safeguard_refusal_error,
)
refusal_details: Final = cast(dict, response["stop_details"]) # cast-ok: gate verified the shape
raise safeguard_refusal_error(model=model, stop_details=refusal_details)
self.success_calls[model_name] += 1
verbose_router_logger.info("ageneric_api_call_with_fallbacks(model=%s)\x1b[32m 200 OK\x1b[0m", model_name)
@ -4964,6 +4981,11 @@ class Router:
# fallback to the original reference for any non-picklable value.
# The original_generic_function is preserved so the per-attempt
# helper knows which underlying API to call on fallback.
# The pre-routing hook stamps its tier selection into this bucket during the primary
# attempt; seeding it before the snapshot gives both the live kwargs and the copy a
# bucket, so the post-call carry-over below always has somewhere to read and write.
kwargs.setdefault("litellm_metadata", {}) # mutable-ok: shared bucket # rebind-ok: stamp must be readable here
fallback_kwargs: Final[dict[str, object]] = kwargs.copy()
if isinstance(fallback_kwargs.get("litellm_metadata"), dict):
fallback_kwargs["litellm_metadata"] = safe_deep_copy(fallback_kwargs["litellm_metadata"])
@ -4973,6 +4995,14 @@ class Router:
response: Final = await self._ageneric_api_call_with_fallbacks(original_function=original_function, **kwargs)
# The snapshot predates the pre-routing hook, so the tier it stamped into the live kwargs
# is carried over write-or-clear: a stale or caller-supplied selection left in the copy
# would key the mid-stream fallback lookup off a tier this attempt never routed to.
clear_pre_routing_selection(fallback_kwargs)
live_pre_routing_selection: Final = get_pre_routing_selection(kwargs)
if live_pre_routing_selection is not None:
record_pre_routing_selection(fallback_kwargs, live_pre_routing_selection)
if kwargs.get("stream") and isinstance(response, BaseResponsesAPIStreamingIterator):
return await self._aresponses_streaming_iterator(
response=response,
@ -5030,6 +5060,10 @@ class Router:
from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import (
aclose_if_supported,
parse_anthropic_error_event,
parse_anthropic_refusal_stop_details,
)
from litellm.llms.anthropic.experimental_pass_through.messages.utils import (
safeguard_refusal_error,
)
source_iterator: Final = response
@ -5068,13 +5102,35 @@ class Router:
continue
if _anthropic_stream_commits_now(chunk, has_generated_content, len(buffered_lifecycle_chunks)):
has_generated_content = True # rebind-ok: real content seen, or the buffer cap was hit
error_event = parse_anthropic_error_event(chunk)
# A transport can split one SSE data line across byte chunks, so pre-content
# detection parses the accumulated buffer plus the current chunk, never the
# chunk alone; the buffer is already capped, which bounds this window too.
parse_window = ( # rebind-ok: freshly computed each iteration, never carried over
b"".join(c for c in (*buffered_lifecycle_chunks, chunk) if isinstance(c, (bytes, bytearray))) # pyright: ignore[reportUnnecessaryIsInstance] # bridge-path chunks are not always bytes at runtime
if not has_generated_content and isinstance(chunk, (bytes, bytearray)) # pyright: ignore[reportUnnecessaryIsInstance] # bridge-path chunks are not always bytes at runtime
else chunk
)
error_event = parse_anthropic_error_event(parse_window)
retriable_pending_error = ( # rebind-ok: freshly computed each iteration, never carried over
not has_generated_content
and error_event is not None
and _is_retriable_anthropic_status(error_event[2])
and not _anthropic_stream_error_is_gateway_verdict(chunk)
)
refusal_stop_details = ( # rebind-ok: freshly computed each iteration, never carried over
parse_anthropic_refusal_stop_details(parse_window)
if not has_generated_content and error_event is None
else None
)
if refusal_stop_details is not None and self._has_content_policy_fallback(model, initial_kwargs):
refusal_error = safeguard_refusal_error(model=model, stop_details=refusal_stop_details)
raise MidStreamFallbackError(
message=refusal_error.message,
model=model,
llm_provider="anthropic",
original_exception=refusal_error,
is_pre_first_chunk=True,
)
if not has_generated_content and not retriable_pending_error and error_event is None:
buffered_lifecycle_chunks = (*buffered_lifecycle_chunks, chunk)
continue
@ -5186,8 +5242,13 @@ class Router:
kwargs=initial_kwargs,
metadata_variable_name="litellm_metadata",
)
# The content-policy dispatch branch matches on the trigger's own type, so a refusal's
# MidStreamFallbackError envelope is unwrapped here or the wrong fallback list is consulted.
fallback_trigger: Final[Exception] = (
e.original_exception if isinstance(e.original_exception, litellm.ContentPolicyViolationError) else e
)
fallback_response = await self.async_function_with_fallbacks_common_utils( # rebind-ok: set on success
e=e,
e=fallback_trigger,
disable_fallbacks=False,
fallbacks=fallbacks,
context_window_fallbacks=context_window_fallbacks,
@ -5243,6 +5304,11 @@ class Router:
# share, leaking primary-deployment metadata into the mid-stream
# fallback request. safe_deep_copy avoids deep-copying the full
# kwargs (which can hold non-deepcopyable logging handles/clients).
# The pre-routing hook stamps its tier selection into this bucket during the primary
# attempt; seeding it before the snapshot gives both the live kwargs and the copy a
# bucket, so the post-call carry-over below always has somewhere to read and write.
kwargs.setdefault("litellm_metadata", {}) # mutable-ok: shared bucket # rebind-ok: stamp must be readable here
fallback_kwargs: Final[dict[str, object]] = kwargs.copy() # mutable-ok: mutated below before re-entry
if isinstance(fallback_kwargs.get("litellm_metadata"), dict):
fallback_kwargs["litellm_metadata"] = safe_deep_copy(fallback_kwargs["litellm_metadata"])
@ -5252,6 +5318,14 @@ class Router:
response: Final = await self._ageneric_api_call_with_fallbacks(original_function=original_function, **kwargs)
# The snapshot predates the pre-routing hook, so the tier it stamped into the live kwargs
# is carried over write-or-clear: a stale or caller-supplied selection left in the copy
# would key the mid-stream fallback lookup off a tier this attempt never routed to.
clear_pre_routing_selection(fallback_kwargs)
live_pre_routing_selection: Final = get_pre_routing_selection(kwargs)
if live_pre_routing_selection is not None:
record_pre_routing_selection(fallback_kwargs, live_pre_routing_selection)
if kwargs.get("stream") and hasattr(response, "__aiter__"):
return await self._aanthropic_messages_streaming_iterator(
response=cast("AsyncIterator[bytes]", response), # cast-ok: stream=True always returns a byte iterator
@ -6807,6 +6881,9 @@ class Router:
original_exception: Final = e
fallback_model_group = None
original_model_group: Final[str | None] = kwargs.get("model")
# A pre-routing hook (complexity / auto / adaptive / quality routers) picks a tier
# behind the router name, and fallbacks are configured per tier, not per router.
lookup_groups: Final[tuple[str, ...]] = fallback_lookup_groups(kwargs, model_group)
fallback_failure_exception_str = ""
if disable_fallbacks is True or original_model_group is None:
@ -6851,15 +6928,15 @@ class Router:
]
# Get external fallbacks — handle both standard and non-standard formats
external_fallback_group: list | None = None
if fallbacks is not None and model_group is not None:
if fallbacks is not None and lookup_groups:
if _check_non_standard_fallback_format(fallbacks=fallbacks):
# Non-standard formats (e.g. ["claude-3-haiku"] or
# [{"model": "...", "messages": [...]}]) are passed through directly
external_fallback_group = fallbacks
else:
external_fallback_group, generic_idx = get_fallback_model_group(
external_fallback_group, generic_idx = get_fallback_model_group_for_lookup_groups(
fallbacks=fallbacks,
model_group=cast(str, model_group),
lookup_groups=lookup_groups,
)
if external_fallback_group is None and generic_idx is not None:
external_fallback_group = fallbacks[generic_idx]["*"]
@ -6917,9 +6994,9 @@ class Router:
if isinstance(e, litellm.ContextWindowExceededError):
if context_window_fallbacks is not None:
context_window_fallback_model_group: Final[list[str] | None] = (
self._get_fallback_model_group_from_fallbacks(
self._get_fallback_model_group_for_lookup_groups(
fallbacks=context_window_fallbacks,
model_group=model_group,
lookup_groups=lookup_groups,
)
)
if context_window_fallback_model_group is None:
@ -6950,9 +7027,9 @@ class Router:
elif isinstance(e, litellm.ContentPolicyViolationError):
if content_policy_fallbacks is not None:
content_policy_fallback_model_group: Final[list[str] | None] = (
self._get_fallback_model_group_from_fallbacks(
self._get_fallback_model_group_for_lookup_groups(
fallbacks=content_policy_fallbacks,
model_group=model_group,
lookup_groups=lookup_groups,
)
)
if content_policy_fallback_model_group is None:
@ -6979,14 +7056,14 @@ class Router:
if litellm.expose_router_debug_in_errors:
e.message += f"\n{error_message}"
if fallbacks is not None and model_group is not None:
if fallbacks is not None and lookup_groups:
verbose_router_logger.debug("inside model fallbacks: %s", mask_sensitive_structure(fallbacks))
(
fallback_model_group,
generic_fallback_idx,
) = get_fallback_model_group(
) = get_fallback_model_group_for_lookup_groups(
fallbacks=fallbacks, # if fallbacks = [{"gpt-3.5-turbo": ["claude-3-haiku"]}]
model_group=cast(str, model_group),
lookup_groups=lookup_groups,
)
## if none, check for generic fallback
if fallback_model_group is None and generic_fallback_idx is not None:
@ -6995,12 +7072,12 @@ class Router:
if fallback_model_group is None:
masked_fallbacks: Final = mask_sensitive_structure(fallbacks)
verbose_router_logger.info(
"No fallback model group found for original model_group=%s. Fallbacks=%s",
model_group,
"No fallback model group found for lookup_groups=%s. Fallbacks=%s",
" -> ".join(lookup_groups),
masked_fallbacks,
)
if hasattr(original_exception, "message") and litellm.expose_router_debug_in_errors:
original_exception.message += f"No fallback model group found for original model_group={model_group}. Fallbacks={masked_fallbacks}"
original_exception.message += f"No fallback model group found for lookup_groups={' -> '.join(lookup_groups)}. Fallbacks={masked_fallbacks}"
raise original_exception
input_kwargs.update(
@ -7046,6 +7123,7 @@ class Router:
If it fails after num_retries, fall back to another model group
"""
model_group: Final[str | None] = kwargs.get("model")
clear_pre_routing_selection(kwargs) # pyright: ignore[reportUnknownArgumentType] # **kwargs is untyped at this boundary
if not isinstance(kwargs.get("attempted_targets"), AttemptedFallbackTargets):
_fallback_metadata_key: Final = _get_router_metadata_variable_name(
function_name=getattr(kwargs.get("original_function"), "__name__", None)
@ -7471,6 +7549,24 @@ class Router:
break
return fallback_model_group
def _get_fallback_model_group_for_lookup_groups(
self,
fallbacks: list[dict[str, list[str]]], # mutable-ok: mirrors the sibling resolver's contract
lookup_groups: tuple[str, ...],
) -> list[str] | None: # mutable-ok: mirrors the sibling resolver's contract
"""First lookup group whose exact-key chain resolves (tier first, then requested group)."""
return next(
(
resolved
for resolved in (
self._get_fallback_model_group_from_fallbacks(fallbacks=fallbacks, model_group=group)
for group in lookup_groups
)
if resolved is not None
),
None,
)
def _get_first_default_fallback(self) -> str | None:
"""
Returns the first model from the default_fallbacks list, if it exists.
@ -7886,6 +7982,31 @@ class Router:
return True
return False
def _has_content_policy_fallback(self, model_group: str, kwargs: Mapping[str, Any]) -> bool:
"""
Whether a content-policy fallback would resolve for this request, keyed the same way
async_function_with_fallbacks_common_utils resolves it: the tier a pre-routing hook
selected wins over the requested group. Raising without this returning True would turn
a deliverable response into an error the fallback chain cannot recover from.
"""
content_policy_fallbacks: Final = kwargs.get("content_policy_fallbacks", self.content_policy_fallbacks)
if content_policy_fallbacks is not None:
return (
self._get_fallback_model_group_for_lookup_groups(
fallbacks=content_policy_fallbacks,
lookup_groups=fallback_lookup_groups(kwargs, model_group),
)
is not None
)
if self._has_default_fallbacks():
return True
verbose_router_logger.debug(
"No content-policy fallback available. Returning original response. model=%s, content_policy_fallbacks=%s",
model_group,
content_policy_fallbacks,
)
return False
def _should_raise_content_policy_error(self, model: str, response: ModelResponse, kwargs: dict) -> bool:
"""
Determines if a content policy error should be raised.
@ -7898,27 +8019,26 @@ class Router:
if response.choices[0].finish_reason != "content_filter":
return False
content_policy_fallbacks: Final = kwargs.get("content_policy_fallbacks", self.content_policy_fallbacks)
return self._has_content_policy_fallback(model, kwargs)
### ONLY RAISE ERROR IF CP FALLBACK AVAILABLE ###
if content_policy_fallbacks is not None:
fallback_model_group = None
for item in content_policy_fallbacks: # [{"gpt-3.5-turbo": ["gpt-4"]}]
if list(item.keys())[0] == model:
fallback_model_group = item[model]
break
if fallback_model_group is not None:
return True
elif self._has_default_fallbacks(): # default fallbacks set
return True
verbose_router_logger.debug(
"Content Policy Error occurred. No available fallbacks. Returning original response. model=%s, content_policy_fallbacks=%s",
model,
content_policy_fallbacks,
def _should_raise_anthropic_refusal_error(
self, model: str, original_generic_function: Callable, response: object, kwargs: Mapping[str, Any]
) -> bool:
"""
The /v1/messages twin of _should_raise_content_policy_error: an Anthropic safeguard
refusal (stop_reason "refusal" carrying stop_details) re-enters the fallback chain only
when a content-policy fallback is configured; a plain refusal without stop_details, or
any response with nothing configured, is returned to the client unchanged.
"""
from litellm.llms.anthropic.experimental_pass_through.messages.utils import (
get_safeguard_refusal_stop_details,
)
return False
if getattr(original_generic_function, "__name__", "") != "anthropic_messages":
return False
if get_safeguard_refusal_stop_details(response) is None:
return False
return self._has_content_policy_fallback(model, kwargs)
def _get_healthy_deployments(self, model: str, parent_otel_span: Span | None):
_all_deployments: list = []
@ -12087,6 +12207,7 @@ class Router:
if pre_routing_hook_response is not None:
model = pre_routing_hook_response.model
messages = pre_routing_hook_response.messages
record_pre_routing_selection(request_kwargs, model)
if pre_routing_hook_response.litellm_params:
accepted_tier_params: Final = self._tier_params_the_target_accepts(
model, pre_routing_hook_response.litellm_params, request_kwargs
@ -12202,6 +12323,7 @@ class Router:
if pre_routing_hook_response is not None:
model = pre_routing_hook_response.model
messages = pre_routing_hook_response.messages
record_pre_routing_selection(request_kwargs, model)
if pre_routing_hook_response.litellm_params:
accepted_tier_params: Final = self._tier_params_the_target_accepts(
model, pre_routing_hook_response.litellm_params, request_kwargs

85
litellm/router_utils/fallback_event_handlers.py
View file

@ -214,6 +214,91 @@ def _check_stripped_model_group(model_group: str, fallback_key: str) -> bool:
return False
PRE_ROUTING_SELECTED_MODEL_KEY: Final = "pre_routing_selected_model"
_ROUTER_METADATA_BUCKETS: Final = ("metadata", "litellm_metadata")
def record_pre_routing_selection(request_kwargs: Mapping[str, Any] | None, selected_model: str) -> None:
"""
Remember which model a pre-routing hook picked, so fallback lookup can key off it.
Fallback resolution runs on an outer kwargs dict that ``**kwargs`` already copied, so
writing the model there is invisible by the time routing picks a tier. The metadata
buckets are nested dicts shared by reference across those copies, which is how the
router already carries values back up.
The write goes through the proxy-internal bucket resolver, never into both buckets:
on /v1/messages the top-level ``metadata`` dict is the provider's own request field,
so a blanket write would forward the tier stamp upstream.
"""
from litellm.litellm_core_utils.core_helpers import get_metadata_variable_name_from_kwargs
if request_kwargs is None:
return
bucket: Final = request_kwargs.get(get_metadata_variable_name_from_kwargs(request_kwargs))
if isinstance(bucket, dict):
bucket[PRE_ROUTING_SELECTED_MODEL_KEY] = selected_model
def clear_pre_routing_selection(request_kwargs: Mapping[str, object] | None) -> None:
"""
Drop any selection the router did not make itself on this hop.
The buckets carry whatever the caller sent, so an inbound value is the caller
choosing a fallback chain rather than the router choosing a tier. A fallback hop
also inherits the previous hop's selection, which would key its own failure off
the tier that already failed. Clearing at the start of every hop leaves only a
value the pre-routing hook wrote while routing that hop.
"""
if request_kwargs is None:
return
for bucket in (request_kwargs.get(name) for name in _ROUTER_METADATA_BUCKETS):
if isinstance(bucket, dict) and PRE_ROUTING_SELECTED_MODEL_KEY in bucket:
del bucket[PRE_ROUTING_SELECTED_MODEL_KEY]
def get_pre_routing_selection(kwargs: Mapping[str, Any]) -> str | None:
"""The model a pre-routing hook selected for this request, if one did."""
buckets: Final = (kwargs.get(name) for name in _ROUTER_METADATA_BUCKETS)
selections: Final = (bucket.get(PRE_ROUTING_SELECTED_MODEL_KEY) for bucket in buckets if isinstance(bucket, dict))
return next((selected for selected in selections if isinstance(selected, str) and selected), None)
def fallback_lookup_groups(kwargs: Mapping[str, Any], model_group: str | None) -> tuple[str, ...]:
"""
Ordered keys for resolving a fallback chain: the tier a pre-routing hook selected wins,
and the requested group still resolves when no tier-keyed chain exists, so configs keyed
on the router name (the documented contract) keep working behind auto-routers.
"""
ordered: Final = (get_pre_routing_selection(kwargs), model_group)
return tuple(dict.fromkeys(group for group in ordered if group))
def _resolved_a_specific_chain(
fallbacks: list[Any], # mutable-ok: mirrors get_fallback_model_group's contract
result: tuple[list[str] | None, int | None], # mutable-ok: mirrors get_fallback_model_group's contract
) -> bool:
resolved, generic_idx = result
if resolved is None:
return False
return generic_idx is None or resolved is not fallbacks[generic_idx]["*"]
def get_fallback_model_group_for_lookup_groups(
fallbacks: list[Any], # mutable-ok: mirrors get_fallback_model_group's contract
lookup_groups: tuple[str, ...],
) -> tuple[list[str] | None, int | None]: # mutable-ok: mirrors get_fallback_model_group's contract
"""
First lookup group with a specifically-keyed chain wins; the generic "*" chain applies
only after every group missed, so a catch-all cannot shadow a later group's own chain.
"""
results: Final = tuple(get_fallback_model_group(fallbacks=fallbacks, model_group=group) for group in lookup_groups)
specific: Final = next((result for result in results if _resolved_a_specific_chain(fallbacks, result)), None)
if specific is not None:
return specific
return next((result for result in results if result[0] is not None), (None, None))
def get_fallback_model_group(fallbacks: list[Any], model_group: str) -> tuple[list[str] | None, int | None]:
"""
Returns:

13
litellm/types/llms/anthropic_messages/anthropic_response.py
View file

@ -78,6 +78,16 @@ class AnthropicUsage(TypedDict, total=False):
server_tool_use: NotRequired[ReadOnly[ServerToolUsage]]
class AnthropicStopDetails(TypedDict, total=False):
"""
Safeguard verdict accompanying a `stop_reason: "refusal"` response:
https://platform.claude.com/docs/en/build-with-claude/refusals-and-fallback
"""
category: ReadOnly[str | None]
explanation: ReadOnly[str | None]
class AnthropicMessagesResponse(TypedDict, total=False):
"""
Anthropic Messages API Response: https://docs.anthropic.com/en/api/messages
@ -90,7 +100,8 @@ class AnthropicMessagesResponse(TypedDict, total=False):
id: str
model: str | None # This represents the Model type from Anthropic
role: Literal["assistant"] | None
stop_reason: Literal["end_turn", "max_tokens", "stop_sequence", "tool_use"] | None
stop_reason: Literal["end_turn", "max_tokens", "stop_sequence", "tool_use", "refusal"] | None
stop_details: NotRequired[ReadOnly[AnthropicStopDetails | None]]
stop_sequence: str | None
type: Literal["message"] | None
usage: AnthropicUsage | None

402
tests/router_unit_tests/test_router_anthropic_messages_fallback.py Normal file
View file

@ -0,0 +1,402 @@
"""
Unit tests for safeguard-refusal fallback on the /v1/messages router surface.
An Anthropic safeguard refusal is an HTTP 200 whose body carries
stop_reason "refusal" plus a stop_details object; the router converts it
into a ContentPolicyViolationError so the content-policy fallback chain
runs, but only when a matching fallback is configured. A plain refusal
without stop_details, or any refusal with nothing configured, must reach
the client byte-identical.
The upstream is faked at the HTTP boundary by intercepting the third-party
transport (httpx.AsyncClient.send), so requests run litellm's real
transformation, allowlist, and streaming pipeline end to end.
"""
import json
from typing import Any, AsyncIterator
from unittest.mock import patch
import httpx
import pytest
from litellm import Router
from litellm.router_utils.fallback_event_handlers import (
PRE_ROUTING_SELECTED_MODEL_KEY,
record_pre_routing_selection,
)
REFUSAL_RESPONSE: dict[str, Any] = {
"id": "msg_refusal",
"type": "message",
"role": "assistant",
"model": "claude-fable-5",
"content": [],
"stop_reason": "refusal",
"stop_sequence": None,
"stop_details": {"category": "cyber", "explanation": "flagged"},
"usage": {"input_tokens": 25, "output_tokens": 1},
}
PLAIN_REFUSAL_RESPONSE: dict[str, Any] = {k: v for k, v in REFUSAL_RESPONSE.items() if k != "stop_details"}
OK_RESPONSE: dict[str, Any] = {
"id": "msg_ok",
"type": "message",
"role": "assistant",
"model": "claude-opus-5",
"content": [{"type": "text", "text": "hello"}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 25, "output_tokens": 2},
}
def _sse(event: str, data: dict[str, Any]) -> bytes:
return f"event: {event}\ndata: {json.dumps(data)}\n\n".encode()
REFUSAL_STREAM_FRAMES: tuple[bytes, ...] = (
_sse("message_start", {"type": "message_start", "message": {**REFUSAL_RESPONSE, "stop_reason": None}}),
_sse(
"message_delta",
{
"type": "message_delta",
"delta": {"stop_reason": "refusal", "stop_details": {"category": "cyber"}},
"usage": {"output_tokens": 1},
},
),
_sse("message_stop", {"type": "message_stop"}),
)
OK_STREAM_FRAMES: tuple[bytes, ...] = (
_sse("message_start", {"type": "message_start", "message": {**OK_RESPONSE, "stop_reason": None}}),
_sse(
"content_block_delta",
{"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "hello"}},
),
_sse("message_stop", {"type": "message_stop"}),
)
def _split_frames_mid_data_line(frames: tuple[bytes, ...]) -> tuple[bytes, ...]:
"""Split each frame's data line in half, modeling a transport chunk boundary."""
return tuple(part for frame in frames for part in (frame[: len(frame) // 2], frame[len(frame) // 2 :]))
class _FrameStream(httpx.AsyncByteStream):
def __init__(self, frames: tuple[bytes, ...]) -> None:
self._frames = frames
async def __aiter__(self) -> AsyncIterator[bytes]:
for frame in self._frames:
yield frame
async def aclose(self) -> None:
return None
class FakeAnthropicUpstream:
"""Intercepts the third-party transport (httpx.AsyncClient.send): refuses on fable
models, answers on others. The router deliberately does not forward caller-injected
clients, so the transport is the seam that exercises the real litellm pipeline."""
def __init__(
self,
refusal_body: dict[str, Any] = REFUSAL_RESPONSE,
refusal_frames: tuple[bytes, ...] = REFUSAL_STREAM_FRAMES,
) -> None:
self.refusal_body = refusal_body
self.refusal_frames = refusal_frames
self.calls: list[str] = []
self.bodies: list[dict[str, Any]] = []
async def send(self, request: httpx.Request, **kwargs: Any) -> httpx.Response:
body = json.loads(request.content or b"{}")
model = body.get("model", "")
self.calls.append(model)
self.bodies.append(body)
refuses = "fable" in model
if body.get("stream"):
frames = self.refusal_frames if refuses else OK_STREAM_FRAMES
return httpx.Response(
200,
stream=_FrameStream(frames),
headers={"content-type": "text/event-stream"},
request=request,
)
return httpx.Response(200, json=self.refusal_body if refuses else OK_RESPONSE, request=request)
def install(self):
async def _send(_client: httpx.AsyncClient, request: httpx.Request, **kwargs: Any) -> httpx.Response:
return await self.send(request, **kwargs)
return patch("httpx.AsyncClient.send", new=_send)
FABLE_TIER = {
"model_name": "fable-tier",
"litellm_params": {"model": "anthropic/claude-fable-5", "api_key": "sk-test"},
}
OPUS_TARGET = {
"model_name": "opus-target",
"litellm_params": {"model": "anthropic/claude-opus-5", "api_key": "sk-test"},
}
def _router(content_policy_fallbacks: list | None) -> Router:
return Router(model_list=[FABLE_TIER, OPUS_TARGET], content_policy_fallbacks=content_policy_fallbacks)
async def _collect(stream: AsyncIterator[bytes]) -> bytes:
return b"".join([chunk async for chunk in stream])
@pytest.mark.asyncio
async def test_non_streaming_refusal_with_fallback_row_returns_fallback_response():
fake = FakeAnthropicUpstream()
router = _router(content_policy_fallbacks=[{"fable-tier": ["opus-target"]}])
with fake.install():
response = await router.aanthropic_messages(
model="fable-tier", max_tokens=16, messages=[{"role": "user", "content": "hi"}]
)
assert response["stop_reason"] == "end_turn"
assert response["id"] == "msg_ok"
assert len(fake.calls) == 2
assert "claude-opus-5" in fake.calls[1]
@pytest.mark.asyncio
@pytest.mark.parametrize(
"content_policy_fallbacks, upstream_body",
[
(None, REFUSAL_RESPONSE),
([{"unrelated-group": ["opus-target"]}], REFUSAL_RESPONSE),
([{"fable-tier": ["opus-target"]}], PLAIN_REFUSAL_RESPONSE),
],
ids=["nothing-configured", "row-for-other-group", "refusal-without-stop-details"],
)
async def test_non_streaming_refusal_passes_through_untouched(content_policy_fallbacks, upstream_body):
fake = FakeAnthropicUpstream(refusal_body=upstream_body)
router = _router(content_policy_fallbacks=content_policy_fallbacks)
with fake.install():
response = await router.aanthropic_messages(
model="fable-tier", max_tokens=16, messages=[{"role": "user", "content": "hi"}]
)
assert response["stop_reason"] == "refusal"
assert response.get("stop_details") == upstream_body.get("stop_details")
assert len(fake.calls) == 1
@pytest.mark.asyncio
async def test_streaming_refusal_with_fallback_row_streams_fallback_frames():
fake = FakeAnthropicUpstream()
router = _router(content_policy_fallbacks=[{"fable-tier": ["opus-target"]}])
with fake.install():
stream = await router.aanthropic_messages(
model="fable-tier", max_tokens=16, stream=True, messages=[{"role": "user", "content": "hi"}]
)
body = await _collect(stream)
assert b'"refusal"' not in body
assert b"text_delta" in body
assert len(fake.calls) == 2
@pytest.mark.asyncio
async def test_streaming_refusal_split_across_chunks_still_falls_back():
fake = FakeAnthropicUpstream(refusal_frames=_split_frames_mid_data_line(REFUSAL_STREAM_FRAMES))
router = _router(content_policy_fallbacks=[{"fable-tier": ["opus-target"]}])
with fake.install():
stream = await router.aanthropic_messages(
model="fable-tier", max_tokens=16, stream=True, messages=[{"role": "user", "content": "hi"}]
)
body = await _collect(stream)
assert b'"refusal"' not in body
assert b"text_delta" in body
assert len(fake.calls) == 2
@pytest.mark.asyncio
async def test_streaming_refusal_without_fallback_row_passes_frames_through():
fake = FakeAnthropicUpstream()
router = _router(content_policy_fallbacks=None)
with fake.install():
stream = await router.aanthropic_messages(
model="fable-tier", max_tokens=16, stream=True, messages=[{"role": "user", "content": "hi"}]
)
body = await _collect(stream)
assert b'"stop_reason": "refusal"' in body
assert b"stop_details" in body
assert len(fake.calls) == 1
@pytest.mark.asyncio
async def test_streaming_refusal_on_routed_tier_matches_tier_keyed_row_without_inbound_metadata():
"""The pre-routing hook's tier stamp must reach the mid-stream fallback lookup even when the
request carries no metadata bucket at all (the snapshot is taken before the request runs)."""
fake = FakeAnthropicUpstream()
smart_router = {
"model_name": "smart-router",
"litellm_params": {
"model": "auto_router/complexity_router",
"complexity_router_config": {
"tiers": {"SIMPLE": "fable-tier", "MEDIUM": "fable-tier", "COMPLEX": "fable-tier"}
},
"complexity_router_default_model": "fable-tier",
},
"model_info": {"id": "router-1", "db_model": True},
}
router = Router(
model_list=[FABLE_TIER, OPUS_TARGET, smart_router],
content_policy_fallbacks=[{"fable-tier": ["opus-target"]}],
ignore_invalid_deployments=True,
)
with fake.install():
stream = await router.aanthropic_messages(
model="smart-router", max_tokens=16, stream=True, messages=[{"role": "user", "content": "hi"}]
)
body = await _collect(stream)
assert b'"refusal"' not in body
assert b"text_delta" in body
assert len(fake.calls) == 2
@pytest.mark.asyncio
async def test_caller_forged_tier_stamp_cannot_pick_the_streaming_fallback_chain():
fake = FakeAnthropicUpstream()
router = _router(content_policy_fallbacks=[{"forged-tier": ["opus-target"]}])
with fake.install():
stream = await router.aanthropic_messages(
model="fable-tier",
max_tokens=16,
stream=True,
messages=[{"role": "user", "content": "hi"}],
litellm_metadata={PRE_ROUTING_SELECTED_MODEL_KEY: "forged-tier"},
)
body = await _collect(stream)
assert b'"stop_reason": "refusal"' in body
assert len(fake.calls) == 1
@pytest.mark.asyncio
async def test_tier_stamp_never_reaches_provider_bound_metadata():
"""On /v1/messages the top-level metadata dict is Anthropic's own request field, so the
routed-tier stamp must never appear in any upstream body even when the client sends one."""
fake = FakeAnthropicUpstream()
smart_router = {
"model_name": "smart-router",
"litellm_params": {
"model": "auto_router/complexity_router",
"complexity_router_config": {
"tiers": {"SIMPLE": "fable-tier", "MEDIUM": "fable-tier", "COMPLEX": "fable-tier"}
},
"complexity_router_default_model": "fable-tier",
},
"model_info": {"id": "router-1", "db_model": True},
}
router = Router(
model_list=[FABLE_TIER, OPUS_TARGET, smart_router],
content_policy_fallbacks=[{"fable-tier": ["opus-target"]}],
ignore_invalid_deployments=True,
)
with fake.install():
response = await router.aanthropic_messages(
model="smart-router",
max_tokens=16,
messages=[{"role": "user", "content": "hi"}],
metadata={"user_id": "u1"},
)
assert response["stop_reason"] == "end_turn"
assert len(fake.bodies) == 2
for body in fake.bodies:
assert body.get("metadata") == {"user_id": "u1"}
def test_record_pre_routing_selection_writes_only_the_internal_bucket():
"""The Anthropic request's own metadata field must never carry the tier stamp."""
kwargs = {"metadata": {"user_id": "u1"}, "litellm_metadata": {}}
record_pre_routing_selection(kwargs, "tier-x")
assert kwargs["litellm_metadata"] == {PRE_ROUTING_SELECTED_MODEL_KEY: "tier-x"}
assert kwargs["metadata"] == {"user_id": "u1"}
def test_refusal_gate_keys_on_pre_routing_tier_stamp():
router = _router(content_policy_fallbacks=[{"tier-group": ["opus-target"]}])
def anthropic_messages(**kwargs: Any) -> None:
return None
refusal_kwargs = {"litellm_metadata": {PRE_ROUTING_SELECTED_MODEL_KEY: "tier-group"}}
assert (
router._should_raise_anthropic_refusal_error(
model="router-group",
original_generic_function=anthropic_messages,
response=dict(REFUSAL_RESPONSE),
kwargs=refusal_kwargs,
)
is True
)
assert (
router._should_raise_anthropic_refusal_error(
model="router-group",
original_generic_function=anthropic_messages,
response=dict(REFUSAL_RESPONSE),
kwargs={},
)
is False
)
def test_has_content_policy_fallback_default_fallbacks_arm():
router = Router(model_list=[OPUS_TARGET], fallbacks=[{"*": ["opus-target"]}])
assert router._has_content_policy_fallback("any-group", {}) is True
assert router._has_content_policy_fallback("any-group", {"content_policy_fallbacks": [{"other": ["x"]}]}) is False
def test_get_fallback_model_group_for_lookup_groups_orders_tier_before_requested():
router = _router(content_policy_fallbacks=None)
fallbacks = [{"tier1": ["backup-a"]}, {"smart-router": ["backup-b"]}]
assert router._get_fallback_model_group_for_lookup_groups(
fallbacks=fallbacks, lookup_groups=("tier1", "smart-router")
) == ["backup-a"]
assert router._get_fallback_model_group_for_lookup_groups(
fallbacks=fallbacks, lookup_groups=("tier9", "smart-router")
) == ["backup-b"]
assert router._get_fallback_model_group_for_lookup_groups(fallbacks=fallbacks, lookup_groups=()) is None
def test_refusal_gate_ignores_other_generic_call_types():
router = _router(content_policy_fallbacks=[{"fable-tier": ["opus-target"]}])
def aresponses(**kwargs: Any) -> None:
return None
assert (
router._should_raise_anthropic_refusal_error(
model="fable-tier",
original_generic_function=aresponses,
response=dict(REFUSAL_RESPONSE),
kwargs={},
)
is False
)

119
tests/test_litellm/router_utils/test_fallback_event_handlers.py
View file

@ -11,7 +11,10 @@ from litellm.router_utils.fallback_event_handlers import (
AttemptedFallbackTargets,
_trigger_cooldown_for_failed_deployment,
fallback_attempt_key,
clear_pre_routing_selection,
get_fallback_model_group,
get_pre_routing_selection,
record_pre_routing_selection,
run_async_fallback,
)
@ -1090,3 +1093,119 @@ async def test_run_async_fallback_preserves_original_model_group_on_nested_fallb
metadata = router.received_kwargs["metadata"]
assert metadata["attempted_fallbacks"] == 2
assert metadata["original_model_group"] == "primary-model"
class TestPreRoutingSelectionCarriesToFallbacks:
"""#38832: a complexity/auto router picks a tier behind the router name, but fallback
lookup kept using the router name, so the tier's configured chain never ran."""
def test_selection_is_recorded_in_the_metadata_bucket(self):
kwargs = {"model": "smart-router", "metadata": {}}
record_pre_routing_selection(kwargs, "tier1")
assert kwargs["metadata"]["pre_routing_selected_model"] == "tier1"
assert get_pre_routing_selection(kwargs) == "tier1"
def test_selection_is_recorded_in_the_litellm_metadata_bucket(self):
kwargs = {"model": "smart-router", "litellm_metadata": {}}
record_pre_routing_selection(kwargs, "tier2")
assert get_pre_routing_selection(kwargs) == "tier2"
def test_a_bucket_survives_the_kwargs_copy_that_fallbacks_run_on(self):
"""The bucket is shared by reference, which is the whole reason this works."""
outer = {"model": "smart-router", "metadata": {}}
inner = {**outer}
record_pre_routing_selection(inner, "tier1")
assert get_pre_routing_selection(outer) == "tier1"
def test_no_selection_reads_as_none(self):
assert get_pre_routing_selection({"model": "smart-router", "metadata": {}}) is None
assert get_pre_routing_selection({"model": "smart-router"}) is None
def test_missing_kwargs_is_a_no_op(self):
"""A caller with no kwargs must not raise, and must not leak the selection anywhere."""
record_pre_routing_selection(None, "tier1")
assert get_pre_routing_selection({}) is None
def test_a_non_dict_bucket_is_ignored(self):
kwargs = {"model": "smart-router", "metadata": "not-a-dict"}
record_pre_routing_selection(kwargs, "tier1")
assert get_pre_routing_selection(kwargs) is None
def test_fallbacks_resolve_against_the_selected_tier(self):
"""The lookup the router performs, keyed on the tier rather than the router name."""
fallbacks = [{"tier1": ["backup-a", "backup-b"]}, {"tier2": ["backup-c"]}]
assert get_fallback_model_group(fallbacks=fallbacks, model_group="tier1")[0] == ["backup-a", "backup-b"]
assert get_fallback_model_group(fallbacks=fallbacks, model_group="smart-router")[0] is None
class TestPreRoutingSelectionIsPerHop:
"""#38832 review: the buckets also carry whatever the caller sent, and a fallback hop
inherits the previous hop's tier, so a hop must start without a selection."""
def test_a_caller_supplied_selection_is_dropped(self):
kwargs = {"model": "plain", "metadata": {"pre_routing_selected_model": "tier1"}}
clear_pre_routing_selection(kwargs)
assert get_pre_routing_selection(kwargs) is None
assert "pre_routing_selected_model" not in kwargs["metadata"]
def test_both_buckets_are_cleared(self):
kwargs = {
"metadata": {"pre_routing_selected_model": "tier1"},
"litellm_metadata": {"pre_routing_selected_model": "tier2"},
}
clear_pre_routing_selection(kwargs)
assert get_pre_routing_selection(kwargs) is None
def test_the_rest_of_the_bucket_is_left_alone(self):
kwargs = {"metadata": {"pre_routing_selected_model": "tier1", "tags": ["a"]}}
clear_pre_routing_selection(kwargs)
assert kwargs["metadata"] == {"tags": ["a"]}
def test_clearing_is_a_no_op_without_a_usable_bucket(self):
kwargs = {"model": "plain", "metadata": "not-a-dict"}
clear_pre_routing_selection(None)
clear_pre_routing_selection(kwargs)
assert kwargs == {"model": "plain", "metadata": "not-a-dict"}
def test_a_selection_recorded_after_clearing_is_kept(self):
"""Clearing runs before routing, so the hook's own write must survive it."""
kwargs = {"model": "smart-router", "metadata": {"pre_routing_selected_model": "stale"}}
clear_pre_routing_selection(kwargs)
record_pre_routing_selection(kwargs, "tier1")
assert get_pre_routing_selection(kwargs) == "tier1"
class TestOrderedFallbackLookupGroups:
def test_tier_first_then_requested_group_deduped(self):
from litellm.router_utils.fallback_event_handlers import (
PRE_ROUTING_SELECTED_MODEL_KEY,
fallback_lookup_groups,
)
kwargs = {"litellm_metadata": {PRE_ROUTING_SELECTED_MODEL_KEY: "tier1"}}
assert fallback_lookup_groups(kwargs, "smart-router") == ("tier1", "smart-router")
assert fallback_lookup_groups(kwargs, "tier1") == ("tier1",)
assert fallback_lookup_groups({}, "smart-router") == ("smart-router",)
assert fallback_lookup_groups({}, None) == ()
def test_first_resolving_group_wins_and_generic_idx_survives_a_miss(self):
from litellm.router_utils.fallback_event_handlers import (
get_fallback_model_group_for_lookup_groups,
)
fallbacks = [{"tier1": ["backup-a"]}, {"smart-router": ["backup-b"]}, {"*": ["backup-c"]}]
assert get_fallback_model_group_for_lookup_groups(fallbacks, ("tier1", "smart-router")) == (["backup-a"], None)
assert get_fallback_model_group_for_lookup_groups(fallbacks, ("tier9", "smart-router")) == (["backup-b"], None)
assert get_fallback_model_group_for_lookup_groups(fallbacks, ("tier9", "no-such")) == (["backup-c"], 2)
assert get_fallback_model_group_for_lookup_groups([{"tier1": ["backup-a"]}], ("no", "nope")) == (None, None)

135
tests/test_litellm/test_router.py
View file

@ -11595,3 +11595,138 @@ class TestTierParamsTheTargetAccepts:
accepted = router._tier_params_the_target_accepts("no-such-group", {"reasoning_effort": "max"}, {})
assert accepted == {"reasoning_effort": "max"}
class TestPreRoutingTierDrivesFallbacks:
"""#38832: a complexity/auto router picks a tier behind the router name, but fallback
lookup stayed on the router name, so the tier's configured chain never ran and a
provider failure on the tier's first hop was returned to the client."""
class _TierRouter(litellm.Router):
async def async_pre_routing_hook(
self, model, request_kwargs, messages=None, input=None, specific_deployment=False
):
from litellm.types.router import PreRoutingHookResponse
if model == "smart-router":
return PreRoutingHookResponse(model="tier1", messages=messages)
return None
@classmethod
def _router(cls, fallbacks) -> "litellm.Router":
return cls._TierRouter(
model_list=[
{
"model_name": "smart-router",
"litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "sk-x"},
},
{
"model_name": "tier1",
"litellm_params": {
"model": "openai/gpt-4o-mini",
"api_key": "sk-x",
"mock_response": "litellm.RateLimitError",
},
},
{
"model_name": "backup-a",
"litellm_params": {
"model": "openai/gpt-4o-mini",
"api_key": "sk-x",
"mock_response": "from backup-a",
},
},
{
"model_name": "backup-b",
"litellm_params": {
"model": "openai/gpt-4o-mini",
"api_key": "sk-x",
"mock_response": "from backup-b",
},
},
{
"model_name": "failing-backup",
"litellm_params": {
"model": "openai/gpt-4o-mini",
"api_key": "sk-x",
"mock_response": "litellm.RateLimitError",
},
},
{
"model_name": "plain",
"litellm_params": {
"model": "openai/gpt-4o-mini",
"api_key": "sk-x",
"mock_response": "litellm.RateLimitError",
},
},
],
fallbacks=fallbacks,
num_retries=0,
)
@pytest.mark.asyncio
async def test_the_selected_tier_fallback_chain_runs(self):
router = self._router([{"tier1": ["backup-a"]}])
response = await router.acompletion(
model="smart-router", messages=[{"role": "user", "content": "hi"}]
)
assert response.choices[0].message.content == "from backup-a"
@pytest.mark.asyncio
async def test_a_chain_keyed_on_the_router_name_is_not_used(self):
"""The router name has no chain of its own, so nothing should rescue this call."""
router = self._router([{"tier2": ["backup-a"]}])
with pytest.raises(litellm.RateLimitError):
await router.acompletion(model="smart-router", messages=[{"role": "user", "content": "hi"}])
@pytest.mark.asyncio
async def test_a_chain_keyed_on_the_router_name_rescues_when_no_tier_chain_exists(self):
"""The documented contract: configs keyed on the requested name keep working behind auto-routers."""
router = self._router([{"smart-router": ["backup-a"]}])
response = await router.acompletion(model="smart-router", messages=[{"role": "user", "content": "hi"}])
assert response.choices[0].message.content == "from backup-a"
@pytest.mark.asyncio
async def test_the_tier_chain_wins_over_the_router_name_chain(self):
router = self._router([{"tier1": ["backup-a"]}, {"smart-router": ["backup-b"]}])
response = await router.acompletion(model="smart-router", messages=[{"role": "user", "content": "hi"}])
assert response.choices[0].message.content == "from backup-a"
@pytest.mark.asyncio
async def test_a_request_without_a_pre_routing_hook_still_uses_its_own_group(self):
router = self._router([{"tier1": ["backup-a"]}])
response = await router.acompletion(
model="tier1", messages=[{"role": "user", "content": "hi"}]
)
assert response.choices[0].message.content == "from backup-a"
@pytest.mark.asyncio
async def test_a_caller_cannot_pick_the_chain_by_sending_the_selection(self):
"""The metadata bucket carries caller-supplied keys, so only the hook may set the tier."""
router = self._router([{"tier1": ["backup-a"]}])
with pytest.raises(litellm.RateLimitError):
await router.acompletion(
model="plain",
messages=[{"role": "user", "content": "hi"}],
metadata={"pre_routing_selected_model": "tier1"},
)
@pytest.mark.asyncio
async def test_each_fallback_hop_resolves_its_own_chain(self):
"""The second hop must key off the group it is running, not the tier that failed."""
router = self._router([{"tier1": ["failing-backup"]}, {"failing-backup": ["backup-b"]}])
response = await router.acompletion(model="smart-router", messages=[{"role": "user", "content": "hi"}])
assert response.choices[0].message.content == "from backup-b"