From 235b63c0046d2fbf4ab1bf810a1eb729f1f3fc27 Mon Sep 17 00:00:00 2001 From: Ben Browning Date: Thu, 11 Jun 2026 16:01:29 -0400 Subject: [PATCH] [Bugfix] Fix Anthropic tool_use content handling dropping args (#45287) Signed-off-by: Ben Browning --- .../test_anthropic_messages_conversion.py | 223 +++++++++++++++++- vllm/entrypoints/anthropic/serving.py | 34 ++- 2 files changed, 252 insertions(+), 5 deletions(-) diff --git a/tests/entrypoints/anthropic/test_anthropic_messages_conversion.py b/tests/entrypoints/anthropic/test_anthropic_messages_conversion.py index ad9fed1d35..21d5154c67 100644 --- a/tests/entrypoints/anthropic/test_anthropic_messages_conversion.py +++ b/tests/entrypoints/anthropic/test_anthropic_messages_conversion.py @@ -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" diff --git a/vllm/entrypoints/anthropic/serving.py b/vllm/entrypoints/anthropic/serving.py index 8f6cccdb0f..266a315421 100644 --- a/vllm/entrypoints/anthropic/serving.py +++ b/vllm/entrypoints/anthropic/serving.py @@ -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(