Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,16 @@ for the architecture.
| Disaggregated | Yes |
| KV routing | Yes |
| Kubernetes multinode tensor parallelism (TP) and attention data parallelism (attention-DP) | Experimental |
| Image, video, and audio input | Yes; disaggregated serving is untested |

The sidecar sends a media kind only when the SGLang gRPC server lists it in
`grpc_features` (`image_inputs`, `video_inputs`, or `audio_inputs`), and returns
400 for other media requests.

> [!WARNING]
> SGLang downloads `http://` and `https://` media URLs itself, from any host by
> default, including `127.0.0.1` and its own HTTP API. Start SGLang with
> `--allowed-media-domains` to limit the hosts it fetches.

## Launch Locally

Expand Down
10 changes: 10 additions & 0 deletions lib/sidecar/sglang/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,16 @@ server. See the [sidecar overview](../README.md) for installation.
| Disaggregated | Yes |
| KV routing | Yes |
| Kubernetes multinode tensor parallelism (TP) and attention data parallelism (attention-DP) | Experimental |
| Image, video, and audio input | Yes; disaggregated serving is untested |

The sidecar sends a media kind only when the SGLang gRPC server lists it in
`grpc_features` (`image_inputs`, `video_inputs`, or `audio_inputs`), and returns
400 for other media requests.

> [!WARNING]
> SGLang downloads `http://` and `https://` media URLs itself, from any host by
> default, including `127.0.0.1` and its own HTTP API. Start SGLang with
> `--allowed-media-domains` to limit the hosts it fetches.

## Run locally

Expand Down
14 changes: 11 additions & 3 deletions lib/sidecar/sglang/proto/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -11,11 +11,19 @@ available there, Dynamo should remove this directory, pin and install the
matching `sglang` wheel as a build dependency, and compile the packaged proto
instead.

The contract was copied from the SGLang v0.5.22 release branch, commit
[`baa9befa381511cd4d56f83980951e5a45fffc6d`](https://github.com/sgl-project/sglang/blob/baa9befa381511cd4d56f83980951e5a45fffc6d/proto/sglang/runtime/v1/sglang.proto).
The contract was copied from the SGLang v0.5.23 release branch, commit
[`297e54fd105e974cdf401e5b4d20253a22c16f90`](https://github.com/sgl-project/sglang/blob/297e54fd105e974cdf401e5b4d20253a22c16f90/proto/sglang/runtime/v1/sglang.proto).
The upstream file's SHA-256 is
`7852d2dbff67c35889d4592516c9c17a9681b6559a514959c36b6acac157c2ff`.
`3e7a78c9d2327530ea22369b2c6159886f720a9887d41a4f2983aebe7f257d2f`.
The local file adds SPDX and temporary-copy comments and applies Dynamo's
`clang-format` style; these changes do not alter the protobuf descriptor. The
SGLang sidecar generates both client and server types and temporarily exposes
them to the Mocker server.

SGLang 0.5.23 adds the `GenerateRequest` media fields 19 to 21 (`image_data`,
`video_data`, and `audio_data`) and the `GetServerInfo` `grpc_features` list.
The sidecar sends each media kind only to a server that lists it in
`grpc_features` (`image_inputs`, `video_inputs`, or `audio_inputs`). An older
server drops unknown fields, and a model processor drops each media kind that
it has no placeholder for. Both drops are silent, so the sidecar refuses each
media kind that the server does not list.
4 changes: 3 additions & 1 deletion lib/sidecar/sglang/proto/sglang.proto
Original file line number Diff line number Diff line change
Expand Up @@ -150,8 +150,10 @@ message GenerateRequest
optional bool require_reasoning = 15;
optional uint32 max_thinking_tokens = 16;
optional KVHintsEnvelope kv_hints = 17;
// Also return the detokenized text, as HTTP /generate does; off by default.
optional bool return_text = 18;
repeated string image_data = 19;
repeated string video_data = 20;
repeated string audio_data = 21;
}

// Versioned KV-hint envelope forwarded from a trusted orchestrator.
Expand Down
62 changes: 60 additions & 2 deletions lib/sidecar/sglang/src/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -529,6 +529,34 @@ pub(crate) fn discovery_mode(server_info: &Value) -> Result<DisaggregationMode,
}
}

/// Media kinds that this SGLang server reads over gRPC.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub(crate) struct MediaSupport {
pub(crate) image: bool,
pub(crate) video: bool,
pub(crate) audio: bool,
}

/// SGLang lists one `grpc_features` entry for each media kind that the model's
/// processor can place in the prompt. It drops the other media silently.
pub(crate) fn media_support(server_info: &Value) -> MediaSupport {
let listed = |name: &str| {
server_info
.get("grpc_features")
.and_then(Value::as_array)
.is_some_and(|features| {
features
.iter()
.any(|feature| feature.as_str() == Some(name))
})
};
MediaSupport {
image: listed("image_inputs"),
video: listed("video_inputs"),
audio: listed("audio_inputs"),
}
}

fn parse_json_object(label: &str, raw: &str) -> Result<Value, DynamoError> {
let value: Value = serde_json::from_str(raw)
.map_err(|err| protocol_error(format!("invalid {label}: {err}")))?;
Expand Down Expand Up @@ -622,8 +650,9 @@ mod tests {
use tonic::transport::Endpoint;

use super::{
NodeMetadata, client_from_channel, discover, discovery_mode, json_u32, json_u64,
parse_discovery, rpc_with_deadline, status_to_dynamo, worker_group_id_from_addresses,
MediaSupport, NodeMetadata, client_from_channel, discover, discovery_mode, json_u32,
json_u64, media_support, parse_discovery, rpc_with_deadline, status_to_dynamo,
worker_group_id_from_addresses,
};
use crate::proto as pb;

Expand Down Expand Up @@ -887,6 +916,35 @@ mod tests {
}
}

#[test]
fn media_support_reads_one_grpc_feature_per_media_kind() {
assert_eq!(
media_support(&json!({"grpc_features": ["other", "image_inputs", "video_inputs"]})),
MediaSupport {
image: true,
video: true,
audio: false
}
);
assert_eq!(
media_support(&json!({"grpc_features": ["audio_inputs"]})),
MediaSupport {
image: false,
video: false,
audio: true
}
);
for info in [
json!({}),
json!({"grpc_features": []}),
json!({"grpc_features": ["multimodal_inputs"]}),
json!({"grpc_features": "image_inputs"}),
json!({"grpc_features": [null, 1]}),
] {
assert_eq!(media_support(&info), MediaSupport::default(), "{info}");
}
}

#[test]
fn discovery_requires_incremental_streaming() {
for info in [
Expand Down
11 changes: 9 additions & 2 deletions lib/sidecar/sglang/src/engine.rs
Original file line number Diff line number Diff line change
Expand Up @@ -27,12 +27,14 @@ use tokio::time::Instant;
use tokio_util::sync::CancellationToken;

use crate::args::Args;
use crate::client::{self, Client, Discovery, KV_CONFIG_KEY, NodeMetadata, Pool, WORKER_GROUP_KEY};
use crate::client::{
self, Client, Discovery, KV_CONFIG_KEY, MediaSupport, NodeMetadata, Pool, WORKER_GROUP_KEY,
};
use crate::native_http::{self, NativeHttp};
use crate::proto as pb;
use crate::protocol::{
build_generate_request, disaggregated_params_to_json, engine_data_from_meta, extract_logprobs,
meta_u32, output_ids_to_u32, terminal_from_meta,
meta_u32, output_ids_to_u32, require_media_support, terminal_from_meta,
};

const RETRY_LOG_INTERVAL: Duration = Duration::from_secs(30);
Expand All @@ -53,6 +55,7 @@ struct StartedState {
native_http: Option<NativeHttp>,
kv_event_sources: Vec<DiscoveredKvEventSource>,
combined_token_limit: Option<u32>,
media_support: MediaSupport,
}

#[derive(Clone, Debug, Eq, PartialEq)]
Expand Down Expand Up @@ -327,19 +330,22 @@ impl LLMEngine for SglangSidecarEngine {
if combined_token_limit.is_none() {
tracing::warn!("requests without max_tokens keep SGLang's 128-token default");
}
let media_support = client::media_support(&discovery.server_info);
self.state
.set(StartedState {
pool,
native_http,
kv_event_sources,
combined_token_limit,
media_support,
})
.map_err(|_| client::engine_shutdown("sglang sidecar already started"))?;
tracing::info!(
model = %config.model,
mode = ?self.disaggregation_mode,
connections = connection_count,
kv_event_sources = kv_event_source_count,
media_support = ?media_support,
"sglang sidecar started"
);
Ok(config)
Expand Down Expand Up @@ -389,6 +395,7 @@ impl LLMEngine for SglangSidecarEngine {
self.bootstrap_host.as_deref(),
self.bootstrap_port,
)?;
require_media_support(&grpc_request, state.media_support)?;
let mut prefill_handoff = if self.disaggregation_mode.is_prefill() {
grpc_request
.disaggregated_params
Expand Down
120 changes: 115 additions & 5 deletions lib/sidecar/sglang/src/protocol.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,14 +6,18 @@
use std::collections::HashMap;

use dynamo_backend_common::{
DisaggregationMode, DynamoError, LLMEngineOutput, LLMEngineOutputExt, PreprocessedRequest,
PromptTokensDetails, StopConditions, StopReason, TopLogprob, usage,
DisaggregationMode, DynamoError, LLMEngineOutput, LLMEngineOutputExt, MultimodalData,
PreprocessedRequest, PromptTokensDetails, StopConditions, StopReason, TopLogprob, usage,
};
use serde_json::{Map, Value};

use crate::client;
use crate::client::{self, MediaSupport};
use crate::proto as pb;

const IMAGE_URL_KEY: &str = "image_url";
const VIDEO_URL_KEY: &str = "video_url";
const AUDIO_URL_KEY: &str = "audio_url";

pub(crate) fn build_generate_request(
request: &PreprocessedRequest,
request_id: &str,
Expand All @@ -22,6 +26,7 @@ pub(crate) fn build_generate_request(
bootstrap_port: Option<u16>,
) -> Result<pb::GenerateRequest, DynamoError> {
validate_request(request)?;
let [image_data, video_data, audio_data] = media_data(request)?;
let input_ids = request
.token_ids
.iter()
Expand Down Expand Up @@ -179,9 +184,99 @@ pub(crate) fn build_generate_request(
bootstrap_host,
bootstrap_port,
)?,
image_data,
video_data,
audio_data,
})
}

/// SGLang matches each list to the placeholder runs of that modality in
/// `input_ids`, so each list keeps the request order.
fn media_data(request: &PreprocessedRequest) -> Result<[Vec<String>; 3], DynamoError> {
let (mut image_data, mut video_data, mut audio_data) = (Vec::new(), Vec::new(), Vec::new());
for (key, items) in request.multi_modal_data.iter().flatten() {
let data = match key.as_str() {
IMAGE_URL_KEY => &mut image_data,
VIDEO_URL_KEY => &mut video_data,
AUDIO_URL_KEY => &mut audio_data,
_ => {
return Err(client::invalid_request(
"SGLang's native Generate RPC accepts only image, video, and audio media",
));
}
};
for item in items {
data.push(media_source(item)?);
}
}
let media = [image_data, video_data, audio_data];
// Text requests ignore these options, but SGLang would decode media with its own defaults.
if request.media_io_kwargs.is_some() && media.iter().any(|data| !data.is_empty()) {
return Err(client::invalid_request(
"media_io_kwargs are not supported by SGLang's native Generate RPC",
));
}
Ok(media)
}

/// SGLang reads `file://` URLs and bare paths from the disk of the SGLang server.
fn media_source(item: &MultimodalData) -> Result<String, DynamoError> {
let source = match item {
MultimodalData::Url(url) => url.as_str(),
MultimodalData::RawUrl(source) => source,
MultimodalData::Decoded(_) => {
return Err(client::invalid_request(
"the SGLang sidecar cannot read frontend-decoded media; configure URL passthrough",
));
}
MultimodalData::UuidOnly(_) => {
return Err(client::invalid_request(
"the SGLang sidecar needs a media URL and cannot resolve UUID-only media",
));
}
};
if ["http://", "https://", "data:"]
.iter()
.any(|scheme| source.starts_with(scheme))
{
Ok(source.to_string())
Comment thread
KrishnanPrash marked this conversation as resolved.
Resolved
} else {
Err(client::invalid_request(
"the SGLang sidecar accepts media only as http://, https://, or data: URLs",
))
}
}

/// SGLang drops media that the server or the model cannot read without an error,
/// so the model would answer without seeing it.
pub(crate) fn require_media_support(
request: &pb::GenerateRequest,
support: MediaSupport,
) -> Result<(), DynamoError> {
for (present, supported, message) in [
(
!request.image_data.is_empty(),
support.image,
"this SGLang server does not accept image inputs over gRPC; serve an image model with an SGLang version that supports gRPC media",
),
(
!request.video_data.is_empty(),
support.video,
"this SGLang server does not accept video inputs over gRPC; serve a video model with an SGLang version that supports gRPC media",
),
(
!request.audio_data.is_empty(),
support.audio,
"this SGLang server does not accept audio inputs over gRPC; serve an audio model with an SGLang version that supports gRPC media",
),
] {
if present && !supported {
return Err(client::invalid_request(message));
}
}
Ok(())
}

fn kv_hint_to_proto(
hint: &dynamo_backend_common::KvHint,
) -> Result<pb::KvHintsEnvelope, DynamoError> {
Expand Down Expand Up @@ -231,9 +326,24 @@ fn validate_request(request: &PreprocessedRequest) -> Result<(), DynamoError> {
if request.token_ids.is_empty() {
return Err(client::invalid_request("token_ids must not be empty"));
}
if request.multi_modal_data.is_some() || request.mm_processor_kwargs.is_some() {
if request.mm_processor_kwargs.is_some() {
return Err(client::invalid_request(
"mm_processor_kwargs are not supported by SGLang's native Generate RPC",
));
}
if request
.multi_modal_uuids
.iter()
.flat_map(|uuids| uuids.values().flatten())
.any(|uuid| uuid.as_deref().is_some_and(|uuid| !uuid.is_empty()))
{
return Err(client::invalid_request(
"media uuids are not supported by SGLang's native Generate RPC",
));
}
if request.encoder_result.is_some() {
return Err(client::invalid_request(
"multimodal payloads are not supported by SGLang's native Generate RPC",
"encoder handoff (encoder_result) is not supported by the SGLang sidecar",
));
}
if request.sampling_options.n.unwrap_or(1) != 1 {
Expand Down
Loading
Loading