mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-11 22:51:28 +00:00
fix: emit input_json_delta for atomic tool call chunks in Anthropic streaming adapter
Signed-off-by: ItsRoy69 <jyotirmoyroy649@gmail.com>
This commit is contained in:
ItsRoy69
parent
5544803b35
commit
df4213c619
2 changed files with 364 additions and 4 deletions
|
|
@ -129,8 +129,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 +142,9 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
"content_block": self.current_content_block_start,
|
||||
}
|
||||
)
|
||||
# Also queue the processed chunk if it carries meaningful streaming chunk, so dropping it loses the arguments.
|
||||
if self._has_meaningful_delta(processed_chunk):
|
||||
self.chunk_queue.append(processed_chunk)
|
||||
self.sent_content_block_finish = False
|
||||
return self.chunk_queue.popleft()
|
||||
|
||||
|
|
@ -305,8 +306,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
|
||||
# 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(
|
||||
|
|
@ -325,6 +324,11 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
}
|
||||
)
|
||||
|
||||
# 3. Also queue the processed chunk if it carries meaningful delta data (e.g. tool-call arguments).
|
||||
# AI Providers deliver name+arguments in a single streaming chunk, so dropping it loses the arguments.
|
||||
if self._has_meaningful_delta(processed_chunk):
|
||||
self.chunk_queue.append(processed_chunk)
|
||||
|
||||
# Reset state for new block
|
||||
self.sent_content_block_finish = False
|
||||
|
||||
|
|
@ -424,6 +428,27 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
|
|||
# For non-dict chunks, forward the original value unchanged
|
||||
yield chunk
|
||||
|
||||
def _has_meaningful_delta(self, processed_chunk: Any) -> bool:
|
||||
"""
|
||||
Check if a processed chunk contains meaningful delta content that
|
||||
should be emitted alongside a content_block_start event.
|
||||
|
||||
AI Providers deliver complete tool calls (name + arguments) in a single streaming chunk. Without emitting the delta, the arguments are silently dropped.
|
||||
"""
|
||||
if processed_chunk.get("type") != "content_block_delta":
|
||||
return False
|
||||
delta = processed_chunk.get("delta", {})
|
||||
delta_type = delta.get("type", "")
|
||||
if delta_type == "input_json_delta":
|
||||
return bool(delta.get("partial_json"))
|
||||
elif delta_type == "text_delta":
|
||||
return bool(delta.get("text"))
|
||||
elif delta_type == "thinking_delta":
|
||||
return bool(delta.get("thinking"))
|
||||
elif delta_type == "signature_delta":
|
||||
return bool(delta.get("signature"))
|
||||
return False
|
||||
|
||||
def _increment_content_block_index(self):
|
||||
self.current_content_block_index += 1
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,335 @@
|
|||
import os
|
||||
import sys
|
||||
from typing import Any, Dict, List
|
||||
|
||||
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,
|
||||
ModelResponseStream,
|
||||
StreamingChoices,
|
||||
Usage,
|
||||
)
|
||||
|
||||
|
||||
def _make_text_chunk(text: str, finish_reason=None) -> ModelResponseStream:
|
||||
"""Helper to build a streaming chunk with text content."""
|
||||
return ModelResponseStream(
|
||||
id="chatcmpl-test",
|
||||
choices=[
|
||||
StreamingChoices(
|
||||
finish_reason=finish_reason,
|
||||
index=0,
|
||||
delta=Delta(
|
||||
content=text,
|
||||
role="assistant",
|
||||
function_call=None,
|
||||
tool_calls=None,
|
||||
audio=None,
|
||||
),
|
||||
)
|
||||
],
|
||||
model="test-model",
|
||||
)
|
||||
|
||||
|
||||
def _make_tool_call_chunk(
|
||||
name: str, arguments: str, tool_call_id: str = "call_abc123", finish_reason=None
|
||||
) -> ModelResponseStream:
|
||||
"""
|
||||
Helper to build a streaming chunk with a tool call that carries BOTH
|
||||
name and arguments in the same chunk (atomic delivery).
|
||||
"""
|
||||
return ModelResponseStream(
|
||||
id="chatcmpl-test",
|
||||
choices=[
|
||||
StreamingChoices(
|
||||
finish_reason=finish_reason,
|
||||
index=0,
|
||||
delta=Delta(
|
||||
content=None,
|
||||
role="assistant",
|
||||
function_call=None,
|
||||
tool_calls=[
|
||||
ChatCompletionDeltaToolCall(
|
||||
id=tool_call_id,
|
||||
function=Function(arguments=arguments, name=name),
|
||||
type="function",
|
||||
index=0,
|
||||
)
|
||||
],
|
||||
audio=None,
|
||||
),
|
||||
)
|
||||
],
|
||||
model="test-model",
|
||||
)
|
||||
|
||||
|
||||
def _make_finish_chunk(stop_reason: str = "tool_use") -> ModelResponseStream:
|
||||
"""Helper to build a streaming chunk with only a finish reason."""
|
||||
return ModelResponseStream(
|
||||
id="chatcmpl-test",
|
||||
choices=[
|
||||
StreamingChoices(
|
||||
finish_reason=stop_reason,
|
||||
index=0,
|
||||
delta=Delta(
|
||||
content=None,
|
||||
role="assistant",
|
||||
function_call=None,
|
||||
tool_calls=None,
|
||||
audio=None,
|
||||
),
|
||||
)
|
||||
],
|
||||
model="test-model",
|
||||
usage=Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Sync iterator tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_emit_input_json_delta_for_atomic_tool_call():
|
||||
"""
|
||||
When a provider delivers name + arguments in a single streaming chunk
|
||||
(atomic delivery), the adapter must emit an input_json_delta event
|
||||
between content_block_start and content_block_stop.
|
||||
|
||||
Regression test for https://github.com/BerriAI/litellm/issues/25561
|
||||
"""
|
||||
chunks = [
|
||||
_make_text_chunk("Let me read that file for you."),
|
||||
_make_tool_call_chunk("Read", '{"file_path": "/etc/hosts"}'),
|
||||
_make_finish_chunk("tool_use"),
|
||||
]
|
||||
|
||||
wrapper = AnthropicStreamWrapper(
|
||||
completion_stream=iter(chunks),
|
||||
model="test-model",
|
||||
)
|
||||
|
||||
events: List[Dict[str, Any]] = list(wrapper)
|
||||
|
||||
# Gather the event types
|
||||
event_types = [e.get("type") for e in events]
|
||||
|
||||
# There must be at least one content_block_delta with input_json_delta
|
||||
input_json_deltas: List[Dict[str, Any]] = [
|
||||
e
|
||||
for e in events
|
||||
if e.get("type") == "content_block_delta"
|
||||
and e.get("delta", {}).get("type") == "input_json_delta"
|
||||
]
|
||||
assert len(input_json_deltas) >= 1, (
|
||||
f"Expected at least one input_json_delta event, got 0. "
|
||||
f"Event types: {event_types}"
|
||||
)
|
||||
|
||||
# The concatenated partial_json must produce the correct arguments
|
||||
combined_json = "".join(
|
||||
d["delta"]["partial_json"] for d in input_json_deltas
|
||||
)
|
||||
assert combined_json == '{"file_path": "/etc/hosts"}'
|
||||
|
||||
|
||||
def test_correct_event_sequence_for_atomic_tool_call():
|
||||
"""
|
||||
Verify the full event sequence: message_start, text block (start/delta/stop),
|
||||
tool_use block (start/delta/stop), message_delta, message_stop.
|
||||
"""
|
||||
chunks = [
|
||||
_make_text_chunk("Hello"),
|
||||
_make_tool_call_chunk("Read", '{"file_path": "/etc/hosts"}'),
|
||||
_make_finish_chunk("tool_use"),
|
||||
]
|
||||
|
||||
wrapper = AnthropicStreamWrapper(
|
||||
completion_stream=iter(chunks),
|
||||
model="test-model",
|
||||
)
|
||||
|
||||
events: List[Dict[str, Any]] = list(wrapper)
|
||||
event_types = [e.get("type") for e in events]
|
||||
|
||||
# Expected sequence pattern
|
||||
assert event_types[0] == "message_start"
|
||||
|
||||
# Text content block
|
||||
assert "content_block_start" in event_types
|
||||
assert "content_block_stop" in event_types
|
||||
|
||||
# tool_use content block
|
||||
tool_block_starts: List[Dict[str, Any]] = [
|
||||
e for e in events
|
||||
if e.get("type") == "content_block_start"
|
||||
and e.get("content_block", {}).get("type") == "tool_use"
|
||||
]
|
||||
assert len(tool_block_starts) == 1
|
||||
assert tool_block_starts[0]["content_block"]["name"] == "Read"
|
||||
|
||||
# input_json_delta must appear between the tool_use start and stop
|
||||
tool_start_idx = events.index(tool_block_starts[0])
|
||||
input_deltas_after_tool_start: List[Dict[str, Any]] = [
|
||||
e for e in events[tool_start_idx:]
|
||||
if e.get("type") == "content_block_delta"
|
||||
and e.get("delta", {}).get("type") == "input_json_delta"
|
||||
]
|
||||
assert len(input_deltas_after_tool_start) >= 1
|
||||
|
||||
# message_stop at the end
|
||||
assert event_types[-1] == "message_stop"
|
||||
|
||||
|
||||
def test_split_tool_call_across_chunks():
|
||||
"""
|
||||
For providers that split name and arguments across multiple chunks,
|
||||
the existing behavior should still work correctly.
|
||||
"""
|
||||
# Chunk 1: tool call with name, empty arguments
|
||||
name_chunk = ModelResponseStream(
|
||||
id="chatcmpl-test",
|
||||
choices=[
|
||||
StreamingChoices(
|
||||
finish_reason=None,
|
||||
index=0,
|
||||
delta=Delta(
|
||||
content=None,
|
||||
role="assistant",
|
||||
function_call=None,
|
||||
tool_calls=[
|
||||
ChatCompletionDeltaToolCall(
|
||||
id="call_abc123",
|
||||
function=Function(arguments="", name="Read"),
|
||||
type="function",
|
||||
index=0,
|
||||
)
|
||||
],
|
||||
audio=None,
|
||||
),
|
||||
)
|
||||
],
|
||||
model="gpt-4",
|
||||
)
|
||||
|
||||
# Chunk 2: arguments only (no name)
|
||||
args_chunk = ModelResponseStream(
|
||||
id="chatcmpl-test",
|
||||
choices=[
|
||||
StreamingChoices(
|
||||
finish_reason=None,
|
||||
index=0,
|
||||
delta=Delta(
|
||||
content=None,
|
||||
role="assistant",
|
||||
function_call=None,
|
||||
tool_calls=[
|
||||
ChatCompletionDeltaToolCall(
|
||||
id=None,
|
||||
function=Function(
|
||||
arguments='{"file_path": "/etc/hosts"}', name=None
|
||||
),
|
||||
type="function",
|
||||
index=0,
|
||||
)
|
||||
],
|
||||
audio=None,
|
||||
),
|
||||
)
|
||||
],
|
||||
model="gpt-4",
|
||||
)
|
||||
|
||||
chunks = [
|
||||
_make_text_chunk("Let me look."),
|
||||
name_chunk,
|
||||
args_chunk,
|
||||
_make_finish_chunk("tool_use"),
|
||||
]
|
||||
|
||||
wrapper = AnthropicStreamWrapper(
|
||||
completion_stream=iter(chunks),
|
||||
model="gpt-4",
|
||||
)
|
||||
|
||||
events: List[Dict[str, Any]] = list(wrapper)
|
||||
|
||||
# Should still produce input_json_delta events
|
||||
input_json_deltas: List[Dict[str, Any]] = [
|
||||
e
|
||||
for e in events
|
||||
if e.get("type") == "content_block_delta"
|
||||
and e.get("delta", {}).get("type") == "input_json_delta"
|
||||
]
|
||||
assert len(input_json_deltas) >= 1
|
||||
|
||||
combined_json = "".join(
|
||||
d["delta"]["partial_json"] for d in input_json_deltas
|
||||
)
|
||||
assert combined_json == '{"file_path": "/etc/hosts"}'
|
||||
|
||||
|
||||
def test_has_meaningful_delta():
|
||||
"""Verify the _has_meaningful_delta helper classifies delta types correctly."""
|
||||
wrapper = AnthropicStreamWrapper(
|
||||
completion_stream=iter([]),
|
||||
model="test-model",
|
||||
)
|
||||
|
||||
# input_json_delta with content -> meaningful
|
||||
assert wrapper._has_meaningful_delta(
|
||||
{
|
||||
"type": "content_block_delta",
|
||||
"index": 0,
|
||||
"delta": {"type": "input_json_delta", "partial_json": '{"a": 1}'},
|
||||
}
|
||||
) is True
|
||||
|
||||
# input_json_delta with empty string -> not meaningful
|
||||
assert wrapper._has_meaningful_delta(
|
||||
{
|
||||
"type": "content_block_delta",
|
||||
"index": 0,
|
||||
"delta": {"type": "input_json_delta", "partial_json": ""},
|
||||
}
|
||||
) is False
|
||||
|
||||
# text_delta with content -> meaningful
|
||||
assert wrapper._has_meaningful_delta(
|
||||
{
|
||||
"type": "content_block_delta",
|
||||
"index": 0,
|
||||
"delta": {"type": "text_delta", "text": "hello"},
|
||||
}
|
||||
) is True
|
||||
|
||||
# text_delta with empty string -> not meaningful
|
||||
assert wrapper._has_meaningful_delta(
|
||||
{
|
||||
"type": "content_block_delta",
|
||||
"index": 0,
|
||||
"delta": {"type": "text_delta", "text": ""},
|
||||
}
|
||||
) is False
|
||||
|
||||
# message_delta (not a content_block_delta) -> not meaningful
|
||||
assert wrapper._has_meaningful_delta(
|
||||
{"type": "message_delta", "delta": {"stop_reason": "end_turn"}}
|
||||
) is False
|
||||
|
||||
# thinking_delta with content -> meaningful
|
||||
assert wrapper._has_meaningful_delta(
|
||||
{
|
||||
"type": "content_block_delta",
|
||||
"index": 0,
|
||||
"delta": {"type": "thinking_delta", "thinking": "Let me think..."},
|
||||
}
|
||||
) is True
|
||||
Loading…
Add table
Reference in a new issue