scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-12 08:42:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 207

fix(router): stamp the failed deployment on every endpoint's attempt (#49941)

Browse source 
* fix(router): stamp the failed deployment on every endpoint's attempt

Weighted failover (#27980) re-picks within a model group by excluding the
deployment named in the error's failed_deployment_id. #27980 stamped that id
in only three attempt functions: _completion, _acompletion and
_ageneric_api_call_with_fallbacks_helper. Embedding, image generation,
transcription, speech, text completion, rerank, adapter completion, the sync
generic call and vector store create never stamped it, so with
enable_weighted_failover on, a failure on one of their deployments skipped the
in-group re-pick and went straight to the cross-group fallbacks, or raised.

Each of those attempts now stamps the deployment it picked, read from the
model_info that _update_kwargs_with_deployment put on the request (so a
client-side-credential call names its own dynamic deployment, as on
completions). A failed pick stamps nothing. File and batch creation and batch
cancellation are left alone: they are bound to a resource on the deployment
they picked, so re-picking another one cannot help.

* test(router): narrow the failed-pick raise to the no-healthy-deployments error

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013FDwtoMjnHa4xxazGJJNSm

* fix(router): exclude the configured deployment on failover for client-side credentials

Weighted failover excluded the per-request id that client-supplied credentials
generate, so the failed configured deployment stayed eligible for the re-pick.
Stamp the configured id alongside it and exclude that. Also drop the vector
store create stamp, which no fallback path reads, and redundant test comments.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013FDwtoMjnHa4xxazGJJNSm

---------

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Nate Armstrong 2026-11-04 11:08:21 -08:00 • committed by GitHub
parent b662f3f7a4
commit 8cbd40893e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 204 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

74
litellm/router_python_core.py
View file

@ -5211,8 +5211,19 @@ class PythonCore(DeprecatedRouterMembers):
# _update_kwargs_with_deployment and stamped into kwargs["model_info"]; stamping
# the static shared deployment's id instead would let one tenant's bad credentials
# cool down the deployment every other tenant sharing this config relies on.
if getattr(exception, "failed_deployment_id", None):
return
effective_model_info: Final = kwargs.get("model_info") or deployment.get("model_info") or MappingProxyType({})
self._set_failed_deployment_id_on_exception(exception, MappingProxyType({"model_info": effective_model_info}))
configured_deployment_id: Final = _MODEL_INFO_ADAPTER.validate_python(effective_model_info).get(
"original_model_id"
)
if isinstance(configured_deployment_id, str) and configured_deployment_id:
exception.failed_configured_deployment_id = configured_deployment_id # pyright: ignore[reportAttributeAccessIssue] # rebind-ok: the exception carries the failure to _maybe_run_weighted_failover
def _stamp_failed_attempt(self, exception: Exception, kwargs: Mapping[str, object], *, picked: bool) -> None:
if picked:
self._stamp_failed_deployment_id_with_effective_model_info(exception, MappingProxyType({}), kwargs)
@staticmethod
def _stamp_retry_skip_deployment_id(exception: Exception, kwargs: Mapping[str, object]) -> None:
@ -6089,9 +6100,10 @@ class PythonCore(DeprecatedRouterMembers):
def _image_generation(self, prompt: str, model: str, **kwargs):
model_name: Final = ""
deployment = None # rebind-ok: pre-init so the except block can stamp a failure with no deployment picked
try:
verbose_router_logger.debug("Inside _image_generation()- model: %s", model)
deployment: Final = self.get_available_deployment(
deployment = self.get_available_deployment( # rebind-ok: set on success, see pre-init
model=model,
messages=[{"role": "user", "content": "prompt"}],
specific_deployment=kwargs.pop("specific_deployment", None),
@ -6127,6 +6139,9 @@ class PythonCore(DeprecatedRouterMembers):
if model_name is not None:
self.fail_calls[model_name] += 1
self._stamp_retry_skip_deployment_id(e, kwargs)
self._stamp_failed_attempt(
e, _REQUEST_KWARGS_ADAPTER.validate_python(kwargs), picked=deployment is not None
)
raise e
async def aimage_generation(self, prompt: str, model: str, **kwargs):
@ -6152,11 +6167,12 @@ class PythonCore(DeprecatedRouterMembers):
async def _aimage_generation(self, prompt: str, model: str, **kwargs):
model_name = model
start_time: Final = datetime.now() # noqa: DTZ005 # SDK logging uses naive local timestamps
deployment = None # rebind-ok: pre-init so the except block can stamp a failure with no deployment picked
try:
verbose_router_logger.debug("Inside _image_generation()- model: %s", model)
parent_otel_span: Final = get_parent_otel_span_from_kwargs(kwargs)
try:
deployment: Final = await self._core_async_get_available_deployment(
deployment = await self._core_async_get_available_deployment( # rebind-ok: set on success, see pre-init
model=model,
messages=[{"role": "user", "content": "prompt"}],
specific_deployment=kwargs.pop("specific_deployment", None),
@ -6217,6 +6233,9 @@ class PythonCore(DeprecatedRouterMembers):
if model_name is not None:
self.fail_calls[model_name] += 1
self._stamp_retry_skip_deployment_id(e, kwargs)
self._stamp_failed_attempt(
e, _REQUEST_KWARGS_ADAPTER.validate_python(kwargs), picked=deployment is not None
)
raise e
async def atranscription(self, file: FileTypes, model: str, **kwargs):
@ -6263,10 +6282,11 @@ class PythonCore(DeprecatedRouterMembers):
async def _atranscription(self, file: FileTypes, model: str, **kwargs):
model_name: Final = model
deployment = None # rebind-ok: pre-init so the except block can stamp a failure with no deployment picked
try:
verbose_router_logger.debug("Inside _atranscription()- model: %s", model)
parent_otel_span: Final = get_parent_otel_span_from_kwargs(kwargs)
deployment: Final = await self._core_async_get_available_deployment(
deployment = await self._core_async_get_available_deployment( # rebind-ok: set on success, see pre-init
model=model,
messages=[{"role": "user", "content": "prompt"}],
specific_deployment=kwargs.pop("specific_deployment", None),
@ -6300,6 +6320,9 @@ class PythonCore(DeprecatedRouterMembers):
if model_name is not None:
self.fail_calls[model_name] += 1
self._stamp_retry_skip_deployment_id(e, kwargs)
self._stamp_failed_attempt(
e, _REQUEST_KWARGS_ADAPTER.validate_python(kwargs), picked=deployment is not None
)
raise e
async def aspeech(self, model: str, input: str, voice: str | None = None, **kwargs):
@ -6356,10 +6379,11 @@ class PythonCore(DeprecatedRouterMembers):
async def _aspeech(self, model: str, input: str, voice: str | None = None, **kwargs):
model_name: Final = model
deployment = None # rebind-ok: pre-init so the except block can stamp a failure with no deployment picked
try:
verbose_router_logger.debug("Inside _aspeech()- model: %s", model)
parent_otel_span: Final = get_parent_otel_span_from_kwargs(kwargs)
deployment: Final = await self._core_async_get_available_deployment(
deployment = await self._core_async_get_available_deployment( # rebind-ok: set on success, see pre-init
model=model,
messages=[{"role": "user", "content": "prompt"}],
specific_deployment=kwargs.pop("specific_deployment", None),
@ -6393,6 +6417,9 @@ class PythonCore(DeprecatedRouterMembers):
if model_name is not None:
self.fail_calls[model_name] += 1
self._stamp_retry_skip_deployment_id(e, kwargs)
self._stamp_failed_attempt(
e, _REQUEST_KWARGS_ADAPTER.validate_python(kwargs), picked=deployment is not None
)
raise e
async def arerank(self, model: str, **kwargs):
@ -6418,9 +6445,10 @@ class PythonCore(DeprecatedRouterMembers):
async def _arerank(self, model: str, **kwargs):
model_name = None
deployment = None # rebind-ok: pre-init so the except block can stamp a failure with no deployment picked
try:
verbose_router_logger.debug("Inside _rerank()- model: %s", model)
deployment: Final = await self._core_async_get_available_deployment(
deployment = await self._core_async_get_available_deployment( # rebind-ok: set on success, see pre-init
model=model,
specific_deployment=kwargs.pop("specific_deployment", None),
request_kwargs=kwargs,
@ -6452,6 +6480,9 @@ class PythonCore(DeprecatedRouterMembers):
if model_name is not None:
self.fail_calls[model_name] += 1
self._stamp_retry_skip_deployment_id(e, kwargs)
self._stamp_failed_attempt(
e, _REQUEST_KWARGS_ADAPTER.validate_python(kwargs), picked=deployment is not None
)
raise e
def text_completion(
@ -6533,10 +6564,11 @@ class PythonCore(DeprecatedRouterMembers):
raise e
async def _atext_completion(self, model: str, prompt: str, **kwargs):
deployment = None # rebind-ok: pre-init so the except block can stamp a failure with no deployment picked
try:
verbose_router_logger.debug("Inside _atext_completion()- model: %s", model)
parent_otel_span: Final = get_parent_otel_span_from_kwargs(kwargs)
deployment: Final = await self._core_async_get_available_deployment(
deployment = await self._core_async_get_available_deployment( # rebind-ok: set on success, see pre-init
model=model,
messages=[{"role": "user", "content": prompt}],
specific_deployment=kwargs.pop("specific_deployment", None),
@ -6572,6 +6604,9 @@ class PythonCore(DeprecatedRouterMembers):
if model is not None:
self.fail_calls[model] += 1
self._stamp_retry_skip_deployment_id(e, kwargs)
self._stamp_failed_attempt(
e, _REQUEST_KWARGS_ADAPTER.validate_python(kwargs), picked=deployment is not None
)
raise e
async def aadapter_completion(
@ -6603,10 +6638,11 @@ class PythonCore(DeprecatedRouterMembers):
raise e
async def _aadapter_completion(self, adapter_id: str, model: str, **kwargs):
deployment = None # rebind-ok: pre-init so the except block can stamp a failure with no deployment picked
try:
verbose_router_logger.debug("Inside _aadapter_completion()- model: %s", model)
parent_otel_span: Final = get_parent_otel_span_from_kwargs(kwargs)
deployment: Final = await self._core_async_get_available_deployment(
deployment = await self._core_async_get_available_deployment( # rebind-ok: set on success, see pre-init
model=model,
messages=[{"role": "user", "content": "default text"}],
specific_deployment=kwargs.pop("specific_deployment", None),
@ -6642,6 +6678,9 @@ class PythonCore(DeprecatedRouterMembers):
if model is not None:
self.fail_calls[model] += 1
self._stamp_retry_skip_deployment_id(e, kwargs)
self._stamp_failed_attempt(
e, _REQUEST_KWARGS_ADAPTER.validate_python(kwargs), picked=deployment is not None
)
raise e
async def _asearch_with_fallbacks(self, original_function: Callable, **kwargs) -> object:
@ -7630,6 +7669,7 @@ class PythonCore(DeprecatedRouterMembers):
"""
handler_name: Final = original_function.__name__
metadata_variable_name: Final = get_router_metadata_variable_name(function_name="generic_api_call")
deployment = None # rebind-ok: pre-init so the except block can stamp a failure with no deployment picked
try:
verbose_router_logger.debug("Inside _generic_api_call() - handler: %s, model: %s", handler_name, model)
self._update_kwargs_before_fallbacks(
@ -7638,7 +7678,7 @@ class PythonCore(DeprecatedRouterMembers):
metadata_variable_name=metadata_variable_name,
)
with self._video_call_pin(handler_name, kwargs), self._route_capability_pin(handler_name):
deployment: Final = self.get_available_deployment(
deployment = self.get_available_deployment( # rebind-ok: set on success, see pre-init
model=model,
messages=kwargs.get("messages", None),
input=kwargs.get("input", None),
@ -7684,6 +7724,9 @@ class PythonCore(DeprecatedRouterMembers):
verbose_router_logger.info("%s(model=%s)\x1b[31m Exception %s\x1b[0m", handler_name, model, e)
if model is not None:
self.fail_calls[model] += 1
self._stamp_failed_attempt(
e, _REQUEST_KWARGS_ADAPTER.validate_python(kwargs), picked=deployment is not None
)
raise e
def embedding(
@ -7705,10 +7748,11 @@ class PythonCore(DeprecatedRouterMembers):
def _embedding(self, input: str | list[object], model: str, **kwargs):
model_name = None
deployment = None # rebind-ok: pre-init so the except block can stamp a failure with no deployment picked
try:
verbose_router_logger.debug("Inside embedding()- model: %s", model)
with embedding_deployment_selection():
deployment: Final = self.get_available_deployment(
deployment = self.get_available_deployment( # rebind-ok: set on success, see pre-init
model=model,
input=input,
specific_deployment=kwargs.pop("specific_deployment", None),
@ -7765,6 +7809,9 @@ class PythonCore(DeprecatedRouterMembers):
if model_name is not None:
self.fail_calls[model_name] += 1
self._stamp_retry_skip_deployment_id(e, kwargs)
self._stamp_failed_attempt(
e, _REQUEST_KWARGS_ADAPTER.validate_python(kwargs), picked=deployment is not None
)
raise e
async def aembedding(
@ -7794,11 +7841,12 @@ class PythonCore(DeprecatedRouterMembers):
async def _aembedding(self, input: str | list[object], model: str, **kwargs):
model_name = None
deployment = None # rebind-ok: pre-init so the except block can stamp a failure with no deployment picked
try:
verbose_router_logger.debug("Inside _aembedding()- model: %s", model)
parent_otel_span: Final = get_parent_otel_span_from_kwargs(kwargs)
with embedding_deployment_selection():
deployment: Final = await self._core_async_get_available_deployment(
deployment = await self._core_async_get_available_deployment( # rebind-ok: set on success, see pre-init
model=model,
input=input,
specific_deployment=kwargs.pop("specific_deployment", None),
@ -7845,6 +7893,9 @@ class PythonCore(DeprecatedRouterMembers):
if model_name is not None:
self.fail_calls[model_name] += 1
self._stamp_retry_skip_deployment_id(e, kwargs)
self._stamp_failed_attempt(
e, _REQUEST_KWARGS_ADAPTER.validate_python(kwargs), picked=deployment is not None
)
raise e
#### FILES API ####
@ -9083,8 +9134,9 @@ class PythonCore(DeprecatedRouterMembers):
kwargs[metadata_variable_name] = meta
if not isinstance(meta, dict):
return None
configured_failed_id: Final[str | None] = getattr(exception, "failed_configured_deployment_id", None)
prev_excluded: Final = set(meta.get("_failover_excluded_ids") or [])
excluded: Final = prev_excluded | {failed_id}
excluded: Final = prev_excluded | {configured_failed_id or failed_id}
all_ids: Final = {
(d.get("model_info") or {}).get("id")

141
tests/unit/test_router_weighted_failover.py
View file

@ -1019,3 +1019,144 @@ async def test_failover_reaches_healthy_sibling_when_affinity_pins_failed_deploy
assert while_a_is_down._hidden_params["model_id"] == "B"
assert while_a_is_down._hidden_params["additional_headers"]["x-litellm-attempted-fallbacks"] == 1
assert after_a_recovers._hidden_params["model_id"] == "A"
def _two_deployment_router(model: str) -> Router:
return Router(
model_list=[
{
"model_name": "test-model",
"litellm_params": {"model": model, "api_key": "bad", "weight": 1},
"model_info": {"id": "A"},
},
{
"model_name": "test-model",
"litellm_params": {"model": model, "api_key": "good", "weight": 0},
"model_info": {"id": "B"},
},
],
routing_strategy="simple-shuffle",
num_retries=0,
disable_cooldowns=True,
enable_weighted_failover=True,
)
_ENDPOINTS: Final = [
("aembedding", "openai/text-embedding-3-small", {"input": ["hi"]}),
("aimage_generation", "openai/dall-e-3", {"prompt": "a cat"}),
("atranscription", "openai/whisper-1", {"file": b"audio"}),
("aspeech", "openai/tts-1", {"input": "hi", "voice": "alloy"}),
("atext_completion", "openai/gpt-3.5-turbo-instruct", {"prompt": "hi"}),
("arerank", "cohere/rerank-english-v3.0", {"query": "q", "documents": ["d"]}),
]
@pytest.mark.asyncio
@pytest.mark.parametrize(("endpoint", "model", "arguments"), _ENDPOINTS, ids=[e[0] for e in _ENDPOINTS])
async def test_weighted_failover_moves_every_endpoint_to_another_deployment(
monkeypatch: pytest.MonkeyPatch, endpoint: str, model: str, arguments: dict[str, object]
) -> None:
keys_called: list[object] = []
answer: Final = object()
async def provider(**kwargs: object) -> object:
keys_called.append(kwargs.get("api_key"))
if kwargs.get("api_key") == "bad":
raise litellm.InternalServerError(message="A is down", llm_provider="openai", model=model)
return answer
monkeypatch.setattr(litellm, endpoint, provider)
router: Final = _two_deployment_router(model)
response: Final = await getattr(router, endpoint)(model="test-model", **arguments)
assert response is answer
assert keys_called == ["bad", "good"]
@pytest.mark.asyncio
async def test_weighted_failover_skips_the_failed_deployment_for_clientside_credentials(
monkeypatch: pytest.MonkeyPatch,
) -> None:
api_bases_called: list[object] = []
answer: Final = object()
async def provider(**kwargs: object) -> object:
api_bases_called.append(kwargs.get("api_base"))
if kwargs.get("api_base") == "http://a":
raise litellm.InternalServerError(message="A is down", llm_provider="openai", model="m")
return answer
monkeypatch.setattr(litellm, "aembedding", provider)
router: Final = Router(
model_list=[
{
"model_name": "test-model",
"litellm_params": {"model": "openai/text-embedding-3-small", "api_base": "http://a", "weight": 1},
"model_info": {"id": "A"},
},
{
"model_name": "test-model",
"litellm_params": {"model": "openai/text-embedding-3-small", "api_base": "http://b", "weight": 0},
"model_info": {"id": "B"},
},
],
routing_strategy="simple-shuffle",
num_retries=0,
disable_cooldowns=True,
enable_weighted_failover=True,
)
response: Final = await router.aembedding(model="test-model", input=["hi"], api_key="tenant-key")
assert response is answer
assert api_bases_called == ["http://a", "http://b"]
@pytest.mark.parametrize(
("endpoint", "attempt", "arguments"),
[
("embedding", "_embedding", {"input": ["hi"]}),
("image_generation", "_image_generation", {"prompt": "a cat"}),
],
)
def test_a_sync_attempt_stamps_the_failed_deployment(
monkeypatch: pytest.MonkeyPatch, endpoint: str, attempt: str, arguments: dict[str, object]
) -> None:
def provider(**kwargs: object) -> object:
raise litellm.InternalServerError(message="down", llm_provider="openai", model="m")
monkeypatch.setattr(litellm, endpoint, provider)
router: Final = Router(
model_list=[
{
"model_name": "test-model",
"litellm_params": {"model": "openai/m", "api_key": "k"},
"model_info": {"id": "A"},
}
],
)
with pytest.raises(litellm.InternalServerError) as raised:
getattr(router, attempt)(model="test-model", **arguments)
assert getattr(raised.value, "failed_deployment_id", None) == "A"
@pytest.mark.asyncio
async def test_a_failed_pick_stamps_no_deployment() -> None:
router: Final = Router(
model_list=[
{
"model_name": "test-model",
"litellm_params": {"model": "openai/text-embedding-3-small", "api_key": "k"},
"model_info": {"id": "A"},
}
],
)
with pytest.raises(litellm.BadRequestError, match="no healthy deployments") as raised:
await router._aembedding(input=["hi"], model="no-such-model")
assert getattr(raised.value, "failed_deployment_id", None) is None