mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-11 08:38:24 +00:00
fix(proxy): keep the reason when a background Responses request fails
A background poll of a request a post-call guardrail blocked saved a null code and message, because the guardrail's error event nests them. A request whose first chunk was an error polled as completed with no error. Both now poll failed with the same code and message a late failure gets.
This commit is contained in:
parent
b2f3614faf
commit
f343c2421a
4 changed files with 116 additions and 9 deletions
|
|
@ -126,6 +126,14 @@ def _response_error_code(details: _FailureDetails) -> str:
|
|||
return _status_error_code(details.status_code)
|
||||
|
||||
|
||||
def responses_error_fields(error: Mapping[str, object]) -> Mapping[str, object]:
|
||||
details: Final = _FailureDetails.model_validate(error)
|
||||
return {
|
||||
"code": _response_error_code(details.model_copy(update={"status_code": _status_of(details.code)})),
|
||||
"message": details.message,
|
||||
}
|
||||
|
||||
|
||||
class ResponsesStreamErrorState:
|
||||
def __init__(self) -> None:
|
||||
self.response_id: str | None = None
|
||||
|
|
@ -157,12 +165,10 @@ class ResponsesStreamErrorState:
|
|||
return prefix.encode("utf-8") + frame if isinstance(frame, bytes) else prefix + frame
|
||||
|
||||
def format_error(self, error: Mapping[str, object]) -> str:
|
||||
details: Final = _FailureDetails.model_validate(error)
|
||||
payload: Final = json.dumps(
|
||||
{
|
||||
"type": ResponsesAPIStreamEvents.ERROR.value,
|
||||
"code": _response_error_code(details.model_copy(update={"status_code": _status_of(details.code)})),
|
||||
"message": details.message,
|
||||
**responses_error_fields(error),
|
||||
"param": error.get("param"),
|
||||
"sequence_number": self.sequence_number + 1,
|
||||
"error": error,
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ https://platform.openai.com/docs/api-reference/responses-streaming
|
|||
import asyncio
|
||||
import json
|
||||
from collections.abc import AsyncIterable, Callable, Mapping, Sequence
|
||||
from http import HTTPStatus
|
||||
from typing import TYPE_CHECKING, Final, Protocol, TypeAlias, runtime_checkable
|
||||
|
||||
from fastapi import Request, Response
|
||||
|
|
@ -20,6 +21,7 @@ from typing_extensions import ReadOnly, TypedDict
|
|||
from litellm._logging import verbose_proxy_logger
|
||||
from litellm.proxy.auth.user_api_key_auth import UserAPIKeyAuth
|
||||
from litellm.proxy.common_request_processing import RESPONSES_LATE_SSE_FRAMING, ProxyBaseLLMRequestProcessing
|
||||
from litellm.proxy.common_utils.responses_stream_errors import responses_error_fields, sse_error_object
|
||||
from litellm.proxy.response_polling.polling_handler import ResponsePollingHandler
|
||||
from litellm.types.llms.openai import ResponsesAPIStatus
|
||||
|
||||
|
|
@ -85,6 +87,14 @@ def _sse_frame_data(frame: str) -> str | None:
|
|||
return next((line[6:].strip() for line in frame.splitlines() if line.startswith("data: ")), None)
|
||||
|
||||
|
||||
def _first_chunk_error(response: object) -> Mapping[str, object] | None:
|
||||
if isinstance(response, _StreamsBody) or not isinstance(response, Response):
|
||||
return None
|
||||
if response.status_code < HTTPStatus.BAD_REQUEST:
|
||||
return None
|
||||
return responses_error_fields(sse_error_object(bytes(response.body)) or {"code": response.status_code})
|
||||
|
||||
|
||||
async def _never_receive() -> Message:
|
||||
await asyncio.Event().wait()
|
||||
raise AssertionError("unreachable")
|
||||
|
|
@ -329,7 +339,7 @@ async def background_streaming_task(
|
|||
|
||||
elif event_type == "error":
|
||||
terminal_status = "failed"
|
||||
terminal_error = {"code": event.get("code"), "message": event.get("message")}
|
||||
terminal_error = responses_error_fields(event.get("error") or event)
|
||||
|
||||
elif event_type in (
|
||||
"response.completed",
|
||||
|
|
@ -387,14 +397,15 @@ async def background_streaming_task(
|
|||
# Final flush to ensure all accumulated state is saved
|
||||
await flush_state_if_needed(force=True)
|
||||
|
||||
# Use the terminal status from the stream, default to "completed"
|
||||
final_status: Final = terminal_status or "completed"
|
||||
early_error: Final = _first_chunk_error(response)
|
||||
final_error: Final = terminal_error if early_error is None else early_error
|
||||
final_status: Final = "failed" if early_error is not None else terminal_status or "completed"
|
||||
|
||||
await polling_handler.update_state(
|
||||
polling_id=polling_id,
|
||||
status=final_status,
|
||||
usage=usage_data,
|
||||
error=terminal_error,
|
||||
error=final_error,
|
||||
reasoning=reasoning_data,
|
||||
tool_choice=tool_choice_data,
|
||||
tools=tools_data,
|
||||
|
|
@ -416,7 +427,7 @@ async def background_streaming_task(
|
|||
"Finished background streaming for %s, status=%s, error=%s, incomplete_details=%s, output_items=%s",
|
||||
polling_id,
|
||||
final_status,
|
||||
terminal_error,
|
||||
final_error,
|
||||
incomplete_details_data,
|
||||
len(output_items),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -98,7 +98,7 @@ class ResponsePollingHandler:
|
|||
polling_id: str,
|
||||
status: ResponsesAPIStatus | None = None,
|
||||
usage: dict | None = None,
|
||||
error: dict | None = None,
|
||||
error: Mapping[str, object] | None = None,
|
||||
incomplete_details: dict | None = None,
|
||||
reasoning: dict | None = None,
|
||||
tool_choice: object | None = None,
|
||||
|
|
|
|||
|
|
@ -1553,6 +1553,96 @@ class TestBackgroundStreamingTerminalEvents:
|
|||
assert final_call.kwargs["status"] == "failed"
|
||||
assert final_call.kwargs["error"] == {"code": "rate_limit_exceeded", "message": "rate limited"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_nested_only_error_event_keeps_its_reason(self):
|
||||
from litellm.proxy.response_polling.background_streaming import (
|
||||
background_streaming_task,
|
||||
)
|
||||
|
||||
nested_error: Final = {"type": "guardrail_error", "code": "400", "message": "output blocked by policy"}
|
||||
error_event: Final = {"type": "error", "sequence_number": 9, "error": nested_error}
|
||||
|
||||
async def _body_iterator():
|
||||
yield b'data: {"type": "response.in_progress"}\n\n'
|
||||
yield f"event: error\ndata: {json.dumps(error_event)}\n\n".encode()
|
||||
|
||||
mock_response: Final = Mock()
|
||||
mock_response.body_iterator = _body_iterator()
|
||||
handler: Final = AsyncMock(spec=ResponsePollingHandler)
|
||||
kwargs: Final = _make_background_streaming_kwargs("poll_nested_error", handler)
|
||||
|
||||
with patch( # test-quality-ok: the processor is built inside the task, same idiom as the sibling tests
|
||||
"litellm.proxy.response_polling.background_streaming.ProxyBaseLLMRequestProcessing"
|
||||
) as MockProcessor:
|
||||
MockProcessor.return_value.base_process_llm_request = AsyncMock(
|
||||
return_value=mock_response
|
||||
)
|
||||
await background_streaming_task(**kwargs)
|
||||
|
||||
final_call: Final = handler.update_state.call_args_list[-1]
|
||||
assert final_call.kwargs["status"] == "failed"
|
||||
assert final_call.kwargs["error"] == {"code": "invalid_request_error", "message": "output blocked by policy"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
("status_code", "body", "expected_error"),
|
||||
[
|
||||
(
|
||||
404,
|
||||
json.dumps(
|
||||
{"error": {"message": "model not found", "type": "invalid_request_error", "param": None, "code": "404"}}
|
||||
),
|
||||
{"code": "not_found_error", "message": "model not found"},
|
||||
),
|
||||
(429, "Too Many Requests", {"code": "rate_limit_exceeded", "message": None}),
|
||||
],
|
||||
)
|
||||
async def test_a_first_chunk_error_polls_failed_like_a_late_one(self, status_code, body, expected_error):
|
||||
from fastapi import Response
|
||||
|
||||
from litellm.proxy.response_polling.background_streaming import (
|
||||
background_streaming_task,
|
||||
)
|
||||
|
||||
first_chunk_error: Final = Response(status_code=status_code, content=body)
|
||||
handler: Final = AsyncMock(spec=ResponsePollingHandler)
|
||||
kwargs: Final = _make_background_streaming_kwargs("poll_first_chunk_error", handler)
|
||||
|
||||
with patch( # test-quality-ok: the processor is built inside the task, same idiom as the sibling tests
|
||||
"litellm.proxy.response_polling.background_streaming.ProxyBaseLLMRequestProcessing"
|
||||
) as MockProcessor:
|
||||
MockProcessor.return_value.base_process_llm_request = AsyncMock(
|
||||
return_value=first_chunk_error
|
||||
)
|
||||
await background_streaming_task(**kwargs)
|
||||
|
||||
final_call: Final = handler.update_state.call_args_list[-1]
|
||||
assert final_call.kwargs["status"] == "failed"
|
||||
assert final_call.kwargs["error"] == expected_error
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_non_streamed_success_response_still_polls_completed(self):
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
from litellm.proxy.response_polling.background_streaming import (
|
||||
background_streaming_task,
|
||||
)
|
||||
|
||||
handler: Final = AsyncMock(spec=ResponsePollingHandler)
|
||||
kwargs: Final = _make_background_streaming_kwargs("poll_non_streamed_success", handler)
|
||||
|
||||
with patch( # test-quality-ok: the processor is built inside the task, same idiom as the sibling tests
|
||||
"litellm.proxy.response_polling.background_streaming.ProxyBaseLLMRequestProcessing"
|
||||
) as MockProcessor:
|
||||
MockProcessor.return_value.base_process_llm_request = AsyncMock(
|
||||
return_value=JSONResponse(status_code=200, content={"id": "resp_123", "status": "completed"})
|
||||
)
|
||||
await background_streaming_task(**kwargs)
|
||||
|
||||
final_call: Final = handler.update_state.call_args_list[-1]
|
||||
assert final_call.kwargs["status"] == "completed"
|
||||
assert final_call.kwargs["error"] is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_response_incomplete_sets_incomplete_status_and_details(self):
|
||||
"""Test that a response.incomplete stream event results in incomplete status"""
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue