mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-09 03:18:44 +00:00
fix(responses): emit text delta when a chunk carries reasoning and content
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Devin AI
parent
e9d1ea59fb
commit
2bd8020d83
2 changed files with 189 additions and 85 deletions
|
|
@ -135,6 +135,80 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
|
|||
|
||||
return delta.content or delta.function_call or delta.tool_calls or chunk.choices[0].finish_reason is not None
|
||||
|
||||
@staticmethod
|
||||
def _split_reasoning_and_content(
|
||||
chunk: ModelResponseStream,
|
||||
) -> tuple[ModelResponseStream, ...]:
|
||||
"""
|
||||
Providers can put `reasoning_content` and `content` in the same chunk. The Responses
|
||||
API has no combined event, so yield a reasoning-only chunk followed by a content-only
|
||||
one and let the per-chunk pipeline emit both deltas in source order.
|
||||
"""
|
||||
if not chunk.choices:
|
||||
return (chunk,)
|
||||
delta: Final = chunk.choices[0].delta
|
||||
if not (getattr(delta, "reasoning_content", None) and delta.content):
|
||||
return (chunk,)
|
||||
|
||||
reasoning_part: Final = chunk.model_copy(deep=True)
|
||||
reasoning_part.choices[0].delta.content = None
|
||||
reasoning_part.choices[0].finish_reason = None
|
||||
|
||||
content_part: Final = chunk.model_copy(deep=True)
|
||||
content_part.choices[0].delta.reasoning_content = None
|
||||
content_part.choices[0].delta.thinking_blocks = None
|
||||
return (reasoning_part, content_part)
|
||||
|
||||
def _queue_reasoning_lifecycle_events(self, chunk: ModelResponseStream) -> None:
|
||||
if not self._reasoning_active or self._reasoning_done_emitted:
|
||||
return
|
||||
delta: Final = chunk.choices[0].delta if chunk.choices else None
|
||||
if delta and getattr(delta, "reasoning_content", None):
|
||||
self._accumulated_reasoning_content_parts.append(delta.reasoning_content)
|
||||
if not self._is_reasoning_end(chunk):
|
||||
return
|
||||
|
||||
reasoning_content: Final = "".join(self._accumulated_reasoning_content_parts)
|
||||
reasoning_item_id: Final = self._reasoning_item_id or self._cached_reasoning_item_id or f"rs_{uuid.uuid4()}"
|
||||
|
||||
self._sequence_number += 1
|
||||
text_done_event: Final = self.create_reasoning_summary_text_done_event(
|
||||
reasoning_item_id=reasoning_item_id,
|
||||
reasoning_content=reasoning_content,
|
||||
sequence_number=self._sequence_number,
|
||||
)
|
||||
self._sequence_number += 1
|
||||
part_done_event: Final = 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
|
||||
output_item_done_event: Final = 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([text_done_event, part_done_event, output_item_done_event])
|
||||
self._reasoning_done_emitted = True
|
||||
self._reasoning_active = False
|
||||
|
||||
def _queue_events_for_chunk(self, chunk: ModelResponseStream) -> None:
|
||||
for part in self._split_reasoning_and_content(chunk):
|
||||
self._ensure_output_item_for_chunk(part)
|
||||
self._queue_reasoning_lifecycle_events(part)
|
||||
response_api_chunk = self._transform_chat_completion_chunk_to_response_api_chunk(part)
|
||||
if response_api_chunk:
|
||||
self._pending_response_events.append(response_api_chunk)
|
||||
|
||||
def _accumulate_provider_specific_fields_from_chunk(self, chunk: ModelResponseStream) -> None:
|
||||
for src in (
|
||||
getattr(chunk, "provider_specific_fields", None),
|
||||
getattr(chunk.choices[0].delta if chunk.choices else None, "provider_specific_fields", None),
|
||||
):
|
||||
if src and isinstance(src, dict):
|
||||
self._merge_provider_specific_fields(src)
|
||||
|
||||
def _queue_tool_call_delta_events(self, tool_calls: object) -> None:
|
||||
"""
|
||||
Convert chat-completions streaming `tool_calls` deltas into Responses API streaming events.
|
||||
|
|
@ -805,71 +879,11 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
|
|||
chunk = await self.litellm_custom_stream_wrapper.__anext__()
|
||||
if chunk is not None:
|
||||
chunk = cast(ModelResponseStream, chunk)
|
||||
self._ensure_output_item_for_chunk(chunk)
|
||||
# Accumulate provider_specific_fields from chunk and delta
|
||||
for src in (
|
||||
getattr(chunk, "provider_specific_fields", None),
|
||||
getattr(
|
||||
chunk.choices[0].delta if chunk.choices else None,
|
||||
"provider_specific_fields",
|
||||
None,
|
||||
),
|
||||
):
|
||||
if src and isinstance(src, dict):
|
||||
self._merge_provider_specific_fields(src)
|
||||
# Proceed to transformation
|
||||
self._accumulate_provider_specific_fields_from_chunk(chunk)
|
||||
self.collected_chat_completion_chunks.append(
|
||||
self._snapshot_chunk_for_stream_chunk_builder(chunk)
|
||||
)
|
||||
if self._reasoning_active and not self._reasoning_done_emitted:
|
||||
# Incrementally accumulate reasoning content instead of
|
||||
# calling stream_chunk_builder on every chunk (O(n²))
|
||||
delta = chunk.choices[0].delta if chunk.choices else None
|
||||
if delta and hasattr(delta, "reasoning_content") and delta.reasoning_content:
|
||||
self._accumulated_reasoning_content_parts.append(delta.reasoning_content)
|
||||
if self._is_reasoning_end(chunk):
|
||||
reasoning_content = "".join(self._accumulated_reasoning_content_parts)
|
||||
|
||||
# Ensure we have a valid reasoning_item_id
|
||||
reasoning_item_id = (
|
||||
self._reasoning_item_id or self._cached_reasoning_item_id or f"rs_{uuid.uuid4()}"
|
||||
)
|
||||
|
||||
# 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(
|
||||
[
|
||||
text_done_event,
|
||||
part_done_event,
|
||||
reasoning_output_item_done_event,
|
||||
]
|
||||
)
|
||||
self._reasoning_done_emitted = True
|
||||
self._reasoning_active = False
|
||||
|
||||
response_api_chunk = self._transform_chat_completion_chunk_to_response_api_chunk(chunk)
|
||||
if response_api_chunk:
|
||||
self._pending_response_events.append(response_api_chunk)
|
||||
self._queue_events_for_chunk(chunk)
|
||||
|
||||
if self._pending_response_events:
|
||||
return self._pending_response_events.pop(0)
|
||||
|
|
@ -902,32 +916,15 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
|
|||
if self._pending_tool_events:
|
||||
return self._pending_tool_events.pop(0)
|
||||
try:
|
||||
chunk = self.litellm_custom_stream_wrapper.__next__()
|
||||
self._ensure_output_item_for_chunk(chunk)
|
||||
# Accumulate provider_specific_fields from chunk and delta
|
||||
for src in (
|
||||
getattr(chunk, "provider_specific_fields", None),
|
||||
getattr(
|
||||
chunk.choices[0].delta if chunk.choices else None,
|
||||
"provider_specific_fields",
|
||||
None,
|
||||
),
|
||||
):
|
||||
if src and isinstance(src, dict):
|
||||
self._merge_provider_specific_fields(src)
|
||||
chunk = cast(ModelResponseStream, self.litellm_custom_stream_wrapper.__next__())
|
||||
self._accumulate_provider_specific_fields_from_chunk(chunk)
|
||||
# Always snapshot before returning any pending events so that
|
||||
# finish_reason (e.g. content_filter) is captured even when
|
||||
# _ensure_output_item_for_chunk queues events on the same chunk.
|
||||
# This mirrors the async path (see __anext__).
|
||||
self.collected_chat_completion_chunks.append(
|
||||
self._snapshot_chunk_for_stream_chunk_builder(cast(ModelResponseStream, chunk))
|
||||
)
|
||||
# Emit any just-queued output_item event
|
||||
# finish_reason (e.g. content_filter) is captured even when the
|
||||
# chunk also queues events. This mirrors the async path (see __anext__).
|
||||
self.collected_chat_completion_chunks.append(self._snapshot_chunk_for_stream_chunk_builder(chunk))
|
||||
self._queue_events_for_chunk(chunk)
|
||||
if self._pending_response_events:
|
||||
return self._pending_response_events.pop(0)
|
||||
response_api_chunk = self._transform_chat_completion_chunk_to_response_api_chunk(chunk)
|
||||
if response_api_chunk:
|
||||
return response_api_chunk
|
||||
# Otherwise, loop to next chunk
|
||||
except StopIteration:
|
||||
return self.common_done_event_logic(sync_mode=True)
|
||||
|
|
|
|||
|
|
@ -2,8 +2,11 @@
|
|||
Test reasoning content preservation in Responses API transformation
|
||||
"""
|
||||
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm.responses.litellm_completion_transformation.streaming_iterator import (
|
||||
LiteLLMCompletionStreamingIterator,
|
||||
)
|
||||
|
|
@ -264,6 +267,110 @@ class TestReasoningContentFinalResponse:
|
|||
assert reasoning_items[0].content[0].text == "Reasoning for first answer"
|
||||
|
||||
|
||||
class _FakeChatCompletionStream:
|
||||
"""Minimal CustomStreamWrapper stand-in that replays pre-built chunks."""
|
||||
|
||||
def __init__(self, chunks):
|
||||
self._chunks = list(chunks)
|
||||
self.logging_obj = SimpleNamespace(
|
||||
_response_cost_calculator=lambda result: 0.0,
|
||||
model_call_details={},
|
||||
)
|
||||
|
||||
def __aiter__(self):
|
||||
return self
|
||||
|
||||
async def __anext__(self):
|
||||
if not self._chunks:
|
||||
raise StopAsyncIteration
|
||||
return self._chunks.pop(0)
|
||||
|
||||
def __iter__(self):
|
||||
return self
|
||||
|
||||
def __next__(self):
|
||||
if not self._chunks:
|
||||
raise StopIteration
|
||||
return self._chunks.pop(0)
|
||||
|
||||
|
||||
def _reasoning_and_text_chunks():
|
||||
def make(content=None, reasoning=None, finish_reason=None):
|
||||
return ModelResponseStream(
|
||||
id="chatcmpl-combined",
|
||||
created=1234567890,
|
||||
model="test-model",
|
||||
object="chat.completion.chunk",
|
||||
choices=[
|
||||
StreamingChoices(
|
||||
finish_reason=finish_reason,
|
||||
index=0,
|
||||
delta=Delta(content=content, role="assistant", reasoning_content=reasoning),
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
return [
|
||||
make(reasoning="Thinking hard"),
|
||||
make(
|
||||
reasoning=" done thinking.",
|
||||
content="MARKER_ALPHA begin middle end",
|
||||
finish_reason="stop",
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
def _new_iterator(chunks):
|
||||
return LiteLLMCompletionStreamingIterator(
|
||||
model="test-model",
|
||||
litellm_custom_stream_wrapper=_FakeChatCompletionStream(chunks),
|
||||
request_input="Test input",
|
||||
responses_api_request={},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_combined_reasoning_and_content_chunk_emits_both_deltas_async():
|
||||
"""A chunk carrying reasoning_content and content must yield both deltas, in source order."""
|
||||
iterator = _new_iterator(_reasoning_and_text_chunks())
|
||||
|
||||
types = []
|
||||
reasoning_deltas = []
|
||||
text_deltas = []
|
||||
done_text = None
|
||||
async for event in iterator:
|
||||
types.append(event.type)
|
||||
if event.type == "response.reasoning_summary_text.delta":
|
||||
reasoning_deltas.append(event.delta)
|
||||
elif event.type == "response.output_text.delta":
|
||||
text_deltas.append(event.delta)
|
||||
elif event.type == "response.output_text.done":
|
||||
done_text = event.text
|
||||
|
||||
assert "".join(reasoning_deltas) == "Thinking hard done thinking."
|
||||
assert "".join(text_deltas) == "MARKER_ALPHA begin middle end"
|
||||
assert "".join(text_deltas) == done_text
|
||||
assert types.index("response.reasoning_summary_text.delta") < types.index(
|
||||
"response.output_text.delta"
|
||||
)
|
||||
assert types.index("response.reasoning_summary_text.done") < types.index(
|
||||
"response.output_text.delta"
|
||||
)
|
||||
|
||||
|
||||
def test_combined_reasoning_and_content_chunk_emits_both_deltas_sync():
|
||||
"""Same guarantee on the sync iterator path."""
|
||||
iterator = _new_iterator(_reasoning_and_text_chunks())
|
||||
|
||||
text_deltas = [
|
||||
event.delta
|
||||
for event in iterator
|
||||
if getattr(event, "type", None) == "response.output_text.delta"
|
||||
]
|
||||
|
||||
assert "".join(text_deltas) == "MARKER_ALPHA begin middle end"
|
||||
|
||||
|
||||
def test_streaming_chunk_id_raw():
|
||||
"""Test that streaming chunk IDs are raw (not encoded) to match OpenAI format"""
|
||||
chunk = ModelResponseStream(
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue