@tduncan / Nvidia-Optimization-Stack-VLLM / commits / e98a8774cb

[Rust Frontend][gRPC] Add multimodal image inference (#50368)

Co-authored-by: Bugen Zhao <i@bugenzhao.com> Signed-off-by: Connor Carpenter <connorc@nvidia.com> Signed-off-by: Bugen Zhao <i@bugenzhao.com>

Connor Carpenter committed Aug 4, 2026 at 02:34 UTC e98a8774cba6f518a8de4433801cc32323a46071

12 files changed +553 -69

rust/Cargo.lock

@@ -5753,6 +5753,7 @@ dependencies = [

5753 "tracing",

5754 "tracing-futures",

5755 "tracing-subscriber",

5756 + "url",

5757 "uuid",

5758 "validator",

5759 "vllm-chat",

rust/proto/inference.proto

+25

@@ -47,6 +47,9 @@ message GenerateRequest {

47 int32 priority = 12;

49 optional string session_id = 13;

50 +

51 + // Multimodal inputs aligned with placeholder markers in token_ids.

52 + repeated MediaItem media = 14;

53 }

55 message RandomSampling {

@@ -202,3 +205,25 @@ message CandidateTokenInfo {

205 message TokenIds {

206 repeated uint32 ids = 1;

207 }

208 +

209 +// ======================================================================================

210 +// Media

211 +// ======================================================================================

212 +

213 +enum Modality {

214 + MODALITY_UNSPECIFIED = 0;

215 + MODALITY_IMAGE = 1;

216 + MODALITY_VIDEO = 2;

217 + MODALITY_AUDIO = 3;

218 +}

219 +

220 +message MediaItem {

221 + Modality modality = 1;

222 + oneof source {

223 + string url = 2; // http:// or https://

224 + string data_uri = 3; // data:

225 + bytes raw_bytes = 4;

226 + }

227 + string mime_type = 5;

228 + string uuid = 6;

229 +}

rust/src/chat/src/lib.rs

+29 -1

@@ -21,6 +21,7 @@ pub use event::{

21 AssistantToolCall, ChatEvent,

22 };

23 use futures::{StreamExt, TryStreamExt as _};

24 +pub use llm_multimodal::MediaContentPart;

25 pub use output::{

26 ChatOutputProcessor, DefaultChatOutputProcessor, DynChatOutputProcessor,

27 HarmonyChatOutputProcessor,

@@ -41,6 +42,7 @@ pub use request::{

42 ChatToolChoice, GenerationPromptMode, ReasoningEffort, SamplingParams,

43 };

44 pub use stream::{ChatEventStream, ChatEventStreamTrait, CollectedAssistantMessage};

45 +pub use vllm_engine_core_client::protocol::multimodal::MmFeatures;

46 pub use vllm_llm::FinishReason;

48 mod backend;

@@ -55,7 +57,6 @@ mod stream;

58 use vllm_engine_core_client::EngineCoreClient;

59 use vllm_engine_core_client::protocol::dtype::ModelDtype;

58 -use vllm_engine_core_client::protocol::multimodal::MmFeatures;

60 use vllm_engine_core_client::protocol::request::ReasoningParserKwargs;

61 use vllm_llm::Llm;

62 use vllm_text::{Prompt, TextLlm, TextRequest};

@@ -136,6 +137,24 @@ impl ChatRequestProcessor {

137 }

138 }

139

140 + /// Prepare media for an already-tokenized request.

141 + async fn prepare_media(

142 + &self,

143 + media: Vec<MediaContentPart>,

144 + token_ids: &mut Vec<u32>,

145 + ) -> Result<Option<MmFeatures>> {

146 + if media.is_empty() {

147 + return Ok(None);

148 + }

149 + let info = self

150 + .backend

151 + .multimodal_model_info()

152 + .ok_or(Error::UnsupportedMultimodalRenderer)?;

153 + let model_dtype = self.model_dtype.ok_or(Error::UnsupportedMultimodalRenderer)?;

154 + let features = info.prepare_multimodal(media, token_ids, model_dtype).await?;

155 + Ok(Some(features))

156 + }

157 +

158 /// Prepare one chat request without submitting it to an engine.

159 pub async fn prepare(

160 &self,

@@ -262,6 +281,15 @@ impl ChatLlm {

281 self.processor.backend.multimodal_model_info().is_some()

282 }

283

284 + /// Prepare media for an already-tokenized request.

285 + pub async fn prepare_media(

286 + &self,

287 + media: Vec<MediaContentPart>,

288 + token_ids: &mut Vec<u32>,

289 + ) -> Result<Option<MmFeatures>> {

290 + self.processor.prepare_media(media, token_ids).await

291 + }

292 +

293 /// Effective tool-call parser name for this model, if parsing is enabled.

294 pub fn tool_call_parser_name(&self) -> Option<&str> {

295 match &self.tool_call_parser {

rust/src/chat/src/multimodal.rs

+1 -1

@@ -696,7 +696,7 @@ impl MultimodalModelInfo {

696 /// `prompt_token_ids` is mutated in place because placeholder expansion

697 /// changes both the final prompt and the offsets recorded in

698 /// `PlaceholderRange`.

699 - async fn prepare_multimodal(

699 + pub(crate) async fn prepare_multimodal(

700 &self,

701 media_parts: Vec<MediaContentPart>,

702 prompt_token_ids: &mut Vec<u32>,

rust/src/cmd/src/cli.rs

+1 -1

@@ -154,7 +154,7 @@ pub struct SharedRuntimeArgs {

154 #[arg(long, value_parser = clap::value_parser!(i32).range(-1..), allow_negative_numbers = true)]

155 #[serde(default)]

156 pub max_logprobs: Option<i32>,

157 - /// TCP port for the gRPC Generate service. When not set, no gRPC server is

157 + /// TCP port for the gRPC Inference service. When not set, no gRPC server is

158 /// started.

159 #[arg(long)]

160 #[serde(default)]

rust/src/server/Cargo.toml

@@ -42,6 +42,7 @@ tower-http.workspace = true

42 tracing.workspace = true

43 tracing-futures.workspace = true

44 tracing-subscriber.workspace = true

45 +url.workspace = true

46 uuid.workspace = true

47 validator.workspace = true

48 vllm-chat.workspace = true

rust/src/server/src/config.rs

+1 -1

@@ -207,7 +207,7 @@ pub struct Config {

207 /// When `true`, suppress periodic stats logging (throughput, queue depth,

208 /// cache usage).

209 pub disable_log_stats: bool,

210 - /// TCP port for the gRPC Generate service. When `None`, no gRPC server is

210 + /// TCP port for the gRPC Inference service. When `None`, no gRPC server is

211 /// started.

212 pub grpc_port: Option<u16>,

213 /// Maximum time to wait for active HTTP/gRPC requests to drain on shutdown.

rust/src/server/src/grpc/convert.rs

+74

@@ -5,7 +5,9 @@

5 //! request/response types.

7 use tonic::Status;

8 +use url::Url;

9 use uuid::Uuid;

10 +use vllm_chat::MediaContentPart;

11 use vllm_engine_core_client::protocol::output::StopReason;

12 use vllm_engine_core_client::protocol::structured_outputs::StructuredOutputsParams;

13 use vllm_text::{

@@ -15,6 +17,78 @@ use vllm_text::{

18 use super::pb;

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 + match item.modality() {

26 + pb::Modality::Image => {}

27 + pb::Modality::Unspecified => {

28 + return Err(Status::invalid_argument(format!(

29 + "media[{index}].modality is required"

30 + )));

31 + }

32 + other => {

33 + return Err(Status::unimplemented(format!(

34 + "media[{index}].modality {other:?} is not supported by the gRPC service"

35 + )));

36 + }

37 + }

38 + let uuid = (!item.uuid.is_empty()).then_some(item.uuid);

39 + let mime_type = (!item.mime_type.is_empty()).then_some(item.mime_type);

40 + let part = match item.source {

41 + Some(pb::media_item::Source::Url(url)) => {

42 + validate_media_uri(index, "url", &url, &["http", "https"])?;

43 + MediaContentPart::ImageUrl {

44 + url,

45 + detail: None,

46 + uuid,

47 + }

48 + }

49 + Some(pb::media_item::Source::DataUri(uri)) => {

50 + validate_media_uri(index, "data_uri", &uri, &["data"])?;

51 + MediaContentPart::ImageUrl {

52 + url: uri,

53 + detail: None,

54 + uuid,

55 + }

56 + }

57 + Some(pb::media_item::Source::RawBytes(bytes)) => MediaContentPart::ImageData {

58 + data: bytes,

59 + mime_type,

60 + uuid,

61 + detail: None,

62 + },

63 + None => {

64 + return Err(Status::invalid_argument(format!(

65 + "media[{index}].source is required"

66 + )));

67 + }

68 + };

69 + parts.push(part);

70 + }

71 + Ok(parts)

72 +}

73 +

74 +fn validate_media_uri(

75 + index: usize,

76 + field: &str,

77 + value: &str,

78 + allowed_schemes: &[&str],

79 +) -> Result<(), Status> {

80 + let uri = Url::parse(value).map_err(|_| {

81 + Status::invalid_argument(format!("media[{index}].{field} is not a valid URI"))

82 + })?;

83 + if !allowed_schemes.contains(&uri.scheme()) {

84 + return Err(Status::invalid_argument(format!(

85 + "media[{index}].{field} must use the {} scheme",

86 + allowed_schemes.join(" or ")

87 + )));

88 + }

89 + Ok(())

90 +}

91 +

92 // ========================================================================================

93 // Request conversion

94 // ========================================================================================

rust/src/server/src/grpc/inference.rs

+189 -56

@@ -3,14 +3,18 @@

4 use std::pin::Pin;

5 use std::sync::Arc;

6 +use std::time::Instant;

8 use futures::{Stream, StreamExt as _};

9 use thiserror_ext::AsReport as _;

10 use tokio::sync::mpsc;

11 use tokio_stream::wrappers::ReceiverStream;

12 use tonic::{Request, Response, Status};

12 -use tracing::info;

13 -use vllm_text::{DecodedTextEvent, TextOutputStreamExt as _};

13 +use tracing::{Span, info, info_span, warn};

14 +use tracing_futures::Instrument as _;

15 +use uuid::Uuid;

16 +use vllm_llm::current_unix_timestamp_secs;

17 +use vllm_text::{DecodedTextEvent, Prompt, TextOutputStreamExt as _, TextRequest};

19 use super::convert::{self, ResponseOpts};

20 use super::{InferenceServer, pb};

@@ -23,10 +27,92 @@ pub struct InferenceServiceImpl {

27 state: Arc<AppState>,

28 }

30 +struct PreparedGrpcRequest {

31 + text_request: TextRequest,

32 + request_span: Span,

33 + started_at: Instant,

34 +}

35 +

36 impl InferenceServiceImpl {

37 pub fn new(state: Arc<AppState>) -> Self {

38 Self { state }

39 }

40 +

41 + async fn prepare_request(

42 + &self,

43 + mut proto_request: pb::GenerateRequest,

44 + stream: bool,

45 + rpc: &'static str,

46 + ) -> Result<PreparedGrpcRequest, Status> {

47 + let started_at = Instant::now();

48 + let arrival_time = current_unix_timestamp_secs();

49 + if proto_request.request_id.is_empty() {

50 + proto_request.request_id = Uuid::new_v4().to_string();

51 + }

52 + let request_id = proto_request.request_id.clone();

53 + let model = if proto_request.model.is_empty() {

54 + self.state.primary_model_name().to_string()

55 + } else {

56 + proto_request.model.clone()

57 + };

58 + let media_count = proto_request.media.len();

59 + let request_span = info_span!(

60 + "grpc_inference",

61 + %request_id,

62 + rpc,

63 + %model,

64 + media_count,

65 + );

66 + info!(parent: &request_span, "gRPC inference request received");

67 +

68 + let result = async {

69 + let media = std::mem::take(&mut proto_request.media);

70 + let mut text_request =

71 + convert::to_text_request(proto_request, stream, self.state.served_model_names())?;

72 + text_request.arrival_time = Some(arrival_time);

73 +

74 + let media = convert::media_parts_from_request(media)?;

75 + if !media.is_empty() {

76 + let Prompt::TokenIds(mut token_ids) = text_request.prompt else {

77 + return Err(Status::invalid_argument(

78 + "multimodal gRPC requests must provide token_ids input",

79 + ));

80 + };

81 + let mm_features = self

82 + .state

83 + .chat

84 + .prepare_media(media, &mut token_ids)

85 + .await

86 + .map_err(|error| Status::internal(error.to_report_string()))?;

87 + text_request.prompt = Prompt::TokenIds(token_ids);

88 + text_request.mm_features = mm_features;

89 + }

90 +

91 + Ok(text_request)

92 + }

93 + .instrument(request_span.clone())

94 + .await;

95 +

96 + let text_request = match result {

97 + Ok(text_request) => text_request,

98 + Err(status) => {

99 + warn!(

100 + parent: &request_span,

101 + grpc_code = ?status.code(),

102 + elapsed_ms = started_at.elapsed().as_millis() as u64,

103 + error = %status.message(),

104 + "gRPC inference request preparation failed"

105 + );

106 + return Err(status);

107 + }

108 + };

109 +

110 + Ok(PreparedGrpcRequest {

111 + text_request,

112 + request_span,

113 + started_at,

114 + })

115 + }

116 }

117

118 #[tonic::async_trait]

@@ -41,16 +127,31 @@ impl pb::inference_server::Inference for InferenceServiceImpl {

127 ) -> Result<Response<pb::GenerateResponse>, Status> {

128 let proto_req = request.into_inner();

129 let response_opts = ResponseOpts::from_proto(proto_req.response.as_ref());

44 - let text_request =

45 - convert::to_text_request(proto_req, false, self.state.served_model_names())?;

130 + let PreparedGrpcRequest {

131 + text_request,

132 + request_span,

133 + started_at,

134 + } = self.prepare_request(proto_req, false, "Generate").await?;

135

47 - let request_id = text_request.request_id.clone();

48 - info!(%request_id, "grpc generate (unary)");

136 + let stream = self

137 + .state

138 + .chat

139 + .text()

140 + .generate(text_request)

141 + .instrument(request_span.clone())

142 + .await

143 + .map_err(|error| log_text_error(&request_span, started_at, "submission", error))?;

144

50 - let stream = self.state.chat.text().generate(text_request).await;

51 - let stream = stream.map_err(text_error_to_status)?;

52 -

53 - let collected = stream.collect_output().await.map_err(text_error_to_status)?;

145 + let collected = stream

146 + .collect_output()

147 + .instrument(request_span.clone())

148 + .await

149 + .map_err(|error| log_text_error(&request_span, started_at, "collection", error))?;

150 + info!(

151 + parent: &request_span,

152 + elapsed_ms = started_at.elapsed().as_millis() as u64,

153 + "gRPC inference request completed"

154 + );

155

156 // Build the single aggregated response.

157 let prompt_info = convert::to_prompt_info(

@@ -87,59 +188,73 @@ impl pb::inference_server::Inference for InferenceServiceImpl {

188 ) -> Result<Response<Self::GenerateStreamStream>, Status> {

189 let proto_req = request.into_inner();

190 let response_opts = ResponseOpts::from_proto(proto_req.response.as_ref());

90 - let text_request =

91 - convert::to_text_request(proto_req, true, self.state.served_model_names())?;

92 -

93 - let request_id = text_request.request_id.clone();

94 - info!(%request_id, "grpc generate (stream)");

191 + let PreparedGrpcRequest {

192 + text_request,

193 + request_span,

194 + started_at,

195 + } = self.prepare_request(proto_req, true, "GenerateStream").await?;

196

96 - let stream = self.state.chat.text().generate(text_request).await;

97 - let stream = stream.map_err(text_error_to_status)?;

197 + let stream = self

198 + .state

199 + .chat

200 + .text()

201 + .generate(text_request)

202 + .instrument(request_span.clone())

203 + .await

204 + .map_err(|error| log_text_error(&request_span, started_at, "submission", error))?;

205

206 let (tx, rx) = mpsc::channel(32);

207

101 - tokio::spawn(async move {

102 - futures::pin_mut!(stream);

103 - while let Some(event) = stream.next().await {

104 - let response = match event {

105 - Err(e) => Err(text_error_to_status(e)),

106 - Ok(DecodedTextEvent::Start {

107 - prompt_token_ids,

108 - prompt_logprobs,

109 - }) => {

110 - let prompt_info = convert::to_prompt_info(

111 - &prompt_token_ids,

112 - prompt_logprobs.as_ref(),

113 - &response_opts,

114 - );

115 - Ok(pb::GenerateResponse {

116 - prompt_info: Some(prompt_info),

117 - outputs: None,

118 - })

119 - }

120 - Ok(DecodedTextEvent::TextDelta {

121 - delta,

122 - token_ids,

123 - logprobs,

124 - finished,

125 - }) => Ok(pb::GenerateResponse {

126 - prompt_info: None,

127 - outputs: Some(convert::to_sequence_output(

128 - &delta,

129 - &token_ids,

130 - logprobs.as_ref(),

131 - finished.as_ref(),

132 - &response_opts,

133 - )),

134 - }),

135 - };

208 + let task_span = request_span.clone();

209 + tokio::spawn(

210 + async move {

211 + futures::pin_mut!(stream);

212 + while let Some(event) = stream.next().await {

213 + let response = match event {

214 + Err(error) => Err(log_text_error(&task_span, started_at, "stream", error)),

215 + Ok(DecodedTextEvent::Start {

216 + prompt_token_ids,

217 + prompt_logprobs,

218 + }) => {

219 + let prompt_info = convert::to_prompt_info(

220 + &prompt_token_ids,

221 + prompt_logprobs.as_ref(),

222 + &response_opts,

223 + );

224 + Ok(pb::GenerateResponse {

225 + prompt_info: Some(prompt_info),

226 + outputs: None,

227 + })

228 + }

229 + Ok(DecodedTextEvent::TextDelta {

230 + delta,

231 + token_ids,

232 + logprobs,

233 + finished,

234 + }) => Ok(pb::GenerateResponse {

235 + prompt_info: None,

236 + outputs: Some(convert::to_sequence_output(

237 + &delta,

238 + &token_ids,

239 + logprobs.as_ref(),

240 + finished.as_ref(),

241 + &response_opts,

242 + )),

243 + }),

244 + };

245

137 - if tx.send(response).await.is_err() {

138 - // Client disconnected.

139 - break;

246 + if tx.send(response).await.is_err() {

247 + break;

248 + }

249 }

250 + info!(

251 + parent: &task_span,

252 + elapsed_ms = started_at.elapsed().as_millis() as u64,

253 + "gRPC inference stream closed"

254 + );

255 }

142 - });

256 + .instrument(request_span),

257 + );

258

259 let response_stream = ReceiverStream::new(rx);

260 Ok(Response::new(Box::pin(response_stream)))

@@ -154,3 +269,21 @@ fn text_error_to_status(error: vllm_text::Error) -> Status {

269 Status::internal(message)

270 }

271 }

272 +

273 +fn log_text_error(

274 + request_span: &Span,

275 + started_at: Instant,

276 + phase: &'static str,

277 + error: vllm_text::Error,

278 +) -> Status {

279 + let status = text_error_to_status(error);

280 + warn!(

281 + parent: request_span,

282 + phase,

283 + grpc_code = ?status.code(),

284 + elapsed_ms = started_at.elapsed().as_millis() as u64,

285 + error = %status.message(),

286 + "gRPC inference request failed"

287 + );

288 + status

289 +}

rust/src/server/src/grpc/tests.rs

+226 -5

@@ -1,6 +1,7 @@

1 // SPDX-License-Identifier: Apache-2.0

2 // SPDX-FileCopyrightText: Copyright contributors to the vLLM project

4 +use std::fs;

5 use std::future::Future;

6 use std::io;

7 use std::pin::Pin;

@@ -204,6 +205,73 @@ impl ChatRenderer for FakeTextBackend {

205 }

206 }

207

208 +struct FakeMultimodalBackend {

209 + model_info: vllm_chat::multimodal::MultimodalModelInfo,

210 +}

211 +

212 +impl TextBackend for FakeMultimodalBackend {

213 + fn tokenizer(&self) -> DynTokenizer {

214 + Arc::new(TestTokenizer::new().with_regular_token("<|image_pad|>", QWEN_IMAGE_TOKEN_ID))

215 + }

216 +

217 + fn model_id(&self) -> &str {

218 + "test-model"

219 + }

220 +}

221 +

222 +impl ChatBackend for FakeMultimodalBackend {

223 + fn chat_renderer(&self) -> DynChatRenderer {

224 + Arc::new(FakeTextBackend)

225 + }

226 +

227 + fn multimodal_model_info(&self) -> Option<&vllm_chat::multimodal::MultimodalModelInfo> {

228 + Some(&self.model_info)

229 + }

230 +

231 + fn new_chat_output_processor(

232 + &self,

233 + request: &mut ChatRequest,

234 + options: NewChatOutputProcessorOptions<'_>,

235 + ) -> vllm_chat::Result<DynChatOutputProcessor> {

236 + Ok(Box::new(DefaultChatOutputProcessor::new(

237 + request,

238 + self.model_id(),

239 + self.tokenizer(),

240 + options.tool_call_parser,

241 + options.reasoning_parser,

242 + )?))

243 + }

244 +}

245 +

246 +const QWEN_IMAGE_TOKEN_ID: u32 = 151655;

247 +const TINY_PNG_DATA_URI: &str = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=";

248 +

249 +fn multimodal_backend() -> Arc<dyn ChatTextBackend> {

250 + let config_path = std::env::temp_dir().join(format!(

251 + "vllm-grpc-qwen-config-{}.json",

252 + uuid::Uuid::new_v4()

253 + ));

254 + fs::write(

255 + &config_path,

256 + r#"{"model_type":"qwen2_vl","image_token_id":151655}"#,

257 + )

258 + .expect("write qwen test config");

259 + let model_info = vllm_chat::multimodal::MultimodalModelInfo::from_paths(

260 + "qwen2-vl-test".to_string(),

261 + Some("qwen2_vl".to_string()),

262 + vllm_chat::multimodal::MultimodalConfigFiles {

263 + config: Some(&config_path),

264 + ..Default::default()

265 + },

266 + Arc::new(TestTokenizer::new().with_regular_token("<|image_pad|>", QWEN_IMAGE_TOKEN_ID)),

267 + Default::default(),

268 + )

269 + .expect("load multimodal info")

270 + .expect("qwen multimodal info is registered");

271 + let _ = fs::remove_file(config_path);

272 + Arc::new(FakeMultimodalBackend { model_info })

273 +}

274 +

275 /// Build the gRPC service + mock engine that serves a single request with the

276 /// given output specs. Shared by the plaintext and TLS server fixtures.

277 async fn setup_grpc_service(

@@ -215,6 +283,24 @@ async fn setup_grpc_service(

283 tokio::sync::watch::Receiver<bool>,

284 MockEngineTask,

285 ) {

286 + setup_grpc_service_with_backend(engine_id, output_specs, Arc::new(FakeTextBackend), |_| {})

287 + .await

288 +}

289 +

290 +async fn setup_grpc_service_with_backend<F>(

291 + engine_id: impl Into<EngineId>,

292 + output_specs: Vec<(Vec<u32>, Option<EngineCoreFinishReason>)>,

293 + backend: Arc<dyn ChatTextBackend>,

294 + check_request: F,

295 +) -> (

296 + InferenceServer<InferenceServiceImpl>,

297 + ControlServer<ControlServiceImpl>,

298 + tokio::sync::watch::Receiver<bool>,

299 + MockEngineTask,

300 +)

301 +where

302 + F: FnOnce(&EngineCoreRequest) + Send + 'static,

303 +{

304 let ipc = IpcNamespace::new().expect("create ipc namespace");

305 let handshake_address = ipc.handshake_endpoint();

306 let engine_id = engine_id.into();

@@ -227,6 +313,7 @@ async fn setup_grpc_service(

313 let add = recv_engine_message(dealer).await;

314 let request: EngineCoreRequest =

315 rmp_serde::from_slice(&add[1]).expect("decode request");

316 + check_request(&request);

317 send_outputs(

318 push,

319 engine_outputs_for_request(&request.request_id, output_specs),

@@ -248,13 +335,11 @@ async fn setup_grpc_service(

335 .expect("connect client");

336 let engine_health = client.subscribe_health();

337

251 - let chat = ChatLlm::from_shared_backend(

252 - Llm::new(client),

253 - Arc::new(FakeTextBackend) as Arc<dyn ChatTextBackend>,

254 - );

338 + let chat = ChatLlm::from_shared_backend(Llm::new(client), backend);

339 let state = Arc::new(AppState::new(vec!["test-model".to_string()], chat));

340 (

257 - InferenceServer::new(InferenceServiceImpl::new(state.clone())),

341 + InferenceServer::new(InferenceServiceImpl::new(state.clone()))

342 + .max_decoding_message_size(crate::DEFAULT_REQUEST_BODY_LIMIT_BYTES),

343 ControlServer::new(ControlServiceImpl::new(state)),

344 engine_health,

345 engine_task,

@@ -560,6 +645,142 @@ async fn unary_generate_with_token_ids_prompt() {

645 server_task.abort();

646 }

647

648 +#[tokio::test(flavor = "multi_thread", worker_threads = 2)]

649 +#[serial]

650 +async fn unary_generate_prepares_multimodal_input_for_engine_core() {

651 + let (inference_service, control_service, engine_health, engine_task) =

652 + setup_grpc_service_with_backend(

653 + b"engine-grpc-multimodal",

654 + default_stream_output_specs(),

655 + multimodal_backend(),

656 + |request| {

657 + let token_ids = request.prompt_token_ids.as_ref().expect("prompt token ids");

658 + let features = request.mm_features.as_ref().expect("multimodal features");

659 + assert_eq!(features.len(), 1);

660 +

661 + let feature = &features[0];

662 + assert_eq!(feature.modality, "image");

663 + assert_eq!(feature.identifier, "image-1");

664 + assert_eq!(feature.mm_position.offset, 1);

665 + assert!(feature.mm_position.length > 1);

666 + assert_eq!(token_ids.len(), feature.mm_position.length + 2);

667 + assert_eq!(token_ids[0], 11);

668 + assert_eq!(token_ids.last(), Some(&12));

669 + assert!(

670 + token_ids[feature.mm_position.offset

671 + ..feature.mm_position.offset + feature.mm_position.length]

672 + .iter()

673 + .all(|token_id| *token_id == QWEN_IMAGE_TOKEN_ID)

674 + );

675 + },

676 + )

677 + .await;

678 + let (channel, server_task) = start_grpc_test_server(

679 + inference_service,

680 + control_service,

681 + engine_health,

682 + tokio_util::sync::CancellationToken::new(),

683 + )

684 + .await;

685 + let mut client = InferenceClient::new(channel);

686 +

687 + client

688 + .generate(pb::GenerateRequest {

689 + request_id: "test-multimodal".to_string(),

690 + model: "test-model".to_string(),

691 + prompt: Some(pb::generate_request::Prompt::TokenIds(pb::TokenIds {

692 + ids: vec![11, QWEN_IMAGE_TOKEN_ID, 12],

693 + })),

694 + media: vec![pb::MediaItem {

695 + modality: pb::Modality::Image as i32,

696 + source: Some(pb::media_item::Source::DataUri(

697 + TINY_PNG_DATA_URI.to_string(),

698 + )),

699 + mime_type: String::new(),

700 + uuid: "image-1".to_string(),

701 + }],

702 + stopping: Some(pb::StoppingCriteria {

703 + max_new_tokens: 10,

704 + ..Default::default()

705 + }),

706 + ..Default::default()

707 + })

708 + .await

709 + .expect("multimodal generate");

710 +

711 + engine_task.await.expect("mock engine task");

712 + server_task.abort();

713 +}

714 +

715 +#[tokio::test(flavor = "multi_thread", worker_threads = 2)]

716 +#[serial]

717 +async fn streaming_generate_rejects_text_prompt_with_media() {

718 + let (mut client, server_task, _engine_task) = grpc_test_server(

719 + b"engine-grpc-stream-media-text",

720 + default_stream_output_specs(),

721 + )

722 + .await;

723 +

724 + let status = client

725 + .generate_stream(pb::GenerateRequest {

726 + request_id: "test-stream-media-text".to_string(),

727 + model: "test-model".to_string(),

728 + prompt: Some(pb::generate_request::Prompt::Text(

729 + "describe this".to_string(),

730 + )),

731 + media: vec![pb::MediaItem {

732 + modality: pb::Modality::Image as i32,

733 + source: Some(pb::media_item::Source::DataUri(

734 + TINY_PNG_DATA_URI.to_string(),

735 + )),

736 + mime_type: String::new(),

737 + uuid: "image-1".to_string(),

738 + }],

739 + ..Default::default()

740 + })

741 + .await

742 + .expect_err("text prompts with media must be rejected");

743 +

744 + assert_eq!(status.code(), tonic::Code::InvalidArgument);

745 + assert_eq!(

746 + status.message(),

747 + "multimodal gRPC requests must provide token_ids input"

748 + );

749 + server_task.abort();

750 +}

751 +

752 +#[tokio::test(flavor = "multi_thread", worker_threads = 2)]

753 +#[serial]

754 +async fn unary_generate_accepts_request_larger_than_tonic_default() {

755 + let (mut client, server_task, _engine_task) =

756 + grpc_test_server(b"engine-grpc-large-media", default_stream_output_specs()).await;

757 +

758 + let status = client

759 + .generate(pb::GenerateRequest {

760 + request_id: "test-large-media".to_string(),

761 + model: "test-model".to_string(),

762 + prompt: Some(pb::generate_request::Prompt::Text(

763 + "describe this".to_string(),

764 + )),

765 + media: vec![pb::MediaItem {

766 + modality: pb::Modality::Image as i32,

767 + source: Some(pb::media_item::Source::RawBytes(vec![0; 5 * 1024 * 1024])),

768 + mime_type: "image/png".to_string(),

769 + uuid: "image-1".to_string(),

770 + }],

771 + ..Default::default()

772 + })

773 + .await

774 + .expect_err("text prompts with media must be rejected after decoding");

775 +

776 + assert_eq!(status.code(), tonic::Code::InvalidArgument);

777 + assert_eq!(

778 + status.message(),

779 + "multimodal gRPC requests must provide token_ids input"

780 + );

781 + server_task.abort();

782 +}

783 +

784 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]

785 #[serial]

786 async fn unary_generate_returns_token_ids_when_requested() {

rust/src/server/src/lib.rs

+3 -1

@@ -59,6 +59,7 @@ const GRPC_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(7200);

59 /// How long the server waits for a keepalive PING reply before dropping the gRPC

60 /// connection. 20s matches the gRPC-core default.

61 const GRPC_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(20);

62 +const DEFAULT_REQUEST_BODY_LIMIT_BYTES: usize = 32 * 1024 * 1024;

64 /// Resolve the public model names accepted by the frontend.

65 fn effective_served_model_names(model: &str, served_model_name: &[String]) -> Vec<String> {

@@ -215,7 +216,8 @@ where

216 let control_service =

217 grpc::ControlGrpcService::new(grpc::ControlServiceImpl::new(state.clone()));

218 let inference_service =

218 - grpc::InferenceGrpcService::new(grpc::InferenceServiceImpl::new(state.clone()));

219 + grpc::InferenceGrpcService::new(grpc::InferenceServiceImpl::new(state.clone()))

220 + .max_decoding_message_size(DEFAULT_REQUEST_BODY_LIMIT_BYTES);

221 let svc = TonicServer::builder()

222 .http2_keepalive_interval(Some(GRPC_KEEPALIVE_INTERVAL))

223 .http2_keepalive_timeout(Some(GRPC_KEEPALIVE_TIMEOUT))

rust/src/server/src/routes.rs

+2 -3

@@ -27,11 +27,10 @@ use axum::routing::{get, post};

27 use tower_http::trace::TraceLayer;

28 use tracing::warn;

30 +use crate::DEFAULT_REQUEST_BODY_LIMIT_BYTES;

31 use crate::middleware;

32 use crate::state::AppState;

33 -const DEFAULT_JSON_BODY_LIMIT_BYTES: usize = 32 * 1024 * 1024;

34 -

34 fn server_dev_mode_enabled() -> bool {

35 std::env::var("VLLM_SERVER_DEV_MODE")

36 .ok()

@@ -127,7 +126,7 @@ fn build_router_with_options(

126 let enable_api_key_auth = state.has_api_keys();

127 let mut router = router

128 .with_state(state.clone())

130 - .layer(DefaultBodyLimit::max(DEFAULT_JSON_BODY_LIMIT_BYTES))

129 + .layer(DefaultBodyLimit::max(DEFAULT_REQUEST_BODY_LIMIT_BYTES))

130 .layer(middleware::request_runtime_layer(state.clone()))

131 .layer(from_fn_with_state(

132 state.clone(),