mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-12 08:42:18 +00:00
Merge 82449f5e7c into fca3d4f513
This commit is contained in:
commit
b13f872206
7 changed files with 289 additions and 36 deletions
|
|
@ -50,6 +50,7 @@ pub(crate) fn should_retry(
|
|||
has_context_window_fallbacks: context_window,
|
||||
has_content_policy_fallbacks: content_policy,
|
||||
has_regular_fallbacks: regular,
|
||||
has_untried_deployment: false,
|
||||
},
|
||||
))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1013,7 +1013,26 @@ impl Core {
|
|||
let Some(failure) = retried_failure(&failed.error) else {
|
||||
return Err(failed);
|
||||
};
|
||||
if !policy_applies && !self.should_retry(snapshot, request, &failure, healthy, all) {
|
||||
let named = failure.failed_deployment_id.as_ref();
|
||||
if let (Some(status), Some(id)) =
|
||||
(failure.status, named.or(failed.deployment_id.as_ref()))
|
||||
&& !should_retry_status(status)
|
||||
&& !inputs.skipped_ids.contains(id)
|
||||
{
|
||||
inputs.skipped_ids.push(id.clone());
|
||||
}
|
||||
let has_untried_deployment = healthy_ids
|
||||
.iter()
|
||||
.any(|id| !inputs.skipped_ids.contains(id));
|
||||
if !policy_applies
|
||||
&& !self.should_retry(
|
||||
snapshot,
|
||||
request,
|
||||
&failure,
|
||||
(healthy, all),
|
||||
has_untried_deployment,
|
||||
)
|
||||
{
|
||||
return Err(failed);
|
||||
}
|
||||
let uncounted = request.uncounted_retries.filter(|_| !policy_applies);
|
||||
|
|
@ -1042,14 +1061,6 @@ impl Core {
|
|||
if retry == 0 {
|
||||
self.log_retry(target, at, 0, &failed.error);
|
||||
}
|
||||
let named = failure.failed_deployment_id.as_ref();
|
||||
if let (Some(status), Some(id)) =
|
||||
(failure.status, named.or(failed.deployment_id.as_ref()))
|
||||
&& !should_retry_status(status)
|
||||
&& !inputs.skipped_ids.contains(id)
|
||||
{
|
||||
inputs.skipped_ids.push(id.clone());
|
||||
}
|
||||
// An endpoint a deployment does not serve is retried on the others, until every
|
||||
// one refused (`_every_deployment_refused`).
|
||||
if failure.is_a("UnsupportedEndpointError")
|
||||
|
|
@ -1097,7 +1108,11 @@ impl Core {
|
|||
response.max_retries = Some(num_retries);
|
||||
return Ok(response);
|
||||
}
|
||||
Err(next) if next.error.attempt().is_some_and(|f| f.agentic_followup) => {
|
||||
Err(next)
|
||||
if next.error.attempt().is_some_and(|f| {
|
||||
f.agentic_followup || f.kind == FailureKind::GuardrailIntervention
|
||||
}) =>
|
||||
{
|
||||
return Err(next);
|
||||
}
|
||||
Err(next) => {
|
||||
|
|
@ -1171,8 +1186,8 @@ impl Core {
|
|||
snapshot: &RouterStateSnapshot,
|
||||
request: &RouterRequest,
|
||||
failure: &AttemptError,
|
||||
healthy: usize,
|
||||
all: usize,
|
||||
(healthy, all): (usize, usize),
|
||||
has_untried_deployment: bool,
|
||||
) -> bool {
|
||||
let lists = Lists::of(snapshot, request);
|
||||
should_retry(
|
||||
|
|
@ -1183,6 +1198,7 @@ impl Core {
|
|||
has_context_window_fallbacks: lists.context_window.is_some(),
|
||||
has_content_policy_fallbacks: lists.content_policy.is_some(),
|
||||
has_regular_fallbacks: lists.has_regular(),
|
||||
has_untried_deployment,
|
||||
},
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -250,7 +250,7 @@ impl Router {
|
|||
if !policy_applies
|
||||
&& !self
|
||||
.core
|
||||
.should_retry(&snapshot, &request, &last, healthy, all)
|
||||
.should_retry(&snapshot, &request, &last, (healthy, all), false)
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -70,6 +70,8 @@ pub struct RetryContext {
|
|||
pub has_content_policy_fallbacks: bool,
|
||||
/// `regular_fallbacks is not None and len(regular_fallbacks) > 0`.
|
||||
pub has_regular_fallbacks: bool,
|
||||
/// A healthy deployment of the group has not failed this call yet.
|
||||
pub has_untried_deployment: bool,
|
||||
}
|
||||
|
||||
/// `should_retry_this_error`, as a predicate: `false` where Python raises the error.
|
||||
|
|
@ -80,15 +82,13 @@ pub fn should_retry(failure: &AttemptError, ctx: RetryContext) -> bool {
|
|||
if failure.is_a("ContentPolicyViolationError") && ctx.has_content_policy_fallbacks {
|
||||
return false;
|
||||
}
|
||||
if let Some(status) = failure.status
|
||||
&& !should_retry_status(status)
|
||||
&& !matches!(status, 401 | 403)
|
||||
&& !failure.is_a("UnsupportedEndpointError")
|
||||
let non_retryable_status = failure
|
||||
.status
|
||||
.is_some_and(|status| !should_retry_status(status) && !matches!(status, 401 | 403));
|
||||
if (non_retryable_status && !failure.is_a("UnsupportedEndpointError"))
|
||||
|| failure.is_a("NotFoundError")
|
||||
{
|
||||
return false;
|
||||
}
|
||||
if failure.is_a("NotFoundError") {
|
||||
return false;
|
||||
return ctx.has_untried_deployment;
|
||||
}
|
||||
if failure.is_a("RateLimitError") && ctx.healthy == 0 && ctx.has_regular_fallbacks {
|
||||
return false;
|
||||
|
|
@ -216,4 +216,25 @@ mod tests {
|
|||
expected
|
||||
);
|
||||
}
|
||||
|
||||
#[rstest]
|
||||
#[case::not_found(FailureKind::NotFound, Some(404))]
|
||||
#[case::bad_request(FailureKind::BadRequest, Some(400))]
|
||||
#[case::unprocessable(FailureKind::Other, Some(422))]
|
||||
fn a_non_retryable_failure_is_retried_only_while_a_deployment_is_untried(
|
||||
#[case] kind: FailureKind,
|
||||
#[case] status: Option<u16>,
|
||||
#[values(true, false)] has_untried_deployment: bool,
|
||||
) {
|
||||
let ctx = RetryContext {
|
||||
healthy: 2,
|
||||
all: 2,
|
||||
has_untried_deployment,
|
||||
..RetryContext::default()
|
||||
};
|
||||
assert_eq!(
|
||||
should_retry(&AttemptError::new(kind, status), ctx),
|
||||
has_untried_deployment
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -225,7 +225,7 @@ async fn a_rate_limit_retries_in_the_same_group(config: RouterConfig) {
|
|||
|
||||
#[rstest]
|
||||
#[tokio::test]
|
||||
async fn a_bad_request_falls_back_without_retrying(config: RouterConfig) {
|
||||
async fn a_bad_request_is_retried_on_the_groups_other_deployment(config: RouterConfig) {
|
||||
let host = Arc::new(ScriptedHost::failing(
|
||||
"primary",
|
||||
vec![AttemptError::new(FailureKind::BadRequest, Some(400))],
|
||||
|
|
@ -239,18 +239,79 @@ async fn a_bad_request_falls_back_without_retrying(config: RouterConfig) {
|
|||
.call(Op::from("acompletion"), request())
|
||||
.await
|
||||
.unwrap();
|
||||
let retried = host
|
||||
.events
|
||||
.lock()
|
||||
.iter()
|
||||
.any(|event| matches!(event, RouterEvent::Retry { .. }));
|
||||
let mut selected = host.selected();
|
||||
selected.sort();
|
||||
assert_eq!(
|
||||
(
|
||||
response.model_group.as_str(),
|
||||
response.attempted_fallbacks,
|
||||
selected
|
||||
),
|
||||
("primary", 0, vec!["a1".to_owned(), "a2".to_owned()])
|
||||
);
|
||||
}
|
||||
|
||||
#[rstest]
|
||||
#[tokio::test]
|
||||
async fn a_bad_request_from_every_deployment_falls_back(config: RouterConfig) {
|
||||
let bad_request = AttemptError::new(FailureKind::BadRequest, Some(400));
|
||||
let host = Arc::new(ScriptedHost::failing(
|
||||
"primary",
|
||||
vec![bad_request.clone(), bad_request],
|
||||
));
|
||||
let router = router(
|
||||
config,
|
||||
json!({"num_retries": 3, "fallbacks": [{"primary": ["backup"]}]}),
|
||||
host.clone(),
|
||||
);
|
||||
let response = router
|
||||
.call(Op::from("acompletion"), request())
|
||||
.await
|
||||
.unwrap();
|
||||
let mut primary_selected: Vec<String> = host
|
||||
.selected()
|
||||
.into_iter()
|
||||
.filter(|id| id.starts_with('a'))
|
||||
.collect();
|
||||
primary_selected.sort();
|
||||
assert_eq!(
|
||||
(
|
||||
response.deployment_id.as_str(),
|
||||
response.attempted_fallbacks,
|
||||
retried
|
||||
primary_selected
|
||||
),
|
||||
("b1", 1, false)
|
||||
("b1", 1, vec!["a1".to_owned(), "a2".to_owned()])
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_guardrail_rejection_on_a_retry_is_not_failed_over_to_a_sibling() {
|
||||
let config = serde_json::from_value(json!({
|
||||
"model_list": [
|
||||
{"model_name": "primary", "litellm_params": {"model": "openai/a"}, "model_info": {"id": "a1"}},
|
||||
{"model_name": "primary", "litellm_params": {"model": "openai/b"}, "model_info": {"id": "a2"}},
|
||||
{"model_name": "primary", "litellm_params": {"model": "openai/c"}, "model_info": {"id": "a3"}},
|
||||
]
|
||||
}))
|
||||
.unwrap();
|
||||
let host = Arc::new(ScriptedHost::failing(
|
||||
"primary",
|
||||
vec![
|
||||
AttemptError::new(FailureKind::NotFound, Some(404)),
|
||||
AttemptError::new(FailureKind::GuardrailIntervention, Some(400)),
|
||||
],
|
||||
));
|
||||
let router = router(config, json!({"num_retries": 2}), host.clone());
|
||||
let error = router
|
||||
.call(Op::from("acompletion"), request())
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert_eq!(
|
||||
(
|
||||
error.attempt().map(|failure| failure.kind),
|
||||
host.selected().len()
|
||||
),
|
||||
(Some(FailureKind::GuardrailIntervention), 2)
|
||||
);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -179,13 +179,10 @@ async def test_lone_deployment_retries_only_retryable_errors(
|
|||
assert len(host.calls_to("acompletion")) == expected_calls
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("builder", "expected_calls"),
|
||||
[(bad_request, 1), (context_window, 1), (content_policy, 1), (server_error, 3), (timeout, 3)],
|
||||
)
|
||||
@pytest.mark.parametrize("builder", [server_error, timeout])
|
||||
@pytest.mark.asyncio
|
||||
async def test_two_deployment_group_retries_only_retryable_errors(
|
||||
make_core: MakeCore, host: ScriptedHost, builder: ErrorBuilder, expected_calls: int
|
||||
async def test_two_deployment_group_retries_retryable_errors_up_to_the_budget(
|
||||
make_core: MakeCore, host: ScriptedHost, builder: ErrorBuilder
|
||||
) -> None:
|
||||
core: Final = make_core(model_list=two_deployments(), num_retries=2)
|
||||
host.answer("*", fresh(builder))
|
||||
|
|
@ -193,7 +190,21 @@ async def test_two_deployment_group_retries_only_retryable_errors(
|
|||
error: Final = await attempt(core)
|
||||
|
||||
assert type(error) is type(builder())
|
||||
assert len(host.calls_to("acompletion")) == expected_calls
|
||||
assert len(host.calls_to("acompletion")) == 3
|
||||
|
||||
|
||||
@pytest.mark.parametrize("builder", [bad_request, context_window, content_policy])
|
||||
@pytest.mark.asyncio
|
||||
async def test_non_retryable_error_tries_each_deployment_of_the_group_once(
|
||||
make_core: MakeCore, host: ScriptedHost, builder: ErrorBuilder
|
||||
) -> None:
|
||||
core: Final = make_core(model_list=two_deployments(), num_retries=2)
|
||||
host.answer("*", fresh(builder))
|
||||
|
||||
error: Final = await attempt(core)
|
||||
|
||||
assert type(error) is type(builder())
|
||||
assert sorted(host.deployments_called("acompletion")) == ["a", "b"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
|
|||
|
|
@ -11460,6 +11460,149 @@ def _retry_skip_deployment(deployment_id, host, litellm_params=None, model_info=
|
|||
}
|
||||
|
||||
|
||||
_UPSTREAM_404 = {
|
||||
"message": "The model `gpt-5.6` does not exist or you do not have access to it.",
|
||||
"type": "invalid_request_error",
|
||||
"code": "model_not_found",
|
||||
}
|
||||
|
||||
_UPSTREAM_SUCCESS = {
|
||||
"id": "chatcmpl-failover",
|
||||
"object": "chat.completion",
|
||||
"created": 1,
|
||||
"model": "gpt-5.6",
|
||||
"choices": [{"index": 0, "message": {"role": "assistant", "content": "hi back"}, "finish_reason": "stop"}],
|
||||
"usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3},
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("status_code,upstream_error", [(404, _UPSTREAM_404), (400, _UPSTREAM_400)])
|
||||
async def test_router_non_retryable_error_fails_over_to_sibling_deployment(
|
||||
monkeypatch: pytest.MonkeyPatch, status_code, upstream_error
|
||||
):
|
||||
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
|
||||
router = litellm.Router(
|
||||
model_list=[
|
||||
_retry_skip_deployment("rejecting", "rejecting", litellm_params={"weight": 1}),
|
||||
_retry_skip_deployment("accepting", "accepting", litellm_params={"weight": 0}),
|
||||
],
|
||||
num_retries=2,
|
||||
)
|
||||
|
||||
with respx.mock(assert_all_called=True) as respx_mock:
|
||||
rejecting = respx_mock.post("https://rejecting.local/v1/chat/completions").mock(
|
||||
return_value=httpx.Response(status_code, json={"error": upstream_error})
|
||||
)
|
||||
accepting = respx_mock.post("https://accepting.local/v1/chat/completions").mock(
|
||||
return_value=httpx.Response(200, json=_UPSTREAM_SUCCESS)
|
||||
)
|
||||
response = await router.acompletion(model="gpt-5.6", messages=[{"role": "user", "content": "hi"}])
|
||||
|
||||
assert (rejecting.call_count, accepting.call_count) == (1, 1)
|
||||
assert response.choices[0].message.content == "hi back"
|
||||
assert response._hidden_params["model_id"] == "accepting"
|
||||
assert response._hidden_params["additional_headers"]["x-litellm-attempted-retries"] == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_router_not_found_failover_still_cools_down_the_refusing_deployment(monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
|
||||
router = litellm.Router(
|
||||
model_list=[
|
||||
_retry_skip_deployment("rejecting", "rejecting", litellm_params={"weight": 1}),
|
||||
_retry_skip_deployment("accepting", "accepting", litellm_params={"weight": 0}),
|
||||
],
|
||||
num_retries=2,
|
||||
)
|
||||
|
||||
with respx.mock(assert_all_called=True) as respx_mock:
|
||||
respx_mock.post("https://rejecting.local/v1/chat/completions").mock(
|
||||
return_value=httpx.Response(404, json={"error": _UPSTREAM_404})
|
||||
)
|
||||
respx_mock.post("https://accepting.local/v1/chat/completions").mock(
|
||||
return_value=httpx.Response(200, json=_UPSTREAM_SUCCESS)
|
||||
)
|
||||
await router.acompletion(model="gpt-5.6", messages=[{"role": "user", "content": "hi"}])
|
||||
|
||||
assert await async_get_cooldown_deployments(ctx=router._core_ctx, parent_otel_span=None) == ["rejecting"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"status_code,upstream_error,expected_exception",
|
||||
[(404, _UPSTREAM_404, litellm.NotFoundError), (400, _UPSTREAM_400, litellm.BadRequestError)],
|
||||
)
|
||||
async def test_router_non_retryable_error_on_single_deployment_group_is_not_retried(
|
||||
monkeypatch: pytest.MonkeyPatch, status_code, upstream_error, expected_exception
|
||||
):
|
||||
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
|
||||
router = litellm.Router(model_list=[_retry_skip_deployment("only", "only")], num_retries=2)
|
||||
|
||||
with respx.mock(assert_all_called=True) as respx_mock:
|
||||
only = respx_mock.post("https://only.local/v1/chat/completions").mock(
|
||||
return_value=httpx.Response(status_code, json={"error": upstream_error})
|
||||
)
|
||||
with pytest.raises(expected_exception):
|
||||
await router.acompletion(model="gpt-5.6", messages=[{"role": "user", "content": "hi"}])
|
||||
|
||||
assert only.call_count == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_router_non_retryable_error_fails_over_past_every_refusing_deployment(monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
|
||||
router = litellm.Router(
|
||||
model_list=[
|
||||
_retry_skip_deployment("first", "first", litellm_params={"weight": 1}),
|
||||
_retry_skip_deployment("second", "second", litellm_params={"weight": 1}),
|
||||
_retry_skip_deployment("accepting", "accepting", litellm_params={"weight": 0}),
|
||||
],
|
||||
num_retries=2,
|
||||
)
|
||||
|
||||
with respx.mock(assert_all_called=True) as respx_mock:
|
||||
refusing = tuple(
|
||||
respx_mock.post(f"https://{host}.local/v1/chat/completions").mock(
|
||||
return_value=httpx.Response(404, json={"error": _UPSTREAM_404})
|
||||
)
|
||||
for host in ("first", "second")
|
||||
)
|
||||
accepting = respx_mock.post("https://accepting.local/v1/chat/completions").mock(
|
||||
return_value=httpx.Response(200, json=_UPSTREAM_SUCCESS)
|
||||
)
|
||||
response = await router.acompletion(model="gpt-5.6", messages=[{"role": "user", "content": "hi"}])
|
||||
|
||||
assert tuple(route.call_count for route in (*refusing, accepting)) == (1, 1, 1)
|
||||
assert response._hidden_params["model_id"] == "accepting"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_router_non_retryable_error_stops_once_every_deployment_refused(monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
|
||||
router = litellm.Router(
|
||||
model_list=[
|
||||
_retry_skip_deployment("first", "first", litellm_params={"weight": 1}),
|
||||
_retry_skip_deployment("second", "second", litellm_params={"weight": 0}),
|
||||
],
|
||||
num_retries=3,
|
||||
disable_cooldowns=True,
|
||||
)
|
||||
|
||||
with respx.mock(assert_all_called=True) as respx_mock:
|
||||
first = respx_mock.post("https://first.local/v1/chat/completions").mock(
|
||||
return_value=httpx.Response(400, json={"error": _UPSTREAM_400})
|
||||
)
|
||||
second = respx_mock.post("https://second.local/v1/chat/completions").mock(
|
||||
return_value=httpx.Response(400, json={"error": _UPSTREAM_400})
|
||||
)
|
||||
with pytest.raises(litellm.BadRequestError) as exc_info:
|
||||
await router.acompletion(model="gpt-5.6", messages=[{"role": "user", "content": "hi"}])
|
||||
|
||||
assert (first.call_count, second.call_count) == (1, 1)
|
||||
assert exc_info.value.failed_deployment_id == "second"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value,expected",
|
||||
[
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue