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

[Bugfix][Frontend] Preserve Cohere and Anthropic streaming errors (#61947)

Browse Source 
Signed-off-by: Kevin <jia__kuan@163.com>
Co-authored-by: OpenAI Codex <noreply@openai.com>
This commit is contained in:
Jovan
2026-10-17 06:10:16 +00:00
committed by GitHub
co-authored by OpenAI Codex
parent 90d6d871d6
commit 5be4266f5c
6 changed files with 531 additions and 375 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vllm/entrypoints/anthropic/serving.py
+206 -229
View File
@@ -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(
vllm/entrypoints/cohere/serving.py
+17 -23
View File
@@ -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
vllm/entrypoints/openai/chat_completion/serving.py
+47 -18
View File
@@ -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,