mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-14 09:03:16 +00:00
fix: raise on failed OpenRouter non-streaming completions (#50880)
* fix: propagate OpenRouter errors and preserve failure usage * test: cover OpenRouter errors and failure accounting * fix: bill a recovered OpenRouter failure on the success log and parse the body once A retry or fallback that recovers from an OpenRouter finish_reason error now carries the failed attempt's usage and cost into the success log, so the key pays for both calls. The response body is parsed once and handed to the error check, and the failure-log test no longer waits on a 10-second deadline * fix: carry a failed attempt's usage before each SDK retry * test: cover OpenRouter's documented failure body without usage Pins the recovered-failure test's start time to a fixed date and adds a regression test for the documented non-streaming error body, which carries no usage --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
parent
cd2e4b6b60
commit
8ef451a187
3 changed files with 324 additions and 2 deletions
|
|
@ -31,6 +31,32 @@ if TYPE_CHECKING:
|
|||
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
|
||||
|
||||
|
||||
_RESPONSE_CHOICES: Final = TypeAdapter(tuple[Mapping[str, object], ...], config=ConfigDict(hide_input_in_errors=True))
|
||||
|
||||
|
||||
def _non_streaming_error(response: Mapping[str, object], headers: httpx.Headers) -> OpenRouterException | None:
|
||||
choices: Final = _RESPONSE_CHOICES.validate_python(response.get("choices") or ())
|
||||
choice: Final = next(
|
||||
(choice for choice in choices if choice.get("finish_reason") == "error" or choice.get("error")), None
|
||||
)
|
||||
if choice is None:
|
||||
return None
|
||||
error: Final = choice.get("error")
|
||||
details: Final[Mapping[str, object]] = _JSON_OBJECT.validate_python(error) if isinstance(error, Mapping) else {}
|
||||
code: Final = details.get("code")
|
||||
message: Final = details.get("message")
|
||||
return OpenRouterException(
|
||||
status_code=code if isinstance(code, int) and 400 <= code < 600 else 502,
|
||||
message=(
|
||||
message
|
||||
if isinstance(message, str) and message
|
||||
else "OpenRouter returned finish_reason 'error': the provider failed during generation"
|
||||
),
|
||||
headers=headers,
|
||||
body=dict(response),
|
||||
)
|
||||
|
||||
|
||||
class CacheControlSupportedModels(str, Enum):
|
||||
"""Models that support cache_control in content blocks."""
|
||||
|
||||
|
|
@ -210,8 +236,8 @@ class OpenrouterConfig(OpenAIGPTConfig):
|
|||
|
||||
# Extract cost from OpenRouter response body
|
||||
# OpenRouter returns cost information in the usage object when usage.include=true
|
||||
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
|
||||
try:
|
||||
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
|
||||
response_provider: Final = response_json.get("provider")
|
||||
if isinstance(response_provider, str) and response_provider:
|
||||
set_hidden_params(
|
||||
|
|
@ -243,6 +269,12 @@ class OpenrouterConfig(OpenAIGPTConfig):
|
|||
# If we can't extract cost, continue without it - don't fail the response
|
||||
pass
|
||||
|
||||
response_error: Final = _non_streaming_error(response_json, raw_response.headers)
|
||||
if response_error is not None:
|
||||
failed_attempt: Final = logging_obj.served_round(model_response)
|
||||
logging_obj.record_served_rounds_for_failure(() if failed_attempt is None else (failed_attempt,))
|
||||
raise response_error
|
||||
|
||||
return model_response
|
||||
|
||||
def get_error_class(self, error_message: str, status_code: int, headers: dict | httpx.Headers) -> BaseLLMException:
|
||||
|
|
|
|||
|
|
@ -6728,6 +6728,16 @@ def _is_not_resent_failure(error: BaseException) -> bool:
|
|||
return isinstance(error, Exception) and not is_resent_failure(error)
|
||||
|
||||
|
||||
class _RetryCallState(Protocol):
|
||||
kwargs: Mapping[str, object]
|
||||
|
||||
|
||||
def _carry_failed_attempt(retry_state: _RetryCallState) -> None:
|
||||
logging_obj: Final = retry_state.kwargs.get("litellm_logging_obj")
|
||||
if isinstance(logging_obj, LiteLLMLoggingObj):
|
||||
logging_obj.carry_failed_attempt_usage()
|
||||
|
||||
|
||||
def completion_with_retries(*args, **kwargs):
|
||||
"""
|
||||
Executes a litellm.completion() with 3 retries
|
||||
|
|
@ -6747,12 +6757,14 @@ def completion_with_retries(*args, **kwargs):
|
|||
retryer = tenacity.Retrying(
|
||||
wait=tenacity.wait_exponential(multiplier=1, max=10),
|
||||
stop=tenacity.stop_after_attempt(num_retries),
|
||||
before=_carry_failed_attempt,
|
||||
retry=tenacity.retry_if_exception(_is_not_resent_failure),
|
||||
reraise=True,
|
||||
)
|
||||
else:
|
||||
retryer = tenacity.Retrying(
|
||||
stop=tenacity.stop_after_attempt(num_retries),
|
||||
before=_carry_failed_attempt,
|
||||
retry=tenacity.retry_if_exception(_is_not_resent_failure),
|
||||
reraise=True,
|
||||
)
|
||||
|
|
@ -6778,12 +6790,14 @@ async def acompletion_with_retries(*args, **kwargs):
|
|||
retryer = tenacity.AsyncRetrying(
|
||||
wait=tenacity.wait_exponential(multiplier=1, max=10),
|
||||
stop=tenacity.stop_after_attempt(num_retries),
|
||||
before=_carry_failed_attempt,
|
||||
retry=tenacity.retry_if_exception(_is_not_resent_failure),
|
||||
reraise=True,
|
||||
)
|
||||
else:
|
||||
retryer = tenacity.AsyncRetrying(
|
||||
stop=tenacity.stop_after_attempt(num_retries),
|
||||
before=_carry_failed_attempt,
|
||||
retry=tenacity.retry_if_exception(_is_not_resent_failure),
|
||||
reraise=True,
|
||||
)
|
||||
|
|
@ -6811,12 +6825,14 @@ def responses_with_retries(*args, **kwargs):
|
|||
retryer = tenacity.Retrying(
|
||||
wait=tenacity.wait_exponential(multiplier=1, max=10),
|
||||
stop=tenacity.stop_after_attempt(num_retries),
|
||||
before=_carry_failed_attempt,
|
||||
retry=tenacity.retry_if_exception(_is_not_resent_failure),
|
||||
reraise=True,
|
||||
)
|
||||
else:
|
||||
retryer = tenacity.Retrying(
|
||||
stop=tenacity.stop_after_attempt(num_retries),
|
||||
before=_carry_failed_attempt,
|
||||
retry=tenacity.retry_if_exception(_is_not_resent_failure),
|
||||
reraise=True,
|
||||
)
|
||||
|
|
@ -6843,12 +6859,14 @@ async def aresponses_with_retries(*args, **kwargs):
|
|||
retryer = tenacity.AsyncRetrying(
|
||||
wait=tenacity.wait_exponential(multiplier=1, max=10),
|
||||
stop=tenacity.stop_after_attempt(num_retries),
|
||||
before=_carry_failed_attempt,
|
||||
retry=tenacity.retry_if_exception(_is_not_resent_failure),
|
||||
reraise=True,
|
||||
)
|
||||
else:
|
||||
retryer = tenacity.AsyncRetrying(
|
||||
stop=tenacity.stop_after_attempt(num_retries),
|
||||
before=_carry_failed_attempt,
|
||||
retry=tenacity.retry_if_exception(_is_not_resent_failure),
|
||||
reraise=True,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,7 +1,10 @@
|
|||
|
||||
import asyncio
|
||||
import copy
|
||||
import json
|
||||
from collections.abc import Iterator
|
||||
from collections.abc import Iterator, Mapping
|
||||
from concurrent.futures import Future
|
||||
from datetime import datetime
|
||||
from itertools import chain
|
||||
from typing import Final
|
||||
from unittest.mock import MagicMock
|
||||
|
|
@ -9,8 +12,11 @@ from unittest.mock import MagicMock
|
|||
import httpx
|
||||
import pytest
|
||||
import respx
|
||||
from pydantic import TypeAdapter
|
||||
|
||||
import litellm
|
||||
from litellm.integrations.custom_logger import CustomLogger
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
|
||||
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
|
||||
from litellm.llms.openai.chat.gpt_transformation import OpenAIGPTConfig
|
||||
from litellm.llms.openrouter.chat.transformation import (
|
||||
|
|
@ -1055,3 +1061,269 @@ def test_openrouter_streaming_chunk_ignores_invalid_provider_guard(provider_fiel
|
|||
).chunk_parser(chunk)
|
||||
|
||||
assert response._hidden_params.get("upstream_provider") is None
|
||||
|
||||
|
||||
class _FailureUsageRecorder(CustomLogger):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.payload: Final[Future[Mapping[str, object]]] = Future()
|
||||
|
||||
def log_failure_event(
|
||||
self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object
|
||||
) -> None:
|
||||
self.payload.set_result(TypeAdapter(Mapping[str, object]).validate_python(kwargs["standard_logging_object"]))
|
||||
|
||||
async def async_log_failure_event(
|
||||
self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object
|
||||
) -> None:
|
||||
self.log_failure_event(kwargs, response_obj, start_time, end_time)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("reported_cost", [0.0, 0.001])
|
||||
@pytest.mark.parametrize("async_call", [False, True])
|
||||
@pytest.mark.parametrize(
|
||||
("finish_reason", "error", "expected_status"),
|
||||
[
|
||||
("error", {"code": 502, "message": "Provider failed"}, 502),
|
||||
("error", None, 502),
|
||||
("stop", {"code": 429, "message": "Provider failed"}, 429),
|
||||
("error", {"code": "bad", "message": "Provider failed"}, 502),
|
||||
("error", {"code": 200, "message": "Provider failed"}, 502),
|
||||
],
|
||||
)
|
||||
@pytest.mark.asyncio
|
||||
async def test_openrouter_non_streaming_failure_raises(
|
||||
async_call: bool, finish_reason: str, error: object, expected_status: int, reported_cost: float
|
||||
) -> None:
|
||||
def respond(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(
|
||||
200,
|
||||
headers={"retry-after": "3"},
|
||||
json={
|
||||
"id": "gen-error",
|
||||
"object": "chat.completion",
|
||||
"created": 1700000000,
|
||||
"model": "openai/gpt-4o",
|
||||
"choices": [
|
||||
{
|
||||
"index": 0,
|
||||
"finish_reason": finish_reason,
|
||||
"message": {"role": "assistant", "content": "partial answer"},
|
||||
"error": error,
|
||||
}
|
||||
],
|
||||
"usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15, "cost": reported_cost},
|
||||
},
|
||||
)
|
||||
|
||||
recorder: Final = _FailureUsageRecorder()
|
||||
transport: Final = httpx.MockTransport(respond)
|
||||
|
||||
async def complete() -> None:
|
||||
if async_call:
|
||||
client: Final = AsyncHTTPHandler(transport=transport)
|
||||
try:
|
||||
await litellm.acompletion(
|
||||
model="openrouter/openai/gpt-4o",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
api_key="sk-test",
|
||||
client=client,
|
||||
num_retries=0,
|
||||
callbacks=[recorder],
|
||||
)
|
||||
finally:
|
||||
await client.close()
|
||||
else:
|
||||
with httpx.Client(transport=transport) as sync_client:
|
||||
litellm.completion(
|
||||
model="openrouter/openai/gpt-4o",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
api_key="sk-test",
|
||||
client=HTTPHandler(client=sync_client),
|
||||
num_retries=0,
|
||||
callbacks=[recorder],
|
||||
)
|
||||
|
||||
with pytest.raises((litellm.APIError, litellm.RateLimitError)) as exc_info:
|
||||
await complete()
|
||||
assert exc_info.value.status_code == expected_status
|
||||
assert "provider failed" in str(exc_info.value).lower()
|
||||
payload: Final = await asyncio.wrap_future(recorder.payload)
|
||||
assert payload["status"] == "failure"
|
||||
assert payload["prompt_tokens"] == 10
|
||||
assert payload["completion_tokens"] == 5
|
||||
assert payload["total_tokens"] == 15
|
||||
assert payload["response_cost"] == pytest.approx(reported_cost)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_openrouter_documented_failure_body_without_usage_raises() -> None:
|
||||
documented_body: Final = {
|
||||
"id": "gen-error",
|
||||
"object": "chat.completion",
|
||||
"created": 1700000000,
|
||||
"model": "openai/gpt-4o",
|
||||
"choices": [
|
||||
{
|
||||
"index": 0,
|
||||
"finish_reason": "error",
|
||||
"message": {"role": "assistant", "content": "partial output..."},
|
||||
"error": {
|
||||
"code": 502,
|
||||
"message": "Provider disconnected mid-stream",
|
||||
"metadata": {"error_type": "provider_unavailable"},
|
||||
},
|
||||
}
|
||||
],
|
||||
}
|
||||
recorder: Final = _FailureUsageRecorder()
|
||||
client: Final = AsyncHTTPHandler(transport=httpx.MockTransport(lambda _: httpx.Response(200, json=documented_body)))
|
||||
try:
|
||||
with pytest.raises(litellm.APIError) as exc_info:
|
||||
await litellm.acompletion(
|
||||
model="openrouter/openai/gpt-4o",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
api_key="sk-test",
|
||||
client=client,
|
||||
num_retries=0,
|
||||
callbacks=[recorder],
|
||||
)
|
||||
finally:
|
||||
await client.close()
|
||||
|
||||
assert exc_info.value.status_code == 502
|
||||
assert "provider disconnected mid-stream" in str(exc_info.value).lower()
|
||||
payload: Final = await asyncio.wrap_future(recorder.payload)
|
||||
assert payload["status"] == "failure"
|
||||
assert payload["total_tokens"] == 0
|
||||
assert payload["response_cost"] == 0
|
||||
|
||||
|
||||
class _SuccessUsageRecorder(CustomLogger):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.payload: Final[Future[Mapping[str, object]]] = Future()
|
||||
|
||||
def log_success_event(
|
||||
self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object
|
||||
) -> None:
|
||||
self.payload.set_result(TypeAdapter(Mapping[str, object]).validate_python(kwargs["standard_logging_object"]))
|
||||
|
||||
async def async_log_success_event(
|
||||
self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object
|
||||
) -> None:
|
||||
self.log_success_event(kwargs, response_obj, start_time, end_time)
|
||||
|
||||
|
||||
def _openrouter_body(finish_reason: str, content: str, usage: Mapping[str, object]) -> Mapping[str, object]:
|
||||
return {
|
||||
"id": "gen-1",
|
||||
"object": "chat.completion",
|
||||
"created": 1700000000,
|
||||
"model": "openai/gpt-4o",
|
||||
"choices": [
|
||||
{
|
||||
"index": 0,
|
||||
"finish_reason": finish_reason,
|
||||
"message": {"role": "assistant", "content": content},
|
||||
**({"error": {"code": 502, "message": "Provider failed"}} if finish_reason == "error" else {}),
|
||||
}
|
||||
],
|
||||
"usage": usage,
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("model_list", "router_settings"),
|
||||
[
|
||||
(
|
||||
[{"model_name": "or", "litellm_params": {"model": "openrouter/openai/gpt-4o", "api_key": "sk-test"}}],
|
||||
{"num_retries": 1},
|
||||
),
|
||||
(
|
||||
[
|
||||
{"model_name": "or", "litellm_params": {"model": "openrouter/openai/gpt-4o", "api_key": "sk-test"}},
|
||||
{"model_name": "or-backup", "litellm_params": {"model": "openrouter/openai/gpt-4o", "api_key": "sk-test"}},
|
||||
],
|
||||
{"num_retries": 0, "fallbacks": [{"or": ["or-backup"]}]},
|
||||
),
|
||||
],
|
||||
ids=["retry", "fallback"],
|
||||
)
|
||||
@pytest.mark.asyncio
|
||||
async def test_openrouter_recovered_failure_bills_the_failed_attempt(
|
||||
model_list: list[dict[str, object]], router_settings: dict[str, object]
|
||||
) -> None:
|
||||
bodies: Final = iter(
|
||||
(
|
||||
_openrouter_body(
|
||||
"error", "partial answer", {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15, "cost": 0.001}
|
||||
),
|
||||
_openrouter_body(
|
||||
"stop", "full answer", {"prompt_tokens": 7, "completion_tokens": 3, "total_tokens": 10, "cost": 0.002}
|
||||
),
|
||||
)
|
||||
)
|
||||
recorder: Final = _SuccessUsageRecorder()
|
||||
client: Final = AsyncHTTPHandler(transport=httpx.MockTransport(lambda _: httpx.Response(200, json=next(bodies))))
|
||||
messages: Final = [{"role": "user", "content": "hi"}]
|
||||
request_logging: Final = LiteLLMLogging(
|
||||
model="or",
|
||||
messages=messages,
|
||||
stream=False,
|
||||
call_type="acompletion",
|
||||
start_time=datetime(2026, 1, 1),
|
||||
litellm_call_id="recovered-openrouter-failure",
|
||||
function_id="recovered-openrouter-failure",
|
||||
dynamic_async_success_callbacks=[recorder],
|
||||
)
|
||||
try:
|
||||
response: Final = await litellm.Router(model_list=model_list, **router_settings).acompletion(
|
||||
model="or",
|
||||
messages=messages,
|
||||
client=client,
|
||||
litellm_logging_obj=request_logging,
|
||||
)
|
||||
finally:
|
||||
await client.close()
|
||||
|
||||
assert response.choices[0].message.content == "full answer"
|
||||
payload: Final = await asyncio.wrap_future(recorder.payload)
|
||||
assert payload["status"] == "success"
|
||||
assert payload["total_tokens"] == 25
|
||||
assert payload["response_cost"] == pytest.approx(0.003)
|
||||
|
||||
|
||||
def _failed_then_recovered_bodies() -> Iterator[Mapping[str, object]]:
|
||||
return iter(
|
||||
(
|
||||
_openrouter_body(
|
||||
"error", "partial answer", {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15, "cost": 0.001}
|
||||
),
|
||||
_openrouter_body(
|
||||
"stop", "full answer", {"prompt_tokens": 7, "completion_tokens": 3, "total_tokens": 10, "cost": 0.002}
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_openrouter_sdk_retry_bills_the_failed_attempt() -> None:
|
||||
bodies: Final = _failed_then_recovered_bodies()
|
||||
recorder: Final = _SuccessUsageRecorder()
|
||||
with httpx.Client(transport=httpx.MockTransport(lambda _: httpx.Response(200, json=next(bodies)))) as sync_client:
|
||||
response: Final = litellm.completion(
|
||||
model="openrouter/openai/gpt-4o",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
api_key="sk-test",
|
||||
num_retries=1,
|
||||
callbacks=[recorder],
|
||||
client=HTTPHandler(client=sync_client),
|
||||
)
|
||||
|
||||
assert response.choices[0].message.content == "full answer"
|
||||
payload: Final = await asyncio.wrap_future(recorder.payload)
|
||||
assert payload["status"] == "success"
|
||||
assert payload["prompt_tokens"] == 17
|
||||
assert payload["completion_tokens"] == 8
|
||||
assert payload["response_cost"] == pytest.approx(0.003)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue