iHore/vLLM
8
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[Bugfix] Fix Anthropic tool_use content handling dropping args (#45287)

Browse Source 
Signed-off-by: Ben Browning <bbrownin@redhat.com>
This commit is contained in:
Ben Browning
2026-06-11 20:01:29 +00:00
committed by GitHub
parent 3b03a2cf47
commit 235b63c004
2 changed files with 252 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/entrypoints/anthropic/test_anthropic_messages_conversion.py
+222 -1
View File
@@ -6,13 +6,29 @@ Tests the image source handling and tool_result content parsing in
AnthropicServingMessages._convert_anthropic_to_openai_request().
Also covers extended-thinking edge cases such as ``redacted_thinking``
blocks echoed back by Anthropic clients.
blocks echoed back by Anthropic clients, and streaming conversion in
``message_stream_converter``.
"""
import json
from unittest.mock import MagicMock
import pytest
from vllm.entrypoints.anthropic.protocol import (
AnthropicMessagesRequest,
)
from vllm.entrypoints.anthropic.serving import AnthropicServingMessages
from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionResponseStreamChoice,
ChatCompletionStreamResponse,
)
from vllm.entrypoints.openai.engine.protocol import (
DeltaFunctionCall,
DeltaMessage,
DeltaToolCall,
UsageInfo,
)
_convert = AnthropicServingMessages._convert_anthropic_to_openai_request
_img_url = AnthropicServingMessages._convert_image_source_to_url
@@ -775,3 +791,208 @@ class TestInlineSystemMessageInMessagesArray:
assert result.messages[0]["role"] == "system"
assert result.messages[0]["content"] == "Top-level prompt.Inline hint."
assert result.messages[1]["role"] == "user"
# ======================================================================
# Streaming conversion: message_stream_converter
# ======================================================================
def _make_stream_converter():
obj = MagicMock(spec=AnthropicServingMessages)
obj.stop_reason_map = {
"stop": "end_turn",
"length": "max_tokens",
"tool_calls": "tool_use",
}
obj.message_stream_converter = (
AnthropicServingMessages.message_stream_converter.__get__(obj)
)
return obj
def _parse_sse_events(raw_events: list[str]) -> list[tuple[str, dict]]:
results = []
for raw in raw_events:
headers = dict(
line.split(": ", 1) for line in raw.strip().split("\n") if ": " in line
)
if "event" in headers and "data" in headers:
results.append((headers["event"], json.loads(headers["data"])))
return results
def _make_stream_chunk(
*,
delta: DeltaMessage | None = None,
finish_reason: str | None = None,
choices: list[ChatCompletionResponseStreamChoice] | None = None,
usage: UsageInfo | None = None,
) -> str:
if choices is None:
choices = [
ChatCompletionResponseStreamChoice(
index=0,
delta=delta or DeltaMessage(),
finish_reason=finish_reason,
)
]
chunk = ChatCompletionStreamResponse(
id="chatcmpl-test",
created=0,
model="test-model",
choices=choices,
usage=usage,
)
return f"data: {chunk.model_dump_json()}"
def _tc(*, args, id=None, name=None):
return DeltaToolCall(
index=0,
id=id,
function=DeltaFunctionCall(name=name, arguments=args),
)
class TestMessageStreamConverterToolUseContentBuffering:
"""Regression test for tool_use arguments being silently dropped.
With speculative decoding or multi-token prediction, a single delta
can carry both the final tool_call argument fragment and trailing
content.
"""
@pytest.mark.asyncio
async def test_tool_use_args_not_dropped_when_content_in_same_chunk(
self,
):
async def sse_input():
yield _make_stream_chunk(
delta=DeltaMessage(role="assistant"),
usage=UsageInfo(prompt_tokens=10, total_tokens=10),
)
yield _make_stream_chunk(
delta=DeltaMessage(
tool_calls=[
_tc(id="call_abc123", name="read_file", args=""),
]
)
)
yield _make_stream_chunk(
delta=DeltaMessage(
tool_calls=[
_tc(args='{"path":"/tmp/f"'),
]
)
)
# BUG TRIGGER: final tool_call args and trailing content in
# one delta, as happens with spec decoding / multi-token
# prediction where multiple tokens land in a single chunk.
yield _make_stream_chunk(
delta=DeltaMessage(
content="\nOkay",
tool_calls=[_tc(args="}")],
)
)
yield _make_stream_chunk(finish_reason="tool_calls")
yield _make_stream_chunk(
choices=[],
usage=UsageInfo(
prompt_tokens=10,
total_tokens=30,
completion_tokens=20,
),
)
yield "data: [DONE]"
converter = _make_stream_converter()
output = []
async for event in converter.message_stream_converter(sse_input()):
output.append(event)
events = _parse_sse_events(output)
assert events[0][0] == "message_start"
arg_fragments = [
data["delta"]["partial_json"]
for _, data in events
if data.get("delta", {}).get("type") == "input_json_delta"
]
full_args = "".join(arg_fragments)
assert full_args == '{"path":"/tmp/f"}'
text_deltas = [
data["delta"]["text"]
for _, data in events
if data.get("delta", {}).get("type") == "text_delta"
]
assert text_deltas == ["\nOkay"]
block_starts = [
(data["content_block"]["type"], data.get("index"))
for ev_type, data in events
if ev_type == "content_block_start"
]
assert block_starts[0] == ("tool_use", 0)
assert block_starts[1] == ("text", 1)
msg_deltas = [data for ev_type, data in events if ev_type == "message_delta"]
assert msg_deltas[0]["delta"]["stop_reason"] == "tool_use"
assert events[-1][0] == "message_stop"
@pytest.mark.asyncio
async def test_buffered_content_flushed_on_done_without_usage_chunk(self):
"""Content buffered during tool_use must be emitted even if the
stream jumps straight from finish_reason to [DONE], skipping the
empty-choices usage chunk."""
async def sse_input():
yield _make_stream_chunk(
delta=DeltaMessage(role="assistant"),
usage=UsageInfo(prompt_tokens=10, total_tokens=10),
)
yield _make_stream_chunk(
delta=DeltaMessage(
tool_calls=[
_tc(id="call_xyz", name="get_weather", args=""),
]
)
)
yield _make_stream_chunk(
delta=DeltaMessage(
tool_calls=[_tc(args='{"city":"NYC"}')],
)
)
yield _make_stream_chunk(
delta=DeltaMessage(content="\nDone"),
finish_reason="tool_calls",
)
# No empty-choices usage chunk — go straight to [DONE].
yield "data: [DONE]"
converter = _make_stream_converter()
output = []
async for event in converter.message_stream_converter(sse_input()):
output.append(event)
events = _parse_sse_events(output)
text_deltas = [
data["delta"]["text"]
for _, data in events
if data.get("delta", {}).get("type") == "text_delta"
]
assert text_deltas == ["\nDone"]
block_starts = [
data["content_block"]["type"]
for ev_type, data in events
if ev_type == "content_block_start"
]
assert "tool_use" in block_starts
assert "text" in block_starts
assert events[-1][0] == "message_stop"
vllm/entrypoints/anthropic/serving.py
+30 -4
View File
@@ -564,6 +564,7 @@ class AnthropicServingMessages(OpenAIServingChat):
self.block_signature: str | None = None
self.signature_emitted: bool = False
self.tool_use_id: str | None = None
self.pending_content: list[str] = []
def reset(self) -> None:
self.block_type = None
@@ -571,6 +572,7 @@ class AnthropicServingMessages(OpenAIServingChat):
self.block_signature = None
self.signature_emitted = False
self.tool_use_id = None
self.pending_content.clear()
def start(self, block: AnthropicContentBlock) -> None:
self.block_type = block.type
@@ -635,10 +637,30 @@ class AnthropicServingMessages(OpenAIServingChat):
state.start(block)
return event
def stop_and_flush() -> list[str]:
buffered = list(state.pending_content)
state.pending_content.clear()
events = stop_active_block()
if not buffered:
return events
text = "".join(buffered)
events.append(start_block(AnthropicContentBlock(type="text", text="")))
pc_chunk = AnthropicStreamEvent(
index=state.block_index,
type="content_block_delta",
delta=AnthropicDelta(type="text_delta", text=text),
)
pc_data = pc_chunk.model_dump_json(exclude_unset=True)
events.append(wrap_data_with_event(pc_data, "content_block_delta"))
events.extend(stop_active_block())
return events
async for item in generator:
if item.startswith("data:"):
data_str = item[5:].strip().rstrip("\n")
if data_str == "[DONE]":
for event in stop_and_flush():
yield event
stop_message = AnthropicStreamEvent(
type="message_stop",
)
@@ -675,7 +697,7 @@ class AnthropicServingMessages(OpenAIServingChat):
# last chunk including usage info
if len(origin_chunk.choices) == 0:
for event in stop_active_block():
for event in stop_and_flush():
yield event
stop_reason = self.stop_reason_map.get(
finish_reason or "stop"
@@ -707,7 +729,7 @@ class AnthropicServingMessages(OpenAIServingChat):
pass
else:
if state.block_type != "thinking":
for event in stop_active_block():
for event in stop_and_flush():
yield event
start_event = start_block(
AnthropicContentBlock(
@@ -733,9 +755,13 @@ class AnthropicServingMessages(OpenAIServingChat):
if origin_chunk.choices[0].delta.content is not None:
if origin_chunk.choices[0].delta.content == "":
pass
elif state.block_type == "tool_use":
state.pending_content.append(
origin_chunk.choices[0].delta.content
)
else:
if state.block_type != "text":
for event in stop_active_block():
for event in stop_and_flush():
yield event
start_event = start_block(
AnthropicContentBlock(type="text", text="")
@@ -773,7 +799,7 @@ class AnthropicServingMessages(OpenAIServingChat):
state.tool_use_id != tool_call.id
and tool_name is not None
):
for event in stop_active_block():
for event in stop_and_flush():
yield event
start_event = start_block(
AnthropicContentBlock(