diff --git a/litellm/completion_extras/litellm_responses_transformation/handler.py b/litellm/completion_extras/litellm_responses_transformation/handler.py index 8f12d8558808..15f5b28e30ed 100644 --- a/litellm/completion_extras/litellm_responses_transformation/handler.py +++ b/litellm/completion_extras/litellm_responses_transformation/handler.py @@ -209,7 +209,15 @@ def completion( json_mode=kwargs.get("json_mode"), ) elif isinstance(result, ModelResponse): - return result + if not stream: + return result + return self._completed_response_as_stream( + response=result, + model=model, + custom_llm_provider=custom_llm_provider, + logging_obj=logging_obj, + json_mode=kwargs.get("json_mode"), + ) elif not stream: responses_api_response = self._collect_response_from_stream(result) return self.transformation_handler.transform_response( @@ -299,7 +307,15 @@ async def acompletion(self, *args, **kwargs) -> Union["ModelResponse", "CustomSt json_mode=kwargs.get("json_mode"), ) elif isinstance(result, ModelResponse): - return result + if not stream: + return result + return self._completed_response_as_stream( + response=result, + model=model, + custom_llm_provider=custom_llm_provider, + logging_obj=logging_obj, + json_mode=kwargs.get("json_mode"), + ) elif not stream: responses_api_response = await self._collect_response_from_stream_async(result) return self.transformation_handler.transform_response( @@ -331,6 +347,25 @@ async def acompletion(self, *args, **kwargs) -> Union["ModelResponse", "CustomSt ) return self._apply_post_stream_processing(streamwrapper, model, custom_llm_provider) + def _completed_response_as_stream( + self, + response: "ModelResponse", + model: str, + custom_llm_provider: str, + logging_obj: "LiteLLMLoggingObj", + json_mode: bool | None, + ) -> "CustomStreamWrapper": + from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper + from litellm.llms.base_llm.base_model_iterator import MockResponseIterator + + streamwrapper = CustomStreamWrapper( + completion_stream=MockResponseIterator(model_response=response, json_mode=json_mode), + model=model, + custom_llm_provider=custom_llm_provider, + logging_obj=logging_obj, + ) + return self._apply_post_stream_processing(streamwrapper, model, custom_llm_provider) + @staticmethod def _apply_post_stream_processing( stream: "CustomStreamWrapper", diff --git a/litellm/completion_extras/litellm_responses_transformation/transformation.py b/litellm/completion_extras/litellm_responses_transformation/transformation.py index 3e50fe660395..89a44fcdeefb 100644 --- a/litellm/completion_extras/litellm_responses_transformation/transformation.py +++ b/litellm/completion_extras/litellm_responses_transformation/transformation.py @@ -1077,6 +1077,7 @@ def _map_responses_status_to_finish_reason(self, status: Optional[str]) -> str: class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator): def __init__(self, streaming_response, sync_stream: bool, json_mode: Optional[bool] = False): super().__init__(streaming_response, sync_stream, json_mode) + self._chat_completion_id: str | None = None def _handle_string_chunk( self, str_line: Union[str, "BaseModel"] @@ -1384,4 +1385,13 @@ def chunk_parser(self, chunk: dict) -> "ModelResponseStream": ModelResponseStream: OpenAI-formatted streaming chunk """ verbose_logger.debug(f"Chat provider: transform_streaming_response called with chunk: {chunk}") - return OpenAiResponsesToChatCompletionStreamIterator.translate_responses_chunk_to_openai_stream(chunk) + return self._with_stream_scoped_id( + OpenAiResponsesToChatCompletionStreamIterator.translate_responses_chunk_to_openai_stream(chunk) + ) + + def _with_stream_scoped_id(self, chunk: "ModelResponseStream") -> "ModelResponseStream": + if self._chat_completion_id is None: + self._chat_completion_id = chunk.id + else: + chunk.id = self._chat_completion_id + return chunk diff --git a/tests/e2e/coverage_registry/llm_conversational.yaml b/tests/e2e/coverage_registry/llm_conversational.yaml index 6cb60c192474..e8fc8067ee0d 100644 --- a/tests/e2e/coverage_registry/llm_conversational.yaml +++ b/tests/e2e/coverage_registry/llm_conversational.yaml @@ -5,6 +5,9 @@ - {id: llm.chat_completions.openai.passthrough.nonstream.cost_logged, module: llm, tier: P1, subject_endpoint: chat_completions, route: openai, capability: basic, streaming: nonstream, assertions: [works, cost_logged], source: "test_passthrough_e2e.py", rationale: "OpenAI-format chat via the raw /openai/{endpoint} passthrough (/openai/v1/chat/completions); proxy swaps in OPENAI_API_KEY and still logs a costed pass_through_endpoint row (LIT-4752)"} - {id: llm.chat_completions.openai.tool_use.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: openai, capability: tool_use, streaming: nonstream, assertions: [works], source: "model_prices json", rationale: "OpenAI function_calling; high usage"} - {id: llm.chat_completions.openai.tool_use.stream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: openai, capability: tool_use, streaming: stream, assertions: [works], source: "model_prices json", rationale: "Tool calls over streaming"} +- {id: llm.chat_completions.openai.basic.stream.bridge_shares_chunk_id, module: llm, tier: P0, subject_endpoint: chat_completions, route: openai, capability: basic, streaming: stream, assertions: [stable_chunk_id], source: "completion_extras/litellm_responses_transformation/transformation.py", rationale: "A responses-only model served over /chat/completions must stream every chunk under one chat completion id; per-chunk ids make id-accumulating SDKs drop the response", fail_before_fix: proven} +- {id: llm.chat_completions.openai.basic.stream.bridge_streams_sse, module: llm, tier: P0, subject_endpoint: chat_completions, route: openai, capability: basic, streaming: stream, assertions: [works], source: "completion_extras/litellm_responses_transformation/handler.py", rationale: "The Responses bridge must answer a streaming chat request with real SSE (content deltas, finish_reason, [DONE]), never a completed response the SSE generator cannot iterate"} +- {id: llm.chat_completions.openai.tool_use.stream.bridge_streams_tool_call, module: llm, tier: P0, subject_endpoint: chat_completions, route: openai, capability: tool_use, streaming: stream, assertions: [works], source: "completion_extras/litellm_responses_transformation/transformation.py", rationale: "Tool calls translated from Responses events must reassemble into one named call with parseable argument JSON over the bridged stream"} - {id: llm.chat_completions.openai.vision.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: openai, capability: vision, streaming: nonstream, assertions: [works], source: "model_prices json", rationale: "gpt-4o vision; high usage"} - {id: llm.chat_completions.openai.prompt_cache_5m.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: openai, capability: prompt_cache_5m, streaming: nonstream, assertions: [works], source: "model_prices json", rationale: "Prompt caching cost optimization"} - {id: llm.chat_completions.openai.service_tier.nonstream.works, module: llm, tier: P1, subject_endpoint: chat_completions, route: openai, capability: service_tier, streaming: nonstream, assertions: [works], source: "OpenAI service_tier param", rationale: "OpenAI scale-tier request option is forwarded and echoed"} diff --git a/tests/e2e/e2e_http.py b/tests/e2e/e2e_http.py index d16e84bd754d..74e57f86b885 100644 --- a/tests/e2e/e2e_http.py +++ b/tests/e2e/e2e_http.py @@ -137,6 +137,7 @@ class StreamingResponse(BaseModel): # quota) arrive as SSE error events inside an otherwise-successful response; # the consumed body is elided, so this is the only place they surface. stream_error: str | None = None + stream_done: bool = False @property def ok(self) -> bool: @@ -408,6 +409,7 @@ def _streaming_outcome(resp: requests.Response, stream: bool) -> StreamingRespon chunks = 0 stream_error: str | None = None stream_events: list[str] = [] + stream_done = False for line in lines: if not line: continue @@ -415,7 +417,9 @@ def _streaming_outcome(resp: requests.Response, stream: bool) -> StreamingRespon decoded_line = line.decode(errors="replace") if decoded_line.startswith("data: "): payload = decoded_line.removeprefix("data: ") - if payload != "[DONE]": + if payload == "[DONE]": + stream_done = True + else: stream_events.append(payload) if stream_error is None and ( line.startswith(b"event: error") @@ -433,6 +437,7 @@ def _streaming_outcome(resp: requests.Response, stream: bool) -> StreamingRespon body="", chunks=chunks, stream_events=stream_events, + stream_done=stream_done, stream_error=stream_error, ) diff --git a/tests/e2e/llm_translation/test_responses_bridge_streaming_e2e.py b/tests/e2e/llm_translation/test_responses_bridge_streaming_e2e.py new file mode 100644 index 000000000000..9a45743a0cd2 --- /dev/null +++ b/tests/e2e/llm_translation/test_responses_bridge_streaming_e2e.py @@ -0,0 +1,173 @@ +"""Live /chat/completions streaming through the Responses API bridge. + +Responses-only models (gpt-5.3-codex here, the same shape as the GPT-5.6 models +customers reach over bedrock_mantle) cannot serve /chat/completions natively, so the +proxy translates the request to /v1/responses and translates each Responses event back +into a chat completion chunk. Two customer-visible contracts only hold on that path: + +- every chunk of one stream carries the same ``id`` (#32854). The bridge builds a chunk + per Responses event, so a regression there hands each chunk a fresh ``chatcmpl-`` + and SDKs that accumulate by id (openai-go's ChatCompletionAccumulator) silently drop + everything after the first chunk while the HTTP response still looks healthy +- the bridge always answers a streaming request with a real SSE stream (#33154). When it + hands back an already-completed response instead, the proxy's SSE generator dies with + "'async for' requires an object with __aiter__ method" mid-stream +""" + +from __future__ import annotations + +import pytest +from pydantic import BaseModel + +from e2e_config import unique_marker +from e2e_http import StreamingResponse +from lifecycle import ResourceManager +from models import ChatBody, ChatMessage, ChatTool, ChatToolFunction, LiteLLMParamsBody +from passthrough_client import PassthroughClient + +pytestmark = pytest.mark.e2e + +RESPONSES_ONLY_BACKEND = "openai/gpt-5.3-codex" + + +class _BridgeToolCallFunction(BaseModel): + name: str | None = None + arguments: str | None = None + + +class _BridgeToolCall(BaseModel): + function: _BridgeToolCallFunction = _BridgeToolCallFunction() + + +class _BridgeDelta(BaseModel): + content: str | None = None + tool_calls: list[_BridgeToolCall] | None = None + + +class _BridgeChoice(BaseModel): + delta: _BridgeDelta = _BridgeDelta() + finish_reason: str | None = None + + +class _BridgeChunk(BaseModel): + id: str + choices: list[_BridgeChoice] = [] + + +class _WeatherArgs(BaseModel): + location: str + + +_WEATHER_TOOL = ChatTool( + function=ChatToolFunction( + name="get_weather", + description="Get the current weather for a location", + parameters={ + "type": "object", + "properties": {"location": {"type": "string"}}, + "required": ["location"], + }, + ) +) + + +def _bridge_chunks(result: StreamingResponse) -> list[_BridgeChunk]: + """Parse the SSE events of a bridged stream, failing loudly on a stream that never + established, carried an error event, or delivered no chunks.""" + assert result.ok and result.is_streaming, f"bridged stream was not established: {result}" + assert result.stream_error is None, f"bridged stream carried an error event: {result.stream_error}" + chunks = [_BridgeChunk.model_validate_json(event) for event in result.stream_events] + assert chunks, f"bridged stream delivered no chunks: {result.body[:500]}" + return chunks + + +class TestResponsesBridgeChatCompletionsStreaming: + @pytest.fixture + def bridged_model(self, client: PassthroughClient, resources: ResourceManager) -> str: + model = f"e2e-bridge-stream-{unique_marker()}" + model_id = client.proxy.create_model( + model, + LiteLLMParamsBody(model=RESPONSES_ONLY_BACKEND, api_key="os.environ/OPENAI_API_KEY"), + ) + resources.defer(lambda: client.proxy.delete_model(model_id)) + return model + + @pytest.mark.covers( + "llm.chat_completions.openai.basic.stream.bridge_shares_chunk_id", + exercised_on=["chat_completions"], + ) + def test_bridged_stream_shares_one_chunk_id( + self, client: PassthroughClient, resources: ResourceManager, bridged_model: str + ) -> None: + result = client.proxy.chat_stream( + resources.key(), + ChatBody( + model=bridged_model, + messages=[ChatMessage(role="user", content=f"Count from 1 to 5, one number per line. {unique_marker()}")], + max_tokens=64, + stream=True, + ), + ) + + chunks = _bridge_chunks(result) + ids = {chunk.id for chunk in chunks} + assert len(ids) == 1, f"bridged stream used {len(ids)} different chunk ids: {sorted(ids)[:5]}" + assert ids.pop().startswith("chatcmpl-"), f"bridged chunk id is not chat-completion shaped: {chunks[0].id}" + + @pytest.mark.covers( + "llm.chat_completions.openai.basic.stream.bridge_streams_sse", + exercised_on=["chat_completions"], + ) + def test_bridged_stream_delivers_content_finish_reason_and_done( + self, client: PassthroughClient, resources: ResourceManager, bridged_model: str + ) -> None: + result = client.proxy.chat_stream( + resources.key(), + ChatBody( + model=bridged_model, + messages=[ChatMessage(role="user", content=f"Reply with the single word pong. {unique_marker()}")], + max_tokens=32, + stream=True, + ), + ) + + chunks = _bridge_chunks(result) + content = "".join(choice.delta.content or "" for chunk in chunks for choice in chunk.choices) + assert content.strip(), f"bridged stream completed with no content deltas: {result.stream_events[:3]}" + assert any( + choice.finish_reason for chunk in chunks for choice in chunk.choices + ), f"bridged stream never emitted a finish_reason: {result.stream_events[-3:]}" + assert result.stream_done, f"bridged stream did not terminate with [DONE]: {result.stream_events[-2:]}" + + @pytest.mark.covers( + "llm.chat_completions.openai.tool_use.stream.bridge_streams_tool_call", + exercised_on=["chat_completions"], + ) + def test_bridged_stream_reassembles_tool_call( + self, client: PassthroughClient, resources: ResourceManager, bridged_model: str + ) -> None: + result = client.proxy.chat_stream( + resources.key(), + ChatBody( + model=bridged_model, + messages=[ + ChatMessage( + role="user", + content="What is the weather in San Francisco? Use the get_weather tool.", + ) + ], + tools=[_WEATHER_TOOL], + tool_choice="required", + max_tokens=256, + stream=True, + ), + ) + + chunks = _bridge_chunks(result) + calls = [call for chunk in chunks for choice in chunk.choices for call in (choice.delta.tool_calls or [])] + assert calls, f"bridged stream returned no tool call for a tool-forced prompt: {result.stream_events[:5]}" + name = "".join(call.function.name or "" for call in calls) + arguments = "".join(call.function.arguments or "" for call in calls) + assert name == "get_weather", f"bridged stream streamed the wrong tool name: {name!r}" + args = _WeatherArgs.model_validate_json(arguments) + assert args.location.strip(), f"bridged tool call arguments missing location: {arguments!r}" diff --git a/tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py b/tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py index a5bc01c2b74c..8ecb7f4c6f09 100644 --- a/tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py +++ b/tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py @@ -203,3 +203,65 @@ async def test_acompletion_preserves_top_level_stream_flag_in_responses_request( assert result is stream assert transform_request.call_args.kwargs["optional_params"]["stream"] is True + + +def _completed_chat_response() -> ModelResponse: + return ModelResponse( + id="chatcmpl-completed", + model="gpt-5.4", + choices=[ + { + "index": 0, + "message": {"role": "assistant", "content": "pong"}, + "finish_reason": "stop", + } + ], + ) + + +@pytest.mark.asyncio +async def test_acompletion_streams_completed_model_response(): + """A streaming request whose bridge call comes back already completed must still be + handed back as an async-iterable stream. Returning the bare ModelResponse crashed the + proxy's SSE generator with "'async for' requires an object with __aiter__ method". + Regression for #33154.""" + completed = _completed_chat_response() + bridge = ResponsesToCompletionBridgeHandler() + + with ( + patch.object( + bridge.transformation_handler, + "transform_request", + return_value={"model": "gpt-5.4", "input": "hi"}, + ), + patch("litellm.aresponses", new=AsyncMock(return_value=completed)), + ): + result = await bridge.acompletion(**_bridge_kwargs(stream=True)) + + assert isinstance(result, CustomStreamWrapper), f"streaming request got {type(result)}" + chunks = [chunk async for chunk in result] + assert "".join( + chunk.choices[0].delta.content or "" for chunk in chunks + ) == "pong", f"completed response did not stream its content: {chunks}" + assert [c for c in chunks if c.choices[0].finish_reason], "stream never emitted a finish_reason" + + +def test_completion_streams_completed_model_response(): + completed = _completed_chat_response() + bridge = ResponsesToCompletionBridgeHandler() + + with ( + patch.object( + bridge.transformation_handler, + "transform_request", + return_value={"model": "gpt-5.4", "input": "hi"}, + ), + patch("litellm.responses", return_value=completed), + ): + result = bridge.completion(**_bridge_kwargs(stream=True)) + + assert isinstance(result, CustomStreamWrapper), f"streaming request got {type(result)}" + chunks = list(result) + assert "".join(chunk.choices[0].delta.content or "" for chunk in chunks) == "pong", ( + f"completed response did not stream its content: {chunks}" + ) diff --git a/tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py b/tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py index 6907e4d0d02e..a111b932f2c1 100644 --- a/tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py +++ b/tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py @@ -2855,6 +2855,41 @@ def stream_tool_id(item_id, call_id): assert stream_tool_id("fc_2", "call_tokyo") == "call_tokyo" +def test_streaming_chunks_share_one_chat_completion_id(): + """Every chunk of one streamed chat completion must carry the same ``id``, per the + OpenAI spec. The bridge builds a fresh ``ModelResponseStream`` per Responses event, + so without a stream-scoped id each chunk got a new ``chatcmpl-`` and clients + that validate id consistency (openai-go's ChatCompletionAccumulator) silently + dropped every chunk after the first. Regression for #32854.""" + from litellm.completion_extras.litellm_responses_transformation.transformation import ( + OpenAiResponsesToChatCompletionStreamIterator, + ) + + iterator = OpenAiResponsesToChatCompletionStreamIterator( + streaming_response=None, sync_stream=True + ) + events = [ + {"type": "response.created", "response": {"id": "resp_abc", "output": []}}, + {"type": "response.output_text.delta", "delta": "Hel"}, + {"type": "response.output_text.delta", "delta": "lo"}, + { + "type": "response.completed", + "response": {"id": "resp_abc", "output": [{"type": "message"}]}, + }, + ] + + ids = [iterator.chunk_parser(event).id for event in events] + + assert len(set(ids)) == 1, f"streamed chunks carried different ids: {ids}" + assert ids[0], "streamed chunks carried an empty id" + + other_stream = OpenAiResponsesToChatCompletionStreamIterator( + streaming_response=None, sync_stream=True + ) + assert ( + other_stream.chunk_parser(events[1]).id != ids[0] + ), "a separate stream must get its own id, not a process-wide one" + @pytest.mark.asyncio @pytest.mark.parametrize(