diff --git a/.claude/skills/benchmark-model-kernels b/.claude/skills/benchmark-model-kernels new file mode 120000 index 00000000000..1bfd1fefe85 --- /dev/null +++ b/.claude/skills/benchmark-model-kernels @@ -0,0 +1 @@ +../../.agents/skills/benchmark-model-kernels \ No newline at end of file diff --git a/examples/speculative_decoding/distributed_generate/launch.sh b/examples/speculative_decoding/distributed_generate/launch.sh index 463f4c2a387..c93f59906e4 100644 --- a/examples/speculative_decoding/distributed_generate/launch.sh +++ b/examples/speculative_decoding/distributed_generate/launch.sh @@ -14,6 +14,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +set -euo pipefail + if [ $# -lt 9 ]; then echo "Usage: $0 " echo "Example: $0 245387 vllm /model/ /input_data/ /output_data/ /scripts/ 0 20 cluster-01,cluster-02 "\"You are a helpful assistant.\""" @@ -32,6 +34,9 @@ NODE_NAME=$9 SYSTEM_PROMPT="${10:-}" IFS=',' read -r -a NODE_LIST <<< "$NODE_NAME" +# Pyxis requires bind-mount sources to exist before srun creates the container. +mkdir -p "$OUTPUT_PATH" + # backend needs to be either vllm or sglang if [ "$BACKEND" != "vllm" ] && [ "$BACKEND" != "sglang" ]; then echo "Invalid backend: $BACKEND" @@ -39,22 +44,37 @@ if [ "$BACKEND" != "vllm" ] && [ "$BACKEND" != "sglang" ]; then fi if [ "$BACKEND" == "vllm" ]; then - CONTAINER_IMAGE="vllm/vllm-openai:v0.8.5" + DEFAULT_CONTAINER_IMAGE="vllm/vllm-openai:v0.24.0" else - CONTAINER_IMAGE="lmsysorg/sglang:v0.4.6.post2-cu124" + DEFAULT_CONTAINER_IMAGE="lmsysorg/sglang:v0.5.3-cu129" fi +CONTAINER_IMAGE=${CONTAINER_IMAGE:-$DEFAULT_CONTAINER_IMAGE} counter=$START_SHARD +worker_pids=() for node in "${NODE_LIST[@]}"; do echo "Processing node: $node" - srun --output=srun_worker_${node}.log --jobid=$JOB_ID -N 1 --ntasks=1 --ntasks-per-node=1 -w $node \ - --mpi pmix --overlap --container-image=$CONTAINER_IMAGE \ - --container-mounts=$MODEL_PATH:/model/,$DATA_PATH:/input_data/,$OUTPUT_PATH:/output_data/,$SCRIPTS_PATH:/scripts/ \ - bash /scripts/distributed_generate/worker.sh $counter $BACKEND $JOBS_PER_NODE "$SYSTEM_PROMPT" & + srun --output="srun_worker_${node}.log" --jobid="$JOB_ID" -N 1 --ntasks=1 --ntasks-per-node=1 -w "$node" \ + --mpi pmix --overlap --container-image="$CONTAINER_IMAGE" \ + --container-mounts="$MODEL_PATH":/model/,"$DATA_PATH":/input_data/,"$OUTPUT_PATH":/output_data/,"$SCRIPTS_PATH":/scripts/ \ + bash /scripts/distributed_generate/worker.sh "$counter" "$BACKEND" "$JOBS_PER_NODE" "$SYSTEM_PROMPT" & echo "srun command for node $node started with PID $!" >> srun_launch.log + worker_pids+=("$!") # increment counter by JOBS_PER_NODE counter=$((counter + JOBS_PER_NODE)) done echo "Started workers, each processing $JOBS_PER_NODE shards of data. Will process shards $START_SHARD through $((counter - 1))." + +worker_status=0 +for worker_pid in "${worker_pids[@]}"; do + if ! wait "$worker_pid"; then + worker_status=1 + fi +done + +if [ "$worker_status" -ne 0 ]; then + echo "ERROR: one or more workers failed." >&2 +fi +exit "$worker_status" diff --git a/examples/speculative_decoding/distributed_generate/launch_multimodal.sh b/examples/speculative_decoding/distributed_generate/launch_multimodal.sh new file mode 100755 index 00000000000..23558644c2a --- /dev/null +++ b/examples/speculative_decoding/distributed_generate/launch_multimodal.sh @@ -0,0 +1,96 @@ +#!/bin/bash +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +set -euo pipefail + +if [ $# -lt 10 ]; then + echo "Usage: $0 [num_frames] [system_prompt]" + echo "Also accepted: $0 [num_frames] [system_prompt]" + echo "Example: $0 245387 sglang /model/ /shards/ /output/ /scripts/ 0 10 /media/ 16 cluster-01" + echo "Optional env: SGLANG_TP_SIZE=8 NUM_TEMPERATURES=8 NUM_THREADS=8 SGLANG_EXTRA_ARGS='--mem-fraction-static 0.75'" + exit 1 +fi + +JOB_ID=$1 +BACKEND=$2 +MODEL_PATH=$3 +DATA_PATH=$4 +OUTPUT_PATH=$5 +SCRIPTS_PATH=$6 +START_SHARD=$7 +JOBS_PER_NODE=$8 +ARG9=$9 +ARG10=${10:-} +ARG11=${11:-} +ARG12=${12:-} + +if [[ "$ARG9" == */* || "$ARG9" == .* ]]; then + MEDIA_PATH=$ARG9 + NUM_FRAMES="${ARG10:-32}" + NODE_NAME=$ARG11 + SYSTEM_PROMPT="$ARG12" +else + NODE_NAME=$ARG9 + MEDIA_PATH=$ARG10 + NUM_FRAMES="${ARG11:-32}" + SYSTEM_PROMPT="$ARG12" +fi + +if [ -z "${NODE_NAME:-}" ] || [ -z "${MEDIA_PATH:-}" ]; then + echo "ERROR: both media_path and node_name are required." >&2 + exit 1 +fi + +IFS=',' read -r -a NODE_LIST <<< "$NODE_NAME" + +if [ "$BACKEND" != "sglang" ]; then + echo "Multimodal generation currently supports backend=sglang." + exit 1 +fi + +mkdir -p "$OUTPUT_PATH" + +# Set CONTAINER_IMAGE to a local .sqsh image to avoid pulling from the registry. +DEFAULT_CONTAINER_IMAGE="lmsysorg/sglang:v0.5.3-cu129" +CONTAINER_IMAGE="${CONTAINER_IMAGE:-$DEFAULT_CONTAINER_IMAGE}" + +counter=$START_SHARD +worker_pids=() +for node in "${NODE_LIST[@]}"; do + echo "Processing node: $node" + srun --output=srun_vlm_worker_${node}.log --jobid=$JOB_ID -N 1 --ntasks=1 --ntasks-per-node=1 -w "$node" \ + --mpi pmix --overlap --container-image="$CONTAINER_IMAGE" \ + --container-mounts="$MODEL_PATH":/model/,"$DATA_PATH":/input_data/,"$OUTPUT_PATH":/output_data/,"$SCRIPTS_PATH":/scripts/,"$MEDIA_PATH":/media_data/ \ + bash /scripts/distributed_generate/worker_multimodal.sh "$counter" "$BACKEND" "$JOBS_PER_NODE" "$NUM_FRAMES" "$SYSTEM_PROMPT" & + + echo "srun multimodal command for node $node started with PID $!" >> srun_launch_multimodal.log + worker_pids+=("$!") + counter=$((counter + JOBS_PER_NODE)) +done + +echo "Started multimodal workers, each processing $JOBS_PER_NODE shards of data. Will process shards $START_SHARD through $((counter - 1))." + +worker_status=0 +for worker_pid in "${worker_pids[@]}"; do + if ! wait "$worker_pid"; then + worker_status=1 + fi +done + +if [ "$worker_status" -ne 0 ]; then + echo "ERROR: one or more multimodal workers failed." >&2 +fi +exit "$worker_status" diff --git a/examples/speculative_decoding/distributed_generate/server_generate_vlm_sglang.py b/examples/speculative_decoding/distributed_generate/server_generate_vlm_sglang.py new file mode 100755 index 00000000000..6987c5b09d7 --- /dev/null +++ b/examples/speculative_decoding/distributed_generate/server_generate_vlm_sglang.py @@ -0,0 +1,589 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Generate multimodal SFT data from video prompts using SGLang native video input.""" + +from __future__ import annotations + +import argparse +import concurrent.futures +import json +import os +import sys +import traceback +from pathlib import Path +from typing import Any +from urllib.parse import quote + +import tqdm + +QWEN_IMAGE_TOKEN = "<|vision_start|><|image_pad|><|vision_end|>" +_UNRESOLVED_MEDIA_PATHS: set[str] = set() + + +def _load_json_or_jsonl(path: str) -> list[dict[str, Any]]: + if path.endswith("jsonl"): + with open(path, encoding="utf-8") as f: + return [json.loads(line) for line in f if line.strip()] + with open(path, encoding="utf-8") as f: + data = json.load(f) + if not isinstance(data, list): + raise ValueError(f"Expected list data in {path}") + return data + + +def _first_user_message(sample: dict[str, Any]) -> dict[str, Any]: + messages = sample.get("messages") or sample.get("conversations") or sample.get("conversation") + if not isinstance(messages, list): + raise ValueError(f"Sample has no messages/conversations list: keys={sorted(sample)}") + for message in messages: + role = (message.get("role") or message.get("from") or "").lower() + if role in ("user", "human"): + return message + raise ValueError("Sample has no user message") + + +def _extract_message_text_and_media(sample: dict[str, Any]) -> tuple[str, str | None, str | None]: + video_path = sample.get("video_path") + image_path = sample.get("image_path") or sample.get("image") + + message = _first_user_message(sample) + content = message.get("content") or message.get("value") + text_parts: list[str] = [] + + if isinstance(content, str): + text_parts.append(content) + elif isinstance(content, list): + for item in content: + if not isinstance(item, dict): + continue + item_type = item.get("type") + if item_type == "text" and isinstance(item.get("text"), str): + text_parts.append(item["text"]) + elif item_type == "video": + video_path = item.get("video") or video_path + elif item_type == "image": + image_path = item.get("image") or image_path + else: + raise ValueError(f"Unsupported user content: {content!r}") + + prompt = "\n".join(part.strip() for part in text_parts if part.strip()) + if not prompt: + raise ValueError("Could not extract text prompt from sample") + return prompt, video_path, image_path + + +def _extract_text_and_media(sample: dict[str, Any]) -> tuple[str, str | None, str | None]: + prompt = sample.get("prompt") + text, video_path, image_path = _extract_message_text_and_media(sample) + if isinstance(prompt, str) and prompt.strip(): + text = prompt.strip() + image_path = sample.get("image_path") or sample.get("image") or image_path + if image_path: + return text, None, image_path + return text, sample.get("video_path") or video_path, None + + +def _resolve_media_path( + path: str | None, media_root: str | None, input_root: str | None +) -> str | None: + if not path: + return None + if path.startswith(("http://", "https://", "data:")): + return path + candidate = Path(path) + if candidate.is_absolute() and candidate.exists(): + return str(candidate) + if input_root: + rooted = Path(input_root) / path + if rooted.exists(): + return str(rooted) + if media_root: + rooted = Path(media_root) / path + if rooted.exists(): + return str(rooted) + # If the record stores an absolute host path, preserve its suffix under media_root. + parts = candidate.parts + if "videos" in parts: + suffix = Path(*parts[parts.index("videos") :]) + rooted = Path(media_root) / suffix + if rooted.exists(): + return str(rooted) + if candidate.exists(): + return str(candidate) + if path not in _UNRESOLVED_MEDIA_PATHS: + print(f"WARNING: could not resolve media path: {path}") + _UNRESOLVED_MEDIA_PATHS.add(path) + return None + + +def _as_openai_media_value( + path: str, media_url_base: str | None, media_root: str | None, input_root: str | None +) -> str: + if path.startswith(("http://", "https://", "data:")): + return path + if media_url_base: + candidate = Path(path) + if candidate.is_absolute(): + if media_root: + try: + path = str(candidate.relative_to(media_root)) + except ValueError: + # The local HTTP server deliberately exposes only media_root. + # Leave paths outside it local for SGLang to resolve directly. + return path + else: + return f"{media_url_base.rstrip('/')}{quote(path, safe='/')}" + return f"{media_url_base.rstrip('/')}/{quote(path, safe='/')}" + # Do not convert local paths to file://. This SGLang build falls through to + # the base64 loader for file:// videos and raises "Incorrect padding". + return path + + +def _openai_chat_url(url: str) -> str: + url = url.rstrip("/") + if url.endswith("/v1/chat/completions"): + return url + if url.endswith("/v1"): + return f"{url}/chat/completions" + return f"{url}/v1/chat/completions" + + +def _messages_for_output(sample: dict[str, Any], answer: str) -> list[dict[str, Any]]: + messages = sample.get("messages") + if isinstance(messages, list): + output_messages = list(messages) + else: + prompt, video_path, image_path = _extract_text_and_media(sample) + content: list[dict[str, Any]] = [{"type": "text", "text": prompt}] + if image_path: + content.append({"type": "image", "image": image_path}) + elif video_path: + content.append({"type": "video", "video": video_path, "fps": 4}) + output_messages = [{"role": "user", "content": content}] + + output_messages.append({"role": "assistant", "content": answer}) + return output_messages + + +def _coerce_text(value: Any) -> str: + if value is None: + return "" + if type(value).__name__ == "ProgramState": + return "" + if isinstance(value, str): + return value.strip() + if isinstance(value, dict): + for key in ("answer", "text", "value", "content"): + text = _coerce_text(value.get(key)) + if text: + return text + return "" + for attr_name in ("text", "value", "content"): + attr = getattr(value, attr_name, None) + if callable(attr): + try: + text = _coerce_text(attr()) + if text: + return text + except Exception: + pass + else: + text = _coerce_text(attr) + if text: + return text + return "" + + +def _answer_from_messages(messages: Any) -> str: + if callable(messages): + try: + messages = messages() + except Exception: + return "" + if not isinstance(messages, list): + return "" + for message in reversed(messages): + if not isinstance(message, dict): + continue + role = (message.get("role") or "").lower() + if role != "assistant": + continue + return _coerce_text(message.get("content")) + return "" + + +def _state_answer(state: Any) -> str: + # SGLang ProgramState normally supports state["answer"]. Some versions or + # failure paths expose variables through helper methods/attributes instead. + try: + text = _coerce_text(state["answer"]) + if text: + return text + except Exception: + pass + + if isinstance(state, dict): + return _coerce_text(state.get("answer") or state.get("value") or state.get("text")) + + for method_name in ("get", "get_var", "get_variable", "var"): + method = getattr(state, method_name, None) + if not callable(method): + continue + try: + text = _coerce_text(method("answer")) + if text: + return text + except Exception: + pass + + text = _answer_from_messages(getattr(state, "messages", None)) + if text: + return text + + for attr_name in ("answer", "variables", "vars"): + text = _coerce_text(getattr(state, attr_name, None)) + if text: + return text + + return "" + + +def _prompt_with_vision_token(prompt: str, media_type: str, token_format: str) -> str: + if token_format == "none": + return prompt + if token_format != "qwen_vl": + raise ValueError(f"Unsupported vision token format: {token_format}") + if media_type == "video": + # SGLang's native sgl.video(...) transport binds video data using its + # own placeholder path. Adding a literal Qwen <|video_pad|> token here + # makes the server look for an extra unbound video iterator. + return prompt + + known_tokens = ( + "<|image_pad|>", + "<|video_pad|>", + "", + "