Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
81 changes: 79 additions & 2 deletions tests/entrypoints/openai/completion/test_completion_error.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
from vllm.entrypoints.openai.models.serving import OpenAIServingModels
from vllm.entrypoints.scale_out.render.serving import ServingRender
from vllm.entrypoints.serve.engine.protocol import ErrorResponse
from vllm.entrypoints.serve.utils.request_logger import RequestLogger
from vllm.exceptions import GenerationError, VLLMValidationError
from vllm.logprobs import Logprob
from vllm.outputs import CompletionOutput, RequestOutput, WeightVersionSpan
Expand Down Expand Up @@ -91,7 +92,11 @@ class MockVllmConfig:
parallel_config: MockParallelConfig


def _build_serving_completion(engine: AsyncLLM) -> OpenAIServingCompletion:
def _build_serving_completion(
engine: AsyncLLM,
request_logger: RequestLogger | None = None,
**kwargs: Any,
) -> OpenAIServingCompletion:
models = OpenAIServingModels(
engine_client=engine,
base_model_paths=BASE_MODEL_PATHS,
Expand All @@ -107,7 +112,8 @@ def _build_serving_completion(engine: AsyncLLM) -> OpenAIServingCompletion:
engine,
models,
online_renderer=online_renderer,
request_logger=None,
request_logger=request_logger,
**kwargs,
)


Expand Down Expand Up @@ -694,6 +700,77 @@ async def test_renderer_only_completion_request_skips_mm_cache():
)


@pytest.mark.asyncio
@pytest.mark.parametrize("stream", [False, True])
async def test_completion_logs_outputs_when_enabled(stream: bool):
"""`--enable-log-outputs` covers /v1/completions like the chat endpoint."""
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)
request_logger = MagicMock(spec=RequestLogger)
serving_completion = _build_serving_completion(
mock_engine, request_logger=request_logger, enable_log_outputs=True
)

def request_output(
request_id: str, text: str, token_ids: list[int], finish_reason: str | None
):
return RequestOutput(
request_id=request_id,
prompt="Test prompt",
prompt_token_ids=[1, 2, 3],
prompt_logprobs=None,
outputs=[
CompletionOutput(
index=0,
text=text,
token_ids=token_ids,
cumulative_logprob=None,
logprobs=None,
finish_reason=finish_reason,
)
],
finished=finish_reason is not None,
)

async def mock_generate(prompt, params, request_id, **kwargs):
if stream:
yield request_output(request_id, "Hello", [100], None)
yield request_output(request_id, " world", [101], "stop")
else:
yield request_output(request_id, "Hello world", [100, 101], "stop")

mock_engine.generate = MagicMock(side_effect=mock_generate)
request = CompletionRequest(
model=MODEL_NAME,
prompt="Test prompt",
max_tokens=10,
stream=stream,
request_id="test-req",
)

response = await serving_completion.create_completion(request)
if stream:
async for _ in response:
pass

logged = [call.kwargs for call in request_logger.log_outputs.call_args_list]
assert {log["request_id"] for log in logged} == {"cmpl-test-req"}
if stream:
assert [(log["outputs"], log["delta"]) for log in logged] == [
("Hello", True),
(" world", True),
("Hello world", False),
]
else:
assert len(logged) == 1
assert logged[0]["outputs"] == "Hello world"
assert logged[0]["output_token_ids"] == [100, 101]
assert logged[0]["finish_reason"] == "stop"


@pytest.mark.asyncio
@pytest.mark.parametrize("include_usage", [False, True])
async def test_completion_error_stream(include_usage: bool):
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ def _serving() -> OpenAIServingCompletion:
serving = OpenAIServingCompletion.__new__(OpenAIServingCompletion)
serving.enable_prompt_tokens_details = False
serving.enable_force_include_usage = False
serving.enable_log_outputs = False
serving.enable_per_request_metrics = False
serving.system_fingerprint = None
serving.return_tokens_as_token_ids = False
Expand Down
2 changes: 2 additions & 0 deletions vllm/entrypoints/generate/api_router.py
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,8 @@ async def init_generate_state(
return_tokens_as_token_ids=args.return_tokens_as_token_ids,
enable_prompt_tokens_details=args.enable_prompt_tokens_details,
enable_force_include_usage=args.enable_force_include_usage,
enable_log_outputs=args.enable_log_outputs,
enable_log_deltas=args.enable_log_deltas,
enable_per_request_metrics=args.enable_per_request_metrics,
)
if "generate" in supported_tasks
Expand Down
41 changes: 41 additions & 0 deletions vllm/entrypoints/openai/completion/serving.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,8 @@ def __init__(
return_tokens_as_token_ids: bool = False,
enable_prompt_tokens_details: bool = False,
enable_force_include_usage: bool = False,
enable_log_outputs: bool = False,
enable_log_deltas: bool = True,
enable_per_request_metrics: bool = False,
):
super().__init__(
Expand All @@ -77,6 +79,8 @@ def __init__(
self.online_renderer = online_renderer
self.enable_prompt_tokens_details = enable_prompt_tokens_details
self.enable_force_include_usage = enable_force_include_usage
self.enable_log_outputs = enable_log_outputs
self.enable_log_deltas = enable_log_deltas
self.enable_per_request_metrics = enable_per_request_metrics

self.default_sampling_params = self.model_config.get_diff_sampling_param()
Expand Down Expand Up @@ -262,6 +266,18 @@ async def _create_completion(
tokenizer,
request_metadata,
)

if self.enable_log_outputs and self.request_logger:
for final_res in final_res_batch_checked:
for output in final_res.outputs:
self.request_logger.log_outputs(
request_id=request_id,
outputs=output.text,
output_token_ids=output.token_ids,
finish_reason=output.finish_reason,
is_streaming=False,
delta=False,
)
except asyncio.CancelledError:
return self.create_error_response("Client disconnected")

Expand Down Expand Up @@ -293,6 +309,7 @@ async def completion_stream_generator(
num_choices = 1 if request.n is None else request.n
next_text_offsets = [0] * num_choices * num_prompts
previous_num_tokens = [0] * num_choices * num_prompts
previous_texts = [""] * num_choices * num_prompts
has_echoed = [False] * num_choices * num_prompts
num_prompt_tokens = [0] * num_prompts
num_cached_tokens = None
Expand Down Expand Up @@ -404,6 +421,19 @@ async def completion_stream_generator(
previous_num_tokens[i] += len(output.token_ids)
stop_reason = output.stop_reason

if self.enable_log_outputs and self.request_logger:
delta_content = output.text
previous_texts[i] += delta_content
if delta_content and self.enable_log_deltas:
self.request_logger.log_outputs(
request_id=request_id,
outputs=delta_content,
output_token_ids=as_list(output.token_ids),
finish_reason=finish_reason,
is_streaming=True,
delta=True,
)

chunk = CompletionStreamResponse(
id=request_id,
object="text_completion",
Expand Down Expand Up @@ -450,6 +480,17 @@ async def completion_stream_generator(
response_json = chunk.model_dump_json(exclude_unset=True)
yield f"data: {response_json}\n\n"

if self.enable_log_outputs and self.request_logger:
for full_text in previous_texts:
self.request_logger.log_outputs(
request_id=request_id,
outputs=full_text,
output_token_ids=None,
finish_reason="streaming_complete",
is_streaming=True,
delta=False,
)

total_prompt_tokens = sum(num_prompt_tokens)
total_completion_tokens = sum(previous_num_tokens)
final_usage_info = UsageInfo(
Expand Down
Loading