mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-13 08:52:02 +00:00
fix(proxy): send the streamed Responses fallback metadata as a typed Responses event
With expose_fallback_errors_to_caller on and include_fallback_errors set, a streamed /v1/responses
call that fell back sent its fallback metadata as a chat.completion.chunk with no type, so the OpenAI
SDK read it as an audio delta event with type None. On the Responses route the frame is now
{"type": "litellm.fallback_metadata", "litellm_fallback": {...}} and goes through the same Responses
framing as every other event, so it gets an event: line whenever its neighbours do. The chat route
keeps its chat chunk unchanged
This commit is contained in:
parent
b25ea48c4b
commit
63a07248e6
3 changed files with 160 additions and 10 deletions
|
|
@ -9907,23 +9907,27 @@ def _get_streaming_fallback_metadata(
|
|||
return True, None, fallback_errors
|
||||
|
||||
|
||||
_RESPONSES_FALLBACK_METADATA_EVENT_TYPE: Final = "litellm.fallback_metadata"
|
||||
|
||||
|
||||
def _format_fallback_metadata_sse_event(
|
||||
*,
|
||||
fallback_model: str | None,
|
||||
fallback_errors: list[dict[str, object]],
|
||||
) -> str:
|
||||
import time
|
||||
|
||||
error_state: ResponsesStreamErrorState | None,
|
||||
) -> str | bytes:
|
||||
litellm_fallback: Final = {"fallback_model": fallback_model, "errors": fallback_errors}
|
||||
if error_state is not None:
|
||||
event: Final = {"type": _RESPONSES_FALLBACK_METADATA_EVENT_TYPE, "litellm_fallback": litellm_fallback}
|
||||
error_state.observe_chunk(event)
|
||||
return error_state.mark_emitted(f"data: {json.dumps(event)}\n\n")
|
||||
payload: Final = {
|
||||
"id": "litellm-fallback-metadata",
|
||||
"object": "chat.completion.chunk",
|
||||
"created": int(time.time()),
|
||||
"model": fallback_model or "",
|
||||
"choices": [],
|
||||
"litellm_fallback": {
|
||||
"fallback_model": fallback_model,
|
||||
"errors": fallback_errors,
|
||||
},
|
||||
"litellm_fallback": litellm_fallback,
|
||||
}
|
||||
return f"data: {json.dumps(payload)}\n\n"
|
||||
|
||||
|
|
@ -10575,6 +10579,7 @@ async def async_data_generator(
|
|||
yield _format_fallback_metadata_sse_event(
|
||||
fallback_model=fallback_model_from_metadata,
|
||||
fallback_errors=fallback_errors,
|
||||
error_state=error_state,
|
||||
)
|
||||
fallback_metadata_event_sent = True
|
||||
continue
|
||||
|
|
@ -10628,6 +10633,7 @@ async def async_data_generator(
|
|||
yield _format_fallback_metadata_sse_event(
|
||||
fallback_model=fallback_model_from_metadata,
|
||||
fallback_errors=fallback_errors,
|
||||
error_state=error_state,
|
||||
)
|
||||
fallback_metadata_event_sent = True
|
||||
|
||||
|
|
|
|||
|
|
@ -16,7 +16,14 @@ from urllib.parse import unquote
|
|||
import httpx
|
||||
import pytest
|
||||
import yaml
|
||||
from integration._support.anthropic_sse import error_body, message_json, message_stream, parse_sse, stream_reply
|
||||
from integration._support.anthropic_sse import (
|
||||
SseEvent,
|
||||
error_body,
|
||||
message_json,
|
||||
message_stream,
|
||||
parse_sse,
|
||||
stream_reply,
|
||||
)
|
||||
from integration._support.client import Gateway, eventually, gateway_from_environment
|
||||
from integration._support.database import read_rows
|
||||
from integration._support.openai_wire import chat_reply, responses_reply
|
||||
|
|
@ -34,6 +41,7 @@ _API_KEY: Final = "synthetic-fallback-errors-key"
|
|||
_ROUTER_ONLY_KEYS: Final = ("include_fallback_errors", "silent_model")
|
||||
_ANSWER: Final = "answered by"
|
||||
_UNAUTHORIZED_MESSAGE: Final = "scripted 401: the primary key was revoked"
|
||||
_RESPONSES_FALLBACK_METADATA_EVENT: Final = "litellm.fallback_metadata"
|
||||
_NONCE: Final = re.compile(r"nonce=([0-9a-f]{32})")
|
||||
_JSON: Final = TypeAdapter(dict[str, JsonValue])
|
||||
_ERRORS: Final = TypeAdapter(list[dict[str, JsonValue]])
|
||||
|
|
@ -447,11 +455,25 @@ def exposed(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Exposed]:
|
|||
yield _Exposed(proxy, wire)
|
||||
|
||||
|
||||
def _mention_the_outage(errors: list[dict[str, JsonValue]]) -> bool:
|
||||
return len(errors) == 1 and _UNAUTHORIZED_MESSAGE in str(errors[0]["message"])
|
||||
|
||||
|
||||
def _errors_mention_the_outage(errors_header: str | None) -> bool:
|
||||
if errors_header is None:
|
||||
return False
|
||||
errors: Final = _ERRORS.validate_json(errors_header)
|
||||
return len(errors) == 1 and _UNAUTHORIZED_MESSAGE in str(errors[0]["message"])
|
||||
return _mention_the_outage(_ERRORS.validate_json(errors_header))
|
||||
|
||||
|
||||
def _fallback_reports(events: tuple[SseEvent, ...]) -> tuple[SseEvent, ...]:
|
||||
return tuple(event for event in events if "litellm_fallback" in event.data)
|
||||
|
||||
|
||||
def _reports_the_outage_served_by_the_backup(report: SseEvent) -> bool:
|
||||
fallback: Final = _JSON.validate_python(report.data["litellm_fallback"])
|
||||
return fallback["fallback_model"] == _EXPOSED_OPENAI_BACKUP and _mention_the_outage(
|
||||
_ERRORS.validate_python(fallback["errors"])
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("stream", _STREAMING)
|
||||
|
|
@ -475,6 +497,10 @@ def test_exposed_chat_fallback_reports_the_errors_without_putting_the_flag_on_th
|
|||
if not stream:
|
||||
assert observed.attempted == "1", observed
|
||||
assert _errors_mention_the_outage(observed.errors_header), observed
|
||||
return
|
||||
reports: Final = _fallback_reports(parse_sse(observed.text))
|
||||
assert tuple(report.data.get("object") for report in reports) == ("chat.completion.chunk",), observed
|
||||
assert _reports_the_outage_served_by_the_backup(reports[0]), observed
|
||||
|
||||
|
||||
def test_exposed_chat_without_a_fallback_keeps_the_flag_off_the_wire_and_the_errors_header_off(
|
||||
|
|
@ -563,6 +589,13 @@ def test_exposed_responses_fallback_keeps_the_flag_off_the_wire(exposed: _Expose
|
|||
assert observed.served_id == _identity("responses", _EXPOSED_OPENAI_BACKUP, sent.nonce), observed
|
||||
assert observed.hit == (_EXPOSED_OPENAI_PRIMARY, _EXPOSED_OPENAI_BACKUP), observed
|
||||
assert observed.leaks == (), observed
|
||||
if not stream:
|
||||
return
|
||||
events: Final = parse_sse(observed.text)
|
||||
reports: Final = _fallback_reports(events)
|
||||
assert tuple(report.data.get("type") for report in reports) == (_RESPONSES_FALLBACK_METADATA_EVENT,), observed
|
||||
assert _reports_the_outage_served_by_the_backup(reports[0]), observed
|
||||
assert all("type" in event.data for event in events if event.data), observed
|
||||
|
||||
|
||||
@pytest.mark.parametrize("stream", _STREAMING)
|
||||
|
|
|
|||
|
|
@ -504,6 +504,7 @@ def test_format_fallback_metadata_sse_event():
|
|||
event = _format_fallback_metadata_sse_event(
|
||||
fallback_model="fallback-model",
|
||||
fallback_errors=fallback_errors,
|
||||
error_state=None,
|
||||
)
|
||||
|
||||
assert isinstance(event, str)
|
||||
|
|
@ -847,6 +848,116 @@ async def test_async_data_generator_emits_fallback_error_metadata_event(monkeypa
|
|||
assert isinstance(metadata_payload["created"], int)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_chat_stream_sends_fallback_metadata_in_place_of_a_stripped_usage_artifact(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
_patch_logging_flags(monkeypatch)
|
||||
monkeypatch.setitem(ps.general_settings, "expose_fallback_errors_to_caller", True)
|
||||
fallback_errors: Final = [
|
||||
{"message": "litellm.RateLimitError: upstream limited request", "type": "RateLimitError", "code": "429"}
|
||||
]
|
||||
usage_artifact: Final = ModelResponseStream(
|
||||
id="chatcmpl-test", choices=[], created=0, model="openai/internal-fallback", object="chat.completion.chunk"
|
||||
)
|
||||
upstream: Final = _FakeStream(
|
||||
[usage_artifact, _simple_chunk(model="openai/internal-fallback", content="hello")],
|
||||
hidden_params={
|
||||
"additional_headers": {
|
||||
"x-litellm-attempted-fallbacks": 1,
|
||||
"x-litellm-model-group": "fallback-model",
|
||||
"x-litellm-fallback-errors": json.dumps(fallback_errors),
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
frames: Final = tuple(
|
||||
[
|
||||
frame.decode() if isinstance(frame, bytes) else frame
|
||||
async for frame in async_data_generator(
|
||||
response=upstream,
|
||||
user_api_key_dict=_user_auth(),
|
||||
request_data={
|
||||
"model": "primary-model",
|
||||
"include_fallback_errors": True,
|
||||
"_litellm_strip_stream_usage": True,
|
||||
},
|
||||
)
|
||||
]
|
||||
)
|
||||
payloads: Final = tuple(json.loads(frame.removeprefix("data: ")) for frame in frames[:-1])
|
||||
|
||||
assert frames[-1] == "data: [DONE]\n\n"
|
||||
assert len(payloads) == 2
|
||||
assert payloads[0]["object"] == "chat.completion.chunk"
|
||||
assert payloads[0]["choices"] == []
|
||||
assert payloads[0]["litellm_fallback"] == {"fallback_model": "fallback-model", "errors": fallback_errors}
|
||||
assert payloads[1]["choices"][0]["delta"]["content"] == "hello"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_responses_stream_sends_fallback_metadata_as_a_typed_event_framed_like_its_neighbours(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
_patch_logging_flags(monkeypatch)
|
||||
monkeypatch.setitem(ps.general_settings, "expose_fallback_errors_to_caller", True)
|
||||
fallback_errors: Final = [
|
||||
{
|
||||
"message": "litellm.AuthenticationError: Incorrect API key provided",
|
||||
"type": "AuthenticationError",
|
||||
"param": None,
|
||||
"code": "401",
|
||||
}
|
||||
]
|
||||
response: Final = ResponsesAPIResponse(id="resp_fallback", created_at=1, model="fallback-model", output=[])
|
||||
upstream: Final = _FakeStream(
|
||||
[
|
||||
ResponseCreatedEvent.model_validate(
|
||||
{"type": "response.created", "sequence_number": 0, "response": response}
|
||||
),
|
||||
ResponseCompletedEvent.model_validate(
|
||||
{"type": "response.completed", "sequence_number": 1, "response": response}
|
||||
),
|
||||
],
|
||||
hidden_params={
|
||||
"additional_headers": {
|
||||
"x-litellm-attempted-fallbacks": 1,
|
||||
"x-litellm-model-group": "fallback-model",
|
||||
"x-litellm-fallback-errors": json.dumps(fallback_errors),
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
frames: Final = tuple(
|
||||
[
|
||||
frame.decode() if isinstance(frame, bytes) else frame
|
||||
async for frame in async_data_generator(
|
||||
response=upstream,
|
||||
user_api_key_dict=_user_auth(),
|
||||
request_data={"model": "primary-model", "include_fallback_errors": True},
|
||||
responses_stream_errors=True,
|
||||
)
|
||||
]
|
||||
)
|
||||
event_frames: Final = frames[:-1]
|
||||
payloads: Final = tuple(
|
||||
json.loads(next(line.removeprefix("data: ") for line in frame.splitlines() if line.startswith("data: ")))
|
||||
for frame in event_frames
|
||||
)
|
||||
|
||||
assert frames[-1] == "data: [DONE]\n\n"
|
||||
assert tuple(payload.get("type") for payload in payloads) == (
|
||||
"response.created",
|
||||
"litellm.fallback_metadata",
|
||||
"response.completed",
|
||||
)
|
||||
assert payloads[1] == {
|
||||
"type": "litellm.fallback_metadata",
|
||||
"litellm_fallback": {"fallback_model": "fallback-model", "errors": fallback_errors},
|
||||
}
|
||||
assert len({frame.startswith("event: ") for frame in event_frames}) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_data_generator_skips_fallback_error_event_without_opt_in(
|
||||
monkeypatch,
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue