mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-14 23:21:35 +00:00
fix(anthropic): correct streaming tool_use framing for /v1/messages
When a provider such as ollama_chat streams a tool call and then a plain stop finish_reason through the Anthropic /v1/messages adapter, the SSE framing was wrong in two ways: the turn was prefixed with a spurious empty text content block, and the final message_delta carried stop_reason end_turn instead of tool_use even though a tool_use block was streamed. Anthropic tool-runners key off stop_reason tool_use to decide whether to execute the tool, so the call was silently dropped. Open the first content block lazily from the first chunk that carries content, and override a translated end_turn to tool_use when a tool_use block was emitted.
This commit is contained in:
Devin AI
parent
24123269cc
commit
fbf9da9af0
3 changed files with 238 additions and 56 deletions
|
|
@ -161,6 +161,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
sent_content_block_start: bool = False
|
||||
sent_content_block_finish: bool = False
|
||||
current_content_block_type: Literal["text", "tool_use", "thinking"] = "text"
|
||||
emitted_tool_use_block: bool = False
|
||||
sent_last_message: bool = False
|
||||
holding_chunk: Optional[Any] = None
|
||||
holding_stop_reason_chunk: Optional[Any] = None
|
||||
|
|
@ -258,6 +259,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
held-chunk flush path stays in sync with the merge path's guarantee
|
||||
when ``self.applied_edits`` is non-empty.
|
||||
"""
|
||||
message_delta_chunk = self._override_stop_reason_for_tool_use(message_delta_chunk)
|
||||
message_delta_chunk = self._ensure_context_management_attached(message_delta_chunk)
|
||||
if self.iterations_usage is None:
|
||||
return message_delta_chunk
|
||||
|
|
@ -393,22 +395,17 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
if compaction_event is not None:
|
||||
return compaction_event
|
||||
|
||||
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": ""},
|
||||
}
|
||||
)
|
||||
return self.chunk_queue.popleft()
|
||||
|
||||
for chunk in self.completion_stream:
|
||||
if chunk == "None" or chunk is None:
|
||||
raise Exception
|
||||
|
||||
if self.sent_content_block_start is False:
|
||||
first_block_action = self._open_first_content_block(chunk)
|
||||
if first_block_action == "skip":
|
||||
continue
|
||||
if first_block_action == "opened":
|
||||
return self.chunk_queue.popleft()
|
||||
|
||||
should_start_new_block = self._should_start_new_content_block(chunk)
|
||||
if should_start_new_block:
|
||||
self._increment_content_block_index()
|
||||
|
|
@ -615,22 +612,17 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
if compaction_event is not None:
|
||||
return compaction_event
|
||||
|
||||
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": ""},
|
||||
}
|
||||
)
|
||||
return self.chunk_queue.popleft()
|
||||
|
||||
async for chunk in self.completion_stream:
|
||||
if chunk == "None" or chunk is None:
|
||||
raise Exception
|
||||
|
||||
if self.sent_content_block_start is False:
|
||||
first_block_action = self._open_first_content_block(chunk)
|
||||
if first_block_action == "skip":
|
||||
continue
|
||||
if first_block_action == "opened":
|
||||
return self.chunk_queue.popleft()
|
||||
|
||||
# Check if we need to start a new content block
|
||||
should_start_new_block = self._should_start_new_content_block(chunk)
|
||||
if should_start_new_block:
|
||||
|
|
@ -900,18 +892,9 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
)
|
||||
|
||||
# Restore original tool name if it was truncated for OpenAI's 64-char limit
|
||||
self._restore_tool_name(block_type, content_block_start)
|
||||
if block_type == "tool_use":
|
||||
# Type narrowing: content_block_start is ToolUseBlock when block_type is "tool_use"
|
||||
from typing import cast
|
||||
|
||||
from litellm.types.llms.anthropic import ToolUseBlock
|
||||
|
||||
tool_block = cast(ToolUseBlock, content_block_start)
|
||||
|
||||
if tool_block.get("name"):
|
||||
truncated_name = tool_block["name"]
|
||||
original_name = self.tool_name_mapping.get(truncated_name, truncated_name)
|
||||
tool_block["name"] = original_name
|
||||
self.emitted_tool_use_block = True
|
||||
|
||||
if block_type != self.current_content_block_type:
|
||||
self.current_content_block_type = block_type
|
||||
|
|
@ -932,3 +915,105 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
return True
|
||||
|
||||
return False
|
||||
|
||||
def _restore_tool_name(
|
||||
self,
|
||||
block_type: Literal["text", "tool_use", "thinking"],
|
||||
content_block_start: "AnthropicStreamWrapper.ContentBlockContentBlockDict",
|
||||
) -> None:
|
||||
"""Restore a tool_use block's original name when it was truncated to fit
|
||||
OpenAI's 64-char tool-name limit. No-op for non-tool blocks.
|
||||
"""
|
||||
if block_type != "tool_use":
|
||||
return
|
||||
|
||||
from typing import cast
|
||||
|
||||
from litellm.types.llms.anthropic import ToolUseBlock
|
||||
|
||||
tool_block = cast(ToolUseBlock, content_block_start)
|
||||
truncated_name = tool_block.get("name")
|
||||
if truncated_name:
|
||||
tool_block["name"] = self.tool_name_mapping.get(truncated_name, truncated_name)
|
||||
|
||||
def _open_first_content_block(self, chunk: "ModelResponseStream") -> Literal["skip", "opened", "fell_through"]:
|
||||
"""Open the very first Anthropic content block lazily, from the first
|
||||
chunk that actually carries content, so a turn that starts with a
|
||||
tool_use (or thinking) block is not prefixed with a spurious empty
|
||||
``{"type": "text", "text": ""}`` block.
|
||||
|
||||
Returns one of:
|
||||
- ``"skip"``: the chunk carries no content yet (e.g. a role-only
|
||||
delta); no block was opened, the caller should keep iterating.
|
||||
- ``"opened"``: a ``content_block_start`` (and, when the trigger chunk
|
||||
bundles the first delta, that delta) was queued; the caller should
|
||||
return the queued head.
|
||||
- ``"fell_through"``: the stream ended before any content arrived, so
|
||||
an empty text block was opened to keep the message well-formed; the
|
||||
caller should continue into the normal ``message_delta`` handling.
|
||||
"""
|
||||
from .transformation import LiteLLMAnthropicMessagesAdapter
|
||||
|
||||
if chunk.choices[0].finish_reason is not None:
|
||||
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": ""},
|
||||
}
|
||||
)
|
||||
return "fell_through"
|
||||
|
||||
adapter = LiteLLMAnthropicMessagesAdapter()
|
||||
(
|
||||
block_type,
|
||||
content_block_start,
|
||||
) = adapter._translate_streaming_openai_chunk_to_anthropic_content_block(choices=chunk.choices)
|
||||
processed_chunk = adapter.translate_streaming_openai_response_to_anthropic(
|
||||
response=chunk,
|
||||
current_content_block_index=self.current_content_block_index,
|
||||
)
|
||||
if block_type == "text" and not self._delta_has_content(processed_chunk):
|
||||
return "skip"
|
||||
|
||||
self._restore_tool_name(block_type, content_block_start)
|
||||
if block_type == "tool_use":
|
||||
self.emitted_tool_use_block = True
|
||||
|
||||
self.current_content_block_type = block_type
|
||||
self.current_content_block_start = content_block_start
|
||||
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": content_block_start,
|
||||
}
|
||||
)
|
||||
if self._delta_has_content(processed_chunk):
|
||||
self.chunk_queue.append(processed_chunk)
|
||||
return "opened"
|
||||
|
||||
def _override_stop_reason_for_tool_use(self, message_delta_chunk: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Force ``stop_reason: "tool_use"`` on the final ``message_delta`` when
|
||||
the streamed message contains a tool_use block but the upstream
|
||||
finish_reason mapped to ``end_turn``.
|
||||
|
||||
Some providers (e.g. ``ollama_chat``) stream the tool call in an earlier
|
||||
chunk and then send a plain ``stop`` finish_reason in the terminating
|
||||
chunk, which maps to ``end_turn``. Anthropic-Messages tool-runners key
|
||||
off ``stop_reason == "tool_use"`` to decide whether to execute the
|
||||
emitted ``tool_use`` block, so this keeps the streaming framing in sync
|
||||
with the non-streaming response and the Anthropic spec.
|
||||
"""
|
||||
if not self.emitted_tool_use_block:
|
||||
return message_delta_chunk
|
||||
delta = message_delta_chunk.get("delta")
|
||||
if not isinstance(delta, dict) or delta.get("stop_reason") != "end_turn":
|
||||
return message_delta_chunk
|
||||
augmented = message_delta_chunk.copy()
|
||||
augmented["delta"] = {**delta, "stop_reason": "tool_use"}
|
||||
return augmented
|
||||
|
|
|
|||
|
|
@ -506,3 +506,118 @@ def test_empty_content_chunk_mid_text_block_is_suppressed_sync():
|
|||
|
||||
assert _text_deltas(events) == ["Hi", " there"]
|
||||
_assert_deltas_match_their_block_type(events)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Regression tests for issue #34692.
|
||||
#
|
||||
# When a provider (e.g. ``ollama_chat``) streams a tool call and then a plain
|
||||
# ``stop`` finish_reason, the Anthropic ``/v1/messages`` framing was wrong in
|
||||
# two ways: the turn was prefixed with a spurious empty ``{"type": "text",
|
||||
# "text": ""}`` content block, and the final ``message_delta`` carried
|
||||
# ``stop_reason: "end_turn"`` instead of ``"tool_use"`` even though a tool_use
|
||||
# block was streamed. Anthropic tool-runners key off ``stop_reason ==
|
||||
# "tool_use"`` to decide whether to execute the tool, so the tool call was
|
||||
# silently dropped downstream.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _content_block_starts(events: List[dict]) -> List[dict]:
|
||||
return [e for e in events if e.get("type") == "content_block_start"]
|
||||
|
||||
|
||||
def _final_stop_reason(events: List[dict]) -> Optional[str]:
|
||||
for event in events:
|
||||
if event.get("type") == "message_delta":
|
||||
return event["delta"].get("stop_reason")
|
||||
return None
|
||||
|
||||
|
||||
def _ollama_style_tool_then_stop_chunks() -> List[MagicMock]:
|
||||
"""``ollama_chat`` streams the whole tool call in a chunk with no
|
||||
finish_reason, then a terminating chunk whose finish_reason is a plain
|
||||
``stop`` (the tool_calls are not repeated on the terminating chunk).
|
||||
"""
|
||||
return [
|
||||
_tool_chunk("call_1", "get_weather", '{"city": "SF"}'),
|
||||
_make_chunk(Delta(content=None), finish_reason="stop"),
|
||||
]
|
||||
|
||||
|
||||
def test_tool_first_turn_has_no_leading_empty_text_block_sync():
|
||||
wrapper = AnthropicStreamWrapper(
|
||||
completion_stream=iter(_ollama_style_tool_then_stop_chunks()), model="qwen3"
|
||||
)
|
||||
events = _drain_sync(wrapper)
|
||||
|
||||
starts = _content_block_starts(events)
|
||||
assert [s["content_block"]["type"] for s in starts] == ["tool_use"]
|
||||
assert starts[0]["index"] == 0
|
||||
assert _input_json_deltas(events) == ['{"city": "SF"}']
|
||||
|
||||
|
||||
def test_stop_reason_is_tool_use_when_tool_streamed_before_plain_stop_sync():
|
||||
wrapper = AnthropicStreamWrapper(
|
||||
completion_stream=iter(_ollama_style_tool_then_stop_chunks()), model="qwen3"
|
||||
)
|
||||
assert _final_stop_reason(_drain_sync(wrapper)) == "tool_use"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_first_turn_has_no_leading_empty_text_block_async():
|
||||
wrapper = AnthropicStreamWrapper(
|
||||
completion_stream=_AsyncStream(_ollama_style_tool_then_stop_chunks()),
|
||||
model="qwen3",
|
||||
)
|
||||
events = await _drain_async(wrapper)
|
||||
|
||||
starts = _content_block_starts(events)
|
||||
assert [s["content_block"]["type"] for s in starts] == ["tool_use"]
|
||||
assert _final_stop_reason(events) == "tool_use"
|
||||
|
||||
|
||||
def test_max_tokens_stop_reason_is_preserved_even_with_tool_use_sync():
|
||||
"""The override only rewrites ``end_turn`` -> ``tool_use``; a genuine
|
||||
``max_tokens`` truncation mid-tool-call must survive so clients can tell
|
||||
the turn was cut short.
|
||||
"""
|
||||
chunks = [
|
||||
_tool_chunk("call_1", "get_weather", '{"city": "S'),
|
||||
_make_chunk(Delta(content=None), finish_reason="length"),
|
||||
]
|
||||
wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="qwen3")
|
||||
assert _final_stop_reason(_drain_sync(wrapper)) == "max_tokens"
|
||||
|
||||
|
||||
def test_plain_text_turn_keeps_end_turn_and_leading_text_block_sync():
|
||||
"""A text-only turn must be unaffected: it opens a single text block and
|
||||
ends with ``end_turn`` (no tool_use override, no dropped block).
|
||||
"""
|
||||
chunks = [
|
||||
_make_chunk(Delta(content="Hello")),
|
||||
_make_chunk(Delta(content=" there")),
|
||||
_make_chunk(Delta(content=None), finish_reason="stop"),
|
||||
]
|
||||
wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="qwen3")
|
||||
events = _drain_sync(wrapper)
|
||||
|
||||
starts = _content_block_starts(events)
|
||||
assert [s["content_block"]["type"] for s in starts] == ["text"]
|
||||
assert _text_deltas(events) == ["Hello", " there"]
|
||||
assert _final_stop_reason(events) == "end_turn"
|
||||
|
||||
|
||||
def test_thinking_first_turn_has_no_leading_empty_text_block_sync():
|
||||
"""A reasoning-first turn must open a ``thinking`` block directly rather
|
||||
than a spurious empty text block ahead of it.
|
||||
"""
|
||||
chunks = [
|
||||
_thinking_chunk("Let me think"),
|
||||
_make_chunk(Delta(content="Answer")),
|
||||
_make_chunk(Delta(content=None), finish_reason="stop"),
|
||||
]
|
||||
wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="qwen3")
|
||||
events = _drain_sync(wrapper)
|
||||
|
||||
starts = _content_block_starts(events)
|
||||
assert [s["content_block"]["type"] for s in starts] == ["thinking", "text"]
|
||||
|
|
|
|||
|
|
@ -134,13 +134,6 @@ def test_anthropic_stream_wrapper_single_tool_call():
|
|||
# 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
|
||||
"content_block_delta", # {"city":
|
||||
"content_block_delta", # "NY"}
|
||||
|
|
@ -151,6 +144,9 @@ def test_anthropic_stream_wrapper_single_tool_call():
|
|||
|
||||
assert expected_types == chunk_types
|
||||
|
||||
message_delta = next(c for c in chunks if c.get("type") == "message_delta")
|
||||
assert message_delta["delta"]["stop_reason"] == "tool_use"
|
||||
|
||||
get_weather_calls = 0
|
||||
|
||||
for chunk in chunks:
|
||||
|
|
@ -196,13 +192,6 @@ def test_anthropic_stream_wrapper_back_to_back_tool_calls():
|
|||
# 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
|
||||
"content_block_delta", # {"city":
|
||||
"content_block_delta", # "NY"}
|
||||
|
|
@ -267,13 +256,6 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text():
|
|||
# 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
|
||||
"content_block_delta", # {"city":
|
||||
"content_block_delta", # "NY"}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue