diff --git a/litellm/completion_extras/litellm_responses_transformation/handler.py b/litellm/completion_extras/litellm_responses_transformation/handler.py index 76d17ea9693..0ee9783a2b5 100644 --- a/litellm/completion_extras/litellm_responses_transformation/handler.py +++ b/litellm/completion_extras/litellm_responses_transformation/handler.py @@ -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, diff --git a/litellm/completion_extras/litellm_responses_transformation/history.py b/litellm/completion_extras/litellm_responses_transformation/history.py index 4435de64cde..06446e8efcb 100644 --- a/litellm/completion_extras/litellm_responses_transformation/history.py +++ b/litellm/completion_extras/litellm_responses_transformation/history.py @@ -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)) diff --git a/litellm/completion_extras/litellm_responses_transformation/transformation.py b/litellm/completion_extras/litellm_responses_transformation/transformation.py index 580e55273f6..e3d375f3b1c 100644 --- a/litellm/completion_extras/litellm_responses_transformation/transformation.py +++ b/litellm/completion_extras/litellm_responses_transformation/transformation.py @@ -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): diff --git a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_history.py b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_history.py index a2efbfef5dc..6f9b3fdf397 100644 --- a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_history.py +++ b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_history.py @@ -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",