From 7f2fcc0b08592fbcccedcc9f27225e1acc0198d9 Mon Sep 17 00:00:00 2001 From: Ratish P <114130421+Ratish1@users.noreply.github.com> Date: Mon, 6 Apr 2026 23:37:24 +0530 Subject: [PATCH] [VLM]: allow Qwen3.5 models for encoder disaggregation (#21849) --- .../srt/disaggregation/encode_server.py | 5 +- .../srt/multimodal/processors/qwen_vl.py | 2 +- python/sglang/srt/server_args.py | 2 + .../distributed/test_epd_disaggregation.py | 184 ++++++++++++++++++ 4 files changed, 190 insertions(+), 3 deletions(-) diff --git a/python/sglang/srt/disaggregation/encode_server.py b/python/sglang/srt/disaggregation/encode_server.py index b425d8c06..72b9f400a 100644 --- a/python/sglang/srt/disaggregation/encode_server.py +++ b/python/sglang/srt/disaggregation/encode_server.py @@ -867,10 +867,11 @@ class MMEncoder: ) # Get additional video metadata if ( - self.model_type in ["qwen3_vl", "qwen3_vl_moe"] + self.model_type + in ["qwen3_vl", "qwen3_vl_moe", "qwen3_5", "qwen3_5_moe"] and video_processor_kwargs.get("video_metadata", None) is not None ): - # For qwen3-vl models, we need to store the video timestamps + # For qwen3-vl/qwen3.5 models, we need to store the video timestamps video_metadata = video_processor_kwargs["video_metadata"] try: merge_size = ( diff --git a/python/sglang/srt/multimodal/processors/qwen_vl.py b/python/sglang/srt/multimodal/processors/qwen_vl.py index 3f102567d..6a757c33e 100644 --- a/python/sglang/srt/multimodal/processors/qwen_vl.py +++ b/python/sglang/srt/multimodal/processors/qwen_vl.py @@ -422,7 +422,7 @@ class QwenVLImageProcessor(SGLangBaseProcessor): audio_seq_lens = (audio_seq_lens - 2) // 2 + 1 if ( - self.model_type in ["qwen3_vl", "qwen3_vl_moe"] + self.model_type in ["qwen3_vl", "qwen3_vl_moe", "qwen3_5", "qwen3_5_moe"] and video_timestamps is not None ): input_ids, offsets, modality_list = self.build_input_ids_with_timestamps( diff --git a/python/sglang/srt/server_args.py b/python/sglang/srt/server_args.py index 646edb414..57dbc1998 100644 --- a/python/sglang/srt/server_args.py +++ b/python/sglang/srt/server_args.py @@ -3326,6 +3326,8 @@ class ServerArgs: "Qwen3VLForConditionalGeneration", "Qwen2_5_VLForConditionalGeneration", "Qwen3VLMoeForConditionalGeneration", + "Qwen3_5ForConditionalGeneration", + "Qwen3_5MoeForConditionalGeneration", "Qwen3OmniMoeForConditionalGeneration", "Qwen2AudioForConditionalGeneration", "Qwen2_5OmniForConditionalGeneration", diff --git a/test/registered/distributed/test_epd_disaggregation.py b/test/registered/distributed/test_epd_disaggregation.py index 9b069209a..301128e19 100644 --- a/test/registered/distributed/test_epd_disaggregation.py +++ b/test/registered/distributed/test_epd_disaggregation.py @@ -33,6 +33,7 @@ from sglang.test.vlm_utils import ( # Omni model for local testing; override via env var EPD_OMNI_MODEL DEFAULT_OMNI_MODEL = "Qwen/Qwen3-Omni-30B-A3B-Instruct" +QWEN35_27B_MODEL = "Qwen/Qwen3.5-27B" register_cuda_ci(est_time=150, suite="stage-c-test-4-gpu-h100") @@ -813,6 +814,189 @@ class TestEPDDisaggregationOneEncoder(PDDisaggregationServerBase): self.assertGreater(mmmu_accuracy, 0.40) +@unittest.skipIf( + is_in_ci(), + "Qwen3.5 EPD image/video test runs locally only", +) +class TestEPDDisaggregationQwen35(PDDisaggregationServerBase): + """EPD disaggregation test for Qwen3.5 image and video requests.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.process_encode = None + cls.model = QWEN35_27B_MODEL + cls.encode_port = f"{int(cls.lb_port) + 300}" + cls.encode_url = f"http://{cls.base_host}:{cls.encode_port}" + cls.language_url = cls.prefill_url + cls.image_man_ironing = IMAGE_MAN_IRONING_URL + cls.video_jobs = VIDEO_JOBS_URL + + print( + f"Setting up Qwen3.5 encoder disaggregation: model={cls.model}, " + f"encode={cls.encode_port}, language={cls.prefill_port}" + ) + + cls.start_encode() + cls.start_prefill() + + cls.wait_server_ready(cls.encode_url + "/health", process=cls.process_encode) + cls.wait_server_ready(cls.language_url + "/health", process=cls.process_prefill) + + cls.api_key = "sk-123456" + + @classmethod + def start_encode(cls): + encode_args = [ + "--trust-remote-code", + "--encoder-only", + "--encoder-transfer-backend", + "zmq_to_scheduler", + "--tp", + "1", + "--port", + cls.encode_port, + "--reasoning-parser", + "qwen3", + "--model-loader-extra-config", + '{"enable_multithread_load": true,"num_threads": 64}', + ] + cls.process_encode = popen_launch_server( + cls.model, + base_url=cls.encode_url, + timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, + other_args=encode_args, + ) + + @classmethod + def start_prefill(cls): + language_args = [ + "--trust-remote-code", + "--language-only", + "--encoder-urls", + cls.encode_url, + "--encoder-transfer-backend", + "zmq_to_scheduler", + "--tp", + "1", + "--base-gpu-id", + "1", + "--port", + cls.prefill_port, + "--reasoning-parser", + "qwen3", + "--model-loader-extra-config", + '{"enable_multithread_load": true,"num_threads": 64}', + ] + cls.process_prefill = popen_launch_server( + cls.model, + base_url=cls.language_url, + timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, + other_args=language_args, + ) + + @classmethod + def tearDownClass(cls): + if cls.process_lb: + kill_process_tree(cls.process_lb.pid) + if cls.process_decode: + kill_process_tree(cls.process_decode.pid) + if cls.process_prefill: + kill_process_tree(cls.process_prefill.pid) + if cls.process_encode: + kill_process_tree(cls.process_encode.pid) + + def _client(self): + return openai.Client(api_key=self.api_key, base_url=f"{self.language_url}/v1") + + def test_image(self): + client = self._client() + response = client.chat.completions.create( + model="default", + messages=[ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": {"url": self.image_man_ironing}, + }, + { + "type": "text", + "text": "Describe this image in a sentence.", + }, + ], + }, + ], + temperature=0, + max_tokens=256, + extra_body={"reasoning_effort": "none"}, + ) + text = response.choices[0].message.content + print(f"[Qwen3.5 EPD] Image response:\n{text}") + self.assertIsNotNone(text) + self.assertGreater(len(text), 0) + + text_lower = text.lower() + self.assertTrue( + any(w in text_lower for w in ("man", "person", "driver")), + f"Image response should mention a person: {text}", + ) + self.assertTrue( + any(w in text_lower for w in ("iron", "cloth", "hang", "holding")), + f"Image response should mention ironing/clothes: {text}", + ) + + def test_video(self): + client = self._client() + response = client.chat.completions.create( + model="default", + messages=[ + { + "role": "user", + "content": [ + {"type": "text", "text": "Describe the video."}, + { + "type": "video_url", + "video_url": {"url": self.video_jobs}, + }, + ], + }, + ], + max_tokens=1024, + stream=False, + ) + text = response.choices[0].message.content + print(f"[Qwen3.5 EPD] Video response:\n{text}") + self.assertIsNotNone(text) + self.assertGreater(len(text), 0) + + text_lower = text.lower() + self.assertTrue( + any( + w in text_lower + for w in ("ipod", "device", "microphone", "smartphone", "phone") + ), + f"Video response should mention a device: {text}", + ) + self.assertTrue( + any( + w in text_lower + for w in ( + "man", + "person", + "individual", + "speaker", + "presenter", + "steve", + "hand", + "hands", + ) + ), + f"Video response should mention a person: {text}", + ) + + class TestEPDDisaggregationMultiEncoders(PDDisaggregationServerBase): """ Test EPD disaggregation with multiple encode servers for load balancing.