mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-20 05:04:24 +00:00
fix(anthropic-adapter): valid Anthropic SSE for reasoning_content and tool-call streams
Peek the first upstream chunk before content_block_start so thinking vs text blocks match reasoning_content models, skip empty reasoning_content, and only re-queue non-empty trigger deltas on block transitions. Adds regression tests for Claude Code SSE failures that caused duplicate non-streaming retries. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
parent
668df9494a
commit
6ed4eeafb3
6 changed files with 780 additions and 116 deletions
|
|
@ -374,14 +374,65 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
|
||||
if self.sent_content_block_start is False:
|
||||
self.sent_content_block_start = True
|
||||
self.sent_content_block_finish = False
|
||||
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,
|
||||
)
|
||||
# Empty / stop-only first chunk: close the block before the
|
||||
# terminal message_delta so the sequence stays spec-compliant.
|
||||
if (
|
||||
isinstance(processed_first, dict)
|
||||
and processed_first.get("type") == "message_delta"
|
||||
):
|
||||
self.chunk_queue.append(
|
||||
{
|
||||
"type": "content_block_stop",
|
||||
"index": self.current_content_block_index,
|
||||
}
|
||||
)
|
||||
self.sent_content_block_finish = True
|
||||
self.chunk_queue.append(processed_first)
|
||||
elif self._trigger_delta_has_content(processed_first):
|
||||
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:
|
||||
|
|
@ -427,19 +478,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
continue
|
||||
|
||||
if should_start_new_block and not self.sent_content_block_finish:
|
||||
# Queue the sequence: content_block_stop -> content_block_start
|
||||
# -> (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(
|
||||
{
|
||||
"type": "content_block_stop",
|
||||
|
|
@ -447,7 +485,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
}
|
||||
)
|
||||
|
||||
# 2. Start new content block
|
||||
self.chunk_queue.append(
|
||||
{
|
||||
"type": "content_block_start",
|
||||
|
|
@ -593,14 +630,60 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
|
||||
if self.sent_content_block_start is False:
|
||||
self.sent_content_block_start = True
|
||||
self.sent_content_block_finish = False
|
||||
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,
|
||||
)
|
||||
if (
|
||||
isinstance(processed_first, dict)
|
||||
and processed_first.get("type") == "message_delta"
|
||||
):
|
||||
self.chunk_queue.append(
|
||||
{
|
||||
"type": "content_block_stop",
|
||||
"index": self.current_content_block_index,
|
||||
}
|
||||
)
|
||||
self.sent_content_block_finish = True
|
||||
self.chunk_queue.append(processed_first)
|
||||
elif self._trigger_delta_has_content(processed_first):
|
||||
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:
|
||||
|
|
@ -641,19 +724,6 @@ 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
|
||||
# -> (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(
|
||||
{
|
||||
"type": "content_block_stop",
|
||||
|
|
@ -667,13 +737,8 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
"content_block": self.current_content_block_start,
|
||||
}
|
||||
)
|
||||
|
||||
# 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
|
||||
self.sent_content_block_finish = False
|
||||
return self.chunk_queue.popleft()
|
||||
|
||||
|
|
|
|||
|
|
@ -1410,13 +1410,13 @@ class LiteLLMAnthropicMessagesAdapter:
|
|||
return "thinking", ChatCompletionThinkingBlock(
|
||||
type="thinking", thinking=thinking, signature=signature
|
||||
)
|
||||
# OpenAI-compatible reasoning backends (e.g. vLLM/SGLang reasoning
|
||||
# parsers) populate ``reasoning_content`` without ``thinking_blocks``.
|
||||
# ``Delta`` deletes the ``thinking_blocks`` attribute when unset, so the
|
||||
# branch above is skipped entirely; open a ``thinking`` block here so the
|
||||
# matching ``thinking_delta`` stream is not emitted into a text block.
|
||||
elif isinstance(choice, StreamingChoices) and getattr(choice.delta, "reasoning_content", None):
|
||||
return "thinking", ChatCompletionThinkingBlock(type="thinking", thinking="", signature="")
|
||||
elif isinstance(choice, StreamingChoices) and hasattr(
|
||||
choice.delta, "reasoning_content"
|
||||
):
|
||||
if choice.delta.reasoning_content:
|
||||
return "thinking", ChatCompletionThinkingBlock(
|
||||
type="thinking", thinking="", signature=""
|
||||
)
|
||||
|
||||
return "text", TextBlock(type="text", text="")
|
||||
|
||||
|
|
@ -1438,8 +1438,7 @@ class LiteLLMAnthropicMessagesAdapter:
|
|||
for choice in choices:
|
||||
if choice.delta.content is not None and len(choice.delta.content) > 0:
|
||||
text += choice.delta.content
|
||||
if choice.delta.tool_calls:
|
||||
partial_json = ""
|
||||
if choice.delta.tool_calls is not None and len(choice.delta.tool_calls) > 0:
|
||||
for tool in choice.delta.tool_calls:
|
||||
if tool.function is not None and tool.function.arguments is not None:
|
||||
partial_json = (partial_json or "") + tool.function.arguments
|
||||
|
|
@ -1459,7 +1458,7 @@ class LiteLLMAnthropicMessagesAdapter:
|
|||
# Handle reasoning_content when thinking_blocks is not present
|
||||
# This handles providers like OpenRouter that return reasoning_content
|
||||
elif isinstance(choice, StreamingChoices) and hasattr(choice.delta, "reasoning_content"):
|
||||
if choice.delta.reasoning_content is not None:
|
||||
if choice.delta.reasoning_content:
|
||||
reasoning_content += choice.delta.reasoning_content
|
||||
|
||||
if reasoning_content and reasoning_signature:
|
||||
|
|
|
|||
|
|
@ -402,6 +402,82 @@ def test_translate_anthropic_messages_to_openai_tool_message_placement():
|
|||
), "Tool message should be placed before user message"
|
||||
|
||||
|
||||
def test_translate_streaming_openai_text_delta_no_tool_calls():
|
||||
"""When tool_calls is None, a text delta with content should be text_delta."""
|
||||
choices = [
|
||||
StreamingChoices(
|
||||
finish_reason=None,
|
||||
index=0,
|
||||
delta=Delta(
|
||||
provider_specific_fields=None,
|
||||
content="Hello",
|
||||
role="assistant",
|
||||
function_call=None,
|
||||
tool_calls=None,
|
||||
audio=None,
|
||||
),
|
||||
logprobs=None,
|
||||
)
|
||||
]
|
||||
|
||||
(
|
||||
type_of_content,
|
||||
content_block_delta,
|
||||
) = LiteLLMAnthropicMessagesAdapter()._translate_streaming_openai_chunk_to_anthropic(
|
||||
choices=choices # type: ignore[arg-type]
|
||||
)
|
||||
|
||||
assert type_of_content == "text_delta"
|
||||
assert content_block_delta["type"] == "text_delta"
|
||||
assert content_block_delta["text"] == "Hello" # type: ignore[typeddict-unknown-key]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"content",
|
||||
[
|
||||
"Hello from DeepSeek",
|
||||
"Let me think about that.",
|
||||
],
|
||||
)
|
||||
def test_translate_streaming_openai_text_delta_empty_tool_calls(content: str):
|
||||
"""When tool_calls is an empty list (len==0), content should still be text_delta.
|
||||
|
||||
This reproduces the bug where DeepSeek/vLLM sends tool_calls: []
|
||||
instead of tool_calls: None. Before the fix, the `is not None` guard
|
||||
let this into the tool_calls branch, and a stray partial_json="" caused
|
||||
an empty input_json_delta to be emitted instead of the text content.
|
||||
"""
|
||||
choices = [
|
||||
StreamingChoices(
|
||||
finish_reason=None,
|
||||
index=0,
|
||||
delta=Delta(
|
||||
provider_specific_fields=None,
|
||||
content=content,
|
||||
role="assistant",
|
||||
function_call=None,
|
||||
tool_calls=[], # empty list — the exact pattern from vLLM/DeepSeek
|
||||
audio=None,
|
||||
),
|
||||
logprobs=None,
|
||||
)
|
||||
]
|
||||
|
||||
(
|
||||
type_of_content,
|
||||
content_block_delta,
|
||||
) = LiteLLMAnthropicMessagesAdapter()._translate_streaming_openai_chunk_to_anthropic(
|
||||
choices=choices # type: ignore[arg-type]
|
||||
)
|
||||
|
||||
assert type_of_content == "text_delta", (
|
||||
f"Expected text_delta but got {type_of_content} — "
|
||||
"empty tool_calls list should not trigger input_json_delta"
|
||||
)
|
||||
assert content_block_delta["type"] == "text_delta"
|
||||
assert content_block_delta["text"] == content # type: ignore[typeddict-unknown-key]
|
||||
|
||||
|
||||
def test_translate_openai_content_to_anthropic_empty_function_arguments():
|
||||
"""Test that empty function arguments are handled safely and don't cause JSON parsing errors."""
|
||||
|
||||
|
|
@ -1839,6 +1915,65 @@ def test_translate_streaming_openai_chunk_to_anthropic_reasoning_content_without
|
|||
assert content_block_delta["thinking"] == "I need to analyze this carefully..."
|
||||
|
||||
|
||||
def test_translate_streaming_openai_chunk_to_anthropic_content_block_empty_reasoning_content():
|
||||
"""Empty reasoning_content must not open a thinking block without thinking_delta."""
|
||||
choices = [
|
||||
StreamingChoices(
|
||||
finish_reason=None,
|
||||
index=0,
|
||||
delta=Delta(
|
||||
reasoning_content="",
|
||||
content="",
|
||||
role="assistant",
|
||||
function_call=None,
|
||||
tool_calls=None,
|
||||
audio=None,
|
||||
),
|
||||
logprobs=None,
|
||||
)
|
||||
]
|
||||
|
||||
(
|
||||
block_type,
|
||||
content_block_start,
|
||||
) = LiteLLMAnthropicMessagesAdapter()._translate_streaming_openai_chunk_to_anthropic_content_block(
|
||||
choices=choices
|
||||
)
|
||||
|
||||
assert block_type == "text"
|
||||
assert content_block_start == {"type": "text", "text": ""}
|
||||
|
||||
|
||||
def test_translate_streaming_openai_chunk_to_anthropic_empty_reasoning_content_no_thinking_delta():
|
||||
"""Empty reasoning_content must not emit thinking_delta on a text block."""
|
||||
choices = [
|
||||
StreamingChoices(
|
||||
finish_reason=None,
|
||||
index=0,
|
||||
delta=Delta(
|
||||
reasoning_content="",
|
||||
content="",
|
||||
role="assistant",
|
||||
function_call=None,
|
||||
tool_calls=None,
|
||||
audio=None,
|
||||
),
|
||||
logprobs=None,
|
||||
)
|
||||
]
|
||||
|
||||
(
|
||||
type_of_content,
|
||||
content_block_delta,
|
||||
) = LiteLLMAnthropicMessagesAdapter()._translate_streaming_openai_chunk_to_anthropic(
|
||||
choices=choices
|
||||
)
|
||||
|
||||
assert type_of_content == "text_delta"
|
||||
assert content_block_delta["type"] == "text_delta"
|
||||
assert content_block_delta["text"] == ""
|
||||
|
||||
|
||||
def test_translate_openai_response_to_anthropic_with_reasoning_content_only():
|
||||
"""
|
||||
Test the full response translation when only reasoning_content is present
|
||||
|
|
|
|||
|
|
@ -0,0 +1,483 @@
|
|||
"""
|
||||
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
|
||||
|
||||
|
||||
def _assert_monotonic_indices(events: list[dict]) -> None:
|
||||
"""Indices on content_block events must be non-decreasing with matched pairs."""
|
||||
open_indices: list[int] = []
|
||||
for event in events:
|
||||
event_type = event.get("type")
|
||||
if event_type not in (
|
||||
"content_block_start",
|
||||
"content_block_delta",
|
||||
"content_block_stop",
|
||||
):
|
||||
continue
|
||||
idx = event.get("index")
|
||||
assert isinstance(idx, int)
|
||||
if event_type == "content_block_start":
|
||||
if open_indices:
|
||||
assert idx >= open_indices[-1]
|
||||
open_indices.append(idx)
|
||||
elif event_type == "content_block_stop":
|
||||
assert open_indices, f"unexpected content_block_stop at index {idx}"
|
||||
assert open_indices.pop() == idx
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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"
|
||||
|
||||
def test_sync_empty_reasoning_content_first_chunk_opens_text_block(self):
|
||||
"""Empty reasoning_content placeholder must not open a thinking block."""
|
||||
chunks = [
|
||||
_make_thinking_chunk(""),
|
||||
_make_text_chunk("Hello"),
|
||||
_make_stop_chunk(),
|
||||
]
|
||||
wrapper = AnthropicStreamWrapper(
|
||||
completion_stream=MockSyncStream(chunks), model="glm-5"
|
||||
)
|
||||
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"
|
||||
|
||||
first_delta = events[2]
|
||||
assert first_delta["type"] == "content_block_delta"
|
||||
assert first_delta["delta"]["type"] == "text_delta"
|
||||
|
||||
@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"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Extra coverage – text-only full sequence, no duplicates, stop-only guard
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestTextOnlyFullSequence:
|
||||
def test_sync_text_only_sequence(self):
|
||||
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)
|
||||
types = [e["type"] for e in events]
|
||||
|
||||
assert types == [
|
||||
"message_start",
|
||||
"content_block_start",
|
||||
"content_block_delta",
|
||||
"content_block_delta",
|
||||
"content_block_stop",
|
||||
"message_delta",
|
||||
"message_stop",
|
||||
]
|
||||
assert events[1]["content_block"]["type"] == "text"
|
||||
assert events[2]["delta"]["text"] == "Hello"
|
||||
assert events[3]["delta"]["text"] == " world"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_text_only_sequence(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)
|
||||
types = [e["type"] for e in events]
|
||||
|
||||
assert types == [
|
||||
"message_start",
|
||||
"content_block_start",
|
||||
"content_block_delta",
|
||||
"content_block_delta",
|
||||
"content_block_stop",
|
||||
"message_delta",
|
||||
"message_stop",
|
||||
]
|
||||
|
||||
|
||||
class TestNoDuplicateFirstDelta:
|
||||
def test_sync_peeked_first_delta_not_duplicated(self):
|
||||
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)
|
||||
|
||||
text_deltas = [
|
||||
e["delta"]["text"]
|
||||
for e in events
|
||||
if e.get("type") == "content_block_delta"
|
||||
and e.get("delta", {}).get("type") == "text_delta"
|
||||
]
|
||||
assert text_deltas == ["Hello", " world"]
|
||||
_assert_monotonic_indices(events)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_peeked_first_delta_not_duplicated(self):
|
||||
chunks = [
|
||||
_make_thinking_chunk("Think."),
|
||||
_make_text_chunk("Answer."),
|
||||
_make_stop_chunk(),
|
||||
]
|
||||
wrapper = AnthropicStreamWrapper(
|
||||
completion_stream=MockAsyncStream(chunks), model="glm-5"
|
||||
)
|
||||
events = await _collect_all_events_async(wrapper)
|
||||
_assert_monotonic_indices(events)
|
||||
|
||||
|
||||
class TestStopOnlyFirstChunk:
|
||||
"""Empty/stop-only first chunk must still emit content_block_stop."""
|
||||
|
||||
def test_sync_stop_only_first_chunk(self):
|
||||
chunks = [_make_stop_chunk()]
|
||||
wrapper = AnthropicStreamWrapper(
|
||||
completion_stream=MockSyncStream(chunks), model="gpt-4o"
|
||||
)
|
||||
events = _collect_all_events(wrapper)
|
||||
types = [e["type"] for e in events]
|
||||
|
||||
assert types == [
|
||||
"message_start",
|
||||
"content_block_start",
|
||||
"content_block_stop",
|
||||
"message_delta",
|
||||
"message_stop",
|
||||
]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_stop_only_first_chunk(self):
|
||||
chunks = [_make_stop_chunk()]
|
||||
wrapper = AnthropicStreamWrapper(
|
||||
completion_stream=MockAsyncStream(chunks), model="gpt-4o"
|
||||
)
|
||||
events = await _collect_all_events_async(wrapper)
|
||||
types = [e["type"] for e in events]
|
||||
|
||||
assert types == [
|
||||
"message_start",
|
||||
"content_block_start",
|
||||
"content_block_stop",
|
||||
"message_delta",
|
||||
"message_stop",
|
||||
]
|
||||
|
|
@ -152,7 +152,7 @@ async def test_async_stream_no_extra_delta_when_tool_args_empty():
|
|||
"""
|
||||
When a provider sends tool name/id WITHOUT arguments in the first chunk
|
||||
(OpenAI-style), the wrapper should NOT emit an extra input_json_delta
|
||||
after content_block_start. This verifies backward compatibility.
|
||||
from the trigger chunk — only the follow-up chunk with real args.
|
||||
"""
|
||||
# Chunk 1: text
|
||||
text_chunk = _make_chunk(Delta(content="Hi", role="assistant", tool_calls=None))
|
||||
|
|
@ -221,9 +221,8 @@ async def test_async_stream_no_extra_delta_when_tool_args_empty():
|
|||
|
||||
assert tool_start_idx is not None
|
||||
|
||||
# Count how many input_json_delta events appear after the tool_use block start.
|
||||
# With empty args in the trigger chunk, only the subsequent tool_args_chunk
|
||||
# should produce one — not the trigger chunk itself.
|
||||
# Count input_json_delta events after the tool_use block start. Only the
|
||||
# follow-up chunk with real arguments should produce a delta.
|
||||
input_json_deltas = [
|
||||
e
|
||||
for e in events[tool_start_idx + 1 :]
|
||||
|
|
|
|||
|
|
@ -131,22 +131,17 @@ 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.
|
||||
# Peek opens tool_use directly (no empty text block). Empty args on the
|
||||
# first tool-name chunk are suppressed by _trigger_delta_has_content.
|
||||
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", # {"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 +188,20 @@ 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.
|
||||
# Empty tool-name trigger deltas are suppressed; only real args stream.
|
||||
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", # {"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", # {"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_delta", # "SF"}
|
||||
"content_block_stop",
|
||||
"message_delta",
|
||||
"message_stop",
|
||||
]
|
||||
|
||||
assert expected_types == chunk_types
|
||||
|
|
@ -264,36 +253,30 @@ 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.
|
||||
# Empty tool-name trigger deltas are suppressed; bundled text deltas are kept.
|
||||
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", # {"city":
|
||||
"content_block_delta", # "NY"}
|
||||
"content_block_stop", # End of first tool_use content block
|
||||
"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":
|
||||
"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_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" text block
|
||||
"content_block_delta", # "The weather is not so nice today." text_delta
|
||||
"content_block_start", # text
|
||||
"content_block_delta", # "The weather is nice today."
|
||||
"content_block_stop",
|
||||
"message_delta", # Stop reason with merged usage
|
||||
"message_stop", # Final message stop
|
||||
"content_block_start", # second tool_use
|
||||
"content_block_delta", # {"city":
|
||||
"content_block_delta", # "SF"}
|
||||
"content_block_stop",
|
||||
"content_block_start", # third tool_use
|
||||
"content_block_delta", # {"city":
|
||||
"content_block_delta", # "CHI"}
|
||||
"content_block_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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue