scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-05 08:07:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 137

fix(anthropic-adapter): correct initial content block type and preserve first delta on block transitions

Browse source 
When models return reasoning_content (e.g. GLM-5 via Vertex AI), the
Anthropic pass-through adapter now:

1. Peeks at the first streaming chunk to determine the correct initial
   content_block_start type (thinking vs text) instead of hardcoding text.

2. Detects reasoning_content (without thinking_blocks) as a thinking
   block type in _translate_streaming_openai_chunk_to_anthropic_content_block.

3. Queues the processed chunk during content block transitions so the
   first token of the new block is not silently dropped.

These fixes ensure Anthropic API consumers (e.g. Claude Code CLI) receive
a valid SSE event sequence where thinking_delta events arrive inside a
thinking content block, not a text block.

Made-with: Cursor
This commit is contained in:
Rocky Li 2026-04-06 14:19:36 +08:00
parent d251238bd7
commit 51553b4621
4 changed files with 444 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -105,13 +105,50 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
if self.sent_content_block_start is False:
self.sent_content_block_start = True
self.chunk_queue.append(
{
"type": "content_block_start",
"index": self.current_content_block_index,
"content_block": {"type": "text", "text": ""},
}
)
# Peek at the first chunk to determine the correct initial
# content block type. Models that use reasoning_content
# (e.g. GLM-5) start with a thinking block, not text.
first_chunk = None
for chunk in self.completion_stream:
if chunk == "None" or chunk is None:
continue
first_chunk = chunk
break
if first_chunk is not None:
(
block_type,
content_block_start,
) = LiteLLMAnthropicMessagesAdapter()._translate_streaming_openai_chunk_to_anthropic_content_block(
choices=first_chunk.choices
)
self.current_content_block_type = block_type
self.current_content_block_start = content_block_start
if block_type == "thinking":
initial_block: dict = {"type": "thinking", "thinking": ""}
elif block_type == "tool_use":
initial_block = dict(content_block_start)
else:
initial_block = {"type": "text", "text": ""}
self.chunk_queue.append(
{
"type": "content_block_start",
"index": self.current_content_block_index,
"content_block": initial_block,
}
)
processed_first = LiteLLMAnthropicMessagesAdapter().translate_streaming_openai_response_to_anthropic(
response=first_chunk,
current_content_block_index=self.current_content_block_index,
)
self.chunk_queue.append(processed_first)
else:
self.chunk_queue.append(
{
"type": "content_block_start",
"index": self.current_content_block_index,
"content_block": {"type": "text", "text": ""},
}
)
return self.chunk_queue.popleft()
for chunk in self.completion_stream:
@ -128,9 +165,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
)
if should_start_new_block and not self.sent_content_block_finish:
# Queue the sequence: content_block_stop -> content_block_start
# The trigger chunk itself is not emitted as a delta since the
# content_block_start already carries the relevant information.
self.chunk_queue.append(
{
"type": "content_block_stop",
@ -144,6 +178,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
"content_block": self.current_content_block_start,
}
)
self.chunk_queue.append(processed_chunk)
self.sent_content_block_finish = False
return self.chunk_queue.popleft()
@ -226,13 +261,47 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
if self.sent_content_block_start is False:
self.sent_content_block_start = True
self.chunk_queue.append(
{
"type": "content_block_start",
"index": self.current_content_block_index,
"content_block": {"type": "text", "text": ""},
}
)
first_chunk = None
async for chunk in self.completion_stream:
if chunk == "None" or chunk is None:
continue
first_chunk = chunk
break
if first_chunk is not None:
(
block_type,
content_block_start,
) = LiteLLMAnthropicMessagesAdapter()._translate_streaming_openai_chunk_to_anthropic_content_block(
choices=first_chunk.choices
)
self.current_content_block_type = block_type
self.current_content_block_start = content_block_start
if block_type == "thinking":
initial_block = {"type": "thinking", "thinking": ""}
elif block_type == "tool_use":
initial_block = dict(content_block_start)
else:
initial_block = {"type": "text", "text": ""}
self.chunk_queue.append(
{
"type": "content_block_start",
"index": self.current_content_block_index,
"content_block": initial_block,
}
)
processed_first = LiteLLMAnthropicMessagesAdapter().translate_streaming_openai_response_to_anthropic(
response=first_chunk,
current_content_block_index=self.current_content_block_index,
)
self.chunk_queue.append(processed_first)
else:
self.chunk_queue.append(
{
"type": "content_block_start",
"index": self.current_content_block_index,
"content_block": {"type": "text", "text": ""},
}
)
return self.chunk_queue.popleft()
async for chunk in self.completion_stream:
@ -304,19 +373,12 @@ 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
# The trigger chunk itself is not emitted as a delta since the
# content_block_start already carries the relevant information.
# 1. Stop current content block
self.chunk_queue.append(
{
"type": "content_block_stop",
"index": max(self.current_content_block_index - 1, 0),
}
)
# 2. Start new content block
self.chunk_queue.append(
{
"type": "content_block_start",
@ -324,11 +386,8 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
"content_block": self.current_content_block_start,
}
)
# Reset state for new block
self.chunk_queue.append(processed_chunk)
self.sent_content_block_finish = False
# Return the first queued item
return self.chunk_queue.popleft()
if (

7
litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py
View file

@ -1397,6 +1397,13 @@ class LiteLLMAnthropicMessagesAdapter:
return "thinking", ChatCompletionThinkingBlock(
type="thinking", thinking=thinking, signature=signature
)
elif isinstance(choice, StreamingChoices) and hasattr(
choice.delta, "reasoning_content"
):
if choice.delta.reasoning_content is not None:
return "thinking", ChatCompletionThinkingBlock(
type="thinking", thinking="", signature=""
)
return "text", TextBlock(type="text", text="")

313
tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_reasoning_content.py Normal file
View file

@ -0,0 +1,313 @@
"""
Tests for streaming_iterator.py fixes:
Fix 2 – Peek at first chunk to determine correct initial content_block_start type.
Models that return reasoning_content (e.g. GLM-5 via Vertex AI) start
the stream with a thinking block, not a text block.
Fix 3 – Queue the processed_chunk (first delta of a new block) when a content
block transition occurs, so the first token is not silently dropped.
"""
import os
import sys
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 (
Delta,
ModelResponseStream,
StreamingChoices,
)
# ---------------------------------------------------------------------------
# Mock streams
# ---------------------------------------------------------------------------
class MockSyncStream:
"""Synchronous mock completion stream yielding a fixed list of chunks."""
def __init__(self, chunks: list[ModelResponseStream]):
self._chunks = iter(chunks)
def __iter__(self):
return self
def __next__(self):
return next(self._chunks)
class MockAsyncStream:
"""Asynchronous mock completion stream yielding a fixed list of chunks."""
def __init__(self, chunks: list[ModelResponseStream]):
self._chunks = iter(chunks)
def __aiter__(self):
return self
async def __anext__(self):
try:
return next(self._chunks)
except StopIteration:
raise StopAsyncIteration
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_thinking_chunk(text: str) -> ModelResponseStream:
"""Create a streaming chunk with reasoning_content (no thinking_blocks)."""
return ModelResponseStream(
choices=[
StreamingChoices(
delta=Delta(
reasoning_content=text,
content="",
role="assistant",
),
index=0,
finish_reason=None,
)
],
)
def _make_text_chunk(text: str) -> ModelResponseStream:
"""Create a streaming chunk with text content."""
return ModelResponseStream(
choices=[
StreamingChoices(
delta=Delta(
content=text,
role="assistant",
),
index=0,
finish_reason=None,
)
],
)
def _make_stop_chunk() -> ModelResponseStream:
"""Create a streaming chunk signalling end of generation."""
return ModelResponseStream(
choices=[
StreamingChoices(
delta=Delta(content=""),
index=0,
finish_reason="stop",
)
],
)
def _collect_all_events(wrapper) -> list[dict]:
"""Collect all events from a sync AnthropicStreamWrapper."""
events = []
for raw in wrapper:
events.append(raw)
return events
async def _collect_all_events_async(wrapper) -> list[dict]:
"""Collect all events from an async AnthropicStreamWrapper."""
events = []
async for raw in wrapper:
events.append(raw)
return events
# ---------------------------------------------------------------------------
# Fix 2 – Initial content_block_start reflects first chunk type
# ---------------------------------------------------------------------------
class TestInitialBlockTypePeek:
"""
When the first chunk from the upstream model contains reasoning_content,
the initial content_block_start must have type "thinking", not "text".
"""
def test_sync_thinking_first_chunk(self):
chunks = [
_make_thinking_chunk("Let me think..."),
_make_thinking_chunk(" about this."),
_make_text_chunk("The answer is 42."),
_make_stop_chunk(),
]
wrapper = AnthropicStreamWrapper(
completion_stream=MockSyncStream(chunks), model="glm-5"
)
events = _collect_all_events(wrapper)
assert events[0]["type"] == "message_start"
content_block_start = events[1]
assert content_block_start["type"] == "content_block_start"
assert content_block_start["content_block"]["type"] == "thinking"
first_delta = events[2]
assert first_delta["type"] == "content_block_delta"
assert first_delta["delta"]["type"] == "thinking_delta"
def test_sync_text_first_chunk(self):
"""Text-first streams should still emit type 'text' (no regression)."""
chunks = [
_make_text_chunk("Hello"),
_make_text_chunk(" world"),
_make_stop_chunk(),
]
wrapper = AnthropicStreamWrapper(
completion_stream=MockSyncStream(chunks), model="gpt-4o"
)
events = _collect_all_events(wrapper)
content_block_start = events[1]
assert content_block_start["type"] == "content_block_start"
assert content_block_start["content_block"]["type"] == "text"
@pytest.mark.asyncio
async def test_async_thinking_first_chunk(self):
chunks = [
_make_thinking_chunk("Let me think..."),
_make_thinking_chunk(" about this."),
_make_text_chunk("The answer is 42."),
_make_stop_chunk(),
]
wrapper = AnthropicStreamWrapper(
completion_stream=MockAsyncStream(chunks), model="glm-5"
)
events = await _collect_all_events_async(wrapper)
assert events[0]["type"] == "message_start"
content_block_start = events[1]
assert content_block_start["type"] == "content_block_start"
assert content_block_start["content_block"]["type"] == "thinking"
first_delta = events[2]
assert first_delta["type"] == "content_block_delta"
assert first_delta["delta"]["type"] == "thinking_delta"
@pytest.mark.asyncio
async def test_async_text_first_chunk(self):
chunks = [
_make_text_chunk("Hello"),
_make_text_chunk(" world"),
_make_stop_chunk(),
]
wrapper = AnthropicStreamWrapper(
completion_stream=MockAsyncStream(chunks), model="gpt-4o"
)
events = await _collect_all_events_async(wrapper)
content_block_start = events[1]
assert content_block_start["type"] == "content_block_start"
assert content_block_start["content_block"]["type"] == "text"
# ---------------------------------------------------------------------------
# Fix 3 – Block transition queues the trigger chunk
# ---------------------------------------------------------------------------
class TestBlockTransitionIncludesFirstDelta:
"""
When the stream transitions from one block type to another (e.g. thinking
→ text), the processed chunk that triggered the transition must be queued
and eventually yielded. Without Fix 3 the first token of the new block
would be silently dropped.
"""
def test_sync_thinking_to_text_no_token_drop(self):
chunks = [
_make_thinking_chunk("Reasoning step."),
_make_text_chunk("Answer text."),
_make_stop_chunk(),
]
wrapper = AnthropicStreamWrapper(
completion_stream=MockSyncStream(chunks), model="glm-5"
)
events = _collect_all_events(wrapper)
text_deltas = [
e
for e in events
if e.get("type") == "content_block_delta"
and e.get("delta", {}).get("type") == "text_delta"
]
assert len(text_deltas) >= 1, (
"The first text delta after a thinking→text transition must not be "
"dropped. Got text_delta events: " + repr(text_deltas)
)
assert text_deltas[0]["delta"]["text"] == "Answer text."
@pytest.mark.asyncio
async def test_async_thinking_to_text_no_token_drop(self):
chunks = [
_make_thinking_chunk("Reasoning step."),
_make_text_chunk("Answer text."),
_make_stop_chunk(),
]
wrapper = AnthropicStreamWrapper(
completion_stream=MockAsyncStream(chunks), model="glm-5"
)
events = await _collect_all_events_async(wrapper)
text_deltas = [
e
for e in events
if e.get("type") == "content_block_delta"
and e.get("delta", {}).get("type") == "text_delta"
]
assert len(text_deltas) >= 1, (
"The first text delta after a thinking→text transition must not be "
"dropped. Got text_delta events: " + repr(text_deltas)
)
assert text_deltas[0]["delta"]["text"] == "Answer text."
def test_sync_event_sequence_is_valid(self):
"""
The full event sequence for a thinking→text stream should follow the
Anthropic SSE spec:
message_start →
content_block_start (thinking) →
content_block_delta (thinking_delta) →
content_block_stop →
content_block_start (text) →
content_block_delta (text_delta) →
content_block_stop →
message_delta →
message_stop
"""
chunks = [
_make_thinking_chunk("Think."),
_make_text_chunk("Answer."),
_make_stop_chunk(),
]
wrapper = AnthropicStreamWrapper(
completion_stream=MockSyncStream(chunks), model="glm-5"
)
events = _collect_all_events(wrapper)
types = [e["type"] for e in events]
assert types[0] == "message_start"
assert types[1] == "content_block_start"
assert events[1]["content_block"]["type"] == "thinking"
assert "content_block_delta" in types
idx_first_stop = types.index("content_block_stop")
assert idx_first_stop > 1
idx_second_start = types.index("content_block_start", idx_first_stop)
assert events[idx_second_start]["content_block"]["type"] == "text"
text_delta_idx = types.index("content_block_delta", idx_second_start)
assert events[text_delta_idx]["delta"]["type"] == "text_delta"

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

@ -131,22 +131,19 @@ def test_anthropic_stream_wrapper_single_tool_call():
chunks.append(chunk)
chunk_types.append(chunk.get("type"))
# Verify the expected sequence of chunk types
# Verify the expected sequence of chunk types.
# The initial content_block_start now peeks at the first upstream chunk
# to determine the correct block type, so we get tool_use directly
# instead of a spurious empty text block.
expected_types = [
"message_start", # Initial message start
# TODO: for future contributors: if the initial content_block_start
# respects the upstream's starting chunk, the initial empty text block
# should be removed (and this test should be updated accordingly)
# ---------------------------------------------------------------------
"content_block_start", # Initial empty text block start
"content_block_stop", # End of empty text block
# ---------------------------------------------------------------------
"content_block_start", # Start of first tool_use content block
"message_start",
"content_block_start", # tool_use (from peek)
"content_block_delta", # first tool chunk (empty args)
"content_block_delta", # {"city":
"content_block_delta", # "NY"}
"content_block_stop", # End of first tool_use content block
"message_delta", # Stop reason with merged usage
"message_stop", # Final message stop
"content_block_stop",
"message_delta",
"message_stop",
]
assert expected_types == chunk_types
@ -193,26 +190,21 @@ def test_anthropic_stream_wrapper_back_to_back_tool_calls():
chunks.append(chunk)
chunk_types.append(chunk.get("type"))
# Verify the expected sequence of chunk types
# Verify the expected sequence of chunk types.
expected_types = [
"message_start", # Initial message start
# TODO: for future contributors: if the initial content_block_start
# respects the upstream's starting chunk, the initial empty text block
# should be removed (and this test should be updated accordingly)
# ---------------------------------------------------------------------
"content_block_start", # Initial empty text block start
"content_block_stop", # End of empty text block
# ---------------------------------------------------------------------
"content_block_start", # Start of first tool_use content block
"message_start",
"content_block_start", # tool_use (from peek)
"content_block_delta", # first tool chunk (empty args)
"content_block_delta", # {"city":
"content_block_delta", # "NY"}
"content_block_stop", # End of first tool_use content block
"content_block_start", # Start of second tool_use content block
"content_block_stop",
"content_block_start", # second tool_use
"content_block_delta", # first chunk of second tool
"content_block_delta", # {"city":
"content_block_delta", # " SF"}
"content_block_stop", # End of second tool_use content block
"message_delta", # Stop reason with merged usage
"message_stop", # Final message stop
"content_block_stop",
"message_delta",
"message_stop",
]
assert expected_types == chunk_types
@ -264,34 +256,32 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text():
chunks.append(chunk)
chunk_types.append(chunk.get("type"))
# Verify the expected sequence of chunk types
# Verify the expected sequence of chunk types.
expected_types = [
"message_start", # Initial message start
# TODO: for future contributors: if the initial content_block_start
# respects the upstream's starting chunk, the initial empty text block
# should be removed (and this test should be updated accordingly)
# ---------------------------------------------------------------------
"content_block_start", # Initial empty text block start
"content_block_stop", # End of empty text block
# ---------------------------------------------------------------------
"content_block_start", # Start of first tool_use content block
"message_start",
"content_block_start", # tool_use (from peek)
"content_block_delta", # first tool chunk (empty args)
"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_stop",
"content_block_start", # Start of second tool_use content block
"content_block_start", # text
"content_block_delta", # "The weather is nice today."
"content_block_stop",
"content_block_start", # second tool_use
"content_block_delta", # first chunk of second tool
"content_block_delta", # {"city":
"content_block_delta", # " SF"}
"content_block_stop", # End of second tool_use content block
"content_block_start", # Start of third tool_use content block
"content_block_stop",
"content_block_start", # third tool_use
"content_block_delta", # first chunk of third tool
"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_stop",
"message_delta", # Stop reason with merged usage
"message_stop", # Final message stop
"content_block_start", # text
"content_block_delta", # "The weather is not so nice today."
"content_block_stop",
"message_delta",
"message_stop",
]
assert expected_types == chunk_types