From b0c20c4bc4494d0545c66d5f2d2b81aeeed01276 Mon Sep 17 00:00:00 2001 From: Junyi Xu <219237550+JunyiXu-nv@users.noreply.github.com> Date: Thu, 18 Dec 2025 07:39:03 +0000 Subject: [PATCH] [TRTLLM-8489][feat] Support stream interval for Responses API Signed-off-by: Junyi Xu <219237550+JunyiXu-nv@users.noreply.github.com> --- tensorrt_llm/serve/harmony_adapter.py | 29 ++++-- tensorrt_llm/serve/responses_utils.py | 138 +++++++++++++------------- 2 files changed, 91 insertions(+), 76 deletions(-) diff --git a/tensorrt_llm/serve/harmony_adapter.py b/tensorrt_llm/serve/harmony_adapter.py index 883483897cee..f5b34f0e8323 100644 --- a/tensorrt_llm/serve/harmony_adapter.py +++ b/tensorrt_llm/serve/harmony_adapter.py @@ -145,15 +145,24 @@ def process_token_batch(self, tokens: list[int]) -> list[dict[str, Any]]: return deltas - def process_token_batch_to_messages(self, - tokens: list[int]) -> list[Message]: + def process_token_batch_to_messages( + self, tokens: list[int]) -> tuple[list[Message], dict[str, str]]: """ Process a batch of tokens while maintaining parsing state. - Returns OpenAI Messages for Responses API + Returns OpenAI Messages for Responses API and parsed contents """ self.tokens_processed += len(tokens) + contents: dict[str, str] = {} for token in tokens: + if self.parser.current_channel: + if self.parser.current_channel not in contents: + contents[self.parser.current_channel] = "" + + contents[ + self.parser. + current_channel] += self.parser.last_content_delta or "" + # Store previous state for transition detection prev_channel = self.parser.current_channel prev_recipient = self.parser.current_recipient @@ -179,7 +188,7 @@ def process_token_batch_to_messages(self, self.channel_started = False self.current_channel_state = None - return self.parser.messages + return self.parser.messages, contents def _create_closing_token_delta(self) -> dict[str, Any] | None: """Create closing token delta for channel transition.""" @@ -1292,7 +1301,8 @@ def stateful_stream_harmony_tokens_to_openai_messages( request_id: str, tokens: list[int], available_tools: list[dict[str, Any]] | None = None, - tool_choice: str | None = None) -> list[Message]: + tool_choice: str | None = None + ) -> tuple[list[Message], dict[str, str]]: """ Process tokens using stateful parsing. @@ -1305,7 +1315,7 @@ def stateful_stream_harmony_tokens_to_openai_messages( available_tools: Available tools for filtering Returns: - List of OpenAI Messages + List of OpenAI Messages and parsed contents """ stream_state = self._stream_states.get(request_id, None) if stream_state is None: @@ -1313,15 +1323,16 @@ def stateful_stream_harmony_tokens_to_openai_messages( tool_choice) try: - messages = stream_state.process_token_batch_to_messages(tokens) - return messages + messages, contents = stream_state.process_token_batch_to_messages( + tokens) + return messages, contents except (HarmonyError, UnicodeDecodeError, ValueError): logger.error( f"Streaming: Failed to process token batch of {len(tokens)} tokens for request {request_id}", ) logger.debug(f"Problematic streaming tokens: {tokens}") - return [] + return [], {} def create_openai_streaming_response( self, diff --git a/tensorrt_llm/serve/responses_utils.py b/tensorrt_llm/serve/responses_utils.py index 35a999ac9a5a..df9a83455d3c 100644 --- a/tensorrt_llm/serve/responses_utils.py +++ b/tensorrt_llm/serve/responses_utils.py @@ -1540,7 +1540,7 @@ def check_parser(parser_id: Optional[str], _responses_debug_log( repr( - f" ---------> delta text: {delta_text}, reasoning delta text: {reasoning_delta_text}, calls: {calls}" + f" ---------> delta text: '{delta_text}', reasoning delta text: '{reasoning_delta_text}', calls: '{calls}', is_finished: '{finished_generation}'" )) # Check if we need to send done events for completed sections @@ -1556,6 +1556,16 @@ def check_parser(parser_id: Optional[str], finished_generation=finished_generation, ) + # Reasoning delta may exist with text delta + if reasoning_delta_text: + if reasoning_delta_text.strip(): + if not streaming_events_helper.is_reasoning_sent: + streaming_events_helper.is_reasoning_sent = True + yield from streaming_events_helper.get_reasoning_output_added_events( + ) + yield streaming_events_helper.get_reasoning_text_delta_event( + reasoning_delta_text) + # Send done events if needed if should_send_reasoning_done and reasoning_full_content: reasoning_item = ResponseReasoningItem( @@ -1574,6 +1584,14 @@ def check_parser(parser_id: Optional[str], streaming_events_helper.is_output_item_added_sent = False streaming_events_helper.is_reasoning_sent = False + # Send delta events for ongoing content + if delta_text: + if delta_text.strip(): + if not streaming_events_helper.is_text_sent: + streaming_events_helper.is_text_sent = True + yield from streaming_events_helper.get_message_output_added_events() + yield streaming_events_helper.get_text_delta_event(delta_text, []) + if should_send_text_done and text_full_content: text_content = ResponseOutputText( text=text_full_content, @@ -1595,22 +1613,6 @@ def check_parser(parser_id: Optional[str], streaming_events_helper.is_output_item_added_sent = False streaming_events_helper.is_text_sent = False - # Send delta events for ongoing content - if delta_text: - if delta_text.strip(): - if not streaming_events_helper.is_text_sent: - streaming_events_helper.is_text_sent = True - yield from streaming_events_helper.get_message_output_added_events() - yield streaming_events_helper.get_text_delta_event(delta_text, []) - elif reasoning_delta_text: - if reasoning_delta_text.strip(): - if not streaming_events_helper.is_reasoning_sent: - streaming_events_helper.is_reasoning_sent = True - yield from streaming_events_helper.get_reasoning_output_added_events( - ) - yield streaming_events_helper.get_reasoning_text_delta_event( - reasoning_delta_text) - def _generate_streaming_event_harmony( harmony_adapter: HarmonyAdapter, @@ -1620,11 +1622,39 @@ def _generate_streaming_event_harmony( streaming_events_helper: ResponsesStreamingEventsHelper, ): tools = [tool.model_dump() for tool in request.tools] - messages = harmony_adapter.stateful_stream_harmony_tokens_to_openai_messages( + messages, contents = harmony_adapter.stateful_stream_harmony_tokens_to_openai_messages( stream_request_id, output.token_ids_diff, tools, request.tool_choice) stream_state = harmony_adapter.get_stream_state(stream_request_id) assert stream_state is not None parser = stream_state.get_parser() + + # send reasoning -> send done -> send text + # to prevent interleaved reasoning and done events + if ("analysis" in contents and contents["analysis"] + and parser.current_recipient is None): + if not streaming_events_helper.is_output_item_added_sent: + streaming_events_helper.is_output_item_added_sent = True + + reasoning_item = ResponseReasoningItem( + id=streaming_events_helper.item_id, + type="reasoning", + summary=[], + status="in_progress", + ) + + reasoning_content = PartReasoningText( + type="reasoning_text", + text="", + ) + + yield streaming_events_helper.get_output_item_added_event( + reasoning_item) + yield streaming_events_helper.get_content_part_added_event( + reasoning_content) + + yield streaming_events_helper.get_reasoning_text_delta_event( + contents["analysis"]) + if parser.state == StreamState.EXPECT_START: streaming_events_helper.output_index_increment() streaming_events_helper.is_output_item_added_sent = False @@ -1674,58 +1704,32 @@ def _generate_streaming_event_harmony( yield streaming_events_helper.get_output_item_done_event( text_item) - if parser.last_content_delta: - if (parser.current_channel == "final" - and parser.current_recipient is None): - if not streaming_events_helper.is_output_item_added_sent: - streaming_events_helper.is_output_item_added_sent = True - - output_item = ResponseOutputMessage( - id=streaming_events_helper.item_id, - type="message", - role="assistant", - content=[], - status="in_progress", - ) - - content_part = ResponseOutputText( - type="output_text", - text="", - annotations=[], - logprobs=[], - ) - yield streaming_events_helper.get_output_item_added_event( - output_item) - yield streaming_events_helper.get_content_part_added_event( - content_part) - - yield streaming_events_helper.get_text_delta_event( - parser.last_content_delta, []) - - elif (parser.current_channel == "analysis" - and parser.current_recipient is None): - if not streaming_events_helper.is_output_item_added_sent: - streaming_events_helper.is_output_item_added_sent = True + if ("final" in contents and contents["final"] + and parser.current_recipient is None): + if not streaming_events_helper.is_output_item_added_sent: + streaming_events_helper.is_output_item_added_sent = True - reasoning_item = ResponseReasoningItem( - id=streaming_events_helper.item_id, - type="reasoning", - summary=[], - status="in_progress", - ) - - reasoning_content = PartReasoningText( - type="reasoning_text", - text="", - ) + output_item = ResponseOutputMessage( + id=streaming_events_helper.item_id, + type="message", + role="assistant", + content=[], + status="in_progress", + ) - yield streaming_events_helper.get_output_item_added_event( - reasoning_item) - yield streaming_events_helper.get_content_part_added_event( - reasoning_content) + content_part = ResponseOutputText( + type="output_text", + text="", + annotations=[], + logprobs=[], + ) + yield streaming_events_helper.get_output_item_added_event( + output_item) + yield streaming_events_helper.get_content_part_added_event( + content_part) - yield streaming_events_helper.get_reasoning_text_delta_event( - parser.last_content_delta) + yield streaming_events_helper.get_text_delta_event( + contents["final"], []) class ResponsesStreamingProcessor: