fix(responses): replay phase on the azure route and tag streamed text deltas with phase

This commit is contained in:
mateo-berri 2026-10-29 09:11:53 -07:00
parent 683b94b462
commit 26483640e1
4 changed files with 155 additions and 46 deletions

View file

@ -251,7 +251,7 @@ class ResponsesToCompletionBridgeHandler:
else:
if self._is_preformatted_cached_chat_stream(result):
return self._apply_post_stream_processing(result, model, custom_llm_provider)
from .transformation import is_structured_output_request
from .transformation import is_structured_output_request, replays_responses_history
completion_stream: Final = self.transformation_handler.get_model_response_iterator(
streaming_response=result,
@ -260,7 +260,7 @@ class ResponsesToCompletionBridgeHandler:
structured_output=is_structured_output_request(
cast("Mapping[str, object]", optional_params) # cast-ok: optional_params is an untyped dict
),
replays_openai_history=custom_llm_provider == "openai",
replays_openai_history=replays_responses_history(custom_llm_provider),
)
streamwrapper: Final = CustomStreamWrapper(
completion_stream=completion_stream,
@ -355,7 +355,7 @@ class ResponsesToCompletionBridgeHandler:
else:
if self._is_preformatted_cached_chat_stream(result):
return self._apply_post_stream_processing(result, model, custom_llm_provider)
from .transformation import is_structured_output_request
from .transformation import is_structured_output_request, replays_responses_history
completion_stream: Final = self.transformation_handler.get_model_response_iterator(
streaming_response=result,
@ -364,7 +364,7 @@ class ResponsesToCompletionBridgeHandler:
structured_output=is_structured_output_request(
cast("Mapping[str, object]", optional_params) # cast-ok: optional_params is an untyped dict
),
replays_openai_history=custom_llm_provider == "openai",
replays_openai_history=replays_responses_history(custom_llm_provider),
)
streamwrapper: Final = CustomStreamWrapper(
completion_stream=completion_stream,

View file

@ -159,6 +159,11 @@ def _parsed_output_item(item: object) -> _OutputItem | None:
return None
def message_item_phase(item: object) -> Phase | None:
parsed: Final = _parsed_output_item(item)
return parsed.phase if isinstance(parsed, _OutputMessage) else None
def _is_message(item: object) -> bool:
try:
_AnyMessage.model_validate(_as_json(item))

View file

@ -26,7 +26,9 @@ from litellm import ModelResponse
from litellm._logging import verbose_logger
from litellm.completion_extras.litellm_responses_transformation.history import (
PHASE_FIELD,
Phase,
UnknownPhase,
message_item_phase,
output_message_text,
reasoning_input_item,
replay_fields,
@ -67,7 +69,7 @@ from litellm.types.llms.openai import (
ResponsesAPIResponse,
ResponsesAPIStreamEvents,
)
from litellm.types.utils import GenericStreamingChunk, ModelResponseStream
from litellm.types.utils import Delta, GenericStreamingChunk, ModelResponseStream
if TYPE_CHECKING:
from openai.types.responses import ResponseInputImageParam, ResponseOutputItem
@ -204,6 +206,13 @@ def _served_provider(litellm_params: Mapping[str, object]) -> str | None:
return custom_llm_provider if isinstance(custom_llm_provider, str) else None
_HISTORY_REPLAY_PROVIDERS: Final = frozenset({"openai", "azure"})
def replays_responses_history(provider: str | None) -> bool:
return provider in _HISTORY_REPLAY_PROVIDERS
def _supports_audio_input(model: str, litellm_params: Mapping[str, object]) -> bool:
provider: Final = _served_provider(litellm_params)
base_model: Final = litellm_params.get("base_model")
@ -557,6 +566,7 @@ def completed_stream_history(
*,
completed: bool,
structured_output: bool,
streamed_phases: bool,
) -> dict[str, object] | None: # mutable-ok: Delta.provider_specific_fields is typed as a dict
items: Final = _STREAMED_OUTPUT_ITEMS.validate_python(output_payload)
calls: Final = tuple(item for item in items if item.get("type") in ("function_call", "custom_tool_call"))
@ -568,12 +578,19 @@ def completed_stream_history(
"provider_specific_fields": tool_search_fields,
}
fields: Final = {
**({PHASE_FIELD: None} if streamed_phases else {}),
**(tool_search_fields or {}),
**replay_fields(items, streamed_message, completed=completed, first_message_only=structured_output),
}
return fields or None
def _text_delta(content: object, phase: Phase | None) -> Delta:
if phase is None:
return Delta(content=content)
return Delta(content=content, provider_specific_fields={PHASE_FIELD: phase})
def _assistant_phase_field(msg: object, replay_openai_history: bool, drop_params: bool) -> Mapping[str, object]:
match requested_phase(msg) if replay_openai_history else None:
case None:
@ -1138,7 +1155,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
keep_prompt_cache_breakpoints=supports_prompt_cache_breakpoint,
replay_tool_search=_requests_tool_search(cast(object, optional_params)), # cast-ok: upcast
include_namespace=True,
replay_openai_history=provider == "openai",
replay_openai_history=replays_responses_history(provider),
)
# OpenAI's Responses API rejects an empty input. For a system-only
# request, carry the system message as a system-role input item instead
@ -1513,10 +1530,11 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
structured_output=is_structured_output_request(
cast("Mapping[str, object]", optional_params) # cast-ok: optional_params is an untyped dict
),
replay_openai_history=_served_provider(
cast("Mapping[str, object]", litellm_params) # cast-ok: litellm_params is an untyped dict
)
== "openai",
replay_openai_history=replays_responses_history(
_served_provider(
cast("Mapping[str, object]", litellm_params) # cast-ok: litellm_params is an untyped dict
)
),
completed=raw_response.status == "completed",
)
@ -1979,6 +1997,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
self._tool_call_index_map: dict[int, int] = {} # mutable-ok: per-stream accumulator state
self._structured_output: Final = structured_output
self._first_message_item_done: bool = False
self._message_phases: Mapping[int, Phase] = MappingProxyType({})
def _handle_string_chunk(
self, str_line: Union[str, "BaseModel"]
@ -2021,6 +2040,8 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
tool_call_index_map: dict[int, int] | None = None, # mutable-ok: per-stream state, remapped in place
replays_openai_history: bool = False,
structured_output: bool = False,
item_phase: Phase | None = None,
streamed_phases: bool = False,
) -> "ModelResponseStream":
"""
Translate a Responses API streaming chunk to OpenAI chat completion streaming format.
@ -2030,6 +2051,8 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
tool_call_index_map: Per-stream output_index -> sequential tool_call index map
replays_openai_history: Record the completed output's item sequence for the next turn
structured_output: Only the first message item became content
item_phase: Phase of the message item this chunk belongs to
streamed_phases: An earlier text delta of this stream carried a phase
Returns:
ModelResponseStream: OpenAI-formatted streaming chunk
@ -2203,7 +2226,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
choices=[
StreamingChoices(
index=0,
delta=Delta(content=content_part),
delta=_text_delta(content_part, item_phase),
finish_reason=None,
)
]
@ -2265,6 +2288,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
tool_search_fields,
completed=event_type == "response.completed",
structured_output=structured_output,
streamed_phases=streamed_phases,
)
if replays_openai_history
else tool_search_fields
@ -2309,6 +2333,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
"""
verbose_logger.debug("Chat provider: transform_streaming_response called with chunk: %s", chunk)
self._remember_served_service_tier(chunk)
self._remember_message_phase(chunk)
translated: Final = (
self._empty_content_chunk()
if self._is_text_delta_after_structured_answer(chunk)
@ -2317,6 +2342,8 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
tool_call_index_map=self._tool_call_index_map,
replays_openai_history=self._replays_openai_history,
structured_output=self._structured_output,
item_phase=self._item_phase(chunk),
streamed_phases=bool(self._message_phases),
)
)
if self._structured_output and _is_message_item_done(chunk):
@ -2338,6 +2365,21 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
return ModelResponseStream(choices=[StreamingChoices(index=0, delta=Delta(content=""), finish_reason=None)])
def _remember_message_phase(self, chunk: Mapping[str, object]) -> None:
if (
not self._replays_openai_history
or _responses_event_type(chunk) != ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED.value
):
return
output_index: Final = chunk.get("output_index")
phase: Final = message_item_phase(chunk.get("item"))
if isinstance(output_index, int) and phase is not None:
self._message_phases = MappingProxyType({**self._message_phases, output_index: phase})
def _item_phase(self, chunk: Mapping[str, object]) -> Phase | None:
output_index: Final = chunk.get("output_index")
return self._message_phases.get(output_index) if isinstance(output_index, int) else None
def _remember_served_service_tier(self, chunk: dict[str, object]) -> None:
response_payload: Final = chunk.get("response")
if not isinstance(response_payload, dict):

View file

@ -1,5 +1,6 @@
import json
from collections.abc import Callable, Iterator, Mapping, Sequence
from types import MappingProxyType
from typing import Final
import httpx
@ -12,7 +13,21 @@ from litellm.exceptions import MidStreamFallbackError
OPENAI_RESPONSES_URL: Final = "https://api.openai.com/v1/responses"
AZURE_API_BASE: Final = "https://litellm-history-test.openai.azure.com"
XAI_API_BASE: Final = "https://api.x.ai/v1"
MODEL: Final = "gpt-5.6"
OPENAI_MODEL: Final = f"openai/responses/{MODEL}"
AZURE_MODEL: Final = f"azure/responses/{MODEL}"
XAI_MODEL: Final = "xai/responses/grok-4.7"
ROUTES: Final = MappingProxyType(
{
"openai": (OPENAI_RESPONSES_URL, MappingProxyType({})),
"azure": (
AZURE_API_BASE,
MappingProxyType({"api_base": AZURE_API_BASE, "api_version": "2025-04-01-preview"}),
),
"xai": (XAI_API_BASE, MappingProxyType({"api_base": XAI_API_BASE})),
}
)
TOOLS: Final = (
{
"type": "function",
@ -183,11 +198,15 @@ def _sse(output: Sequence[Mapping[str, object]], terminal_event: str | None = "r
return httpx.Response(200, content=body, headers={"content-type": "text/event-stream"})
def _route(model: str) -> tuple[str, Mapping[str, str]]:
return ROUTES[model.split("/", 1)[0]]
def _assistant_turn(
output: Sequence[Mapping[str, object]],
*,
stream: bool,
model: str = f"openai/responses/{MODEL}",
model: str = OPENAI_MODEL,
terminal_event: str | None = "response.completed",
status: str = "completed",
response_format: Mapping[str, object] | None = None,
@ -195,19 +214,17 @@ def _assistant_turn(
reply: Final = httpx.Response(
200, content=json.dumps(_response(output, status)), headers={"content-type": "application/json"}
)
upstream_url, route_params = _route(model)
with respx.mock(assert_all_called=True) as router:
router.post(url__startswith=OPENAI_RESPONSES_URL if model.startswith("openai/") else AZURE_API_BASE).mock(
return_value=_sse(output, terminal_event) if stream else reply
)
router.post(url__startswith=upstream_url).mock(return_value=_sse(output, terminal_event) if stream else reply)
response: Final = litellm.completion(
model=model,
messages=[USER],
tools=list(TOOLS),
api_key="sk-test",
api_base=None if model.startswith("openai/") else AZURE_API_BASE,
api_version=None if model.startswith("openai/") else "2025-04-01-preview",
stream=stream,
response_format=response_format,
**route_params,
)
message: Final = (
(litellm.stream_chunk_builder(list(response), messages=[USER]) if stream else response).choices[0].message
@ -217,22 +234,22 @@ def _assistant_turn(
def _next_turn_input(
assistant: Mapping[str, object],
model: str = f"openai/responses/{MODEL}",
model: str = OPENAI_MODEL,
tools: Sequence[Mapping[str, object]] = TOOLS,
drop_params: bool = False,
) -> list[Mapping[str, object]]:
upstream_url, route_params = _route(model)
with respx.mock(assert_all_called=True) as router:
upstream: Final = router.post(
url__startswith=OPENAI_RESPONSES_URL if model.startswith("openai/") else AZURE_API_BASE
).mock(return_value=httpx.Response(200, json=_response((FOLLOW_UP,))))
upstream: Final = router.post(url__startswith=upstream_url).mock(
return_value=httpx.Response(200, json=_response((FOLLOW_UP,)))
)
litellm.completion(
model=model,
messages=[USER, dict(assistant), TOOL_RESULT],
tools=list(tools),
api_key="sk-test",
api_base=None if model.startswith("openai/") else AZURE_API_BASE,
api_version=None if model.startswith("openai/") else "2025-04-01-preview",
drop_params=drop_params,
**route_params,
)
return json.loads(upstream.calls[0].request.content)["input"]
@ -246,11 +263,12 @@ def _types(items: Sequence[Mapping[str, object]]) -> list[object]:
return [item.get("type") for item in items]
@pytest.mark.parametrize("model", [OPENAI_MODEL, AZURE_MODEL], ids=["openai", "azure"])
@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"])
def test_next_turn_replays_every_item_phase_and_boundary_once_in_order(stream: bool) -> None:
assistant: Final = _assistant_turn(OUTPUT, stream=stream)
def test_next_turn_replays_every_item_phase_and_boundary_once_in_order(stream: bool, model: str) -> None:
assistant: Final = _assistant_turn(OUTPUT, stream=stream, model=model)
sent: Final = _next_turn_input(assistant)
sent: Final = _next_turn_input(assistant, model=model)
assert assistant["content"] == "Checking the weather.Paris: light rain."
assert _replayed_fields(assistant)["phase"] == "final_answer"
@ -341,7 +359,7 @@ def test_an_unphased_answer_after_commentary_is_never_labeled_commentary(stream:
sent: Final = _next_turn_input(_edit_content(assistant))
assert "phase" not in _replayed_fields(assistant)
assert _replayed_fields(assistant).get("phase") is None
assert _types(sent[1:-1]) == ["message", "function_call"]
assert "phase" not in sent[1]
@ -372,7 +390,7 @@ def test_a_last_message_the_bridge_cannot_read_leaves_the_turn_without_a_phase(s
unreadable: Final = {**_with_text(FINAL_ANSWER, "It is raining in Paris."), "phase": "summary"}
assistant: Final = _assistant_turn((COMMENTARY, WEATHER_CALL, unreadable), stream=stream)
assert "phase" not in _replayed_fields(assistant)
assert _replayed_fields(assistant).get("phase") is None
def test_a_phase_openai_does_not_define_is_dropped_under_drop_params() -> None:
@ -393,7 +411,7 @@ def test_a_phase_openai_does_not_define_is_rejected_without_drop_params() -> Non
)
with pytest.raises(litellm.BadRequestError, match="'analysis'"):
litellm.completion(
model=f"openai/responses/{MODEL}",
model=OPENAI_MODEL,
messages=[USER, {**assistant, "phase": "analysis"}, TOOL_RESULT],
tools=list(TOOLS),
api_key="sk-test",
@ -486,28 +504,71 @@ def _chunks_before_error(stream: Iterator[object]) -> Iterator[object]:
return
def test_an_interrupted_stream_records_no_replay() -> None:
def _streamed_chunks(model: str, terminal_event: str | None = "response.completed") -> tuple[object, ...]:
upstream_url, route_params = _route(model)
with respx.mock(assert_all_called=True) as router:
router.post(OPENAI_RESPONSES_URL).mock(return_value=_sse(OUTPUT, terminal_event=None))
router.post(url__startswith=upstream_url).mock(return_value=_sse(OUTPUT, terminal_event))
stream: Final = litellm.completion(
model=f"openai/responses/{MODEL}", messages=[USER], tools=list(TOOLS), api_key="sk-test", stream=True
model=model, messages=[USER], tools=list(TOOLS), api_key="sk-test", stream=True, **route_params
)
received: Final = tuple(_chunks_before_error(iter(stream)))
return tuple(_chunks_before_error(iter(stream)))
async def _async_streamed_chunks(model: str) -> tuple[object, ...]:
upstream_url, route_params = _route(model)
with respx.mock(assert_all_called=True) as router:
router.post(url__startswith=upstream_url).mock(return_value=_sse(OUTPUT))
stream: Final = await litellm.acompletion(
model=model, messages=[USER], tools=list(TOOLS), api_key="sk-test", stream=True, **route_params
)
return tuple([chunk async for chunk in stream])
def test_an_interrupted_stream_records_no_replay() -> None:
received: Final = _streamed_chunks(OPENAI_MODEL, terminal_event=None)
assistant: Final = litellm.stream_chunk_builder(list(received), messages=[USER]).choices[0].message.model_dump()
assert assistant["content"] == "Checking the weather.Paris: light rain."
assert _replayed_fields(assistant) == {}
assert _replayed_fields(assistant) == {"phase": "final_answer"}
def _text_delta_phases(chunks: Sequence[object]) -> list[tuple[object, object]]:
deltas: Final = tuple(chunk.choices[0].delta for chunk in chunks if chunk.choices and chunk.choices[0].delta.content)
return [(delta.content, (delta.provider_specific_fields or {}).get("phase")) for delta in deltas]
@pytest.mark.parametrize(
("model", "commentary_phase", "answer_phase"),
[
(OPENAI_MODEL, "commentary", "final_answer"),
(AZURE_MODEL, "commentary", "final_answer"),
(XAI_MODEL, None, None),
],
ids=["openai", "azure", "xai"],
)
@pytest.mark.parametrize("use_async", [False, True], ids=["sync", "async"])
async def test_each_streamed_text_delta_carries_the_phase_of_its_message(
model: str, commentary_phase: str | None, answer_phase: str | None, use_async: bool, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
chunks: Final = await _async_streamed_chunks(model) if use_async else _streamed_chunks(model)
assert _text_delta_phases(chunks) == [
("Checking the weather.", commentary_phase),
("Paris: ", answer_phase),
("light rain.", answer_phase),
]
@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"])
def test_a_non_openai_route_records_no_phase_or_replay(stream: bool) -> None:
assistant: Final = _assistant_turn(OUTPUT, stream=stream, model=f"azure/responses/{MODEL}")
def test_a_route_without_replay_records_no_phase_or_replay(stream: bool) -> None:
assistant: Final = _assistant_turn(OUTPUT, stream=stream, model=XAI_MODEL)
assert _replayed_fields(assistant) == {}
def test_a_non_openai_route_ignores_a_recorded_replay() -> None:
sent: Final = _next_turn_input(_assistant_turn(OUTPUT, stream=False), model=f"azure/responses/{MODEL}")
def test_a_route_without_replay_ignores_a_recorded_replay() -> None:
sent: Final = _next_turn_input(_assistant_turn(OUTPUT, stream=False), model=XAI_MODEL)
assert _types(sent[1:-1]) == ["reasoning", "reasoning", "message", "function_call"]
assert all("phase" not in item for item in sent)
@ -520,12 +581,13 @@ LYON_CALL: Final = {**WEATHER_CALL, "id": "fc_lyon", "call_id": "call_weather_ly
@pytest.mark.parametrize(
("model", "expected_ids", "expected_text"),
[
(f"openai/responses/{MODEL}", ["rs_first", "rs_second"], "Paris is likely wet. Lyon is likely dry."),
(f"azure/responses/{MODEL}", ["rs_second"], "Lyon is likely dry."),
(OPENAI_MODEL, ["rs_first", "rs_second"], "Paris is likely wet. Lyon is likely dry."),
(AZURE_MODEL, ["rs_first", "rs_second"], "Paris is likely wet. Lyon is likely dry."),
(XAI_MODEL, ["rs_second"], "Lyon is likely dry."),
],
ids=["openai", "azure"],
ids=["openai", "azure", "xai"],
)
def test_only_the_openai_route_keeps_every_reasoning_item_of_a_tool_call_turn(
def test_only_replaying_routes_keep_every_reasoning_item_of_a_tool_call_turn(
model: str, expected_ids: list[str], expected_text: str
) -> None:
assistant: Final = _assistant_turn(
@ -540,10 +602,10 @@ def test_only_the_openai_route_keeps_every_reasoning_item_of_a_tool_call_turn(
@pytest.mark.parametrize(
("model", "expected_content"),
[(f"openai/responses/{MODEL}", None), (f"azure/responses/{MODEL}", "")],
ids=["openai", "azure"],
[(OPENAI_MODEL, None), (AZURE_MODEL, None), (XAI_MODEL, "")],
ids=["openai", "azure", "xai"],
)
def test_only_the_openai_route_ends_a_streamed_tool_call_turn_with_null_content(
def test_only_replaying_routes_end_a_streamed_tool_call_turn_with_null_content(
model: str, expected_content: str | None
) -> None:
assistant: Final = _assistant_turn((FIRST_REASONING, WEATHER_CALL), stream=True, model=model)
@ -603,7 +665,7 @@ def test_a_custom_tool_call_replays_with_its_custom_output(stream: bool) -> None
return_value=httpx.Response(200, json=_response((FOLLOW_UP,)))
)
litellm.completion(
model=f"openai/responses/{MODEL}",
model=OPENAI_MODEL,
messages=[USER, assistant, {"role": "tool", "tool_call_id": "call_patch_readme", "content": "applied"}],
tools=[{"type": "custom", "custom": {"name": "apply_patch"}}],
api_key="sk-test",