diff --git a/rust/src/chat/src/event.rs b/rust/src/chat/src/event.rs index 5fe25a9969..4056476b48 100644 --- a/rust/src/chat/src/event.rs +++ b/rust/src/chat/src/event.rs @@ -5,7 +5,7 @@ use std::ops::Deref; use std::sync::Arc; use serde::{Deserialize, Serialize}; -use vllm_llm::TokenUsage; +use vllm_llm::{RequestTimestamps, TokenUsage}; use vllm_text::{DecodedLogprobs, DecodedPromptLogprobs}; use crate::FinishReason; @@ -235,6 +235,8 @@ pub enum ChatEvent { Done { message: AssistantMessage, usage: ChatTokenUsage, + /// Engine lifecycle times used by frontend response metrics. + timestamps: RequestTimestamps, finish_reason: FinishReason, /// Connector-specific KV transfer parameters for disaggregated serving. /// Boxed, like `ec_transfer_params`, to keep the per-token variants of diff --git a/rust/src/chat/src/output/default/unified.rs b/rust/src/chat/src/output/default/unified.rs index 9f872d9794..a6238c440e 100644 --- a/rust/src/chat/src/output/default/unified.rs +++ b/rust/src/chat/src/output/default/unified.rs @@ -277,6 +277,7 @@ pub(crate) async fn unified_event_stream( engine: finished.usage, reasoning_tokens: state.reasoning_tokens, }, + timestamps: finished.timestamps, finish_reason: finished.finish_reason, kv_transfer_params: finished.kv_transfer_params, ec_transfer_params: finished.ec_transfer_params, @@ -419,6 +420,7 @@ mod tests { kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, + timestamps: Default::default(), })), } } @@ -742,6 +744,7 @@ mod tests { }, AssistantEvent::Done { usage: done_usage(0), + timestamps: Default::default(), finish_reason: crate::FinishReason::Stop(None), kv_transfer_params: None, ec_transfer_params: None, @@ -782,6 +785,7 @@ mod tests { }, AssistantEvent::Done { usage: done_usage(5), + timestamps: Default::default(), finish_reason: crate::FinishReason::Stop(None), kv_transfer_params: None, ec_transfer_params: None, @@ -814,6 +818,7 @@ mod tests { }, AssistantEvent::Done { usage: done_usage(2), + timestamps: Default::default(), finish_reason: crate::FinishReason::Stop(None), kv_transfer_params: None, ec_transfer_params: None, @@ -866,6 +871,7 @@ mod tests { }, AssistantEvent::Done { usage: done_usage(5), + timestamps: Default::default(), finish_reason: crate::FinishReason::Stop(None), kv_transfer_params: None, ec_transfer_params: None, @@ -912,6 +918,7 @@ mod tests { }, AssistantEvent::Done { usage: done_usage(0), + timestamps: Default::default(), finish_reason: crate::FinishReason::Stop(None), kv_transfer_params: None, ec_transfer_params: None, diff --git a/rust/src/chat/src/output/harmony/mod.rs b/rust/src/chat/src/output/harmony/mod.rs index 01e0390622..8ec873c6f2 100644 --- a/rust/src/chat/src/output/harmony/mod.rs +++ b/rust/src/chat/src/output/harmony/mod.rs @@ -380,6 +380,7 @@ async fn harmony_assistant_event_stream( if let Some(finished) = finished { y.yield_ok(AssistantEvent::Done { usage: finished.usage.into(), + timestamps: finished.timestamps, finish_reason: finished.finish_reason, kv_transfer_params: finished.kv_transfer_params, ec_transfer_params: finished.ec_transfer_params, diff --git a/rust/src/chat/src/output/harmony/tests.rs b/rust/src/chat/src/output/harmony/tests.rs index 5b89e53bb9..a529888ccf 100644 --- a/rust/src/chat/src/output/harmony/tests.rs +++ b/rust/src/chat/src/output/harmony/tests.rs @@ -59,6 +59,7 @@ fn finished() -> Finished { kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, + timestamps: Default::default(), } } @@ -132,6 +133,7 @@ fn interrupted_final_message_is_preserved() { output_token_count: 0, cached_token_count: 0, }), + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, @@ -183,6 +185,7 @@ fn interrupted_analysis_message_is_preserved() { output_token_count: 0, cached_token_count: 0, }), + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, diff --git a/rust/src/chat/src/output/mod.rs b/rust/src/chat/src/output/mod.rs index 8c1dae37f4..14ac725206 100644 --- a/rust/src/chat/src/output/mod.rs +++ b/rust/src/chat/src/output/mod.rs @@ -7,6 +7,7 @@ use std::sync::Arc; use futures::Stream; use trait_set::trait_set; use uuid::Uuid; +use vllm_llm::RequestTimestamps; use vllm_parser::output_grammar::BuiltOutputGrammar; use vllm_text::output::{DecodedLogprobs, DecodedPromptLogprobs, DecodedTextEvent}; @@ -52,6 +53,7 @@ pub(crate) enum AssistantEvent { ToolCallArgumentsDelta { delta: String }, Done { usage: ChatTokenUsage, + timestamps: RequestTimestamps, finish_reason: FinishReason, /// Connector-specific KV transfer parameters for disaggregated serving. /// Boxed, like `ec_transfer_params`, to keep the per-token variants of diff --git a/rust/src/chat/src/output/structured.rs b/rust/src/chat/src/output/structured.rs index 80035db4cb..3ed1a4e852 100644 --- a/rust/src/chat/src/output/structured.rs +++ b/rust/src/chat/src/output/structured.rs @@ -10,6 +10,7 @@ use asynk_strim_attr::{TryYielder, try_stream}; use futures::{StreamExt as _, pin_mut}; +use vllm_llm::RequestTimestamps; use vllm_text::DecodedLogprobs; use super::{AssistantEvent, AssistantEventStream}; @@ -149,6 +150,7 @@ impl StructuredEventState { fn finish( &mut self, usage: ChatTokenUsage, + timestamps: RequestTimestamps, finish_reason: FinishReason, kv_transfer_params: Option>, ec_transfer_params: Option>, @@ -159,6 +161,7 @@ impl StructuredEventState { events.push(ChatEvent::Done { message: self.message.clone(), usage, + timestamps, finish_reason, kv_transfer_params, ec_transfer_params, @@ -312,13 +315,18 @@ pub(crate) async fn structured_chat_event_stream( } AssistantEvent::Done { usage, + timestamps, finish_reason, kv_transfer_params, ec_transfer_params, } => { - for next in - state.finish(usage, finish_reason, kv_transfer_params, ec_transfer_params)? - { + for next in state.finish( + usage, + timestamps, + finish_reason, + kv_transfer_params, + ec_transfer_params, + )? { y.yield_ok(next).await; } } @@ -353,6 +361,7 @@ mod tests { output_token_count: 1, cached_token_count: 0, }), + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, @@ -408,6 +417,7 @@ mod tests { output_token_count: 1, cached_token_count: 0, }), + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, @@ -461,6 +471,7 @@ mod tests { output_token_count: 1, cached_token_count: 0, }), + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, @@ -514,6 +525,7 @@ mod tests { output_token_count: 1, cached_token_count: 0, }), + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, @@ -582,6 +594,7 @@ mod tests { output_token_count: 1, cached_token_count: 0, }), + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, @@ -633,6 +646,7 @@ mod tests { }, reasoning_tokens: 3, }, + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, diff --git a/rust/src/chat/src/stream.rs b/rust/src/chat/src/stream.rs index 509628e8e2..fd58249461 100644 --- a/rust/src/chat/src/stream.rs +++ b/rust/src/chat/src/stream.rs @@ -7,6 +7,7 @@ use std::task::{Context, Poll}; use futures::Stream; use trait_set::trait_set; +use vllm_llm::RequestTimestamps; use vllm_text::{DecodedLogprobs, DecodedPositionLogprobs, DecodedPromptLogprobs}; use crate::FinishReason; @@ -22,6 +23,7 @@ pub struct CollectedAssistantMessage { pub logprobs: Option, pub token_ids: Vec, pub usage: ChatTokenUsage, + pub timestamps: RequestTimestamps, pub finish_reason: FinishReason, /// Connector-specific KV transfer parameters for disaggregated serving. pub kv_transfer_params: Option, @@ -81,6 +83,7 @@ impl ChatEventStream { ChatEvent::Done { message: done, usage, + timestamps, finish_reason, kv_transfer_params, ec_transfer_params, @@ -94,6 +97,7 @@ impl ChatEventStream { }), token_ids, usage, + timestamps, finish_reason, kv_transfer_params: kv_transfer_params.map(|value| *value), ec_transfer_params: ec_transfer_params.map(|value| *value), @@ -200,6 +204,7 @@ mod tests { output_token_count: 1, cached_token_count: 0, }), + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, @@ -241,6 +246,7 @@ mod tests { output_token_count: 1, cached_token_count: 0, }), + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, diff --git a/rust/src/chat/tests/roundtrip.rs b/rust/src/chat/tests/roundtrip.rs index 41f06d7734..d2891ed027 100644 --- a/rust/src/chat/tests/roundtrip.rs +++ b/rust/src/chat/tests/roundtrip.rs @@ -1051,6 +1051,7 @@ fn decoded_completion_stream( kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, + timestamps: Default::default(), })), } }); @@ -1063,6 +1064,7 @@ fn decoded_completion_stream( kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, + timestamps: Default::default(), }); events.push(DecodedTextEvent::TextDelta { decoded: chunk.delta, diff --git a/rust/src/cmd/src/cli.rs b/rust/src/cmd/src/cli.rs index eeb458c86e..66b48f13d3 100644 --- a/rust/src/cmd/src/cli.rs +++ b/rust/src/cmd/src/cli.rs @@ -348,6 +348,11 @@ pub struct SharedRuntimeArgs { #[serde(default)] pub enable_prompt_tokens_details: bool, + /// Include per-request timing metrics in API responses. + #[arg(long, default_missing_value = "true", num_args = 0..=1)] + #[serde(default)] + pub enable_per_request_metrics: bool, + /// If specified, API server will add X-Request-Id header to responses. #[arg( long, @@ -635,6 +640,7 @@ impl SharedRuntimeArgs { ApiServerOptions { enable_log_requests: self.enable_log_requests, enable_prompt_tokens_details: self.enable_prompt_tokens_details, + enable_per_request_metrics: self.enable_per_request_metrics, enable_request_id_headers: self.enable_request_id_headers, enable_scale_out: self.enable_scale_out, sse_keep_alive_interval: (self.sse_keep_alive_interval > 0) diff --git a/rust/src/cmd/src/cli/tests.rs b/rust/src/cmd/src/cli/tests.rs index cc545ce33e..8f4318ac16 100644 --- a/rust/src/cmd/src/cli/tests.rs +++ b/rust/src/cmd/src/cli/tests.rs @@ -199,6 +199,7 @@ fn serve_args_forward_python_flags_with_separator() { chat_template_content_format: Auto, enable_log_requests: false, enable_prompt_tokens_details: false, + enable_per_request_metrics: false, enable_request_id_headers: false, enable_scale_out: false, sse_keep_alive_interval: 0, @@ -970,6 +971,40 @@ fn serve_args_reject_unsupported_flag_arg() { "#]].assert_eq(&error.to_string()); } +#[test] +fn serve_args_enable_per_request_metrics_requires_engine_stats() { + let cli = Cli::try_parse_from([ + "vllm-rs", + "serve", + "Qwen/Qwen3-0.6B", + "--enable-per-request-metrics", + ]) + .expect("per-request metrics flag is supported"); + let Command::Serve(args) = cli.command else { + panic!("expected serve args"); + }; + let config = args.to_frontend_config("tcp://127.0.0.1:0".into()); + assert!(config.api_server_options.enable_per_request_metrics); + assert!(config.validate().is_ok()); + + let cli = Cli::try_parse_from([ + "vllm-rs", + "serve", + "Qwen/Qwen3-0.6B", + "--enable-per-request-metrics", + "--disable-log-stats", + ]) + .expect("flags parse independently"); + let Command::Serve(args) = cli.command else { + panic!("expected serve args"); + }; + let error = args.to_frontend_config("tcp://127.0.0.1:0".into()).validate().unwrap_err(); + assert!( + format!("{error:#}") + .contains("--enable-per-request-metrics requires engine statistics logging") + ); +} + #[test] fn serve_args_reject_custom_generation_config_source() { let error = Cli::try_parse_from([ @@ -1067,6 +1102,7 @@ fn frontend_args_accept_json() { chat_template_content_format: Auto, enable_log_requests: false, enable_prompt_tokens_details: false, + enable_per_request_metrics: false, enable_request_id_headers: false, enable_scale_out: false, sse_keep_alive_interval: 0, @@ -1738,6 +1774,7 @@ fn serve_args_accept_handshake_aliases() { chat_template_content_format: Auto, enable_log_requests: false, enable_prompt_tokens_details: false, + enable_per_request_metrics: false, enable_request_id_headers: false, enable_scale_out: false, sse_keep_alive_interval: 0, @@ -1895,6 +1932,7 @@ fn serve_frontend_config_uses_dp_address_as_advertised_host() { api_server_options: ApiServerOptions { enable_log_requests: false, enable_prompt_tokens_details: false, + enable_per_request_metrics: false, enable_request_id_headers: false, enable_scale_out: false, sse_keep_alive_interval: None, @@ -1989,6 +2027,7 @@ fn serve_frontend_config_keeps_tcp_transport_for_non_local_only_topology() { api_server_options: ApiServerOptions { enable_log_requests: false, enable_prompt_tokens_details: false, + enable_per_request_metrics: false, enable_request_id_headers: false, enable_scale_out: false, sse_keep_alive_interval: None, @@ -2127,6 +2166,7 @@ fn frontend_config_uses_external_coordinator_when_coordinator_address_is_present api_server_options: ApiServerOptions { enable_log_requests: false, enable_prompt_tokens_details: false, + enable_per_request_metrics: false, enable_request_id_headers: false, enable_scale_out: false, sse_keep_alive_interval: None, diff --git a/rust/src/cmd/src/cli/unsupported.rs b/rust/src/cmd/src/cli/unsupported.rs index 0a14d1b841..02e1377ff0 100644 --- a/rust/src/cmd/src/cli/unsupported.rs +++ b/rust/src/cmd/src/cli/unsupported.rs @@ -476,15 +476,6 @@ pub struct ServerUnsupportedArgs { #[arg(long)] pub max_log_len: Option, - /// If set to True, include per-request timing metrics in API responses. - #[arg( - long, - visible_alias = "no-enable-per-request-metrics", - default_missing_value = "true", - num_args = 0..=1 - )] - pub enable_per_request_metrics: Option, - /// If set to True, enable tracking server_load_metrics in the app state. #[arg( long, diff --git a/rust/src/llm/src/lib.rs b/rust/src/llm/src/lib.rs index c94c1d80f7..1e15f11c12 100644 --- a/rust/src/llm/src/lib.rs +++ b/rust/src/llm/src/lib.rs @@ -19,7 +19,7 @@ pub use output::{ GenerateOutputStreamExt, GeneratePromptInfo, TokenUsage, WeightVersionSpan, }; pub use request::GenerateRequest; -pub use request_metrics::current_unix_timestamp_secs; +pub use request_metrics::{RequestTimestamps, current_unix_timestamp_secs}; pub use vllm_engine_core_client::protocol::logprobs::{Logprobs, PositionLogprobs, TokenLogprob}; use crate::inflight::InflightRequests; diff --git a/rust/src/llm/src/output.rs b/rust/src/llm/src/output.rs index 81dbb3fda5..3f0b030062 100644 --- a/rust/src/llm/src/output.rs +++ b/rust/src/llm/src/output.rs @@ -20,6 +20,7 @@ use vllm_engine_core_client::{ AbortCause, EngineCoreOutputStream, EngineCoreStreamDelivery, EngineCoreStreamOutput, }; +use crate::RequestTimestamps; use crate::error::Result; use crate::inflight::RequestGuard; use crate::request_metrics::{RequestMetricsTracker, current_unix_timestamp_secs}; @@ -183,6 +184,12 @@ pub struct GenerateOutput { pub spec_decode_metrics: Option>, /// Weight-version spans over all output tokens, set on terminal outputs only. pub weight_versions: Option>, + /// Snapshot of engine lifecycle timestamps after this output. + /// + /// Every output carries a snapshot because text decoding may finish the + /// request locally after matching a stop string, before the engine emits a + /// terminal output. + pub timestamps: RequestTimestamps, } impl GenerateOutput { @@ -237,6 +244,7 @@ impl GenerateOutput { self.ec_transfer_params = next.ec_transfer_params; self.spec_decode_metrics = next.spec_decode_metrics; self.weight_versions = next.weight_versions; + self.timestamps = next.timestamps; } } @@ -266,6 +274,7 @@ impl GenerateOutput { sampling_mask: None, spec_decode_metrics: None, weight_versions: None, + timestamps: RequestTimestamps::default(), } } } @@ -402,6 +411,7 @@ impl GenerateOutputStream { sampling_mask, spec_decode_metrics: raw.spec_decode_metrics, weight_versions, + timestamps: self.request_metrics.timestamps(), }; Ok(output) diff --git a/rust/src/llm/src/request_metrics.rs b/rust/src/llm/src/request_metrics.rs index 71decdd281..eebf0596b0 100644 --- a/rust/src/llm/src/request_metrics.rs +++ b/rust/src/llm/src/request_metrics.rs @@ -18,6 +18,15 @@ const PROMPT_TOKEN_SOURCE_LOCAL_COMPUTE: &str = "local_compute"; const PROMPT_TOKEN_SOURCE_LOCAL_CACHE_HIT: &str = "local_cache_hit"; const PROMPT_TOKEN_SOURCE_EXTERNAL_KV_TRANSFER: &str = "external_kv_transfer"; +/// Engine lifecycle timestamps used by frontend per-request response metrics. +#[derive(Debug, Clone, Copy, Default, PartialEq)] +pub struct RequestTimestamps { + pub queued_ts: f64, + pub scheduled_ts: f64, + pub first_token_ts: f64, + pub last_token_ts: f64, +} + /// Request-scoped metrics state tracked across streamed engine-core updates. /// /// This is the Rust-side counterpart of the Python frontend's request-lifecycle @@ -227,6 +236,16 @@ impl RequestMetricsTracker { } } + /// Engine lifecycle timestamps recorded so far. + pub(crate) fn timestamps(&self) -> RequestTimestamps { + RequestTimestamps { + queued_ts: self.queued_ts, + scheduled_ts: self.scheduled_ts, + first_token_ts: self.first_token_ts, + last_token_ts: self.last_token_ts, + } + } + /// Record prompt token counters through cached metric handles. fn record_prompt_tokens(&self, prefill_stats: &PrefillStats) { let computed = prefill_stats.num_computed_tokens as u64; @@ -388,7 +407,7 @@ mod tests { use vllm_engine_core_client::protocol::output::{EngineCoreEvent, EngineCoreEventType}; use vllm_engine_core_client::protocol::stats::PrefillStats; - use super::{RequestMetricsTracker, diff_or_zero}; + use super::{RequestMetricsTracker, RequestTimestamps, diff_or_zero}; #[test] fn tracker_updates_timing_state_across_prefill_decode_and_finish() { @@ -445,6 +464,15 @@ mod tests { assert_eq!(tracker.scheduled_ts, 9.0); assert_eq!(tracker.first_token_ts, 10.0); assert_eq!(tracker.last_token_ts, 11.5); + assert_eq!( + tracker.timestamps(), + RequestTimestamps { + queued_ts: 8.0, + scheduled_ts: 9.0, + first_token_ts: 10.0, + last_token_ts: 11.5, + } + ); assert!((tracker.first_token_latency - 0.2).abs() < 1e-9); assert_eq!( diff_or_zero(tracker.last_token_ts, tracker.first_token_ts), diff --git a/rust/src/llm/tests/generate.rs b/rust/src/llm/tests/generate.rs index 0a1eacd73f..cc6fd9d169 100644 --- a/rust/src/llm/tests/generate.rs +++ b/rust/src/llm/tests/generate.rs @@ -352,7 +352,9 @@ async fn generate_merges_outputs_by_stream_interval() { (&[6], "v7", None), (&[7], "v7", Some(EngineCoreFinishReason::Length)), ]; - for (token_ids, weight_version, finish_reason) in steps { + for (step, (token_ids, weight_version, finish_reason)) in + steps.into_iter().enumerate() + { send_outputs( push, RequestBatchOutputs { @@ -364,6 +366,7 @@ async fn generate_merges_outputs_by_stream_interval() { None, )], weight_version: Some(weight_version.into()), + timestamp: step as f64 + 1.0, ..Default::default() } .into(), @@ -430,6 +433,9 @@ async fn generate_merges_outputs_by_stream_interval() { Some(vec![span("v7", 0, 2), span("v8", 2, 5), span("v7", 5, 7)]) ] ); + let last_token_timestamps: Vec<_> = + outputs.iter().map(|output| output.timestamps.last_token_ts).collect(); + assert_eq!(last_token_timestamps, [1.0, 4.0, 6.0]); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -549,6 +555,7 @@ async fn collect_output_rejects_partial_sampling_mask() { }), spec_decode_metrics: None, weight_versions: None, + timestamps: Default::default(), }; let error = futures::stream::iter([Ok(output)]).collect_output().await.unwrap_err(); diff --git a/rust/src/server/src/config.rs b/rust/src/server/src/config.rs index ca91ba40cd..d70933fb44 100644 --- a/rust/src/server/src/config.rs +++ b/rust/src/server/src/config.rs @@ -56,6 +56,8 @@ pub struct ApiServerOptions { pub enable_log_requests: bool, /// When `true`, include prompt token cache details in response usage. pub enable_prompt_tokens_details: bool, + /// Include request timing metrics in OpenAI chat and completion responses. + pub enable_per_request_metrics: bool, /// When `true`, set `X-Request-Id` on every HTTP response. pub enable_request_id_headers: bool, /// When `true`, register the scale-out `/inference/v1/generate` route. @@ -296,6 +298,11 @@ impl Config { /// Validate frontend configuration that can be checked before engine /// startup. pub fn validate(&self) -> Result<()> { + if self.api_server_options.enable_per_request_metrics && self.disable_log_stats { + bail!( + "--enable-per-request-metrics requires engine statistics logging; remove --disable-log-stats" + ); + } vllm_chat::validate_parser_overrides(&self.tool_call_parser, &self.reasoning_parser)?; self.cors.validate()?; if let Some(tls) = &self.tls { diff --git a/rust/src/server/src/grpc/convert.rs b/rust/src/server/src/grpc/convert.rs index 8e31bb84d2..1eec1d0fe3 100644 --- a/rust/src/server/src/grpc/convert.rs +++ b/rust/src/server/src/grpc/convert.rs @@ -947,6 +947,7 @@ mod tests { kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, + timestamps: Default::default(), } } diff --git a/rust/src/server/src/grpc/inference.rs b/rust/src/server/src/grpc/inference.rs index 1484a9f9e3..fae2794ba3 100644 --- a/rust/src/server/src/grpc/inference.rs +++ b/rust/src/server/src/grpc/inference.rs @@ -259,6 +259,7 @@ impl pb::inference_server::Inference for InferenceServiceImpl { kv_transfer_params: collected.kv_transfer_params.map(Box::new), ec_transfer_params: collected.ec_transfer_params.map(Box::new), sampling_mask: collected.sampling_mask, + timestamps: collected.timestamps, }; let outputs = convert::to_sequence_output( diff --git a/rust/src/server/src/routes/inference/generate.rs b/rust/src/server/src/routes/inference/generate.rs index 930c698c30..42716de8de 100644 --- a/rust/src/server/src/routes/inference/generate.rs +++ b/rust/src/server/src/routes/inference/generate.rs @@ -637,6 +637,7 @@ mod tests { sampling_mask: None, spec_decode_metrics: None, weight_versions: None, + timestamps: Default::default(), }), Ok(GenerateOutput { request_id: String::new(), @@ -654,6 +655,7 @@ mod tests { sampling_mask: None, spec_decode_metrics: None, weight_versions: None, + timestamps: Default::default(), }), ]); @@ -741,6 +743,7 @@ mod tests { sampling_mask: None, spec_decode_metrics: None, weight_versions: None, + timestamps: Default::default(), } } diff --git a/rust/src/server/src/routes/openai/chat_completions.rs b/rust/src/server/src/routes/openai/chat_completions.rs index a87c52fb55..5d8dcda536 100644 --- a/rust/src/server/src/routes/openai/chat_completions.rs +++ b/rust/src/server/src/routes/openai/chat_completions.rs @@ -38,6 +38,7 @@ use crate::routes::openai::chat_completions::types::{ use crate::routes::openai::utils::logprobs::{ decoded_logprobs_to_openai_chat, prompt_logprobs_to_maps, }; +use crate::routes::openai::utils::metrics::PerRequestMetrics; use crate::routes::openai::utils::types::{ ChatLogProbs, FunctionCallDelta, FunctionCallResponse, StreamResponseEnvelope, ToolCall, ToolCallDelta, Usage, @@ -128,6 +129,7 @@ async fn collect_chat_completion( ApiServerOptions { enable_log_requests, enable_prompt_tokens_details, + enable_per_request_metrics, .. }: ApiServerOptions, ResponseOptions { @@ -158,6 +160,7 @@ async fn collect_chat_completion( logprobs, token_ids, usage, + timestamps, finish_reason, kv_transfer_params, ec_transfer_params, @@ -202,6 +205,8 @@ async fn collect_chat_completion( } else { None }; + let metrics = enable_per_request_metrics + .then(|| PerRequestMetrics::from_timestamps(timestamps, usage.output_token_count)); let usage = Usage::from_token_usage(usage, enable_prompt_tokens_details); if enable_log_requests { @@ -236,6 +241,7 @@ async fn collect_chat_completion( token_ids: (return_token_ids && include_output_metadata).then_some(token_ids), }], usage: Some(usage), + metrics, system_fingerprint: None, prompt_logprobs, prompt_token_ids: return_token_ids.then(|| prompt_token_ids.to_vec()), @@ -254,6 +260,7 @@ async fn chat_completion_chunk_stream( ApiServerOptions { enable_log_requests, enable_prompt_tokens_details, + enable_per_request_metrics, .. }: ApiServerOptions, ResponseOptions { @@ -428,6 +435,7 @@ async fn chat_completion_chunk_stream( } Ok(ChatEvent::Done { usage: final_usage, + timestamps, finish_reason, .. }) => { @@ -472,9 +480,16 @@ async fn chat_completion_chunk_stream( } if include_usage { + let metrics = enable_per_request_metrics.then(|| { + PerRequestMetrics::from_timestamps( + timestamps, + final_usage.output_token_count, + ) + }); y.yield_ok(usage_chunk( &envelope, Usage::from_token_usage(final_usage, enable_prompt_tokens_details), + metrics, )) .await; } @@ -496,9 +511,11 @@ async fn chat_completion_chunk_stream( fn usage_chunk( envelope: &Arc, usage: Usage, + metrics: Option, ) -> ChatCompletionStreamResponse { let mut chunk = ChatCompletionStreamResponse::new(envelope); chunk.usage = Some(usage); + chunk.metrics = metrics.map(Into::into); chunk } @@ -962,6 +979,12 @@ mod tests { Ok(ChatEvent::Done { message: Default::default(), usage: done_usage(1, 1, 1), + timestamps: vllm_llm::RequestTimestamps { + queued_ts: 10.0, + scheduled_ts: 10.2, + first_token_ts: 10.5, + last_token_ts: 11.0, + }, finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, @@ -975,6 +998,7 @@ mod tests { 1, ApiServerOptions { enable_prompt_tokens_details: true, + enable_per_request_metrics: true, ..Default::default() }, ResponseOptions { @@ -1006,6 +1030,10 @@ mod tests { .map(|details| details.cached_tokens), Some(1) ); + assert!(chunks[..3].iter().all(|chunk| chunk.metrics.is_none())); + let metrics = chunks[3].metrics.as_ref().expect("per-request metrics"); + assert!((metrics.time_to_first_token_ms.unwrap() - 300.0).abs() < 1e-9); + assert!((metrics.tokens_per_second.unwrap() - 1.25).abs() < 1e-9); } #[tokio::test] @@ -1041,6 +1069,7 @@ mod tests { Ok(ChatEvent::Done { message: Default::default(), usage: done_usage(1, 1, 0), + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, @@ -1095,6 +1124,7 @@ mod tests { Ok(ChatEvent::Done { message: Default::default(), usage: done_usage(1, 2, 0), + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, @@ -1175,6 +1205,7 @@ mod tests { Ok(ChatEvent::Done { message: Default::default(), usage: done_usage(1, 2, 0), + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, @@ -1307,6 +1338,7 @@ mod tests { Ok(ChatEvent::Done { message: Default::default(), usage: done_usage(1, 4, 0), + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, @@ -1385,6 +1417,7 @@ mod tests { Ok(ChatEvent::Done { message: Default::default(), usage: done_usage(1, 1, 0), + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, @@ -1468,6 +1501,7 @@ mod tests { }, reasoning_tokens: 2, }, + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, @@ -1539,6 +1573,7 @@ mod tests { }, reasoning_tokens: 2, }, + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, @@ -1593,6 +1628,7 @@ mod tests { Ok(ChatEvent::Done { message: Default::default(), usage: done_usage(1, 1, 0), + timestamps: Default::default(), finish_reason: FinishReason::stop_eos(), kv_transfer_params: None, ec_transfer_params: None, diff --git a/rust/src/server/src/routes/openai/chat_completions/types.rs b/rust/src/server/src/routes/openai/chat_completions/types.rs index 21776dd075..d73480b45d 100644 --- a/rust/src/server/src/routes/openai/chat_completions/types.rs +++ b/rust/src/server/src/routes/openai/chat_completions/types.rs @@ -13,6 +13,7 @@ use validator::Validate; use vllm_engine_core_client::protocol::sampling::RepetitionDetectionParams; use vllm_text::TruncationSide; +use crate::routes::openai::utils::metrics::{PerRequestMetrics, StreamPerRequestMetrics}; use crate::routes::openai::utils::structured_outputs::ResponseFormat; use crate::routes::openai::utils::types::{ ChatLogProbs, ChatMessage, Normalizable, PromptLogprobs, ReasoningEffort, StreamOptions, @@ -365,6 +366,7 @@ pub(super) struct ChatCompletionResponse { pub model: String, pub choices: Vec, pub usage: Option, + pub metrics: Option, pub system_fingerprint: Option, pub prompt_logprobs: Option, pub prompt_token_ids: Option>, @@ -412,6 +414,7 @@ pub(super) struct ChatCompletionStreamResponse { pub envelope: Arc, pub choices: Vec, pub usage: Option, + pub metrics: Option, pub prompt_token_ids: Option>, } @@ -422,6 +425,7 @@ impl ChatCompletionStreamResponse { envelope: Arc::clone(envelope), choices: Vec::new(), usage: None, + metrics: None, prompt_token_ids: None, } } diff --git a/rust/src/server/src/routes/openai/completions.rs b/rust/src/server/src/routes/openai/completions.rs index 646a67d1fb..05dd579ece 100644 --- a/rust/src/server/src/routes/openai/completions.rs +++ b/rust/src/server/src/routes/openai/completions.rs @@ -33,6 +33,7 @@ use super::utils::logprobs::{ collected_logprobs_to_openai, decoded_logprobs_to_openai, decoded_prompt_logprobs_to_openai, prompt_logprobs_to_maps, text_len, }; +use super::utils::metrics::PerRequestMetrics; use super::utils::types::{StreamResponseEnvelope, Usage}; use crate::config::ApiServerOptions; use crate::error::{ApiError, bail_server_error, server_error, text_submit_error}; @@ -140,6 +141,7 @@ async fn collect_completion( ApiServerOptions { enable_log_requests, enable_prompt_tokens_details, + enable_per_request_metrics, .. }: ApiServerOptions, ResponseOptions { @@ -197,6 +199,9 @@ async fn collect_completion( Some(prompt) => format!("{prompt}{}", collected.text), }; let finish_reason = completion_finish_reason_to_openai(&finish_reason)?.to_string(); + let metrics = enable_per_request_metrics.then(|| { + PerRequestMetrics::from_timestamps(collected.timestamps, collected.usage.output_token_count) + }); let usage = Usage::from_token_usage(collected.usage, enable_prompt_tokens_details); if enable_log_requests { @@ -225,6 +230,7 @@ async fn collect_completion( prompt_token_ids: return_token_ids.then(|| collected.prompt_token_ids.to_vec()), }], usage: Some(usage), + metrics, system_fingerprint: None, kv_transfer_params: collected.kv_transfer_params, ec_transfer_params: collected.ec_transfer_params, @@ -241,6 +247,7 @@ async fn completion_chunk_stream( ApiServerOptions { enable_log_requests, enable_prompt_tokens_details, + enable_per_request_metrics, .. }: ApiServerOptions, ResponseOptions { @@ -356,12 +363,19 @@ async fn completion_chunk_stream( yield_chunk!(final_chunk); if include_usage { + let metrics = enable_per_request_metrics.then(|| { + PerRequestMetrics::from_timestamps( + finished.timestamps, + finished.usage.output_token_count, + ) + }); y.yield_ok(CompletionSseChunk::Usage(usage_chunk( &envelope, Usage::from_token_usage( finished.usage, enable_prompt_tokens_details, ), + metrics, ))) .await; } @@ -411,9 +425,16 @@ async fn completion_chunk_stream( yield_chunk!(final_chunk); if include_usage { + let metrics = enable_per_request_metrics.then(|| { + PerRequestMetrics::from_timestamps( + finished.timestamps, + finished.usage.output_token_count, + ) + }); y.yield_ok(CompletionSseChunk::Usage(usage_chunk( &envelope, Usage::from_token_usage(finished.usage, enable_prompt_tokens_details), + metrics, ))) .await; } @@ -509,9 +530,14 @@ fn prompt_only_logprobs_to_openai( )) } -fn usage_chunk(envelope: &Arc, usage: Usage) -> CompletionStreamResponse { +fn usage_chunk( + envelope: &Arc, + usage: Usage, + metrics: Option, +) -> CompletionStreamResponse { let mut chunk = CompletionStreamResponse::new(envelope); chunk.usage = Some(usage); + chunk.metrics = metrics.map(Into::into); chunk } @@ -578,7 +604,7 @@ mod tests { use super::{ ApiServerOptions, CompletionSseChunk, CompletionStreamResponse, ResponseOptions, - StreamResponseEnvelope, completion_chunk_stream, final_chunk, + StreamResponseEnvelope, collect_completion, completion_chunk_stream, final_chunk, }; fn decoded_delta( @@ -639,6 +665,66 @@ mod tests { assert!(final_chunk(&stream_envelope(), FinishReason::Error).is_err()); } + #[tokio::test] + async fn collected_completion_includes_metrics_only_when_enabled() { + let make_stream = || { + stream::iter(vec![ + Ok(DecodedTextEvent::Start { + prompt_token_ids: vec![1].into(), + prompt_logprobs: None, + }), + Ok(decoded_delta( + "hi", + vec![2], + None, + Some(Finished { + usage: vllm_llm::TokenUsage { + prompt_token_count: 1, + output_token_count: 1, + cached_token_count: 0, + }, + finish_reason: FinishReason::stop_eos(), + kv_transfer_params: None, + ec_transfer_params: None, + sampling_mask: None, + timestamps: vllm_llm::RequestTimestamps { + queued_ts: 10.0, + scheduled_ts: 10.2, + first_token_ts: 10.5, + last_token_ts: 10.5, + }, + }), + )), + ]) + }; + + for enabled in [false, true] { + let response = collect_completion( + make_stream(), + "cmpl-1".into(), + "model".into(), + 1, + ApiServerOptions { + enable_per_request_metrics: enabled, + ..Default::default() + }, + ResponseOptions::default(), + ) + .await + .expect("completion response"); + let json = serde_json::to_value(response).expect("serializable response"); + if enabled { + assert!( + (json["metrics"]["time_to_first_token_ms"].as_f64().unwrap() - 300.0).abs() + < 1e-9 + ); + assert_eq!(json["metrics"]["mean_itl_ms"], serde_json::Value::Null); + } else { + assert_eq!(json["metrics"], serde_json::Value::Null); + } + } + } + #[tokio::test] async fn completion_chunk_stream_maps_streaming_logprobs() { let stream = stream::iter(vec![ @@ -702,6 +788,12 @@ mod tests { kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, + timestamps: vllm_llm::RequestTimestamps { + queued_ts: 10.0, + scheduled_ts: 10.2, + first_token_ts: 10.5, + last_token_ts: 11.0, + }, }), )), ]); @@ -713,6 +805,7 @@ mod tests { 1, ApiServerOptions { enable_prompt_tokens_details: true, + enable_per_request_metrics: true, ..Default::default() }, ResponseOptions { @@ -779,6 +872,10 @@ mod tests { match &chunks[3] { CompletionSseChunk::Usage(chunk) => { + assert!(chunks[..3].iter().all(|chunk| chunk_response(chunk).metrics.is_none())); + let metrics = chunk.metrics.as_ref().expect("per-request metrics"); + assert!((metrics.queue_time_ms.unwrap() - 200.0).abs() < 1e-9); + assert!((metrics.tokens_per_second.unwrap() - 2.5).abs() < 1e-9); assert_eq!( chunk .usage @@ -815,6 +912,7 @@ mod tests { kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, + timestamps: Default::default(), }), )), ]); @@ -879,6 +977,7 @@ mod tests { kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, + timestamps: Default::default(), }), )), ]); @@ -932,6 +1031,7 @@ mod tests { kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, + timestamps: Default::default(), }), )), ]); @@ -988,6 +1088,7 @@ mod tests { kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, + timestamps: Default::default(), }), )), ]); @@ -1061,6 +1162,7 @@ mod tests { kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, + timestamps: Default::default(), }), )), ]); @@ -1136,6 +1238,7 @@ mod tests { kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, + timestamps: Default::default(), }), )), ]); diff --git a/rust/src/server/src/routes/openai/completions/types.rs b/rust/src/server/src/routes/openai/completions/types.rs index 9f7027511c..fb6aa22c60 100644 --- a/rust/src/server/src/routes/openai/completions/types.rs +++ b/rust/src/server/src/routes/openai/completions/types.rs @@ -11,6 +11,7 @@ use validator::Validate; use vllm_engine_core_client::protocol::sampling::RepetitionDetectionParams; use vllm_text::{Prompt, TruncationSide}; +use crate::routes::openai::utils::metrics::{PerRequestMetrics, StreamPerRequestMetrics}; use crate::routes::openai::utils::types::{ LogProbs, Normalizable, PromptLogprobs, StreamOptions, StreamResponseEnvelope, StringOrArray, Usage, default_true, deserialize_request_top_k, validate_stop, @@ -239,6 +240,7 @@ pub(super) struct CompletionResponse { pub model: String, pub choices: Vec, pub usage: Option, + pub metrics: Option, pub system_fingerprint: Option, pub kv_transfer_params: Option, pub ec_transfer_params: Option, @@ -265,6 +267,7 @@ pub(super) struct CompletionStreamResponse { pub envelope: Arc, pub choices: Vec, pub usage: Option, + pub metrics: Option, } impl CompletionStreamResponse { @@ -274,6 +277,7 @@ impl CompletionStreamResponse { envelope: Arc::clone(envelope), choices: Vec::new(), usage: None, + metrics: None, } } } diff --git a/rust/src/server/src/routes/openai/utils/metrics.rs b/rust/src/server/src/routes/openai/utils/metrics.rs new file mode 100644 index 0000000000..0c1155f733 --- /dev/null +++ b/rust/src/server/src/routes/openai/utils/metrics.rs @@ -0,0 +1,122 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +use serde::Serialize; +use vllm_llm::RequestTimestamps; + +/// Per-request response timing fields matching Python's `PerRequestMetrics`. +#[derive(Debug, Clone, PartialEq, Serialize)] +pub struct PerRequestMetrics { + pub time_to_first_token_ms: Option, + pub generation_time_ms: Option, + pub queue_time_ms: Option, + pub mean_itl_ms: Option, + pub tokens_per_second: Option, +} + +impl PerRequestMetrics { + pub fn from_timestamps(times: RequestTimestamps, num_generation_tokens: usize) -> Self { + let duration = |end: f64, start: f64| (end > 0.0 && start > 0.0).then_some(end - start); + let decode = duration(times.last_token_ts, times.first_token_ts); + let inference = duration(times.last_token_ts, times.scheduled_ts); + + Self { + time_to_first_token_ms: duration(times.first_token_ts, times.scheduled_ts) + .map(|seconds| seconds * 1000.0), + generation_time_ms: decode.map(|seconds| seconds * 1000.0), + queue_time_ms: duration(times.scheduled_ts, times.queued_ts) + .map(|seconds| seconds * 1000.0), + mean_itl_ms: decode + .filter(|_| num_generation_tokens > 1) + .map(|seconds| seconds * 1000.0 / (num_generation_tokens - 1) as f64), + tokens_per_second: inference + .filter(|seconds| *seconds > 0.0) + .map(|seconds| num_generation_tokens as f64 / seconds), + } + } +} + +/// Streaming form of [`PerRequestMetrics`]. +/// +/// Python serializes the final usage chunk with recursive `exclude_none=True`, +/// so unset fields are omitted instead of serialized as `null`. +#[serde_with::skip_serializing_none] +#[derive(Debug, Clone, PartialEq, Serialize)] +pub struct StreamPerRequestMetrics { + pub time_to_first_token_ms: Option, + pub generation_time_ms: Option, + pub queue_time_ms: Option, + pub mean_itl_ms: Option, + pub tokens_per_second: Option, +} + +impl From for StreamPerRequestMetrics { + fn from(metrics: PerRequestMetrics) -> Self { + let PerRequestMetrics { + time_to_first_token_ms, + generation_time_ms, + queue_time_ms, + mean_itl_ms, + tokens_per_second, + } = metrics; + Self { + time_to_first_token_ms, + generation_time_ms, + queue_time_ms, + mean_itl_ms, + tokens_per_second, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn matches_python_timing_formulas_and_missing_fields() { + let times = RequestTimestamps { + queued_ts: 10.0, + scheduled_ts: 10.2, + first_token_ts: 10.5, + last_token_ts: 11.0, + }; + let metrics = PerRequestMetrics::from_timestamps(times, 6); + assert!((metrics.time_to_first_token_ms.unwrap() - 300.0).abs() < 1e-9); + assert_eq!(metrics.generation_time_ms, Some(500.0)); + assert!((metrics.queue_time_ms.unwrap() - 200.0).abs() < 1e-9); + assert_eq!(metrics.mean_itl_ms, Some(100.0)); + assert!((metrics.tokens_per_second.unwrap() - 7.5).abs() < 1e-9); + + let single_token = PerRequestMetrics::from_timestamps(times, 1); + assert_eq!(single_token.mean_itl_ms, None); + + let missing = PerRequestMetrics::from_timestamps(RequestTimestamps::default(), 0); + assert_eq!(missing.time_to_first_token_ms, None); + assert_eq!(missing.generation_time_ms, None); + assert_eq!(missing.queue_time_ms, None); + assert_eq!(missing.mean_itl_ms, None); + assert_eq!(missing.tokens_per_second, None); + } + + #[test] + fn stream_serialization_omits_unset_fields() { + let metrics = PerRequestMetrics::from_timestamps( + RequestTimestamps { + queued_ts: 10.0, + scheduled_ts: 10.2, + first_token_ts: 10.5, + last_token_ts: 10.5, + }, + 1, + ); + let value = serde_json::to_value(StreamPerRequestMetrics::from(metrics)).unwrap(); + let metrics = value.as_object().unwrap(); + assert_eq!(metrics.len(), 4); + assert!(metrics.contains_key("time_to_first_token_ms")); + assert!(metrics.contains_key("generation_time_ms")); + assert!(metrics.contains_key("queue_time_ms")); + assert!(metrics.contains_key("tokens_per_second")); + assert!(!metrics.contains_key("mean_itl_ms")); + } +} diff --git a/rust/src/server/src/routes/openai/utils/mod.rs b/rust/src/server/src/routes/openai/utils/mod.rs index 4bbea924a3..a5f00ae0ea 100644 --- a/rust/src/server/src/routes/openai/utils/mod.rs +++ b/rust/src/server/src/routes/openai/utils/mod.rs @@ -2,6 +2,7 @@ // SPDX-FileCopyrightText: Copyright contributors to the vLLM project pub mod logprobs; +pub mod metrics; pub mod structured_outputs; pub mod types; pub mod usage; diff --git a/rust/src/text/src/output/decoded.rs b/rust/src/text/src/output/decoded.rs index 64df0492bf..6308bdf6e1 100644 --- a/rust/src/text/src/output/decoded.rs +++ b/rust/src/text/src/output/decoded.rs @@ -10,7 +10,7 @@ use tracing::{Level, debug, trace}; use vllm_engine_core_client::AbortCause; use vllm_engine_core_client::protocol::output::StopReason; use vllm_engine_core_client::protocol::sampling_mask::SamplingMask; -use vllm_llm::{FinishReason, GenerateOutput, TokenUsage}; +use vllm_llm::{FinishReason, GenerateOutput, RequestTimestamps, TokenUsage}; use vllm_tokenizer::{DecodedText, DynTokenizer, IncrementalDecoder}; use super::logprobs::{ @@ -55,6 +55,7 @@ pub struct Finished { pub ec_transfer_params: Option>, /// Sampling support sets aligned with all generated token positions. pub sampling_mask: Option, + pub timestamps: RequestTimestamps, } /// Sample metadata emitted by one engine output update. @@ -317,6 +318,7 @@ pub async fn decoded_text_event_stream( kv_transfer_params, ec_transfer_params, sampling_mask, + timestamps: output.timestamps, })), }) .await; @@ -543,7 +545,19 @@ mod tests { name: "utf8 byte fallback resolves across engine updates", outputs: vec![ GenerateOutput::for_test(Some(Arc::clone(&prompt)), vec![0xe4], None), - GenerateOutput::for_test(None, vec![0xbd, 0xa0], Some(FinishReason::Length)), + GenerateOutput { + timestamps: RequestTimestamps { + queued_ts: 8.0, + scheduled_ts: 9.0, + first_token_ts: 10.0, + last_token_ts: 11.5, + }, + ..GenerateOutput::for_test( + None, + vec![0xbd, 0xa0], + Some(FinishReason::Length), + ) + }, ], expected: vec![ DecodedTextEvent::Start { @@ -565,6 +579,12 @@ mod tests { kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, + timestamps: RequestTimestamps { + queued_ts: 8.0, + scheduled_ts: 9.0, + first_token_ts: 10.0, + last_token_ts: 11.5, + }, }), ), ], @@ -597,6 +617,7 @@ mod tests { kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, + timestamps: Default::default(), }), ), ], @@ -634,12 +655,19 @@ mod tests { async fn stream_stop_string_sets_task_local_abort_cause_on_raw_stream_drop() { let prompt: Arc<[u32]> = Arc::from([]); let dropped_cause = Arc::new(Mutex::new(None)); + let timestamps = RequestTimestamps { + queued_ts: 1.0, + scheduled_ts: 2.0, + first_token_ts: 3.0, + last_token_ts: 4.0, + }; let raw_stream = DropRecordingStream { - next: Some(Ok(GenerateOutput::for_test( - Some(prompt), - ascii_tokens("hello"), - Some(FinishReason::Length), - ))), + // The engine output is not terminal. The frontend terminates the + // request after decoding the configured stop string. + next: Some(Ok(GenerateOutput { + timestamps, + ..GenerateOutput::for_test(Some(prompt), ascii_tokens("hello"), None) + })), dropped_cause: Arc::clone(&dropped_cause), }; let tokenizer: DynTokenizer = Arc::new(TestTokenizer::new()); @@ -657,6 +685,7 @@ mod tests { assert_eq!(output.text, "he"); assert!(output.finish_reason.is_stop()); + assert_eq!(output.timestamps, timestamps); assert_eq!( *dropped_cause.lock().unwrap(), Some(AbortCause::StopStringMatched) diff --git a/rust/src/text/src/output/mod.rs b/rust/src/text/src/output/mod.rs index 8e09f07618..b69d439e3f 100644 --- a/rust/src/text/src/output/mod.rs +++ b/rust/src/text/src/output/mod.rs @@ -31,6 +31,7 @@ pub struct CollectedTextOutput { pub token_ids: Vec, pub finish_reason: FinishReason, pub usage: vllm_llm::TokenUsage, + pub timestamps: vllm_llm::RequestTimestamps, /// Connector-specific KV transfer parameters for disaggregated serving. pub kv_transfer_params: Option, /// Connector-specific encoder cache transfer parameters for disaggregated @@ -91,6 +92,7 @@ impl T { token_ids: delta_token_ids, finish_reason: FinishReason::Error, usage: vllm_llm::TokenUsage::default(), + timestamps: vllm_llm::RequestTimestamps::default(), kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, @@ -101,6 +103,7 @@ impl T { let mut collected = collected.unwrap(); collected.finish_reason = finished.finish_reason; collected.usage = finished.usage; + collected.timestamps = finished.timestamps; collected.kv_transfer_params = finished.kv_transfer_params.map(|value| *value); collected.ec_transfer_params = @@ -181,6 +184,7 @@ mod tests { kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, + timestamps: Default::default(), })), }), ]); @@ -304,6 +308,7 @@ mod tests { kv_transfer_params: None, ec_transfer_params: None, sampling_mask: None, + timestamps: Default::default(), })), }), ]); diff --git a/tests/entrypoints/openai/chat_completion/test_chat_completion.py b/tests/entrypoints/openai/chat_completion/test_chat_completion.py index e0422a520a..ff6595fb0c 100644 --- a/tests/entrypoints/openai/chat_completion/test_chat_completion.py +++ b/tests/entrypoints/openai/chat_completion/test_chat_completion.py @@ -20,6 +20,7 @@ def default_server_args(): "--max-num-seqs", "128", "--enforce-eager", + "--enable-per-request-metrics", ] @@ -247,3 +248,59 @@ async def test_kv_transfer_prompt_token_ids_streaming(client: openai.AsyncOpenAI # streamed text-out, reconstructed from deltas, with generated token ids. assert content assert delta_token_ids + + +def _assert_complete_timing_metrics(metrics: object) -> None: + assert isinstance(metrics, dict) + timing_fields = { + "time_to_first_token_ms", + "generation_time_ms", + "queue_time_ms", + "mean_itl_ms", + "tokens_per_second", + } + assert timing_fields <= metrics.keys() + assert all(metrics[field] >= 0 for field in timing_fields) + + +@pytest.mark.asyncio +async def test_per_request_metrics_non_streaming(client: openai.AsyncOpenAI): + response = await client.chat.completions.create( + model=MODEL_NAME, + messages=[{"role": "user", "content": "Count to four."}], + max_completion_tokens=4, + temperature=0, + extra_body={"min_tokens": 4}, + ) + + _assert_complete_timing_metrics(response.model_extra.get("metrics")) + + +@pytest.mark.asyncio +async def test_per_request_metrics_on_final_usage_chunk(client: openai.AsyncOpenAI): + stream = await client.chat.completions.create( + model=MODEL_NAME, + messages=[{"role": "user", "content": "Say hello."}], + max_completion_tokens=1, + temperature=0, + stream=True, + stream_options={"include_usage": True}, + extra_body={"min_tokens": 1}, + ) + + chunks = [chunk async for chunk in stream] + assert len(chunks) > 1 + for chunk in chunks[:-1]: + assert chunk.usage is None + assert "metrics" not in (chunk.model_extra or {}) + + usage_chunk = chunks[-1] + assert usage_chunk.usage is not None + assert usage_chunk.choices == [] + metrics = (usage_chunk.model_extra or {}).get("metrics") + assert isinstance(metrics, dict) + assert metrics["time_to_first_token_ms"] >= 0 + assert metrics["generation_time_ms"] >= 0 + assert metrics["queue_time_ms"] >= 0 + assert metrics["tokens_per_second"] >= 0 + assert "mean_itl_ms" not in metrics