diff --git a/rust/proto/control.proto b/rust/proto/control.proto index 428ad6460499..bf0aee04b8c0 100644 --- a/rust/proto/control.proto +++ b/rust/proto/control.proto @@ -39,6 +39,8 @@ message ServerInfo { uint64 max_running_requests = 8; uint64 max_batched_tokens = 9; RlCapabilities rl_capabilities = 11; + bool supports_native_sampling_params_json = 12; + bool supports_preprocessed_media_features = 13; } message RlCapabilities { @@ -54,6 +56,8 @@ message ParallelismInfo { uint32 data_parallel_size = 3; uint32 data_parallel_rank = 4; uint32 decode_context_parallel_size = 5; + // Process world size for one data-parallel engine (TP * PP * PCP). + uint64 world_size = 6; } message GetModelInfoRequest {} diff --git a/rust/proto/inference.proto b/rust/proto/inference.proto index 897319ca9c07..ed2885f1365d 100644 --- a/rust/proto/inference.proto +++ b/rust/proto/inference.proto @@ -50,6 +50,11 @@ message GenerateRequest { // Multimodal inputs aligned with placeholder markers in token_ids. repeated MediaItem media = 14; + + // Exact JSON object from native /inference/v1/generate sampling_params. + // When non-empty, this is the sampling authority; structured fields remain + // the compatibility path for generic gRPC clients. + bytes native_sampling_params_json = 16; } message RandomSampling { @@ -225,7 +230,20 @@ message MediaItem { string url = 2; // http:// or https:// string data_uri = 3; // data: bytes raw_bytes = 4; + PreprocessedMediaFeatures features = 7; } string mime_type = 5; string uuid = 6; } + +// Engine-ready image features produced by a version-compatible frontend. +// `kwargs` is the MessagePack wire representation of one MmKwargsItem. +message PreprocessedMediaFeatures { + // Required inline payload. Cache-only references are not accepted. + optional bytes kwargs = 1; + string identifier = 2; + uint64 offset = 3; + uint64 length = 4; + optional string mm_hash = 5; + repeated bool is_embed = 6; +} diff --git a/rust/src/engine-core-client/src/protocol/output.rs b/rust/src/engine-core-client/src/protocol/output.rs index a9695bd1cf11..80a8a1289eda 100644 --- a/rust/src/engine-core-client/src/protocol/output.rs +++ b/rust/src/engine-core-client/src/protocol/output.rs @@ -73,6 +73,20 @@ pub struct EngineCoreEvent { pub timestamp: f64, } +/// Compatibility-only engine output value that is decoded without allocation. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +pub struct DiscardedEngineOutputValue; + +impl<'de> Deserialize<'de> for DiscardedEngineOutputValue { + fn deserialize(deserializer: D) -> std::result::Result + where + D: Deserializer<'de>, + { + serde::de::IgnoredAny::deserialize(deserializer)?; + Ok(Self) + } +} + /// Engine-core output for a single request. /// /// Original Python definition: @@ -125,6 +139,12 @@ pub struct EngineCoreOutput { pub mm_cache_miss_hashes: Option>, #[serde(default)] pub new_sampling_mask: Option, + /// Per-request speculative-decoding metrics appended by newer v0.28 wheels. + #[serde(default)] + pub spec_decode_metrics: Option, + /// Routed-expert metadata appended by newer v0.28 wheels. + #[serde(default)] + pub routed_experts_payload: Option, } impl EngineCoreOutput { @@ -443,6 +463,8 @@ mod tests { num_nans_in_logits: 0, mm_cache_miss_hashes: None, new_sampling_mask: None, + spec_decode_metrics: None, + routed_experts_payload: None, }, ], scheduler_stats: None, diff --git a/rust/src/engine-core-client/src/tests/client.rs b/rust/src/engine-core-client/src/tests/client.rs index 8f07a8059e92..acaa11648bd3 100644 --- a/rust/src/engine-core-client/src/tests/client.rs +++ b/rust/src/engine-core-client/src/tests/client.rs @@ -2612,6 +2612,12 @@ fn python_msgpack_fixtures_match_rust_encoding() { num_nans_in_logits: 0, mm_cache_miss_hashes: None, new_sampling_mask: None, + spec_decode_metrics: Some( + DiscardedEngineOutputValue, + ), + routed_experts_payload: Some( + DiscardedEngineOutputValue, + ), }, ], scheduler_stats: None, diff --git a/rust/src/engine-core-client/src/tests/python_compat.py b/rust/src/engine-core-client/src/tests/python_compat.py index 389e97cd9996..1e7a526682a4 100755 --- a/rust/src/engine-core-client/src/tests/python_compat.py +++ b/rust/src/engine-core-client/src/tests/python_compat.py @@ -99,6 +99,8 @@ class EngineCoreOutput( num_nans_in_logits: int = 0 mm_cache_miss_hashes: list[str] | None = None new_sampling_mask: object | None = None + spec_decode_metrics: object | None = None + routed_experts_payload: object | None = None class EngineCoreOutputs( @@ -198,6 +200,8 @@ class EngineCoreOutputs( request_id="req-1", new_token_ids=[7, 8], finish_reason=FinishReason.LENGTH, + spec_decode_metrics={"accepted": 1}, + routed_experts_payload={"format": "test"}, ) ], finished_requests={"req-1"}, diff --git a/rust/src/server/Cargo.toml b/rust/src/server/Cargo.toml index 8b8f57a0dac5..134f7c32e973 100644 --- a/rust/src/server/Cargo.toml +++ b/rust/src/server/Cargo.toml @@ -21,6 +21,7 @@ llm-multimodal.workspace = true openssl.workspace = true prost.workspace = true prost-types.workspace = true +rmp-serde.workspace = true rmpv.workspace = true serde.workspace = true serde_json.workspace = true @@ -60,7 +61,6 @@ async-openai = { workspace = true, features = ["full"] } bytes.workspace = true clap.workspace = true expect-test.workspace = true -rmp-serde.workspace = true serial_test.workspace = true tempfile.workspace = true tokio = { workspace = true, features = ["test-util"] } diff --git a/rust/src/server/src/grpc/control.rs b/rust/src/server/src/grpc/control.rs index c7d9af88c82f..e97e7b5d65c5 100644 --- a/rust/src/server/src/grpc/control.rs +++ b/rust/src/server/src/grpc/control.rs @@ -46,6 +46,7 @@ impl ControlServiceImpl { data_parallel_size: self.state.data_parallel_size().min(u32::MAX as usize) as u32, data_parallel_rank: ready.data_parallel_rank, decode_context_parallel_size: ready.decode_context_parallel_size, + world_size: ready.world_size, } } @@ -151,6 +152,8 @@ impl pb::control_server::Control for ControlServiceImpl { max_running_requests: ready.max_num_seqs, max_batched_tokens: ready.max_num_batched_tokens, rl_capabilities: Some(self.rl_capabilities()), + supports_native_sampling_params_json: true, + supports_preprocessed_media_features: true, })) } diff --git a/rust/src/server/src/grpc/convert.rs b/rust/src/server/src/grpc/convert.rs index 4460f6b22249..8164d037cb50 100644 --- a/rust/src/server/src/grpc/convert.rs +++ b/rust/src/server/src/grpc/convert.rs @@ -4,12 +4,20 @@ //! Conversion between gRPC protobuf types and internal `vllm-text` //! request/response types. +use std::io::Cursor; + +use serde::Deserialize as _; +use sha2::{Digest as _, Sha256}; use tonic::Status; use url::Url; use uuid::Uuid; use vllm_chat::MediaContentPart; +use vllm_engine_core_client::protocol::multimodal::{ + MmFeatureSpec, MmFeatures, MmKwargValue, MmKwargsItem, PlaceholderRange, +}; use vllm_engine_core_client::protocol::output::StopReason; use vllm_engine_core_client::protocol::structured_outputs::StructuredOutputsParams; +use vllm_engine_core_client::protocol::tensor::{WireArrayData, WireTensor}; use vllm_text::{ DecodedLogprobs, DecodedPromptLogprobs, FinishReason, Finished, Prompt, SamplingParams, TextDecodeOptions, TextRequest, @@ -17,10 +25,22 @@ use vllm_text::{ use super::pb; -pub fn media_parts_from_request( +const MAX_MM_DEPTH: usize = 32; +const MAX_MM_NODES: usize = 65_536; +const MAX_MM_TENSOR_RANK: usize = 32; +const PREPROCESSED_MM_ID_DOMAIN: &[u8] = b"vllm.grpc.preprocessed-mm.v1"; + +pub enum GrpcMultimodalInput { + Raw(Vec), + Preprocessed(MmFeatures), +} + +pub fn multimodal_input_from_request( media: Vec, -) -> Result, Status> { - let mut parts = Vec::with_capacity(media.len()); +) -> Result { + let mut parts = Vec::new(); + let mut features = Vec::new(); + let mut wire_nodes = 0usize; for (index, item) in media.into_iter().enumerate() { match item.modality() { pb::Modality::Image => {} @@ -35,10 +55,23 @@ pub fn media_parts_from_request( ))); } } + let source = item.source.ok_or_else(|| { + Status::invalid_argument(format!("media[{index}].source is required")) + })?; + if let pb::media_item::Source::Features(feature) = source { + if !parts.is_empty() { + return Err(mixed_media_error()); + } + features.push(preprocessed_feature(index, feature, &mut wire_nodes)?); + continue; + } + if !features.is_empty() { + return Err(mixed_media_error()); + } let uuid = (!item.uuid.is_empty()).then_some(item.uuid); let mime_type = (!item.mime_type.is_empty()).then_some(item.mime_type); - let part = match item.source { - Some(pb::media_item::Source::Url(url)) => { + let part = match source { + pb::media_item::Source::Url(url) => { validate_media_uri(index, "url", &url, &["http", "https"])?; MediaContentPart::ImageUrl { url, @@ -46,7 +79,7 @@ pub fn media_parts_from_request( uuid, } } - Some(pb::media_item::Source::DataUri(uri)) => { + pb::media_item::Source::DataUri(uri) => { validate_media_uri(index, "data_uri", &uri, &["data"])?; MediaContentPart::ImageUrl { url: uri, @@ -54,21 +87,340 @@ pub fn media_parts_from_request( uuid, } } - Some(pb::media_item::Source::RawBytes(bytes)) => MediaContentPart::ImageData { + pb::media_item::Source::RawBytes(bytes) => MediaContentPart::ImageData { data: bytes, mime_type, uuid, detail: None, }, - None => { - return Err(Status::invalid_argument(format!( - "media[{index}].source is required" - ))); - } + pb::media_item::Source::Features(_) => unreachable!("handled above"), }; parts.push(part); } - Ok(parts) + if features.is_empty() { + return Ok(GrpcMultimodalInput::Raw(parts)); + } + + features.sort_by_key(|feature| feature.mm_position.offset); + for pair in features.windows(2) { + let previous_end = pair[0] + .mm_position + .offset + .checked_add(pair[0].mm_position.length) + .expect("placeholder range validated above"); + if previous_end > pair[1].mm_position.offset { + return Err(Status::invalid_argument( + "preprocessed image feature placeholder ranges overlap", + )); + } + } + Ok(GrpcMultimodalInput::Preprocessed(features)) +} + +fn mixed_media_error() -> Status { + Status::invalid_argument("raw media and preprocessed image features cannot be mixed") +} + +fn preprocessed_feature( + index: usize, + feature: pb::PreprocessedMediaFeatures, + wire_nodes: &mut usize, +) -> Result { + let offset = usize::try_from(feature.offset).map_err(|_| { + Status::invalid_argument(format!( + "media[{index}].features.offset exceeds platform limits" + )) + })?; + let length = usize::try_from(feature.length).map_err(|_| { + Status::invalid_argument(format!( + "media[{index}].features.length exceeds platform limits" + )) + })?; + if length == 0 { + return Err(Status::invalid_argument(format!( + "media[{index}].features.length must be positive" + ))); + } + offset.checked_add(length).ok_or_else(|| { + Status::invalid_argument(format!( + "media[{index}].features placeholder range overflows" + )) + })?; + let is_embed = if feature.is_embed.is_empty() { + None + } else { + if feature.is_embed.len() != length { + return Err(Status::invalid_argument(format!( + "media[{index}].features.is_embed length must equal placeholder length" + ))); + } + Some( + WireTensor::from_bool(vec![length], feature.is_embed).map_err(|error| { + Status::invalid_argument(format!( + "media[{index}].features.is_embed is invalid: {error}" + )) + })?, + ) + }; + let raw = feature.kwargs.ok_or_else(|| { + Status::invalid_argument(format!( + "media[{index}].features.kwargs is required; cache-only features are unsupported" + )) + })?; + let data = decode_mm_kwargs(index, &raw, wire_nodes)?; + validate_mm_kwargs(index, &data)?; + + let mut hasher = Sha256::new(); + hasher.update(PREPROCESSED_MM_ID_DOMAIN); + hasher.update(5_u64.to_be_bytes()); + hasher.update(b"image"); + hasher.update((raw.len() as u64).to_be_bytes()); + hasher.update(&raw); + + let fingerprint = format!("{:x}", hasher.finalize()); + + Ok(MmFeatureSpec { + data: Some(data), + modality: "image".to_string(), + identifier: format!("grpc-mm:{fingerprint}"), + mm_position: PlaceholderRange { + offset, + length, + is_embed, + }, + mm_hash: None, + }) +} + +fn validate_mm_kwargs(index: usize, kwargs: &MmKwargsItem) -> Result<(), Status> { + if kwargs.is_empty() { + return Err(Status::invalid_argument(format!( + "media[{index}].features.kwargs must not be empty" + ))); + } + for element in kwargs.values() { + let value = element.data.as_ref().ok_or_else(|| { + Status::invalid_argument(format!( + "media[{index}].features.kwargs must contain inline data" + )) + })?; + validate_mm_value(index, value, 0)?; + } + Ok(()) +} + +fn validate_mm_value(index: usize, value: &MmKwargValue, depth: usize) -> Result<(), Status> { + if depth > MAX_MM_DEPTH { + return Err(Status::resource_exhausted(format!( + "media[{index}].features.kwargs nesting exceeds {MAX_MM_DEPTH} levels" + ))); + } + match value { + MmKwargValue::Tensor(tensor) => validate_mm_tensor(index, tensor), + MmKwargValue::List(values) => { + for value in values { + validate_mm_value(index, value, depth + 1)?; + } + Ok(()) + } + MmKwargValue::Int(_) | MmKwargValue::Float(_) => Ok(()), + } +} + +fn validate_mm_tensor(index: usize, tensor: &WireTensor) -> Result<(), Status> { + if tensor.shape.len() > MAX_MM_TENSOR_RANK { + return Err(Status::invalid_argument(format!( + "media[{index}].features.kwargs tensor rank exceeds {MAX_MM_TENSOR_RANK}" + ))); + } + let width: usize = match tensor.dtype.as_str() { + "bool" | "uint8" | "int8" => 1, + "float16" | "bfloat16" | "uint16" | "int16" => 2, + "float32" | "uint32" | "int32" => 4, + "float64" | "uint64" | "int64" => 8, + _ => { + return Err(Status::invalid_argument(format!( + "media[{index}].features.kwargs has an unsupported tensor dtype" + ))); + } + }; + let expected = tensor + .shape + .iter() + .try_fold(width, |bytes, dim| bytes.checked_mul(*dim)) + .ok_or_else(|| { + Status::invalid_argument(format!( + "media[{index}].features.kwargs tensor size overflows" + )) + })?; + match &tensor.data { + WireArrayData::RawView(bytes) if bytes.len() == expected => Ok(()), + WireArrayData::RawView(bytes) => Err(Status::invalid_argument(format!( + "media[{index}].features.kwargs tensor byte length {} does not match expected {expected}", + bytes.len() + ))), + WireArrayData::AuxIndex(aux_index) => Err(Status::invalid_argument(format!( + "media[{index}].features.kwargs references auxiliary frame {aux_index}" + ))), + } +} + +fn decode_mm_kwargs( + index: usize, + raw: &[u8], + wire_nodes: &mut usize, +) -> Result { + let mut cursor = 0usize; + scan_msgpack(raw, &mut cursor, 0, wire_nodes)?; + if cursor != raw.len() { + return Err(Status::invalid_argument(format!( + "media[{index}].features.kwargs contains trailing MessagePack data" + ))); + } + let mut deserializer = rmp_serde::Deserializer::new(Cursor::new(raw)); + deserializer.set_max_depth(MAX_MM_DEPTH); + MmKwargsItem::deserialize(&mut deserializer).map_err(|error| { + Status::invalid_argument(format!( + "media[{index}].features.kwargs is invalid MessagePack: {error}" + )) + }) +} + +fn scan_msgpack( + raw: &[u8], + cursor: &mut usize, + depth: usize, + nodes: &mut usize, +) -> Result<(), Status> { + if depth > MAX_MM_DEPTH { + return Err(Status::resource_exhausted( + "preprocessed image kwargs is nested too deeply", + )); + } + *nodes = nodes + .checked_add(1) + .ok_or_else(|| Status::resource_exhausted("preprocessed image kwargs is too complex"))?; + if *nodes > MAX_MM_NODES { + return Err(Status::resource_exhausted( + "preprocessed image kwargs contains too many values", + )); + } + + let marker = take_msgpack(raw, cursor, 1)?[0]; + match marker { + 0x00..=0x7f | 0xc0 | 0xc2 | 0xc3 | 0xe0..=0xff => Ok(()), + 0x80..=0x8f => { + scan_msgpack_children(raw, cursor, depth, nodes, usize::from(marker & 0x0f) * 2) + } + 0x90..=0x9f => scan_msgpack_children(raw, cursor, depth, nodes, usize::from(marker & 0x0f)), + 0xa0..=0xbf => skip_msgpack(raw, cursor, usize::from(marker & 0x1f)), + 0xc1 => Err(Status::invalid_argument("reserved MessagePack marker")), + 0xc4 | 0xd9 => { + let len = usize::from(read_msgpack_u8(raw, cursor)?); + skip_msgpack(raw, cursor, len) + } + 0xc5 | 0xda => { + let len = usize::from(read_msgpack_u16(raw, cursor)?); + skip_msgpack(raw, cursor, len) + } + 0xc6 | 0xdb => { + let len = usize::try_from(read_msgpack_u32(raw, cursor)?) + .map_err(|_| Status::resource_exhausted("MessagePack value is too large"))?; + skip_msgpack(raw, cursor, len) + } + 0xc7 => { + let len = usize::from(read_msgpack_u8(raw, cursor)?); + skip_msgpack(raw, cursor, len + 1) + } + 0xc8 => { + let len = usize::from(read_msgpack_u16(raw, cursor)?); + skip_msgpack(raw, cursor, len + 1) + } + 0xc9 => { + let len = usize::try_from(read_msgpack_u32(raw, cursor)?) + .map_err(|_| Status::resource_exhausted("MessagePack value is too large"))?; + skip_msgpack(raw, cursor, len + 1) + } + 0xca => skip_msgpack(raw, cursor, 4), + 0xcb => skip_msgpack(raw, cursor, 8), + 0xcc | 0xd0 => skip_msgpack(raw, cursor, 1), + 0xcd | 0xd1 => skip_msgpack(raw, cursor, 2), + 0xce | 0xd2 => skip_msgpack(raw, cursor, 4), + 0xcf | 0xd3 => skip_msgpack(raw, cursor, 8), + 0xd4 => skip_msgpack(raw, cursor, 2), + 0xd5 => skip_msgpack(raw, cursor, 3), + 0xd6 => skip_msgpack(raw, cursor, 5), + 0xd7 => skip_msgpack(raw, cursor, 9), + 0xd8 => skip_msgpack(raw, cursor, 17), + 0xdc => { + let count = usize::from(read_msgpack_u16(raw, cursor)?); + scan_msgpack_children(raw, cursor, depth, nodes, count) + } + 0xdd => { + let count = usize::try_from(read_msgpack_u32(raw, cursor)?) + .map_err(|_| Status::resource_exhausted("MessagePack array is too large"))?; + scan_msgpack_children(raw, cursor, depth, nodes, count) + } + 0xde => { + let count = usize::from(read_msgpack_u16(raw, cursor)?) * 2; + scan_msgpack_children(raw, cursor, depth, nodes, count) + } + 0xdf => { + let count = usize::try_from(read_msgpack_u32(raw, cursor)?) + .map_err(|_| Status::resource_exhausted("MessagePack map is too large"))? + .checked_mul(2) + .ok_or_else(|| Status::resource_exhausted("MessagePack map is too large"))?; + scan_msgpack_children(raw, cursor, depth, nodes, count) + } + } +} + +fn scan_msgpack_children( + raw: &[u8], + cursor: &mut usize, + depth: usize, + nodes: &mut usize, + count: usize, +) -> Result<(), Status> { + if count > MAX_MM_NODES.saturating_sub(*nodes) { + return Err(Status::resource_exhausted( + "preprocessed image kwargs contains too many values", + )); + } + for _ in 0..count { + scan_msgpack(raw, cursor, depth + 1, nodes)?; + } + Ok(()) +} + +fn take_msgpack<'a>(raw: &'a [u8], cursor: &mut usize, len: usize) -> Result<&'a [u8], Status> { + let end = cursor + .checked_add(len) + .filter(|end| *end <= raw.len()) + .ok_or_else(|| Status::invalid_argument("truncated preprocessed image kwargs"))?; + let bytes = &raw[*cursor..end]; + *cursor = end; + Ok(bytes) +} + +fn skip_msgpack(raw: &[u8], cursor: &mut usize, len: usize) -> Result<(), Status> { + take_msgpack(raw, cursor, len).map(|_| ()) +} + +fn read_msgpack_u8(raw: &[u8], cursor: &mut usize) -> Result { + Ok(take_msgpack(raw, cursor, 1)?[0]) +} + +fn read_msgpack_u16(raw: &[u8], cursor: &mut usize) -> Result { + Ok(u16::from_be_bytes( + take_msgpack(raw, cursor, 2)?.try_into().expect("fixed two-byte slice"), + )) +} + +fn read_msgpack_u32(raw: &[u8], cursor: &mut usize) -> Result { + Ok(u32::from_be_bytes( + take_msgpack(raw, cursor, 4)?.try_into().expect("fixed four-byte slice"), + )) } fn validate_media_uri( @@ -135,8 +487,15 @@ pub fn to_text_request( let response = req.response.as_ref(); let kv = req.kv.as_ref(); - let mut sampling_params = - build_sampling_params(req.temperature, sampling, decoding, stopping, response)?; + let mut sampling_params = if req.native_sampling_params_json.is_empty() { + build_sampling_params(req.temperature, sampling, decoding, stopping, response)? + } else { + serde_json::from_slice::(&req.native_sampling_params_json).map_err( + |error| { + Status::invalid_argument(format!("native_sampling_params_json is invalid: {error}")) + }, + )? + }; // Thread KVCacheParameters → SamplingParams fields. if let Some(kv) = kv { @@ -577,11 +936,160 @@ impl ResponseOpts { #[cfg(test)] mod tests { + use std::collections::BTreeMap; + + use vllm_engine_core_client::protocol::multimodal::{ + MmBatchedField, MmField, MmFieldElem, MmKwargValue, + }; use vllm_engine_core_client::protocol::output::StopReason; + use vllm_engine_core_client::protocol::tensor::{WireArrayData, WireTensor}; use vllm_text::{FinishReason, Finished, Prompt}; use super::pb::finish_info::{FinishReason as PbFinishReason, StopReason as PbStopReason}; - use super::{ResponseOpts, pb, to_finish_info, to_sequence_output, to_text_request}; + use super::{ + GrpcMultimodalInput, ResponseOpts, multimodal_input_from_request, pb, to_finish_info, + to_sequence_output, to_text_request, + }; + + fn encoded_kwargs(data: WireArrayData) -> Vec { + let kwargs = BTreeMap::from([( + "pixel_values".to_string(), + MmFieldElem { + data: Some(MmKwargValue::Tensor(WireTensor { + dtype: "uint8".to_string(), + shape: vec![1], + data, + })), + field: MmField::Batched(MmBatchedField { keep_on_cpu: false }), + }, + )]); + rmp_serde::to_vec_named(&kwargs).expect("encode multimodal kwargs") + } + + fn preprocessed_image(offset: u64, length: u64, kwargs: Vec) -> pb::MediaItem { + pb::MediaItem { + modality: pb::Modality::Image as i32, + source: Some(pb::media_item::Source::Features( + pb::PreprocessedMediaFeatures { + kwargs: Some(kwargs), + identifier: "producer-id".to_string(), + offset, + length, + mm_hash: Some("image-hash".to_string()), + is_embed: Vec::new(), + }, + )), + mime_type: String::new(), + uuid: String::new(), + } + } + + #[test] + fn preprocessed_images_decode_and_sort() { + let second = + preprocessed_image(4, 1, encoded_kwargs(WireArrayData::RawView(vec![2].into()))); + let first = + preprocessed_image(1, 2, encoded_kwargs(WireArrayData::RawView(vec![1].into()))); + + let GrpcMultimodalInput::Preprocessed(features) = + multimodal_input_from_request(vec![second, first]).expect("valid image features") + else { + panic!("expected preprocessed features"); + }; + assert_eq!(features.len(), 2); + assert_eq!(features[0].mm_position.offset, 1); + assert_eq!(features[1].mm_position.offset, 4); + assert_eq!(features[0].modality, "image"); + assert!(features[0].identifier.starts_with("grpc-mm:")); + assert_ne!(features[0].identifier, features[1].identifier); + assert_eq!(features[0].mm_hash, None); + assert_eq!(features[1].mm_hash, None); + } + + #[test] + fn dynamo_msgpack_fixture_decodes() { + let kwargs = vec![ + 0x81, 0xac, 0x70, 0x69, 0x78, 0x65, 0x6c, 0x5f, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x73, + 0x82, 0xa4, 0x64, 0x61, 0x74, 0x61, 0x93, 0xa5, 0x75, 0x69, 0x6e, 0x74, 0x38, 0x91, + 0x03, 0xc7, 0x03, 0x03, 0x01, 0x02, 0x03, 0xa5, 0x66, 0x69, 0x65, 0x6c, 0x64, 0x92, + 0xa7, 0x62, 0x61, 0x74, 0x63, 0x68, 0x65, 0x64, 0x81, 0xab, 0x6b, 0x65, 0x65, 0x70, + 0x5f, 0x6f, 0x6e, 0x5f, 0x63, 0x70, 0x75, 0xc2, + ]; + let GrpcMultimodalInput::Preprocessed(features) = + multimodal_input_from_request(vec![preprocessed_image(0, 1, kwargs)]) + .expect("Dynamo MessagePack fixture") + else { + panic!("expected preprocessed features"); + }; + + assert_eq!(features.len(), 1); + assert!(features[0].data.as_ref().unwrap().contains_key("pixel_values")); + } + + #[test] + fn preprocessed_images_reject_auxiliary_frames() { + let image = preprocessed_image(0, 1, encoded_kwargs(WireArrayData::AuxIndex(1))); + let error = match multimodal_input_from_request(vec![image]) { + Err(error) => error, + Ok(_) => panic!("auxiliary frame must be rejected"), + }; + assert!(error.message().contains("auxiliary frame 1")); + } + + #[test] + fn preprocessed_images_reject_overlapping_ranges() { + let kwargs = || encoded_kwargs(WireArrayData::RawView(vec![1].into())); + let error = match multimodal_input_from_request(vec![ + preprocessed_image(1, 2, kwargs()), + preprocessed_image(2, 2, kwargs()), + ]) { + Err(error) => error, + Ok(_) => panic!("overlapping ranges must be rejected"), + }; + assert!(error.message().contains("overlap")); + } + + #[test] + fn raw_and_preprocessed_images_cannot_be_mixed() { + let raw = pb::MediaItem { + modality: pb::Modality::Image as i32, + source: Some(pb::media_item::Source::RawBytes(vec![1])), + mime_type: "image/png".to_string(), + uuid: String::new(), + }; + let feature = + preprocessed_image(0, 1, encoded_kwargs(WireArrayData::RawView(vec![1].into()))); + let error = match multimodal_input_from_request(vec![raw, feature]) { + Err(error) => error, + Ok(_) => panic!("mixed media sources must be rejected"), + }; + assert!(error.message().contains("cannot be mixed")); + } + + #[test] + fn preprocessed_images_bound_messagepack_depth() { + let mut value = MmKwargValue::Int(1); + for _ in 0..40 { + value = MmKwargValue::List(vec![value]); + } + let kwargs = BTreeMap::from([( + "pixel_values".to_string(), + MmFieldElem { + data: Some(value), + field: MmField::Batched(MmBatchedField { keep_on_cpu: false }), + }, + )]); + let image = preprocessed_image( + 0, + 1, + rmp_serde::to_vec_named(&kwargs).expect("encode nested kwargs"), + ); + let error = match multimodal_input_from_request(vec![image]) { + Err(error) => error, + Ok(_) => panic!("deeply nested kwargs must be rejected"), + }; + assert_eq!(error.code(), tonic::Code::ResourceExhausted); + } fn base_request() -> pb::GenerateRequest { pb::GenerateRequest { @@ -610,6 +1118,60 @@ mod tests { assert_eq!(text.sampling_params.temperature, Some(0.0)); } + #[test] + fn native_sampling_params_preserve_exact_values() { + let req = pb::GenerateRequest { + temperature: Some(0.7), + sampling: Some(pb::RandomSampling { + top_k: 7, + ..Default::default() + }), + decoding: Some(pb::DecodingParameters { + repetition_penalty: 1.1, + ..Default::default() + }), + response: Some(pb::ResponseOptions { + skip_special_tokens: Some(false), + ..Default::default() + }), + native_sampling_params_json: br#"{ + "temperature": 0.0, + "top_k": 0, + "repetition_penalty": 2.5, + "skip_reading_prefix_cache": false + }"# + .to_vec(), + ..base_request() + }; + + let text = to_text_request(req, false, &["test-model".to_string()]).expect("convert ok"); + assert_eq!(text.sampling_params.temperature, Some(0.0)); + assert_eq!(text.sampling_params.top_k, Some(0)); + assert_eq!(text.sampling_params.repetition_penalty, Some(2.5)); + assert_eq!(text.sampling_params.skip_reading_prefix_cache, Some(false)); + assert!(!text.decode_options.skip_special_tokens); + } + + #[test] + fn native_sampling_params_preserve_omitted_temperature() { + let req = pb::GenerateRequest { + native_sampling_params_json: br#"{"top_k": 0}"#.to_vec(), + ..base_request() + }; + + let text = to_text_request(req, false, &["test-model".to_string()]).expect("convert ok"); + assert_eq!(text.sampling_params.temperature, None); + } + + #[test] + fn invalid_native_sampling_params_are_rejected() { + let req = pb::GenerateRequest { + native_sampling_params_json: b"not-json".to_vec(), + ..base_request() + }; + assert!(to_text_request(req, false, &["test-model".to_string()]).is_err()); + } + #[test] fn absent_seed_is_none() { let req = pb::GenerateRequest { diff --git a/rust/src/server/src/grpc/inference.rs b/rust/src/server/src/grpc/inference.rs index 8eebdb721002..dcadd440c063 100644 --- a/rust/src/server/src/grpc/inference.rs +++ b/rust/src/server/src/grpc/inference.rs @@ -76,21 +76,48 @@ impl InferenceServiceImpl { text_request.arrival_time = Some(arrival_time); text_request.data_parallel_rank = data_parallel_rank; - let media = convert::media_parts_from_request(media)?; - if !media.is_empty() { - let Prompt::TokenIds(mut token_ids) = text_request.prompt else { - return Err(Status::invalid_argument( - "multimodal gRPC requests must provide token_ids input", - )); - }; - let mm_features = self - .state - .chat - .prepare_media(media, &mut token_ids) - .await - .map_err(|error| Status::internal(error.to_report_string()))?; - text_request.prompt = Prompt::TokenIds(token_ids); - text_request.mm_features = mm_features; + match convert::multimodal_input_from_request(media)? { + convert::GrpcMultimodalInput::Raw(media) if !media.is_empty() => { + let Prompt::TokenIds(mut token_ids) = text_request.prompt else { + return Err(Status::invalid_argument( + "multimodal gRPC requests must provide token_ids input", + )); + }; + let mm_features = self + .state + .chat + .prepare_media(media, &mut token_ids) + .await + .map_err(|error| Status::internal(error.to_report_string()))?; + text_request.prompt = Prompt::TokenIds(token_ids); + text_request.mm_features = mm_features; + } + convert::GrpcMultimodalInput::Raw(_) => {} + convert::GrpcMultimodalInput::Preprocessed(features) => { + if !self.state.chat.supports_multimodal() { + return Err(Status::invalid_argument( + "model does not support multimodal input", + )); + } + let Prompt::TokenIds(token_ids) = &text_request.prompt else { + return Err(Status::invalid_argument( + "multimodal gRPC requests must provide token_ids input", + )); + }; + for (index, feature) in features.iter().enumerate() { + let in_prompt = feature + .mm_position + .offset + .checked_add(feature.mm_position.length) + .is_some_and(|end| end <= token_ids.len()); + if !in_prompt { + return Err(Status::invalid_argument(format!( + "media[{index}].features placeholder range exceeds token_ids" + ))); + } + } + text_request.mm_features = Some(features); + } } Ok(text_request) diff --git a/rust/src/server/src/grpc/tests.rs b/rust/src/server/src/grpc/tests.rs index d5ee87a87f28..1e23cd414f1a 100644 --- a/rust/src/server/src/grpc/tests.rs +++ b/rust/src/server/src/grpc/tests.rs @@ -1,6 +1,7 @@ // SPDX-License-Identifier: Apache-2.0 // SPDX-FileCopyrightText: Copyright contributors to the vLLM project +use std::collections::BTreeMap; use std::fs; use std::future::Future; use std::io; @@ -33,11 +34,15 @@ use vllm_engine_core_client::mock_engine::{ }; use vllm_engine_core_client::protocol::decode_value; use vllm_engine_core_client::protocol::handshake::{EngineCoreReadyResponse, KvEventsConfig}; +use vllm_engine_core_client::protocol::multimodal::{ + MmBatchedField, MmField, MmFieldElem, MmKwargValue, +}; use vllm_engine_core_client::protocol::output::{ EngineCoreFinishReason, EngineCoreOutput, EngineCoreOutputs, RequestBatchOutputs, UtilityCallOutput, }; use vllm_engine_core_client::protocol::request::EngineCoreRequest; +use vllm_engine_core_client::protocol::tensor::WireTensor; use vllm_engine_core_client::protocol::utility::{UtilityOutput, UtilityResultEnvelope}; use vllm_engine_core_client::test_utils::{IpcNamespace, spawn_mock_engine_task_with_ready}; use vllm_engine_core_client::{EngineCoreClient, EngineCoreClientConfig, EngineId, TransportMode}; @@ -238,6 +243,35 @@ impl ChatBackend for FakeMultimodalBackend { const QWEN_IMAGE_TOKEN_ID: u32 = 151655; const TINY_PNG_DATA_URI: &str = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII="; +fn preprocessed_image(offset: u64, length: u64) -> pb::MediaItem { + let kwargs = BTreeMap::from([( + "pixel_values".to_string(), + MmFieldElem { + data: Some(MmKwargValue::Tensor(WireTensor::from_raw( + "uint8", + vec![1], + vec![7], + ))), + field: MmField::Batched(MmBatchedField { keep_on_cpu: false }), + }, + )]); + pb::MediaItem { + modality: pb::Modality::Image as i32, + source: Some(pb::media_item::Source::Features( + pb::PreprocessedMediaFeatures { + kwargs: Some(rmp_serde::to_vec_named(&kwargs).expect("encode image feature")), + identifier: "producer-id".to_string(), + offset, + length, + mm_hash: Some("image-hash".to_string()), + is_embed: Vec::new(), + }, + )), + mime_type: String::new(), + uuid: String::new(), + } +} + fn multimodal_backend() -> Arc { let config_path = std::env::temp_dir().join(format!( "vllm-grpc-qwen-config-{}.json", @@ -728,6 +762,100 @@ async fn unary_generate_prepares_multimodal_input_for_engine_core() { server_task.abort(); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[serial] +async fn unary_generate_forwards_preprocessed_image_features() { + let (inference_service, control_service, engine_health, engine_task) = + setup_grpc_service_with_backend( + b"engine-grpc-preprocessed-image", + default_stream_output_specs(), + multimodal_backend(), + |request| { + let token_ids = request.prompt_token_ids.as_ref().expect("prompt token ids"); + assert_eq!( + token_ids, + &[11, QWEN_IMAGE_TOKEN_ID, QWEN_IMAGE_TOKEN_ID, 12] + ); + let features = request.mm_features.as_ref().expect("multimodal features"); + assert_eq!(features.len(), 1); + assert_eq!(features[0].modality, "image"); + assert_eq!(features[0].mm_position.offset, 1); + assert_eq!(features[0].mm_position.length, 2); + assert!(features[0].identifier.starts_with("grpc-mm:")); + }, + ) + .await; + let (channel, server_task) = start_grpc_test_server( + inference_service, + control_service, + engine_health, + tokio_util::sync::CancellationToken::new(), + ) + .await; + let mut client = InferenceClient::new(channel); + + client + .generate(pb::GenerateRequest { + request_id: "test-preprocessed-image".to_string(), + model: "test-model".to_string(), + prompt: Some(pb::generate_request::Prompt::TokenIds(pb::TokenIds { + ids: vec![11, QWEN_IMAGE_TOKEN_ID, QWEN_IMAGE_TOKEN_ID, 12], + })), + media: vec![preprocessed_image(1, 2)], + stopping: Some(pb::StoppingCriteria { + max_new_tokens: 10, + ..Default::default() + }), + ..Default::default() + }) + .await + .expect("preprocessed image generate"); + + engine_task.await.expect("mock engine task"); + server_task.abort(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[serial] +async fn preprocessed_image_range_must_fit_prompt() { + let (inference_service, control_service, engine_health, _engine_task) = + setup_grpc_service_with_backend( + b"engine-grpc-preprocessed-image-bounds", + default_stream_output_specs(), + multimodal_backend(), + |_| panic!("invalid request must not reach EngineCore"), + ) + .await; + let (channel, server_task) = start_grpc_test_server( + inference_service, + control_service, + engine_health, + tokio_util::sync::CancellationToken::new(), + ) + .await; + let mut client = InferenceClient::new(channel); + + let error = client + .generate(pb::GenerateRequest { + request_id: "test-preprocessed-image-bounds".to_string(), + model: "test-model".to_string(), + prompt: Some(pb::generate_request::Prompt::TokenIds(pb::TokenIds { + ids: vec![11, QWEN_IMAGE_TOKEN_ID, 12], + })), + media: vec![preprocessed_image(1, 3)], + stopping: Some(pb::StoppingCriteria { + max_new_tokens: 10, + ..Default::default() + }), + ..Default::default() + }) + .await + .expect_err("out-of-bounds image range must be rejected"); + assert_eq!(error.code(), tonic::Code::InvalidArgument); + assert!(error.message().contains("exceeds token_ids")); + server_task.abort(); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[serial] async fn streaming_generate_rejects_text_prompt_with_media() { @@ -1529,12 +1657,15 @@ async fn control_reports_server_and_model_info() { assert_eq!(server.total_kv_blocks, DEFAULT_MOCK_NUM_GPU_BLOCKS); assert_eq!(server.max_running_requests, 256); assert_eq!(server.max_batched_tokens, 8_192); + assert!(server.supports_native_sampling_params_json); + assert!(server.supports_preprocessed_media_features); let parallelism = server.parallelism.expect("parallelism metadata"); assert_eq!(parallelism.tensor_parallel_size, 1); assert_eq!(parallelism.pipeline_parallel_size, 1); assert_eq!(parallelism.data_parallel_size, 1); assert_eq!(parallelism.data_parallel_rank, 0); assert_eq!(parallelism.decode_context_parallel_size, 1); + assert_eq!(parallelism.world_size, 1); let rl = server.rl_capabilities.expect("RL capabilities"); assert!(!rl.weight_transfer_enabled); assert!(rl.weight_transfer_backend.is_empty());