main
py 553 lines 18 KB
Raw
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 """Tests for Moondream3 multimodal processing.
4
5 Includes:
6 - Processor creation and application tests
7 - Image tokenization and placeholder expansion tests
8 - Tiling and cropping logic tests (CPU-based)
9 - Pixel normalization tests
10 """
11
12 import numpy as np
13 import pytest
14 import torch
15
16 from vllm.multimodal import MULTIMODAL_REGISTRY
17
18 from ....conftest import ImageTestAssets
19 from ...utils import build_model_context
20
21 MOONDREAM3_MODEL_ID = "moondream/moondream3-preview"
22 # Expected multimodal prefix: BOS + 729 image tokens.
23 EXPECTED_IMAGE_TOKENS = 730
24 # Vision encoder constants
25 CROP_SIZE = 378
26 PATCH_SIZE = 14
27 MAX_CROPS = 12
28
29
30 @pytest.mark.parametrize("model_id", [MOONDREAM3_MODEL_ID])
31 def test_processor_creation(model_id: str):
32 """Test that Moondream3 processor can be created."""
33 ctx = build_model_context(
34 model_id,
35 limit_mm_per_prompt={"image": 1},
36 )
37 processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config)
38 assert processor is not None
39
40
41 @pytest.mark.parametrize("model_id", [MOONDREAM3_MODEL_ID])
42 def test_processor_apply(
43 image_assets: ImageTestAssets,
44 model_id: str,
45 ):
46 """Test that Moondream3 processor can process inputs.
47
48 NOTE: The prompt includes the leading BOS token because Moondream3
49 pre-fills BOS and image embeddings together.
50 """
51 ctx = build_model_context(
52 model_id,
53 limit_mm_per_prompt={"image": 1},
54 )
55 processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config)
56
57 prompt = "<|endoftext|><image><|md_reserved_0|>query<|md_reserved_1|>What is this?<|md_reserved_2|>" # noqa: E501
58 mm_data = {"image": [image_assets[0].pil_image]}
59
60 processed_inputs = processor(
61 prompt,
62 mm_items=processor.info.parse_mm_data(mm_data),
63 hf_processor_mm_kwargs={},
64 )
65
66 assert "prompt_token_ids" in processed_inputs
67 image_placeholders = processed_inputs["mm_placeholders"]["image"]
68 assert len(image_placeholders) == 1
69 assert image_placeholders[0].length == EXPECTED_IMAGE_TOKENS
70
71
72 @pytest.mark.parametrize("model_id", [MOONDREAM3_MODEL_ID])
73 def test_processor_pixel_values(
74 image_assets: ImageTestAssets,
75 model_id: str,
76 ):
77 """Test that pixel values are correctly produced."""
78 ctx = build_model_context(
79 model_id,
80 limit_mm_per_prompt={"image": 1},
81 )
82 processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config)
83
84 prompt = "<|endoftext|><image><|md_reserved_0|>query<|md_reserved_1|>What is this?<|md_reserved_2|>" # noqa: E501
85 mm_data = {"image": [image_assets[0].pil_image]}
86
87 processed_inputs = processor(
88 prompt,
89 mm_items=processor.info.parse_mm_data(mm_data),
90 hf_processor_mm_kwargs={},
91 )
92
93 # Check mm_kwargs contains pixel_values
94 mm_kwargs = processed_inputs.get("mm_kwargs")
95 assert mm_kwargs is not None
96 mm_data_result = mm_kwargs.get_data()
97 assert "pixel_values" in mm_data_result
98
99 # Verify pixel_values shape
100 pixel_values = mm_data_result["pixel_values"]
101 assert pixel_values.dim() == 5 # [batch, num_crops, C, H, W]
102 assert pixel_values.shape[2] == 3 # RGB channels
103 assert pixel_values.shape[3] == 378 # crop height
104 assert pixel_values.shape[4] == 378 # crop width
105
106
107 @pytest.mark.parametrize("model_id", [MOONDREAM3_MODEL_ID])
108 def test_processor_image_token_expansion(
109 image_assets: ImageTestAssets,
110 model_id: str,
111 ):
112 """Test that <image> placeholder is expanded to correct number of tokens."""
113 ctx = build_model_context(
114 model_id,
115 limit_mm_per_prompt={"image": 1},
116 )
117 processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config)
118
119 prompt = "<|endoftext|><image><|md_reserved_0|>query<|md_reserved_1|>Describe.<|md_reserved_2|>" # noqa: E501
120 mm_data = {"image": [image_assets[0].pil_image]}
121
122 processed_inputs = processor(
123 prompt,
124 mm_items=processor.info.parse_mm_data(mm_data),
125 hf_processor_mm_kwargs={},
126 )
127 image_placeholders = processed_inputs["mm_placeholders"]["image"]
128 assert len(image_placeholders) == 1
129 assert image_placeholders[0].length == EXPECTED_IMAGE_TOKENS
130
131
132 @pytest.mark.parametrize("model_id", [MOONDREAM3_MODEL_ID])
133 def test_multi_crop_tiling(
134 model_id: str,
135 ):
136 """Test that large images produce correct multi-crop tiling."""
137 from PIL import Image
138
139 from vllm.transformers_utils.processors.moondream3 import Moondream3Processor
140
141 processor = Moondream3Processor.from_pretrained(model_id, trust_remote_code=True)
142
143 # Create a large image that requires multiple crops
144 large_image = Image.new("RGB", (1000, 1000), color="blue")
145 pixel_values, tiling = processor.preprocess_image(large_image)
146
147 # Large images should produce more than 1x1 tiling
148 assert tiling[0] >= 1 and tiling[1] >= 1
149 # Check that we have global crop + local crops
150 expected_crops = tiling[0] * tiling[1] + 1
151 assert pixel_values.shape[0] == expected_crops
152
153
154 @pytest.mark.parametrize(
155 "image_size",
156 [
157 (500, 500),
158 (800, 600),
159 (1920, 1080),
160 ],
161 )
162 @pytest.mark.parametrize("model_id", [MOONDREAM3_MODEL_ID])
163 def test_tiling_various_sizes(
164 image_size: tuple[int, int],
165 model_id: str,
166 ):
167 """Test tiling with various image sizes."""
168 from PIL import Image
169
170 from vllm.transformers_utils.processors.moondream3 import Moondream3Processor
171
172 processor = Moondream3Processor.from_pretrained(model_id, trust_remote_code=True)
173
174 width, height = image_size
175 image = Image.new("RGB", (width, height), color="red")
176 pixel_values, tiling = processor.preprocess_image(image)
177
178 # Basic shape checks
179 assert pixel_values.dim() == 4 # [num_crops, C, H, W]
180 assert pixel_values.shape[1] == 3 # RGB
181 assert pixel_values.shape[2] == 378 # crop height
182 assert pixel_values.shape[3] == 378 # crop width
183
184 # Tiling should respect max_crops (12)
185 assert tiling[0] * tiling[1] <= 12
186
187
188 @pytest.mark.parametrize("model_id", [MOONDREAM3_MODEL_ID])
189 def test_pixel_normalization(
190 model_id: str,
191 ):
192 """Test that pixel values are normalized to [-1, 1] range."""
193 from PIL import Image
194
195 from vllm.transformers_utils.processors.moondream3 import Moondream3Processor
196
197 processor = Moondream3Processor.from_pretrained(model_id, trust_remote_code=True)
198
199 # Create test image
200 image = Image.new("RGB", (378, 378), color="green")
201 pixel_values, _ = processor.preprocess_image(image)
202
203 # Normalization: (x - 0.5) / 0.5 = 2*x - 1
204 # For input [0, 1], output should be [-1, 1]
205 assert pixel_values.min() >= -1.0
206 assert pixel_values.max() <= 1.0
207
208
209 @pytest.mark.parametrize("model_id", [MOONDREAM3_MODEL_ID])
210 def test_chat_template_with_image(
211 image_assets: ImageTestAssets,
212 model_id: str,
213 ):
214 """Test that chat template correctly formats BOS + image + prompt."""
215 ctx = build_model_context(
216 model_id,
217 limit_mm_per_prompt={"image": 1},
218 )
219 processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config)
220 tokenizer = ctx.tokenizer
221
222 # Use the chat template format
223 prompt = "<|endoftext|><image><|md_reserved_0|>query<|md_reserved_1|>What is this?<|md_reserved_2|>" # noqa: E501
224 mm_data = {"image": [image_assets[0].pil_image]}
225
226 processed_inputs = processor(
227 prompt,
228 mm_items=processor.info.parse_mm_data(mm_data),
229 hf_processor_mm_kwargs={},
230 )
231 token_ids = processed_inputs["prompt_token_ids"]
232
233 # BOS token (<|endoftext|>) should be token ID 0
234 bos_token_id = tokenizer.encode("<|endoftext|>", add_special_tokens=False)[0]
235 assert bos_token_id == 0
236
237 # First token should be BOS
238 assert token_ids[0] == bos_token_id
239
240
241 @pytest.mark.parametrize(
242 "content",
243 [
244 pytest.param(
245 [
246 {
247 "type": "image_url",
248 "image_url": {"url": "https://example.invalid/image.png"},
249 },
250 {"type": "text", "text": "What is in this image?"},
251 ],
252 id="image-first",
253 ),
254 pytest.param(
255 [
256 {"type": "text", "text": "What is in this image?"},
257 {
258 "type": "image_url",
259 "image_url": {"url": "https://example.invalid/image.png"},
260 },
261 ],
262 id="text-first",
263 ),
264 ],
265 )
266 @pytest.mark.parametrize("model_id", [MOONDREAM3_MODEL_ID])
267 def test_chat_template_content_list_uses_moondream_image_prefix(
268 image_assets: ImageTestAssets,
269 content: list[dict[str, object]],
270 model_id: str,
271 ):
272 ctx = build_model_context(
273 model_id,
274 limit_mm_per_prompt={"image": 1},
275 )
276 processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config)
277 hf_processor = processor.info.get_hf_processor()
278
279 prompt = hf_processor.tokenizer.apply_chat_template(
280 [{"role": "user", "content": content}],
281 chat_template=hf_processor.chat_template,
282 tokenize=False,
283 )
284
285 expected_prompt = (
286 "<|endoftext|><image><|md_reserved_0|>query<|md_reserved_1|>"
287 "What is in this image?<|md_reserved_2|>"
288 )
289 assert prompt == expected_prompt
290
291 processed_inputs = processor(
292 prompt,
293 mm_items=processor.info.parse_mm_data({"image": [image_assets[0].pil_image]}),
294 hf_processor_mm_kwargs={},
295 )
296 image_placeholders = processed_inputs["mm_placeholders"]["image"]
297 assert len(image_placeholders) == 1
298 assert image_placeholders[0].length == EXPECTED_IMAGE_TOKENS
299
300
301 @pytest.mark.parametrize("model_id", [MOONDREAM3_MODEL_ID])
302 def test_bos_token_always_first(
303 image_assets: ImageTestAssets,
304 model_id: str,
305 ):
306 """Test that BOS token (ID 0) is always at position 0."""
307 ctx = build_model_context(
308 model_id,
309 limit_mm_per_prompt={"image": 1},
310 )
311 processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config)
312
313 # Start with BOS token explicitly
314 prompt = "<|endoftext|><image><|md_reserved_0|>query<|md_reserved_1|>Describe this image.<|md_reserved_2|>" # noqa: E501
315 mm_data = {"image": [image_assets[0].pil_image]}
316
317 processed_inputs = processor(
318 prompt,
319 mm_items=processor.info.parse_mm_data(mm_data),
320 hf_processor_mm_kwargs={},
321 )
322 token_ids = processed_inputs["prompt_token_ids"]
323
324 # Token ID 0 (<|endoftext|>) should be the first token
325 assert token_ids[0] == 0, (
326 f"Expected BOS token (0) at position 0, got {token_ids[0]}"
327 )
328
329
330 @pytest.mark.parametrize("model_id", [MOONDREAM3_MODEL_ID])
331 def test_processor_with_small_image(
332 model_id: str,
333 ):
334 """Test processor with image smaller than crop size."""
335 from PIL import Image
336
337 from vllm.transformers_utils.processors.moondream3 import Moondream3Processor
338
339 processor = Moondream3Processor.from_pretrained(model_id, trust_remote_code=True)
340
341 # Small image (smaller than crop size)
342 small_image = Image.new("RGB", (100, 100), color="yellow")
343 pixel_values, tiling = processor.preprocess_image(small_image)
344
345 # Small images should use 1x1 tiling
346 assert tiling == (1, 1)
347 # Should have 2 crops (global + 1 local)
348 assert pixel_values.shape[0] == 2
349
350
351 @pytest.mark.parametrize(
352 "image_kind",
353 [
354 pytest.param("numpy_hwc", id="numpy-hwc"),
355 pytest.param("numpy_chw", id="numpy-chw"),
356 pytest.param("torch_chw", id="torch-chw"),
357 ],
358 )
359 @pytest.mark.parametrize("model_id", [MOONDREAM3_MODEL_ID])
360 def test_preprocess_image_accepts_non_pil_inputs(
361 image_assets: ImageTestAssets,
362 image_kind: str,
363 model_id: str,
364 ):
365 from vllm.transformers_utils.processors.moondream3 import Moondream3Processor
366
367 processor = Moondream3Processor.from_pretrained(model_id, trust_remote_code=True)
368 pil_image = image_assets[0].pil_image.convert("RGB")
369 hwc_array = np.asarray(pil_image)
370 expected_pixel_values, expected_tiling = processor.preprocess_image(pil_image)
371
372 if image_kind == "numpy_hwc":
373 image = hwc_array
374 elif image_kind == "numpy_chw":
375 image = np.transpose(hwc_array, (2, 0, 1))
376 else:
377 image = torch.from_numpy(np.transpose(hwc_array, (2, 0, 1)).copy())
378
379 pixel_values, tiling = processor.preprocess_image(image)
380
381 assert tiling == expected_tiling
382 assert pixel_values.shape == expected_pixel_values.shape
383 assert pixel_values.dtype == torch.bfloat16
384 assert torch.equal(pixel_values, expected_pixel_values)
385
386
387 @pytest.mark.parametrize("image_kind", ["numpy_chw", "torch_chw"])
388 @pytest.mark.parametrize("model_id", [MOONDREAM3_MODEL_ID])
389 def test_processor_apply_accepts_non_pil_image_inputs(
390 image_assets: ImageTestAssets,
391 image_kind: str,
392 model_id: str,
393 ):
394 ctx = build_model_context(
395 model_id,
396 limit_mm_per_prompt={"image": 1},
397 )
398 processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config)
399
400 prompt = "<|endoftext|><image><|md_reserved_0|>query<|md_reserved_1|>What is this?<|md_reserved_2|>" # noqa: E501
401 hwc_array = np.asarray(image_assets[0].pil_image.convert("RGB"))
402 chw_array = np.transpose(hwc_array, (2, 0, 1)).copy()
403 image = chw_array if image_kind == "numpy_chw" else torch.from_numpy(chw_array)
404
405 processed_inputs = processor(
406 prompt,
407 mm_items=processor.info.parse_mm_data({"image": [image]}),
408 hf_processor_mm_kwargs={},
409 )
410
411 image_placeholders = processed_inputs["mm_placeholders"]["image"]
412 assert len(image_placeholders) == 1
413 assert image_placeholders[0].length == EXPECTED_IMAGE_TOKENS
414
415 mm_kwargs = processed_inputs["mm_kwargs"].get_data()
416 assert mm_kwargs["pixel_values"].shape[2:] == (3, 378, 378)
417
418
419 class TestMoondream3TilingLogic:
420 """CPU-based tests for Moondream3 tiling selection logic.
421
422 These tests validate the select_tiling() function which determines
423 how images are divided into crops for the vision encoder.
424 """
425
426 def test_small_image_no_tiling(self):
427 """Small images should use 1x1 tiling."""
428 from vllm.transformers_utils.processors.moondream3 import select_tiling
429
430 tiling = select_tiling(
431 height=300, width=300, crop_size=CROP_SIZE, max_crops=MAX_CROPS
432 )
433 assert tiling == (1, 1)
434
435 def test_exact_crop_size(self):
436 """Image exactly at crop size should use 1x1."""
437 from vllm.transformers_utils.processors.moondream3 import select_tiling
438
439 tiling = select_tiling(
440 height=CROP_SIZE, width=CROP_SIZE, crop_size=CROP_SIZE, max_crops=MAX_CROPS
441 )
442 assert tiling == (1, 1)
443
444 def test_large_square_image(self):
445 """Large square image should use multiple tiles."""
446 from vllm.transformers_utils.processors.moondream3 import select_tiling
447
448 tiling = select_tiling(
449 height=800, width=800, crop_size=CROP_SIZE, max_crops=MAX_CROPS
450 )
451 h_tiles, w_tiles = tiling
452 assert h_tiles >= 2
453 assert w_tiles >= 2
454 assert h_tiles * w_tiles <= MAX_CROPS
455
456 def test_wide_image(self):
457 """Wide image should have more width tiles."""
458 from vllm.transformers_utils.processors.moondream3 import select_tiling
459
460 tiling = select_tiling(
461 height=400, width=1200, crop_size=CROP_SIZE, max_crops=MAX_CROPS
462 )
463 h_tiles, w_tiles = tiling
464 assert w_tiles >= h_tiles
465
466 def test_tall_image(self):
467 """Tall image should have more height tiles."""
468 from vllm.transformers_utils.processors.moondream3 import select_tiling
469
470 tiling = select_tiling(
471 height=1200, width=400, crop_size=CROP_SIZE, max_crops=MAX_CROPS
472 )
473 h_tiles, w_tiles = tiling
474 assert h_tiles >= w_tiles
475
476 def test_respects_max_crops(self):
477 """Tiling should not exceed max_crops."""
478 from vllm.transformers_utils.processors.moondream3 import select_tiling
479
480 tiling = select_tiling(
481 height=2000, width=2000, crop_size=CROP_SIZE, max_crops=4
482 )
483 h_tiles, w_tiles = tiling
484 assert h_tiles * w_tiles <= 4
485
486
487 class TestMoondream3VisionShapes:
488 """CPU-based tests for vision encoder expected shapes.
489
490 These tests verify the mathematical relationships between
491 crop size, patch size, and token counts.
492 """
493
494 def test_expected_patch_count(self):
495 """Test 378/14 = 27 patches per side, 729 total."""
496 patches_per_side = CROP_SIZE // PATCH_SIZE
497 total_patches = patches_per_side**2
498
499 assert patches_per_side == 27
500 assert total_patches == EXPECTED_IMAGE_TOKENS - 1
501
502 def test_patch_embedding_input_dim(self):
503 """Test patch embedding input dimension."""
504 channels = 3
505 input_dim = PATCH_SIZE * PATCH_SIZE * channels
506
507 assert input_dim == 14 * 14 * 3
508 assert input_dim == 588
509
510
511 class TestMoondream3TauAttention:
512 """CPU-based tests for tau attention scaling components.
513
514 These tests validate the tau attention formula used in Moondream3:
515 - Token-based: tok_q = tanh(gelu(qkv) @ tau_wq.T)
516 - Position-based: tau_pos = 1 + (sigmoid(alpha * log(pos+1)) - 0.5)
517 """
518
519 def test_tau_position_range(self):
520 """Test tau position scaling produces values in valid range."""
521 num_heads = 32
522 seq_len = 100
523
524 tau_alpha = torch.randn(num_heads)
525 positions = torch.arange(seq_len)
526
527 pos_float = (positions.float() + 1.0).clamp(min=1e-6)
528 pos_log = pos_float.log()
529 tau_pos = 1.0 + (torch.sigmoid(tau_alpha[:, None] * pos_log[None, :]) - 0.5)
530
531 assert tau_pos.shape == (num_heads, seq_len)
532 # tau_pos should be between 0.5 and 1.5
533 assert tau_pos.min() >= 0.5
534 assert tau_pos.max() <= 1.5
535
536 def test_tau_token_output_range(self):
537 """Test tau token scaling output is bounded by tanh."""
538 import torch.nn.functional as F
539
540 seq_len = 100
541 qkv_dim = 6144 # 2048 * 3
542 num_heads = 32
543
544 qkv = torch.randn(seq_len, qkv_dim)
545 tau_wq = torch.randn(num_heads, qkv_dim)
546
547 tok_feat = F.gelu(qkv)
548 tok_q = torch.tanh(tok_feat @ tau_wq.t())
549
550 assert tok_q.shape == (seq_len, num_heads)
551 # tanh output is bounded by [-1, 1]
552 assert tok_q.min() >= -1.0
553 assert tok_q.max() <= 1.0