scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-15 09:12:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 311

Merge 2206535ead into 200ecdb6e6

Browse source
This commit is contained in:
Mohamed Jadib 2026-11-03 15:14:16 -05:00 • committed by GitHub
commit ef875ea66b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 111 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

66
litellm/llms/anthropic/messages/adapters/streaming_iterator.py
View file

@ -674,6 +674,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
self._paused_turn: bool = False
self._claude_refusal_stop_details: AnthropicStopDetails | None = None
self.sent_compaction_block: bool = False
self._current_chunk_is_finish: bool = False
# Per-phase flags so the compaction block's start/delta/stop events
# are emitted (and the public state machine is advanced) in
# lock-step with the caller actually consuming each event. Pre-
@ -935,7 +936,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
continue
should_start_new_block = self._should_start_new_content_block(chunk)
is_opening_first_block = self.sent_content_block_start is False
is_opening_first_block = self.sent_content_block_start is False and not self._current_chunk_is_finish
if is_opening_first_block and self._is_blank_delta(chunk):
continue
if is_opening_first_block:
@ -1035,19 +1036,22 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
if processed_chunk["type"] == "message_delta" and self.sent_content_block_finish is False:
# Queue both the content_block_stop and the message_delta
self.chunk_queue.append(
{
"type": "content_block_stop",
"index": self.current_content_block_index,
}
)
self.sent_content_block_finish = True
if self.sent_content_block_start:
self.chunk_queue.append(
{
"type": "content_block_stop",
"index": self.current_content_block_index,
}
)
self.sent_content_block_finish = True
if processed_chunk.get("delta", {}).get("stop_reason") is not None:
self.holding_stop_reason_chunk = processed_chunk
else:
processed_chunk = self._augment_message_delta_usage(processed_chunk)
self.chunk_queue.append(processed_chunk)
return self.chunk_queue.popleft()
if self.chunk_queue:
return self.chunk_queue.popleft()
continue
elif self.holding_chunk is not None:
self.chunk_queue.append(self.holding_chunk)
if processed_chunk.get("type") == "message_delta":
@ -1079,7 +1083,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
# valid Anthropic order (... -> content_block_stop ->
# message_delta). Emit ``content_block_stop`` here if
# the active content block was not already closed.
if not self.sent_content_block_finish:
if self.sent_content_block_start and not self.sent_content_block_finish:
self.chunk_queue.append(
{
"type": "content_block_stop",
@ -1108,14 +1112,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
# Anthropic SSE ordering is preserved (content_block_stop ->
# message_delta).
if self.holding_stop_reason_chunk is not None:
if not self.sent_content_block_finish:
self.sent_content_block_finish = True
self.chunk_queue.append(self._augment_message_delta_usage(self.holding_stop_reason_chunk))
self.holding_stop_reason_chunk = None
return {
"type": "content_block_stop",
"index": self.current_content_block_index,
}
held: Final = self._augment_message_delta_usage(self.holding_stop_reason_chunk)
self.holding_stop_reason_chunk = None
return held
@ -1170,7 +1166,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
continue
should_start_new_block = self._should_start_new_content_block(chunk)
is_opening_first_block = self.sent_content_block_start is False
is_opening_first_block = self.sent_content_block_start is False and not self._current_chunk_is_finish
if is_opening_first_block and self._is_blank_delta(chunk):
continue
if is_opening_first_block:
@ -1262,19 +1258,22 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
if processed_chunk["type"] == "message_delta" and self.sent_content_block_finish is False:
# Queue both the content_block_stop and the holding chunk
self.chunk_queue.append(
{
"type": "content_block_stop",
"index": self.current_content_block_index,
}
)
self.sent_content_block_finish = True
if self.sent_content_block_start:
self.chunk_queue.append(
{
"type": "content_block_stop",
"index": self.current_content_block_index,
}
)
self.sent_content_block_finish = True
if processed_chunk.get("delta", {}).get("stop_reason") is not None:
self.holding_stop_reason_chunk = processed_chunk
else:
processed_chunk = self._augment_message_delta_usage(processed_chunk)
self.chunk_queue.append(processed_chunk)
return self.chunk_queue.popleft()
if self.chunk_queue:
return self.chunk_queue.popleft()
continue
elif self.holding_chunk is not None:
# Queue both chunks
self.chunk_queue.append(self.holding_chunk)
@ -1307,7 +1306,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
# valid Anthropic order (... -> content_block_stop ->
# message_delta). Emit ``content_block_stop`` here if
# the active content block was not already closed.
if not self.sent_content_block_finish:
if self.sent_content_block_start and not self.sent_content_block_finish:
self.chunk_queue.append(
{
"type": "content_block_stop",
@ -1341,14 +1340,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
# Anthropic SSE ordering is preserved (content_block_stop ->
# message_delta).
if self.holding_stop_reason_chunk is not None:
if not self.sent_content_block_finish:
self.sent_content_block_finish = True
self.chunk_queue.append(self._augment_message_delta_usage(self.holding_stop_reason_chunk))
self.holding_stop_reason_chunk = None
return {
"type": "content_block_stop",
"index": self.current_content_block_index,
}
held: Final = self._augment_message_delta_usage(self.holding_stop_reason_chunk)
self.holding_stop_reason_chunk = None
return held
@ -1544,7 +1535,8 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
from .transformation import LiteLLMAnthropicMessagesAdapter
self._record_anthropic_stop_fields(chunk.choices[0].delta)
if chunk.choices[0].finish_reason is not None:
self._current_chunk_is_finish = chunk.choices[0].finish_reason is not None
if self._current_chunk_is_finish:
return False
refusal_text: Final = refusal_reply_text(chunk.choices[0].delta)

87
tests/unit/llms/anthropic/messages/adapters/test_streaming_iterator_first_delta.py
View file

@ -817,9 +817,9 @@ async def test_role_only_lead_chunk_does_not_open_text_block_before_reasoning_co
def test_finish_first_chunk_is_not_deferred_sync():
"""A stream whose first upstream chunk is already the finish event must not
be skipped by the blank-delta deferral. ``_is_blank_delta`` returns False
for a finish chunk so the message_delta still flows (with an empty text
block opened and closed first); without that guard the deferral would drop
the terminal event entirely.
for a finish chunk so the message_delta still flows directly without
opening/closing a spurious empty text block (Issue #47576); without that
guard the deferral would drop the terminal event entirely.
"""
chunks = [_make_chunk(Delta(content=None), finish_reason="stop")]
wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x")
@ -827,8 +827,6 @@ def test_finish_first_chunk_is_not_deferred_sync():
assert [e["type"] for e in events] == [
"message_start",
"content_block_start",
"content_block_stop",
"message_delta",
"message_stop",
]
@ -1534,3 +1532,82 @@ async def test_signed_thinking_blocks_keep_their_own_text_and_signature(is_async
assert _last_delta_types_per_thinking_block(events) == ("signature_delta", "signature_delta")
assert _text_deltas(events) == ["Done"]
_assert_deltas_match_their_block_type(events)
@pytest.mark.parametrize("is_async", [False, True])
@pytest.mark.asyncio
async def test_finish_only_first_chunk_emits_no_synthetic_content_block(is_async: bool):
"""A stream whose first choice chunk is finish-only (0-output stream) must
not synthesize a fake empty text block {"type": "text", "text": ""}.
The emitted sequence must strictly be message_start -> message_delta -> message_stop.
Regression test for issue #47576."""
chunks = [
_make_chunk(Delta(content=None), finish_reason="stop"),
]
if is_async:
wrapper = AnthropicStreamWrapper(completion_stream=_AsyncStream(chunks), model="claude-x")
events = await _drain_async(wrapper)
else:
wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x")
events = _drain_sync(wrapper)
event_types = [e.get("type") for e in events]
assert event_types == ["message_start", "message_delta", "message_stop"]
assert not any(e.get("type") == "content_block_start" for e in events)
assert not any(e.get("type") == "content_block_stop" for e in events)
message_delta = next(e for e in events if e.get("type") == "message_delta")
assert message_delta["delta"]["stop_reason"] == "end_turn"
@pytest.mark.parametrize("is_async", [False, True])
@pytest.mark.asyncio
async def test_finish_only_chunk_with_trailing_usage_emits_no_synthetic_content_block(is_async: bool):
"""A 0-output stream with a trailing usage chunk must not emit synthetic
content blocks and must correctly merge usage into the terminal message_delta.
Regression test for issue #47576."""
usage_chunk = MagicMock()
usage_chunk.choices = []
usage_chunk.usage = Usage(prompt_tokens=15, completion_tokens=0, total_tokens=15)
usage_chunk._hidden_params = {}
chunks = [
_make_chunk(Delta(content=None), finish_reason="stop"),
usage_chunk,
]
if is_async:
wrapper = AnthropicStreamWrapper(completion_stream=_AsyncStream(chunks), model="claude-x")
events = await _drain_async(wrapper)
else:
wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x")
events = _drain_sync(wrapper)
event_types = [e.get("type") for e in events]
assert event_types == ["message_start", "message_delta", "message_stop"]
assert not any(e.get("type") == "content_block_start" for e in events)
assert not any(e.get("type") == "content_block_stop" for e in events)
message_delta = next(e for e in events if e.get("type") == "message_delta")
assert message_delta["delta"]["stop_reason"] == "end_turn"
assert message_delta["usage"]["input_tokens"] == 15
assert message_delta["usage"]["output_tokens"] == 0
@pytest.mark.parametrize("is_async", [False, True])
@pytest.mark.asyncio
async def test_blank_deltas_before_finish_chunk_emit_no_synthetic_content_block(is_async: bool):
"""Blank deltas before the terminal chunk must be skipped, and the finish
chunk must not open a synthetic content block.
Regression test for issue #47576."""
chunks = [
_make_chunk(Delta(content=None)),
_make_chunk(Delta(content=None), finish_reason="stop"),
]
if is_async:
wrapper = AnthropicStreamWrapper(completion_stream=_AsyncStream(chunks), model="claude-x")
events = await _drain_async(wrapper)
else:
wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x")
events = _drain_sync(wrapper)
event_types = [e.get("type") for e in events]
assert event_types == ["message_start", "message_delta", "message_stop"]
assert not any(e.get("type") == "content_block_start" for e in events)