mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-11 08:38:24 +00:00
fix(proxy): send the streamed Responses fallback metadata as a typed Responses event (#50804)
* 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
* test(proxy): stop pinning the Responses stream terminator in the fallback metadata test
Main ends Responses streams without data: [DONE] since #47980, so the test reads the event frames whether or not a terminator follows them, and passes both before and after that change merges
* fix(proxy): keep the top-level model on the Responses fallback metadata event
* fix(proxy): keep the fallback metadata id on the Responses event and omit an empty model
* fix(proxy): leave the top-level model off the Responses fallback event when raw model names are on
The complexity router's return_raw_model_name leaves every other event unstamped, so the fallback event now matches. A new test streams a raw upstream frame before the event to cover how main names it.
* chore: push again after GitHub dropped the previous push event
* fix(proxy): send the Responses fallback event without an event line so older OpenAI SDKs read it
---------
Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
parent
845f109524
commit
a7696a2bc2
4 changed files with 184 additions and 12 deletions
|
|
@ -381,7 +381,7 @@ def resolve_litellm_call_id(client_call_id: str | None) -> str:
|
|||
return str(uuid.uuid4())
|
||||
|
||||
|
||||
def should_return_raw_model_name(request_data: dict[str, object]) -> bool:
|
||||
def should_return_raw_model_name(request_data: Mapping[str, object]) -> bool:
|
||||
return any(
|
||||
isinstance(metadata, dict) and metadata.get(RETURN_RAW_MODEL_NAME_METADATA_KEY) is True
|
||||
for metadata in (request_data.get("metadata"), request_data.get("litellm_metadata"))
|
||||
|
|
|
|||
|
|
@ -10303,23 +10303,34 @@ def _get_streaming_fallback_metadata(
|
|||
return True, None, fallback_errors
|
||||
|
||||
|
||||
_FALLBACK_METADATA_ID: Final = "litellm-fallback-metadata"
|
||||
_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]],
|
||||
responses_stream: bool,
|
||||
return_raw_model_name: bool,
|
||||
) -> str:
|
||||
import time
|
||||
|
||||
litellm_fallback: Final = {"fallback_model": fallback_model, "errors": fallback_errors}
|
||||
if responses_stream:
|
||||
stamped_model: Final = None if return_raw_model_name else fallback_model
|
||||
event: Final = {
|
||||
"type": _RESPONSES_FALLBACK_METADATA_EVENT_TYPE,
|
||||
"id": _FALLBACK_METADATA_ID,
|
||||
**({"model": stamped_model} if stamped_model else {}),
|
||||
"litellm_fallback": litellm_fallback,
|
||||
}
|
||||
return f"data: {json.dumps(event)}\n\n"
|
||||
payload: Final = {
|
||||
"id": "litellm-fallback-metadata",
|
||||
"id": _FALLBACK_METADATA_ID,
|
||||
"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"
|
||||
|
||||
|
|
@ -10328,7 +10339,7 @@ def _restamp_streaming_chunk_model(
|
|||
*,
|
||||
chunk: object,
|
||||
requested_model_from_client: str,
|
||||
request_data: dict,
|
||||
request_data: Mapping[str, object],
|
||||
model_mismatch_logged: bool,
|
||||
fallback_was_attempted: bool = False,
|
||||
fallback_model_from_metadata: str | None = None,
|
||||
|
|
@ -10906,6 +10917,7 @@ async def async_data_generator(
|
|||
model_mismatch_logged = False
|
||||
fallback_metadata_event_sent = False
|
||||
include_fallback_errors: Final = _should_include_fallback_errors(request_data)
|
||||
return_raw_model_name: Final = should_return_raw_model_name(request_data)
|
||||
# Use a running string instead of list + join to avoid O(n^2) overhead.
|
||||
# Previously "".join(str_so_far_parts) was called every chunk, re-joining
|
||||
# the entire accumulated response. String += is O(n) amortized total.
|
||||
|
|
@ -11003,6 +11015,8 @@ async def async_data_generator(
|
|||
yield _format_fallback_metadata_sse_event(
|
||||
fallback_model=fallback_model_from_metadata,
|
||||
fallback_errors=fallback_errors,
|
||||
responses_stream=error_state is not None,
|
||||
return_raw_model_name=return_raw_model_name,
|
||||
)
|
||||
fallback_metadata_event_sent = True
|
||||
continue
|
||||
|
|
@ -11059,6 +11073,8 @@ async def async_data_generator(
|
|||
yield _format_fallback_metadata_sse_event(
|
||||
fallback_model=fallback_model_from_metadata,
|
||||
fallback_errors=fallback_errors,
|
||||
responses_stream=error_state is not None,
|
||||
return_raw_model_name=return_raw_model_name,
|
||||
)
|
||||
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]])
|
||||
|
|
@ -445,11 +453,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)
|
||||
|
|
@ -473,6 +495,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(
|
||||
|
|
@ -561,6 +587,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)
|
||||
|
|
|
|||
|
|
@ -874,6 +874,8 @@ def test_format_fallback_metadata_sse_event():
|
|||
event = _format_fallback_metadata_sse_event(
|
||||
fallback_model="fallback-model",
|
||||
fallback_errors=fallback_errors,
|
||||
responses_stream=False,
|
||||
return_raw_model_name=False,
|
||||
)
|
||||
|
||||
assert isinstance(event, str)
|
||||
|
|
@ -1469,6 +1471,127 @@ 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
|
||||
@pytest.mark.parametrize("fallback_model", ["fallback-model", None])
|
||||
@pytest.mark.parametrize("return_raw_model_name", [False, True])
|
||||
async def test_responses_stream_sends_fallback_metadata_as_a_typed_event_without_an_event_line(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
fallback_model: str | None,
|
||||
return_raw_model_name: bool,
|
||||
) -> 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} if fallback_model else {}),
|
||||
"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,
|
||||
"metadata": {RETURN_RAW_MODEL_NAME_METADATA_KEY: return_raw_model_name},
|
||||
},
|
||||
responses_stream_errors=True,
|
||||
)
|
||||
]
|
||||
)
|
||||
event_frames: Final = tuple(frame for frame in frames if frame != "data: [DONE]\n\n")
|
||||
payloads: Final = tuple(
|
||||
json.loads(next(line.removeprefix("data: ") for line in frame.splitlines() if line.startswith("data: ")))
|
||||
for frame in event_frames
|
||||
)
|
||||
|
||||
assert tuple(payload.get("type") for payload in payloads) == (
|
||||
"response.created",
|
||||
"litellm.fallback_metadata",
|
||||
"response.completed",
|
||||
)
|
||||
stamped_model: Final = None if return_raw_model_name else fallback_model
|
||||
assert payloads[1] == {
|
||||
"type": "litellm.fallback_metadata",
|
||||
"id": "litellm-fallback-metadata",
|
||||
**({"model": stamped_model} if stamped_model else {}),
|
||||
"litellm_fallback": {"fallback_model": fallback_model, "errors": fallback_errors},
|
||||
}
|
||||
assert {payload.get("model") for payload in payloads} == {stamped_model}
|
||||
assert event_frames[1] == f"data: {json.dumps(payloads[1])}\n\n"
|
||||
|
||||
|
||||
@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