[Rust Frontend][Multimodal] Accept preprocessed multimodal gRPC features (#55047)
Signed-off-by: Biswa Panda <biswa.panda@gmail.com> Signed-off-by: Bugen Zhao <i@bugenzhao.com> Co-authored-by: Bugen Zhao <i@bugenzhao.com> Co-authored-by: Codex <noreply@openai.com>
Biswa Panda committed
Sep 11, 2026 at 19:29 UTC
31759ccb6d560e150849273b47e15680a0120169
16 files changed
+986
-113
rust/Cargo.lock
+1
-1
@@ -5940,7 +5940,7 @@ dependencies = [
5940
5941
[[package]]
5942
name = "vllm-proto"
5943
-version = "0.1.0"
5943
+version = "0.2.0"
5944
dependencies = [
5945
"prost",
5946
"prost-types",
rust/Cargo.toml
+1
-1
@@ -147,7 +147,7 @@ vllm-llm = { path = "src/llm" }
147
vllm-managed-engine = { path = "src/managed-engine" }
148
vllm-metrics = { path = "src/metrics" }
149
vllm-parser = { path = "src/parser" }
150
-vllm-proto = { version = "0.1.0", path = "proto" }
150
+vllm-proto = { version = "0.2.0", path = "proto" }
151
vllm-server = { path = "src/server" }
152
vllm-text = { path = "src/text" }
153
vllm-tokenizer = { path = "src/tokenizer" }
rust/proto/Cargo.toml
+1
-1
@@ -1,6 +1,6 @@
1
[package]
2
name = "vllm-proto"
3
-version = "0.1.0"
3
+version = "0.2.0"
4
edition.workspace = true
5
license.workspace = true
6
description = "Protobuf messages and gRPC bindings for the vLLM Rust frontend"
rust/proto/README.md
+1
-1
@@ -17,7 +17,7 @@ Repository setup requires a `BUF_TOKEN` GitHub Actions secret with permission to
17
18
```toml
19
[dependencies]
20
-vllm-proto = "0.1"
20
+vllm-proto = "0.2"
21
```
22
23
For example, import `vllm_proto::inference_client::InferenceClient` or
rust/proto/inference.proto
+13
@@ -228,7 +228,20 @@ message MediaItem {
228
string url = 2; // http:// or https://
229
string data_uri = 3; // data:
230
bytes raw_bytes = 4;
231
+ PreprocessedMediaFeatures features = 7;
232
}
233
string mime_type = 5;
234
string uuid = 6;
235
}
236
+
237
+// Engine-ready multimodal features produced by a version-compatible frontend.
238
+// `kwargs` is the MessagePack wire representation of one MmKwargsItem.
239
+message PreprocessedMediaFeatures {
240
+ // Required inline payload. Cache-only references are not accepted.
241
+ optional bytes kwargs = 1;
242
+ string identifier = 2;
243
+ uint64 offset = 3;
244
+ uint64 length = 4;
245
+ optional string mm_hash = 5;
246
+ repeated bool is_embed = 6;
247
+}
rust/src/chat/src/error.rs
+3
@@ -27,6 +27,8 @@ pub enum Error {
27
UnsupportedModality { modality: String },
28
#[error("At most {limit} {modality}(s) may be provided in one prompt.")]
29
MmLimitExceeded { modality: String, limit: usize },
30
+ #[error("invalid inline multimodal features: {message}")]
31
+ InvalidPreprocessedMultimodal { message: String },
32
#[error("multimodal preprocessing error: {0}")]
33
Multimodal(#[message] String),
34
#[error("{kind} parsing is not available for model `{model_id}`")]
@@ -106,6 +108,7 @@ impl Error {
108
Self::UnsupportedMultimodalRenderer
109
| Self::UnsupportedMultimodalContent(_)
110
| Self::UnsupportedModality { .. }
111
+ | Self::InvalidPreprocessedMultimodal { .. }
112
| Self::MmLimitExceeded { .. } => true,
113
114
_ => false,
rust/src/chat/src/lib.rs
+14
-5
@@ -130,7 +130,7 @@ impl ChatRequestProcessor {
130
/// Prepare media for an already-tokenized request.
131
async fn prepare_media(
132
&self,
133
- media: Vec<MediaContentPart>,
133
+ media: multimodal::MultimodalInput,
134
token_ids: &mut Vec<u32>,
135
) -> Result<Option<MmFeatures>> {
136
if media.is_empty() {
@@ -140,8 +140,15 @@ impl ChatRequestProcessor {
140
.backend
141
.multimodal_model_info()
142
.ok_or(Error::UnsupportedMultimodalRenderer)?;
143
- let model_dtype = self.model_dtype.ok_or(Error::UnsupportedMultimodalRenderer)?;
144
- let features = info.prepare_multimodal(media, token_ids, model_dtype).await?;
143
+ let features = match media {
144
+ multimodal::MultimodalInput::Raw(parts) => {
145
+ let model_dtype = self.model_dtype.ok_or(Error::UnsupportedMultimodalRenderer)?;
146
+ info.prepare_multimodal(parts, token_ids, model_dtype).await?
147
+ }
148
+ multimodal::MultimodalInput::Preprocessed(features) => {
149
+ info.prepare_preprocessed(features, token_ids.len())?
150
+ }
151
+ };
152
Ok(Some(features))
153
}
154
@@ -286,12 +293,14 @@ impl ChatLlm {
293
}
294
295
/// Prepare media for an already-tokenized request.
296
+ /// Raw content is preprocessed; inline features are checked against model
297
+ /// capabilities, modality limits, and the final prompt positions.
298
pub async fn prepare_media(
299
&self,
291
- media: Vec<MediaContentPart>,
300
+ media: impl Into<multimodal::MultimodalInput>,
301
token_ids: &mut Vec<u32>,
302
) -> Result<Option<MmFeatures>> {
294
- self.processor.prepare_media(media, token_ids).await
303
+ self.processor.prepare_media(media.into(), token_ids).await
304
}
305
306
/// Effective tool-call parser name for this model, if parsing is enabled.
rust/src/chat/src/multimodal.rs
+38
-4
@@ -39,11 +39,14 @@ use crate::request::{ChatContent, ChatContentPart, ChatMessage, ChatRequest};
39
mod audio;
40
mod expand;
41
mod image;
42
+mod input;
43
mod item;
44
+mod preprocessed;
45
mod tensor;
46
mod video;
47
48
use self::expand::expand_prompt_token_ids;
49
+pub use self::input::MultimodalInput;
50
51
/// Resolved multimodal support for one loaded model.
52
#[derive(Clone)]
@@ -654,11 +657,16 @@ impl MultimodalModelInfo {
657
///
658
/// Modalities without a configured count are unlimited.
659
fn validate_mm_limits(&self, media_parts: &[MediaContentPart]) -> Result<()> {
660
+ self.validate_modality_limits(media_parts.iter().filter_map(media_part_limit_modality))
661
+ }
662
+
663
+ fn validate_modality_limits(
664
+ &self,
665
+ modalities: impl IntoIterator<Item = MmModality>,
666
+ ) -> Result<()> {
667
let mut counts: HashMap<MmModality, usize> = HashMap::new();
658
- for part in media_parts {
659
- if let Some(modality) = media_part_limit_modality(part) {
660
- *counts.entry(modality).or_default() += 1;
661
- }
668
+ for modality in modalities {
669
+ *counts.entry(modality).or_default() += 1;
670
}
671
672
for (modality, count) in counts {
@@ -677,6 +685,32 @@ impl MultimodalModelInfo {
685
Ok(())
686
}
687
688
+ /// Validate inline storage, batching, and placeholder ranges, then check
689
+ /// this model's supported modalities and item-count limits.
690
+ ///
691
+ /// `prompt_len` must include all expanded multimodal placeholders.
692
+ pub(crate) fn prepare_preprocessed(
693
+ &self,
694
+ mut features: MmFeatures,
695
+ prompt_len: usize,
696
+ ) -> Result<MmFeatures> {
697
+ preprocessed::validate_features(&mut features, prompt_len)?;
698
+ for feature in &features {
699
+ let supported = match feature.modality {
700
+ MmModality::Image => self.image.is_some(),
701
+ MmModality::Video => self.video.is_some(),
702
+ MmModality::Audio => self.audio.is_some(),
703
+ };
704
+ if !supported {
705
+ return Err(Error::UnsupportedModality {
706
+ modality: feature.modality.as_str().to_owned(),
707
+ });
708
+ }
709
+ }
710
+ self.validate_modality_limits(features.iter().map(|feature| feature.modality))?;
711
+ Ok(features)
712
+ }
713
+
714
/// Run media fetch, per-modality preprocessing, prompt expansion, and
715
/// feature build.
716
///
rust/src/chat/src/multimodal/input.rs
new
+29
@@ -0,0 +1,29 @@
1
+// SPDX-License-Identifier: Apache-2.0
2
+// SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
+
4
+use llm_multimodal::MediaContentPart;
5
+use vllm_engine_core_client::protocol::multimodal::MmFeatures;
6
+
7
+/// Media preparation stage supplied by a frontend handler.
8
+#[derive(Debug)]
9
+pub enum MultimodalInput {
10
+ /// Raw media to fetch and preprocess, expanding prompt placeholders as needed.
11
+ Raw(Vec<MediaContentPart>),
12
+ /// Preprocessed features to validate against storage, batching, model, and prompt constraints.
13
+ Preprocessed(MmFeatures),
14
+}
15
+
16
+impl MultimodalInput {
17
+ pub fn is_empty(&self) -> bool {
18
+ match self {
19
+ Self::Raw(parts) => parts.is_empty(),
20
+ Self::Preprocessed(features) => features.is_empty(),
21
+ }
22
+ }
23
+}
24
+
25
+impl From<Vec<MediaContentPart>> for MultimodalInput {
26
+ fn from(parts: Vec<MediaContentPart>) -> Self {
27
+ Self::Raw(parts)
28
+ }
29
+}
rust/src/chat/src/multimodal/preprocessed.rs
new
+274
@@ -0,0 +1,274 @@
1
+// SPDX-License-Identifier: Apache-2.0
2
+// SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
+
4
+use std::collections::HashMap;
5
+
6
+use vllm_engine_core_client::protocol::dtype::TensorDtype;
7
+use vllm_engine_core_client::protocol::multimodal::{
8
+ MmFeatureSpec, MmField, MmKwargValue, MmKwargsItem,
9
+};
10
+
11
+use crate::error::{
12
+ Error, Result, bail_invalid_preprocessed_multimodal, invalid_preprocessed_multimodal,
13
+};
14
+
15
+const MAX_DEPTH: usize = 32;
16
+
17
+/// Validate inline storage, placeholder ranges, and batching compatibility,
18
+/// then stably sort features by their prompt offsets.
19
+/// Encoder cache identifiers are supplied by the producer, which must bind
20
+/// them to the feature content.
21
+pub(super) fn validate_features(features: &mut [MmFeatureSpec], prompt_len: usize) -> Result<()> {
22
+ let mut fields = HashMap::new();
23
+ for feature in features.iter() {
24
+ if feature.identifier.is_empty() {
25
+ bail_invalid_preprocessed_multimodal!("feature identifier must be nonempty");
26
+ }
27
+ let data = feature.data.as_ref().ok_or_else(|| {
28
+ invalid_preprocessed_multimodal!(
29
+ "inline data is required; cache-only features are unsupported"
30
+ )
31
+ })?;
32
+ validate_kwargs(data)?;
33
+ for (key, elem) in data {
34
+ if let Some(previous) = fields.insert((feature.modality, key), &elem.field)
35
+ && !same_batching(previous, &elem.field)
36
+ {
37
+ bail_invalid_preprocessed_multimodal!(
38
+ "incompatible batching for {} field {key:?}",
39
+ feature.modality.as_str()
40
+ );
41
+ }
42
+ }
43
+ let position = &feature.mm_position;
44
+ if position.length == 0 || position.offset.checked_add(position.length).is_none() {
45
+ bail_invalid_preprocessed_multimodal!(
46
+ "placeholder range must have a positive length and fit in usize"
47
+ );
48
+ }
49
+ if position.offset + position.length > prompt_len {
50
+ bail_invalid_preprocessed_multimodal!(
51
+ "{} placeholder range exceeds prompt token IDs",
52
+ feature.modality.as_str()
53
+ );
54
+ }
55
+ if let Some(mask) = &position.is_embed {
56
+ if mask.dtype != TensorDtype::Bool || mask.shape != [position.length] {
57
+ bail_invalid_preprocessed_multimodal!(
58
+ "is_embed must be a boolean tensor of shape [placeholder length]"
59
+ );
60
+ }
61
+ mask.validate_inline()
62
+ .map_err(|message| Error::InvalidPreprocessedMultimodal { message })?;
63
+ }
64
+ }
65
+ // Mirror the Python frontend (`argsort_mm_positions`): features are
66
+ // ordered by their placeholder position in the prompt.
67
+ features.sort_by_key(|feature| feature.mm_position.offset);
68
+ Ok(())
69
+}
70
+
71
+fn validate_kwargs(kwargs: &MmKwargsItem) -> Result<()> {
72
+ if kwargs.is_empty() {
73
+ bail_invalid_preprocessed_multimodal!("kwargs must contain at least one field");
74
+ }
75
+ for (key, elem) in kwargs {
76
+ if key.is_empty() {
77
+ bail_invalid_preprocessed_multimodal!("kwargs field name must be nonempty");
78
+ }
79
+ let data = elem.data.as_ref().ok_or_else(|| {
80
+ invalid_preprocessed_multimodal!("field {key:?} requires inline data")
81
+ })?;
82
+ validate_value(data, 0)?;
83
+ if let MmField::Flat(field) = &elem.field {
84
+ match data {
85
+ MmKwargValue::Tensor(tensor) => {
86
+ let rank = tensor.shape.len() as i32;
87
+ if rank == 0 || field.dim < -rank || field.dim >= rank {
88
+ bail_invalid_preprocessed_multimodal!(
89
+ "flat field {key:?} dim is outside the tensor rank"
90
+ );
91
+ }
92
+ }
93
+ MmKwargValue::List(_) if field.dim == 0 => {}
94
+ _ => bail_invalid_preprocessed_multimodal!(
95
+ "flat field {key:?} requires a tensor or a list with dim=0"
96
+ ),
97
+ }
98
+ }
99
+ }
100
+ Ok(())
101
+}
102
+
103
+fn validate_value(value: &MmKwargValue, depth: usize) -> Result<()> {
104
+ if depth > MAX_DEPTH {
105
+ bail_invalid_preprocessed_multimodal!("kwargs nesting exceeds {MAX_DEPTH} levels");
106
+ }
107
+ match value {
108
+ MmKwargValue::Tensor(tensor) => tensor
109
+ .validate_inline()
110
+ .map_err(|message| Error::InvalidPreprocessedMultimodal { message }),
111
+ MmKwargValue::List(values) => {
112
+ for value in values {
113
+ validate_value(value, depth + 1)?;
114
+ }
115
+ Ok(())
116
+ }
117
+ MmKwargValue::Int(_) | MmKwargValue::Float(_) => Ok(()),
118
+ }
119
+}
120
+
121
+fn same_batching(left: &MmField, right: &MmField) -> bool {
122
+ // Slices and batch_size describe the producer's original batch. Python
123
+ // reduce_data combines the already-split item data without using them.
124
+ match (left, right) {
125
+ (MmField::Batched(a), MmField::Batched(b)) => a == b,
126
+ (MmField::Flat(a), MmField::Flat(b)) => a.dim == b.dim && a.keep_on_cpu == b.keep_on_cpu,
127
+ (MmField::Shared(a), MmField::Shared(b)) => a.keep_on_cpu == b.keep_on_cpu,
128
+ _ => false,
129
+ }
130
+}
131
+
132
+#[cfg(test)]
133
+mod tests {
134
+ use std::collections::BTreeMap;
135
+
136
+ use vllm_engine_core_client::protocol::multimodal::{
137
+ MmBatchedField, MmFieldElem, MmFlatField, MmModality, MmSharedField, MmSlice,
138
+ PlaceholderRange, SliceSpec,
139
+ };
140
+ use vllm_engine_core_client::protocol::tensor::{WireArrayData, WireTensor};
141
+
142
+ use super::*;
143
+ use crate::multimodal::tests::qwen3_vl_info;
144
+
145
+ fn feature(offset: usize, field: MmField) -> MmFeatureSpec {
146
+ MmFeatureSpec {
147
+ data: Some(BTreeMap::from([(
148
+ "pixels".to_owned(),
149
+ MmFieldElem {
150
+ data: Some(MmKwargValue::Tensor(
151
+ WireTensor::from_f32(vec![2], vec![1., 2.]).unwrap(),
152
+ )),
153
+ field,
154
+ },
155
+ )])),
156
+ modality: MmModality::Image,
157
+ identifier: format!("image-{offset}"),
158
+ mm_position: PlaceholderRange {
159
+ offset,
160
+ length: 2,
161
+ is_embed: None,
162
+ },
163
+ mm_hash: Some("processor-hash".to_owned()),
164
+ }
165
+ }
166
+
167
+ fn batched() -> MmField {
168
+ MmField::Batched(MmBatchedField { keep_on_cpu: false })
169
+ }
170
+
171
+ #[test]
172
+ fn inline_features_preserve_overlaps_and_stably_sort_prompt_positions() {
173
+ let info = qwen3_vl_info();
174
+ let mut a = feature(2, batched());
175
+ a.mm_position.is_embed = Some(WireTensor::from_bool(vec![2], vec![true, false]).unwrap());
176
+ let b = feature(1, batched());
177
+ let mut c = feature(2, batched());
178
+ c.modality = MmModality::Video;
179
+ let expected = vec![b.clone(), a.clone(), c.clone()];
180
+ let features = info.prepare_preprocessed(vec![a, b, c], 4).unwrap();
181
+ assert_eq!(features, expected);
182
+ }
183
+
184
+ #[test]
185
+ fn independently_preprocessed_items_keep_their_original_batch_metadata() {
186
+ let info = qwen3_vl_info();
187
+ let flat = |length| {
188
+ MmField::Flat(MmFlatField {
189
+ slices: vec![MmSlice::Slice(SliceSpec {
190
+ start: Some(0),
191
+ stop: Some(length),
192
+ step: None,
193
+ })],
194
+ dim: 0,
195
+ keep_on_cpu: false,
196
+ })
197
+ };
198
+ let shared = |batch_size| {
199
+ MmField::Shared(MmSharedField {
200
+ batch_size,
201
+ keep_on_cpu: true,
202
+ })
203
+ };
204
+ for (a, b) in [(flat(2), flat(7)), (shared(1), shared(3))] {
205
+ let features = vec![feature(0, a), feature(2, b)];
206
+ let result = info.prepare_preprocessed(features.clone(), 4).unwrap();
207
+ assert_eq!(result, features);
208
+ }
209
+ }
210
+
211
+ #[test]
212
+ fn inline_features_reject_cache_references_and_invalid_tensor_storage() {
213
+ let info = qwen3_vl_info();
214
+ let original = feature(0, batched());
215
+ let mut cached = original.clone();
216
+ cached.data = None;
217
+ assert!(info.prepare_preprocessed(vec![cached], 4).is_err());
218
+ let mut field_cached = original.clone();
219
+ field_cached.data.as_mut().unwrap().get_mut("pixels").unwrap().data = None;
220
+ assert!(info.prepare_preprocessed(vec![field_cached], 4).is_err());
221
+ for tensor in [
222
+ WireTensor {
223
+ dtype: TensorDtype::F32,
224
+ shape: vec![2],
225
+ data: WireArrayData::AuxIndex(1),
226
+ },
227
+ WireTensor::from_raw(TensorDtype::F32, vec![2], vec![0; 4]),
228
+ WireTensor::from_raw(TensorDtype::F32, vec![usize::MAX, 2], vec![]),
229
+ ] {
230
+ let mut item = original.clone();
231
+ item.data.as_mut().unwrap().get_mut("pixels").unwrap().data =
232
+ Some(MmKwargValue::Tensor(tensor));
233
+ assert!(info.prepare_preprocessed(vec![item], 4).is_err());
234
+ }
235
+ }
236
+
237
+ #[test]
238
+ fn inline_features_validate_placeholder_and_mask_bounds() {
239
+ let info = qwen3_vl_info();
240
+ for (offset, length) in [(usize::MAX, 2), (0, 0)] {
241
+ let mut item = feature(offset, batched());
242
+ item.mm_position.length = length;
243
+ assert!(info.prepare_preprocessed(vec![item], 4).is_err());
244
+ }
245
+ assert!(info.prepare_preprocessed(vec![feature(2, batched())], 3).is_err());
246
+ for mask in [
247
+ WireTensor::from_bool(vec![1], vec![true]).unwrap(),
248
+ WireTensor::from_u32(vec![2], vec![1, 0]).unwrap(),
249
+ ] {
250
+ let mut item = feature(0, batched());
251
+ item.mm_position.is_embed = Some(mask);
252
+ assert!(info.prepare_preprocessed(vec![item], 4).is_err());
253
+ }
254
+ }
255
+
256
+ #[test]
257
+ fn inline_features_reject_incompatible_batching_and_flat_axes() {
258
+ let info = qwen3_vl_info();
259
+ let flat = MmField::Flat(MmFlatField {
260
+ slices: vec![],
261
+ dim: 2,
262
+ keep_on_cpu: false,
263
+ });
264
+ assert!(info.prepare_preprocessed(vec![feature(0, flat)], 4).is_err());
265
+ let shared = MmField::Shared(MmSharedField {
266
+ batch_size: 1,
267
+ keep_on_cpu: false,
268
+ });
269
+ assert!(
270
+ info.prepare_preprocessed(vec![feature(0, batched()), feature(2, shared)], 4)
271
+ .is_err()
272
+ );
273
+ }
274
+}
rust/src/server/Cargo.toml
+1
-1
@@ -20,6 +20,7 @@ libc.workspace = true
20
llm-multimodal.workspace = true
21
openssl.workspace = true
22
prost-types.workspace = true
23
+rmp-serde.workspace = true
24
rmpv.workspace = true
25
serde.workspace = true
26
serde_json.workspace = true
@@ -58,7 +59,6 @@ async-openai = { workspace = true, features = ["full"] }
59
bytes.workspace = true
60
clap.workspace = true
61
expect-test.workspace = true
61
-rmp-serde.workspace = true
62
serial_test.workspace = true
63
tempfile.workspace = true
64
tokio = { workspace = true, features = ["test-util"] }
rust/src/server/src/grpc/convert.rs
-91
@@ -5,9 +5,7 @@
5
//! request/response types.
6
7
use tonic::Status;
8
-use url::Url;
8
use uuid::Uuid;
10
-use vllm_chat::MediaContentPart;
9
use vllm_engine_core_client::protocol::output::StopReason;
10
use vllm_engine_core_client::protocol::structured_outputs::StructuredOutputsParams;
11
use vllm_text::{
@@ -17,95 +15,6 @@ use vllm_text::{
15
16
use super::pb;
17
20
-pub fn media_parts_from_request(
21
- media: Vec<pb::MediaItem>,
22
-) -> Result<Vec<MediaContentPart>, Status> {
23
- let mut parts = Vec::with_capacity(media.len());
24
- for (index, item) in media.into_iter().enumerate() {
25
- let modality = item.modality();
26
- if modality == pb::Modality::Unspecified {
27
- return Err(Status::invalid_argument(format!(
28
- "media[{index}].modality is required"
29
- )));
30
- }
31
- let uuid = (!item.uuid.is_empty()).then_some(item.uuid);
32
- let mime_type = (!item.mime_type.is_empty()).then_some(item.mime_type);
33
- let source = item.source.ok_or_else(|| {
34
- Status::invalid_argument(format!("media[{index}].source is required"))
35
- })?;
36
- match &source {
37
- pb::media_item::Source::Url(url) => {
38
- validate_media_uri(index, "url", url, &["http", "https"])?;
39
- }
40
- pb::media_item::Source::DataUri(uri) => {
41
- validate_media_uri(index, "data_uri", uri, &["data"])?;
42
- }
43
- pb::media_item::Source::RawBytes(_) => {}
44
- }
45
- let part = match (modality, source) {
46
- (
47
- pb::Modality::Image,
48
- pb::media_item::Source::Url(url) | pb::media_item::Source::DataUri(url),
49
- ) => MediaContentPart::ImageUrl {
50
- url,
51
- detail: None,
52
- uuid,
53
- },
54
- (pb::Modality::Image, pb::media_item::Source::RawBytes(data)) => {
55
- MediaContentPart::ImageData {
56
- data,
57
- mime_type,
58
- uuid,
59
- detail: None,
60
- }
61
- }
62
- (
63
- pb::Modality::Video,
64
- pb::media_item::Source::Url(url) | pb::media_item::Source::DataUri(url),
65
- ) => MediaContentPart::VideoUrl { url, uuid },
66
- (pb::Modality::Video, pb::media_item::Source::RawBytes(data)) => {
67
- MediaContentPart::VideoData {
68
- data,
69
- mime_type,
70
- uuid,
71
- }
72
- }
73
- (
74
- pb::Modality::Audio,
75
- pb::media_item::Source::Url(url) | pb::media_item::Source::DataUri(url),
76
- ) => MediaContentPart::AudioUrl { url, uuid },
77
- (pb::Modality::Audio, pb::media_item::Source::RawBytes(data)) => {
78
- MediaContentPart::AudioData {
79
- data,
80
- mime_type,
81
- uuid,
82
- }
83
- }
84
- (pb::Modality::Unspecified, _) => unreachable!("modality validated above"),
85
- };
86
- parts.push(part);
87
- }
88
- Ok(parts)
89
-}
90
-
91
-fn validate_media_uri(
92
- index: usize,
93
- field: &str,
94
- value: &str,
95
- allowed_schemes: &[&str],
96
-) -> Result<(), Status> {
97
- let uri = Url::parse(value).map_err(|_| {
98
- Status::invalid_argument(format!("media[{index}].{field} is not a valid URI"))
99
- })?;
100
- if !allowed_schemes.contains(&uri.scheme()) {
101
- return Err(Status::invalid_argument(format!(
102
- "media[{index}].{field} must use the {} scheme",
103
- allowed_schemes.join(" or ")
104
- )));
105
- }
106
- Ok(())
107
-}
108
-
18
// ========================================================================================
19
// Request conversion
20
// ========================================================================================
rust/src/server/src/grpc/inference.rs
+11
-7
@@ -149,19 +149,23 @@ impl InferenceServiceImpl {
149
})?);
150
}
151
152
- let media = convert::media_parts_from_request(media)?;
152
+ let media = super::media::from_proto(media)?;
153
if !media.is_empty() {
154
let Prompt::TokenIds(mut token_ids) = text_request.prompt else {
155
return Err(Status::invalid_argument(
156
"multimodal gRPC requests must provide token_ids input",
157
));
158
};
159
- let mm_features = self
160
- .state
161
- .chat
162
- .prepare_media(media, &mut token_ids)
163
- .await
164
- .map_err(|error| Status::internal(error.to_report_string()))?;
159
+ let mm_features =
160
+ self.state.chat.prepare_media(media, &mut token_ids).await.map_err(
161
+ |error| {
162
+ if error.is_request_validation_error() {
163
+ Status::invalid_argument(error.to_report_string())
164
+ } else {
165
+ Status::internal(error.to_report_string())
166
+ }
167
+ },
168
+ )?;
169
text_request.prompt = Prompt::TokenIds(token_ids);
170
text_request.mm_features = mm_features;
171
}
rust/src/server/src/grpc/media.rs
new
+341
@@ -0,0 +1,341 @@
1
+// SPDX-License-Identifier: Apache-2.0
2
+// SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
+
4
+//! Protobuf media adaptation into the shared multimodal preparation boundary.
5
+
6
+use std::io::Cursor;
7
+
8
+use serde::Deserialize as _;
9
+use thiserror_ext::AsReport as _;
10
+use tonic::Status;
11
+use url::Url;
12
+use vllm_chat::MediaContentPart;
13
+use vllm_chat::multimodal::MultimodalInput;
14
+use vllm_engine_core_client::protocol::multimodal::{
15
+ MmFeatureSpec, MmKwargsItem, MmModality, PlaceholderRange,
16
+};
17
+use vllm_engine_core_client::protocol::tensor::WireTensor;
18
+
19
+use super::pb;
20
+
21
+/// Maximum total serialized kwargs size accepted in one gRPC request.
22
+const MAX_INLINE_MM_BYTES: usize = 16 * 1024 * 1024;
23
+const MAX_MSGPACK_DEPTH: usize = 32;
24
+
25
+/// Adapt a request containing either raw media or preprocessed features.
26
+///
27
+/// Require a single input kind and bound the total kwargs size before decoding.
28
+/// Shared media preparation validates features, model support, item limits, and prompt bounds.
29
+pub(super) fn from_proto(media: Vec<pb::MediaItem>) -> Result<MultimodalInput, Status> {
30
+ if !media
31
+ .iter()
32
+ .any(|item| matches!(item.source, Some(pb::media_item::Source::Features(_))))
33
+ {
34
+ return raw_parts_from_proto(media).map(MultimodalInput::Raw);
35
+ }
36
+ let mut encoded_bytes = 0usize;
37
+ for item in &media {
38
+ let Some(pb::media_item::Source::Features(feature)) = &item.source else {
39
+ return Err(mixed_media_error());
40
+ };
41
+ encoded_bytes = encoded_bytes.saturating_add(feature.kwargs.as_ref().map_or(0, Vec::len));
42
+ }
43
+ if encoded_bytes > MAX_INLINE_MM_BYTES {
44
+ return Err(Status::resource_exhausted(format!(
45
+ "inline multimodal payload exceeds {MAX_INLINE_MM_BYTES} bytes"
46
+ )));
47
+ }
48
+ let features = media
49
+ .into_iter()
50
+ .enumerate()
51
+ .map(|(index, item)| {
52
+ feature_from_proto(item).map_err(|error| {
53
+ Status::new(error.code(), format!("media[{index}]: {}", error.message()))
54
+ })
55
+ })
56
+ .collect::<Result<Vec<_>, _>>()?;
57
+ Ok(MultimodalInput::Preprocessed(features))
58
+}
59
+
60
+fn feature_from_proto(item: pb::MediaItem) -> Result<MmFeatureSpec, Status> {
61
+ let modality = match item.modality() {
62
+ pb::Modality::Image => MmModality::Image,
63
+ pb::Modality::Video => MmModality::Video,
64
+ pb::Modality::Audio => MmModality::Audio,
65
+ pb::Modality::Unspecified => return Err(Status::invalid_argument("modality is required")),
66
+ };
67
+ let Some(pb::media_item::Source::Features(feature)) = item.source else {
68
+ return Err(mixed_media_error());
69
+ };
70
+ let bytes = feature.kwargs.ok_or_else(|| {
71
+ Status::invalid_argument("features.kwargs is required; cache-only features are unsupported")
72
+ })?;
73
+ let offset = usize::try_from(feature.offset)
74
+ .map_err(|_| Status::invalid_argument("features.offset exceeds platform limits"))?;
75
+ let length = usize::try_from(feature.length)
76
+ .map_err(|_| Status::invalid_argument("features.length exceeds platform limits"))?;
77
+ let is_embed = if feature.is_embed.is_empty() {
78
+ None
79
+ } else {
80
+ Some(
81
+ WireTensor::from_bool(vec![length], feature.is_embed)
82
+ .map_err(Status::invalid_argument)?,
83
+ )
84
+ };
85
+ Ok(MmFeatureSpec {
86
+ modality,
87
+ data: Some(decode_inline_mm_kwargs(&bytes)?),
88
+ identifier: feature.identifier,
89
+ mm_hash: feature.mm_hash,
90
+ mm_position: PlaceholderRange {
91
+ offset,
92
+ length,
93
+ is_embed,
94
+ },
95
+ })
96
+}
97
+
98
+/// Decode one inline item using the engine protocol's typed serde schema.
99
+fn decode_inline_mm_kwargs(bytes: &[u8]) -> Result<MmKwargsItem, Status> {
100
+ let mut decoder = rmp_serde::Deserializer::new(Cursor::new(bytes));
101
+ decoder.set_max_depth(MAX_MSGPACK_DEPTH);
102
+ let kwargs = MmKwargsItem::deserialize(&mut decoder).map_err(|error| {
103
+ Status::invalid_argument(format!(
104
+ "invalid multimodal kwargs MessagePack: {}",
105
+ error.as_report()
106
+ ))
107
+ })?;
108
+ if decoder.position() != bytes.len() as u64 {
109
+ return Err(Status::invalid_argument(
110
+ "kwargs contains trailing MessagePack data",
111
+ ));
112
+ }
113
+ Ok(kwargs)
114
+}
115
+
116
+fn mixed_media_error() -> Status {
117
+ Status::invalid_argument("raw media and preprocessed media features cannot be mixed")
118
+}
119
+
120
+fn raw_parts_from_proto(media: Vec<pb::MediaItem>) -> Result<Vec<MediaContentPart>, Status> {
121
+ let mut parts = Vec::with_capacity(media.len());
122
+ for (index, item) in media.into_iter().enumerate() {
123
+ let modality = item.modality();
124
+ if modality == pb::Modality::Unspecified {
125
+ return Err(Status::invalid_argument(format!(
126
+ "media[{index}].modality is required"
127
+ )));
128
+ }
129
+ let uuid = (!item.uuid.is_empty()).then_some(item.uuid);
130
+ let mime_type = (!item.mime_type.is_empty()).then_some(item.mime_type);
131
+ let source = item.source.ok_or_else(|| {
132
+ Status::invalid_argument(format!("media[{index}].source is required"))
133
+ })?;
134
+ match &source {
135
+ pb::media_item::Source::Url(url) => {
136
+ validate_media_uri(index, "url", url, &["http", "https"])?;
137
+ }
138
+ pb::media_item::Source::DataUri(uri) => {
139
+ validate_media_uri(index, "data_uri", uri, &["data"])?;
140
+ }
141
+ pb::media_item::Source::RawBytes(_) => {}
142
+ pb::media_item::Source::Features(_) => return Err(mixed_media_error()),
143
+ }
144
+ let part = match (modality, source) {
145
+ (
146
+ pb::Modality::Image,
147
+ pb::media_item::Source::Url(url) | pb::media_item::Source::DataUri(url),
148
+ ) => MediaContentPart::ImageUrl {
149
+ url,
150
+ detail: None,
151
+ uuid,
152
+ },
153
+ (pb::Modality::Image, pb::media_item::Source::RawBytes(data)) => {
154
+ MediaContentPart::ImageData {
155
+ data,
156
+ mime_type,
157
+ uuid,
158
+ detail: None,
159
+ }
160
+ }
161
+ (
162
+ pb::Modality::Video,
163
+ pb::media_item::Source::Url(url) | pb::media_item::Source::DataUri(url),
164
+ ) => MediaContentPart::VideoUrl { url, uuid },
165
+ (pb::Modality::Video, pb::media_item::Source::RawBytes(data)) => {
166
+ MediaContentPart::VideoData {
167
+ data,
168
+ mime_type,
169
+ uuid,
170
+ }
171
+ }
172
+ (
173
+ pb::Modality::Audio,
174
+ pb::media_item::Source::Url(url) | pb::media_item::Source::DataUri(url),
175
+ ) => MediaContentPart::AudioUrl { url, uuid },
176
+ (pb::Modality::Audio, pb::media_item::Source::RawBytes(data)) => {
177
+ MediaContentPart::AudioData {
178
+ data,
179
+ mime_type,
180
+ uuid,
181
+ }
182
+ }
183
+ (_, pb::media_item::Source::Features(_)) => return Err(mixed_media_error()),
184
+ (pb::Modality::Unspecified, _) => unreachable!("modality validated above"),
185
+ };
186
+ parts.push(part);
187
+ }
188
+ Ok(parts)
189
+}
190
+
191
+fn validate_media_uri(
192
+ index: usize,
193
+ field: &str,
194
+ value: &str,
195
+ allowed_schemes: &[&str],
196
+) -> Result<(), Status> {
197
+ let uri = Url::parse(value).map_err(|_| {
198
+ Status::invalid_argument(format!("media[{index}].{field} is not a valid URI"))
199
+ })?;
200
+ if !allowed_schemes.contains(&uri.scheme()) {
201
+ return Err(Status::invalid_argument(format!(
202
+ "media[{index}].{field} must use the {} scheme",
203
+ allowed_schemes.join(" or ")
204
+ )));
205
+ }
206
+ Ok(())
207
+}
208
+
209
+#[cfg(test)]
210
+mod tests {
211
+ use std::collections::BTreeMap;
212
+
213
+ use super::*;
214
+ use vllm_engine_core_client::protocol::multimodal::{
215
+ MmBatchedField, MmField, MmFieldElem, MmFlatField, MmKwargValue, MmSlice, SliceSpec,
216
+ };
217
+
218
+ fn preprocessed() -> pb::MediaItem {
219
+ let kwargs = BTreeMap::from([(
220
+ "x",
221
+ MmFieldElem {
222
+ data: Some(MmKwargValue::Int(1)),
223
+ field: MmField::Batched(MmBatchedField { keep_on_cpu: false }),
224
+ },
225
+ )]);
226
+ pb::MediaItem {
227
+ modality: pb::Modality::Image as i32,
228
+ source: Some(pb::media_item::Source::Features(
229
+ pb::PreprocessedMediaFeatures {
230
+ kwargs: Some(rmp_serde::to_vec_named(&kwargs).unwrap()),
231
+ identifier: "image-id".to_owned(),
232
+ length: 1,
233
+ ..Default::default()
234
+ },
235
+ )),
236
+ ..Default::default()
237
+ }
238
+ }
239
+
240
+ #[test]
241
+ fn mixed_media_is_rejected_in_both_orders() {
242
+ let raw = pb::MediaItem {
243
+ modality: pb::Modality::Image as i32,
244
+ source: Some(pb::media_item::Source::RawBytes(vec![1])),
245
+ ..Default::default()
246
+ };
247
+ for media in [vec![raw.clone(), preprocessed()], vec![preprocessed(), raw]] {
248
+ assert_eq!(
249
+ from_proto(media).unwrap_err().code(),
250
+ tonic::Code::InvalidArgument
251
+ );
252
+ }
253
+ }
254
+
255
+ #[test]
256
+ fn protobuf_features_require_modality_inline_kwargs_and_a_matching_mask() {
257
+ let mut missing_modality = preprocessed();
258
+ missing_modality.modality = 99;
259
+ let mut missing_kwargs = preprocessed();
260
+ let Some(pb::media_item::Source::Features(feature)) = &mut missing_kwargs.source else {
261
+ unreachable!()
262
+ };
263
+ feature.kwargs = None;
264
+ let mut invalid_mask = preprocessed();
265
+ let Some(pb::media_item::Source::Features(feature)) = &mut invalid_mask.source else {
266
+ unreachable!()
267
+ };
268
+ feature.is_embed = vec![true, false];
269
+ for media in [missing_modality, missing_kwargs, invalid_mask] {
270
+ let error = from_proto(vec![media]).unwrap_err();
271
+ assert_eq!(error.code(), tonic::Code::InvalidArgument);
272
+ assert!(error.message().starts_with("media[0]:"));
273
+ }
274
+ }
275
+
276
+ #[test]
277
+ fn aggregate_payload_limit_is_checked_before_decoding_items() {
278
+ let mut item = preprocessed();
279
+ let Some(pb::media_item::Source::Features(feature)) = &mut item.source else {
280
+ unreachable!()
281
+ };
282
+ feature.kwargs = Some(vec![0; MAX_INLINE_MM_BYTES / 2 + 1]);
283
+ let error = from_proto(vec![item.clone(), item]).unwrap_err();
284
+ assert_eq!(error.code(), tonic::Code::ResourceExhausted);
285
+ }
286
+
287
+ #[test]
288
+ fn kwargs_decoder_rejects_trailing_truncated_and_deeply_nested_input() {
289
+ let kwargs = BTreeMap::from([(
290
+ "pixels".to_owned(),
291
+ MmFieldElem {
292
+ data: Some(MmKwargValue::Int(1)),
293
+ field: MmField::Batched(MmBatchedField { keep_on_cpu: false }),
294
+ },
295
+ )]);
296
+ let valid = rmp_serde::to_vec_named(&kwargs).unwrap();
297
+ assert_eq!(decode_inline_mm_kwargs(&valid).unwrap(), kwargs);
298
+ let mut trailing = valid.clone();
299
+ trailing.push(0);
300
+ let mut nested_kwargs = kwargs.clone();
301
+ let mut value = MmKwargValue::Int(1);
302
+ for _ in 0..64 {
303
+ value = MmKwargValue::List(vec![value]);
304
+ }
305
+ nested_kwargs.get_mut("pixels").unwrap().data = Some(value);
306
+ let nested = rmp_serde::to_vec_named(&nested_kwargs).unwrap();
307
+ // A declared array length must not drive a proportional allocation.
308
+ let huge_array = vec![
309
+ 0x81, 0xa1, b'x', 0x81, 0xa4, b'd', b'a', b't', b'a', 0xdd, 0xff, 0xff, 0xff, 0xff,
310
+ ];
311
+ for bytes in [&valid[..valid.len() - 1], &trailing, &nested, &huge_array] {
312
+ assert!(decode_inline_mm_kwargs(bytes).is_err());
313
+ }
314
+ }
315
+
316
+ #[test]
317
+ fn kwargs_decode_python_encoder_tensor_and_factory_tuple() {
318
+ // MsgpackEncoder(size_threshold=1 << 30), vLLM 6cbb3c154e:
319
+ // pixel_values = tensor([1., 2.]), MultiModalFlatField(slices=[slice(0, 2)]).
320
+ let bytes = b"\x81\xacpixel_values\x82\xa4data\x93\xa7float32\x91\x02\xd7\x03\
321
+ \x00\x00\x80\x3f\x00\x00\x00\x40\
322
+ \xa5field\x92\xa4flat\x83\xabkeep_on_cpu\xc2\
323
+ \xa6slices\x91\x93\x00\x02\xc0\xa3dim\x00";
324
+ let kwargs = decode_inline_mm_kwargs(bytes).unwrap();
325
+ let expected = MmFieldElem {
326
+ data: Some(MmKwargValue::Tensor(
327
+ WireTensor::from_f32(vec![2], vec![1., 2.]).unwrap(),
328
+ )),
329
+ field: MmField::Flat(MmFlatField {
330
+ slices: vec![MmSlice::Slice(SliceSpec {
331
+ start: Some(0),
332
+ stop: Some(2),
333
+ step: None,
334
+ })],
335
+ dim: 0,
336
+ keep_on_cpu: false,
337
+ }),
338
+ };
339
+ assert_eq!(kwargs["pixel_values"], expected);
340
+ }
341
+}
rust/src/server/src/grpc/mod.rs
+1
@@ -7,6 +7,7 @@ mod control;
7
mod convert;
8
mod health;
9
mod inference;
10
+mod media;
11
12
/// Generated protobuf/gRPC types for the `vllm` package.
13
pub use vllm_proto as pb;
rust/src/server/src/grpc/tests.rs
+257
-1
@@ -1,6 +1,7 @@
1
// SPDX-License-Identifier: Apache-2.0
2
// SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
4
+use std::collections::BTreeMap;
5
use std::fs;
6
use std::future::Future;
7
use std::io;
@@ -23,6 +24,7 @@ use tonic_health::pb::health_check_response::ServingStatus as HealthServingStatu
24
use tonic_health::pb::health_client::HealthClient;
25
use tonic_health::server::health_reporter;
26
use tower::service_fn;
27
+use vllm_chat::multimodal::{MmLimitPerPrompt, MmLimitSpec, MmModality};
28
use vllm_chat::{
29
ChatBackend, ChatLlm, ChatRenderer, ChatRequest, ChatTextBackend, DefaultChatOutputProcessor,
30
DynChatOutputProcessor, DynChatRenderer, NewChatOutputProcessorOptions, RenderedPrompt,
@@ -32,12 +34,17 @@ use vllm_engine_core_client::mock_engine::{
34
default_ready_response,
35
};
36
use vllm_engine_core_client::protocol::decode_value;
37
+use vllm_engine_core_client::protocol::dtype::TensorDtype;
38
use vllm_engine_core_client::protocol::handshake::{EngineCoreReadyResponse, KvEventsConfig};
39
+use vllm_engine_core_client::protocol::multimodal::{
40
+ MmBatchedField, MmField, MmFieldElem, MmKwargValue,
41
+};
42
use vllm_engine_core_client::protocol::output::{
43
EngineCoreFinishReason, EngineCoreOutput, EngineCoreOutputs, RequestBatchOutputs,
44
UtilityCallOutput,
45
};
46
use vllm_engine_core_client::protocol::request::EngineCoreRequest;
47
+use vllm_engine_core_client::protocol::tensor::WireTensor;
48
use vllm_engine_core_client::protocol::utility::{UtilityOutput, UtilityResultEnvelope};
49
use vllm_engine_core_client::test_utils::{IpcNamespace, spawn_mock_engine_task_with_ready};
50
use vllm_engine_core_client::{EngineCoreClient, EngineCoreClientConfig, EngineId, TransportMode};
@@ -292,6 +299,12 @@ const QWEN_IMAGE_TOKEN_ID: u32 = 151655;
299
const TINY_PNG_DATA_URI: &str = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=";
300
301
fn multimodal_backend() -> Arc<dyn ChatTextBackend> {
302
+ multimodal_backend_with_limits(Default::default())
303
+}
304
+
305
+fn multimodal_backend_with_limits(
306
+ limit_mm_per_prompt: MmLimitPerPrompt,
307
+) -> Arc<dyn ChatTextBackend> {
308
let config_path = std::env::temp_dir().join(format!(
309
"vllm-grpc-qwen-config-{}.json",
310
uuid::Uuid::new_v4()
@@ -309,7 +322,7 @@ fn multimodal_backend() -> Arc<dyn ChatTextBackend> {
322
..Default::default()
323
},
324
Arc::new(TestTokenizer::new().with_regular_token("<|image_pad|>", QWEN_IMAGE_TOKEN_ID)),
312
- Default::default(),
325
+ limit_mm_per_prompt,
326
)
327
.expect("load multimodal info")
328
.expect("qwen multimodal info is registered");
@@ -877,6 +890,249 @@ async fn unary_generate_prepares_multimodal_input_for_engine_core() {
890
server_task.abort();
891
}
892
893
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
894
+#[serial]
895
+async fn generate_forwards_preprocessed_multimodal_features() {
896
+ for stream in [false, true] {
897
+ let kwargs = BTreeMap::from([(
898
+ "pixel_values".to_string(),
899
+ MmFieldElem {
900
+ data: Some(MmKwargValue::Tensor(WireTensor::from_raw(
901
+ TensorDtype::U8,
902
+ vec![3],
903
+ vec![1, 2, 3],
904
+ ))),
905
+ field: MmField::Batched(MmBatchedField { keep_on_cpu: false }),
906
+ },
907
+ )]);
908
+ let encoded_kwargs = rmp_serde::to_vec_named(&kwargs).expect("encode multimodal kwargs");
909
+ let expected_kwargs = kwargs.clone();
910
+ let (inference_service, control_service, engine_health, engine_task) =
911
+ setup_grpc_service_with_backend(
912
+ b"engine-grpc-preprocessed-multimodal",
913
+ default_stream_output_specs(),
914
+ multimodal_backend(),
915
+ move |request| {
916
+ assert_eq!(
917
+ request.prompt_token_ids.as_deref(),
918
+ Some(&[11, 99, 99, 12][..])
919
+ );
920
+ let features = request.mm_features.as_ref().expect("multimodal features");
921
+ assert_eq!(features.len(), 1);
922
+ let feature = &features[0];
923
+ assert_eq!(feature.modality.as_str(), "image");
924
+ assert_eq!(feature.identifier, "image-hash-a");
925
+ assert_eq!(feature.mm_hash.as_deref(), Some("image-hash-a"));
926
+ assert_eq!(feature.mm_position.offset, 1);
927
+ assert_eq!(feature.mm_position.length, 2);
928
+ assert_eq!(feature.data.as_ref(), Some(&expected_kwargs));
929
+ assert_eq!(
930
+ feature.mm_position.is_embed,
931
+ Some(WireTensor::from_bool(vec![2], vec![true, false]).unwrap())
932
+ );
933
+ },
934
+ )
935
+ .await;
936
+ let (channel, server_task) = start_grpc_test_server(
937
+ inference_service,
938
+ control_service,
939
+ engine_health,
940
+ tokio_util::sync::CancellationToken::new(),
941
+ )
942
+ .await;
943
+ let mut client = InferenceClient::new(channel);
944
+
945
+ let request = pb::GenerateRequest {
946
+ request_id: "test-preprocessed-multimodal".to_string(),
947
+ model: "test-model".to_string(),
948
+ prompt: Some(pb::generate_request::Prompt::TokenIds(pb::TokenIds {
949
+ ids: vec![11, 99, 99, 12],
950
+ })),
951
+ media: vec![pb::MediaItem {
952
+ modality: pb::Modality::Image as i32,
953
+ source: Some(pb::media_item::Source::Features(
954
+ pb::PreprocessedMediaFeatures {
955
+ kwargs: Some(encoded_kwargs),
956
+ identifier: "image-hash-a".to_string(),
957
+ offset: 1,
958
+ length: 2,
959
+ mm_hash: Some("image-hash-a".to_string()),
960
+ is_embed: vec![true, false],
961
+ },
962
+ )),
963
+ mime_type: String::new(),
964
+ uuid: String::new(),
965
+ }],
966
+ stopping: Some(pb::StoppingCriteria {
967
+ max_new_tokens: 10,
968
+ ..Default::default()
969
+ }),
970
+ ..Default::default()
971
+ };
972
+ if stream {
973
+ let mut response =
974
+ client.generate_stream(request).await.expect("preprocessed stream").into_inner();
975
+ while response.message().await.expect("stream message").is_some() {}
976
+ } else {
977
+ client.generate(request).await.expect("preprocessed generate");
978
+ }
979
+
980
+ engine_task.await.expect("mock engine task");
981
+ server_task.abort();
982
+ }
983
+}
984
+
985
+fn preprocessed_grpc_media(
986
+ modality: pb::Modality,
987
+ identifier: &str,
988
+ offset: u64,
989
+ length: u64,
990
+) -> pb::MediaItem {
991
+ let kwargs = BTreeMap::from([(
992
+ "pixel_values".to_string(),
993
+ MmFieldElem {
994
+ data: Some(MmKwargValue::Tensor(WireTensor::from_raw(
995
+ TensorDtype::U8,
996
+ vec![1],
997
+ vec![1],
998
+ ))),
999
+ field: MmField::Batched(MmBatchedField { keep_on_cpu: false }),
1000
+ },
1001
+ )]);
1002
+ pb::MediaItem {
1003
+ modality: modality as i32,
1004
+ source: Some(pb::media_item::Source::Features(
1005
+ pb::PreprocessedMediaFeatures {
1006
+ kwargs: Some(rmp_serde::to_vec_named(&kwargs).expect("encode kwargs")),
1007
+ identifier: identifier.to_string(),
1008
+ offset,
1009
+ length,
1010
+ mm_hash: Some(identifier.to_string()),
1011
+ is_embed: Vec::new(),
1012
+ },
1013
+ )),
1014
+ mime_type: String::new(),
1015
+ uuid: String::new(),
1016
+ }
1017
+}
1018
+
1019
+fn preprocessed_grpc_request(media: Vec<pb::MediaItem>) -> pb::GenerateRequest {
1020
+ pb::GenerateRequest {
1021
+ request_id: "test-preprocessed-validation".to_string(),
1022
+ model: "test-model".to_string(),
1023
+ prompt: Some(pb::generate_request::Prompt::TokenIds(pb::TokenIds {
1024
+ ids: vec![11, 99, 99, 12],
1025
+ })),
1026
+ media,
1027
+ stopping: Some(pb::StoppingCriteria {
1028
+ max_new_tokens: 10,
1029
+ ..Default::default()
1030
+ }),
1031
+ ..Default::default()
1032
+ }
1033
+}
1034
+
1035
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1036
+#[serial]
1037
+async fn unary_generate_rejects_preprocessed_features_for_text_only_model() {
1038
+ let (inference_service, control_service, engine_health, _engine_task) =
1039
+ setup_grpc_service_with_backend(
1040
+ b"engine-grpc-preprocessed-text-only",
1041
+ default_stream_output_specs(),
1042
+ Arc::new(FakeTextBackend),
1043
+ |_| {},
1044
+ )
1045
+ .await;
1046
+ let (channel, server_task) = start_grpc_test_server(
1047
+ inference_service,
1048
+ control_service,
1049
+ engine_health,
1050
+ tokio_util::sync::CancellationToken::new(),
1051
+ )
1052
+ .await;
1053
+ let mut client = InferenceClient::new(channel);
1054
+
1055
+ let status = client
1056
+ .generate(preprocessed_grpc_request(vec![preprocessed_grpc_media(
1057
+ pb::Modality::Image,
1058
+ "image-a",
1059
+ 1,
1060
+ 2,
1061
+ )]))
1062
+ .await
1063
+ .expect_err("text-only models must reject preprocessed features");
1064
+
1065
+ assert_eq!(status.code(), tonic::Code::InvalidArgument);
1066
+ server_task.abort();
1067
+}
1068
+
1069
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1070
+#[serial]
1071
+async fn generate_validates_preprocessed_model_limits_and_prompt() {
1072
+ let limits = MmLimitPerPrompt::from([(MmModality::Image, MmLimitSpec::Count(1))]);
1073
+ let (inference_service, control_service, engine_health, _engine_task) =
1074
+ setup_grpc_service_with_backend(
1075
+ b"engine-grpc-preprocessed-limit",
1076
+ default_stream_output_specs(),
1077
+ multimodal_backend_with_limits(limits),
1078
+ |_| {},
1079
+ )
1080
+ .await;
1081
+ let (channel, server_task) = start_grpc_test_server(
1082
+ inference_service,
1083
+ control_service,
1084
+ engine_health,
1085
+ tokio_util::sync::CancellationToken::new(),
1086
+ )
1087
+ .await;
1088
+ let mut client = InferenceClient::new(channel);
1089
+
1090
+ let over_limit = preprocessed_grpc_request(vec![
1091
+ preprocessed_grpc_media(pb::Modality::Image, "image-a", 0, 1),
1092
+ preprocessed_grpc_media(pb::Modality::Image, "image-b", 2, 1),
1093
+ ]);
1094
+ let bad_range = preprocessed_grpc_request(vec![preprocessed_grpc_media(
1095
+ pb::Modality::Image,
1096
+ "image",
1097
+ 3,
1098
+ 2,
1099
+ )]);
1100
+ let unsupported = preprocessed_grpc_request(vec![preprocessed_grpc_media(
1101
+ pb::Modality::Audio,
1102
+ "audio",
1103
+ 0,
1104
+ 1,
1105
+ )]);
1106
+ let mut text = preprocessed_grpc_request(vec![preprocessed_grpc_media(
1107
+ pb::Modality::Image,
1108
+ "image",
1109
+ 0,
1110
+ 1,
1111
+ )]);
1112
+ let mut truncated = text.clone();
1113
+ text.prompt = Some(pb::generate_request::Prompt::Text("hello".to_owned()));
1114
+ truncated.truncate_prompt_tokens = 1;
1115
+ let empty_identifier =
1116
+ preprocessed_grpc_request(vec![preprocessed_grpc_media(pb::Modality::Image, "", 0, 1)]);
1117
+ for request in [
1118
+ over_limit,
1119
+ bad_range,
1120
+ unsupported,
1121
+ text,
1122
+ truncated,
1123
+ empty_identifier,
1124
+ ] {
1125
+ let status = client.generate(request.clone()).await.expect_err("invalid multimodal input");
1126
+ assert_eq!(status.code(), tonic::Code::InvalidArgument);
1127
+ let status = client
1128
+ .generate_stream(request)
1129
+ .await
1130
+ .expect_err("invalid streaming multimodal input");
1131
+ assert_eq!(status.code(), tonic::Code::InvalidArgument);
1132
+ }
1133
+ server_task.abort();
1134
+}
1135
+
1136
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1137
#[serial]
1138
async fn streaming_generate_rejects_text_prompt_with_media() {