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

[Rust Frontend] Add per-request timing metrics to chat and completions (#59314)

Browse Source 
Signed-off-by: wxin <wxin@nvidia.com>
Signed-off-by: Bugen Zhao <i@bugenzhao.com>
Co-authored-by: Bugen Zhao <i@bugenzhao.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
xinnywinne
2026-10-12 15:39:10 -07:00
committed by GitHub
co-authored by Bugen Zhao Claude Opus 5.5
parent 802feb6770
commit 1bc6e072b0
28 changed files with 517 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rust/src/chat/src/event.rs
+3 -1
View File
@@ -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
rust/src/chat/src/output/default/unified.rs
+7
View File
@@ -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,
rust/src/chat/src/output/harmony/mod.rs
+1
View File
@@ -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,
rust/src/chat/src/output/harmony/tests.rs
+3
View File
@@ -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,
rust/src/chat/src/output/mod.rs
+2
View File
@@ -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
rust/src/chat/src/output/structured.rs
+17 -3
View File
@@ -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<Box<serde_json::Value>>,
ec_transfer_params: Option<Box<serde_json::Value>>,
@@ -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,
rust/src/chat/src/stream.rs
+6
View File
@@ -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<DecodedLogprobs>,
pub token_ids: Vec<u32>,
pub usage: ChatTokenUsage,
pub timestamps: RequestTimestamps,
pub finish_reason: FinishReason,
/// Connector-specific KV transfer parameters for disaggregated serving.
pub kv_transfer_params: Option<serde_json::Value>,
@@ -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,
rust/src/chat/tests/roundtrip.rs
+2
View File
@@ -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,
rust/src/cmd/src/cli.rs
+6
View File
@@ -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)
rust/src/cmd/src/cli/tests.rs
+40
View File
@@ -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,
rust/src/cmd/src/cli/unsupported.rs
-9
View File
@@ -476,15 +476,6 @@ pub struct ServerUnsupportedArgs {
#[arg(long)]
pub max_log_len: Option<Unsupported>,
/// 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<Unsupported>,
/// If set to True, enable tracking server_load_metrics in the app state.
#[arg(
long,
rust/src/llm/src/lib.rs
+1 -1
View File
@@ -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;
rust/src/llm/src/output.rs
+10
View File
@@ -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<Box<RequestSpecDecodeMetrics>>,
/// Weight-version spans over all output tokens, set on terminal outputs only.
pub weight_versions: Option<Vec<WeightVersionSpan>>,
/// 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)
rust/src/llm/src/request_metrics.rs
+29 -1
View File
@@ -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),
rust/src/llm/tests/generate.rs
+8 -1
View File
@@ -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();
rust/src/server/src/config.rs
+7
View File
@@ -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 {
rust/src/server/src/grpc/convert.rs
+1
View File
@@ -947,6 +947,7 @@ mod tests {
kv_transfer_params: None,
ec_transfer_params: None,
sampling_mask: None,
timestamps: Default::default(),
}
}
rust/src/server/src/grpc/inference.rs
+1
View File
@@ -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(
rust/src/server/src/routes/inference/generate.rs
+3
View File
@@ -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(),
}
}
rust/src/server/src/routes/openai/chat_completions.rs
+36
View File
@@ -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<StreamResponseEnvelope>,
usage: Usage,
metrics: Option<PerRequestMetrics>,
) -> 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,
rust/src/server/src/routes/openai/chat_completions/types.rs
+4
View File
@@ -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<ChatCompletionChoice>,
pub usage: Option<Usage>,
pub metrics: Option<PerRequestMetrics>,
pub system_fingerprint: Option<String>,
pub prompt_logprobs: Option<PromptLogprobs>,
pub prompt_token_ids: Option<Vec<u32>>,
@@ -412,6 +414,7 @@ pub(super) struct ChatCompletionStreamResponse {
pub envelope: Arc<StreamResponseEnvelope>,
pub choices: Vec<ChatCompletionStreamChoice>,
pub usage: Option<Usage>,
pub metrics: Option<StreamPerRequestMetrics>,
pub prompt_token_ids: Option<Vec<u32>>,
}
@@ -422,6 +425,7 @@ impl ChatCompletionStreamResponse {
envelope: Arc::clone(envelope),
choices: Vec::new(),
usage: None,
metrics: None,
prompt_token_ids: None,
}
}
rust/src/server/src/routes/openai/completions.rs
+105 -2
View File
@@ -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<StreamResponseEnvelope>, usage: Usage) -> CompletionStreamResponse {
fn usage_chunk(
envelope: &Arc<StreamResponseEnvelope>,
usage: Usage,
metrics: Option<PerRequestMetrics>,
) -> 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(),
}),
)),
]);
rust/src/server/src/routes/openai/completions/types.rs
+4
View File
@@ -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<CompletionChoice>,
pub usage: Option<Usage>,
pub metrics: Option<PerRequestMetrics>,
pub system_fingerprint: Option<String>,
pub kv_transfer_params: Option<Value>,
pub ec_transfer_params: Option<Value>,
@@ -265,6 +267,7 @@ pub(super) struct CompletionStreamResponse {
pub envelope: Arc<StreamResponseEnvelope>,
pub choices: Vec<CompletionStreamChoice>,
pub usage: Option<Usage>,
pub metrics: Option<StreamPerRequestMetrics>,
}
impl CompletionStreamResponse {
@@ -274,6 +277,7 @@ impl CompletionStreamResponse {
envelope: Arc::clone(envelope),
choices: Vec::new(),
usage: None,
metrics: None,
}
}
}
rust/src/server/src/routes/openai/utils/metrics.rs
+122
View File
@@ -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<f64>,
pub generation_time_ms: Option<f64>,
pub queue_time_ms: Option<f64>,
pub mean_itl_ms: Option<f64>,
pub tokens_per_second: Option<f64>,
}
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<f64>,
pub generation_time_ms: Option<f64>,
pub queue_time_ms: Option<f64>,
pub mean_itl_ms: Option<f64>,
pub tokens_per_second: Option<f64>,
}
impl From<PerRequestMetrics> 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"));
}
}
rust/src/server/src/routes/openai/utils/mod.rs
+1
View File
@@ -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;
rust/src/text/src/output/decoded.rs
+36 -7
View File
@@ -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<Box<serde_json::Value>>,
/// Sampling support sets aligned with all generated token positions.
pub sampling_mask: Option<SamplingMask>,
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)
rust/src/text/src/output/mod.rs
+5
View File
@@ -31,6 +31,7 @@ pub struct CollectedTextOutput {
pub token_ids: Vec<u32>,
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<serde_json::Value>,
/// Connector-specific encoder cache transfer parameters for disaggregated
@@ -91,6 +92,7 @@ impl<T: TextOutputStream> 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: TextOutputStream> 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(),
})),
}),
]);
tests/entrypoints/openai/chat_completion/test_chat_completion.py
+57
View File
@@ -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