[Bugfix][Frontend] Return streaming errors before the first token (#40986)
Signed-off-by: 1fanwang <1fannnw@gmail.com> Co-authored-by: Robert Shaw <114415538+robertgshaw2-redhat@users.noreply.github.com>
This commit is contained in:
co-authored by
Robert Shaw
parent
89439db727
commit
b0eb87fe49
@@ -248,7 +248,8 @@ async def test_renderer_only_chat_request_skips_mm_cache():
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_chat_error_stream():
|
||||
@pytest.mark.parametrize("include_usage", [False, True])
|
||||
async def test_chat_error_stream(include_usage: bool):
|
||||
"""Test finish_reason='error' returns 500 InternalServerError (streaming)."""
|
||||
mock_engine = MagicMock(spec=AsyncLLM)
|
||||
mock_engine.errored = False
|
||||
@@ -313,6 +314,67 @@ async def test_chat_error_stream():
|
||||
messages=[{"role": "user", "content": "Test prompt"}],
|
||||
max_tokens=10,
|
||||
stream=True,
|
||||
stream_options={"include_usage": include_usage},
|
||||
)
|
||||
|
||||
response = await serving_chat.create_chat_completion(request)
|
||||
|
||||
chunks = []
|
||||
async for chunk in response:
|
||||
chunks.append(chunk)
|
||||
|
||||
assert len(chunks) >= 2
|
||||
assert any("Internal server error" in chunk for chunk in chunks), (
|
||||
f"Expected error message in chunks: {chunks}"
|
||||
)
|
||||
assert chunks[-1] == "data: [DONE]\n\n"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("include_usage", [False, True])
|
||||
async def test_chat_error_stream_first_chunk(include_usage: bool):
|
||||
"""An error before the first token must reach the streaming client."""
|
||||
mock_engine = MagicMock(spec=AsyncLLM)
|
||||
mock_engine.errored = False
|
||||
mock_engine.model_config = MockModelConfig()
|
||||
mock_engine.input_processor = MagicMock()
|
||||
mock_engine.renderer = _build_renderer(mock_engine.model_config)
|
||||
|
||||
serving_chat = _build_serving_chat(mock_engine)
|
||||
|
||||
completion_output = CompletionOutput(
|
||||
index=0,
|
||||
text="",
|
||||
token_ids=[],
|
||||
cumulative_logprob=None,
|
||||
logprobs=None,
|
||||
finish_reason="error",
|
||||
)
|
||||
|
||||
request_output = RequestOutput(
|
||||
request_id="test-id",
|
||||
prompt="Test prompt",
|
||||
prompt_token_ids=[1, 2, 3],
|
||||
prompt_logprobs=None,
|
||||
outputs=[completion_output],
|
||||
finished=True,
|
||||
metrics=None,
|
||||
lora_request=None,
|
||||
encoder_prompt=None,
|
||||
encoder_prompt_token_ids=None,
|
||||
)
|
||||
|
||||
async def mock_generate(*args, **kwargs):
|
||||
yield request_output
|
||||
|
||||
mock_engine.generate = MagicMock(side_effect=mock_generate)
|
||||
|
||||
request = ChatCompletionRequest(
|
||||
model=MODEL_NAME,
|
||||
messages=[{"role": "user", "content": "Test prompt"}],
|
||||
max_tokens=10,
|
||||
stream=True,
|
||||
stream_options={"include_usage": include_usage},
|
||||
)
|
||||
|
||||
response = await serving_chat.create_chat_completion(request)
|
||||
|
||||
@@ -383,10 +383,14 @@ async def test_completion_suffix_supports_text_prompt_list():
|
||||
|
||||
assert result == [{"ok": True}]
|
||||
assert online_renderer.preprocess_completion.call_args.kwargs["prompt_input"] == [
|
||||
f"{DEEPSEEK_V4_FIM_BEGIN}prefix A"
|
||||
f"{DEEPSEEK_V4_FIM_HOLE} suffix{DEEPSEEK_V4_FIM_END}",
|
||||
f"{DEEPSEEK_V4_FIM_BEGIN}prefix B"
|
||||
f"{DEEPSEEK_V4_FIM_HOLE} suffix{DEEPSEEK_V4_FIM_END}",
|
||||
(
|
||||
f"{DEEPSEEK_V4_FIM_BEGIN}prefix A"
|
||||
f"{DEEPSEEK_V4_FIM_HOLE} suffix{DEEPSEEK_V4_FIM_END}"
|
||||
),
|
||||
(
|
||||
f"{DEEPSEEK_V4_FIM_BEGIN}prefix B"
|
||||
f"{DEEPSEEK_V4_FIM_HOLE} suffix{DEEPSEEK_V4_FIM_END}"
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@@ -619,7 +623,8 @@ async def test_renderer_only_completion_request_skips_mm_cache():
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_completion_error_stream():
|
||||
@pytest.mark.parametrize("include_usage", [False, True])
|
||||
async def test_completion_error_stream(include_usage: bool):
|
||||
"""Test finish_reason='error' returns 500 InternalServerError (streaming)."""
|
||||
mock_engine = MagicMock(spec=AsyncLLM)
|
||||
mock_engine.errored = False
|
||||
@@ -684,6 +689,67 @@ async def test_completion_error_stream():
|
||||
prompt="Test prompt",
|
||||
max_tokens=10,
|
||||
stream=True,
|
||||
stream_options={"include_usage": include_usage},
|
||||
)
|
||||
|
||||
response = await serving_completion.create_completion(request)
|
||||
|
||||
chunks = []
|
||||
async for chunk in response:
|
||||
chunks.append(chunk)
|
||||
|
||||
assert len(chunks) >= 2
|
||||
assert any("Internal server error" in chunk for chunk in chunks), (
|
||||
f"Expected error message in chunks: {chunks}"
|
||||
)
|
||||
assert chunks[-1] == "data: [DONE]\n\n"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("include_usage", [False, True])
|
||||
async def test_completion_error_stream_first_chunk(include_usage: bool):
|
||||
"""An error before the first token must reach the streaming client."""
|
||||
mock_engine = MagicMock(spec=AsyncLLM)
|
||||
mock_engine.errored = False
|
||||
mock_engine.model_config = MockModelConfig()
|
||||
mock_engine.input_processor = MagicMock()
|
||||
mock_engine.renderer = _build_renderer(mock_engine.model_config)
|
||||
|
||||
serving_completion = _build_serving_completion(mock_engine)
|
||||
|
||||
completion_output = CompletionOutput(
|
||||
index=0,
|
||||
text="",
|
||||
token_ids=[],
|
||||
cumulative_logprob=None,
|
||||
logprobs=None,
|
||||
finish_reason="error",
|
||||
)
|
||||
|
||||
request_output = RequestOutput(
|
||||
request_id="test-id",
|
||||
prompt="Test prompt",
|
||||
prompt_token_ids=[1, 2, 3],
|
||||
prompt_logprobs=None,
|
||||
outputs=[completion_output],
|
||||
finished=True,
|
||||
metrics=None,
|
||||
lora_request=None,
|
||||
encoder_prompt=None,
|
||||
encoder_prompt_token_ids=None,
|
||||
)
|
||||
|
||||
async def mock_generate(*args, **kwargs):
|
||||
yield request_output
|
||||
|
||||
mock_engine.generate = MagicMock(side_effect=mock_generate)
|
||||
|
||||
request = CompletionRequest(
|
||||
model=MODEL_NAME,
|
||||
prompt="Test prompt",
|
||||
max_tokens=10,
|
||||
stream=True,
|
||||
stream_options={"include_usage": include_usage},
|
||||
)
|
||||
|
||||
response = await serving_completion.create_completion(request)
|
||||
|
||||
@@ -631,6 +631,8 @@ class OpenAIServingChat(GenerateBaseServing):
|
||||
if finish_reason_sent[i]:
|
||||
continue
|
||||
|
||||
self._raise_if_error(output.finish_reason, request_id)
|
||||
|
||||
if request.logprobs and (
|
||||
request.top_logprobs is not None or request.logprob_token_ids
|
||||
):
|
||||
@@ -758,10 +760,6 @@ class OpenAIServingChat(GenerateBaseServing):
|
||||
|
||||
# if the model is finished generating
|
||||
else:
|
||||
# check for error finish reason and abort streaming
|
||||
# finish_reason='error' indicates a retryable error
|
||||
self._raise_if_error(output.finish_reason, request_id)
|
||||
|
||||
# Send the finish response for each request.n only once
|
||||
# In OpenAI's API, when a tool is called, the
|
||||
# finish_reason is:
|
||||
|
||||
@@ -326,6 +326,8 @@ class OpenAIServingCompletion(GenerateBaseServing):
|
||||
|
||||
for output in res.outputs:
|
||||
i = output.index + prompt_idx * num_choices
|
||||
finish_reason = output.finish_reason
|
||||
self._raise_if_error(finish_reason, request_id)
|
||||
|
||||
# Useful when request.return_token_ids is True
|
||||
# Returning prompt token IDs shares the same logic
|
||||
@@ -392,11 +394,8 @@ class OpenAIServingCompletion(GenerateBaseServing):
|
||||
|
||||
previous_text_lens[i] += len(delta_text)
|
||||
previous_num_tokens[i] += len(output.token_ids)
|
||||
finish_reason = output.finish_reason
|
||||
stop_reason = output.stop_reason
|
||||
|
||||
self._raise_if_error(finish_reason, request_id)
|
||||
|
||||
chunk = CompletionStreamResponse(
|
||||
id=request_id,
|
||||
object="text_completion",
|
||||
|
||||
Reference in New Issue
Block a user