Skip to content

[TRTLLM-8489][feat] Support stream interval for Responses API #10109

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
JunyiXu-nv wants to merge 1 commit into from
Closed

[TRTLLM-8489][feat] Support stream interval for Responses API #10109

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 20 additions & 9 deletions tensorrt_llm/serve/harmony_adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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."""
Expand Down Expand Up @@ -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.

Expand All @@ -1305,23 +1315,24 @@ 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:
stream_state = self.create_stream_state(request_id, available_tools,
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,
Expand Down
138 changes: 71 additions & 67 deletions tensorrt_llm/serve/responses_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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)
Comment on lines +1559 to +1567

@coderabbitai coderabbitai Bot Feb 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Reasoning delta emitted before output-item-added when text is whitespace-only.

If reasoning_delta_text is truthy but contains only whitespace (e.g. " "), the strip() check on Line 1561 is falsy so the output-item-added / content-part-added events are skipped, yet the reasoning text delta event on Line 1566 is still yielded. This violates the Responses API streaming protocol, which expects response.output_item.added and response.content_part.added to precede any delta event for that item.

The same pattern exists for delta_text on Lines 1588-1593.

Proposed fix: gate the delta yield behind the added-events guard, or always send added events when any delta is emitted 
     # 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(
-            )
+        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 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()
+        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, [])

Alternatively, if there is a deliberate reason to skip whitespace-only leading deltas, move the delta yield inside the if .strip(): block as well.

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@tensorrt_llm/serve/responses_utils.py` around lines 1559 - 1567, The code
currently yields a reasoning/text delta event even when the delta is
whitespace-only, which can emit a delta before the corresponding
output-item-added/content-part-added events; update the logic in the block
handling reasoning_delta_text (and the analogous block for delta_text) so that
the yield of
streaming_events_helper.get_reasoning_text_delta_event(reasoning_delta_text)
(and streaming_events_helper.get_text_delta_event(delta_text)) is only executed
after you have emitted the added events — either by moving the delta yield
inside the if reasoning_delta_text.strip() branch or by forcing
streaming_events_helper.get_reasoning_output_added_events()/get_output_added_events()
to run whenever a delta is truthy; ensure you reference the existing helpers
(streaming_events_helper.is_reasoning_sent, get_reasoning_output_added_events,
get_reasoning_text_delta_event, get_output_added_events, get_text_delta_event)
when making the change.


# Send done events if needed
if should_send_reasoning_done and reasoning_full_content:
reasoning_item = ResponseReasoningItem(
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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
Expand Down Expand Up @@ -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:
Expand Down
Loading