diff --git a/litellm/llms/openai/responses/guardrail_translation/handler.py b/litellm/llms/openai/responses/guardrail_translation/handler.py index 84ff790ae48..23ab84fee61 100644 --- a/litellm/llms/openai/responses/guardrail_translation/handler.py +++ b/litellm/llms/openai/responses/guardrail_translation/handler.py @@ -1767,7 +1767,7 @@ class OpenAIResponsesHandler(BaseTranslation): "annotations": (), "logprobs": None, } - return ( + events: Final = ( *_open_item_closing_events(responses_so_far), OutputItemAddedEvent( type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, @@ -1812,6 +1812,10 @@ class OpenAIResponsesHandler(BaseTranslation): response=_blocked_response(exc, response_id=response_id, model=model, output_item=item), ), ) + return tuple( + event.model_copy(update={"sequence_number": number}) + for number, event in enumerate(events, start=_next_stream_sequence_number(responses_so_far)) + ) class _BlockedContentPart(TypedDict): diff --git a/litellm/proxy/guardrails/guardrail_hooks/llm_shield_proxy/stream_restorers.py b/litellm/proxy/guardrails/guardrail_hooks/llm_shield_proxy/stream_restorers.py index 4ef8bbdbb3e..17076c81f47 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/llm_shield_proxy/stream_restorers.py +++ b/litellm/proxy/guardrails/guardrail_hooks/llm_shield_proxy/stream_restorers.py @@ -247,7 +247,8 @@ class ResponsesStreamRestorer: own window, keyed by the family, the item id and the part index. - Any `*.done` event closes the stream of the same family. Whatever its window still holds goes out first, as a copy of that stream's last delta event -- so it carries - the stream's own ids, and repeats that event's `sequence_number`. Then every text + the stream's own ids. It takes the done event's `sequence_number`, and every later + event's number moves up by one, so the numbers stay unique and in order. Then every text field on the done event is restored in full: its string fields other than identifiers, plus any `part` or `item` it repeats. - `response.completed` / `response.incomplete` repeat the whole reply, and are @@ -259,6 +260,8 @@ class ResponsesStreamRestorer: self._rehydrate: Final = rehydrate self._carries: Final[dict[ResponsesStreamKey, str]] = {} # mutable-ok: per-stream windows advanced in place. self._last_deltas: Final[dict[ResponsesStreamKey, object]] = {} # mutable-ok: newest delta per stream. + self._inserted: int = 0 + self._last_sequence_number: int | None = None async def restore(self, event: object) -> tuple[object, ...]: """The events to emit in place of `event`: any flush, then the event itself.""" @@ -267,7 +270,7 @@ class ResponsesStreamRestorer: return (event,) if kind.endswith(".delta") and kind not in RESPONSES_BINARY_DELTAS: await self._restore_delta(event, kind) - return (event,) + return self._numbered((), event) slots: Final[SlotSink] = [] flushed: Final = await self._flush(responses_stream_key(event, kind)) if kind.endswith(".done") else () if kind.endswith(".done"): @@ -280,12 +283,29 @@ class ResponsesStreamRestorer: for item in read_list(read_field(event, "response"), "output"): collect_response_item(item, slots) await rehydrate_slots(slots, self._rehydrate) - return (*flushed, event) + return self._numbered(flushed, event) async def finish(self) -> tuple[object, ...]: """Flushes every stream the provider never closed, e.g. a truncated reply.""" - flushed: Final = tuple([await self._flush(key) for key in tuple(self._carries)]) - return tuple(itertools.chain.from_iterable(flushed)) + flushed: Final = tuple(itertools.chain.from_iterable([await self._flush(key) for key in tuple(self._carries)])) + last: Final = self._last_sequence_number + if last is not None: + for offset, flush in enumerate(flushed, start=1): + write_field(flush, "sequence_number", last + offset) + return flushed + + def _numbered(self, flushed: tuple[object, ...], event: object) -> tuple[object, ...]: + """Gives each flush the slot `event` arrived in and moves `event` and later events up.""" + number: Final = read_field(event, "sequence_number") + if not isinstance(number, int): + return (*flushed, event) + for offset, flush in enumerate(flushed): + write_field(flush, "sequence_number", number + self._inserted + offset) + self._inserted += len(flushed) + self._last_sequence_number = number + self._inserted + if self._inserted: + write_field(event, "sequence_number", self._last_sequence_number) + return (*flushed, event) async def _restore_delta(self, event: object, kind: str) -> None: text: Final = read_field(event, "delta") diff --git a/litellm/responses/litellm_completion_transformation/streaming_iterator.py b/litellm/responses/litellm_completion_transformation/streaming_iterator.py index 17bd2d33b08..39a7f06a85a 100644 --- a/litellm/responses/litellm_completion_transformation/streaming_iterator.py +++ b/litellm/responses/litellm_completion_transformation/streaming_iterator.py @@ -22,7 +22,11 @@ from litellm.responses.litellm_completion_transformation.tool_search import ( from litellm.responses.litellm_completion_transformation.transformation import ( LiteLLMCompletionResponsesConfig, ) -from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator, stamp_responses_usage_cost +from litellm.responses.streaming_iterator import ( + SEQUENCE_NUMBER_STAMPED_ON_EMISSION, + ResponsesAPIStreamingIterator, + stamp_responses_usage_cost, +) from litellm.responses.utils import ResponsesAPIRequestUtils from litellm.types.llms.openai import ( PART_UNION_TYPES, @@ -181,14 +185,13 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): return True def _queue_output_item_added(self, output_index: int, item: BaseLiteLLMOpenAIResponseObject) -> None: - self._sequence_number += 1 - added: Final = OutputItemAddedEvent( - type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, - output_index=output_index, - item=item, + self._pending_tool_events.append( + OutputItemAddedEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, + output_index=output_index, + item=item, + ) ) - added.__dict__["sequence_number"] = self._sequence_number - self._pending_tool_events.append(added) def _queue_tool_search_call_added(self, call_id: str, output_index: int, args_delta: str) -> None: if call_id not in self._tool_args_by_call_id: @@ -203,12 +206,10 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): item: Final = build_tool_search_call( call_id, final_args or self._tool_args_by_call_id.get(call_id, ""), "completed" ) - self._sequence_number += 1 self._pending_tool_events.append( OutputItemDoneEvent( type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, output_index=output_index, - sequence_number=self._sequence_number, item=BaseLiteLLMOpenAIResponseObject(**item.model_dump()), ) ) @@ -374,15 +375,12 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): chunk_size = 10 # Match typical OpenAI delta size for i in range(0, len(fn_args_delta), chunk_size): delta_chunk = fn_args_delta[i : i + chunk_size] - self._sequence_number += 1 delta_event: BaseLiteLLMOpenAIResponseObject = FunctionCallArgumentsDeltaEvent( type=ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA, item_id=self._tool_item_id_by_call_id.get(call_id, call_id), output_index=output_index, delta=delta_chunk, ) - # Add sequence_number as extra field (BaseLiteLLMOpenAIResponseObject allows extra fields) - delta_event.__dict__["sequence_number"] = self._sequence_number self._pending_tool_events.append(delta_event) def _queue_final_tool_call_done_events(self, litellm_complete_object: ModelResponse) -> None: @@ -450,33 +448,27 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): chunk_size = 10 # Match typical OpenAI delta size for i in range(0, len(remaining_args), chunk_size): delta_chunk = remaining_args[i : i + chunk_size] - self._sequence_number += 1 delta_event = FunctionCallArgumentsDeltaEvent( type=ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA, item_id=self._tool_item_id_by_call_id.get(call_id, call_id), output_index=output_index, delta=delta_chunk, ) - delta_event.__dict__["sequence_number"] = self._sequence_number self._pending_tool_events.append(delta_event) - self._sequence_number += 1 done_event = FunctionCallArgumentsDoneEvent( type=ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE, item_id=self._tool_item_id_by_call_id.get(call_id, call_id), output_index=output_index, arguments=final_args, ) - done_event.__dict__["sequence_number"] = self._sequence_number self._pending_tool_events.append(done_event) - self._sequence_number += 1 item_kwargs = self._tool_call_item_kwargs(call_id, fn_name, final_args, "completed") item_kwargs["id"] = self._tool_item_id_by_call_id.setdefault(call_id, item_kwargs["id"]) item_done_event = OutputItemDoneEvent( type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, output_index=output_index, - sequence_number=self._sequence_number, item=BaseLiteLLMOpenAIResponseObject(**item_kwargs), ) self._pending_tool_events.append(item_done_event) @@ -506,16 +498,11 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): (ResponsesAPIStreamEvents.WEB_SEARCH_CALL_SEARCHING, WebSearchCallSearchingEvent), (ResponsesAPIStreamEvents.WEB_SEARCH_CALL_COMPLETED, WebSearchCallCompletedEvent), ): - self._sequence_number += 1 - event = event_class(type=event_type, output_index=output_index, item_id=item.id) - event.__dict__["sequence_number"] = self._sequence_number - self._pending_tool_events.append(event) - self._sequence_number += 1 + self._pending_tool_events.append(event_class(type=event_type, output_index=output_index, item_id=item.id)) self._pending_tool_events.append( OutputItemDoneEvent( type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, output_index=output_index, - sequence_number=self._sequence_number, item=BaseLiteLLMOpenAIResponseObject(**item.model_dump()), ) ) @@ -625,31 +612,24 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): """ response_created_event_data: Final = self._default_response_created_event_data() - self._sequence_number += 1 - event: Final = ResponseCreatedEvent( + return ResponseCreatedEvent( type=ResponsesAPIStreamEvents.RESPONSE_CREATED, response=self._with_encoded_response_id(ResponsesAPIResponse(**response_created_event_data)), ) - event.__dict__["sequence_number"] = self._sequence_number - return event def create_response_in_progress_event(self) -> ResponseInProgressEvent: response_in_progress_event_data: Final = self._default_response_created_event_data() response_in_progress_event_data["status"] = "in_progress" - self._sequence_number += 1 - event: Final = ResponseInProgressEvent( + return ResponseInProgressEvent( type=ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS, response=self._with_encoded_response_id(ResponsesAPIResponse(**response_in_progress_event_data)), ) - event.__dict__["sequence_number"] = self._sequence_number - return event def create_output_item_added_event(self) -> OutputItemAddedEvent: if self._cached_item_id is None: self._cached_item_id = f"msg_{uuid.uuid4()}" - self._sequence_number += 1 - event: Final = OutputItemAddedEvent( + return OutputItemAddedEvent( type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, output_index=self._message_index(), item=BaseLiteLLMOpenAIResponseObject( @@ -662,30 +642,24 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): } ), ) - event.__dict__["sequence_number"] = self._sequence_number - return event def create_content_part_added_event(self) -> ContentPartAddedEvent: if self._cached_item_id is None: self._cached_item_id = f"msg_{uuid.uuid4()}" - self._sequence_number += 1 - event: Final = ContentPartAddedEvent( + return ContentPartAddedEvent( type=ResponsesAPIStreamEvents.CONTENT_PART_ADDED, item_id=self._cached_item_id, output_index=self._message_index(), content_index=0, part=BaseLiteLLMOpenAIResponseObject(**{"type": "output_text", "text": "", "annotations": []}), ) - event.__dict__["sequence_number"] = self._sequence_number - return event def _queue_message_item_added_events(self) -> None: if self._cached_item_id is None: self._cached_item_id = f"msg_{uuid.uuid4()}" self.sent_message_item_added_event = True self.sent_content_part_added_event = True - self._sequence_number += 1 event: Final = OutputItemAddedEvent( type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, output_index=self._message_index(), @@ -699,7 +673,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): } ), ) - event.__dict__["sequence_number"] = self._sequence_number self._pending_response_events.append(event) self._pending_response_events.append(self.create_content_part_added_event()) @@ -755,7 +728,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self, reasoning_item_id: str, reasoning_content: str, - sequence_number: int, ) -> ReasoningSummaryTextDoneEvent: """ Create response.reasoning_summary_text.done event. @@ -765,7 +737,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): "type": "response.reasoning_summary_text.done", "item_id": "rs_0c5dae30e53172980069708ba2f59c8197b71ca9820edad07c", "output_index": 0, - "sequence_number": 97, "summary_index": 0, "text": "**Clarifying the first humans**\n\nThe I'm addressing the user's specific interest." } @@ -774,7 +745,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): type=ResponsesAPIStreamEvents.REASONING_SUMMARY_TEXT_DONE, item_id=reasoning_item_id, output_index=self._reasoning_index(), - sequence_number=sequence_number, + sequence_number=SEQUENCE_NUMBER_STAMPED_ON_EMISSION, summary_index=0, text=reasoning_content, ) @@ -783,7 +754,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self, reasoning_item_id: str, reasoning_content: str, - sequence_number: int, ) -> ReasoningSummaryPartDoneEvent: """ Create response.reasoning_summary_part.done event. @@ -797,7 +767,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): "type": "summary_text", "text": "**Clarifying the first humans**\n\nThe earlier hominins. It feels important to ensure I'm addressing the user's specific interest." }, - "sequence_number": 98, "summary_index": 0 } """ @@ -805,7 +774,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): type=ResponsesAPIStreamEvents.REASONING_SUMMARY_PART_DONE, item_id=reasoning_item_id, output_index=self._reasoning_index(), - sequence_number=sequence_number, + sequence_number=SEQUENCE_NUMBER_STAMPED_ON_EMISSION, summary_index=0, part=BaseLiteLLMOpenAIResponseObject( **{ @@ -869,7 +838,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): return OutputItemDoneEvent( type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, output_index=self._message_index(), - sequence_number=1, item=BaseLiteLLMOpenAIResponseObject( **{ "id": self._cached_item_id, @@ -891,7 +859,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self, reasoning_item_id: str, reasoning_content: str, - sequence_number: int, ) -> OutputItemDoneEvent: """ Create response.output_item.done event for reasoning items. @@ -900,7 +867,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): { "type": "response.output_item.done", "output_index": 0, - "sequence_number": 99, "item": { "id": "rs_0c5dae30e53172980069708ba2f59c8197b71ca9820edad07c", "type": "reasoning", @@ -916,7 +882,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): return OutputItemDoneEvent( type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, output_index=self._reasoning_index(), - sequence_number=sequence_number, item=BaseLiteLLMOpenAIResponseObject( **{ "id": reasoning_item_id, @@ -1016,7 +981,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): if chunk.choices[0].finish_reason is None and not _delta_carries_output(delta): return - self._sequence_number += 1 self.sent_output_item_added_event = True # Reasoning-first @@ -1026,20 +990,20 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._cached_reasoning_item_id = mint_reasoning_item_id() self._reasoning_item_id = self._cached_reasoning_item_id - event = OutputItemAddedEvent( - type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, - output_index=self._reasoning_index(), - item=BaseLiteLLMOpenAIResponseObject( - **{ - "id": self._cached_reasoning_item_id, - "type": "reasoning", - "status": "in_progress", - "summary": None, - } - ), + self._pending_response_events.append( + OutputItemAddedEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, + output_index=self._reasoning_index(), + item=BaseLiteLLMOpenAIResponseObject( + **{ + "id": self._cached_reasoning_item_id, + "type": "reasoning", + "status": "in_progress", + "summary": None, + } + ), + ) ) - event.__dict__["sequence_number"] = self._sequence_number - self._pending_response_events.append(event) return # Tool-first @@ -1052,8 +1016,18 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._queue_message_item_added_events() return + def _stamp_sequence_number(self, event: BaseLiteLLMOpenAIResponseObject) -> BaseLiteLLMOpenAIResponseObject: + stamped: Final = event.model_copy(update={"sequence_number": self._sequence_number}) + self._sequence_number += 1 + return stamped + async def __anext__( self, + ) -> ResponsesAPIStreamingResponse | ResponseCompletedEvent | BaseLiteLLMOpenAIResponseObject: + return self._stamp_sequence_number(await self._anext_unstamped_event()) + + async def _anext_unstamped_event( + self, ) -> ResponsesAPIStreamingResponse | ResponseCompletedEvent | BaseLiteLLMOpenAIResponseObject: try: while True: @@ -1112,27 +1086,17 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): ) reasoning_item_id = self._cached_reasoning_item_id - # Create text.done event first with its own sequence number - self._sequence_number += 1 text_done_event = self.create_reasoning_summary_text_done_event( reasoning_item_id=reasoning_item_id, reasoning_content=reasoning_content, - sequence_number=self._sequence_number, ) - - # Create part.done event second with its own sequence number - self._sequence_number += 1 part_done_event = self.create_reasoning_summary_part_done_event( reasoning_item_id=reasoning_item_id, reasoning_content=reasoning_content, - sequence_number=self._sequence_number, ) - - self._sequence_number += 1 reasoning_output_item_done_event = self.create_reasoning_output_item_done_event( reasoning_item_id=reasoning_item_id, reasoning_content=reasoning_content, - sequence_number=self._sequence_number, ) self._pending_response_events.extend( [ @@ -1164,6 +1128,11 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): def __next__( self, + ) -> ResponsesAPIStreamingResponse | ResponseCompletedEvent | BaseLiteLLMOpenAIResponseObject: + return self._stamp_sequence_number(self._next_unstamped_event()) + + def _next_unstamped_event( + self, ) -> ResponsesAPIStreamingResponse | ResponseCompletedEvent | BaseLiteLLMOpenAIResponseObject: try: while True: @@ -1281,16 +1250,13 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): if delta_content: if not self.sent_message_item_added_event: self._queue_message_item_added_events() - self._sequence_number += 1 - text_delta_event: Final = OutputTextDeltaEvent( + return OutputTextDeltaEvent( type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, item_id=item_id, output_index=self._message_index(), content_index=0, delta=delta_content, ) - text_delta_event.__dict__["sequence_number"] = self._sequence_number - return text_delta_event # Priority 3: Handle tool call deltas (if any) -> queue events and emit them # For each tool call delta, we emit events one at a time to match OpenAI's streaming behavior diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index cb745384403..7a6685ae824 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -314,6 +314,7 @@ def _mid_stream_fallback_eligible(mapped_exception: Exception) -> bool: PRE_OUTPUT_LIFECYCLE_EVENT_TYPES: Final = frozenset({"response.created", "response.in_progress", "response.queued"}) +SEQUENCE_NUMBER_STAMPED_ON_EMISSION: Final = -1 class BaseResponsesAPIStreamingIterator: @@ -1739,7 +1740,6 @@ def build_synthetic_response_events( _build_response_status_event(openai_types.ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS, transformed), ] - sequence_number = 0 output_items: Final[Sequence[object]] = getattr(transformed, "output", []) or [] for output_index, output_item in enumerate(output_items): output_item_payload = _dump_response_object(output_item) @@ -1849,35 +1849,31 @@ def build_synthetic_response_events( delta=summary_text[i : i + chunk_size], ) ) - sequence_number += 1 events.append( openai_types.ReasoningSummaryTextDoneEvent( type=openai_types.ResponsesAPIStreamEvents.REASONING_SUMMARY_TEXT_DONE, item_id=item_id, output_index=output_index, - sequence_number=sequence_number, + sequence_number=SEQUENCE_NUMBER_STAMPED_ON_EMISSION, summary_index=summary_index, text=summary_text, ) ) - sequence_number += 1 events.append( openai_types.ReasoningSummaryPartDoneEvent( type=openai_types.ResponsesAPIStreamEvents.REASONING_SUMMARY_PART_DONE, item_id=item_id, output_index=output_index, - sequence_number=sequence_number, + sequence_number=SEQUENCE_NUMBER_STAMPED_ON_EMISSION, summary_index=summary_index, part=openai_types.BaseLiteLLMOpenAIResponseObject(**summary_payload), ) ) - sequence_number += 1 events.append( openai_types.OutputItemDoneEvent( type=openai_types.ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, output_index=output_index, - sequence_number=sequence_number, item=openai_types.BaseLiteLLMOpenAIResponseObject(**output_item_payload), ) ) @@ -1888,7 +1884,7 @@ def build_synthetic_response_events( response=transformed, ) ) - return events + return [event.model_copy(update={"sequence_number": number}) for number, event in enumerate(events)] # --------------------------------------------------------------------------- diff --git a/litellm/router.py b/litellm/router.py index cc44161a5cf..41aa568ab4a 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -424,6 +424,7 @@ if TYPE_CHECKING: ResponseIncompleteEvent, ResponseInputParam, ResponsesAPIResponse, + ResponsesAPIStreamingResponse, ) Span = _Span @@ -3652,12 +3653,13 @@ class Router: fallback_headers_adopted: bool = False - def __init__(self, async_generator: AsyncGenerator): + def __init__(self, async_generator: "AsyncGenerator[ResponsesAPIStreamingResponse, None]"): import time from datetime import datetime self._async_generator = async_generator self._billing_source: object = source_iterator + self._last_sequence_number = -1 # Mirror every attribute BaseResponsesAPIStreamingIterator.__init__ # would have set. The wrapper bypasses super().__init__ (it has no # httpx.Response of its own and no provider config to drive), so @@ -3759,7 +3761,16 @@ class Router: # the very key that created the container (#30210). if self.completed_response is None and getattr(chunk, "type", None) in _RESPONSES_TERMINAL_EVENT_TYPES: self.completed_response = chunk - return chunk + sequence_number: Final = getattr(chunk, "sequence_number", None) + if not isinstance(sequence_number, int): + return chunk + if not self.fallback_headers_adopted: + self._last_sequence_number = max(self._last_sequence_number, sequence_number) + return chunk + self._last_sequence_number = max(self._last_sequence_number + 1, sequence_number) + if self._last_sequence_number == sequence_number: + return chunk + return chunk.model_copy(update={"sequence_number": self._last_sequence_number}) async def aclose(self): # async generators always expose aclose — no defensive check needed. @@ -3803,7 +3814,9 @@ class Router: exc, ) - wrapper: Final = FallbackResponsesStreamWrapper(stream_with_fallbacks()) + wrapper: Final = FallbackResponsesStreamWrapper( + stream_with_fallbacks() # pyright: ignore[reportArgumentType] # relays response events typed as object + ) return wrapper async def _aresponses_fallback_attempt( diff --git a/litellm/types/llms/openai.py b/litellm/types/llms/openai.py index 8cc9e912e80..f42e99b0a58 100644 --- a/litellm/types/llms/openai.py +++ b/litellm/types/llms/openai.py @@ -1,7 +1,8 @@ import builtins -from collections.abc import Iterable, Mapping +from collections.abc import ItemsView, Iterable, Mapping from enum import Enum from os import PathLike +from types import MappingProxyType from typing import IO, Any, Final, Generic, Literal, Optional, TypeAlias, Union import httpx @@ -1672,17 +1673,40 @@ class ResponsesAPIStreamEvents(str, Enum): ERROR = "error" -class ResponseCreatedEvent(BaseLiteLLMOpenAIResponseObject): +_NO_EXTRAS: Final[Mapping[str, object]] = MappingProxyType({}) + + +class BaseResponsesStreamEvent(BaseLiteLLMOpenAIResponseObject): # frozen-ok: guardrails rewrite events in place + def __getitem__(self, key: str) -> object: + return self._mapping_holding(key)[key] + + def get(self, key: str, default: object = None) -> object: + return self._mapping_holding(key).get(key, default) + + def __contains__(self, key: object) -> bool: + return key in self.__dict__ or key in self._extras() + + def items(self) -> ItemsView[str, object]: + return MappingProxyType({**self.__dict__, **self._extras()}).items() + + def _extras(self) -> Mapping[str, object]: + return self.__pydantic_extra__ or _NO_EXTRAS + + def _mapping_holding(self, key: str) -> Mapping[str, object]: + return self.__dict__ if key in self.__dict__ else self._extras() + + +class ResponseCreatedEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.RESPONSE_CREATED] response: ResponsesAPIResponse -class ResponseInProgressEvent(BaseLiteLLMOpenAIResponseObject): +class ResponseInProgressEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS] response: ResponsesAPIResponse -class ResponseCompletedEvent(BaseLiteLLMOpenAIResponseObject): +class ResponseCompletedEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.RESPONSE_COMPLETED] response: ResponsesAPIResponse _hidden_params: dict = PrivateAttr(default_factory=dict) @@ -1696,12 +1720,12 @@ class ResponseCompletedEvent(BaseLiteLLMOpenAIResponseObject): self._hidden_params = hidden_params -class ResponseFailedEvent(BaseLiteLLMOpenAIResponseObject): +class ResponseFailedEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.RESPONSE_FAILED] response: ResponsesAPIResponse -class ResponseIncompleteEvent(BaseLiteLLMOpenAIResponseObject): +class ResponseIncompleteEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE] response: ResponsesAPIResponse @@ -1709,7 +1733,7 @@ class ResponseIncompleteEvent(BaseLiteLLMOpenAIResponseObject): ResponsesTerminalEvent: TypeAlias = ResponseCompletedEvent | ResponseIncompleteEvent | ResponseFailedEvent -class ResponsePartAddedEvent(BaseLiteLLMOpenAIResponseObject): +class ResponsePartAddedEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.RESPONSE_PART_ADDED] item_id: str output_index: int @@ -1717,7 +1741,7 @@ class ResponsePartAddedEvent(BaseLiteLLMOpenAIResponseObject): part: dict -class ReasoningSummaryTextDeltaEvent(BaseLiteLLMOpenAIResponseObject): +class ReasoningSummaryTextDeltaEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.REASONING_SUMMARY_TEXT_DELTA] item_id: str output_index: int @@ -1725,7 +1749,7 @@ class ReasoningSummaryTextDeltaEvent(BaseLiteLLMOpenAIResponseObject): delta: str -class ReasoningSummaryTextDoneEvent(BaseLiteLLMOpenAIResponseObject): +class ReasoningSummaryTextDoneEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.REASONING_SUMMARY_TEXT_DONE] item_id: str output_index: int @@ -1734,7 +1758,7 @@ class ReasoningSummaryTextDoneEvent(BaseLiteLLMOpenAIResponseObject): text: str -class ReasoningSummaryPartDoneEvent(BaseLiteLLMOpenAIResponseObject): +class ReasoningSummaryPartDoneEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.REASONING_SUMMARY_PART_DONE] item_id: str output_index: int @@ -1743,13 +1767,13 @@ class ReasoningSummaryPartDoneEvent(BaseLiteLLMOpenAIResponseObject): part: BaseLiteLLMOpenAIResponseObject -class OutputItemAddedEvent(BaseLiteLLMOpenAIResponseObject): +class OutputItemAddedEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED] output_index: int item: BaseLiteLLMOpenAIResponseObject | None -class OutputItemDoneEvent(BaseLiteLLMOpenAIResponseObject): +class OutputItemDoneEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE] output_index: int sequence_number: int = 1 @@ -1769,7 +1793,7 @@ class OpenAIChatCompletionLogprobsContent(TypedDict, total=False): top_logprobs: list[OpenAIChatCompletionLogprobsContentTopLogprobs] -class ContentPartAddedEvent(BaseLiteLLMOpenAIResponseObject): +class ContentPartAddedEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.CONTENT_PART_ADDED] item_id: str output_index: int @@ -1797,7 +1821,7 @@ class ContentPartDonePartReasoningText(BaseLiteLLMOpenAIResponseObject): PART_UNION_TYPES = ContentPartDonePartOutputText | ContentPartDonePartRefusal | ContentPartDonePartReasoningText -class ContentPartDoneEvent(BaseLiteLLMOpenAIResponseObject): +class ContentPartDoneEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.CONTENT_PART_DONE] item_id: str output_index: int @@ -1805,7 +1829,7 @@ class ContentPartDoneEvent(BaseLiteLLMOpenAIResponseObject): part: PART_UNION_TYPES -class OutputTextDeltaEvent(BaseLiteLLMOpenAIResponseObject): +class OutputTextDeltaEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA] item_id: str output_index: int @@ -1813,7 +1837,7 @@ class OutputTextDeltaEvent(BaseLiteLLMOpenAIResponseObject): delta: str -class OutputTextAnnotationAddedEvent(BaseLiteLLMOpenAIResponseObject): +class OutputTextAnnotationAddedEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.OUTPUT_TEXT_ANNOTATION_ADDED] item_id: str output_index: int @@ -1822,7 +1846,7 @@ class OutputTextAnnotationAddedEvent(BaseLiteLLMOpenAIResponseObject): annotation: dict -class OutputTextDoneEvent(BaseLiteLLMOpenAIResponseObject): +class OutputTextDoneEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE] item_id: str output_index: int @@ -1830,7 +1854,7 @@ class OutputTextDoneEvent(BaseLiteLLMOpenAIResponseObject): text: str -class RefusalDeltaEvent(BaseLiteLLMOpenAIResponseObject): +class RefusalDeltaEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.REFUSAL_DELTA] item_id: str output_index: int @@ -1838,7 +1862,7 @@ class RefusalDeltaEvent(BaseLiteLLMOpenAIResponseObject): delta: str -class RefusalDoneEvent(BaseLiteLLMOpenAIResponseObject): +class RefusalDoneEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.REFUSAL_DONE] item_id: str output_index: int @@ -1846,72 +1870,72 @@ class RefusalDoneEvent(BaseLiteLLMOpenAIResponseObject): refusal: str -class FunctionCallArgumentsDeltaEvent(BaseLiteLLMOpenAIResponseObject): +class FunctionCallArgumentsDeltaEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA] item_id: str output_index: int delta: str -class FunctionCallArgumentsDoneEvent(BaseLiteLLMOpenAIResponseObject): +class FunctionCallArgumentsDoneEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE] item_id: str output_index: int arguments: str -class FileSearchCallInProgressEvent(BaseLiteLLMOpenAIResponseObject): +class FileSearchCallInProgressEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.FILE_SEARCH_CALL_IN_PROGRESS] output_index: int item_id: str -class FileSearchCallSearchingEvent(BaseLiteLLMOpenAIResponseObject): +class FileSearchCallSearchingEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.FILE_SEARCH_CALL_SEARCHING] output_index: int item_id: str -class FileSearchCallCompletedEvent(BaseLiteLLMOpenAIResponseObject): +class FileSearchCallCompletedEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.FILE_SEARCH_CALL_COMPLETED] output_index: int item_id: str -class WebSearchCallInProgressEvent(BaseLiteLLMOpenAIResponseObject): +class WebSearchCallInProgressEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.WEB_SEARCH_CALL_IN_PROGRESS] output_index: int item_id: str -class WebSearchCallSearchingEvent(BaseLiteLLMOpenAIResponseObject): +class WebSearchCallSearchingEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.WEB_SEARCH_CALL_SEARCHING] output_index: int item_id: str -class WebSearchCallCompletedEvent(BaseLiteLLMOpenAIResponseObject): +class WebSearchCallCompletedEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.WEB_SEARCH_CALL_COMPLETED] output_index: int item_id: str # MCP List Tools Events -class MCPListToolsInProgressEvent(BaseLiteLLMOpenAIResponseObject): +class MCPListToolsInProgressEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.MCP_LIST_TOOLS_IN_PROGRESS] sequence_number: int output_index: int item_id: str -class MCPListToolsCompletedEvent(BaseLiteLLMOpenAIResponseObject): +class MCPListToolsCompletedEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.MCP_LIST_TOOLS_COMPLETED] sequence_number: int output_index: int item_id: str -class MCPListToolsFailedEvent(BaseLiteLLMOpenAIResponseObject): +class MCPListToolsFailedEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.MCP_LIST_TOOLS_FAILED] sequence_number: int output_index: int @@ -1919,14 +1943,14 @@ class MCPListToolsFailedEvent(BaseLiteLLMOpenAIResponseObject): # MCP Call Events -class MCPCallInProgressEvent(BaseLiteLLMOpenAIResponseObject): +class MCPCallInProgressEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.MCP_CALL_IN_PROGRESS] sequence_number: int output_index: int item_id: str -class MCPCallArgumentsDeltaEvent(BaseLiteLLMOpenAIResponseObject): +class MCPCallArgumentsDeltaEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.MCP_CALL_ARGUMENTS_DELTA] output_index: int item_id: str @@ -1934,7 +1958,7 @@ class MCPCallArgumentsDeltaEvent(BaseLiteLLMOpenAIResponseObject): sequence_number: int -class MCPCallArgumentsDoneEvent(BaseLiteLLMOpenAIResponseObject): +class MCPCallArgumentsDoneEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.MCP_CALL_ARGUMENTS_DONE] output_index: int item_id: str @@ -1942,21 +1966,21 @@ class MCPCallArgumentsDoneEvent(BaseLiteLLMOpenAIResponseObject): sequence_number: int -class MCPCallCompletedEvent(BaseLiteLLMOpenAIResponseObject): +class MCPCallCompletedEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.MCP_CALL_COMPLETED] sequence_number: int item_id: str output_index: int -class MCPCallFailedEvent(BaseLiteLLMOpenAIResponseObject): +class MCPCallFailedEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.MCP_CALL_FAILED] sequence_number: int item_id: str output_index: int -class ImageGenerationPartialImageEvent(BaseLiteLLMOpenAIResponseObject): +class ImageGenerationPartialImageEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.IMAGE_GENERATION_PARTIAL_IMAGE] partial_image_index: int b64_json: str @@ -1971,7 +1995,7 @@ class ErrorEventError(BaseLiteLLMOpenAIResponseObject): param: str | dict[str, object] | None = None -class ErrorEvent(BaseLiteLLMOpenAIResponseObject): +class ErrorEvent(BaseResponsesStreamEvent): type: Literal[ResponsesAPIStreamEvents.ERROR] sequence_number: int error: ErrorEventError | None = None @@ -1980,7 +2004,7 @@ class ErrorEvent(BaseLiteLLMOpenAIResponseObject): param: str | Mapping[str, object] | None = None -class GenericEvent(BaseLiteLLMOpenAIResponseObject): +class GenericEvent(BaseResponsesStreamEvent): type: str model_config = ConfigDict(extra="allow", protected_namespaces=()) diff --git a/tests/unit/llms/openai/responses/test_openai_responses_guardrail_handler.py b/tests/unit/llms/openai/responses/test_openai_responses_guardrail_handler.py index bf871210074..8a811b83d4c 100644 --- a/tests/unit/llms/openai/responses/test_openai_responses_guardrail_handler.py +++ b/tests/unit/llms/openai/responses/test_openai_responses_guardrail_handler.py @@ -3786,6 +3786,34 @@ class TestBuildBlockSseChunks: assert completed["output"][0]["content"][0]["text"] == "Blocked by policy." assert completed["usage"] == {"input_tokens": 7, "output_tokens": 21, "total_tokens": 28} + def test_continuation_numbers_every_event_after_the_last_one_relayed(self): + yielded: Final = [ + {"type": "response.created", "sequence_number": 0, "response": {"id": "resp_live", "output": []}}, + { + "type": "response.output_item.added", + "sequence_number": 1, + "output_index": 0, + "item": {"id": "msg_orig", "type": "message", "role": "assistant"}, + }, + { + "type": "response.output_text.delta", + "sequence_number": 2, + "item_id": "msg_orig", + "output_index": 0, + "content_index": 0, + "delta": "partial", + }, + ] + + payloads: Final = self._payloads( + OpenAIResponsesHandler().build_block_sse_chunks( + self._exc(original_response=yielded), stream_started=True, responses_so_far=yielded + ) + ) + + assert payloads[-1]["type"] == "response.completed" + assert [payload.get("sequence_number") for payload in payloads] == list(range(3, 3 + len(payloads))) + def test_continuation_reads_usage_from_typed_completed_event(self): from litellm.types.llms.openai import ( ResponseCompletedEvent, diff --git a/tests/unit/proxy/guardrails/guardrail_hooks/test_llm_shield_proxy.py b/tests/unit/proxy/guardrails/guardrail_hooks/test_llm_shield_proxy.py index 63fcdfbb0c5..ea20d55ff1d 100644 --- a/tests/unit/proxy/guardrails/guardrail_hooks/test_llm_shield_proxy.py +++ b/tests/unit/proxy/guardrails/guardrail_hooks/test_llm_shield_proxy.py @@ -1676,6 +1676,48 @@ class TestResponsesStreamRestoration: assert "".join(e.delta for e in out) == "see a@example.com a[" + @pytest.mark.asyncio + async def test_a_flushed_delta_takes_the_done_events_number_and_later_events_move_up(self): + guardrail, _ = _shielded(self.VAULT) + done = OutputTextDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE, + item_id="msg_1", + output_index=0, + content_index=0, + text="Mail [EMAIL_1] x[", + sequence_number=3, + ) + + out = await _restore_stream( + guardrail, + [ + self._text_delta("Mail [EMA", 1), + self._text_delta("IL_1] x[", 2), + done, + self._text_delta("next", 4, content_index=1), + ], + ) + + assert [(e.type, e.sequence_number) for e in out] == [ + (ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, 1), + (ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, 2), + (ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, 3), + (ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE, 4), + (ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, 5), + ] + assert out[2].delta == "[" + + @pytest.mark.asyncio + async def test_a_truncated_stream_flush_continues_after_the_last_number(self): + guardrail, _ = _shielded(self.VAULT) + + out = await _restore_stream( + guardrail, [self._text_delta("see [EMAIL_1] a[", 7), self._text_delta(" b", 8, content_index=1)] + ) + + assert [e.sequence_number for e in out] == [7, 8, 9] + assert out[-1].delta == "[" + @pytest.mark.asyncio async def test_completed_response_is_restored(self): """The terminal event repeats the whole reply, and clients read it as the answer.""" diff --git a/tests/unit/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py b/tests/unit/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py index 4117c303940..3fd580792b8 100644 --- a/tests/unit/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py +++ b/tests/unit/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py @@ -11,6 +11,7 @@ spend tracking stores, so a follow-up previous_response_id still finds the conve """ import json +from collections.abc import Callable, Iterable from typing import Final from unittest.mock import AsyncMock, MagicMock @@ -473,7 +474,6 @@ def test_tool_call_arguments_are_chunked_to_match_openai_behavior(): assert evt is not None assert evt.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED assert evt.output_index == 0 - assert hasattr(evt, "__dict__") and "sequence_number" in evt.__dict__ # Collect all remaining delta events from the pending queue by creating empty chunks delta_events = [] @@ -507,17 +507,11 @@ def test_tool_call_arguments_are_chunked_to_match_openai_behavior(): assert len(evt.delta) <= 10 assert evt.item_id == "fc_call_test" assert evt.output_index == 0 - assert hasattr(evt, "__dict__") and "sequence_number" in evt.__dict__ # Verify all deltas concatenated equal the original arguments concatenated = "".join(evt.delta for evt in delta_events) assert concatenated == large_arguments - # Verify sequence numbers are increasing - sequence_numbers = [evt.__dict__["sequence_number"] for evt in delta_events] - assert sequence_numbers == sorted(sequence_numbers) - assert len(set(sequence_numbers)) == len(sequence_numbers) # All unique - def test_tool_call_delta_without_id_uses_index_mapping(): iterator = LiteLLMCompletionStreamingIterator( @@ -1467,3 +1461,47 @@ async def test_caller_function_named_like_the_search_tool_streams_as_function_ca assert [(item["type"], item["name"]) for item in output] == [("function_call", "tool_search_tool_regex")] assert "response.function_call_arguments.done" in [event["type"] for event in events] + + +def _serialized_sequence_numbers(events: Iterable[BaseLiteLLMOpenAIResponseObject]) -> list[int]: + payloads: Final = tuple(json.loads(event.model_dump_json()) for event in events) + assert all("sequence_number" in payload for payload in payloads), [ + payload["type"] for payload in payloads if "sequence_number" not in payload + ] + return [payload["sequence_number"] for payload in payloads] + + +def _long_text_chunks() -> tuple[ModelResponseStream, ...]: + return (*(_chunk(f"word{i} ") for i in range(300)), _chunk("end", finish_reason="stop")) + + +@pytest.mark.parametrize( + "stream", + [ + lambda: (_chunk("Hello"), _chunk("!", finish_reason="stop")), + _long_text_chunks, + lambda: (_chunk("Checking. "), _tool_call_chunk(finish_reason="tool_calls")), + lambda: (_reasoning_chunk("let me think"), _chunk("Hello"), _chunk("!", finish_reason="stop")), + lambda: (_reasoning_chunk("let me think"), _annotation_only_chunk(), _tool_call_chunk("tool_calls")), + lambda: (_tool_call_chunk(), _chunk("", finish_reason="tool_calls")), + ], + ids=["text", "long_text", "text_then_tool", "reasoning_then_text", "citation_tool", "tool"], +) +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.asyncio +async def test_every_event_serializes_its_position_in_the_stream_from_zero( + stream: Callable[[], tuple[ModelResponseStream, ...]], sync_mode: bool +): + iterator: Final = _build_iterator(stream()) + events: Final = list(iterator) if sync_mode else [event async for event in iterator] + + assert _serialized_sequence_numbers(events) == list(range(len(events))) + + +def test_bridged_events_answer_dict_reads_of_their_number(): + events: Final = list(_build_iterator((_chunk("Hello"), _chunk("!", finish_reason="stop")))) + positions: Final = list(range(len(events))) + + assert [event["sequence_number"] for event in events] == positions + assert [event.get("sequence_number") for event in events] == positions + assert all("sequence_number" in event for event in events) diff --git a/tests/unit/responses/test_streaming_iterator.py b/tests/unit/responses/test_streaming_iterator.py index 18953f8d7bc..54d5b91c225 100644 --- a/tests/unit/responses/test_streaming_iterator.py +++ b/tests/unit/responses/test_streaming_iterator.py @@ -2557,3 +2557,54 @@ def test_faked_stream_builds_each_refusal_summary_and_custom_tool_input_once_for (1, "content", 1): "No forecasts for Mars.", (2, "input", 0): "curl wttr.in/Berlin", } + + +@pytest.mark.parametrize( + "output", + [ + [ + { + "type": "message", + "id": "msg_seq", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "hello world", "annotations": []}], + } + ], + [ + { + "type": "reasoning", + "id": "rs_seq", + "summary": [{"type": "summary_text", "text": "thinking about it"}], + }, + { + "type": "function_call", + "id": "fc_seq", + "call_id": "call_seq", + "name": "lookup", + "arguments": '{"id":1}', + }, + ], + ], + ids=["message", "reasoning_and_tool"], +) +def test_synthetic_stream_serializes_every_event_position_from_zero(output: list[dict[str, object]]): + transformed: Final = ResponsesAPIResponse.model_validate( + { + "id": "resp_seq", + "created_at": 1741476542, + "status": "completed", + "model": "claude-sonnet-5-5", + "object": "response", + "output": output, + } + ) + + events: Final = streaming_module.build_synthetic_response_events( + transformed=transformed, + logging_obj=None, + chunk_size=5, + ) + + payloads: Final = tuple(json.loads(event.model_dump_json()) for event in events) + assert [payload.get("sequence_number") for payload in payloads] == list(range(len(payloads))) diff --git a/tests/unit/test_router/test_router.py b/tests/unit/test_router/test_router.py index beda09fae3f..69ba7726703 100644 --- a/tests/unit/test_router/test_router.py +++ b/tests/unit/test_router/test_router.py @@ -5209,6 +5209,151 @@ async def test_aresponses_streaming_content_policy_error_event_routes_to_content assert fallback_call.await_args.kwargs["model"] == "gemini/gemini-2.5-flash" +@pytest.mark.asyncio +async def test_aresponses_stream_keeps_numbering_after_a_mid_stream_fallback(): + from litellm.types.llms.openai import OutputTextDeltaEvent, ResponsesAPIStreamEvents + + def delta(sequence_number: int, text: str) -> OutputTextDeltaEvent: + return OutputTextDeltaEvent( + type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, + item_id="msg_1", + output_index=0, + content_index=0, + delta=text, + sequence_number=sequence_number, + ) + + from litellm.exceptions import MidStreamFallbackError + + router: Final = _InjectedFallbackRouter(_AsyncList([delta(0, "c"), delta(1, "d")])) + primary: Final = _make_responses_iterator( + chunks=[delta(0, "a"), delta(1, "b")], + error=MidStreamFallbackError( + message="primary died", + model="primary", + llm_provider="anthropic", + is_pre_first_chunk=False, + generated_content="ab", + ), + model="primary", + ) + + wrapped: Final = await router._aresponses_streaming_iterator( + response=primary, + initial_kwargs={"model": "primary", "stream": True, "input": "Hi", "original_generic_function": litellm.aresponses}, + ) + events: Final = [event async for event in wrapped] + + assert [(event.delta, event.sequence_number) for event in events] == [("a", 0), ("b", 1), ("c", 2), ("d", 3)] + + + +@pytest.mark.asyncio +async def test_aresponses_stream_fallback_keeps_higher_numbers_and_passes_unnumbered_events(): + from litellm.exceptions import MidStreamFallbackError + from litellm.types.llms.openai import OutputTextDeltaEvent, ResponsesAPIStreamEvents + + def delta(text: str) -> OutputTextDeltaEvent: + return OutputTextDeltaEvent( + type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, item_id="msg_1", output_index=0, content_index=0, delta=text + ) + + def numbered(sequence_number: int, text: str) -> OutputTextDeltaEvent: + return delta(text).model_copy(update={"sequence_number": sequence_number}) + + router: Final = _InjectedFallbackRouter(_AsyncList([numbered(7, "c"), delta("d"), numbered(8, "e")])) + primary: Final = _make_responses_iterator( + chunks=[numbered(0, "a"), numbered(1, "b")], + error=MidStreamFallbackError( + message="primary died", + model="primary", + llm_provider="anthropic", + is_pre_first_chunk=False, + generated_content="ab", + ), + model="primary", + ) + + wrapped: Final = await router._aresponses_streaming_iterator( + response=primary, + initial_kwargs={"model": "primary", "stream": True, "input": "Hi", "original_generic_function": litellm.aresponses}, + ) + events: Final = [event async for event in wrapped] + + assert [(event.delta, event.get("sequence_number")) for event in events] == [ + ("a", 0), + ("b", 1), + ("c", 7), + ("d", None), + ("e", 8), + ] + + +@pytest.mark.asyncio +async def test_aresponses_stream_without_fallback_passes_upstream_numbers_unchanged(): + from litellm.types.llms.openai import OutputTextDeltaEvent, ResponsesAPIStreamEvents + + def delta(sequence_number: int, text: str) -> OutputTextDeltaEvent: + return OutputTextDeltaEvent( + type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, + item_id="msg_1", + output_index=0, + content_index=0, + delta=text, + sequence_number=sequence_number, + ) + + router: Final = _InjectedFallbackRouter(_AsyncList([])) + primary: Final = _make_responses_iterator( + chunks=[delta(0, "a"), delta(1, "b"), delta(1, "c"), delta(0, "d")], model="primary" + ) + + wrapped: Final = await router._aresponses_streaming_iterator( + response=primary, + initial_kwargs={"model": "primary", "stream": True, "input": "Hi", "original_generic_function": litellm.aresponses}, + ) + events: Final = [event async for event in wrapped] + + assert [(event.delta, event.sequence_number) for event in events] == [("a", 0), ("b", 1), ("c", 1), ("d", 0)] + + +@pytest.mark.asyncio +async def test_aresponses_stream_fallback_continues_above_the_highest_number_the_primary_sent(): + from litellm.exceptions import MidStreamFallbackError + from litellm.types.llms.openai import OutputTextDeltaEvent, ResponsesAPIStreamEvents + + def delta(sequence_number: int, text: str) -> OutputTextDeltaEvent: + return OutputTextDeltaEvent( + type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, + item_id="msg_1", + output_index=0, + content_index=0, + delta=text, + sequence_number=sequence_number, + ) + + router: Final = _InjectedFallbackRouter(_AsyncList([delta(0, "d")])) + primary: Final = _make_responses_iterator( + chunks=[delta(0, "a"), delta(3, "b"), delta(1, "c")], + error=MidStreamFallbackError( + message="primary died", + model="primary", + llm_provider="anthropic", + is_pre_first_chunk=False, + generated_content="abc", + ), + model="primary", + ) + + wrapped: Final = await router._aresponses_streaming_iterator( + response=primary, + initial_kwargs={"model": "primary", "stream": True, "input": "Hi", "original_generic_function": litellm.aresponses}, + ) + events: Final = [event async for event in wrapped] + + assert [(event.delta, event.sequence_number) for event in events] == [("a", 0), ("b", 3), ("c", 1), ("d", 4)] + + @pytest.mark.asyncio async def test_aresponses_streaming_iterator_unwraps_content_policy_trigger_for_fallback_dispatch(): """The fallback dispatch matches on the trigger's own type, so the wrapper must hand it the diff --git a/tests/unit/types/llms/test_types_llms_openai.py b/tests/unit/types/llms/test_types_llms_openai.py index e90d48b64af..2271b049015 100644 --- a/tests/unit/types/llms/test_types_llms_openai.py +++ b/tests/unit/types/llms/test_types_llms_openai.py @@ -11,8 +11,10 @@ from litellm.types.llms.openai import ( HttpxBinaryResponseContent, OpenAIModerationResponse, OpenAIVideoObject, + OutputTextDeltaEvent, ResponseCompletedEvent, ResponsesAPIResponse, + ResponsesAPIStreamEvents, ) @@ -686,3 +688,32 @@ def test_rebuilding_a_dumped_responses_answer_keeps_the_tool_search_call_keys(ex rebuilt: Final = ResponsesAPIResponse(**answer.model_dump()) assert rebuilt.model_dump()["output"] == [search_call] + + +def _text_delta() -> OutputTextDeltaEvent: + return OutputTextDeltaEvent( + type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, item_id="msg_1", output_index=0, content_index=0, delta="hi" + ) + + +def test_stream_event_dict_reads_see_a_field_set_after_construction(): + event: Final = _text_delta().model_copy(update={"sequence_number": 3}) + + assert event["sequence_number"] == 3 + assert event.get("sequence_number") == 3 + assert "sequence_number" in event + assert dict(event.items())["sequence_number"] == 3 + assert event["delta"] == "hi" + + +def test_stream_event_dict_reads_see_an_undeclared_field_named_keys(): + assert _text_delta().model_copy(update={"keys": "value"})["keys"] == "value" + + +def test_stream_event_dict_reads_still_miss_absent_keys(): + event: Final = _text_delta() + + assert event.get("sequence_number", "fallback") == "fallback" + assert "sequence_number" not in event + with pytest.raises(KeyError): + event["sequence_number"]