mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-08 08:08:37 +00:00
fix(proxy): keep the HTTP status on late Responses errors and end Responses streams on error events
A late error body without a code now takes the response's status, an upstream error event counts as the stream's terminal event, and a Responses stream that fails before its first chunk ends with an error event instead of data: [DONE].
This commit is contained in:
mateo-berri
parent
7dfa539315
commit
86360077c6
4 changed files with 93 additions and 14 deletions
|
|
@ -1196,6 +1196,10 @@ def _sse_error_frames(error_obj: Mapping[str, object]) -> tuple[str, str]:
|
|||
return f"data: {json.dumps({'error': error_obj}, default=str)}\n\n", "data: [DONE]\n\n"
|
||||
|
||||
|
||||
def _responses_error_frames(error_obj: Mapping[str, object]) -> tuple[str, ...]:
|
||||
return (ResponsesStreamErrorState().format_error(error_obj),)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class LateSseFraming:
|
||||
"""How a response the keepalive already opened ends once its upstream fails late."""
|
||||
|
|
@ -1216,14 +1220,14 @@ def _openai_late_body_frames(produced: Response) -> tuple[bytes, ...]:
|
|||
def _responses_late_failure_frames(exc: Exception) -> tuple[str, ...]:
|
||||
if isinstance(exc, ModifyResponseException):
|
||||
return tuple(chunk.decode() for chunk in OpenAIResponsesHandler().build_block_sse_chunks(exc))
|
||||
_, error_obj = sse_error_payload(exc)
|
||||
return (ResponsesStreamErrorState().format_error(error_obj),)
|
||||
return _responses_error_frames(sse_error_payload(exc)[1])
|
||||
|
||||
|
||||
def _responses_late_body_frames(produced: Response) -> tuple[bytes, ...]:
|
||||
body: Final = bytes(produced.body)
|
||||
error_obj: Final = sse_error_object(body or b"{}") or {"message": body.decode(), "code": str(produced.status_code)}
|
||||
return (ResponsesStreamErrorState().format_error(error_obj).encode(),)
|
||||
error_obj: Final = sse_error_object(body or b"{}") or {"message": body.decode()}
|
||||
with_status: Final = {**error_obj, "code": error_obj.get("code") or str(produced.status_code)}
|
||||
return tuple(frame.encode() for frame in _responses_error_frames(with_status))
|
||||
|
||||
|
||||
OPENAI_LATE_SSE_FRAMING: Final = LateSseFraming(failure=_openai_late_failure_frames, body=_openai_late_body_frames)
|
||||
|
|
@ -1232,6 +1236,20 @@ RESPONSES_LATE_SSE_FRAMING: Final = LateSseFraming(
|
|||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class FirstChunkSseFraming:
|
||||
"""How a stream's first chunk is read for an error, and how the stream ends when reading it fails."""
|
||||
|
||||
data_payload: SseDataPayload
|
||||
error_frames: Callable[[Mapping[str, object]], tuple[str, ...]]
|
||||
|
||||
|
||||
OPENAI_FIRST_CHUNK_FRAMING: Final = FirstChunkSseFraming(data_payload=_sse_data_payload, error_frames=_sse_error_frames)
|
||||
RESPONSES_FIRST_CHUNK_FRAMING: Final = FirstChunkSseFraming(
|
||||
data_payload=_responses_event_data_payload, error_frames=_responses_error_frames
|
||||
)
|
||||
|
||||
|
||||
def _sse_stream_headers(headers: Mapping[str, str]) -> Mapping[str, str]:
|
||||
"""`headers` plus the two that stop reverse proxies from buffering SSE (issue #28384)."""
|
||||
return MappingProxyType({**headers, **_TTFT_KEEPALIVE_HEADERS})
|
||||
|
|
@ -1260,7 +1278,7 @@ async def create_response(
|
|||
request: Request | None = None,
|
||||
refresh_headers: Callable[[], Awaitable[Mapping[str, str]]] | None = None,
|
||||
general_settings: Mapping[str, object] = _NO_GENERAL_SETTINGS,
|
||||
first_chunk_payload: SseDataPayload = _sse_data_payload,
|
||||
framing: FirstChunkSseFraming = OPENAI_FIRST_CHUNK_FRAMING,
|
||||
) -> StreamingResponse | JSONResponse:
|
||||
"""
|
||||
Create streaming response, checking if the first chunk is an error.
|
||||
|
|
@ -1296,7 +1314,9 @@ async def create_response(
|
|||
|
||||
if first_chunk_value is not None:
|
||||
try:
|
||||
error_code_from_chunk: Final = await _parse_event_data_for_error(first_chunk_value, first_chunk_payload)
|
||||
error_code_from_chunk: Final = await _parse_event_data_for_error(
|
||||
first_chunk_value, framing.data_payload
|
||||
)
|
||||
if error_code_from_chunk is not None:
|
||||
# First chunk is an error, stream hasn't really started yet
|
||||
# Should return standard JSON error response instead of SSE format
|
||||
|
|
@ -1309,7 +1329,7 @@ async def create_response(
|
|||
# Parse error content
|
||||
error_dict: Final = with_call_id(
|
||||
JSON_OBJECT.validate_python(
|
||||
_extract_error_from_sse_chunk(first_chunk_value, first_chunk_payload)
|
||||
_extract_error_from_sse_chunk(first_chunk_value, framing.data_payload)
|
||||
),
|
||||
error_body_call_id(general_settings, resolved_headers.get(LITELLM_CALL_ID_HEADER)),
|
||||
)
|
||||
|
|
@ -1352,7 +1372,7 @@ async def create_response(
|
|||
error_status, error_obj = sse_error_payload(e)
|
||||
|
||||
async def error_gen_message() -> AsyncGenerator[str, None]:
|
||||
for frame in _sse_error_frames(error_obj):
|
||||
for frame in framing.error_frames(error_obj):
|
||||
yield frame
|
||||
|
||||
return StreamingResponse(
|
||||
|
|
@ -3286,8 +3306,8 @@ class ProxyBaseLLMRequestProcessing:
|
|||
request=request,
|
||||
refresh_headers=refresh_stream_headers,
|
||||
general_settings=general_settings,
|
||||
first_chunk_payload=(
|
||||
_responses_event_data_payload if route_type == "aresponses" else _sse_data_payload
|
||||
framing=(
|
||||
RESPONSES_FIRST_CHUNK_FRAMING if route_type == "aresponses" else OPENAI_FIRST_CHUNK_FRAMING
|
||||
),
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -172,7 +172,7 @@ class ResponsesStreamErrorState:
|
|||
self.model = event.response.model or self.model
|
||||
if event.response.created_at is not None:
|
||||
self.created_at = event.response.created_at
|
||||
if event.type in ("response.completed", "response.failed", "response.incomplete"):
|
||||
if event.type in ("response.completed", "response.failed", "response.incomplete", "error"):
|
||||
self.terminal_emitted = True
|
||||
if not event.type or "\n" in event.type or "\r" in event.type:
|
||||
return frame
|
||||
|
|
|
|||
|
|
@ -146,6 +146,33 @@ async def test_responses_stream_dispatches_named_events_without_a_chat_done_sent
|
|||
assert payload["type"] == event_type
|
||||
assert json.loads(frames[1].split("data: ", 1)[1])["delta"] == "hello"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_responses_stream_sends_no_failure_event_after_an_upstream_error_event() -> None:
|
||||
response: Final = ResponsesAPIResponse(id="resp_error_then_raise", created_at=1, model="local-model", output=[])
|
||||
created: Final = ResponseCreatedEvent.model_validate(
|
||||
{"type": "response.created", "sequence_number": 0, "response": response}
|
||||
)
|
||||
error_event: Final = ErrorEvent.model_validate(
|
||||
{"type": "error", "sequence_number": 1, "code": "server_error", "message": "upstream failed", "param": None}
|
||||
)
|
||||
|
||||
async def upstream() -> AsyncIterator[BaseModel]:
|
||||
yield created
|
||||
yield error_event
|
||||
raise litellm.APIError(status_code=500, message="Stream close failed", llm_provider="openai", model="local-model")
|
||||
|
||||
frames: Final = tuple(
|
||||
[
|
||||
frame.decode() if isinstance(frame, bytes) else frame
|
||||
async for frame in select_data_generator(
|
||||
response=upstream(), user_api_key_dict=_user_auth(), request_data={}, responses_stream_errors=True
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
assert tuple(frame.split("\n", 1)[0] for frame in frames) == ("event: response.created", "event: error")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_guardrail_error_string_mid_responses_stream_ends_it_as_a_named_error_event() -> None:
|
||||
response: Final = ResponsesAPIResponse(id="resp_hook_error", created_at=1, model="local-model", output=[])
|
||||
|
|
|
|||
|
|
@ -50,7 +50,8 @@ from litellm.proxy.common_request_processing import (
|
|||
_get_cost_breakdown_from_logging_obj,
|
||||
CostBreakdownHeaderValues,
|
||||
_has_attribute_error_in_chain,
|
||||
_responses_event_data_payload,
|
||||
OPENAI_FIRST_CHUNK_FRAMING,
|
||||
RESPONSES_FIRST_CHUNK_FRAMING,
|
||||
include_guardrail_response_requested,
|
||||
is_azure_model_router_request,
|
||||
RESPONSES_LATE_SSE_FRAMING,
|
||||
|
|
@ -2250,6 +2251,14 @@ class TestProxyBaseLLMRequestProcessing:
|
|||
)
|
||||
|
||||
|
||||
def _fail_stream_setup() -> str:
|
||||
raise RuntimeError("stream setup failed")
|
||||
|
||||
|
||||
async def _stream_failing_before_its_first_chunk() -> AsyncGenerator[str, None]:
|
||||
yield _fail_stream_setup()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestCommonRequestProcessingHelpers:
|
||||
async def consume_stream(self, streaming_response: StreamingResponse) -> list:
|
||||
|
|
@ -2335,7 +2344,7 @@ class TestCommonRequestProcessingHelpers:
|
|||
yield f"event: error\ndata: {json.dumps({'type': 'error', 'sequence_number': 0, 'error': error})}\n\n"
|
||||
|
||||
response: Final = await create_response(
|
||||
responses_stream(), "text/event-stream", {}, first_chunk_payload=_responses_event_data_payload
|
||||
responses_stream(), "text/event-stream", {}, framing=RESPONSES_FIRST_CHUNK_FRAMING
|
||||
)
|
||||
|
||||
assert isinstance(response, JSONResponse)
|
||||
|
|
@ -2350,13 +2359,30 @@ class TestCommonRequestProcessingHelpers:
|
|||
yield f"event: error\ndata: {event}\n\n"
|
||||
|
||||
response: Final = await create_response(
|
||||
responses_stream(), "text/event-stream", {}, first_chunk_payload=_responses_event_data_payload
|
||||
responses_stream(), "text/event-stream", {}, framing=RESPONSES_FIRST_CHUNK_FRAMING
|
||||
)
|
||||
|
||||
assert isinstance(response, JSONResponse)
|
||||
assert response.status_code == 400
|
||||
assert json.loads(response.body) == {"error": error}
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"framing,frame_starts",
|
||||
[
|
||||
(OPENAI_FIRST_CHUNK_FRAMING, ('data: {"error": ', "data: [DONE]")),
|
||||
(RESPONSES_FIRST_CHUNK_FRAMING, ('event: error\ndata: {"type": "error", "code": "server_error"',)),
|
||||
],
|
||||
)
|
||||
async def test_a_stream_that_fails_before_its_first_chunk_ends_in_its_own_api_shape(self, framing, frame_starts):
|
||||
response: Final = await create_response(
|
||||
_stream_failing_before_its_first_chunk(), "text/event-stream", {}, framing=framing
|
||||
)
|
||||
frames: Final = await self.consume_stream(response)
|
||||
|
||||
assert response.status_code == 500
|
||||
assert len(frames) == len(frame_starts)
|
||||
assert all(frame.startswith(start) for frame, start in zip(frames, frame_starts, strict=True))
|
||||
|
||||
async def test_create_streaming_response_first_chunk_not_error(self):
|
||||
async def mock_generator():
|
||||
yield 'data: {"content": "first part"}\n\n'
|
||||
|
|
@ -10591,6 +10617,11 @@ _POLICY_BLOCK_DETAIL: Final = MappingProxyType(
|
|||
)
|
||||
|
||||
|
||||
async def _slow_codeless_rate_limit_body() -> Response:
|
||||
await asyncio.sleep(0.2)
|
||||
return JSONResponse(status_code=429, content={"error": {"message": "slow down", "code": None}})
|
||||
|
||||
|
||||
async def _slow_dict_detail_failure() -> Response:
|
||||
await asyncio.sleep(0.2)
|
||||
raise HTTPException(status_code=400, detail=dict(_POLICY_BLOCK_DETAIL))
|
||||
|
|
@ -10600,6 +10631,7 @@ async def _slow_dict_detail_failure() -> Response:
|
|||
"produce,expected_code,expected_error",
|
||||
[
|
||||
(_slow_rate_limit_failure, "rate_limit_exceeded", {"message": "rate limited", "code": "429"}),
|
||||
(_slow_codeless_rate_limit_body, "rate_limit_exceeded", {"message": "slow down", "code": "429"}),
|
||||
(
|
||||
_slow_guardrail_block_body,
|
||||
"invalid_request_error",
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue