fix(interactions): never drop streamed text deltas; always emit terminal completion (#28394)
* fix(interactions): never drop streamed text deltas; always emit terminal completion The interactions streaming bridge had two bugs flagged by Greptile on PR #28153: 1. The first OutputTextDeltaEvent (and the second, when no ResponseCreatedEvent precedes the deltas) was consumed to emit a synthetic interaction.created / step.start event, but the chunk's text payload was never forwarded as a step.delta. The text only reappeared in the terminal step.stop, which defeats the purpose of incremental streaming. 2. When the upstream Responses API stream ended via StopIteration without a ResponseCompletedEvent, the iterator emitted step.stop but never the terminal interaction.completed event carrying the full collected text. This refactors the iterator to translate each upstream chunk into a list of events (instead of a single event) and buffers them in a deque. A text delta now expands into [interaction.created, step.start, step.delta] on the first chunk so no token is dropped, and the StopIteration / StopAsyncIteration fallback always flushes a terminal interaction.completed event when one hasn't already been sent. Both behaviors are covered by new unit tests: - test_no_text_token_is_dropped_during_streaming - test_response_created_then_text_delta_emits_step_start_and_delta - test_stop_iteration_fallback_emits_completion_event - test_response_completed_emits_stop_then_completion (no double-emit) Co-authored-by: Mateo Wang <mateo-berri@users.noreply.github.com> * fix(interactions): correlate EOF terminal events with stream's interaction id The StopIteration fallback path previously built the terminal step.stop / interaction.completed events with id=None (legacy content.stop) and a memory-address fallback string (interaction.completed), neither of which matched the item_id used by the earlier interaction.created / step.start / step.delta events in the same stream. Downstream consumers correlating events by id would see a mismatch. Persist the interaction id derived from the first upstream chunk (item_id on an OutputTextDeltaEvent, or response.id on a ResponseCreatedEvent) and reuse it when flushing the terminal events on EOF. Author: mateo-berri <277851410+mateo-berri@users.noreply.github.com> * ci(windows): raise UV_HTTP_TIMEOUT to 300s for uv sync The using_litellm_on_windows job has been hitting flaky PyPI download timeouts during 'uv sync --frozen --group dev' — different packages on each rerun (six, pydantic-core), all surfacing the same uv error: Failed to download distribution due to network timeout. Try increasing UV_HTTP_TIMEOUT (current value: 30s). uv's default 30s per-request timeout is too tight for the Windows runner on this project (50+ deps, several multi-MB wheels), so bump it to 300s to let slow individual downloads complete instead of failing the build. * fix(interactions): correlate ResponseCompletedEvent terminal events with stream's interaction id When a stream starts directly with OutputTextDeltaEvent (no preceding ResponseCreatedEvent), interaction.created carries item_id while interaction.completed previously carried response.id from ResponseCompletedEvent. The two ids can differ, leaving consumers that correlate events by id unable to match the start and completion events. Fall back to self._interaction_id (set on the first chunk that derives an id) before response.id, mirroring the EOF terminal path. --------- Co-authored-by: Cursor Agent <cursoragent@cursor.com> Co-authored-by: Mateo Wang <mateo-berri@users.noreply.github.com>
This commit is contained in:
Mateo Wang
GitHub
parent
718c4637a8
commit
8acf64e16c
@ -158,6 +158,8 @@ jobs:
|
||||
CHOCOLATEY_CONFIRM_ALL: "true"
|
||||
- run:
|
||||
name: Install Dependencies
|
||||
environment:
|
||||
UV_HTTP_TIMEOUT: "300"
|
||||
command: |
|
||||
$installer = Join-Path $env:TEMP "uv-install.ps1"
|
||||
Invoke-WebRequest -Uri https://astral.sh/uv/0.10.9/install.ps1 -OutFile $installer
|
||||
|
||||
@ -2,7 +2,17 @@
|
||||
Streaming iterator for transforming Responses API stream to Interactions API stream.
|
||||
"""
|
||||
|
||||
from typing import Any, AsyncIterator, Dict, Iterator, Optional, cast
|
||||
from collections import deque
|
||||
from typing import (
|
||||
Any,
|
||||
AsyncIterator,
|
||||
Deque,
|
||||
Dict,
|
||||
Iterator,
|
||||
List,
|
||||
Optional,
|
||||
cast,
|
||||
)
|
||||
|
||||
from litellm.responses.streaming_iterator import (
|
||||
BaseResponsesAPIStreamingIterator,
|
||||
@ -33,9 +43,9 @@ class LiteLLMResponsesInteractionsStreamingIterator:
|
||||
|
||||
Schema selection:
|
||||
- New schema (default, use_legacy_interactions_schema=False):
|
||||
interaction.created → step.start → step.delta … → step.stop → interaction.completed
|
||||
interaction.created -> step.start -> step.delta ... -> step.stop -> interaction.completed
|
||||
- Legacy schema (use_legacy_interactions_schema=True, remove after June 8 2026):
|
||||
interaction.start → content.start → content.delta … → content.stop → interaction.complete
|
||||
interaction.start -> content.start -> content.delta ... -> content.stop -> interaction.complete
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
@ -63,86 +73,152 @@ class LiteLLMResponsesInteractionsStreamingIterator:
|
||||
# emitted by this stream use a consistent schema, even if the global
|
||||
# flag is mutated mid-stream (e.g. by a config reload).
|
||||
self._use_legacy: bool = litellm.use_legacy_interactions_schema
|
||||
# Buffer of events that have been derived from upstream chunks but not
|
||||
# yet returned to the caller. A single Responses API chunk may expand
|
||||
# into multiple Interactions API events (e.g. the first text delta
|
||||
# produces interaction.created + step.start + step.delta), and the
|
||||
# terminal sequence on stream end may also span multiple events
|
||||
# (step.stop + interaction.completed).
|
||||
self._pending_events: Deque[InteractionsAPIStreamingResponse] = deque()
|
||||
# Tracks whether we've already emitted a terminal completion event so
|
||||
# the StopIteration fallback path doesn't double-emit.
|
||||
self._sent_completion_event = False
|
||||
# ID resolved from the first upstream chunk (item_id on a text delta or
|
||||
# response.id on response.created). Persisted so the EOF terminal
|
||||
# events stay correlated with the start events delivered earlier.
|
||||
self._interaction_id: Optional[str] = None
|
||||
|
||||
def _transform_responses_chunk_to_interactions_chunk(
|
||||
self,
|
||||
responses_chunk: ResponsesAPIStreamingResponse,
|
||||
) -> Optional[InteractionsAPIStreamingResponse]:
|
||||
# ------------------------------------------------------------------
|
||||
# Event builders
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _build_interaction_start_event(
|
||||
self, interaction_id: str
|
||||
) -> InteractionsAPIStreamingResponse:
|
||||
event_type = "interaction.start" if self._use_legacy else "interaction.created"
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type=event_type,
|
||||
id=interaction_id,
|
||||
object="interaction",
|
||||
status="in_progress",
|
||||
model=self.model,
|
||||
)
|
||||
|
||||
def _build_content_start_event(
|
||||
self, interaction_id: str
|
||||
) -> InteractionsAPIStreamingResponse:
|
||||
if self._use_legacy:
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type="content.start",
|
||||
id=interaction_id,
|
||||
object="content",
|
||||
delta={"type": "text", "text": ""},
|
||||
)
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type="step.start",
|
||||
index=0,
|
||||
step={"type": "model_output", "content": []},
|
||||
)
|
||||
|
||||
def _build_text_delta_event(
|
||||
self, interaction_id: str, delta_text: str
|
||||
) -> InteractionsAPIStreamingResponse:
|
||||
if self._use_legacy:
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type="content.delta",
|
||||
id=interaction_id,
|
||||
object="content",
|
||||
delta={"type": "text", "text": delta_text},
|
||||
)
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type="step.delta",
|
||||
index=0,
|
||||
delta={"type": "text", "text": delta_text},
|
||||
)
|
||||
|
||||
def _build_content_stop_event(
|
||||
self, interaction_id: Optional[str]
|
||||
) -> InteractionsAPIStreamingResponse:
|
||||
if self._use_legacy:
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type="content.stop",
|
||||
id=interaction_id,
|
||||
object="content",
|
||||
delta={"type": "text", "text": self.collected_text},
|
||||
)
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type="step.stop",
|
||||
index=0,
|
||||
)
|
||||
|
||||
def _build_completion_event(
|
||||
self, response_id: str
|
||||
) -> InteractionsAPIStreamingResponse:
|
||||
if self._use_legacy:
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type="interaction.complete",
|
||||
id=response_id,
|
||||
object="interaction",
|
||||
status="completed",
|
||||
model=self.model,
|
||||
outputs=[{"type": "text", "text": self.collected_text}],
|
||||
)
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type="interaction.completed",
|
||||
id=response_id,
|
||||
object="interaction",
|
||||
status="completed",
|
||||
model=self.model,
|
||||
steps=[
|
||||
{
|
||||
"type": "model_output",
|
||||
"content": [{"type": "text", "text": self.collected_text}],
|
||||
}
|
||||
],
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Per-chunk transform (returns a list of events to enqueue)
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _events_for_chunk(
|
||||
self, responses_chunk: ResponsesAPIStreamingResponse
|
||||
) -> List[InteractionsAPIStreamingResponse]:
|
||||
"""
|
||||
Transform a Responses API streaming chunk to an Interactions API streaming chunk.
|
||||
Translate a single upstream Responses API chunk into the list of
|
||||
Interactions API events it should produce.
|
||||
|
||||
Emits new-schema events by default; falls back to legacy events when
|
||||
``litellm.use_legacy_interactions_schema`` is True.
|
||||
Remove legacy branch after June 8, 2026.
|
||||
Returning a list (rather than a single event) lets a chunk emit any
|
||||
synthetic start events that haven't been sent yet *together with* the
|
||||
actual delta event, so we never silently drop the chunk's payload.
|
||||
"""
|
||||
if not responses_chunk:
|
||||
return None
|
||||
return []
|
||||
|
||||
use_legacy = self._use_legacy
|
||||
|
||||
# Handle OutputTextDeltaEvent
|
||||
# Text delta: emit any missing start events, then the delta itself.
|
||||
if isinstance(responses_chunk, OutputTextDeltaEvent):
|
||||
delta_text = (
|
||||
responses_chunk.delta if isinstance(responses_chunk.delta, str) else ""
|
||||
)
|
||||
self.collected_text += delta_text
|
||||
item_id = (
|
||||
interaction_id = (
|
||||
getattr(responses_chunk, "item_id", None) or f"interaction_{id(self)}"
|
||||
)
|
||||
if self._interaction_id is None:
|
||||
self._interaction_id = interaction_id
|
||||
|
||||
# Send the "interaction started" event on the first delta
|
||||
events: List[InteractionsAPIStreamingResponse] = []
|
||||
if not self.sent_interaction_start:
|
||||
self.sent_interaction_start = True
|
||||
if use_legacy:
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type="interaction.start",
|
||||
id=item_id,
|
||||
object="interaction",
|
||||
status="in_progress",
|
||||
model=self.model,
|
||||
)
|
||||
else:
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type="interaction.created",
|
||||
id=item_id,
|
||||
object="interaction",
|
||||
status="in_progress",
|
||||
model=self.model,
|
||||
)
|
||||
|
||||
# Send the "content/step started" event on the second delta
|
||||
events.append(self._build_interaction_start_event(interaction_id))
|
||||
if not self.sent_content_start:
|
||||
self.sent_content_start = True
|
||||
if use_legacy:
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type="content.start",
|
||||
id=item_id,
|
||||
object="content",
|
||||
delta={"type": "text", "text": ""},
|
||||
)
|
||||
else:
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type="step.start",
|
||||
index=0,
|
||||
step={"type": "model_output", "content": []},
|
||||
)
|
||||
events.append(self._build_content_start_event(interaction_id))
|
||||
events.append(self._build_text_delta_event(interaction_id, delta_text))
|
||||
return events
|
||||
|
||||
# Emit the delta itself
|
||||
if use_legacy:
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type="content.delta",
|
||||
id=item_id,
|
||||
object="content",
|
||||
delta={"type": "text", "text": delta_text},
|
||||
)
|
||||
else:
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type="step.delta",
|
||||
index=0,
|
||||
delta={"type": "text", "text": delta_text},
|
||||
)
|
||||
|
||||
# Handle ResponseCreatedEvent or ResponseInProgressEvent
|
||||
# Response created / in-progress: synthesize interaction start if we
|
||||
# haven't already sent one.
|
||||
if isinstance(responses_chunk, (ResponseCreatedEvent, ResponseInProgressEvent)):
|
||||
if not self.sent_interaction_start:
|
||||
self.sent_interaction_start = True
|
||||
@ -151,224 +227,135 @@ class LiteLLMResponsesInteractionsStreamingIterator:
|
||||
if hasattr(responses_chunk, "response")
|
||||
else None
|
||||
) or f"interaction_{id(self)}"
|
||||
event_type = (
|
||||
"interaction.start" if use_legacy else "interaction.created"
|
||||
)
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type=event_type,
|
||||
id=response_id,
|
||||
object="interaction",
|
||||
status="in_progress",
|
||||
model=self.model,
|
||||
)
|
||||
if self._interaction_id is None:
|
||||
self._interaction_id = response_id
|
||||
return [self._build_interaction_start_event(response_id)]
|
||||
return []
|
||||
|
||||
# Handle ResponseCompletedEvent
|
||||
# Response completed: emit step.stop (if content was started) followed
|
||||
# by the terminal completion event. Prefer the interaction id already
|
||||
# established by earlier events so consumers can correlate the start
|
||||
# and completion events by id (response.id may differ from the item_id
|
||||
# used to derive the initial id when the stream starts directly with a
|
||||
# text delta).
|
||||
if isinstance(responses_chunk, ResponseCompletedEvent):
|
||||
self.finished = True
|
||||
response = responses_chunk.response
|
||||
response_id = getattr(response, "id", None) or f"interaction_{id(self)}"
|
||||
response_id = (
|
||||
self._interaction_id
|
||||
or getattr(response, "id", None)
|
||||
or f"interaction_{id(self)}"
|
||||
)
|
||||
|
||||
if use_legacy:
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type="interaction.complete",
|
||||
id=response_id,
|
||||
object="interaction",
|
||||
status="completed",
|
||||
model=self.model,
|
||||
outputs=[{"type": "text", "text": self.collected_text}],
|
||||
)
|
||||
else:
|
||||
return InteractionsAPIStreamingResponse(
|
||||
event_type="interaction.completed",
|
||||
id=response_id,
|
||||
object="interaction",
|
||||
status="completed",
|
||||
model=self.model,
|
||||
steps=[
|
||||
{
|
||||
"type": "model_output",
|
||||
"content": [{"type": "text", "text": self.collected_text}],
|
||||
}
|
||||
],
|
||||
)
|
||||
terminal: List[InteractionsAPIStreamingResponse] = []
|
||||
if self.sent_content_start:
|
||||
terminal.append(self._build_content_stop_event(response_id))
|
||||
terminal.append(self._build_completion_event(response_id))
|
||||
self._sent_completion_event = True
|
||||
return terminal
|
||||
|
||||
# For other event types, return None (skip)
|
||||
return None
|
||||
return []
|
||||
|
||||
def _build_terminal_events_on_eof(
|
||||
self,
|
||||
) -> List[InteractionsAPIStreamingResponse]:
|
||||
"""
|
||||
Build the events to flush when the upstream stream ends without a
|
||||
ResponseCompletedEvent. Ensures consumers always observe a terminal
|
||||
interaction.completed/interaction.complete carrying the full text.
|
||||
"""
|
||||
if self._sent_completion_event:
|
||||
return []
|
||||
|
||||
fallback_id = self._interaction_id or f"interaction_{id(self)}"
|
||||
terminal: List[InteractionsAPIStreamingResponse] = []
|
||||
if self.sent_content_start:
|
||||
terminal.append(self._build_content_stop_event(fallback_id))
|
||||
if self.sent_interaction_start or self.collected_text:
|
||||
terminal.append(self._build_completion_event(fallback_id))
|
||||
self._sent_completion_event = True
|
||||
return terminal
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Iteration
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def __iter__(self) -> Iterator[InteractionsAPIStreamingResponse]:
|
||||
"""Sync iterator implementation."""
|
||||
return self
|
||||
|
||||
def __next__(self) -> InteractionsAPIStreamingResponse:
|
||||
"""Get next chunk in sync mode."""
|
||||
# Check for a pending interaction.complete/completed event BEFORE the
|
||||
# finished check — otherwise the buffered completion event (which
|
||||
# carries the full text) would be dropped after `self.finished` is set.
|
||||
if hasattr(self, "_pending_interaction_complete"):
|
||||
pending: InteractionsAPIStreamingResponse = getattr(
|
||||
self, "_pending_interaction_complete"
|
||||
)
|
||||
delattr(self, "_pending_interaction_complete")
|
||||
return pending
|
||||
if self._pending_events:
|
||||
return self._pending_events.popleft()
|
||||
|
||||
if self.finished:
|
||||
raise StopIteration
|
||||
|
||||
# Use a loop instead of recursion to avoid stack overflow
|
||||
sync_iterator = cast(
|
||||
SyncResponsesAPIStreamingIterator, self.responses_stream_iterator
|
||||
)
|
||||
while True:
|
||||
try:
|
||||
# Get next chunk from responses API stream
|
||||
chunk = next(sync_iterator)
|
||||
|
||||
# Transform chunk (chunk is already a ResponsesAPIStreamingResponse)
|
||||
transformed = self._transform_responses_chunk_to_interactions_chunk(
|
||||
chunk
|
||||
)
|
||||
|
||||
if transformed:
|
||||
completion_event_type = (
|
||||
"interaction.complete"
|
||||
if self._use_legacy
|
||||
else "interaction.completed"
|
||||
)
|
||||
stop_event_type = (
|
||||
"content.stop" if self._use_legacy else "step.stop"
|
||||
)
|
||||
# If content was started, send the stop event before the completion event.
|
||||
if (
|
||||
self.finished
|
||||
and self.sent_content_start
|
||||
and transformed.event_type == completion_event_type
|
||||
):
|
||||
stop_kwargs: Dict[str, Any] = {
|
||||
"event_type": stop_event_type,
|
||||
"index": 0,
|
||||
}
|
||||
if self._use_legacy:
|
||||
stop_kwargs["id"] = transformed.id
|
||||
stop_kwargs["object"] = "content"
|
||||
stop_kwargs["delta"] = {
|
||||
"type": "text",
|
||||
"text": self.collected_text,
|
||||
}
|
||||
stop_chunk = InteractionsAPIStreamingResponse(**stop_kwargs)
|
||||
self._pending_interaction_complete = transformed
|
||||
return stop_chunk
|
||||
return transformed
|
||||
|
||||
# If no transformation, continue to next chunk (loop continues)
|
||||
|
||||
except StopIteration:
|
||||
self.finished = True
|
||||
self._pending_events.extend(self._build_terminal_events_on_eof())
|
||||
if self._pending_events:
|
||||
return self._pending_events.popleft()
|
||||
raise
|
||||
|
||||
# Send final stop event if content was started
|
||||
if self.sent_content_start:
|
||||
stop_event_type = (
|
||||
"content.stop" if self._use_legacy else "step.stop"
|
||||
)
|
||||
stop_kwargs = {
|
||||
"event_type": stop_event_type,
|
||||
"index": 0,
|
||||
}
|
||||
if self._use_legacy:
|
||||
stop_kwargs["object"] = "content"
|
||||
stop_kwargs["delta"] = {
|
||||
"type": "text",
|
||||
"text": self.collected_text,
|
||||
}
|
||||
return InteractionsAPIStreamingResponse(**stop_kwargs)
|
||||
|
||||
raise StopIteration
|
||||
events = self._events_for_chunk(chunk)
|
||||
if events:
|
||||
self._pending_events.extend(events)
|
||||
return self._pending_events.popleft()
|
||||
|
||||
def __aiter__(self) -> AsyncIterator[InteractionsAPIStreamingResponse]:
|
||||
"""Async iterator implementation."""
|
||||
return self
|
||||
|
||||
async def __anext__(self) -> InteractionsAPIStreamingResponse:
|
||||
"""Get next chunk in async mode."""
|
||||
# Check for a pending interaction.complete/completed event BEFORE the
|
||||
# finished check — otherwise the buffered completion event (which
|
||||
# carries the full text) would be dropped after `self.finished` is set.
|
||||
if hasattr(self, "_pending_interaction_complete"):
|
||||
pending: InteractionsAPIStreamingResponse = getattr(
|
||||
self, "_pending_interaction_complete"
|
||||
)
|
||||
delattr(self, "_pending_interaction_complete")
|
||||
return pending
|
||||
if self._pending_events:
|
||||
return self._pending_events.popleft()
|
||||
|
||||
if self.finished:
|
||||
raise StopAsyncIteration
|
||||
|
||||
# Use a loop instead of recursion to avoid stack overflow
|
||||
async_iterator = cast(
|
||||
ResponsesAPIStreamingIterator, self.responses_stream_iterator
|
||||
)
|
||||
while True:
|
||||
try:
|
||||
# Get next chunk from responses API stream
|
||||
chunk = await async_iterator.__anext__()
|
||||
|
||||
# Transform chunk (chunk is already a ResponsesAPIStreamingResponse)
|
||||
transformed = self._transform_responses_chunk_to_interactions_chunk(
|
||||
chunk
|
||||
)
|
||||
|
||||
if transformed:
|
||||
completion_event_type = (
|
||||
"interaction.complete"
|
||||
if self._use_legacy
|
||||
else "interaction.completed"
|
||||
)
|
||||
stop_event_type = (
|
||||
"content.stop" if self._use_legacy else "step.stop"
|
||||
)
|
||||
# If content was started, send the stop event before the completion event.
|
||||
if (
|
||||
self.finished
|
||||
and self.sent_content_start
|
||||
and transformed.event_type == completion_event_type
|
||||
):
|
||||
stop_kwargs_async: Dict[str, Any] = {
|
||||
"event_type": stop_event_type,
|
||||
"index": 0,
|
||||
}
|
||||
if self._use_legacy:
|
||||
stop_kwargs_async["id"] = transformed.id
|
||||
stop_kwargs_async["object"] = "content"
|
||||
stop_kwargs_async["delta"] = {
|
||||
"type": "text",
|
||||
"text": self.collected_text,
|
||||
}
|
||||
stop_chunk = InteractionsAPIStreamingResponse(
|
||||
**stop_kwargs_async
|
||||
)
|
||||
self._pending_interaction_complete = transformed
|
||||
return stop_chunk
|
||||
return transformed
|
||||
|
||||
# If no transformation, continue to next chunk (loop continues)
|
||||
|
||||
except StopAsyncIteration:
|
||||
self.finished = True
|
||||
self._pending_events.extend(self._build_terminal_events_on_eof())
|
||||
if self._pending_events:
|
||||
return self._pending_events.popleft()
|
||||
raise
|
||||
|
||||
# Send final stop event if content was started
|
||||
if self.sent_content_start:
|
||||
stop_event_type = (
|
||||
"content.stop" if self._use_legacy else "step.stop"
|
||||
)
|
||||
stop_kwargs_async = {
|
||||
"event_type": stop_event_type,
|
||||
"index": 0,
|
||||
}
|
||||
if self._use_legacy:
|
||||
stop_kwargs_async["object"] = "content"
|
||||
stop_kwargs_async["delta"] = {
|
||||
"type": "text",
|
||||
"text": self.collected_text,
|
||||
}
|
||||
return InteractionsAPIStreamingResponse(**stop_kwargs_async)
|
||||
events = self._events_for_chunk(chunk)
|
||||
if events:
|
||||
self._pending_events.extend(events)
|
||||
return self._pending_events.popleft()
|
||||
|
||||
raise StopAsyncIteration
|
||||
# ------------------------------------------------------------------
|
||||
# Backwards-compatible single-chunk transform (used by tests and any
|
||||
# external callers that drove the iterator chunk-by-chunk pre-fix).
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _transform_responses_chunk_to_interactions_chunk(
|
||||
self,
|
||||
responses_chunk: ResponsesAPIStreamingResponse,
|
||||
) -> Optional[InteractionsAPIStreamingResponse]:
|
||||
"""
|
||||
Compatibility shim: returns the *first* event produced for this chunk
|
||||
and queues any remaining events on ``self._pending_events`` so they
|
||||
are surfaced on subsequent calls/iterations.
|
||||
|
||||
Prefer ``_events_for_chunk`` in new code.
|
||||
"""
|
||||
events = self._events_for_chunk(responses_chunk)
|
||||
if not events:
|
||||
return None
|
||||
first = events[0]
|
||||
if len(events) > 1:
|
||||
self._pending_events.extend(events[1:])
|
||||
return first
|
||||
|
||||
@ -25,6 +25,7 @@ from litellm.llms.gemini.interactions.transformation import (
|
||||
)
|
||||
from litellm.types.llms.openai import (
|
||||
OutputTextDeltaEvent,
|
||||
ResponseCompletedEvent,
|
||||
ResponseCreatedEvent,
|
||||
)
|
||||
from litellm.types.router import GenericLiteLLMParams
|
||||
@ -151,7 +152,12 @@ class TestTransformRequest:
|
||||
request_body = config.transform_request(
|
||||
model=None,
|
||||
agent="my-custom-slides-agent",
|
||||
input=[{"type": "text", "text": "Create a 5-slide presentation about AI trends."}],
|
||||
input=[
|
||||
{
|
||||
"type": "text",
|
||||
"text": "Create a 5-slide presentation about AI trends.",
|
||||
}
|
||||
],
|
||||
optional_params={
|
||||
"environment": "remote",
|
||||
"stream": False,
|
||||
@ -306,7 +312,55 @@ class TestStreamingIterator:
|
||||
assert chunk.id == "resp_123"
|
||||
|
||||
def test_text_delta_sequence_new_schema(self):
|
||||
"""First two OutputTextDeltaEvents emit created + step.start; third emits step.delta."""
|
||||
"""First chunk yields created + step.start + step.delta; later chunks yield step.delta."""
|
||||
it = self._make_iterator(use_legacy=False)
|
||||
|
||||
first_events = it._events_for_chunk(self._make_text_delta("Hello"))
|
||||
assert [e.event_type for e in first_events] == [
|
||||
"interaction.created",
|
||||
"step.start",
|
||||
"step.delta",
|
||||
]
|
||||
assert first_events[-1].delta == {"type": "text", "text": "Hello"}
|
||||
assert it.sent_interaction_start is True
|
||||
assert it.sent_content_start is True
|
||||
|
||||
second_events = it._events_for_chunk(self._make_text_delta(" World"))
|
||||
assert [e.event_type for e in second_events] == ["step.delta"]
|
||||
assert second_events[0].delta == {"type": "text", "text": " World"}
|
||||
|
||||
third_events = it._events_for_chunk(self._make_text_delta("!"))
|
||||
assert [e.event_type for e in third_events] == ["step.delta"]
|
||||
assert third_events[0].delta == {"type": "text", "text": "!"}
|
||||
|
||||
def test_text_delta_sequence_legacy_schema(self):
|
||||
"""Legacy: first chunk yields interaction.start + content.start + content.delta."""
|
||||
it = self._make_iterator(use_legacy=True)
|
||||
|
||||
first_events = it._events_for_chunk(self._make_text_delta("Hello"))
|
||||
assert [e.event_type for e in first_events] == [
|
||||
"interaction.start",
|
||||
"content.start",
|
||||
"content.delta",
|
||||
]
|
||||
assert first_events[-1].delta == {"type": "text", "text": "Hello"}
|
||||
|
||||
second_events = it._events_for_chunk(self._make_text_delta(" World"))
|
||||
assert [e.event_type for e in second_events] == ["content.delta"]
|
||||
assert second_events[0].delta == {"type": "text", "text": " World"}
|
||||
|
||||
def test_first_text_delta_without_item_id_uses_fallback_id(self):
|
||||
it = self._make_iterator(use_legacy=False)
|
||||
event = self._make_text_delta("Hi")
|
||||
event.item_id = None
|
||||
|
||||
events = it._events_for_chunk(event)
|
||||
|
||||
assert events[0].event_type == "interaction.created"
|
||||
assert events[0].id == f"interaction_{id(it)}"
|
||||
|
||||
def test_first_text_delta_emits_text_via_compat_shim(self):
|
||||
"""The legacy single-chunk shim must surface the synthetic events AND the delta."""
|
||||
it = self._make_iterator(use_legacy=False)
|
||||
|
||||
first = it._transform_responses_chunk_to_interactions_chunk(
|
||||
@ -314,57 +368,134 @@ class TestStreamingIterator:
|
||||
)
|
||||
assert first is not None
|
||||
assert first.event_type == "interaction.created"
|
||||
assert it.sent_interaction_start is True
|
||||
assert it.sent_content_start is False
|
||||
|
||||
second = it._transform_responses_chunk_to_interactions_chunk(
|
||||
self._make_text_delta(" World")
|
||||
)
|
||||
second = it.__next__() if it._pending_events else None
|
||||
assert second is not None
|
||||
assert second.event_type == "step.start"
|
||||
assert it.sent_content_start is True
|
||||
|
||||
third = it._transform_responses_chunk_to_interactions_chunk(
|
||||
self._make_text_delta("!")
|
||||
)
|
||||
third = it.__next__() if it._pending_events else None
|
||||
assert third is not None
|
||||
assert third.event_type == "step.delta"
|
||||
assert third.delta == {"type": "text", "text": "!"}
|
||||
assert third.delta == {"type": "text", "text": "Hello"}
|
||||
|
||||
def test_text_delta_sequence_legacy_schema(self):
|
||||
"""Legacy: interaction.start → content.start → content.delta."""
|
||||
it = self._make_iterator(use_legacy=True)
|
||||
|
||||
first = it._transform_responses_chunk_to_interactions_chunk(
|
||||
self._make_text_delta("Hello")
|
||||
)
|
||||
assert first is not None
|
||||
assert first.event_type == "interaction.start"
|
||||
|
||||
second = it._transform_responses_chunk_to_interactions_chunk(
|
||||
self._make_text_delta(" World")
|
||||
)
|
||||
assert second is not None
|
||||
assert second.event_type == "content.start"
|
||||
assert second.delta == {"type": "text", "text": ""}
|
||||
|
||||
third = it._transform_responses_chunk_to_interactions_chunk(
|
||||
self._make_text_delta("!")
|
||||
)
|
||||
assert third is not None
|
||||
assert third.event_type == "content.delta"
|
||||
assert third.delta == {"type": "text", "text": "!"}
|
||||
|
||||
def test_first_text_delta_without_item_id_uses_fallback_id(self):
|
||||
def test_response_created_then_text_delta_emits_step_start_and_delta(self):
|
||||
"""Realistic flow: response.created arrives first, then text delta."""
|
||||
it = self._make_iterator(use_legacy=False)
|
||||
event = self._make_text_delta("Hi")
|
||||
event.item_id = None
|
||||
|
||||
chunk = it._transform_responses_chunk_to_interactions_chunk(event)
|
||||
first = it._events_for_chunk(self._make_response_created())
|
||||
assert [e.event_type for e in first] == ["interaction.created"]
|
||||
|
||||
assert chunk is not None
|
||||
assert chunk.event_type == "interaction.created"
|
||||
assert chunk.id == f"interaction_{id(it)}"
|
||||
second = it._events_for_chunk(self._make_text_delta("Hello"))
|
||||
assert [e.event_type for e in second] == ["step.start", "step.delta"]
|
||||
assert second[-1].delta == {"type": "text", "text": "Hello"}
|
||||
|
||||
def test_no_text_token_is_dropped_during_streaming(self):
|
||||
"""Concatenated step.delta payloads must equal the upstream text."""
|
||||
it = self._make_iterator(use_legacy=False)
|
||||
|
||||
chunks = ["Hello", " ", "world", "!"]
|
||||
emitted_text = ""
|
||||
for c in chunks:
|
||||
for ev in it._events_for_chunk(self._make_text_delta(c)):
|
||||
if ev.event_type == "step.delta":
|
||||
assert ev.delta is not None
|
||||
emitted_text += ev.delta["text"]
|
||||
|
||||
assert emitted_text == "Hello world!"
|
||||
|
||||
def test_stop_iteration_fallback_emits_completion_event(self):
|
||||
"""If upstream ends without ResponseCompletedEvent, terminal events still flow."""
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
text_event = self._make_text_delta("hi")
|
||||
sync_iter = MagicMock()
|
||||
sync_iter.__iter__ = lambda self: self
|
||||
sync_iter.__next__ = MagicMock(side_effect=[text_event, StopIteration])
|
||||
|
||||
original = litellm.use_legacy_interactions_schema
|
||||
litellm.use_legacy_interactions_schema = False
|
||||
try:
|
||||
it = LiteLLMResponsesInteractionsStreamingIterator(
|
||||
model="gpt-5.4",
|
||||
litellm_custom_stream_wrapper=sync_iter,
|
||||
request_input="hi",
|
||||
optional_params={},
|
||||
)
|
||||
finally:
|
||||
litellm.use_legacy_interactions_schema = original
|
||||
|
||||
emitted: list = []
|
||||
try:
|
||||
while True:
|
||||
emitted.append(next(it))
|
||||
except StopIteration:
|
||||
pass
|
||||
|
||||
event_types = [e.event_type for e in emitted]
|
||||
assert event_types == [
|
||||
"interaction.created",
|
||||
"step.start",
|
||||
"step.delta",
|
||||
"step.stop",
|
||||
"interaction.completed",
|
||||
]
|
||||
terminal = emitted[-1]
|
||||
assert terminal.steps == [
|
||||
{
|
||||
"type": "model_output",
|
||||
"content": [{"type": "text", "text": "hi"}],
|
||||
}
|
||||
]
|
||||
# EOF-flushed terminal event must carry the same id as interaction.created.
|
||||
assert terminal.id == emitted[0].id == "item_1"
|
||||
|
||||
def test_response_completed_emits_stop_then_completion(self):
|
||||
"""ResponseCompletedEvent expands into step.stop + interaction.completed."""
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
text_event = self._make_text_delta("hi")
|
||||
completed = MagicMock(spec=ResponseCompletedEvent)
|
||||
completed.response = MagicMock(id="resp_999")
|
||||
|
||||
sync_iter = MagicMock()
|
||||
sync_iter.__iter__ = lambda self: self
|
||||
sync_iter.__next__ = MagicMock(side_effect=[text_event, completed])
|
||||
|
||||
original = litellm.use_legacy_interactions_schema
|
||||
litellm.use_legacy_interactions_schema = False
|
||||
try:
|
||||
it = LiteLLMResponsesInteractionsStreamingIterator(
|
||||
model="gpt-5.4",
|
||||
litellm_custom_stream_wrapper=sync_iter,
|
||||
request_input="hi",
|
||||
optional_params={},
|
||||
)
|
||||
finally:
|
||||
litellm.use_legacy_interactions_schema = original
|
||||
|
||||
emitted: list = []
|
||||
try:
|
||||
while True:
|
||||
emitted.append(next(it))
|
||||
except StopIteration:
|
||||
pass
|
||||
|
||||
event_types = [e.event_type for e in emitted]
|
||||
assert event_types == [
|
||||
"interaction.created",
|
||||
"step.start",
|
||||
"step.delta",
|
||||
"step.stop",
|
||||
"interaction.completed",
|
||||
]
|
||||
# StopIteration fallback path must NOT add a duplicate completion event.
|
||||
assert event_types.count("interaction.completed") == 1
|
||||
# When the stream starts directly with a text delta (no preceding
|
||||
# response.created), the terminal events must reuse the id derived from
|
||||
# the first chunk's item_id rather than switching to response.id, so
|
||||
# consumers can correlate the start and completion events by id.
|
||||
assert emitted[0].id == "item_1"
|
||||
assert emitted[-1].id == "item_1"
|
||||
|
||||
|
||||
class TestInteractionOperationUrls:
|
||||
|
||||
Loading…
Reference in New Issue
Block a user