From 5be4266f5c7c10c69e4be5f3ec01d6a1d8bae9a1 Mon Sep 17 00:00:00 2001 From: Jovan <101700786+KuanJia@users.noreply.github.com> Date: Sat, 17 Oct 2026 14:10:16 +0800 Subject: [PATCH] [Bugfix][Frontend] Preserve Cohere and Anthropic streaming errors (#61947) Signed-off-by: Kevin Co-authored-by: OpenAI Codex --- .../test_anthropic_messages_conversion.py | 85 ++-- .../cohere/test_serving_streaming.py | 135 +++--- .../chat_completion/test_serving_chat.py | 146 +++++- vllm/entrypoints/anthropic/serving.py | 435 +++++++++--------- vllm/entrypoints/cohere/serving.py | 40 +- .../openai/chat_completion/serving.py | 65 ++- 6 files changed, 531 insertions(+), 375 deletions(-) diff --git a/tests/entrypoints/anthropic/test_anthropic_messages_conversion.py b/tests/entrypoints/anthropic/test_anthropic_messages_conversion.py index 399c898d0b..5e82072d18 100644 --- a/tests/entrypoints/anthropic/test_anthropic_messages_conversion.py +++ b/tests/entrypoints/anthropic/test_anthropic_messages_conversion.py @@ -49,6 +49,9 @@ from vllm.entrypoints.openai.chat_completion.protocol import ( ChatMessage, ) from vllm.entrypoints.serve.engine.protocol import PromptTokenUsageInfo, UsageInfo +from vllm.entrypoints.serve.exception_handling.error_response import ( + create_error_response, +) from vllm.entrypoints.serve.exception_handling.handlers.validation import ( validation_exception_handler, ) @@ -1034,7 +1037,7 @@ def _make_stream_chunk( stop_reason: int | str | None = None, choices: list[ChatCompletionResponseStreamChoice] | None = None, usage: UsageInfo | None = None, -) -> str: +) -> ChatCompletionStreamResponse: if choices is None: choices = [ ChatCompletionResponseStreamChoice( @@ -1051,7 +1054,7 @@ def _make_stream_chunk( choices=choices, usage=usage, ) - return f"data: {chunk.model_dump_json()}" + return chunk def _tc(*, args, id=None, name=None): @@ -1062,6 +1065,39 @@ def _tc(*, args, id=None, name=None): ) +@pytest.mark.asyncio +@pytest.mark.parametrize("block_type", [None, "text", "thinking", "tool_use"]) +async def test_stream_error_preserves_message_and_closes_block(block_type): + async def source(): + if block_type is not None: + yield _make_stream_chunk(delta=DeltaMessage(role="assistant")) + if block_type == "text": + delta = DeltaMessage(content="Hello") + elif block_type == "thinking": + delta = DeltaMessage(reasoning="Thinking") + else: + delta = DeltaMessage( + tool_calls=[_tc(id="call_1", name="search", args='{"q":')] + ) + yield _make_stream_chunk(delta=delta) + yield create_error_response(ValueError("backend failed")) + + converter = _make_stream_converter() + events = _parse_sse_events( + [event async for event in converter.message_stream_converter(source())] + ) + assert events[-1] == ( + "error", + { + "type": "error", + "error": {"type": "internal_error", "message": "backend failed"}, + }, + ) + assert not any(event_type == "message_stop" for event_type, _ in events) + if block_type is not None: + assert events[-2][0] == "content_block_stop" + + class TestMessageStreamConverterToolUseContentBuffering: """Regression test for tool_use arguments being silently dropped. @@ -1074,7 +1110,7 @@ class TestMessageStreamConverterToolUseContentBuffering: async def test_tool_use_args_not_dropped_when_content_in_same_chunk( self, ): - async def sse_input(): + async def chunk_input(): yield _make_stream_chunk( delta=DeltaMessage(role="assistant"), usage=UsageInfo(prompt_tokens=10, total_tokens=10), @@ -1111,11 +1147,10 @@ class TestMessageStreamConverterToolUseContentBuffering: completion_tokens=20, ), ) - yield "data: [DONE]" converter = _make_stream_converter() output = [] - async for event in converter.message_stream_converter(sse_input()): + async for event in converter.message_stream_converter(chunk_input()): output.append(event) events = _parse_sse_events(output) @@ -1156,7 +1191,7 @@ class TestMessageStreamConverterToolUseContentBuffering: stream jumps straight from finish_reason to [DONE], skipping the empty-choices usage chunk.""" - async def sse_input(): + async def chunk_input(): yield _make_stream_chunk( delta=DeltaMessage(role="assistant"), usage=UsageInfo(prompt_tokens=10, total_tokens=10), @@ -1178,11 +1213,10 @@ class TestMessageStreamConverterToolUseContentBuffering: 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()): + async for event in converter.message_stream_converter(chunk_input()): output.append(event) events = _parse_sse_events(output) @@ -1216,7 +1250,7 @@ class TestMessageStartIncludesTypeAndRole: @pytest.mark.asyncio async def test_message_start_contains_message_type_and_role(self): - async def sse_input(): + async def chunk_input(): yield _make_stream_chunk( delta=DeltaMessage(content="Hello"), usage=UsageInfo( @@ -1226,11 +1260,10 @@ class TestMessageStartIncludesTypeAndRole: ), ) yield _make_stream_chunk(finish_reason="stop") - yield "data: [DONE]" converter = _make_stream_converter() output = [] - async for event in converter.message_stream_converter(sse_input()): + async for event in converter.message_stream_converter(chunk_input()): output.append(event) events = _parse_sse_events(output) @@ -1260,7 +1293,7 @@ class TestStreamingCacheUsageSemantics: message_start omits cache fields. The final chunk carries prompt_tokens_details, so message_delta carries resolved values.""" - async def sse_input(): + async def chunk_input(): yield _make_stream_chunk( delta=DeltaMessage(role="assistant", content="hi"), usage=UsageInfo(prompt_tokens=100, total_tokens=100), @@ -1277,11 +1310,10 @@ class TestStreamingCacheUsageSemantics: ), ), ) - yield "data: [DONE]" converter = _make_stream_converter() output = [] - async for event in converter.message_stream_converter(sse_input()): + async for event in converter.message_stream_converter(chunk_input()): output.append(event) events = _parse_sse_events(output) @@ -1305,7 +1337,7 @@ class TestStreamingCacheUsageSemantics: """When the final chunk reports cached_tokens=0, message_delta carries cache fields = 0 (cache miss); message_start still omits them.""" - async def sse_input(): + async def chunk_input(): yield _make_stream_chunk( delta=DeltaMessage(role="assistant"), usage=UsageInfo(prompt_tokens=50, total_tokens=50), @@ -1322,11 +1354,10 @@ class TestStreamingCacheUsageSemantics: ), ), ) - yield "data: [DONE]" converter = _make_stream_converter() output = [] - async for event in converter.message_stream_converter(sse_input()): + async for event in converter.message_stream_converter(chunk_input()): output.append(event) events = _parse_sse_events(output) @@ -1346,7 +1377,7 @@ class TestStreamingCacheUsageSemantics: """If --enable-prompt-tokens-details is off, no chunk carries cache info; both message_start and message_delta omit cache fields.""" - async def sse_input(): + async def chunk_input(): yield _make_stream_chunk( delta=DeltaMessage(role="assistant"), usage=UsageInfo(prompt_tokens=30, total_tokens=30), @@ -1356,11 +1387,10 @@ class TestStreamingCacheUsageSemantics: choices=[], usage=UsageInfo(prompt_tokens=30, completion_tokens=2, total_tokens=32), ) - yield "data: [DONE]" converter = _make_stream_converter() output = [] - async for event in converter.message_stream_converter(sse_input()): + async for event in converter.message_stream_converter(chunk_input()): output.append(event) events = _parse_sse_events(output) @@ -1628,7 +1658,7 @@ class TestStopSequenceReason: @pytest.mark.asyncio async def test_streaming_stop_string_maps_to_stop_sequence(self): - async def sse_input(): + async def chunk_input(): yield _make_stream_chunk( delta=DeltaMessage(content="hi"), usage=UsageInfo(prompt_tokens=5, total_tokens=5, completion_tokens=0), @@ -1638,11 +1668,10 @@ class TestStopSequenceReason: choices=[], usage=UsageInfo(prompt_tokens=5, total_tokens=8, completion_tokens=3), ) - yield "data: [DONE]" converter = _make_stream_converter() output = [] - async for event in converter.message_stream_converter(sse_input()): + async for event in converter.message_stream_converter(chunk_input()): output.append(event) events = _parse_sse_events(output) @@ -1654,7 +1683,7 @@ class TestStopSequenceReason: async def test_streaming_no_stop_string_emits_explicit_null_stop_sequence(self): """exclude_unset=True drops stop_sequence unless it is set explicitly.""" - async def sse_input(): + async def chunk_input(): yield _make_stream_chunk(delta=DeltaMessage(role="assistant")) yield _make_stream_chunk(delta=DeltaMessage(content="hi")) yield _make_stream_chunk(finish_reason="stop") @@ -1662,11 +1691,10 @@ class TestStopSequenceReason: choices=[], usage=UsageInfo(prompt_tokens=5, total_tokens=8, completion_tokens=3), ) - yield "data: [DONE]" converter = _make_stream_converter() output = [] - async for event in converter.message_stream_converter(sse_input()): + async for event in converter.message_stream_converter(chunk_input()): output.append(event) events = _parse_sse_events(output) @@ -1724,7 +1752,7 @@ class TestToolUseStopReason: ], ) async def test_streaming(self, args, finish_reason, stop_reason, expected): - async def sse_input(): + async def chunk_input(): yield _make_stream_chunk(delta=DeltaMessage(role="assistant")) yield _make_stream_chunk( delta=DeltaMessage( @@ -1738,11 +1766,10 @@ class TestToolUseStopReason: choices=[], usage=UsageInfo(prompt_tokens=5, total_tokens=8, completion_tokens=3), ) - yield "data: [DONE]" converter = _make_stream_converter() events = _parse_sse_events( - [e async for e in converter.message_stream_converter(sse_input())] + [e async for e in converter.message_stream_converter(chunk_input())] ) msg_deltas = [data for ev_type, data in events if ev_type == "message_delta"] diff --git a/tests/entrypoints/cohere/test_serving_streaming.py b/tests/entrypoints/cohere/test_serving_streaming.py index 8c2e8623ad..c0dc86bf9a 100644 --- a/tests/entrypoints/cohere/test_serving_streaming.py +++ b/tests/entrypoints/cohere/test_serving_streaming.py @@ -5,7 +5,7 @@ The stream-translation entry point is :meth:`CohereServingChatV2._chat_completion_stream_to_v2`, which turns an -async iterable of OpenAI SSE chunks into Cohere's +async iterable of OpenAI chunk objects into Cohere's ``message-start → (content|tool-call|citation)* → message-end → [DONE]`` event stream. @@ -15,7 +15,7 @@ state machine. """ import json -from collections.abc import AsyncGenerator +from collections.abc import AsyncGenerator, Sequence from typing import Any import pytest @@ -26,10 +26,7 @@ from vllm.entrypoints.cohere.cohere_chat_message import ( from vllm.entrypoints.cohere.cohere_chat_message import ( CitationSource, ) -from vllm.entrypoints.cohere.protocol import ( - CohereChatV2Request, - MessageStartEvent, -) +from vllm.entrypoints.cohere.protocol import MessageStartEvent from vllm.entrypoints.cohere.serving import ( _DONE_FRAME, CohereServingChatV2, @@ -41,6 +38,7 @@ from vllm.entrypoints.cohere.serving import ( from vllm.entrypoints.openai.chat_completion.protocol import ( ChatCompletionStreamResponse, ) +from vllm.entrypoints.serve.engine.protocol import ErrorResponse # ---------------------------------------------------------------------- # Helpers @@ -85,9 +83,8 @@ def _make_chunk( usage: dict[str, Any] | None = None, omit_choices: bool = False, citations: list[Any] | None = None, -) -> str: - """Build the ``data: {...}\\n\\n`` SSE frame the production code - consumes.""" +) -> ChatCompletionStreamResponse: + """Build the raw chat chunk the production converter consumes.""" delta: dict[str, Any] = {} if role is not None: delta["role"] = role @@ -119,27 +116,23 @@ def _make_chunk( payload["choices"] = [] if usage is not None: payload["usage"] = usage - # Use ChatCompletionStreamResponse to normalize the payload shape. - chunk = ChatCompletionStreamResponse.model_validate(payload) - return f"data: {chunk.model_dump_json(exclude_none=False)}\n\n" + return ChatCompletionStreamResponse.model_validate(payload) -def _make_done() -> str: - return "data: [DONE]\n\n" - - -async def _stream_from(items: list[str]) -> AsyncGenerator[str, None]: +async def _stream_from( + items: Sequence[ChatCompletionStreamResponse | ErrorResponse], +) -> AsyncGenerator[ChatCompletionStreamResponse | ErrorResponse, None]: for item in items: yield item -async def _drain(serving: CohereServingChatV2, items: list[str]) -> list[str]: +async def _drain( + serving: CohereServingChatV2, + items: Sequence[ChatCompletionStreamResponse | ErrorResponse], +) -> list[str]: """Drive ``_chat_completion_stream_to_v2`` over ``items`` and collect the emitted SSE frames.""" - request = CohereChatV2Request( - model="m", messages=[{"role": "user", "content": "hi"}], stream=True - ) - gen = serving._chat_completion_stream_to_v2(_stream_from(items), request) + gen = serving._chat_completion_stream_to_v2(_stream_from(items)) return [frame async for frame in gen] @@ -793,15 +786,14 @@ class TestChatCompletionStreamToV2: assert "content-end" not in types @pytest.mark.asyncio - async def test_done_marker_in_middle_closes_open_block(self): - # Some upstreams send [DONE] without a trailing usage-only chunk. + async def test_stream_exhaustion_closes_open_block_without_usage(self): + # The raw stream may finish without a trailing usage-only chunk. # The translator must still emit message-end before [DONE] so # Cohere clients don't hang. serving = _serving(is_reasoning_model=True) items = [ _make_chunk(role="assistant"), _make_chunk(content="Hi"), - _make_done(), ] frames = await _drain(serving, items) types = [_parse_event(f)["type"] for f in frames[:-1]] @@ -815,72 +807,65 @@ class TestChatCompletionStreamToV2: assert frames[-1] == _DONE_FRAME @pytest.mark.asyncio - async def test_skips_empty_and_non_data_lines(self): + @pytest.mark.parametrize("block_type", [None, "text", "thinking", "tool_call"]) + async def test_upstream_error_preserves_message(self, block_type): serving = _serving() - items = [ - "\n", - "event: ping\n\n", - _make_chunk(role="assistant"), - "data: \n\n", # empty data - _make_chunk(content="Hi"), - _make_chunk( - omit_choices=True, - usage={ - "prompt_tokens": 1, - "completion_tokens": 1, - "total_tokens": 2, - }, - ), - ] + items: list[ChatCompletionStreamResponse | ErrorResponse] = [] + if block_type is not None: + items.append(_make_chunk(role="assistant")) + if block_type == "text": + items.append(_make_chunk(content="Hi")) + elif block_type == "thinking": + items.append(_make_chunk(reasoning="Thinking")) + else: + items.append( + _make_chunk( + tool_calls=[ + { + "index": 0, + "id": "call_1", + "function": {"name": "search", "arguments": '{"q":'}, + } + ] + ) + ) + items.append(serving.create_error_response(ValueError("backend failed"))) + frames = await _drain(serving, items) assert frames[-1] == _DONE_FRAME - # Should still produce a complete lifecycle. - types = [_parse_event(f)["type"] for f in frames[:-1]] - assert types[0] == "message-start" - assert types[-1] == "message-end" + events = [_parse_event(frame) for frame in frames[:-1]] + error_ends = [event for event in events if event["type"] == "message-end"] + assert len(error_ends) == 1 + assert error_ends[0]["delta"] == { + "finish_reason": "ERROR", + "error": "backend failed", + } + assert events[-1] == error_ends[0] + if block_type is not None: + assert events[-2]["type"] == ( + "tool-call-end" if block_type == "tool_call" else "content-end" + ) @pytest.mark.asyncio - async def test_exception_in_chunk_parsing_emits_error_message_end(self): - # An invalid JSON payload after the first chunk triggers the - # error path: a synthetic message-end with finish_reason=ERROR - # followed by [DONE]. - serving = _serving() - items = [ - _make_chunk(role="assistant"), - "data: {not valid json}\n\n", - ] - frames = await _drain(serving, items) - assert frames[-1] == _DONE_FRAME - # Find the error-shaped message-end. - error_end = _parse_event(frames[-2]) - assert error_end["type"] == "message-end" - assert error_end["delta"]["finish_reason"] == "ERROR" - assert "error" in error_end["delta"] - - @pytest.mark.asyncio - async def test_error_message_end_is_sanitized(self): + @pytest.mark.parametrize("started", [False, True]) + async def test_error_message_end_is_sanitized(self, started): # The client-visible ``error`` field on the terminal message-end # must be routed through ``sanitize_message`` so tracebacks, # host filesystem paths, and Python object memory addresses # can't leak from a mid-stream exception into the SSE payload. serving = _serving() - async def _raises_mid_stream() -> AsyncGenerator[str, None]: - yield _make_chunk(role="assistant") + async def _raises_mid_stream() -> AsyncGenerator[ + ChatCompletionStreamResponse | ErrorResponse, None + ]: + if started: + yield _make_chunk(role="assistant") raise RuntimeError( "boom reading /Users/dev/vllm/x.py" ) - request = CohereChatV2Request( - model="m", - messages=[{"role": "user", "content": "hi"}], - stream=True, - ) frames = [ - f - async for f in serving._chat_completion_stream_to_v2( - _raises_mid_stream(), request - ) + f async for f in serving._chat_completion_stream_to_v2(_raises_mid_stream()) ] assert frames[-1] == _DONE_FRAME error_end = _parse_event(frames[-2]) diff --git a/tests/entrypoints/openai/chat_completion/test_serving_chat.py b/tests/entrypoints/openai/chat_completion/test_serving_chat.py index 85404af06c..0c7526cfc9 100644 --- a/tests/entrypoints/openai/chat_completion/test_serving_chat.py +++ b/tests/entrypoints/openai/chat_completion/test_serving_chat.py @@ -28,6 +28,7 @@ from vllm.entrypoints.generate.base.serving import build_per_request_timing_metr from vllm.entrypoints.openai.chat_completion.protocol import ( ChatCompletionRequest, ChatCompletionResponse, + ChatCompletionStreamResponse, ) from vllm.entrypoints.openai.chat_completion.serving import ( OpenAIServingChat, @@ -43,7 +44,7 @@ from vllm.entrypoints.openai.models.serving import ( from vllm.entrypoints.openai.parser.harmony_utils import get_encoding from vllm.entrypoints.serve.engine.protocol import ErrorResponse from vllm.entrypoints.serve.utils.tool_calls_utils import resolve_finish_reason -from vllm.exceptions import QueueOverflowError, VLLMValidationError +from vllm.exceptions import GenerationError, QueueOverflowError, VLLMValidationError from vllm.inputs import TokensPrompt from vllm.logprobs import Logprob from vllm.multimodal.inputs import PlaceholderRange @@ -716,6 +717,149 @@ async def _collect_metrics_stream_chunks( return chunks +@pytest.mark.asyncio +@pytest.mark.parametrize("raw", [False, True]) +async def test_streaming_chunks_keep_usage_and_metadata(raw): + serving = _build_minimal_metrics_serving_chat(False) + request = ChatCompletionRequest( + model="test-model", + messages=[{"role": "user", "content": "Test"}], + stream=True, + stream_options={"include_usage": True}, + ) + metadata = RequestResponseMetadata(request_id="chatcmpl-test") + method = ( + serving.chat_completion_stream_generator_raw + if raw + else serving.chat_completion_stream_generator + ) + chunks = [ + chunk + async for chunk in method( + request, + _single_request_output(_make_metrics_request_output()), + "chatcmpl-test", + "test-model", + conversation=[], + tokenizer=MagicMock(), + request_metadata=metadata, + ) + ] + if raw: + assert all(isinstance(chunk, ChatCompletionStreamResponse) for chunk in chunks) + else: + assert chunks[-1] == "data: [DONE]\n\n" + first = json.loads(chunks[0].removeprefix("data: ")) + assert first["choices"][0]["logprobs"] is None + usage = json.loads(chunks[-2].removeprefix("data: ")) + assert "system_fingerprint" not in usage + assert "prompt_tokens_details" not in usage["usage"] + chunks = [ + ChatCompletionStreamResponse.model_validate_json( + chunk.removeprefix("data: ") + ) + for chunk in chunks[:-1] + ] + assert len(chunks) == 3 + assert chunks[0].choices[0].delta.role == "assistant" + assert chunks[1].choices[0].delta.content == "Hello" + assert chunks[1].choices[0].finish_reason == "stop" + assert chunks[2].choices == [] + assert chunks[2].usage.prompt_tokens == 3 + assert chunks[2].usage.completion_tokens == 2 + assert metadata.final_usage_info.total_tokens == 5 + + +@pytest.mark.asyncio +async def test_streaming_serialization_failure_emits_error_and_done(): + serving = _build_minimal_metrics_serving_chat(False) + serving.parser_cls = HarmonyParser + parser = MagicMock() + parser.parse_delta.return_value = DeltaMessage( + content="Hello", extra_object=object() + ) + serving._make_parser = MagicMock(return_value=parser) + request = ChatCompletionRequest( + model="test-model", messages=[{"role": "user", "content": "Test"}], stream=True + ) + frames = [ + frame + async for frame in serving.chat_completion_stream_generator( + request, + _single_request_output(_make_metrics_request_output()), + "chatcmpl-test", + "test-model", + conversation=[], + tokenizer=MagicMock(), + request_metadata=RequestResponseMetadata(request_id="chatcmpl-test"), + ) + ] + assert frames[-1] == "data: [DONE]\n\n" + error = ErrorResponse.model_validate_json(frames[-2].removeprefix("data: ")) + assert "serialize" in error.error.message + + +@pytest.mark.asyncio +@pytest.mark.parametrize("raw", [False, True]) +@pytest.mark.parametrize( + "failure", ["parser", "before_first", "after_content", "generation"] +) +async def test_streaming_errors_preserve_envelope(raw, failure): + serving = _build_minimal_metrics_serving_chat(False) + error = ( + GenerationError("backend failed") + if failure == "generation" + else ValueError("backend failed") + ) + if failure == "parser": + serving.parser_cls = HarmonyParser + serving._make_parser = MagicMock(side_effect=error) + + async def results(): + if failure == "after_content": + output = _make_metrics_request_output() + output.outputs[0].finish_reason = None + output.finished = False + yield output + raise error + + request = ChatCompletionRequest( + model="test-model", messages=[{"role": "user", "content": "Test"}], stream=True + ) + method = ( + serving.chat_completion_stream_generator_raw + if raw + else serving.chat_completion_stream_generator + ) + chunks = [ + chunk + async for chunk in method( + request, + results(), + "chatcmpl-test", + "test-model", + conversation=[], + tokenizer=MagicMock(), + request_metadata=RequestResponseMetadata(request_id="chatcmpl-test"), + ) + ] + if raw: + assert all( + isinstance(chunk, (ChatCompletionStreamResponse, ErrorResponse)) + for chunk in chunks + ) + assert isinstance(chunks[-1], ErrorResponse) + error_response = chunks[-1] + else: + assert chunks[-1] == "data: [DONE]\n\n" + assert chunks.count("data: [DONE]\n\n") == 1 + error_response = ErrorResponse.model_validate_json( + chunks[-2].removeprefix("data: ") + ) + assert error_response.error.message == "backend failed" + assert error_response.error.code == (500 if failure == "generation" else 400) + + def test_build_per_request_timing_metrics_valid_timestamps(): metrics = build_per_request_timing_metrics( _PER_REQUEST_STATS, num_generation_tokens=10 diff --git a/vllm/entrypoints/anthropic/serving.py b/vllm/entrypoints/anthropic/serving.py index 4b7f2b95a5..8eb6448550 100644 --- a/vllm/entrypoints/anthropic/serving.py +++ b/vllm/entrypoints/anthropic/serving.py @@ -820,7 +820,14 @@ class AnthropicServingMessages(OpenAIServingChat): elif isinstance(generator, ChatCompletionResponse): return self.messages_full_converter(generator) - return self.message_stream_converter(generator) + return generator + + def chat_completion_stream_generator( + self, *args: Any, **kwargs: Any + ) -> AsyncGenerator[str, None]: + return self.message_stream_converter( + self.chat_completion_stream_generator_raw(*args, **kwargs) + ) def messages_full_converter( self, @@ -881,7 +888,7 @@ class AnthropicServingMessages(OpenAIServingChat): async def message_stream_converter( self, - generator: AsyncGenerator[str, None], + generator: AsyncGenerator[ChatCompletionStreamResponse | ErrorResponse, None], ) -> AsyncGenerator[str, None]: try: @@ -991,238 +998,208 @@ class AnthropicServingMessages(OpenAIServingChat): 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", - ) - data = stop_message.model_dump_json( - exclude_unset=True, exclude_none=True - ) - yield wrap_data_with_event(data, "message_stop") - else: - origin_chunk = ChatCompletionStreamResponse.model_validate_json( - data_str - ) - - if first_item: - chunk = AnthropicStreamEvent( - type="message_start", - message=AnthropicMessagesResponse( - id=origin_chunk.id, - # Set explicitly: this event is serialized - # with exclude_unset=True, which drops - # default-valued fields, while strict - # Anthropic SDK clients require - # message.type/role (issue #45367). - type="message", - role="assistant", - content=[], - model=origin_chunk.model, - stop_reason=None, - stop_sequence=None, - usage=_build_anthropic_usage(origin_chunk.usage), - ), - ) - first_item = False - data = chunk.model_dump_json(exclude_unset=True) - yield wrap_data_with_event(data, "message_start") - continue - - # last chunk including usage info - if len(origin_chunk.choices) == 0: - for event in stop_and_flush(): - yield event - anthropic_stop_reason, anthropic_stop_sequence = ( - _to_anthropic_stop_reason( - finish_reason or "stop", - stop_sequence, - tool_used=tool_used, - tool_complete=( - tool_used and _is_json("".join(last_tool_args)) - ), - ) - ) - stop_delta = AnthropicDelta( - stop_reason=anthropic_stop_reason, - # Set explicitly so exclude_unset=True keeps it. - stop_sequence=anthropic_stop_sequence, - ) - chunk = AnthropicStreamEvent( - type="message_delta", - delta=stop_delta, - usage=_build_anthropic_usage(origin_chunk.usage), - ) - data = chunk.model_dump_json(exclude_unset=True) - yield wrap_data_with_event(data, "message_delta") - continue - - if origin_chunk.choices[0].finish_reason is not None: - finish_reason = origin_chunk.choices[0].finish_reason - stop_sequence = origin_chunk.choices[0].stop_reason - # continue - - # thinking / text content - reasoning_delta = origin_chunk.choices[0].delta.reasoning - if reasoning_delta is not None: - if reasoning_delta == "": - pass - else: - if state.block_type != "thinking": - for event in stop_and_flush(): - yield event - start_event = start_block( - AnthropicContentBlock( - type="thinking", thinking="" - ) - ) - yield start_event - chunk = AnthropicStreamEvent( - index=( - state.block_index - if state.block_index is not None - else state.content_block_index - ), - type="content_block_delta", - delta=AnthropicDelta( - type="thinking_delta", - thinking=reasoning_delta, - ), - ) - data = chunk.model_dump_json(exclude_unset=True) - yield wrap_data_with_event(data, "content_block_delta") - - 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_and_flush(): - yield event - start_event = start_block( - AnthropicContentBlock(type="text", text="") - ) - yield start_event - chunk = AnthropicStreamEvent( - index=( - state.block_index - if state.block_index is not None - else state.content_block_index - ), - type="content_block_delta", - delta=AnthropicDelta( - type="text_delta", - text=origin_chunk.choices[0].delta.content, - ), - ) - data = chunk.model_dump_json(exclude_unset=True) - yield wrap_data_with_event(data, "content_block_delta") - - # tool calls - process all tool calls in the delta - if len(origin_chunk.choices[0].delta.tool_calls) > 0: - for tool_call in origin_chunk.choices[0].delta.tool_calls: - if tool_call.id is not None: - # Update mapping for incremental updates - tool_index_to_id[tool_call.index] = tool_call.id - # Only create new block if different tool call - # AND has a name - tool_name = ( - tool_call.function.name - if tool_call.function - else None - ) - if ( - state.tool_use_id != tool_call.id - and tool_name is not None - ): - for event in stop_and_flush(): - yield event - tool_used = True - last_tool_args = [] - start_event = start_block( - AnthropicContentBlock( - type="tool_use", - id=tool_call.id, - name=tool_name, - input={}, - ) - ) - yield start_event - # Handle initial arguments if present - if ( - tool_call.function - and tool_call.function.arguments - and state.tool_use_id == tool_call.id - ): - last_tool_args.append( - tool_call.function.arguments - ) - chunk = AnthropicStreamEvent( - index=( - state.block_index - if state.block_index is not None - else state.content_block_index - ), - type="content_block_delta", - delta=AnthropicDelta( - type="input_json_delta", - partial_json=tool_call.function.arguments, - ), - ) - data = chunk.model_dump_json(exclude_unset=True) - yield wrap_data_with_event( - data, "content_block_delta" - ) - else: - # Incremental update - use index to find tool_use_id - tool_use_id = tool_index_to_id.get(tool_call.index) - if ( - tool_use_id is not None - and tool_call.function - and tool_call.function.arguments - and state.tool_use_id == tool_use_id - ): - last_tool_args.append( - tool_call.function.arguments - ) - chunk = AnthropicStreamEvent( - index=( - state.block_index - if state.block_index is not None - else state.content_block_index - ), - type="content_block_delta", - delta=AnthropicDelta( - type="input_json_delta", - partial_json=tool_call.function.arguments, - ), - ) - data = chunk.model_dump_json(exclude_unset=True) - yield wrap_data_with_event( - data, "content_block_delta" - ) - continue - else: - error_response = AnthropicStreamEvent( - type="error", - error=AnthropicError( - type="internal_error", - message="Invalid data format received", + async for origin_chunk in generator: + if isinstance(origin_chunk, ErrorResponse): + raise RuntimeError(origin_chunk.error.message) + if first_item: + chunk = AnthropicStreamEvent( + type="message_start", + message=AnthropicMessagesResponse( + id=origin_chunk.id, + # Set explicitly: this event is serialized + # with exclude_unset=True, which drops + # default-valued fields, while strict + # Anthropic SDK clients require + # message.type/role (issue #45367). + type="message", + role="assistant", + content=[], + model=origin_chunk.model, + stop_reason=None, + stop_sequence=None, + usage=_build_anthropic_usage(origin_chunk.usage), ), ) - data = error_response.model_dump_json(exclude_unset=True) - yield wrap_data_with_event(data, "error") + first_item = False + data = chunk.model_dump_json(exclude_unset=True) + yield wrap_data_with_event(data, "message_start") + continue + + # last chunk including usage info + if len(origin_chunk.choices) == 0: + for event in stop_and_flush(): + yield event + anthropic_stop_reason, anthropic_stop_sequence = ( + _to_anthropic_stop_reason( + finish_reason or "stop", + stop_sequence, + tool_used=tool_used, + tool_complete=( + tool_used and _is_json("".join(last_tool_args)) + ), + ) + ) + stop_delta = AnthropicDelta( + stop_reason=anthropic_stop_reason, + # Set explicitly so exclude_unset=True keeps it. + stop_sequence=anthropic_stop_sequence, + ) + chunk = AnthropicStreamEvent( + type="message_delta", + delta=stop_delta, + usage=_build_anthropic_usage(origin_chunk.usage), + ) + data = chunk.model_dump_json(exclude_unset=True) + yield wrap_data_with_event(data, "message_delta") + continue + + if origin_chunk.choices[0].finish_reason is not None: + finish_reason = origin_chunk.choices[0].finish_reason + stop_sequence = origin_chunk.choices[0].stop_reason + # continue + + # thinking / text content + reasoning_delta = origin_chunk.choices[0].delta.reasoning + if reasoning_delta is not None: + if reasoning_delta == "": + pass + else: + if state.block_type != "thinking": + for event in stop_and_flush(): + yield event + start_event = start_block( + AnthropicContentBlock(type="thinking", thinking="") + ) + yield start_event + chunk = AnthropicStreamEvent( + index=( + state.block_index + if state.block_index is not None + else state.content_block_index + ), + type="content_block_delta", + delta=AnthropicDelta( + type="thinking_delta", + thinking=reasoning_delta, + ), + ) + data = chunk.model_dump_json(exclude_unset=True) + yield wrap_data_with_event(data, "content_block_delta") + + 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_and_flush(): + yield event + start_event = start_block( + AnthropicContentBlock(type="text", text="") + ) + yield start_event + chunk = AnthropicStreamEvent( + index=( + state.block_index + if state.block_index is not None + else state.content_block_index + ), + type="content_block_delta", + delta=AnthropicDelta( + type="text_delta", + text=origin_chunk.choices[0].delta.content, + ), + ) + data = chunk.model_dump_json(exclude_unset=True) + yield wrap_data_with_event(data, "content_block_delta") + + # tool calls - process all tool calls in the delta + if len(origin_chunk.choices[0].delta.tool_calls) > 0: + for tool_call in origin_chunk.choices[0].delta.tool_calls: + if tool_call.id is not None: + # Update mapping for incremental updates + tool_index_to_id[tool_call.index] = tool_call.id + # Only create new block if different tool call + # AND has a name + tool_name = ( + tool_call.function.name if tool_call.function else None + ) + if ( + state.tool_use_id != tool_call.id + and tool_name is not None + ): + for event in stop_and_flush(): + yield event + tool_used = True + last_tool_args = [] + start_event = start_block( + AnthropicContentBlock( + type="tool_use", + id=tool_call.id, + name=tool_name, + input={}, + ) + ) + yield start_event + # Handle initial arguments if present + if ( + tool_call.function + and tool_call.function.arguments + and state.tool_use_id == tool_call.id + ): + last_tool_args.append(tool_call.function.arguments) + chunk = AnthropicStreamEvent( + index=( + state.block_index + if state.block_index is not None + else state.content_block_index + ), + type="content_block_delta", + delta=AnthropicDelta( + type="input_json_delta", + partial_json=tool_call.function.arguments, + ), + ) + data = chunk.model_dump_json(exclude_unset=True) + yield wrap_data_with_event(data, "content_block_delta") + else: + # Incremental update - use index to find tool_use_id + tool_use_id = tool_index_to_id.get(tool_call.index) + if ( + tool_use_id is not None + and tool_call.function + and tool_call.function.arguments + and state.tool_use_id == tool_use_id + ): + last_tool_args.append(tool_call.function.arguments) + chunk = AnthropicStreamEvent( + index=( + state.block_index + if state.block_index is not None + else state.content_block_index + ), + type="content_block_delta", + delta=AnthropicDelta( + type="input_json_delta", + partial_json=tool_call.function.arguments, + ), + ) + data = chunk.model_dump_json(exclude_unset=True) + yield wrap_data_with_event(data, "content_block_delta") + continue + for event in stop_and_flush(): + yield event + stop_message = AnthropicStreamEvent(type="message_stop") + data = stop_message.model_dump_json(exclude_unset=True, exclude_none=True) + yield wrap_data_with_event(data, "message_stop") except Exception as e: logger.exception("Error in message stream converter.") + for event in stop_active_block(): + yield event error_response = AnthropicStreamEvent( type="error", error=AnthropicError( diff --git a/vllm/entrypoints/cohere/serving.py b/vllm/entrypoints/cohere/serving.py index d4bf81d8c2..73fcda2e2f 100644 --- a/vllm/entrypoints/cohere/serving.py +++ b/vllm/entrypoints/cohere/serving.py @@ -321,7 +321,14 @@ class CohereServingChatV2(OpenAIServingChat): case ChatCompletionResponse(): return self._chat_completion_to_v2(generator, request) case _: - return self._chat_completion_stream_to_v2(generator, request) + return generator + + def chat_completion_stream_generator( + self, *args: Any, **kwargs: Any + ) -> AsyncGenerator[str, None]: + return self._chat_completion_stream_to_v2( + self.chat_completion_stream_generator_raw(*args, **kwargs) + ) def to_chat_completion_request( self, request: CohereChatV2Request @@ -1428,10 +1435,9 @@ class CohereServingChatV2(OpenAIServingChat): async def _chat_completion_stream_to_v2( self, - generator: AsyncGenerator[str, None], - request: CohereChatV2Request, + generator: AsyncGenerator[ChatCompletionStreamResponse | ErrorResponse, None], ) -> AsyncGenerator[str, None]: - """Translate an OpenAI-style chat completion SSE stream into Cohere's + """Translate raw OpenAI chat completion chunks into Cohere's v2 stream-event format. Cohere's v2 stream lifecycle is: @@ -1445,19 +1451,9 @@ class CohereServingChatV2(OpenAIServingChat): state = _StreamState() try: - async for item in generator: - if not item.startswith("data:"): - continue - data_str = item[len("data:") :].strip().rstrip("\n") - if not data_str: - continue - if data_str == "[DONE]": - # OpenAI's stream terminator. Fall through to the - # post-loop cleanup so we always emit ``message-end`` - # even if the usage-only chunk was skipped. - break - - chunk = ChatCompletionStreamResponse.model_validate_json(data_str) + async for chunk in generator: + if isinstance(chunk, ErrorResponse): + raise RuntimeError(chunk.error.message) state.last_chunk_id = chunk.id if not state.started: @@ -1514,7 +1510,9 @@ class CohereServingChatV2(OpenAIServingChat): except Exception as exc: logger.exception("Error converting chat completion stream to v2") - if state.started and not state.ended: + if not state.ended: + for ev in self._close_open_blocks(state): + yield ev yield _sse( json.dumps( { @@ -1530,11 +1528,7 @@ class CohereServingChatV2(OpenAIServingChat): yield _DONE_FRAME return - # Normal completion or ``[DONE]``: ensure ``message-end`` is always - # emitted. Upstream may close the stream without sending the final - # usage-only chunk (e.g. on shutdown, or when ``[DONE]`` is the only - # terminator); without this fallback Cohere clients would hang - # waiting for the closing event. + # Ensure message-end is emitted even without a final usage chunk. if state.started and not state.ended: for ev in self._close_open_blocks(state): yield ev diff --git a/vllm/entrypoints/openai/chat_completion/serving.py b/vllm/entrypoints/openai/chat_completion/serving.py index 7e1782e241..aac1ee77ff 100644 --- a/vllm/entrypoints/openai/chat_completion/serving.py +++ b/vllm/entrypoints/openai/chat_completion/serving.py @@ -478,6 +478,44 @@ class OpenAIServingChat(GenerateBaseServing): chat_template_kwargs: dict[str, Any] | None = None, mm_token_counts: dict[str, int] | None = None, ) -> AsyncGenerator[str, None]: + try: + async for chunk in self.chat_completion_stream_generator_raw( + request, + result_generator, + request_id, + model_name, + conversation, + tokenizer, + request_metadata, + chat_template_kwargs=chat_template_kwargs, + mm_token_counts=mm_token_counts, + ): + if isinstance(chunk, ErrorResponse): + data = chunk.model_dump_json() + else: + data = chunk.model_dump_json( + exclude_unset=True, exclude_none=not chunk.choices + ) + yield f"data: {data}\n\n" + except Exception as e: + logger.exception("Error serializing chat completion stream.") + data = self.create_streaming_error_response(e) + yield f"data: {data}\n\n" + yield "data: [DONE]\n\n" + + async def chat_completion_stream_generator_raw( + self, + request: ChatCompletionRequest, + result_generator: AsyncIterator[RequestOutput], + request_id: str, + model_name: str, + conversation: list[ConversationMessage], + tokenizer: TokenizerLike, + request_metadata: RequestResponseMetadata, + chat_template_kwargs: dict[str, Any] | None = None, + mm_token_counts: dict[str, int] | None = None, + ) -> AsyncGenerator[ChatCompletionStreamResponse | ErrorResponse, None]: + """Yield chat chunks or errors without serializing them into SSE frames.""" created_time = int(time.time()) chunk_object_type: Final = "chat.completion.chunk" first_iteration = True @@ -510,9 +548,7 @@ class OpenAIServingChat(GenerateBaseServing): parsers = [None] * num_choices except Exception as e: logger.exception("Error in parser creation.") - data = self.create_streaming_error_response(e) - yield f"data: {data}\n\n" - yield "data: [DONE]\n\n" + yield self.create_error_response(e) return stream_options = request.stream_options @@ -583,8 +619,7 @@ class OpenAIServingChat(GenerateBaseServing): ), ) - data = chunk.model_dump_json(exclude_unset=True) - yield f"data: {data}\n\n" + yield chunk # Send response to echo the input portion of the # last message @@ -624,8 +659,7 @@ class OpenAIServingChat(GenerateBaseServing): ), ) - data = chunk.model_dump_json(exclude_unset=True) - yield f"data: {data}\n\n" + yield chunk first_iteration = False for output in res.outputs: @@ -821,8 +855,7 @@ class OpenAIServingChat(GenerateBaseServing): ), ) - data = chunk.model_dump_json(exclude_unset=True) - yield f"data: {data}\n\n" + yield chunk if self._include_reasoning_tokens_details and not include_continuous_usage: for i, parser in enumerate(parsers): @@ -882,10 +915,7 @@ class OpenAIServingChat(GenerateBaseServing): system_fingerprint=self.system_fingerprint, metrics=stream_per_request_metrics, ) - final_usage_data = final_usage_chunk.model_dump_json( - exclude_unset=True, exclude_none=True - ) - yield f"data: {final_usage_data}\n\n" + yield final_usage_chunk # report to FastAPI middleware aggregate usage across all choices num_completion_tokens = sum(previous_num_tokens) @@ -919,13 +949,12 @@ class OpenAIServingChat(GenerateBaseServing): ) except GenerationError as e: - yield f"data: {self._convert_generation_error_to_streaming_response(e)}\n\n" + yield self.create_error_response( + str(e), err_type="InternalServerError", status_code=e.status_code + ) except Exception as e: logger.exception("Error in chat completion stream generator.") - data = self.create_streaming_error_response(e) - yield f"data: {data}\n\n" - # Send the final done message after all response.n are finished - yield "data: [DONE]\n\n" + yield self.create_error_response(e) async def chat_completion_full_generator( self,