From 6d1d7c6ae4f7eb9978791e559bb7e6195f5eca6f Mon Sep 17 00:00:00 2001 From: zhaopenghao Date: Wed, 29 Jul 2026 02:25:30 +0000 Subject: [PATCH] [Feat] Add Qwen3.5 SFT example Add text and multimodal Qwen3.5 SFT configuration and an 8-GPU launcher. Support video_url samples and ignore empty text samples discovered during end-to-end training validation. --- examples/v1/config/sft_qwen3p5.py | 143 ++++++++++++++++++ examples/v1/scripts/run_sft.sh | 46 ++++++ .../datasets/test_openai_tokenize_fn_empty.py | 12 ++ tests/datasets/test_qwen35_vl_tokenize_fn.py | 30 ++++ xtuner/v1/data_proto/messages/qwen35_chat.py | 4 +- .../mllm_tokenize_fn/qwen3_vl_tokenize_fn.py | 5 +- xtuner/v1/datasets/sft_tokenize_fn/openai.py | 4 + 7 files changed, 240 insertions(+), 4 deletions(-) create mode 100644 examples/v1/config/sft_qwen3p5.py create mode 100644 examples/v1/scripts/run_sft.sh create mode 100644 tests/datasets/test_openai_tokenize_fn_empty.py diff --git a/examples/v1/config/sft_qwen3p5.py b/examples/v1/config/sft_qwen3p5.py new file mode 100644 index 0000000000..9075f4953d --- /dev/null +++ b/examples/v1/config/sft_qwen3p5.py @@ -0,0 +1,143 @@ +import os + +from xtuner.v1.config import FSDPConfig, LRConfig, MuonConfig +from xtuner.v1.datasets import OpenaiTokenizeFunctionConfig, Qwen3VLTokenizeFnConfig +from xtuner.v1.datasets.config import DataloaderConfig, DatasetConfig +from xtuner.v1.float8.config import Float8Config, ScalingGranularity +from xtuner.v1.loss import CELossConfig +from xtuner.v1.model import Qwen3_5_VLMoE35BA3Config +from xtuner.v1.model.moe.moe import MTPConfig +from xtuner.v1.train import ResumeConfig, TrainerConfig + + +def _get_bool_env(name: str, default: bool = False) -> bool: + return os.environ.get(name, "1" if default else "0").lower() in ("1", "true", "yes", "on") + + +WORK_DIR = os.environ.get("WORK_DIR", "work_dirs/qwen3p5_sft") +QWEN3_5_MOE_PATH = os.environ["QWEN3_5_MOE_PATH"] +ALPACA_PATH = os.environ.get("ALPACA_PATH") +DATA_PATH = os.environ.get("DATA_PATH", ALPACA_PATH) +MEDIA_ROOT = os.environ.get("MEDIA_ROOT", "") +if DATA_PATH is None: + raise RuntimeError("DATA_PATH or ALPACA_PATH is required") + +sample_max_length = int(os.environ.get("SAMPLE_MAX_LENGTH", str(128 * 1024))) +pack_max_length = int(os.environ.get("PACK_MAX_LENGTH", str(128 * 1024))) +global_batch_size = int(os.environ.get("GLOBAL_BATCH_SIZE", "16")) +ep_size = int(os.environ.get("EP_SIZE", "1")) +model_compile = _get_bool_env("MODEL_COMPILE", True) +total_step = int(os.environ["TOTAL_STEP"]) if "TOTAL_STEP" in os.environ else None +is_multimodal = bool(MEDIA_ROOT) + +model_cfg = Qwen3_5_VLMoE35BA3Config(only_llm_forward=not is_multimodal, compile_cfg=model_compile) +model_cfg.text_config.ep_size = ep_size +if ep_size > 1: + model_cfg.text_config.dispatcher = os.environ.get("DISPATCHER", "deepep") + +if _get_bool_env("FP8"): + model_cfg.text_config.float8_cfg = Float8Config( + scaling_granularity_gemm=ScalingGranularity.TILEWISE, + scaling_granularity_grouped_gemm=ScalingGranularity.TILEWISE, + ) + +# Shared physical weights retain the reference config's four normal MTP prediction depths. +normal_mtp_layers = int(os.environ.get("NORMAL_MTP_LAYERS", "4")) +model_cfg.text_config.mtp_config = MTPConfig( + num_layers=normal_mtp_layers, + share_weights=normal_mtp_layers > 1, + loss_scaling_factor=float(os.environ.get("NORMAL_MTP_FACTOR", "1.0")), +) + +# Qwen3VLTokenizeFnConfig is for multimodal data. Pure-text data such as +# Alpaca can use OpenaiTokenizeFunctionConfig. +if is_multimodal: + tokenize_fn = Qwen3VLTokenizeFnConfig( + chat_template="qwen3.5-vl", + llm_pack_weight=-3.2, + visual_pack_weight=5.0, + max_length=sample_max_length, + processor_path=QWEN3_5_MOE_PATH, + rand_video_max_frames=int(os.environ.get("RAND_VIDEO_MAX_FRAMES", "24")), + max_pixels=int(os.environ.get("MAX_PIXELS", str(16384 * 32 * 32))), + ) +else: + tokenize_fn = OpenaiTokenizeFunctionConfig( + chat_template="qwen3.5-vl", + max_length=sample_max_length, + ) + +dataset_config = [ + { + "dataset": DatasetConfig( + name="multimodal" if is_multimodal else "alpaca", + anno_path=DATA_PATH, + class_name="VLMJsonlDataset" if is_multimodal else "JsonlDataset", + media_root=MEDIA_ROOT, + sample_ratio=float(os.environ.get("DATASET_SAMPLE_RATIO", "1.0")), + cache_dir=os.path.join(WORK_DIR, "jsonl_cache"), + cache_tag=os.environ.get( + "CACHE_TAG", + f"qwen3p5_{'vl' if is_multimodal else 'text'}_{sample_max_length}", + ), + ), + "tokenize_fn": tokenize_fn, + } +] + +dataloader_config = DataloaderConfig( + dataset_config_list=dataset_config, + pack_level="soft", + pack_max_length=pack_max_length, + pack_to_max_length=True, + pack_chunk_size=int(os.environ.get("PACK_CHUNK_SIZE", "10000")), + pack_workers=int(os.environ.get("PACK_WORKERS", "4")), + global_pack=_get_bool_env("GLOBAL_PACK", True), + group_by_length=_get_bool_env("GROUP_BY_LENGTH", True), + collator="qwen3_vl_sft_collator" if is_multimodal else "sft_llm_collator", + pack_extra_buffer_size=int(os.environ.get("PACK_EXTRA_BUFFER_SIZE", "20")), + num_workers=int(os.environ.get("DATALOADER_NUM_WORKERS", "4")), +) + +optim_cfg = MuonConfig( + lr=float(os.environ.get("LR", "2e-5")), + weight_decay=float(os.environ.get("WEIGHT_DECAY", "0.05")), +) +lr_cfg = LRConfig( + lr_type="cosine", + warmup_ratio=float(os.environ.get("WARMUP_RATIO", "0.1")), + lr_min=float(os.environ.get("LR_MIN", "1e-6")), +) +fsdp_cfg = FSDPConfig( + recompute_ratio=float(os.environ.get("RECOMPUTE_RATIO", "1.0")), + torch_compile=model_compile, + ep_size=ep_size, + checkpoint_preserve_rng_state=False, +) +loss_cfg = CELossConfig( + mode="chunk", + chunk_size=int(os.environ.get("LOSS_CHUNK_SIZE", "1024")), + loss_reduction=os.environ.get("LOSS_REDUCTION", "square"), +) + +trainer = TrainerConfig( + model_cfg=model_cfg, + load_from=QWEN3_5_MOE_PATH, + tokenizer_path=QWEN3_5_MOE_PATH, + resume_cfg=ResumeConfig(auto_resume=True), + fsdp_cfg=fsdp_cfg, + optim_cfg=optim_cfg, + dataloader_cfg=dataloader_config, + lr_cfg=lr_cfg, + loss_cfg=loss_cfg, + global_batch_size=global_batch_size, + total_step=total_step, + total_epoch=None if total_step is not None else int(os.environ.get("TOTAL_EPOCH", "1")), + sp_size=int(os.environ.get("SP_SIZE", "4")), + checkpoint_interval=int(os.environ.get("CHECKPOINT_INTERVAL", "500")), + checkpoint_maxkeep=int(os.environ.get("CHECKPOINT_MAX_KEEP", "2")), + hf_interval=int(os.environ.get("HF_INTERVAL", "500")), + hf_max_keep=int(os.environ.get("HF_MAX_KEEP", "2")), + work_dir=WORK_DIR, + debug_skip_save=_get_bool_env("DEBUG_SKIP_SAVE"), +) diff --git a/examples/v1/scripts/run_sft.sh b/examples/v1/scripts/run_sft.sh new file mode 100644 index 0000000000..da805186e6 --- /dev/null +++ b/examples/v1/scripts/run_sft.sh @@ -0,0 +1,46 @@ +#!/usr/bin/env bash +set -euo pipefail + +# QWEN3_5_MOE_PATH and ALPACA_PATH can be initialized by sourcing zdev/env.sh. +# Set DATA_PATH and MEDIA_ROOT instead when training multimodal data. +: "${QWEN3_5_MOE_PATH:?QWEN3_5_MOE_PATH is required}" +: "${DATA_PATH:=${ALPACA_PATH:?DATA_PATH or ALPACA_PATH is required}}" +export QWEN3_5_MOE_PATH DATA_PATH +export MEDIA_ROOT="${MEDIA_ROOT:-}" + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "${SCRIPT_DIR}/../../.." && pwd)" +CONFIG_PATH="${1:-${REPO_ROOT}/examples/v1/config/sft_qwen3p5.py}" +export WORK_DIR="${2:-${WORK_DIR:-work_dirs/qwen3p5_sft}}" + +XTUNER_PATH="${XTUNER_PATH:-${REPO_ROOT}}" +export PYTHONPATH="${XTUNER_PATH}${PYTHONPATH:+:${PYTHONPATH}}" + +export XTUNER_ACTIVATION_OFFLOAD="${XTUNER_ACTIVATION_OFFLOAD:-0}" +export XTUNER_GC_ENABLE="${XTUNER_GC_ENABLE:-1}" +export XTUNER_USE_FA3="${XTUNER_USE_FA3:-1}" +# FA3 does not support deterministic backward for Qwen3.5's head_dim=256. +if [[ "${XTUNER_USE_FA3}" == "1" ]]; then + export XTUNER_DETERMINISTIC=false +fi +export TORCH_LOGS="${TORCH_LOGS:-recompiles}" +export PYTORCH_CUDA_ALLOC_CONF="${PYTORCH_CUDA_ALLOC_CONF:-expandable_segments:True}" + +NNODES="${NNODES:-${NODE_COUNT:-1}}" +NODE_RANK="${NODE_RANK:-0}" +MASTER_ADDR="${MASTER_ADDR:-127.0.0.1}" +MASTER_PORT="${MASTER_PORT:-6000}" + +ulimit -n 65536 +mkdir -p "${WORK_DIR}" + +torchrun \ + --nproc-per-node="${NPROC_PER_NODE:-8}" \ + --master-addr="${MASTER_ADDR}" \ + --master-port="${MASTER_PORT}" \ + --nnodes="${NNODES}" \ + --node-rank="${NODE_RANK}" \ + --tee 3 \ + -m xtuner.v1.train.cli.sft \ + --config "${CONFIG_PATH}" \ + 2>&1 | tee -a "${WORK_DIR}/node_${NODE_RANK}.txt" diff --git a/tests/datasets/test_openai_tokenize_fn_empty.py b/tests/datasets/test_openai_tokenize_fn_empty.py new file mode 100644 index 0000000000..269b404e65 --- /dev/null +++ b/tests/datasets/test_openai_tokenize_fn_empty.py @@ -0,0 +1,12 @@ +from unittest.mock import Mock + +from transformers import PreTrainedTokenizer +from xtuner.v1.datasets import OpenaiTokenizeFunctionConfig + + +class TestOpenaiTokenizeFunctionEmptyMessages: + def test_empty_messages_are_marked_as_damaged_in_cache(self): + tokenize_fn = OpenaiTokenizeFunctionConfig(chat_template="qwen3.5-vl").build(Mock(spec=PreTrainedTokenizer)) + tokenize_fn.set_state("cache") + + assert tokenize_fn([]) == {"num_tokens": 0, "proxy_attn_flops": 0.0} diff --git a/tests/datasets/test_qwen35_vl_tokenize_fn.py b/tests/datasets/test_qwen35_vl_tokenize_fn.py index 3f8cbf5295..2149989766 100644 --- a/tests/datasets/test_qwen35_vl_tokenize_fn.py +++ b/tests/datasets/test_qwen35_vl_tokenize_fn.py @@ -215,6 +215,36 @@ def test_calc_frame_info(self): self.assertEqual(origin_fps_list, [20]) self.assertEqual(timestamps_list, [[0.25, 1.5]]) + def test_qwen35_vl_video_url_cache(self): + sample = { + "messages": [ + { + "role": "user", + "content": [ + { + "type": "video_url", + "video_url": { + "url": "tennis_frames_2fps/", + "image_wh": [1280, 720], + "origin_video_length": 182, + "origin_fps": 30.0, + "processed_video_length": 13, + "processed_fps": 2, + }, + }, + {"type": "text", "text": "Describe the video."}, + ], + }, + {"role": "assistant", "content": "A tennis match is being played."}, + ] + } + self.tokenize_fn.set_state("cache") + + result = self.tokenize_fn(sample) + + self.assertGreater(result["num_tokens"], 0) + self.assertGreater(sum(result["num_img_tokens"]), 0) + @parametrize.parametrize("add_vision_id", [(True,), (False,)]) def test_qwen3_vl_sft_video(self, add_vision_id): QWEN35_VL_PATH = os.environ["QWEN3_5_MOE_PATH"] diff --git a/xtuner/v1/data_proto/messages/qwen35_chat.py b/xtuner/v1/data_proto/messages/qwen35_chat.py index 99e5a8e241..6d2e052b0d 100644 --- a/xtuner/v1/data_proto/messages/qwen35_chat.py +++ b/xtuner/v1/data_proto/messages/qwen35_chat.py @@ -56,13 +56,13 @@ def render_content(content, do_vision_count, image_count, video_count, add_visio if add_vision_id: result += f"Picture {image_count}: " result += "<|vision_start|><|image_pad|><|vision_end|>" - elif "video" in item or item.get("type") == "video": + elif "video" in item or "video_url" in item or item.get("type") in ("video", "video_url"): if do_vision_count: video_count += 1 if add_vision_id: result += f"Video {video_count}: " - video_content = item.get("video", {}) + video_content = item.get("video", item.get("video_url", {})) assert isinstance(video_content, dict), f"video_content must be a dict, but got {type(video_content)}" timestamps = video_content.get("timestamps", []) if len(timestamps) > 0: diff --git a/xtuner/v1/datasets/mllm_tokenize_fn/qwen3_vl_tokenize_fn.py b/xtuner/v1/datasets/mllm_tokenize_fn/qwen3_vl_tokenize_fn.py index d89e74af8f..1e3f33545e 100644 --- a/xtuner/v1/datasets/mllm_tokenize_fn/qwen3_vl_tokenize_fn.py +++ b/xtuner/v1/datasets/mllm_tokenize_fn/qwen3_vl_tokenize_fn.py @@ -217,8 +217,9 @@ def replace_video_timestamps_and_num_frame( content = msg["content"] if isinstance(content, list): for item in content: - if "video" in item: - video_content = item["video"] + if "video" in item or "video_url" in item: + video_content = item.get("video", item.get("video_url")) + assert isinstance(video_content, dict) if len(timestamps_list) > 0: timestamps = timestamps_list[video_cnt] video_content["timestamps"] = timestamps diff --git a/xtuner/v1/datasets/sft_tokenize_fn/openai.py b/xtuner/v1/datasets/sft_tokenize_fn/openai.py index 1ca9f44a6c..38253271e3 100644 --- a/xtuner/v1/datasets/sft_tokenize_fn/openai.py +++ b/xtuner/v1/datasets/sft_tokenize_fn/openai.py @@ -47,6 +47,10 @@ def __call__(self, item: dict | list, **kwargs) -> DataItem | CacheItem: if isinstance(item, dict) and "dialogs" in item: item = item["dialogs"] + # JsonlDataset removes damaged samples whose cached token count is zero. + if not item: + return CacheItem(num_tokens=0) + if self.chat_template_name == "qwen3.5-vl": messages = Qwen35ChatMessages(messages=item, tools=tools) elif self.chat_template_name == "glm5.2":