scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-13 23:11:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 152

fix(anthropic-adapter): re-emit first delta on streaming content-block transitions (#30024)

Browse source 
* fix(anthropic-adapter): re-emit first delta on streaming content-block transitions

The `/v1/messages` -> `/v1/chat/completions` streaming adapter
(`AnthropicStreamWrapper`) silently dropped the first non-empty delta of
every content block that started via a *transition* (e.g. text -> tool_use ->
text, text -> thinking).

When an upstream chunk both triggers a new content block (its type differs
from the active block) and carries that block's first delta, the wrapper
emitted `content_block_stop` -> `content_block_start` and then only re-queued
the trigger chunk when it was an `input_json_delta` (bundled tool args). The
synthesized `content_block_start` always carries an empty body, so the first
`text_delta` / `thinking_delta` was lost — the client output started from the
second token (e.g. "Hi, how can I help you?" rendered as ", how can I help
you?", or text resuming after a tool call lost its first sentence). This is
especially visible with Claude Code-style clients that consume Anthropic
Messages streaming events strictly.

Fix: re-queue the trigger chunk's translated delta whenever it carries
non-empty content (text/thinking/signature/tool args), via a shared
`_trigger_delta_has_content` helper used by both the sync and async paths.
Empty trigger deltas are still suppressed so no spurious empty
`content_block_delta` is introduced.

Fixes #30014

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* test(anthropic-adapter): cover all _trigger_delta_has_content branches

Add a direct parametrized unit test for the re-emit predicate so every delta
type (text/input_json/thinking/signature), the empty-payload guards, and the
malformed/non-delta cases are exercised independently of upstream chunk
translation. Raises patch coverage for the new helper.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: shin-berri <shin-laptop@berri.ai>
Co-authored-by: yuneng-jiang <yuneng@berri.ai>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Martín Alcalá Rubí 2026-06-11 09:02:38 -03:00 • committed by GitHub
parent 706ea8f3d6
commit 4782aa953b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 388 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

87
litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py
View file

@ -469,12 +469,16 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
if should_start_new_block and not self.sent_content_block_finish:
# Queue the sequence: content_block_stop -> content_block_start
# For text blocks the trigger chunk is not emitted as a separate
# delta because content_block_start carries the information.
# For tool_use blocks we must also emit the trigger chunk's delta
# when it carries input_json_delta data, because some providers
# (e.g. xAI, Gemini) include tool arguments in the same streaming
# chunk as the function name/id.
# -> (optionally) the trigger chunk's delta.
#
# The synthesized content_block_start always carries an
# empty body, so the chunk that *triggered* the transition
# also carries the new block's first delta. It must be
# re-emitted or the first token of the new block is lost.
# This applies to text_delta and thinking_delta (the first
# non-empty text/thinking token) as well as input_json_delta
# (providers like xAI/Gemini bundle tool arguments with the
# function name/id in a single chunk).
# 1. Stop current content block
self.chunk_queue.append(
@ -493,14 +497,9 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
}
)
# 3. If the trigger chunk carries tool argument data, queue it
# so the input_json_delta is not silently dropped.
if (
processed_chunk.get("type") == "content_block_delta"
and isinstance(processed_chunk.get("delta"), dict)
and processed_chunk["delta"].get("type") == "input_json_delta"
and processed_chunk["delta"].get("partial_json")
):
# 3. If the trigger chunk carries delta content, queue it
# so the first delta of the new block is not silently dropped.
if self._trigger_delta_has_content(processed_chunk):
self.chunk_queue.append(processed_chunk)
self.sent_content_block_finish = False
@ -711,12 +710,16 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
if not self.queued_usage_chunk:
if should_start_new_block and not self.sent_content_block_finish:
# Queue the sequence: content_block_stop -> content_block_start
# For text blocks the trigger chunk is not emitted as a separate
# delta because content_block_start carries the information.
# For tool_use blocks we must also emit the trigger chunk's delta
# when it carries input_json_delta data, because some providers
# (e.g. xAI, Gemini) include tool arguments in the same streaming
# chunk as the function name/id.
# -> (optionally) the trigger chunk's delta.
#
# The synthesized content_block_start always carries an
# empty body, so the chunk that *triggered* the transition
# also carries the new block's first delta. It must be
# re-emitted or the first token of the new block is lost.
# This applies to text_delta and thinking_delta (the
# first non-empty text/thinking token) as well as
# input_json_delta (providers like xAI/Gemini bundle tool
# arguments with the function name/id in a single chunk).
# 1. Stop current content block
self.chunk_queue.append(
@ -733,15 +736,9 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
}
)
# 3. If the trigger chunk carries tool argument data, queue it
# so the input_json_delta is not silently dropped.
if (
processed_chunk.get("type") == "content_block_delta"
and isinstance(processed_chunk.get("delta"), dict)
and processed_chunk["delta"].get("type")
== "input_json_delta"
and processed_chunk["delta"].get("partial_json")
):
# 3. If the trigger chunk carries delta content, queue it
# so the first delta of the new block is not silently dropped.
if self._trigger_delta_has_content(processed_chunk):
self.chunk_queue.append(processed_chunk)
# Reset state for new block
@ -898,6 +895,38 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
def _increment_content_block_index(self):
self.current_content_block_index += 1
@staticmethod
def _trigger_delta_has_content(processed_chunk: Dict[str, Any]) -> bool:
"""Return True if a translated trigger chunk carries a non-empty
``content_block_delta`` payload that must be re-emitted after a
block transition.
When an upstream chunk both *triggers* a new content block (its type
differs from the active block) and *carries* delta content, that
content belongs to the new block. The synthesized
``content_block_start`` only ever carries an empty body — see
``_translate_streaming_openai_chunk_to_anthropic_content_block``,
which returns an empty ``TextBlock``/``ToolUseBlock``/thinking block —
so the trigger chunk's delta must be re-queued or the first token of
the new block (the first non-empty text/thinking delta, or bundled
tool arguments) is silently dropped.
"""
if processed_chunk.get("type") != "content_block_delta":
return False
delta = processed_chunk.get("delta")
if not isinstance(delta, dict):
return False
delta_type = delta.get("type")
if delta_type == "text_delta":
return bool(delta.get("text"))
if delta_type == "input_json_delta":
return bool(delta.get("partial_json"))
if delta_type == "thinking_delta":
return bool(delta.get("thinking"))
if delta_type == "signature_delta":
return bool(delta.get("signature"))
return False
def _should_start_new_content_block(self, chunk: "ModelResponseStream") -> bool:
"""
Determine if we should start a new content block based on the processed chunk.

312
tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_first_delta.py Normal file
View file

@ -0,0 +1,312 @@
"""
Regression tests for issue #30014.
When LiteLLM proxies ``client -> /v1/messages -> /v1/chat/completions`` and a
streaming chunk both *triggers* a new Anthropic content block (its type differs
from the active block) and *carries* the first delta of that new block, the
trigger chunk's delta must be re-emitted as a ``content_block_delta``.
The synthesized ``content_block_start`` always carries an empty body, so before
the fix the first non-empty ``text_delta`` of every transitioned block was
silently dropped — e.g. text resuming after a tool call started from the second
token ("The weather is nice." was lost, "Hi" rendered as ""). Bundled
``input_json_delta`` tool arguments were already preserved and must stay
preserved, and empty trigger deltas must not produce spurious events.
"""
import os
import sys
from typing import List, Optional
from unittest.mock import MagicMock
import pytest
sys.path.insert(0, os.path.abspath("../../../../.."))
from litellm.llms.anthropic.experimental_pass_through.adapters.streaming_iterator import (
AnthropicStreamWrapper,
)
from litellm.types.utils import (
ChatCompletionDeltaToolCall,
Delta,
Function,
StreamingChoices,
)
def _make_chunk(delta: Delta, finish_reason: Optional[str] = None) -> MagicMock:
chunk = MagicMock()
chunk.choices = [
StreamingChoices(
finish_reason=finish_reason,
index=0,
delta=delta,
logprobs=None,
)
]
chunk.usage = None
chunk._hidden_params = {}
return chunk
def _tool_chunk(
call_id: str, name: Optional[str], arguments: Optional[str]
) -> MagicMock:
return _make_chunk(
Delta(
content=None,
tool_calls=[
ChatCompletionDeltaToolCall(
id=call_id,
function=Function(name=name, arguments=arguments),
type="function",
index=0,
)
],
)
)
class _AsyncStream:
def __init__(self, items: List[MagicMock]):
self._it = iter(items)
def __aiter__(self):
return self
async def __anext__(self):
try:
return next(self._it)
except StopIteration:
raise StopAsyncIteration
def _drain_sync(wrapper: AnthropicStreamWrapper) -> List[dict]:
return list(wrapper)
async def _drain_async(wrapper: AnthropicStreamWrapper) -> List[dict]:
return [event async for event in wrapper]
def _text_deltas(events: List[dict]) -> List[str]:
return [
e["delta"]["text"]
for e in events
if e.get("type") == "content_block_delta"
and e["delta"].get("type") == "text_delta"
]
def _input_json_deltas(events: List[dict]) -> List[str]:
return [
e["delta"]["partial_json"]
for e in events
if e.get("type") == "content_block_delta"
and e["delta"].get("type") == "input_json_delta"
]
def test_first_text_delta_after_tool_use_is_not_dropped_sync():
"""A tool_use -> text transition (text resuming after a tool call) carries
the resumed text's first token in the trigger chunk. Without the fix it was
dropped, so "The weather is nice." vanished and the answer began at " Bye.".
"""
chunks = [
_make_chunk(Delta(content="Let me check.")),
_tool_chunk("call_1", "get_weather", '{"city":'),
_tool_chunk("call_1", None, ' "NY"}'),
_make_chunk(Delta(content="The weather is nice.")),
_make_chunk(Delta(content=" Bye.")),
_make_chunk(Delta(content=None), finish_reason="stop"),
]
wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x")
events = _drain_sync(wrapper)
assert _input_json_deltas(events) == ['{"city":', ' "NY"}']
assert _text_deltas(events) == [
"Let me check.",
"The weather is nice.",
" Bye.",
]
@pytest.mark.asyncio
async def test_first_text_delta_after_tool_use_is_not_dropped_async():
"""Async path mirrors the sync regression — the proxy serves the async
iterator, so it must preserve the first resumed text delta too.
"""
chunks = [
_make_chunk(Delta(content="Let me check.")),
_tool_chunk("call_1", "get_weather", '{"city": "NY"}'),
_make_chunk(Delta(content="The weather is nice.")),
_make_chunk(Delta(content=" Bye.")),
_make_chunk(Delta(content=None), finish_reason="stop"),
]
wrapper = AnthropicStreamWrapper(
completion_stream=_AsyncStream(chunks), model="claude-x"
)
events = await _drain_async(wrapper)
assert _input_json_deltas(events) == ['{"city": "NY"}']
assert _text_deltas(events) == [
"Let me check.",
"The weather is nice.",
" Bye.",
]
def test_single_first_text_token_after_tool_use_preserved_sync():
"""Minimal reproduction of the issue's example: a single short text token
("Hi") resuming after a tool call. Without the fix the whole answer is
dropped because its only delta sits in the transition trigger chunk.
"""
chunks = [
_tool_chunk("call_1", "get_weather", '{"city": "NY"}'),
_make_chunk(Delta(content="Hi")),
_make_chunk(Delta(content=None), finish_reason="stop"),
]
wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x")
events = _drain_sync(wrapper)
assert _text_deltas(events) == ["Hi"]
def test_multiple_text_deltas_after_tool_use_preserved_sync():
"""Multiple-delta edge case: only the *first* text delta sits in the
transition trigger chunk; the rest stream normally. All of them — leading
one included — must reach the client in order.
"""
chunks = [
_tool_chunk("call_1", "get_weather", '{"city": "NY"}'),
_make_chunk(Delta(content="Hi")),
_make_chunk(Delta(content=", how ")),
_make_chunk(Delta(content="can I help ")),
_make_chunk(Delta(content="you?")),
_make_chunk(Delta(content=None), finish_reason="stop"),
]
wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x")
events = _drain_sync(wrapper)
assert _text_deltas(events) == ["Hi", ", how ", "can I help ", "you?"]
assert "".join(_text_deltas(events)) == "Hi, how can I help you?"
def test_empty_trigger_delta_is_not_re_emitted_sync():
"""A transition whose trigger chunk carries no content (empty text) must
NOT produce a spurious empty ``content_block_delta`` — only the synthesized
``content_block_start`` is emitted for the new block. Here a ``tool_use ->
text`` transition is triggered by an empty-content chunk; the re-emit guard
must reject it so the new text block opens without a leading empty delta.
"""
chunks = [
_tool_chunk("call_1", "get_weather", '{"city": "NY"}'),
# tool_use -> text transition triggered by an empty content chunk; the
# real text arrives in the following chunk.
_make_chunk(Delta(content="")),
_make_chunk(Delta(content="real text")),
_make_chunk(Delta(content=None), finish_reason="stop"),
]
wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x")
events = _drain_sync(wrapper)
# No empty-string text_delta should be present.
assert "" not in _text_deltas(events)
assert "".join(_text_deltas(events)) == "real text"
def test_bundled_tool_args_on_transition_still_preserved_sync():
"""Existing behavior guard: when the trigger chunk that opens a tool_use
block also carries arguments (xAI/Gemini style), the ``input_json_delta``
must still be emitted after ``content_block_start``.
"""
chunks = [
_make_chunk(Delta(content="Calling a tool.")),
_tool_chunk("call_1", "get_weather", '{"city": "NY"}'),
_make_chunk(Delta(content=None), finish_reason="tool_calls"),
]
wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x")
events = _drain_sync(wrapper)
assert _text_deltas(events) == ["Calling a tool."]
assert _input_json_deltas(events) == ['{"city": "NY"}']
@pytest.mark.parametrize(
"processed_chunk, expected",
[
# Non-empty deltas of every type must be re-emitted.
(
{
"type": "content_block_delta",
"delta": {"type": "text_delta", "text": "x"},
},
True,
),
(
{
"type": "content_block_delta",
"delta": {"type": "input_json_delta", "partial_json": "{}"},
},
True,
),
(
{
"type": "content_block_delta",
"delta": {"type": "thinking_delta", "thinking": "t"},
},
True,
),
(
{
"type": "content_block_delta",
"delta": {"type": "signature_delta", "signature": "s"},
},
True,
),
# Empty deltas must NOT be re-emitted (no spurious events).
(
{
"type": "content_block_delta",
"delta": {"type": "text_delta", "text": ""},
},
False,
),
(
{
"type": "content_block_delta",
"delta": {"type": "input_json_delta", "partial_json": ""},
},
False,
),
(
{
"type": "content_block_delta",
"delta": {"type": "thinking_delta", "thinking": ""},
},
False,
),
(
{
"type": "content_block_delta",
"delta": {"type": "signature_delta", "signature": ""},
},
False,
),
# Unknown delta type / non-content_block_delta / malformed delta.
(
{"type": "content_block_delta", "delta": {"type": "other_delta"}},
False,
),
({"type": "message_delta", "delta": {"stop_reason": "stop"}}, False),
({"type": "content_block_delta", "delta": None}, False),
],
)
def test_trigger_delta_has_content_branches(processed_chunk, expected):
"""Directly exercise the re-emit predicate across all delta types and the
empty/malformed guards, so the helper's behavior is pinned independently of
upstream chunk-translation details.
"""
assert (
AnthropicStreamWrapper._trigger_delta_has_content(processed_chunk) is expected
)

20
tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_parallel_tool_calls.py
View file

@ -278,7 +278,8 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text():
"content_block_delta", # {"city":
"content_block_delta", # "NY"}
"content_block_stop", # End of first tool_use content block
"content_block_start", # "The weather is nice today"
"content_block_start", # "The weather is nice today" text block
"content_block_delta", # "The weather is nice today." text_delta
"content_block_stop",
"content_block_start", # Start of second tool_use content block
"content_block_delta", # {"city":
@ -288,7 +289,8 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text():
"content_block_delta", # {"city":
"content_block_delta", # " CHI"}
"content_block_stop", # End of third tool_use content block
"content_block_start", # "The weather is not so nice today"
"content_block_start", # "The weather is not so nice today" text block
"content_block_delta", # "The weather is not so nice today." text_delta
"content_block_stop",
"message_delta", # Stop reason with merged usage
"message_stop", # Final message stop
@ -296,6 +298,20 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text():
assert expected_types == chunk_types
# Regression: the first (and only) text delta of each text block sits in
# the chunk that *triggered* the tool_use -> text transition. It must be
# re-emitted as a content_block_delta instead of being silently dropped.
text_deltas = [
chunk["delta"]["text"]
for chunk in chunks
if chunk.get("type") == "content_block_delta"
and chunk["delta"].get("type") == "text_delta"
]
assert text_deltas == [
"The weather is nice today.",
"The weather is not so nice today.",
]
get_weather_calls = 0
for chunk in chunks: