mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-01 07:03:02 +00:00
fix(router): free the least-busy pick of a sync generic call that fails before its provider
This commit is contained in:
mateo-berri
parent
c60a56a1c7
commit
c4e8731915
2 changed files with 63 additions and 50 deletions
|
|
@ -7392,65 +7392,66 @@ class Router(DeprecatedRouterMembers):
|
|||
"""
|
||||
handler_name: Final = original_function.__name__
|
||||
metadata_variable_name: Final = get_router_metadata_variable_name(function_name="generic_api_call")
|
||||
try:
|
||||
verbose_router_logger.debug("Inside _generic_api_call() - handler: %s, model: %s", handler_name, model)
|
||||
self._update_kwargs_before_fallbacks(
|
||||
model=model,
|
||||
kwargs=kwargs,
|
||||
metadata_variable_name=metadata_variable_name,
|
||||
)
|
||||
with (
|
||||
self._video_call_pin(handler_name, kwargs),
|
||||
self._route_capability_pin(handler_name),
|
||||
picks_reserve(True),
|
||||
):
|
||||
deployment: Final = self.get_available_deployment(
|
||||
with owning_picks(self._core_least_busy_reservations, call_id_of(CALL_IDENTITY.validate_python(kwargs))):
|
||||
try:
|
||||
verbose_router_logger.debug("Inside _generic_api_call() - handler: %s, model: %s", handler_name, model)
|
||||
self._update_kwargs_before_fallbacks(
|
||||
model=model,
|
||||
messages=kwargs.get("messages", None),
|
||||
input=kwargs.get("input", None),
|
||||
specific_deployment=kwargs.pop("specific_deployment", None),
|
||||
request_kwargs=kwargs,
|
||||
kwargs=kwargs,
|
||||
metadata_variable_name=metadata_variable_name,
|
||||
)
|
||||
with (
|
||||
self._video_call_pin(handler_name, kwargs),
|
||||
self._route_capability_pin(handler_name),
|
||||
picks_reserve(True),
|
||||
):
|
||||
deployment: Final = self.get_available_deployment(
|
||||
model=model,
|
||||
messages=kwargs.get("messages", None),
|
||||
input=kwargs.get("input", None),
|
||||
specific_deployment=kwargs.pop("specific_deployment", None),
|
||||
request_kwargs=kwargs,
|
||||
)
|
||||
self._update_kwargs_with_deployment(
|
||||
deployment=deployment,
|
||||
kwargs=kwargs,
|
||||
function_name="generic_api_call",
|
||||
splits_stream_timeout=handler_name in _STREAM_TIMEOUT_AWARE_GENERIC_HANDLERS,
|
||||
)
|
||||
self._update_kwargs_with_deployment(
|
||||
deployment=deployment,
|
||||
kwargs=kwargs,
|
||||
function_name="generic_api_call",
|
||||
splits_stream_timeout=handler_name in _STREAM_TIMEOUT_AWARE_GENERIC_HANDLERS,
|
||||
)
|
||||
|
||||
data: Final = deployment["litellm_params"].copy()
|
||||
model_name: Final = data["model"]
|
||||
data: Final = deployment["litellm_params"].copy()
|
||||
model_name: Final = data["model"]
|
||||
|
||||
self.total_calls[model_name] += 1
|
||||
self.total_calls[model_name] += 1
|
||||
|
||||
# For passthrough routes, use the actual model from deployment
|
||||
# and swap model name in endpoint if present
|
||||
if "endpoint" in kwargs and kwargs["endpoint"]:
|
||||
kwargs["endpoint"] = kwargs["endpoint"].replace(model, model_name)
|
||||
kwargs["model"] = model_name
|
||||
# For passthrough routes, use the actual model from deployment
|
||||
# and swap model name in endpoint if present
|
||||
if "endpoint" in kwargs and kwargs["endpoint"]:
|
||||
kwargs["endpoint"] = kwargs["endpoint"].replace(model, model_name)
|
||||
kwargs["model"] = model_name
|
||||
|
||||
# Perform pre-call checks for routing strategy
|
||||
self.routing_strategy_pre_call_checks(deployment=deployment)
|
||||
# Perform pre-call checks for routing strategy
|
||||
self.routing_strategy_pre_call_checks(deployment=deployment)
|
||||
|
||||
custom_llm_provider: Final = provider_for_generic_call(data)
|
||||
custom_llm_provider: Final = provider_for_generic_call(data)
|
||||
|
||||
response: Final = original_function(
|
||||
**{
|
||||
**data,
|
||||
"custom_llm_provider": custom_llm_provider,
|
||||
"caching": self._core_cache_responses,
|
||||
**kwargs,
|
||||
}
|
||||
)
|
||||
response: Final = original_function(
|
||||
**{
|
||||
**data,
|
||||
"custom_llm_provider": custom_llm_provider,
|
||||
"caching": self._core_cache_responses,
|
||||
**kwargs,
|
||||
}
|
||||
)
|
||||
|
||||
self.success_calls[model_name] += 1
|
||||
verbose_router_logger.info("%s(model=%s)\x1b[32m 200 OK\x1b[0m", handler_name, model_name)
|
||||
return response
|
||||
except Exception as e:
|
||||
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
|
||||
raise e
|
||||
self.success_calls[model_name] += 1
|
||||
verbose_router_logger.info("%s(model=%s)\x1b[32m 200 OK\x1b[0m", handler_name, model_name)
|
||||
return response
|
||||
except Exception as e:
|
||||
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
|
||||
raise e
|
||||
|
||||
def embedding(
|
||||
self,
|
||||
|
|
|
|||
|
|
@ -939,6 +939,18 @@ def test_a_sync_text_completion_does_not_keep_its_deployment_busy(
|
|||
assert _router_pick(router, TEXT_GROUP) == _called_deployment(route)
|
||||
|
||||
|
||||
|
||||
@pytest.mark.parametrize("call_identity", _CALL_IDENTITIES)
|
||||
def test_a_sync_responses_call_that_fails_before_reaching_its_provider_frees_its_pick(
|
||||
call_identity: dict[str, str],
|
||||
) -> None:
|
||||
router: Final = _least_busy_router()
|
||||
|
||||
with pytest.raises(litellm.InternalServerError):
|
||||
router.responses(model=ROUTER_GROUP, input="hi", api_base="not-a-url", **call_identity)
|
||||
|
||||
assert _router_pick(router) == ROUTER_DEPLOYMENT_IDS[0]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_batch_cancel_does_not_keep_its_deployment_busy(
|
||||
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue