Compare commits

...
Author SHA1 Message Date
Lucas WilkinsonandOpenAI Codex 6099f540ae Handle older FlashInfer FP8 MoE signatures
Co-authored-by: OpenAI Codex <codex@openai.com>

Signed-off-by: Lucas Wilkinson <lwilkins@redhat.com>
2026-07-05 02:00:30 +00:00
34b560b725 [Bugfix][Gemma4] Fix FA4 mm_prefix mask: add sliding window and absolute q_idx (#47332)
Signed-off-by: Luciano Martins <lucianommartins@users.noreply.github.com>
Co-authored-by: Luciano Martins <lucianommartins@users.noreply.github.com>
2026-07-04 17:46:40 -07:00
Ting SUNandGitHub 91b5647300 [Bugfix][Model] Allow Run:ai memory_limit sentinel values (#47337)
Signed-off-by: Ting Sun <suntcrick@gmail.com>
2026-07-05 00:08:34 +00:00
Carl PerssonandGitHub 4a6bf3c77f [ROCm][CI] Fix Kernels and Kernels attention test failures (#47519)
Signed-off-by: Carl Persson <carl.persson@amd.com>
2026-07-04 15:59:51 -05:00
Ting SUNandGitHub d2afe39647 [Bugfix][Frontend] Preserve default sampling params in batch chat (#47597)
Signed-off-by: Ting Sun <suntcrick@gmail.com>
2026-07-04 19:06:39 +00:00
Wentao YeandGitHub 2a9113f998 [Perf] Remove redundant op for GLM 5.2 (#47198)
Signed-off-by: yewentao256 <zhyanwentao@126.com>
2026-07-04 13:25:02 -04:00
yzong-rhandGitHub 0cd6f767e3 [Bugfix][Frontend][gpt-oss] Recover raw tail when Harmony parser ends non-terminal (#47379) 2026-07-04 10:46:24 -04:00
Harry MellorandGitHub f1445f6dbd [CI] Bump huggingface-hub from v1.10.2 to v1.22.0 (#47551)
Signed-off-by: Harry Mellor <19981378+hmellor@users.noreply.github.com>
2026-07-04 07:45:45 -07:00
1d354c694e [Misc] Validate Pooling cache_salt Values (#46966)
Signed-off-by: Taneem Ibrahim <taneem.ibrahim@gmail.com>
Co-authored-by: Wentao Ye <44945378+yewentao256@users.noreply.github.com>
2026-07-04 10:19:28 -04:00
Taneem IbrahimandGitHub 2f21224527 [Misc] Update request-extras parity for batch chat completion (#47333)
Signed-off-by: Taneem Ibrahim <taneem.ibrahim@gmail.com>
2026-07-04 10:19:04 -04:00
Taneem IbrahimGitHubmergify[bot] <37929162+mergify[bot]@users.noreply.github.com>Wentao Ye
fa1fa968c4 [Misc] Forward request-level prompt extras for cross-encoder scoring (#46939)
Signed-off-by: Taneem Ibrahim <taneem.ibrahim@gmail.com>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
Co-authored-by: Wentao Ye <44945378+yewentao256@users.noreply.github.com>
2026-07-04 10:18:36 -04:00
Taneem IbrahimGitHubmergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
6eac8e0070 [Misc] Preserve cross-encoder pooling extra kwargs (#47082)
Signed-off-by: Taneem Ibrahim <taneem.ibrahim@gmail.com>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
2026-07-04 08:14:13 -04:00
1a308c449c [XPU] Add W8A8 FP8 linear kernel with multi-granularity quant support (#43645)
Signed-off-by: Chaojun,Zhang <chaojun.zhang@intel.com>
Signed-off-by: Chaojun Zhang <chaojun.zhang@intel.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
2026-07-04 18:10:01 +08:00
e7c9df9449 [Bugfix][Structured Output][Spec Decode] Constrain bitmask and trim grammar advance at the reasoning boundary (#44297)
Signed-off-by: Allen.Yu <yuyue0225sc@163.com>
Signed-off-by: yue.yu <yuyue0225sc@163.com>
Co-authored-by: Benjamin Chislett <chislett.ben@gmail.com>
2026-07-04 09:08:45 +00:00
gausah01andGitHub 26eb87204d [Bugfix] Fix CPU split-KV scratchpad sizing (#45844)
Signed-off-by: Gauri Sahnan <gauri.sahnan@arm.com>
2026-07-04 06:47:23 +00:00
4c3c17d43b [ROCm] Disable persistent sparse-MLA kernel for chunked-prefill continuations (#47567)
Signed-off-by: Rohan Potdar <rohan.potdar@amd.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-04 01:21:43 -05:00
f329ce405b [ROCm][CI][Bugfix] Use VllmRunner for voxtral_realtime tests to avoid OOM on AMD GPU (#47536)
Signed-off-by: Shanshan Shen <87969357+shen-shanshan@users.noreply.github.com>
Signed-off-by: Andreas Karatzas <akaratza@amd.com>
Co-authored-by: Andreas Karatzas <akaratza@amd.com>
2026-07-04 12:26:10 +08:00
07516fda67 [MRV2][SD] Make Dynamic SD comatible with Full Cuda Graphs (#45953)
Signed-off-by: Ekagra Ranjan <3116519+ekagra-ranjan@users.noreply.github.com>
Co-authored-by: Benjamin Chislett <chislett.ben@gmail.com>
Co-authored-by: Nick Hill <nickhill123@gmail.com>
2026-07-03 23:58:27 -04:00
sychen52GitHubmergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
67ff0ae30f Support nvfp4 kv with kv-cache-dtype-skip-layers sliding_window (#42890)
Signed-off-by: Shiyang Chen <shiychen@nvidia.com>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
2026-07-04 02:29:13 +00:00
Bugen ZhaoandGitHub ab3b6d97aa [Frontend] Limit SO_REUSEPORT to multi-worker serving (#47529)
Signed-off-by: Bugen Zhao <i@bugenzhao.com>
2026-07-04 01:26:24 +00:00
Ben BrowningandGitHub fb5291b35b [Frontend] [Parser] Port DeepSeek V4 to streaming parser engine framework (#45877)
Signed-off-by: Ben Browning <bbrownin@redhat.com>
2026-07-03 20:55:23 -04:00
labAxiaomingandGitHub d6d39c111e [GLM4V] Avoid GLM4V processor init during startup metadata reads (#47155)
Signed-off-by: xiaoming <1259730330@qq.com>
2026-07-03 15:03:16 -07:00
379950191f [Bugfix][Multimodal] Normalize direct PIL image inputs (#47566)
Signed-off-by: Ting Sun <suntcrick@gmail.com>
Co-authored-by: Cyrus Leung <tlleungac@connect.ust.hk>
2026-07-03 14:27:14 -07:00
576bf75d0e [AMD][EPLB] Enable EPLB for Quark OCP MXFP4 MoE (#47220)
Signed-off-by: okorzh-amd <okorzh-amd@users.noreply.github.com>
Co-authored-by: okorzh-amd <okorzh-amd@users.noreply.github.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Andreas Karatzas <akaratza@amd.com>
2026-07-03 14:41:52 -05:00
TresandGitHub f006e5a24c [CI][AMD] Allow git operations on previously created work trees (#47554)
Signed-off-by: Tres Popp <tres.popp@amd.com>
2026-07-03 14:41:01 -05:00
f63dca6838 [ROCm] Fix encoder-decoder cross-attention KV layout aliasing (#47035)
Signed-off-by: Djordje Ramic <djoramic@amd.com>
Co-authored-by: Andreas Karatzas <akaratza@amd.com>
2026-07-03 13:53:29 -05:00
Bugen ZhaoandGitHub 8651f043b8 [Rust Frontend] Speed up chat roundtrip tests (#47523)
Signed-off-by: Bugen Zhao <i@bugenzhao.com>
2026-07-03 19:25:06 +01:00
Andreas KaratzasandGitHub 3775d5fcab [ROCm][CI] Adding test groups for parity with upstream (#47479)
Signed-off-by: Andreas Karatzas <akaratza@amd.com>
2026-07-03 19:15:01 +04:00
Roberto L. CastroGitHubmergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
d7192cfccf [CI Bugfix] Lazily import Qwen warmup dependencies (#47539)
Signed-off-by: LopezCastroRoberto <rocastro@redhat.com>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
2026-07-03 23:10:49 +08:00
AgenticSparkandGitHub 978de83353 [Bugfix][CPU] Ship examples/ in the CPU release image (#47447)
Signed-off-by: liejiang <jianglie2023@gmail.com>
2026-07-03 11:46:24 +00:00
105 changed files with 4235 additions and 1140 deletions
+48 -1
View File
@@ -49,6 +49,30 @@ steps:
VLLM_XPU_FUSED_MOE_USE_REF=1 python3 examples/basic/offline_inference/generate.py --model Qwen/Qwen3-30B-A3B-Instruct-2507-FP8 --enforce-eager -tp 2 --max-model-len 8192 &&
python3 examples/basic/offline_inference/generate.py --model INCModel/Qwen3-30B-A3B-Instruct-2507-MXFP4-LLMC --enforce-eager -tp 2 --max-model-len 8192
'
- label: "XPU W8A8 FP8 Linear Examples"
depends_on:
- image-build-xpu
timeout_in_minutes: 60
device: intel_gpu
agent_tags:
label: production
gpu: 1+
mem: 24+
no_plugin: true
env:
REGISTRY: "public.ecr.aws/q9t5s3a7"
REPO: "vllm-ci-test-repo"
VLLM_TEST_DEVICE: "xpu"
source_file_dependencies:
- vllm/
- .buildkite/intel_jobs/test-intel.yaml
commands:
- >-
bash .buildkite/scripts/hardware_ci/run-intel-test.sh
'python3 examples/basic/offline_inference/generate.py --linear-backend xpu --model RedHatAI/Meta-Llama-3.1-8B-Instruct-FP8 --enforce-eager --max-model-len 4096 &&
python3 examples/basic/offline_inference/generate.py --linear-backend xpu --model neuralmagic/Llama-3.2-1B-Instruct-FP8-dynamic --enforce-eager --max-model-len 4096 &&
python3 examples/basic/offline_inference/generate.py --linear-backend xpu --model meta-llama/Llama-3.2-1B-Instruct --quantization fp8 --enforce-eager --max-model-len 4096
'
- label: "XPU V1 test"
depends_on:
- image-build-xpu
@@ -120,4 +144,27 @@ steps:
- >-
bash .buildkite/scripts/hardware_ci/run-intel-test.sh
'cd tests &&
pytest -v -s quantization/test_auto_round.py'
pytest -v -s quantization/test_auto_round.py'
- label: "XPU compressed tensors FP8 test"
depends_on:
- image-build-xpu
timeout_in_minutes: 60
device: intel_gpu
agent_tags:
label: production
gpu: 1+
mem: 16+
no_plugin: true
env:
REGISTRY: "public.ecr.aws/q9t5s3a7"
REPO: "vllm-ci-test-repo"
VLLM_TEST_DEVICE: "xpu"
source_file_dependencies:
- vllm/
- tests/quantization/test_compressed_tensors.py
- .buildkite/intel_jobs/test-intel.yaml
commands:
- >-
bash .buildkite/scripts/hardware_ci/run-intel-test.sh
'cd tests &&
pytest -v -s quantization/test_compressed_tensors.py::test_compressed_tensors_fp8'
@@ -448,9 +448,11 @@ checkout="${BUILDKITE_BUILD_CHECKOUT_PATH:-}"
if [[ -z "${checkout}" || ! -d "${checkout}" ]]; then
checkout="."
fi
if git -C "${checkout}" rev-parse --is-inside-work-tree >/dev/null 2>&1; then
# Pass safe.directory per-command (-c) because buildkite runs will always fail
# the next check on git 2.35.2+ due to mixed uses of root and buildkite-agent/uids.
if git -c "safe.directory=${checkout}" -C "${checkout}" rev-parse --is-inside-work-tree >/dev/null 2>&1; then
vllm_standalone_merge_base="$(
git -C "${checkout}" merge-base HEAD origin/main 2>/dev/null || true
git -c "safe.directory=${checkout}" -C "${checkout}" merge-base HEAD origin/main 2>/dev/null || true
)"
fi
if [[ -z "${vllm_standalone_merge_base}" ]]; then
+188 -3
View File
@@ -1025,6 +1025,23 @@ steps:
commands:
- pytest -s -v evals/gsm8k/test_gsm8k_correctness.py --config-list-file=configs/models-small.txt
- label: MRCR Eval Small Models # TBD
timeout_in_minutes: 180
mirror_hardwares: [amdexperimental, amdproduction, amdgfx942nightly, amdmi300]
agent_pool: mi300_1
optional: true
working_dir: "/vllm-workspace/tests"
source_file_dependencies:
- vllm/model_executor/models/
- vllm/model_executor/model_loader/
- vllm/v1/attention/backends/
- vllm/v1/attention/selector.py
- vllm/_aiter_ops.py
- vllm/platforms/rocm.py
- tests/evals/mrcr/
commands:
- pytest -s -v evals/mrcr/test_mrcr_correctness.py --config-list-file=evals/mrcr/configs/models-small.txt
- label: LM Eval Small Models (MI300) # TBD
timeout_in_minutes: 180
mirror_hardwares: [amdexperimental, amdproduction, amdgfx942nightly, amdmi300]
@@ -1283,6 +1300,21 @@ steps:
#---------------------------------------------------------- mi300 · kernels ----------------------------------------------------------#
- label: vLLM IR Tests # TBD
timeout_in_minutes: 30
mirror_hardwares: [amdexperimental, amdproduction, amdgfx942nightly, amdmi300]
agent_pool: mi300_1
optional: true
working_dir: "/vllm-workspace/"
source_file_dependencies:
- vllm/ir
- vllm/kernels
- vllm/_aiter_ops.py
- vllm/platforms/rocm.py
commands:
- pytest -v -s tests/ir
- pytest -v -s tests/kernels/ir
- label: Kernels Attention Test %N # TBD
timeout_in_minutes: 100
mirror_hardwares: [amdexperimental, amdproduction, amdgfx942nightly, amdmi300]
@@ -1317,6 +1349,21 @@ steps:
commands:
- pytest -v -s kernels/core --ignore=kernels/core/test_minimax_reduce_rms.py kernels/test_concat_mla_q.py kernels/test_top_k_per_row.py
- label: Kernels KDA Test # TBD
timeout_in_minutes: 30
mirror_hardwares: [amdexperimental, amdproduction, amdgfx942nightly, amdmi300]
agent_pool: mi300_1
optional: true
working_dir: "/vllm-workspace/tests"
source_file_dependencies:
- vllm/model_executor/layers/fla/ops/kda.py
- vllm/model_executor/layers/fla/ops/chunk_delta_h.py
- vllm/model_executor/layers/fla/ops/l2norm.py
- tests/kernels/test_kda.py
- vllm/platforms/rocm.py
commands:
- pytest -v -s kernels/test_kda.py
- label: Kernels MoE Test %N # TBD
timeout_in_minutes: 95
mirror_hardwares: [amdexperimental, amdproduction, amdgfx942nightly, amdmi300]
@@ -1429,6 +1476,128 @@ steps:
- pytest -v -s model_executor -m '(not slow_test)'
- pytest -v -s entrypoints/openai/completion/test_tensorizer_entrypoint.py
#---------------------------------------------------- mi300 · model_runner_v2 -------------------------------------------------------#
- label: Model Runner V2 Core Tests # TBD
timeout_in_minutes: 60
mirror_hardwares: [amdexperimental, amdproduction, amdgfx942nightly, amdmi300]
agent_pool: mi300_1
optional: true
working_dir: "/vllm-workspace/tests"
source_file_dependencies:
- vllm/v1/worker/gpu/
- vllm/v1/worker/gpu_worker.py
- vllm/v1/core/sched/
- vllm/v1/attention/
- tests/v1/engine/test_llm_engine.py
- tests/v1/e2e/
- tests/entrypoints/llm/test_struct_output_generate.py
- vllm/platforms/rocm.py
commands:
- set -x
- export VLLM_USE_V2_MODEL_RUNNER=1
- pytest -v -s v1/engine/test_llm_engine.py -k "not test_engine_metrics"
- ENFORCE_EAGER=1 pytest -v -s v1/e2e/general/test_async_scheduling.py -k "not ngram"
- pytest -v -s v1/e2e/general/test_context_length.py
- pytest -v -s v1/e2e/general/test_min_tokens.py
- pytest -v -s entrypoints/llm/test_struct_output_generate.py -k "xgrammar and not speculative_config6 and not speculative_config7 and not speculative_config8 and not speculative_config0"
- label: Model Runner V2 Examples # TBD
timeout_in_minutes: 60
mirror_hardwares: [amdexperimental, amdproduction, amdgfx942nightly, amdmi300]
agent_pool: mi300_1
optional: true
working_dir: "/vllm-workspace/examples"
source_file_dependencies:
- vllm/v1/worker/gpu/
- vllm/v1/core/sched/
- vllm/v1/worker/gpu_worker.py
- examples/basic/offline_inference/
- examples/generate/multimodal/
- examples/features/
- examples/pooling/embed/vision_embedding_offline.py
- examples/features/tensorize_vllm_model.py
- examples/deployment/llm_engine_example.py
- vllm/platforms/rocm.py
commands:
- set -x
- export VLLM_USE_V2_MODEL_RUNNER=1
- pip install tensorizer
- python3 basic/offline_inference/chat.py
- python3 basic/offline_inference/generate.py --model facebook/opt-125m
- python3 generate/multimodal/audio_language_offline.py --seed 0
- python3 generate/multimodal/vision_language_offline.py --seed 0
- python3 generate/multimodal/vision_language_multi_image_offline.py --seed 0
- python3 generate/multimodal/encoder_decoder_multimodal_offline.py --model-type whisper --seed 0
- python3 pooling/embed/vision_embedding_offline.py --seed 0
- python3 features/automatic_prefix_caching/prefix_caching_offline.py
- python3 deployment/llm_engine_example.py
- python3 features/tensorize_vllm_model.py --model facebook/opt-125m serialize --serialized-directory /tmp/ --suffix v1 && python3 features/tensorize_vllm_model.py --model facebook/opt-125m deserialize --path-to-tensors /tmp/vllm/facebook/opt-125m/v1/model.tensors
- python3 features/speculative_decoding/spec_decode_offline.py --test --method eagle --num_spec_tokens 3 --dataset-name hf --dataset-path philschmid/mt-bench --num-prompts 80 --temp 0 --top-p 1.0 --top-k -1 --tp 1 --enable-chunked-prefill --max-model-len 2048
- python3 features/speculative_decoding/spec_decode_offline.py --test --method eagle3 --num_spec_tokens 3 --dataset-name hf --dataset-path philschmid/mt-bench --num-prompts 80 --temp 0 --top-p 1.0 --top-k -1 --tp 1 --enable-chunked-prefill --max-model-len 1536
- label: Model Runner V2 Distributed (2 GPUs) # TBD
timeout_in_minutes: 60
mirror_hardwares: [amdexperimental, amdproduction, amdgfx942nightly, amdmi300]
agent_pool: mi300_2
num_gpus: 2
optional: true
working_dir: "/vllm-workspace/tests"
source_file_dependencies:
- vllm/v1/worker/gpu/
- vllm/v1/worker/gpu_worker.py
- tests/basic_correctness/test_basic_correctness.py
- tests/v1/distributed/test_async_llm_dp.py
- tests/v1/distributed/test_eagle_dp.py
- vllm/platforms/rocm.py
commands:
- set -x
- export VLLM_USE_V2_MODEL_RUNNER=1
- TARGET_TEST_SUITE=MI300 pytest -v -s basic_correctness/test_basic_correctness.py -m 'distributed(num_gpus=2)' -k "not ray and not True"
- TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_async_llm_dp.py -k "not ray"
- TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_eagle_dp.py
- label: Model Runner V2 Pipeline Parallelism (4 GPUs) # TBD
timeout_in_minutes: 60
mirror_hardwares: [amdexperimental, amdproduction, amdgfx942nightly, amdmi300]
agent_pool: mi300_4
num_gpus: 4
optional: true
working_dir: "/vllm-workspace/tests"
source_file_dependencies:
- vllm/v1/worker/gpu/
- vllm/v1/worker/gpu_worker.py
- tests/distributed/test_pipeline_parallel.py
- tests/distributed/test_pp_cudagraph.py
- vllm/platforms/rocm.py
commands:
- set -x
- export VLLM_USE_V2_MODEL_RUNNER=1
- pytest -v -s distributed/test_pipeline_parallel.py -k "not ray and not Jamba"
- pytest -v -s distributed/test_pp_cudagraph.py -k "not ray"
- label: Model Runner V2 Spec Decode # TBD
timeout_in_minutes: 45
mirror_hardwares: [amdexperimental, amdproduction, amdgfx942nightly, amdmi300]
agent_pool: mi300_1
optional: true
working_dir: "/vllm-workspace/tests"
source_file_dependencies:
- vllm/v1/worker/gpu/
- vllm/v1/worker/gpu_worker.py
- tests/v1/spec_decode/test_max_len.py
- tests/v1/spec_decode/test_rejection_sampler_utils.py
- tests/v1/spec_decode/test_synthetic_rejection_sampler_utils.py
- tests/v1/e2e/spec_decode/test_spec_decode.py
- vllm/platforms/rocm.py
commands:
- set -x
- export VLLM_USE_V2_MODEL_RUNNER=1
- pytest -v -s v1/spec_decode/test_max_len.py -k "eagle or mtp"
- pytest -v -s v1/spec_decode/test_rejection_sampler_utils.py
- pytest -v -s v1/spec_decode/test_synthetic_rejection_sampler_utils.py
- pytest -v -s v1/e2e/spec_decode/test_spec_decode.py -k "eagle or mtp"
#------------------------------------------------------ mi300 · models / basic -------------------------------------------------------#
- label: Basic Models Tests (Extra Initialization) %N # TBD
@@ -1745,6 +1914,22 @@ steps:
- pytest -v -s plugins_tests/test_oot_registration_offline.py # it needs a clean process
- pytest -v -s plugins_tests/lora_resolvers # unit tests for in-tree lora resolver plugins
- label: GGUF Plugin # TBD
timeout_in_minutes: 30
mirror_hardwares: [amdexperimental, amdproduction, amdgfx942nightly, amdmi300]
agent_pool: mi300_1
optional: true
soft_fail: true
working_dir: "/vllm-workspace/tests"
source_file_dependencies:
- vllm/model_executor/layers/quantization
- tests/plugins_tests/test_gguf_plugin.py
- tests/plugins_tests/gguf
- vllm/platforms/rocm.py
commands:
- pip install "vllm-gguf-plugin >= 0.0.2"
- pytest -v -s plugins_tests/gguf
#------------------------------------------------------- mi300 · quantization --------------------------------------------------------#
- label: Quantization # TBD
@@ -2063,7 +2248,7 @@ steps:
- pytest -v -s v1/worker
- pytest -v -s -m 'not cpu_test' v1/kv_connector/unit
- pytest -v -s -m 'not cpu_test' v1/metrics
- pip install -U git+https://github.com/robertgshaw2-redhat/lm-evaluation-harness.git@streaming-api
- pip install -U git+https://github.com/vllm-project/lm-evaluation-harness.git@streaming-api
# - export HSA_NO_SCRATCH_RECLAIM=1
- pytest -v -s entrypoints/openai/correctness/test_lmeval.py::test_lm_eval_accuracy_v1_engine
@@ -2770,7 +2955,7 @@ steps:
#--------------------------------------------------------- mi355 · examples ----------------------------------------------------------#
- label: Examples # TBD
timeout_in_minutes: 45
timeout_in_minutes: 100
mirror_hardwares: [amdexperimental, amdproduction, amdgfx950nightly, amdmi355]
agent_pool: mi355_1
working_dir: "/vllm-workspace/examples"
@@ -3133,7 +3318,7 @@ steps:
- pytest -v -s v1/worker
- pytest -v -s -m 'not cpu_test' v1/kv_connector/unit
- pytest -v -s -m 'not cpu_test' v1/metrics
- pip install -U git+https://github.com/robertgshaw2-redhat/lm-evaluation-harness.git@streaming-api
- pip install -U git+https://github.com/vllm-project/lm-evaluation-harness.git@streaming-api
- pytest -v -s entrypoints/openai/correctness/test_lmeval.py::test_lm_eval_accuracy_v1_engine
- label: V1 Sample + Logits # TBD
+1 -1
View File
@@ -103,7 +103,7 @@ steps:
- pytest -v -s -m 'not cpu_test' v1/kv_connector/unit
- pytest -v -s -m 'not cpu_test' v1/metrics
# Integration test for streaming correctness (requires special branch).
- pip install -U git+https://github.com/robertgshaw2-redhat/lm-evaluation-harness.git@streaming-api
- pip install -U git+https://github.com/vllm-project/lm-evaluation-harness.git@streaming-api
- pytest -v -s entrypoints/openai/correctness/test_lmeval.py::test_lm_eval_accuracy_v1_engine
mirror:
amd:
+56 -14
View File
@@ -417,8 +417,10 @@ class AttentionScheduler {
has_decode_request = has_decode_request || (q_token_num == 1);
decode_only_batch = decode_only_batch && (q_token_num == 1);
}
int32_t q_head_per_kv = input.num_heads_q / input.num_heads_kv;
const bool supports_gqa = q_head_per_kv <= max_num_q_per_iter;
const int32_t original_q_head_per_kv =
input.num_heads_q / input.num_heads_kv;
int32_t q_head_per_kv = original_q_head_per_kv;
const bool supports_gqa = original_q_head_per_kv <= max_num_q_per_iter;
const bool use_gqa_fast_path = supports_gqa && decode_only_batch;
const bool use_gqa_scratchpad = supports_gqa && has_decode_request;
if (!use_gqa_scratchpad) {
@@ -671,22 +673,62 @@ class AttentionScheduler {
metadata_ptr->effective_thread_num = effective_thread_num;
{
// when q_tile_size = max_num_q_per_iter, requires max
// attention_scratchpad_size
AttentionScratchPad sc(0, *metadata_ptr, 0x0);
int64_t n = AttentionScheduler::calcu_tile_size_with_constant_q(
cache_size, input.head_dim, input.elem_size, input.q_buffer_elem_size,
input.logits_buffer_elem_size, input.output_buffer_elem_size,
max_num_q_per_iter, kv_len_alignment, max_num_q_per_iter, true);
sc.update(input.head_dim, input.q_buffer_elem_size,
input.logits_buffer_elem_size, input.output_buffer_elem_size,
max_num_q_per_iter, max_num_q_per_iter, n);
int64_t max_attention_scratchpad_size = 0;
for (const AttentionWorkItemGroup& item : workitems) {
const bool curr_use_gqa =
use_gqa_fast_path || (supports_gqa && item.q_token_num == 1);
const int32_t curr_q_heads_per_kv =
curr_use_gqa ? original_q_head_per_kv : 1;
const int32_t curr_default_q_tile_token_num =
default_tile_size / curr_q_heads_per_kv;
for (int32_t q_token_offset = 0; q_token_offset < item.q_token_num;
q_token_offset += curr_default_q_tile_token_num) {
const int32_t actual_q_token_num = std::min(
curr_default_q_tile_token_num, item.q_token_num - q_token_offset);
const int32_t q_head_tile_size =
actual_q_token_num * curr_q_heads_per_kv;
const int32_t rounded_q_head_tile_size =
((q_head_tile_size + max_num_q_per_iter - 1) /
max_num_q_per_iter) *
max_num_q_per_iter;
const int64_t n = AttentionScheduler::calcu_tile_size_with_constant_q(
cache_size, input.head_dim, input.elem_size,
input.q_buffer_elem_size, input.logits_buffer_elem_size,
input.output_buffer_elem_size, max_num_q_per_iter,
kv_len_alignment, rounded_q_head_tile_size,
rounded_q_head_tile_size <= max_num_q_per_iter);
sc.update(input.head_dim, input.q_buffer_elem_size,
input.logits_buffer_elem_size,
input.output_buffer_elem_size, max_num_q_per_iter,
rounded_q_head_tile_size, n);
max_attention_scratchpad_size = std::max(
max_attention_scratchpad_size, sc.get_thread_scratchpad_size());
}
}
metadata_ptr->attention_scratchpad_size_per_thread =
((sc.get_thread_scratchpad_size() + 63) / 64) * 64;
((max_attention_scratchpad_size + 63) / 64) * 64;
int32_t max_reduction_q_head_tile_size = 0;
for (const ReductionWorkItemGroup& item : reduce_workitems) {
const bool curr_use_gqa =
use_gqa_fast_path || (supports_gqa && item.q_token_id_num == 1);
const int32_t curr_q_heads_per_kv =
curr_use_gqa ? original_q_head_per_kv : 1;
max_reduction_q_head_tile_size =
std::max(max_reduction_q_head_tile_size,
item.q_token_id_num * curr_q_heads_per_kv);
}
sc.update(0, metadata_ptr->reduction_split_num, input.head_dim,
q_head_per_kv * split_kv_q_token_num_threshold,
input.output_buffer_elem_size);
max_reduction_q_head_tile_size, input.output_buffer_elem_size);
metadata_ptr->reduction_scratchpad_size_per_kv_head =
((sc.get_reduction_scratchpad_size() + 63) / 64) * 64;
}
+6
View File
@@ -301,6 +301,12 @@ LABEL ai.vllm.build.cpu-x86="${VLLM_CPU_X86:-false}"
LABEL ai.vllm.build.cpu-arm-bf16="${VLLM_CPU_ARM_BF16:-false}"
LABEL ai.vllm.build.python-version="${PYTHON_VERSION:-3.12}"
# Copy the examples directory (including the chat/tool templates) so it is
# present in the released image, as the CUDA image ships it too. The vllm-test
# stage above adds examples/ for testing only, so without this the published
# vllm-openai-cpu image would not ship examples/*.jinja.
COPY examples examples
ENTRYPOINT ["vllm", "serve"]
+2
View File
@@ -62,6 +62,8 @@ weight name. Unset fields fall back to the `--quantization` shorthand's
defaults, or for already-quantized checkpoints to whatever the checkpoint
declares.
On XPU, non-block FP8 scaled-mm linear layers default to W8A16; setting `--linear-backend xpu` forces W8A8. Use `--linear-backend xpu_woq` to explicitly select weight-only quantization (W8A16).
The CLI accepts the same shape as JSON or as dotted keys:
```bash
@@ -49,6 +49,32 @@ You can configure how the quantization scales are computed in vLLM using three d
- `kv_cache_dtype="fp8_e4m3"`: Supported on CUDA 11.8+ and ROCm (AMD GPUs)
- `kv_cache_dtype="fp8_e5m2"`: Supported on CUDA 11.8+
### Skipping Specific Layers from KV-Cache Quantization
Some attention layer types (e.g. sliding-window) are more sensitive to KV-cache quantization. The `--kv-cache-dtype-skip-layers` flag leaves the specified layers at the model's native dtype while keeping the rest of the layers under the chosen quantized dtype. The flag accepts either layer indices or layer-type names:
```bash
# Skip every sliding-window attention layer.
vllm serve <model> \
--kv-cache-dtype fp8 \
--kv-cache-dtype-skip-layers sliding_window
# Skip specific layer indices.
vllm serve <model> \
--kv-cache-dtype fp8 \
--kv-cache-dtype-skip-layers 0 1 23
```
Programmatic usage:
```python
llm = LLM(
model="meta-llama/Llama-3.1-8B-Instruct",
kv_cache_dtype="fp8",
kv_cache_dtype_skip_layers=["sliding_window"],
)
```
---
## Examples
@@ -71,8 +71,5 @@ VLLM_USE_V2_MODEL_RUNNER=0 vllm serve meta-llama/Llama-3.1-8B-Instruct \
## Limitations
* only tested with Eagle and Eagle-3. Other SD methods may or may not work out of the box
* only usable with Model Runner V1
* not compatible with full cuda graph so we force piece-wise cuda graph with this feature
We are working on enabling it on MRv2 with full cuda graph support.
* Tested with Eagle, Eagle-3, and DFlash. Other SD methods may or may not work out of the box
* Full Cudagraph only works with Model Runner V2. MRv1 only supports piece-wise cuda graph with this feature
+4 -4
View File
@@ -112,9 +112,10 @@ charset-normalizer==3.4.0
# via requests
chz==0.3.0
# via gpt-oss
click==8.1.7
click==8.4.2
# via
# black
# huggingface-hub
# jiwer
# nltk
# ray
@@ -309,7 +310,7 @@ h2==4.3.0
# via httpx
harfile==0.5.0
# via schemathesis
hf-xet==1.4.3
hf-xet==1.5.1
# via huggingface-hub
hiredis==3.0.0
# via tensorizer
@@ -335,7 +336,7 @@ httpx==0.27.2
# schemathesis
httpx-sse==0.4.3
# via mcp
huggingface-hub==1.10.2
huggingface-hub==1.22.0
# via
# accelerate
# datasets
@@ -1182,7 +1183,6 @@ typer==0.26.8
# fastapi-cli
# fastapi-cloud-cli
# fastsafetensors
# huggingface-hub
# perceptron
# transformers
typing-extensions==4.15.0
+4 -4
View File
@@ -117,9 +117,10 @@ charset-normalizer==3.4.0
# via requests
chz==0.3.0
# via gpt-oss
click==8.1.7
click==8.4.2
# via
# black
# huggingface-hub
# jiwer
# nltk
# ray
@@ -330,7 +331,7 @@ h2==4.3.0
# via httpx
harfile==0.5.0
# via schemathesis
hf-xet==1.4.3
hf-xet==1.5.1
# via huggingface-hub
hiredis==3.0.0
# via tensorizer
@@ -356,7 +357,7 @@ httpx==0.27.2
# schemathesis
httpx-sse==0.4.3
# via mcp
huggingface-hub==1.10.2
huggingface-hub==1.22.0
# via
# accelerate
# datasets
@@ -1285,7 +1286,6 @@ typer==0.26.8
# fastapi-cli
# fastapi-cloud-cli
# fastsafetensors
# huggingface-hub
# perceptron
# transformers
typing-extensions==4.15.0
+4 -4
View File
@@ -116,9 +116,10 @@ choreographer==1.2.1
# via kaleido
chz==0.4.0
# via gpt-oss
click==8.3.1
click==8.4.2
# via
# black
# huggingface-hub
# jiwer
# nltk
# ray
@@ -323,7 +324,7 @@ h2==4.3.0
# via httpx
harfile==0.5.0
# via schemathesis
hf-xet==1.4.3
hf-xet==1.5.1
# via huggingface-hub
hiredis==3.3.1
# via tensorizer
@@ -349,7 +350,7 @@ httpx==0.27.2
# schemathesis
httpx-sse==0.4.3
# via mcp
huggingface-hub==1.10.2
huggingface-hub==1.22.0
# via
# accelerate
# datasets
@@ -1244,7 +1245,6 @@ typer==0.24.1
# fastapi-cli
# fastapi-cloud-cli
# fastsafetensors
# huggingface-hub
# perceptron
# transformers
typing-extensions==4.15.0
+4 -4
View File
@@ -83,8 +83,9 @@ charset-normalizer==3.4.6
# via requests
chz==0.4.0
# via gpt-oss
click==8.3.1
click==8.4.2
# via
# huggingface-hub
# jiwer
# nltk
# rich-toolkit
@@ -206,7 +207,7 @@ h11==0.16.0
# uvicorn
harfile==0.4.0
# via schemathesis
hf-xet==1.4.3
hf-xet==1.5.1
# via huggingface-hub
html2text==2025.4.15
# via gpt-oss
@@ -227,7 +228,7 @@ httpx==0.28.1
# schemathesis
httpx-sse==0.4.3
# via mcp
huggingface-hub==1.10.2
huggingface-hub==1.22.0
# via
# accelerate
# datasets
@@ -959,7 +960,6 @@ typer==0.24.1
# via
# fastapi-cli
# fastapi-cloud-cli
# huggingface-hub
# transformers
typing-extensions==4.15.0
# via
+7
View File
@@ -144,6 +144,13 @@ too_many_arguments = "allow"
[profile.dev]
panic = "abort"
# Speed up cold tokenizer construction in tests.
[profile.dev.package]
fastokens = { opt-level = 3 }
regex-automata = { opt-level = 3 }
serde_json = { opt-level = 3 }
tokenizers = { opt-level = 3 }
[profile.release]
lto = "thin"
panic = "abort"
+27 -19
View File
@@ -214,14 +214,17 @@ macro_rules! roundtrip_tests {
($($case:ident => [$($(#[$fixture_attr:meta])* $fixture:ident),* $(,)?]),+ $(,)?) => {
paste::paste! {
$(
$(
#[tokio::test]
$(#[$fixture_attr])*
#[file_serial([<hf_ $case>])]
async fn [<roundtrip_ $case _ $fixture>]() -> Result<()> {
[<run_roundtrip_ $fixture>](RoundtripCase::$case()).await
}
)*
#[tokio::test]
#[file_serial([<hf_ $case>])]
async fn [<roundtrip_ $case>]() -> Result<()> {
let case = RoundtripCase::$case();
let backends = load_roundtrip_backends(&case).await?;
$(
$(#[$fixture_attr])*
[<run_roundtrip_ $fixture>](&case, &backends).await?;
)*
Ok(())
}
)+
}
};
@@ -241,18 +244,21 @@ roundtrip_tests! {
}
/// Run the fixed reasoning+content fixture for one model/parser case.
async fn run_roundtrip_reasoning_and_content(case: RoundtripCase) -> Result<()> {
async fn run_roundtrip_reasoning_and_content(
case: &RoundtripCase,
backends: &vllm_chat::LoadedModelBackends,
) -> Result<()> {
for thinking in case.thinking_behavior.fixtures() {
run_roundtrip_reasoning_and_content_inner(case.clone(), thinking).await?;
run_roundtrip_reasoning_and_content_inner(case, backends, thinking).await?;
}
Ok(())
}
async fn run_roundtrip_reasoning_and_content_inner(
case: RoundtripCase,
case: &RoundtripCase,
backends: &vllm_chat::LoadedModelBackends,
thinking: Option<bool>,
) -> Result<()> {
let backends = load_roundtrip_backends(&case).await?;
let request = roundtrip_request(
"roundtrip-reasoning-content",
vec![ChatMessage::text(ChatRole::User, "What is 2 + 2?")],
@@ -275,7 +281,7 @@ async fn run_roundtrip_reasoning_and_content_inner(
});
AssistantMessage { content }
};
let result = run_roundtrip(&case, &backends, &request, assistant).await?;
let result = run_roundtrip(case, backends, &request, assistant).await?;
assert_eq!(
result.parsed_message.reasoning().as_deref().map(str::trim),
@@ -293,8 +299,10 @@ async fn run_roundtrip_reasoning_and_content_inner(
}
/// Run the fixed reasoning+multiple-tools fixture for one model/parser case.
async fn run_roundtrip_tool_call_mix(case: RoundtripCase) -> Result<()> {
let backends = load_roundtrip_backends(&case).await?;
async fn run_roundtrip_tool_call_mix(
case: &RoundtripCase,
backends: &vllm_chat::LoadedModelBackends,
) -> Result<()> {
let request = roundtrip_request(
"roundtrip-reasoning-tools",
vec![ChatMessage::text(
@@ -308,8 +316,8 @@ async fn run_roundtrip_tool_call_mix(case: RoundtripCase) -> Result<()> {
let expected_text = "I will call the tools.";
let result = run_roundtrip(
&case,
&backends,
case,
backends,
&request,
AssistantMessage {
content: vec![
@@ -353,12 +361,12 @@ async fn run_roundtrip_tool_call_mix(case: RoundtripCase) -> Result<()> {
assert_eq!(tool_calls[0].name, "get_weather");
assert_eq!(
tool_calls[0].arguments,
expected_arguments(&case, r#"{"location": "Shanghai"}"#)?,
expected_arguments(case, r#"{"location": "Shanghai"}"#)?,
);
assert_eq!(tool_calls[1].name, "add");
assert_eq!(
tool_calls[1].arguments,
expected_arguments(&case, r#"{"y": 1.0, "x": 2, "items": ["left", "right"]}"#)?,
expected_arguments(case, r#"{"y": 1.0, "x": 2, "items": ["left", "right"]}"#)?,
);
assert_eq!(
+1 -1
View File
@@ -21,7 +21,7 @@ def _patch_hf_api(side_effect):
@pytest.fixture(scope="session")
def hf_tokenizer() -> PreTrainedTokenizerBase:
return AutoTokenizer.from_pretrained("gpt2")
return AutoTokenizer.from_pretrained("openai-community/gpt2")
_FAKE_ROWS = {
+1 -1
View File
@@ -12,7 +12,7 @@ from vllm.benchmarks.datasets import get_samples
@pytest.fixture(scope="session")
def hf_tokenizer() -> PreTrainedTokenizerBase:
return AutoTokenizer.from_pretrained("gpt2")
return AutoTokenizer.from_pretrained("openai-community/gpt2")
def _write_jsonl(path: Path, n_rows: int) -> None:
+1 -1
View File
@@ -17,7 +17,7 @@ from vllm.benchmarks.datasets import (
@pytest.fixture(scope="session")
def hf_tokenizer() -> PreTrainedTokenizerBase:
# Use a small, commonly available tokenizer
return AutoTokenizer.from_pretrained("gpt2")
return AutoTokenizer.from_pretrained("openai-community/gpt2")
class Params(NamedTuple):
@@ -16,7 +16,7 @@ from vllm.benchmarks.datasets import RandomMultiModalDataset, SampleRequest
@pytest.fixture(scope="session")
def hf_tokenizer() -> PreTrainedTokenizerBase:
"""Use a small, commonly available tokenizer."""
return AutoTokenizer.from_pretrained("gpt2")
return AutoTokenizer.from_pretrained("openai-community/gpt2")
@pytest.fixture
+2 -2
View File
@@ -13,7 +13,7 @@ from vllm.benchmarks.datasets.create_txt_slices_dataset import create_txt_slices
@pytest.fixture(scope="session")
def hf_tokenizer() -> PreTrainedTokenizerBase:
# Use a small, commonly available tokenizer
return AutoTokenizer.from_pretrained("gpt2")
return AutoTokenizer.from_pretrained("openai-community/gpt2")
text_content = """
@@ -39,7 +39,7 @@ def test_create_txt_slices_jsonl(
create_txt_slices_jsonl(
input_path=str(txt_path),
output_path=str(jsonl_path),
tokenizer_name="gpt2",
tokenizer_name="openai-community/gpt2",
num_prompts=10,
input_len=10,
output_len=10,
+2 -2
View File
@@ -502,7 +502,7 @@ def test_gpt2_cache_hit(monkeypatch: pytest.MonkeyPatch):
m.setenv("VLLM_USE_AOT_COMPILE", "1")
# First compilation - initialize model and generate
llm_model = LLM(
model="gpt2",
model="openai-community/gpt2",
compilation_config=CompilationConfig(
mode=CompilationMode.VLLM_COMPILE,
),
@@ -519,7 +519,7 @@ def test_gpt2_cache_hit(monkeypatch: pytest.MonkeyPatch):
# Second compilation - should hit cache
m.setenv("VLLM_FORCE_AOT_LOAD", "1")
llm_model = LLM(
model="gpt2",
model="openai-community/gpt2",
compilation_config=CompilationConfig(
mode=CompilationMode.VLLM_COMPILE,
),
@@ -24,7 +24,7 @@ from vllm.utils.torch_utils import is_torch_equal_or_newer
def get_test_models():
"""Get list of models to test based on PyTorch version"""
models = [
"gpt2",
"openai-community/gpt2",
"Qwen/Qwen2-7B-Instruct",
"meta-llama/Llama-3.1-8B",
]
+1 -1
View File
@@ -114,7 +114,7 @@ TEXT_GENERATION_MODELS = {
"tiiuae/falcon-7b": PPTestSettings.fast(),
"google/gemma-1.1-2b-it": PPTestSettings.fast(),
"google/gemma-2-9b": PPTestSettings.fast(),
"gpt2": PPTestSettings.fast(),
"openai-community/gpt2": PPTestSettings.fast(),
"EleutherAI/gpt-j-6b": PPTestSettings.fast(),
"EleutherAI/pythia-1.4b": PPTestSettings.fast(),
"ibm/PowerLM-3b": PPTestSettings.fast(),
@@ -18,7 +18,7 @@ from vllm.renderers.embed_utils import safe_load_prompt_embeds
@pytest.mark.asyncio
async def test_empty_prompt():
model_name = "gpt2"
model_name = "openai-community/gpt2"
server_args = ["--enforce-eager"]
with RemoteOpenAIServer(model_name, server_args) as remote_server:
client = remote_server.get_async_client()
@@ -38,7 +38,7 @@ async def test_empty_prompt():
@pytest.mark.asyncio
async def test_out_of_vocab_token_ids():
model_name = "gpt2"
model_name = "openai-community/gpt2"
server_args = ["--enforce-eager"]
with RemoteOpenAIServer(model_name, server_args) as remote_server:
client = remote_server.get_async_client()
@@ -68,7 +68,8 @@ def server(request):
if request.param is not None:
args += ["--attention-backend", request.param]
if "AITER" in request.param:
env_dict = _AITER_ENV
# TODO: re-enable once AITER reenables fp16 unified attention.
pytest.skip("ROCM_AITER_UNIFIED_ATTN does not support fp16")
with RemoteOpenAIServer(MODEL_NAME, args, env_dict=env_dict) as remote_server:
yield remote_server
+22 -13
View File
@@ -74,7 +74,7 @@ class FakeHarmonyParser(HarmonyParser):
self.reasoning_parser = None
self.tool_parser = None
self._chunk_results: list[ChunkResult] = []
self._flush_results: list[Segment | None] = []
self._flush_results: list[list[Segment]] = []
self.processed_chunks: list[list[int]] = []
def enqueue_chunk_result(
@@ -89,7 +89,7 @@ class FakeHarmonyParser(HarmonyParser):
)
)
def enqueue_flush_result(self, segment: Segment | None) -> None:
def enqueue_flush_result(self, segment: list[Segment]) -> None:
self._flush_results.append(segment)
def process_chunk(self, token_ids) -> ChunkResult:
@@ -98,10 +98,10 @@ class FakeHarmonyParser(HarmonyParser):
return self._chunk_results.pop(0)
return ChunkResult(segments=[], reasoning_token_count=0)
def flush(self) -> Segment | None:
def flush(self) -> list[Segment]:
if self._flush_results:
return self._flush_results.pop(0)
return None
return []
def make_harmony_context(
@@ -598,13 +598,21 @@ async def test_streaming_message_synchronization():
content=[TextContent(text=response_text)],
recipient=Role.USER,
)
flush_segment = Segment(
channel="commentary",
recipient=None,
delta="",
completed_message=message,
)
parser.enqueue_flush_result(flush_segment)
flush_segments = [
Segment(
channel="final",
recipient=None,
delta=response_text,
completed_message=None,
),
Segment(
channel="final",
recipient=None,
delta="",
completed_message=message,
),
]
parser.enqueue_flush_result(flush_segments)
# Create another output to trigger synchronization via flush()
context.append_output(
@@ -618,8 +626,9 @@ async def test_streaming_message_synchronization():
assert context.num_init_messages == 1
assert context._messages[2].content[0].text == response_text
assert context.last_append_flush_status is True
assert len(context.last_append_segments) == 1
assert context.last_append_segments[0].completed_message is message
assert len(context.last_append_segments) == 2
assert context.last_append_segments[-2].delta == response_text
assert context.last_append_segments[-1].completed_message is message
def test_turn_metrics_copy_and_reset():
@@ -113,7 +113,7 @@ def _build_decode_metadata():
# stub with the attribute is enough for metadata construction.
layer_name = "placeholder"
vllm_config.compilation_config.static_forward_context[layer_name] = (
types.SimpleNamespace(prefill_backend=None)
types.SimpleNamespace(prefill_backend=torch.empty((1,)))
)
init_workspace_manager(device)
+10 -1
View File
@@ -2,6 +2,7 @@
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
import importlib.metadata
import types
from dataclasses import dataclass
from importlib.util import find_spec
@@ -10,7 +11,7 @@ import torch
from packaging import version
from tests.kernels.moe.utils import check_accuracy
from vllm._aiter_ops import is_aiter_found
from vllm._aiter_ops import is_aiter_found, rocm_aiter_ops
from vllm.platforms import current_platform
from vllm.utils.flashinfer import has_flashinfer
@@ -1247,6 +1248,7 @@ def test_rocm_mxfp4_moe_oracle(
num_tokens: int,
hidden_size: int,
intermediate_size: int,
monkeypatch: pytest.MonkeyPatch,
):
"""
Test ROCm MXFP4 MoE using oracle functions.
@@ -1268,6 +1270,7 @@ def test_rocm_mxfp4_moe_oracle(
if config["requires_gfx950"] and not ROCM_GFX950:
pytest.skip(f"Backend {backend_name} requires GFX950")
import vllm.distributed.parallel_state as ps
from vllm.config import VllmConfig, set_current_vllm_config
from vllm.model_executor.layers.fused_moe.activation import MoEActivation
from vllm.model_executor.layers.fused_moe.oracle.mxfp4 import (
@@ -1282,6 +1285,12 @@ def test_rocm_mxfp4_moe_oracle(
# Initialize workspace manager (needed for modular kernels)
init_workspace_manager(torch.accelerator.current_device_index())
# Set up the TP Group to prevent failure on should_use_cdna4_mx_scale_swizzle check
monkeypatch.setattr(ps, "_TP", types.SimpleNamespace(world_size=1))
# AITER must be enabled or aiter_mxfp4_w4a8_moe asserts before dispatch.
monkeypatch.setattr(rocm_aiter_ops, "_AITER_ENABLED", True)
# Map string to enum
backend = Mxfp4MoeBackend[backend_name]
+1
View File
@@ -115,6 +115,7 @@ def test_chatglm3_lora_tp4_fully_sharded_loras(chatglm3_lora_files):
enable_lora=True,
max_loras=2,
max_lora_rank=64,
max_num_seqs=16,
tensor_parallel_size=4,
trust_remote_code=True,
fully_sharded_loras=True,
@@ -119,6 +119,6 @@ def test_runai_invalid_extra_config_leaves_environ_untouched():
# os.environ (all values are validated before any global mutation).
with patch.dict(os.environ, {}, clear=False):
os.environ.pop("RUNAI_STREAMER_CONCURRENCY", None)
with pytest.raises(ValueError, match="memory_limit must be a positive integer"):
with pytest.raises(ValueError, match="memory_limit must be an integer >= -1"):
_runai_loader({"concurrency": 16, "memory_limit": -5})
assert "RUNAI_STREAMER_CONCURRENCY" not in os.environ
@@ -1,7 +1,5 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
import contextlib
import pytest
import pytest_asyncio
from mistral_common.protocol.transcription.request import (
@@ -12,7 +10,7 @@ from mistral_common.tokens.tokenizers.audio import Audio
from mistral_common.tokens.tokenizers.mistral import MistralTokenizer
from mistral_common.tokens.tokenizers.tekken import SpecialTokenPolicy
from vllm import LLM, EngineArgs, SamplingParams
from vllm import LLM, SamplingParams
from vllm.assets.audio import AudioAsset
from vllm.engine.arg_utils import AsyncEngineArgs
from vllm.utils.math_utils import cdiv
@@ -100,77 +98,88 @@ def tokenizer() -> MistralTokenizer:
return MistralTokenizer.from_hf_hub(MODEL_NAME)
@pytest.fixture
def engine(monkeypatch: pytest.MonkeyPatch):
# Disable multiprocessing allows us to access model executor from LLM engine
monkeypatch.setenv("VLLM_ENABLE_V1_MULTIPROCESSING", "0")
engine_args = EngineArgs(**ENGINE_CONFIG)
llm = LLM.from_engine_args(engine_args)
try:
yield llm
finally:
with contextlib.suppress(Exception):
llm.llm_engine.engine_core.shutdown()
import torch
torch.accelerator.empty_cache()
@pytest_asyncio.fixture
async def async_engine():
gpu_memory_utilization = ENGINE_CONFIG.get("gpu_memory_utilization", 0.9)
from vllm.platforms import current_platform
if current_platform.is_rocm():
from tests.utils import wait_for_rocm_memory_to_settle
wait_for_rocm_memory_to_settle(threshold_ratio=1.0 - gpu_memory_utilization)
engine_args = AsyncEngineArgs(**ENGINE_CONFIG)
llm = AsyncLLM.from_engine_args(engine_args)
try:
yield llm
finally:
llm.shutdown()
shutdown_timeout = 60.0 if current_platform.is_rocm() else None
llm.shutdown(timeout=shutdown_timeout)
del llm
import torch
torch._dynamo.reset()
from vllm.distributed import cleanup_dist_env_and_memory
cleanup_dist_env_and_memory()
from tests.utils import wait_for_rocm_memory_to_settle
wait_for_rocm_memory_to_settle(threshold_ratio=1.0 - gpu_memory_utilization)
def test_voxtral_realtime_forward(audio_assets, tokenizer, engine):
assert_encoder_kv_cache_spec(engine)
audio_config = tokenizer.instruct_tokenizer.tokenizer.audio
def test_voxtral_realtime_forward(audio_assets, tokenizer, vllm_runner, monkeypatch):
monkeypatch.setenv("VLLM_ENABLE_V1_MULTIPROCESSING", "0")
def from_file(file_path: str):
audio = Audio.from_file(file_path, strict=False)
req = TranscriptionRequest(
audio=audio.to_base64(audio.format),
streaming=StreamingMode.OFFLINE,
language=None,
vllm_kwargs = {**ENGINE_CONFIG}
vllm_kwargs["model_name"] = vllm_kwargs.pop("model")
with vllm_runner(**vllm_kwargs) as vllm_model:
assert_encoder_kv_cache_spec(vllm_model.llm)
audio_config = tokenizer.instruct_tokenizer.tokenizer.audio
def from_file(file_path: str):
audio = Audio.from_file(file_path, strict=False)
req = TranscriptionRequest(
audio=audio.to_base64(audio.format),
streaming=StreamingMode.OFFLINE,
language=None,
)
tokenized = tokenizer.instruct_tokenizer.encode_transcription(req)
return (tokenized.tokens, tokenized.audios[0].audio_array)
tokenized_list = [
from_file(audio_asset.get_local_path()) for audio_asset in audio_assets
]
inputs = []
sampling_params = []
for tokens, audio_array in tokenized_list:
num_samples = audio_array.shape[0]
max_tokens = audio_config.num_audio_tokens(num_samples) - len(tokens) - 1
sampling_params.append(
SamplingParams(temperature=0.0, max_tokens=max_tokens)
)
input_dict = {
"multi_modal_data": {"audio": [(audio_array, None)]},
"prompt_token_ids": tokens,
}
inputs.append(input_dict)
outputs = vllm_model.llm.generate(
inputs,
sampling_params=sampling_params,
)
tokenized = tokenizer.instruct_tokenizer.encode_transcription(req)
return (tokenized.tokens, tokenized.audios[0].audio_array)
tokenized_list = [
from_file(audio_asset.get_local_path()) for audio_asset in audio_assets
]
inputs = []
sampling_params = []
for tokens, audio_array in tokenized_list:
num_samples = audio_array.shape[0]
max_tokens = audio_config.num_audio_tokens(num_samples) - len(tokens) - 1
sampling_params.append(SamplingParams(temperature=0.0, max_tokens=max_tokens))
input_dict = {
"multi_modal_data": {"audio": [(audio_array, None)]},
"prompt_token_ids": tokens,
}
inputs.append(input_dict)
outputs = engine.generate(
inputs,
sampling_params=sampling_params,
)
texts = _normalize([out.outputs[0].text for out in outputs])
for i, (got, expected) in enumerate(zip(texts, EXPECTED_TEXT)):
assert got == expected, (
f"Output mismatch at index {i}:\n"
f" got: {got!r}\n"
f" expected: {expected!r}"
)
texts = _normalize([out.outputs[0].text for out in outputs])
for i, (got, expected) in enumerate(zip(texts, EXPECTED_TEXT)):
assert got == expected, (
f"Output mismatch at index {i}:\n"
f" got: {got!r}\n"
f" expected: {expected!r}"
)
@pytest.mark.asyncio
+1 -1
View File
@@ -296,7 +296,7 @@ _TEXT_GENERATION_EXAMPLE_MODELS = {
"GlmMoeDsaForCausalLM": _HfExamplesInfo(
"zai-org/GLM-5", min_transformers_version="5.0.1", is_available_online=False
),
"GPT2LMHeadModel": _HfExamplesInfo("openai-community/gpt2", {"alias": "gpt2"}),
"GPT2LMHeadModel": _HfExamplesInfo("openai-community/gpt2"),
"GPTBigCodeForCausalLM": _HfExamplesInfo(
"bigcode/starcoder",
extras={
+268
View File
@@ -0,0 +1,268 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Tests for DeepSeek V3.2 parser engine semantics.
V3.2 uses the same DSML parameter format as V4 but wraps tool calls in
``<DSMLfunction_calls>`` instead of ``<DSMLtool_calls>`` and has
no reasoning (``<think>``/``</think>``) support.
"""
import json
import pytest
from tests.parser.engine.conftest import make_mock_tokenizer
from tests.parser.engine.streaming_helpers import (
collect_content,
collect_function_name,
collect_tool_arguments,
simulate_tool_streaming,
)
from vllm.parser.deepseek_v4 import (
DSML_INVOKE_END,
DSML_INVOKE_NAME_END,
DSML_INVOKE_PREFIX,
)
from vllm.parser.deepseek_v32 import (
DSML_FUNC_END,
DSML_FUNC_START,
DeepSeekV32Parser,
)
from vllm.parser.engine.parser_engine_config import ParserState
_PARAM_OPEN = 'DSMLparameter name="{name}" string="{is_str}">'
_PARAM_CLOSE = "</DSMLparameter>"
def _param(name: str, is_str: str, value: str) -> str:
return f"<{_PARAM_OPEN.format(name=name, is_str=is_str)}{value}{_PARAM_CLOSE}"
def _invoke(name: str, *params: str) -> str:
body = "\n".join(params)
return (
f"{DSML_INVOKE_PREFIX}{name}{DSML_INVOKE_NAME_END}\n{body}\n{DSML_INVOKE_END}"
)
def _func_calls(*invocations: str) -> str:
body = "\n".join(invocations)
return f"{DSML_FUNC_START}\n{body}\n{DSML_FUNC_END}"
def _make_tool(name, properties):
from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionToolsParam,
)
return ChatCompletionToolsParam(
type="function",
function={
"name": name,
"parameters": {
"type": "object",
"properties": properties,
},
},
)
@pytest.fixture
def mock_tokenizer():
return make_mock_tokenizer({})
@pytest.fixture
def mock_request():
from unittest.mock import MagicMock
from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionRequest,
)
req = MagicMock(spec=ChatCompletionRequest)
req.tools = []
req.tool_choice = "auto"
return req
# ── Non-streaming extraction ────────────────────────────────────────
class TestNonStreaming:
def test_no_tool_call(self, mock_tokenizer, mock_request):
parser = DeepSeekV32Parser(mock_tokenizer)
result = parser.extract_tool_calls("Hello world", mock_request)
assert not result.tools_called
assert result.content == "Hello world"
def test_single_tool(self, mock_tokenizer, mock_request):
text = _func_calls(
_invoke("get_weather", _param("city", "true", "SF")),
)
parser = DeepSeekV32Parser(mock_tokenizer)
result = parser.extract_tool_calls(text, mock_request)
assert result.tools_called
assert len(result.tool_calls) == 1
assert result.tool_calls[0].function.name == "get_weather"
args = json.loads(result.tool_calls[0].function.arguments)
assert args == {"city": "SF"}
def test_parallel_tools(self, mock_tokenizer, mock_request):
text = _func_calls(
_invoke("get_weather", _param("city", "true", "SF")),
_invoke("get_weather", _param("city", "true", "NYC")),
)
parser = DeepSeekV32Parser(mock_tokenizer)
result = parser.extract_tool_calls(text, mock_request)
assert result.tools_called
assert len(result.tool_calls) == 2
assert json.loads(result.tool_calls[0].function.arguments) == {"city": "SF"}
assert json.loads(result.tool_calls[1].function.arguments) == {"city": "NYC"}
def test_content_before_tool_call(self, mock_tokenizer, mock_request):
text = "Let me check. " + _func_calls(
_invoke("search", _param("q", "true", "vllm")),
)
parser = DeepSeekV32Parser(mock_tokenizer)
result = parser.extract_tool_calls(text, mock_request)
assert result.tools_called
assert result.content is not None
assert "Let me check" in result.content
def test_non_string_params_json_parsed(self, mock_tokenizer, mock_request):
text = _func_calls(
_invoke(
"toggle",
_param("enabled", "false", "true"),
_param("count", "false", "42"),
),
)
parser = DeepSeekV32Parser(mock_tokenizer)
result = parser.extract_tool_calls(text, mock_request)
args = json.loads(result.tool_calls[0].function.arguments)
assert args["enabled"] is True
assert args["count"] == 42
def test_wrapper_unwrapping(self, mock_tokenizer, mock_request):
tool = _make_tool("get_weather", {"location": {"type": "string"}})
mock_request.tools = [tool]
text = _func_calls(
_invoke(
"get_weather",
_param("arguments", "false", '{"location":"Beijing"}'),
),
)
parser = DeepSeekV32Parser(mock_tokenizer, tools=[tool])
result = parser.extract_tool_calls(text, mock_request)
args = json.loads(result.tool_calls[0].function.arguments)
assert args == {"location": "Beijing"}
# ── Initial state ────────────────────────────────────────────────────
class TestInitialState:
def test_always_content(self, mock_tokenizer):
parser = DeepSeekV32Parser(mock_tokenizer)
cfg = parser.parser_engine_config
assert cfg.initial_state == ParserState.CONTENT
def test_ignores_thinking_kwargs(self, mock_tokenizer):
parser = DeepSeekV32Parser(
mock_tokenizer,
chat_template_kwargs={"thinking": True, "enable_thinking": True},
)
cfg = parser.parser_engine_config
assert cfg.initial_state == ParserState.CONTENT
# ── Streaming ────────────────────────────────────────────────────────
class TestStreaming:
def test_single_tool_streaming(self, mock_tokenizer, mock_request):
text = _func_calls(
_invoke("get_weather", _param("city", "true", "SF")),
)
parser = DeepSeekV32Parser(mock_tokenizer)
results = simulate_tool_streaming(parser, mock_request, list(text))
assert collect_function_name(results) == "get_weather"
args_json = collect_tool_arguments(results)
assert json.loads(args_json) == {"city": "SF"}
def test_content_before_tool_streaming(self, mock_tokenizer, mock_request):
text = "Checking... " + _func_calls(
_invoke("fn", _param("k", "true", "v")),
)
parser = DeepSeekV32Parser(mock_tokenizer)
results = simulate_tool_streaming(parser, mock_request, list(text))
content = collect_content(results)
assert "Checking" in content
def test_parallel_tools_streaming(self, mock_tokenizer, mock_request):
text = _func_calls(
_invoke("fn_a", _param("x", "true", "1")),
_invoke("fn_b", _param("y", "true", "2")),
)
parser = DeepSeekV32Parser(mock_tokenizer)
results = simulate_tool_streaming(parser, mock_request, list(text))
names = []
for delta, _ in results:
if delta and delta.tool_calls:
for tc in delta.tool_calls:
if tc.function and tc.function.name:
names.append(tc.function.name)
assert "fn_a" in names
assert "fn_b" in names
def test_no_tool_content_only(self, mock_tokenizer, mock_request):
text = "Just some text, no tools."
parser = DeepSeekV32Parser(mock_tokenizer)
results = simulate_tool_streaming(parser, mock_request, list(text))
content = collect_content(results)
assert "Just some text" in content
args = collect_tool_arguments(results)
assert args == ""
def test_streaming_wrapper_unwrap_consistency(self, mock_tokenizer, mock_request):
tool = _make_tool("get_weather", {"location": {"type": "string"}})
mock_request.tools = [tool]
parser = DeepSeekV32Parser(mock_tokenizer, tools=[tool])
chunks = [
DSML_FUNC_START,
_invoke(
"get_weather",
_param("arguments", "false", '{"location": "NYC"}'),
),
DSML_FUNC_END,
]
results = simulate_tool_streaming(parser, mock_request, chunks)
streamed_args = collect_tool_arguments(results)
final_delta, _ = results[-1]
finish_delta = parser.finish_streaming()
extracted = parser._build_extracted_result(final_delta, finish_delta)
assert extracted.tools_called is True
assert len(extracted.tool_calls) == 1
final_args = extracted.tool_calls[0].function.arguments
assert json.loads(final_args) == {"location": "NYC"}
assert '"arguments"' not in streamed_args
assert final_args.startswith(streamed_args)
def test_missing_invoke_end(self, mock_tokenizer, mock_request):
text = (
f"{DSML_FUNC_START}\n"
f"{DSML_INVOKE_PREFIX}fn{DSML_INVOKE_NAME_END}\n"
f"{_param('k', 'true', 'v')}\n"
f"{DSML_FUNC_END}"
)
parser = DeepSeekV32Parser(mock_tokenizer)
results = simulate_tool_streaming(parser, mock_request, list(text))
assert collect_function_name(results) == "fn"
args = json.loads(collect_tool_arguments(results))
assert args == {"k": "v"}
+922
View File
@@ -0,0 +1,922 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Tests for DeepSeek V4-specific parser engine semantics."""
import json
import pytest
from tests.parser.engine.conftest import make_mock_tokenizer
from tests.parser.engine.replay_harness import (
DUMMY_TOOLS,
MockTokenizer,
_test_request,
collect_output,
replay_streaming,
)
from tests.parser.engine.streaming_helpers import (
collect_content,
collect_function_name,
collect_tool_arguments,
simulate_reasoning_streaming,
simulate_tool_streaming,
)
from vllm.parser.abstract_parser import DelegatingParser
from vllm.parser.deepseek_v4 import (
DSML_INVOKE_END,
DSML_INVOKE_NAME_END,
DSML_INVOKE_PREFIX,
DSML_THINK_END,
DSML_THINK_START,
DSML_TOOL_END,
DSML_TOOL_START,
DeepSeekV4Parser,
_dsml_arg_converter,
_unwrap_wrapper_args,
deepseek_v4_config,
)
from vllm.parser.engine.registered_adapters import (
DeepSeekV4ParserReasoningAdapter,
DeepSeekV4ParserToolAdapter,
)
_THINK_START_ID = 50
_THINK_END_ID = 51
_PARAM_OPEN = 'DSMLparameter name="{name}" string="{is_str}">'
_PARAM_CLOSE = "</DSMLparameter>"
def _param(name: str, is_str: str, value: str) -> str:
return f"<{_PARAM_OPEN.format(name=name, is_str=is_str)}{value}{_PARAM_CLOSE}"
@pytest.fixture
def mock_tokenizer():
return make_mock_tokenizer(
{
DSML_THINK_START: _THINK_START_ID,
DSML_THINK_END: _THINK_END_ID,
}
)
# ── Arg converter unit tests ─────────────────────────────────────────
class TestArgConverter:
def _raw(self, *params: tuple[str, str, str]) -> str:
lines = [_param(n, s, v) for n, s, v in params]
return "\n" + "\n".join(lines) + "\n"
def test_string_param(self):
raw = self._raw(("city", "true", "杭州"))
result = json.loads(_dsml_arg_converter(raw, partial=False))
assert result == {"city": "杭州"}
def test_string_with_spaces_and_quotes(self):
raw = self._raw(("msg", "true", 'He said "hello world"'))
result = json.loads(_dsml_arg_converter(raw, partial=False))
assert result["msg"] == 'He said "hello world"'
def test_integer_param(self):
raw = self._raw(("count", "false", "42"))
result = json.loads(_dsml_arg_converter(raw, partial=False))
assert result["count"] == 42
assert isinstance(result["count"], int)
def test_float_param(self):
raw = self._raw(("ratio", "false", "3.14"))
result = json.loads(_dsml_arg_converter(raw, partial=False))
assert abs(result["ratio"] - 3.14) < 1e-9
def test_bool_param(self):
raw = self._raw(("flag", "false", "true"))
result = json.loads(_dsml_arg_converter(raw, partial=False))
assert result["flag"] is True
def test_array_param(self):
raw = self._raw(("items", "false", '["a", "b", "c"]'))
result = json.loads(_dsml_arg_converter(raw, partial=False))
assert result["items"] == ["a", "b", "c"]
def test_object_param(self):
raw = self._raw(("opts", "false", '{"key": "val"}'))
result = json.loads(_dsml_arg_converter(raw, partial=False))
assert result["opts"] == {"key": "val"}
def test_mixed_types(self):
raw = self._raw(
("location", "true", "Tokyo"),
("limit", "false", "10"),
("active", "false", "false"),
)
result = json.loads(_dsml_arg_converter(raw, partial=False))
assert result == {"location": "Tokyo", "limit": 10, "active": False}
def test_empty_args(self):
result = json.loads(_dsml_arg_converter("", partial=False))
assert result == {}
def test_invalid_json_fallback(self):
raw = self._raw(("data", "false", "[broken"))
result = json.loads(_dsml_arg_converter(raw, partial=False))
assert result["data"] == "[broken"
def test_chinese_chars_preserved_in_json(self):
raw = self._raw(("query", "true", "你好世界"))
raw_json = _dsml_arg_converter(raw, partial=False)
assert "你好世界" in raw_json
result = json.loads(raw_json)
assert result["query"] == "你好世界"
def test_partial_complete_plus_in_progress(self):
raw = self._raw(("city", "true", "Tokyo"))
raw += f"<{_PARAM_OPEN.format(name='unit', is_str='true')}celsi"
result = json.loads(_dsml_arg_converter(raw, partial=True))
assert result["city"] == "Tokyo"
assert result["unit"] == "celsi"
def test_partial_no_in_progress(self):
raw = self._raw(("city", "true", "Tokyo"))
result = json.loads(_dsml_arg_converter(raw, partial=True))
assert result == {"city": "Tokyo"}
def test_partial_value_with_angle_bracket(self):
raw = f"<{_PARAM_OPEN.format(name='code', is_str='true')}a<b"
result = json.loads(_dsml_arg_converter(raw, partial=True))
assert result == {"code": "a<b"}
def test_partial_value_with_angle_bracket_and_complete_param(self):
raw = self._raw(("city", "true", "Tokyo"))
raw += f"<{_PARAM_OPEN.format(name='expr', is_str='true')}x<5"
result = json.loads(_dsml_arg_converter(raw, partial=True))
assert result["city"] == "Tokyo"
assert result["expr"] == "x<5"
def test_null_string_false(self):
raw = self._raw(("val", "false", "null"))
result = json.loads(_dsml_arg_converter(raw, partial=False))
assert result["val"] is None
def test_string_true_not_json_parsed(self):
raw = self._raw(("n", "true", "42"))
result = json.loads(_dsml_arg_converter(raw, partial=False))
assert result["n"] == "42"
assert isinstance(result["n"], str)
# ── Bare </think> absorption and duplicate <think> absorption ─────────
class TestThinkTagAbsorption:
def test_bare_think_end_not_leaked(self, mock_tokenizer):
parser = DeepSeekV4Parser(mock_tokenizer)
chunks = ["</think>", "Here is the direct answer."]
reasoning, content = simulate_reasoning_streaming(parser, chunks)
assert reasoning == ""
assert "</think>" not in content
assert "Here is the direct answer" in content
def test_duplicate_think_start_absorbed(self, mock_tokenizer):
parser = DeepSeekV4Parser(
mock_tokenizer, chat_template_kwargs={"thinking": True}
)
chunks = [
"<think>\n",
"Some reasoning.\n",
"</think>\n",
"Answer.",
]
reasoning, content = simulate_reasoning_streaming(parser, chunks)
assert "Some reasoning" in reasoning
assert "Answer" in content
# ── Missing </DSMLinvoke> before </DSMLtool_calls> ────────────
class TestMissingInvokeEnd:
def test_non_streaming(self, mock_tokenizer, mock_request):
parser = DeepSeekV4Parser(mock_tokenizer)
text = (
f"{DSML_TOOL_START}"
f"{DSML_INVOKE_PREFIX}get_weather{DSML_INVOKE_NAME_END}\n"
f"{_param('location', 'true', 'NYC')}\n"
f"{DSML_TOOL_END}"
)
result = parser.extract_tool_calls(text, mock_request)
assert result.tools_called is True
assert len(result.tool_calls) == 1
assert result.tool_calls[0].function.name == "get_weather"
args = json.loads(result.tool_calls[0].function.arguments)
assert args == {"location": "NYC"}
def test_streaming_with_trailing_content(self, mock_tokenizer, mock_request):
parser = DeepSeekV4Parser(mock_tokenizer)
chunks = [
DSML_TOOL_START,
f"{DSML_INVOKE_PREFIX}get_weather{DSML_INVOKE_NAME_END}\n"
f"{_param('location', 'true', 'NYC')}\n",
DSML_TOOL_END,
"Done.",
]
results = simulate_tool_streaming(parser, mock_request, chunks)
assert collect_function_name(results) == "get_weather"
args = json.loads(collect_tool_arguments(results))
assert args == {"location": "NYC"}
assert "Done." in collect_content(results)
# ── Thinking mode initial state ──────────────────────────────────────
class TestThinkingModeConfig:
def test_thinking_true_starts_in_reasoning(self):
cfg = deepseek_v4_config(thinking=True)
assert cfg.initial_state.name == "REASONING"
def test_thinking_false_starts_in_content(self):
cfg = deepseek_v4_config(thinking=False)
assert cfg.initial_state.name == "CONTENT"
def test_enable_thinking_kwarg(self, mock_tokenizer):
p = DeepSeekV4Parser(
mock_tokenizer, chat_template_kwargs={"enable_thinking": True}
)
assert p.parser_engine_config.initial_state.name == "REASONING"
def test_no_thinking_kwarg_defaults_to_content(self, mock_tokenizer):
p = DeepSeekV4Parser(mock_tokenizer)
assert p.parser_engine_config.initial_state.name == "CONTENT"
def test_thinking_mode_reasoning_without_tags(self, mock_tokenizer):
parser = DeepSeekV4Parser(
mock_tokenizer, chat_template_kwargs={"thinking": True}
)
chunks = [
"\n\nLet me consider ",
"this carefully.\n",
"</think>\n",
"Here is the result.",
]
reasoning, content = simulate_reasoning_streaming(parser, chunks)
assert "Let me consider" in reasoning
assert "Here is the result" in content
def test_thinking_mode_all_reasoning_no_end_tag(self, mock_tokenizer):
parser = DeepSeekV4Parser(
mock_tokenizer, chat_template_kwargs={"thinking": True}
)
chunks = ["I'll review ", "the PR."]
reasoning, content = simulate_reasoning_streaming(parser, chunks)
assert "review" in reasoning
assert "the PR" in reasoning
assert content == ""
def test_reasoning_effort_none_overrides_enable_thinking(self, mock_tokenizer):
p = DeepSeekV4Parser(
mock_tokenizer,
chat_template_kwargs={
"enable_thinking": True,
"reasoning_effort": "none",
},
)
assert p.parser_engine_config.initial_state.name == "CONTENT"
# ── Implicit reasoning end (missing </think> before tool calls) ─────
class TestImplicitReasoningEnd:
"""Tool call markers end reasoning implicitly when </think> is missing.
DeepSeek V4 models occasionally omit </think> before emitting tool calls.
The (REASONING, TOOL_START) transition handles this gracefully.
"""
@pytest.fixture
def thinking_parser(self, mock_tokenizer):
return DeepSeekV4Parser(mock_tokenizer, chat_template_kwargs={"thinking": True})
def _reasoning_then_tool(self, reasoning_text: str) -> str:
return reasoning_text + _tool_calls(
_invoke("get_weather", ("location", "true", "NYC")),
)
def test_non_streaming_extract_reasoning_implicit_end(self, thinking_parser):
text = self._reasoning_then_tool("Let me look up the weather.\n\n")
reasoning, content = thinking_parser.extract_reasoning(text, None)
assert reasoning == "Let me look up the weather."
assert DSML_TOOL_START not in reasoning
assert DSML_INVOKE_PREFIX not in reasoning
assert content is None
def test_non_streaming_extract_tool_calls_implicit_end(
self, thinking_parser, mock_request
):
text = self._reasoning_then_tool("Let me look up the weather.\n\n")
result = thinking_parser.extract_tool_calls(text, mock_request)
assert result.tools_called is True
assert len(result.tool_calls) == 1
assert result.tool_calls[0].function.name == "get_weather"
args = json.loads(result.tool_calls[0].function.arguments)
assert args == {"location": "NYC"}
def test_non_streaming_parse_implicit_end(self, thinking_parser, mock_request):
text = self._reasoning_then_tool("Let me look up the weather.\n\n")
reasoning, content, tool_calls = thinking_parser.parse(text, mock_request)
assert reasoning == "Let me look up the weather."
assert content is None
assert tool_calls is not None
assert len(tool_calls) == 1
assert tool_calls[0].name == "get_weather"
args = json.loads(tool_calls[0].arguments)
assert args == {"location": "NYC"}
def test_streaming_reasoning_implicit_end(self, thinking_parser):
chunks = [
"Let me look up the weather.\n\n",
DSML_TOOL_START,
DSML_INVOKE_PREFIX + "get_weather" + DSML_INVOKE_NAME_END,
]
reasoning, content = simulate_reasoning_streaming(thinking_parser, chunks)
assert reasoning == "Let me look up the weather."
assert DSML_TOOL_START not in reasoning
assert DSML_INVOKE_PREFIX not in reasoning
def test_streaming_tool_extraction_implicit_end(
self, thinking_parser, mock_request
):
chunks = [
"Let me check.\n\n",
DSML_TOOL_START,
DSML_INVOKE_PREFIX
+ "get_weather"
+ DSML_INVOKE_NAME_END
+ "\n"
+ _param("location", "true", "NYC")
+ "\n"
+ DSML_INVOKE_END,
DSML_TOOL_END,
]
results = simulate_tool_streaming(thinking_parser, mock_request, chunks)
assert collect_function_name(results) == "get_weather"
args = json.loads(collect_tool_arguments(results))
assert args == {"location": "NYC"}
def test_thinking_false_explicit_think_then_tool_call(self, mock_tokenizer):
parser = DeepSeekV4Parser(mock_tokenizer)
chunks = [
DSML_THINK_START,
"Let me check the weather.",
DSML_TOOL_START,
DSML_INVOKE_PREFIX + "get_weather" + DSML_INVOKE_NAME_END,
]
reasoning, content = simulate_reasoning_streaming(parser, chunks)
assert "Let me check the weather" in reasoning
assert DSML_TOOL_START not in reasoning
assert DSML_THINK_START not in reasoning
def test_non_streaming_parallel_tools_after_implicit_end(
self, thinking_parser, mock_request
):
text = "I need both.\n\n" + _tool_calls(
_invoke("get_weather", ("location", "true", "NYC")),
_invoke("get_time", ("timezone", "true", "EST")),
)
result = thinking_parser.extract_tool_calls(text, mock_request)
assert result.tools_called is True
assert len(result.tool_calls) == 2
assert result.tool_calls[0].function.name == "get_weather"
assert result.tool_calls[1].function.name == "get_time"
def test_streaming_implicit_end_trailing_whitespace_stripped(self, thinking_parser):
chunks = [
"Reasoning.\n\n\n",
DSML_TOOL_START,
DSML_INVOKE_PREFIX + "func" + DSML_INVOKE_NAME_END,
]
reasoning, content = simulate_reasoning_streaming(thinking_parser, chunks)
assert reasoning == "Reasoning."
# ── Wrapper argument unwrapping ──────────────────────────────────────
class TestWrapperUnwrapping:
def test_unwrap_arguments_wrapper(self):
from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionToolsParam,
)
tool = ChatCompletionToolsParam(
type="function",
function={
"name": "get_weather",
"parameters": {
"type": "object",
"properties": {"location": {"type": "string"}},
},
},
)
result = _unwrap_wrapper_args(
'{"arguments": {"location": "Beijing"}}',
[tool],
"get_weather",
)
assert json.loads(result) == {"location": "Beijing"}
def test_unwrap_input_wrapper(self):
from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionToolsParam,
)
tool = ChatCompletionToolsParam(
type="function",
function={
"name": "get_weather",
"parameters": {
"type": "object",
"properties": {"location": {"type": "string"}},
},
},
)
result = _unwrap_wrapper_args(
'{"input": {"location": "Beijing"}}',
[tool],
"get_weather",
)
assert json.loads(result) == {"location": "Beijing"}
def test_no_unwrap_when_key_in_schema(self):
from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionToolsParam,
)
tool = ChatCompletionToolsParam(
type="function",
function={
"name": "func",
"parameters": {
"type": "object",
"properties": {"arguments": {"type": "string"}},
},
},
)
result = _unwrap_wrapper_args(
'{"arguments": "some value"}',
[tool],
"func",
)
assert json.loads(result) == {"arguments": "some value"}
def test_no_unwrap_when_no_tools(self):
result = _unwrap_wrapper_args(
'{"arguments": {"location": "Beijing"}}',
None,
"get_weather",
)
assert json.loads(result) == {"arguments": {"location": "Beijing"}}
def test_unwrap_json_string_inner(self):
from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionToolsParam,
)
tool = ChatCompletionToolsParam(
type="function",
function={
"name": "get_weather",
"parameters": {
"type": "object",
"properties": {"location": {"type": "string"}},
},
},
)
result = _unwrap_wrapper_args(
'{"arguments": "{\\"location\\": \\"Beijing\\"}"}',
[tool],
"get_weather",
)
assert json.loads(result) == {"location": "Beijing"}
# ── Parallel tool call wrapper unwrapping ───────────────────────────
def _make_tool(name, properties):
from vllm.entrypoints.openai.chat_completion.protocol import ( # noqa: E501
ChatCompletionToolsParam,
)
return ChatCompletionToolsParam(
type="function",
function={
"name": name,
"parameters": {
"type": "object",
"properties": properties,
},
},
)
def _invoke(name, *params):
body = "\n".join(_param(n, s, v) for n, s, v in params)
return (
f"{DSML_INVOKE_PREFIX}{name}{DSML_INVOKE_NAME_END}\n{body}\n{DSML_INVOKE_END}"
)
def _tool_calls(*invokes):
return DSML_TOOL_START + "\n".join(invokes) + DSML_TOOL_END
class TestParallelUnwrapping:
@pytest.fixture
def weather_tool(self):
return _make_tool(
"get_weather",
{
"location": {"type": "string"},
"unit": {"type": "string"},
},
)
@pytest.fixture
def time_tool(self):
return _make_tool(
"get_time",
{"timezone": {"type": "string"}},
)
@pytest.mark.parametrize(
"weather_args, expected",
[
(
'{"location": "NYC", "unit": "celsius"}',
{"location": "NYC", "unit": "celsius"},
),
('{"location": "NYC"}', {"location": "NYC"}),
],
ids=["all_props", "subset_props"],
)
def test_unwrap_parallel_uses_correct_schema(
self,
mock_tokenizer,
mock_request,
weather_tool,
time_tool,
weather_args,
expected,
):
tools = [weather_tool, time_tool]
parser = DeepSeekV4Parser(mock_tokenizer, tools=tools)
mock_request.tools = tools
text = _tool_calls(
_invoke("get_weather", ("arguments", "false", weather_args)),
_invoke("get_time", ("timezone", "true", "EST")),
)
result = parser.extract_tool_calls(text, mock_request)
assert result.tools_called is True
assert len(result.tool_calls) == 2
assert result.tool_calls[0].function.name == "get_weather"
args0 = json.loads(result.tool_calls[0].function.arguments)
assert args0 == expected
assert result.tool_calls[1].function.name == "get_time"
args1 = json.loads(result.tool_calls[1].function.arguments)
assert args1 == {"timezone": "EST"}
def test_unwrap_parallel_streaming(
self, mock_tokenizer, mock_request, weather_tool, time_tool
):
tools = [weather_tool, time_tool]
parser = DeepSeekV4Parser(mock_tokenizer, tools=tools)
mock_request.tools = tools
chunks = [
DSML_TOOL_START,
_invoke(
"get_weather",
("arguments", "false", '{"location": "NYC"}'),
),
_invoke("get_time", ("timezone", "true", "EST")),
DSML_TOOL_END,
]
results = simulate_tool_streaming(parser, mock_request, chunks)
final_delta, _ = results[-1]
finish_delta = parser.finish_streaming()
extracted = parser._build_extracted_result(final_delta, finish_delta)
assert extracted.tools_called is True
assert len(extracted.tool_calls) == 2
args0 = json.loads(extracted.tool_calls[0].function.arguments)
assert args0 == {"location": "NYC"}
args1 = json.loads(extracted.tool_calls[1].function.arguments)
assert args1 == {"timezone": "EST"}
def test_no_unwrap_parallel_when_no_match(
self, mock_tokenizer, mock_request, weather_tool, time_tool
):
tools = [weather_tool, time_tool]
parser = DeepSeekV4Parser(mock_tokenizer, tools=tools)
mock_request.tools = tools
text = _tool_calls(
_invoke(
"get_weather",
("arguments", "false", '{"unknown_key": "val"}'),
),
_invoke("get_time", ("timezone", "true", "EST")),
)
result = parser.extract_tool_calls(text, mock_request)
assert len(result.tool_calls) == 2
args0 = json.loads(result.tool_calls[0].function.arguments)
assert args0 == {"arguments": {"unknown_key": "val"}}
args1 = json.loads(result.tool_calls[1].function.arguments)
assert args1 == {"timezone": "EST"}
def test_unwrap_single_tool_still_works(self, mock_tokenizer, mock_request):
tool = _make_tool("get_weather", {"location": {"type": "string"}})
tools = [tool]
parser = DeepSeekV4Parser(mock_tokenizer, tools=tools)
mock_request.tools = tools
text = _tool_calls(
_invoke(
"get_weather",
("arguments", "false", '{"location": "Beijing"}'),
),
)
result = parser.extract_tool_calls(text, mock_request)
assert result.tools_called is True
assert len(result.tool_calls) == 1
args = json.loads(result.tool_calls[0].function.arguments)
assert args == {"location": "Beijing"}
# ── Streaming wrapper consistency ─────────────────────────────────────
class TestStreamingWrapperConsistency:
"""Streamed arg deltas must stay consistent with final extraction
when wrapper params like 'arguments' are unwrapped."""
def test_streaming_wrapper_unwrap_consistency(self, mock_tokenizer, mock_request):
tool = _make_tool("get_weather", {"location": {"type": "string"}})
tools = [tool]
parser = DeepSeekV4Parser(mock_tokenizer, tools=tools)
mock_request.tools = tools
chunks = [
DSML_TOOL_START,
_invoke(
"get_weather",
("arguments", "false", '{"location": "NYC"}'),
),
DSML_TOOL_END,
]
results = simulate_tool_streaming(parser, mock_request, chunks)
streamed_args = collect_tool_arguments(results)
final_delta, _ = results[-1]
finish_delta = parser.finish_streaming()
extracted = parser._build_extracted_result(final_delta, finish_delta)
assert extracted.tools_called is True
assert len(extracted.tool_calls) == 1
final_args = extracted.tool_calls[0].function.arguments
assert json.loads(final_args) == {"location": "NYC"}
assert '"arguments"' not in streamed_args, (
f"Streamed args should not contain wrapper key, got: {streamed_args!r}"
)
assert final_args.startswith(streamed_args), (
f"Extracted args {final_args!r} "
f"should start with streamed args {streamed_args!r}"
)
# ── DelegatingParser: large delta with </think> + tool calls ─────────
_DSV4_FULL_VOCAB = {
DSML_THINK_START: 128821,
DSML_THINK_END: 128822,
DSML_TOOL_START: 128823,
DSML_TOOL_END: 128824,
}
class _DeepSeekV4Delegating(DelegatingParser):
reasoning_parser_cls = DeepSeekV4ParserReasoningAdapter
tool_parser_cls = DeepSeekV4ParserToolAdapter
def _dsv4_tokens(
reasoning: str,
tool_name: str,
params: list[tuple[str, str, str]],
) -> list[tuple[int, str]]:
"""Build a token sequence: reasoning + </think> + DSML tool block."""
tokens: list[tuple[int, str]] = []
tid = 100
for word in reasoning.split(" "):
prefix = " " if tokens else ""
tokens.append((tid, prefix + word))
tid += 1
tokens.append((_DSV4_FULL_VOCAB[DSML_THINK_END], DSML_THINK_END))
tokens.append((tid, "\n\n"))
tid += 1
tokens.append((_DSV4_FULL_VOCAB[DSML_TOOL_START], DSML_TOOL_START))
tokens.append((tid, "\n"))
tid += 1
invoke_prefix_text = f"{DSML_INVOKE_PREFIX}{tool_name}{DSML_INVOKE_NAME_END}"
tokens.append((tid, invoke_prefix_text))
tid += 1
tokens.append((tid, "\n"))
tid += 1
for name, is_str, value in params:
param_text = _param(name, is_str, value)
tokens.append((tid, param_text))
tid += 1
tokens.append((tid, "\n"))
tid += 1
tokens.append((tid, DSML_INVOKE_END))
tid += 1
tokens.append((tid, "\n"))
tid += 1
tokens.append((_DSV4_FULL_VOCAB[DSML_TOOL_END], DSML_TOOL_END))
return tokens
class TestDelegatingParserLargeDelta:
"""Regression: tool calls lost when </think> + DSML arrive in same delta.
The DelegatingParser used by the serving layer splits reasoning and
tool parsing across two separate engine instances. When </think> and
the entire DSML tool block arrive in a single large streaming delta,
the content transfer from reasoning adapter to tool adapter must
preserve the tool call text.
"""
@pytest.fixture
def dsv4_tokens(self):
return _dsv4_tokens(
reasoning="The user wants the current weather in Berlin.",
tool_name="get_weather",
params=[
("location", "true", "Berlin"),
("units", "true", "celsius"),
],
)
@pytest.fixture
def dsv4_tokenizer(self, dsv4_tokens):
return MockTokenizer(
vocab=dict(_DSV4_FULL_VOCAB),
tokens=dsv4_tokens,
)
@pytest.mark.parametrize(
"chunk_size",
[1, 2, 3, 5, None],
ids=lambda c: f"chunk={c}",
)
def test_tool_calls_extracted_at_all_chunk_sizes(
self, dsv4_tokenizer, dsv4_tokens, chunk_size
):
parser = _DeepSeekV4Delegating(
dsv4_tokenizer,
chat_template_kwargs={"thinking": True},
)
deltas = replay_streaming(
parser,
dsv4_tokens,
chunk_size=chunk_size,
finished_on_last=True,
tools=DUMMY_TOOLS,
)
output = collect_output(deltas)
assert "The user wants" in output.reasoning
assert len(output.tool_calls) == 1, (
f"Expected 1 tool call but got {len(output.tool_calls)}; "
f"reasoning={output.reasoning!r}, content={output.content!r}"
)
assert output.tool_calls[0]["name"] == "get_weather"
args = json.loads(output.tool_calls[0]["arguments"])
assert args == {"location": "Berlin", "units": "celsius"}
def test_eos_drop_token_does_not_swallow_tool_calls(self):
"""Tool calls must survive when an EOS DROP token's ID is in
delta_token_ids but its text is absent from delta_text.
At large stream_interval the EOS token ID arrives in the same
delta as </think> + tool calls but the detokenizer strips the
EOS text. The scanner's _rebuild_from_anchors defers all text
after </think> when it can't find the EOS anchor text. The
reasoning adapter's finish_streaming must flush deferred text
as content (with skip_tool_parsing), not as tool calls.
"""
eos_text = "<end▁of▁sentence>"
eos_id = 128801
vocab = {
DSML_THINK_START: 128821,
DSML_THINK_END: 128822,
eos_text: eos_id,
}
reasoning = "The user wants weather."
tool_block = (
"\n\n"
+ DSML_TOOL_START
+ "\n"
+ DSML_INVOKE_PREFIX
+ "get_weather"
+ DSML_INVOKE_NAME_END
+ "\n"
+ _param("location", "true", "Berlin")
+ "\n"
+ DSML_INVOKE_END
+ "\n"
+ DSML_TOOL_END
)
# delta_text does NOT include EOS text (detokenizer strips it)
full_text = reasoning + DSML_THINK_END + tool_block
# Build token list: word-split reasoning, then special tokens,
# then word-split tool block content, then EOS.
# EOS ID is present but its text is NOT in delta_text.
tokens: list[tuple[int, str]] = []
tid = 100
for word in reasoning.split(" "):
pfx = " " if tokens else ""
tokens.append((tid, pfx + word))
tid += 1
tokens.append((128822, DSML_THINK_END))
for ch in tool_block:
tokens.append((tid, ch))
tid += 1
tokens.append((eos_id, eos_text))
all_ids = [t[0] for t in tokens]
tokenizer = MockTokenizer(vocab=vocab, tokens=tokens)
request = _test_request(tools=DUMMY_TOOLS)
# All-in-one delta: EOS ID in token_ids but text NOT in
# delta_text (detokenizer strips EOS). This is the scenario
# at large stream_interval.
parser = _DeepSeekV4Delegating(
tokenizer,
chat_template_kwargs={"thinking": True},
)
deltas = [
parser.parse_delta(
full_text,
all_ids,
request,
prompt_token_ids=[],
finished=True,
)
]
output = collect_output(deltas)
assert "The user wants" in output.reasoning
assert len(output.tool_calls) == 1, (
f"Expected 1 tool call but got {len(output.tool_calls)}; "
f"reasoning={output.reasoning!r}, content={output.content!r}"
)
assert output.tool_calls[0]["name"] == "get_weather"
args = json.loads(output.tool_calls[0]["arguments"])
assert args == {"location": "Berlin"}
+15 -12
View File
@@ -20,6 +20,7 @@ import pytest
from tests.parser.engine.replay_harness import (
DUMMY_TOOLS,
MockTokenizer,
Sample,
_test_request,
assert_no_terminal_leakage,
assert_parse_output,
@@ -100,6 +101,14 @@ def _discover_parsers() -> list[_ParserInfo]:
_PARSERS = _discover_parsers()
def _make_parser(parser_cls: type[ParserEngine], tokenizer, sample: Sample, **extra):
kwargs = dict(extra)
if sample.chat_template_kwargs:
kwargs["chat_template_kwargs"] = sample.chat_template_kwargs
return parser_cls(tokenizer, sample.tools, **kwargs)
_ENGINE_PARSERS: dict[str, type[ParserEngine]] = {
f"{p.name}_engine": p.parser_cls for p in _PARSERS
}
@@ -123,7 +132,7 @@ class TestReplayWithHoldback:
def test_replay(self, parser_cls, sample, terminals, chunk_size, holdback):
tokenizer = make_mock_tokenizer(sample)
parser = parser_cls(tokenizer, sample.tools)
parser = _make_parser(parser_cls, tokenizer, sample)
deltas = replay_streaming(
parser,
sample.tokens,
@@ -160,7 +169,7 @@ class TestTextHoldback:
def test_replay(self, parser_cls, sample, terminals, delay):
tokenizer = make_mock_tokenizer(sample)
parser = parser_cls(tokenizer, sample.tools)
parser = _make_parser(parser_cls, tokenizer, sample)
deltas = replay_with_text_holdback(
parser,
sample.tokens,
@@ -190,7 +199,7 @@ class TestReplay:
def test_replay(self, parser_cls, sample, terminals, chunk_size):
tokenizer = make_mock_tokenizer(sample)
parser = parser_cls(tokenizer, sample.tools)
parser = _make_parser(parser_cls, tokenizer, sample)
deltas = replay_streaming(
parser,
sample.tokens,
@@ -227,7 +236,7 @@ class TestDeferralFinish:
def test_misaligned_last_delta_with_finish(self, parser_cls, sample, tool_end_text):
tokenizer = make_mock_tokenizer(sample)
parser = parser_cls(tokenizer, sample.tools)
parser = _make_parser(parser_cls, tokenizer, sample)
request = _test_request()
@@ -528,10 +537,7 @@ class TestDropTokenReplay:
for sample in parser_info.samples:
injected = _inject_drop_tokens(sample)
tokenizer = make_mock_tokenizer(injected)
parser = parser_info.parser_cls(
tokenizer,
tools=sample.tools,
)
parser = _make_parser(parser_info.parser_cls, tokenizer, sample)
results = replay_streaming(
parser,
@@ -562,10 +568,7 @@ class TestDropTokenNonStreaming:
for sample in parser_info.samples:
injected = _inject_drop_tokens(sample)
tokenizer = make_mock_tokenizer(injected)
parser = parser_info.parser_cls(
tokenizer,
tools=sample.tools,
)
parser = _make_parser(parser_info.parser_cls, tokenizer, sample)
request = _test_request(tools=sample.tools)
output = parse_non_streaming(parser, injected, request)
+145 -3
View File
@@ -29,6 +29,8 @@ from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionToolsParam,
)
from vllm.parser.engine.registered_adapters import (
DeepSeekV4Parser,
DeepSeekV32Parser,
Gemma4Parser,
Glm47MoeParser,
KimiK2Parser,
@@ -644,6 +646,144 @@ def _build_seed_oss(scenario: Scenario, validate: bool = True) -> Sample:
return sample
# ── DeepSeek V4 (DSML tool format) ──────────────────────────────────
_DSML = "DSML"
_DSV4_VOCAB: dict[str, int] = {
"<think>": 128821,
"</think>": 128822,
f"<{_DSML}tool_calls>": 128823,
f"</{_DSML}tool_calls>": 128824,
}
def _dsv4_param_text(key: str, value: Any) -> str:
is_string = isinstance(value, str)
if is_string:
val_str = value
elif isinstance(value, bool):
val_str = "true" if value else "false"
elif isinstance(value, (int, float)):
val_str = str(value)
else:
val_str = json.dumps(value, ensure_ascii=False)
string_attr = "true" if is_string else "false"
return (
f'<{_DSML}parameter name="{key}" string="{string_attr}">'
f"{val_str}</{_DSML}parameter>\n"
)
def _dsv4_tool_text(tc: ToolCallSpec) -> str:
parts = [f'<{_DSML}invoke name="{tc.name}">\n']
for key, value in tc.arguments.items():
parts.append(_dsv4_param_text(key, value))
parts.append(f"</{_DSML}invoke>\n")
return "".join(parts)
def _dsml_tool_segs(
scenario: Scenario,
tag: str,
) -> list[tuple[str, bool]]:
if not scenario.tool_calls:
return []
parts = ["\n"]
for tc in scenario.tool_calls:
parts.append(_dsv4_tool_text(tc))
return [
(f"<{_DSML}{tag}>", True),
("".join(parts), False),
(f"</{_DSML}{tag}>", True),
]
def _dsv4_segments(scenario: Scenario, thinking: bool) -> list[tuple[str, bool]]:
segs: list[tuple[str, bool]] = []
if thinking:
if scenario.reasoning is not None:
segs.append((scenario.reasoning, False))
if scenario.content is not None or scenario.tool_calls:
segs.append(("</think>", True))
else:
if scenario.reasoning is not None:
segs.append(("<think>", True))
segs.append((scenario.reasoning, False))
segs.append(("</think>", True))
if scenario.content is not None:
segs.append((scenario.content, False))
segs.extend(_dsml_tool_segs(scenario, "tool_calls"))
return segs
def _build_deepseek_v4(scenario: Scenario, validate: bool = True) -> Sample:
thinking = scenario.reasoning is not None
chat_kwargs = {"thinking": True} if thinking else None
if thinking:
expected_reasoning: str | None = scenario.reasoning or ""
else:
expected_reasoning = None
sample = _make_sample(
sample_id=f"deepseek_v4-{scenario.id}",
description=scenario.description,
vocab=_DSV4_VOCAB,
segments=_dsv4_segments(scenario, thinking),
expected_reasoning=expected_reasoning,
expected_content=_qwen3_expected_content(scenario),
expected_tool_calls=_expected_tc(scenario),
tools=_expected_tools(scenario),
chat_template_kwargs=chat_kwargs,
)
if validate:
kwargs = {}
if chat_kwargs:
kwargs["chat_template_kwargs"] = chat_kwargs
_validate_sample(sample, DeepSeekV4Parser, **kwargs)
return sample
# ── DeepSeek V3.2 (DSML tool format, no reasoning) ──────────────────
_DSV32_VOCAB: dict[str, int] = {
f"<{_DSML}function_calls>": 128830,
f"</{_DSML}function_calls>": 128831,
}
def _dsv32_segments(scenario: Scenario) -> list[tuple[str, bool]]:
segs: list[tuple[str, bool]] = []
if scenario.content is not None:
segs.append((scenario.content, False))
segs.extend(_dsml_tool_segs(scenario, "function_calls"))
return segs
def _build_deepseek_v32(scenario: Scenario, validate: bool = True) -> Sample | None:
if scenario.reasoning is not None:
return None
sample = _make_sample(
sample_id=f"deepseek_v32-{scenario.id}",
description=scenario.description,
vocab=_DSV32_VOCAB,
segments=_dsv32_segments(scenario),
expected_reasoning=None,
expected_content=_qwen3_expected_content(scenario),
expected_tool_calls=_expected_tc(scenario),
tools=_expected_tools(scenario),
)
if validate:
_validate_sample(sample, DeepSeekV32Parser)
return sample
# ── GLM-4.7 MoE (XML tool format, starts in REASONING) ──────────────
_GLM47_MOE_VOCAB: dict[str, int] = {
@@ -811,13 +951,15 @@ _KIMI_K2_SCENARIOS = [
# ── Registry and public API ──────────────────────────────────────────
_BUILDERS: dict[str, Any] = {
"qwen3": _build_qwen3,
"deepseek_v32": _build_deepseek_v32,
"deepseek_v4": _build_deepseek_v4,
"gemma4": _build_gemma4,
"minimax_m2": _build_minimax_m2,
"nemotron_v3": _build_nemotron_v3,
"seed_oss": _build_seed_oss,
"glm47_moe": _build_glm47_moe,
"kimi_k2": _build_kimi_k2,
"qwen3": _build_qwen3,
}
@@ -826,10 +968,10 @@ def build_samples(model: str) -> tuple[Sample, ...]:
"""Build all scenario samples for a model, self-validated."""
builder = _BUILDERS[model]
scenarios = _KIMI_K2_SCENARIOS if model == "kimi_k2" else SCENARIOS
return tuple(builder(s) for s in scenarios)
return tuple(s for s in (builder(sc) for sc in scenarios) if s is not None)
def build_sample(model: str, scenario: Scenario) -> Sample:
def build_sample(model: str, scenario: Scenario) -> Sample | None:
"""Build a single sample for one model + scenario."""
return _BUILDERS[model](scenario)
+70 -34
View File
@@ -7,7 +7,6 @@ from collections.abc import Sequence
import pytest
from openai_harmony import (
Conversation,
HarmonyError,
Message,
RenderConversationConfig,
Role,
@@ -51,6 +50,15 @@ def chat_request():
)
@pytest.fixture
def malformed_msgs_str() -> list[str]:
return [
"<|channel|>analysis<|message|>thinking<|end|>",
"<|start|>assistant<|channel|>commentary<|message|>thinking<|end|>",
'<|start|>assistant<|channel|>final {"answer": "hi"}<|return|>',
]
def encode_output(harmony_str: str) -> list[int]:
return get_encoding().encode(harmony_str, allowed_special="all")
@@ -131,13 +139,22 @@ def tool_call_entries(delta_message) -> list[tuple[int, str | None, str | None]]
]
def assert_parser_is_reset(harmony_parser: HarmonyParser):
assert harmony_parser._parser is None
assert harmony_parser._num_processed_messages == 0
assert harmony_parser._current_message_tokens == []
class TestFlush:
def test_flush(self, harmony_parser):
harmony_parser.process_chunk(
encode_output("<|channel|>analysis<|message|>Think")
)
flushed = harmony_parser.flush()
flushed_segments = harmony_parser.flush()
assert flushed_segments is not None
assert len(flushed_segments) == 1
flushed = flushed_segments[0]
assert flushed is not None
assert flushed.channel == "analysis"
@@ -145,15 +162,27 @@ class TestFlush:
assert flushed.delta == ""
assert flushed.completed_message is not None
assert get_text(flushed.completed_message) == "Think"
assert harmony_parser._parser is None
assert_parser_is_reset(harmony_parser)
def test_flush_raises_and_resets_on_non_terminal_eos(self, harmony_parser):
harmony_parser.process_chunk(encode_output("<|channel|>analysis"))
def test_flush_recovers_invalid_output(self, harmony_parser, malformed_msgs_str):
for msg_str in malformed_msgs_str[:-1]:
chunk = harmony_parser.process_chunk(encode_output(msg_str))
assert "".join(segment.delta for segment in chunk.segments) == "thinking"
with pytest.raises(HarmonyError):
harmony_parser.flush()
last_msg_str = malformed_msgs_str[-1]
harmony_parser.process_chunk(encode_output(last_msg_str))
flushed_segments = harmony_parser.flush()
assert len(flushed_segments) == 2
delta_segment = flushed_segments[0]
message_segment = flushed_segments[1]
assert harmony_parser._parser is None
assert delta_segment.channel == "final"
assert delta_segment.recipient is None
assert delta_segment.delta == last_msg_str
assert message_segment.channel == "final"
assert message_segment.recipient is None
assert get_text(message_segment.completed_message) == last_msg_str
assert_parser_is_reset(harmony_parser)
class TestParse:
@@ -364,7 +393,7 @@ class TestParse:
assert reasoning is None
assert content == "I'm in the middle of answering"
assert tool_calls is None
assert harmony_parser._parser is None
assert_parser_is_reset(harmony_parser)
def test_interrupted_reasoning_first_message(self, harmony_parser, chat_request):
reasoning, content, tool_calls = harmony_parser.parse(
@@ -378,7 +407,7 @@ class TestParse:
assert reasoning == "I'm in the middle of thinking"
assert content is None
assert tool_calls is None
assert harmony_parser._parser is None
assert_parser_is_reset(harmony_parser)
def test_truncated_output(self, harmony_parser, chat_request):
reasoning, content, tool_calls = harmony_parser.parse(
@@ -394,24 +423,23 @@ class TestParse:
assert reasoning == "I'm thinking."
assert content == "I'm in the middle of answering"
assert tool_calls is None
assert harmony_parser._parser is None
assert_parser_is_reset(harmony_parser)
def test_malformed_final_recovers_raw_content(self, harmony_parser, chat_request):
raw_output = (
"<|channel|>analysis<|message|>thinking<|end|>"
'<|start|>assistant<|channel|>final {"answer": "hi"}<|return|>'
)
def test_malformed_msgs_recovers_raw_content(
self, harmony_parser, chat_request, malformed_msgs_str
):
combined_output = "".join(malformed_msgs_str)
reasoning, content, tool_calls = harmony_parser.parse(
raw_output,
"",
chat_request,
model_output_token_ids=encode_output(raw_output),
model_output_token_ids=encode_output(combined_output),
)
assert content == raw_output
assert reasoning is None
assert reasoning == "thinking"
assert content == "thinking\n" + malformed_msgs_str[-1]
assert tool_calls is None
assert harmony_parser._parser is None
assert_parser_is_reset(harmony_parser)
@pytest.mark.parametrize(
("harmony_str", "expected_content"),
@@ -489,7 +517,7 @@ class TestParseDelta:
assert second_delta is not None
assert second_delta.content == "Answer"
assert second_delta.reasoning is None
assert parser._parser is None
assert_parser_is_reset(parser)
def test_multi_token(self, gpt_oss_tokenizer, chat_request):
parser = HarmonyParser(gpt_oss_tokenizer)
@@ -506,25 +534,33 @@ class TestParseDelta:
assert delta.reasoning is None
assert not delta.tool_calls
def test_malformed_final_recovers_raw_content(
self, gpt_oss_tokenizer, chat_request
def test_malformed_msgs_recovers_raw_content(
self, gpt_oss_tokenizer, chat_request, malformed_msgs_str
):
parser = HarmonyParser(gpt_oss_tokenizer)
delta = parser.parse_delta(
delta_text='final {"answer": "hi"}',
delta_token_ids=encode_output(
'<|channel|>final {"answer": "hi"}<|return|>'
),
for msg_str in malformed_msgs_str[:-1]:
delta = parser.parse_delta(
delta_text="",
delta_token_ids=encode_output(msg_str),
request=chat_request,
finished=False,
)
assert delta.reasoning or delta.content == "thinking"
assert not delta.tool_calls
last_delta = parser.parse_delta(
delta_text="",
delta_token_ids=encode_output(malformed_msgs_str[-1]),
request=chat_request,
finished=True,
)
assert delta is not None
assert delta.content == 'final {"answer": "hi"}'
assert delta.reasoning is None
assert not delta.tool_calls
assert parser._parser is None
assert last_delta is not None
assert last_delta.content == malformed_msgs_str[-1]
assert last_delta.reasoning is None
assert not last_delta.tool_calls
assert_parser_is_reset(parser)
@pytest.mark.parametrize("tool_channel", ["commentary", "analysis"])
def test_tool_call_split_across_deltas(
@@ -11,6 +11,8 @@ from vllm.reasoning.deepseek_r1_reasoning_parser import DeepSeekR1ReasoningParse
from vllm.reasoning.deepseek_v3_reasoning_parser import DeepSeekV3ReasoningParser
from vllm.reasoning.identity_reasoning_parser import IdentityReasoningParser
pytestmark = pytest.mark.skip_global_cleanup
REASONING_MODEL_NAME = "deepseek-ai/DeepSeek-V3.1"
@@ -35,9 +37,13 @@ def test_parser_selection(tokenizer, thinking, expected_parser_type):
def test_deepseek_v4_reasoning_parser_alias():
from vllm.reasoning.deepseek_v4_engine_reasoning_parser import (
DeepSeekV4ParserReasoningAdapter,
)
parser_cls = ReasoningParserManager.get_reasoning_parser("deepseek_v4")
assert parser_cls is DeepSeekV3ReasoningParser
assert parser_cls is DeepSeekV4ParserReasoningAdapter
def test_identity_reasoning_parser_basic(tokenizer):
@@ -40,7 +40,7 @@ from vllm.renderers.hf import (
# Qwen2TokenizerFast (SentencePiece BPE variant)
# BertTokenizerFast (WordPiece)
TOKENIZER_IDS: Final[list[str]] = [
"gpt2",
"openai-community/gpt2",
"Qwen/Qwen2.5-1.5B-Instruct",
"bert-base-uncased",
]
+31
View File
@@ -34,6 +34,7 @@ from vllm.config.vllm import (
OptimizationLevel,
)
from vllm.platforms import current_platform
from vllm.v1.attention.backend import AttentionCGSupport
DEVICE_TYPE = current_platform.device_type
@@ -67,6 +68,36 @@ def test_v2_model_runner_env_tri_state(monkeypatch, env_value, expected):
assert envs.VLLM_USE_V2_MODEL_RUNNER is expected
@pytest.mark.parametrize(
("use_v2_model_runner", "expected_capture_sizes"),
[
(False, [4, 8, 12, 16]),
(True, list(range(1, 17))),
],
)
def test_resolve_cudagraph_mode_adjusts_spec_decode_sizes_only_for_v1(
use_v2_model_runner,
expected_capture_sizes,
):
compilation_config = CompilationConfig(
cudagraph_mode=CUDAGraphMode.FULL_AND_PIECEWISE,
cudagraph_capture_sizes=list(range(1, 17)),
)
compilation_config.max_cudagraph_capture_size = 16
compilation_config.post_init_cudagraph_sizes()
cudagraph_mode = compilation_config.resolve_cudagraph_mode_and_sizes(
AttentionCGSupport.ALWAYS,
"FakeAttentionBackend",
uniform_decode_query_len=4,
use_v2_model_runner=use_v2_model_runner,
tensor_parallel_size=1,
)
assert cudagraph_mode == CUDAGraphMode.FULL_AND_PIECEWISE
assert compilation_config.cudagraph_capture_sizes == expected_capture_sizes
@pytest.mark.parametrize(
("model_config", "expected"),
[
+4 -2
View File
@@ -23,7 +23,7 @@ def _assert_tokenizer_like(tokenizer: object):
def test_tokenizer_like_protocol():
tokenizer = get_tokenizer("gpt2", use_fast=True)
tokenizer = get_tokenizer("openai-community/gpt2", use_fast=True)
assert isinstance(tokenizer, PreTrainedTokenizerFast)
_assert_tokenizer_like(tokenizer)
@@ -43,7 +43,9 @@ def test_tokenizer_like_protocol():
_assert_tokenizer_like(tokenizer)
@pytest.mark.parametrize("tokenizer_name", ["facebook/opt-125m", "gpt2"])
@pytest.mark.parametrize(
"tokenizer_name", ["facebook/opt-125m", "openai-community/gpt2"]
)
def test_tokenizer_revision(tokenizer_name: str):
# Assume that "main" branch always exists
tokenizer = get_tokenizer(tokenizer_name, revision="main")
+1 -1
View File
@@ -33,7 +33,7 @@ TRUTH = [
TOKENIZERS = [
"facebook/opt-125m",
"gpt2",
"openai-community/gpt2",
"bigcode/tiny_starcoder_py",
"EleutherAI/gpt-j-6b",
"EleutherAI/pythia-70m",
+2 -2
View File
@@ -14,7 +14,7 @@ from vllm.tokenizers.hf import (
)
@pytest.mark.parametrize("model_id", ["gpt2", "zai-org/chatglm3-6b"])
@pytest.mark.parametrize("model_id", ["openai-community/gpt2", "zai-org/chatglm3-6b"])
def test_cached_tokenizer(model_id: str):
reference_tokenizer = AutoTokenizer.from_pretrained(
model_id, trust_remote_code=True
@@ -47,7 +47,7 @@ def _check_consistency(target: TokenizerLike, expected: TokenizerLike):
assert target.encode("prompt") == expected.encode("prompt")
@pytest.mark.parametrize("model_id", ["gpt2"])
@pytest.mark.parametrize("model_id", ["openai-community/gpt2"])
def test_thread_pool_tokenizer_pickle(model_id: str):
"""Regression test for issue #45433: the thread-pool tokenizer wrapper
reconstructs through maybe_make_thread_pool on unpickling, which used to
+1 -1
View File
@@ -9,4 +9,4 @@ from vllm.tokenizers import TokenizerLike
@pytest.fixture(scope="module")
def default_tokenizer() -> TokenizerLike:
return AutoTokenizer.from_pretrained("gpt2")
return AutoTokenizer.from_pretrained("openai-community/gpt2")
@@ -1,7 +1,7 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Unit tests for DeepSeekV32ToolParser.
"""Unit tests for DeepSeekV32EngineToolParser.
These tests use a minimal mock tokenizer so no real model weights are required.
"""
@@ -17,7 +17,11 @@ from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionToolsParam,
FunctionDefinition,
)
from vllm.tool_parsers.deepseekv32_tool_parser import DeepSeekV32ToolParser
from vllm.tool_parsers.deepseekv32_engine_tool_parser import (
DeepSeekV32EngineToolParser,
)
pytestmark = pytest.mark.skip_global_cleanup
# ---------------------------------------------------------------------------
# Helpers
@@ -30,8 +34,8 @@ MOCK_TOKENIZER.get_vocab.return_value = {}
MOCK_TOKENIZER.tokenize.return_value = []
def make_parser(tools=None) -> DeepSeekV32ToolParser:
return DeepSeekV32ToolParser(MOCK_TOKENIZER, tools=tools)
def make_parser(tools=None) -> DeepSeekV32EngineToolParser:
return DeepSeekV32EngineToolParser(MOCK_TOKENIZER, tools=tools)
def make_tool_param(name: str, params: dict) -> MagicMock:
@@ -167,9 +171,9 @@ class TestExtractToolCalls:
assert isinstance(args["enabled"], bool)
assert isinstance(args["count"], int)
def test_string_attr_true_preserves_literal_despite_schema(self):
"""string="true" must keep the value as a string even
if the schema says integer."""
def test_string_attr_true_coerced_by_schema(self):
"""string="true" delivers a string, but the engine's schema-aware
type fixer coerces it to the schema type (integer)."""
tool = ChatCompletionToolsParam(
function=FunctionDefinition(
name="score",
@@ -192,8 +196,8 @@ class TestExtractToolCalls:
result = parser.extract_tool_calls(model_output, None)
assert result.tools_called
args = json.loads(result.tool_calls[0].function.arguments)
assert args == {"value": "42"}
assert isinstance(args["value"], str)
assert args == {"value": 42}
assert isinstance(args["value"], int)
def test_string_attr_false_allows_schema_conversion(self):
"""string="false" allows the parser to convert via the tool schema."""
@@ -222,7 +226,6 @@ class TestExtractToolCalls:
assert args == {"value": 42}
assert isinstance(args["value"], int)
@pytest.mark.skip_global_cleanup
def test_composed_schema_converts_object_and_array_params(self):
"""Composed JSON Schema types must still drive DSML type coercion."""
tool = ChatCompletionToolsParam(
@@ -282,8 +285,9 @@ class TestExtractToolCalls:
assert isinstance(args["wait"], dict)
assert isinstance(args["patches"], list)
@pytest.mark.skip_global_cleanup
def test_string_attr_true_preserves_literal_for_composed_schema(self):
def test_string_attr_true_coerced_by_composed_schema(self):
"""string="true" delivers a JSON string, but the engine's schema-aware
type fixer coerces it to the composed schema type (object)."""
tool = ChatCompletionToolsParam(
function=FunctionDefinition(
name="set_timer",
@@ -313,7 +317,7 @@ class TestExtractToolCalls:
result = parser.extract_tool_calls(model_output, None)
assert result.tools_called
args = json.loads(result.tool_calls[0].function.arguments)
assert args == {"wait": '{"type":"for","minutes":2880}'}
assert args == {"wait": {"type": "for", "minutes": 2880}}
def test_arguments_wrapper_repaired(self):
"""A single 'arguments' wrapper parameter must be unwrapped when it
@@ -486,8 +490,9 @@ class TestExtractToolCalls:
args = json.loads(result.tool_calls[0].function.arguments)
assert args["value"] is None
def test_null_not_coerced_without_null_in_schema(self):
"""Literal 'null' must stay as a string when the schema is just 'string'."""
def test_null_coerced_back_to_string_by_schema(self):
"""string="false" with 'null' is json-parsed to None, but the
engine's schema fixer coerces it back to "null" for string schemas."""
tool = ChatCompletionToolsParam(
function=FunctionDefinition(
name="echo",
@@ -512,8 +517,8 @@ class TestExtractToolCalls:
assert args["text"] == "null"
assert isinstance(args["text"], str)
def test_no_schema_keeps_strings(self):
"""Without a tool schema, all string='false' params default to string."""
def test_no_schema_parses_json(self):
"""Without a tool schema, string='false' params are JSON-parsed."""
parser = make_parser(tools=None)
model_output = (
f"{FC_START}\n"
@@ -526,8 +531,8 @@ class TestExtractToolCalls:
result = parser.extract_tool_calls(model_output, None)
assert result.tools_called
args = json.loads(result.tool_calls[0].function.arguments)
assert args["count"] == "42"
assert args["flag"] == "true"
assert args["count"] == 42
assert args["flag"] is True
# ---------------------------------------------------------------------------
@@ -648,8 +653,9 @@ class TestExtractToolCallsStreaming:
args_str = self._reconstruct_args(deltas)
assert json.loads(args_str) == {"x": 3, "y": 4}
def test_string_attr_true_preserves_literal_in_streaming(self):
"""Streaming: string='true' must keep the value literal despite schema."""
def test_string_attr_true_coerced_by_schema_streaming(self):
"""Streaming: string='true' delivers a string but the engine's
schema fixer coerces it to the schema type (integer)."""
tool = ChatCompletionToolsParam(
function=FunctionDefinition(
name="score",
@@ -672,10 +678,9 @@ class TestExtractToolCallsStreaming:
deltas = self._stream(parser, full_text)
args_str = self._reconstruct_args(deltas)
args = json.loads(args_str)
assert args == {"value": "42"}
assert isinstance(args["value"], str)
assert args == {"value": 42}
assert isinstance(args["value"], int)
@pytest.mark.skip_global_cleanup
def test_composed_schema_conversion_in_streaming(self):
tool = ChatCompletionToolsParam(
function=FunctionDefinition(
@@ -821,13 +826,13 @@ class TestExtractToolCallsStreaming:
assert json.loads(self._reconstruct_args(deltas, tool_index=0)) == {"p": "v1"}
assert json.loads(self._reconstruct_args(deltas, tool_index=1)) == {"q": "v2"}
def test_state_reset_on_new_stream(self, parser):
"""A second stream (previous_text == '') must reset state cleanly."""
def test_state_reset_on_new_stream(self):
"""A fresh parser instance must produce identical results."""
full_text = build_tool_call("fn", {"k": "v"})
# First stream
self._stream(parser, full_text)
# Second stream - should produce identical results
deltas2 = self._stream(parser, full_text)
self._stream(make_parser(), full_text)
# Second stream with fresh parser
deltas2 = self._stream(make_parser(), full_text)
assert json.loads(self._reconstruct_args(deltas2)) == {"k": "v"}
def test_empty_arguments_streaming(self, parser):
@@ -860,26 +865,6 @@ class TestExtractToolCallsStreaming:
assert len(ids) == 2
assert ids[0] != ids[1]
def test_eos_after_tool_calls(self, parser):
"""EOS token (empty delta_text, non-empty delta_token_ids) returns
a non-None DeltaMessage so the serving framework can finalize."""
full_text = build_tool_call("fn", {"k": "v"})
# Drive through the full text first
deltas = self._stream(parser, full_text)
assert any(d.tool_calls for d in deltas)
# Now simulate EOS: empty delta_text, but token ids present
prev = full_text
result = parser.extract_tool_calls_streaming(
previous_text=prev,
current_text=prev,
delta_text="",
previous_token_ids=[],
current_token_ids=[],
delta_token_ids=[2], # EOS token id
request=make_request(),
)
assert result is not None
def test_streaming_matches_non_streaming(self, parser):
"""Streaming and non-streaming must produce the same result."""
full_text = build_tool_call(
@@ -968,7 +953,6 @@ class TestExtractToolCallsStreaming:
def test_emits_arguments_before_invoke_completes(self, parser):
"""Argument deltas should stream before the invoke block closes."""
# Stream only a partial invoke (no closing tag)
partial_text = (
f"{FC_START}\n"
f'{INV_START}fn">\n'
@@ -981,7 +965,9 @@ class TestExtractToolCallsStreaming:
for tc in delta.tool_calls or []
if tc.function and tc.function.arguments is not None
]
assert "".join(arg_chunks) == '{"k":"val"'
combined = "".join(arg_chunks)
assert combined # some partial args emitted
assert combined.startswith('{"k"')
def test_no_marker_leak_chunked(self, parser):
"""Chunked streaming must NOT leak DSML start-marker fragments
@@ -1,7 +1,7 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Unit tests for DeepSeekV4ToolParser."""
"""Unit tests for DeepSeekV4EngineToolParser."""
import json
from unittest.mock import MagicMock
@@ -17,7 +17,11 @@ from vllm.entrypoints.openai.chat_completion.protocol import (
FunctionDefinition,
)
from vllm.tool_parsers import ToolParserManager
from vllm.tool_parsers.deepseekv4_tool_parser import DeepSeekV4ToolParser
from vllm.tool_parsers.deepseekv4_engine_tool_parser import (
DeepSeekV4EngineToolParser,
)
pytestmark = pytest.mark.skip_global_cleanup
MOCK_TOKENIZER = MagicMock()
MOCK_TOKENIZER.get_vocab.return_value = {}
@@ -67,8 +71,8 @@ def sample_tools() -> list[ChatCompletionToolsParam]:
]
def make_parser(tools=None) -> DeepSeekV4ToolParser:
return DeepSeekV4ToolParser(MOCK_TOKENIZER, tools=tools)
def make_parser(tools=None) -> DeepSeekV4EngineToolParser:
return DeepSeekV4EngineToolParser(MOCK_TOKENIZER, tools=tools)
def make_request(tools=None) -> MagicMock:
@@ -84,7 +88,7 @@ def build_tool_call(func_name: str, params: dict[str, str]) -> str:
return f'{TC_START}\n{INV_START}{func_name}">\n{param_strs}{INV_END}\n{TC_END}'
def stream(parser: DeepSeekV4ToolParser, full_text: str, chunk_size: int = 7):
def stream(parser: DeepSeekV4EngineToolParser, full_text: str, chunk_size: int = 7):
deltas = []
previous_text = ""
for start in range(0, len(full_text), chunk_size):
@@ -120,7 +124,9 @@ def reconstruct_args(deltas, tool_index: int = 0) -> str:
def test_registered():
assert ToolParserManager.get_tool_parser("deepseek_v4") is DeepSeekV4ToolParser
assert (
ToolParserManager.get_tool_parser("deepseek_v4") is DeepSeekV4EngineToolParser
)
def test_extract_tool_calls():
@@ -285,7 +291,7 @@ def test_extract_tool_calls_arguments_wrapper():
},
)
parser = DeepSeekV4ToolParser(mock_tokenizer, tools=[tool])
parser = DeepSeekV4EngineToolParser(mock_tokenizer, tools=[tool])
request = MagicMock()
request.tools = [tool]
@@ -303,7 +309,64 @@ def test_extract_tool_calls_arguments_wrapper():
assert args == {"location": "Beijing"}
@pytest.mark.skip_global_cleanup
_ANGLE_BRACKET_TOOL = ChatCompletionToolsParam(
function=FunctionDefinition(
name="run_command",
parameters={
"type": "object",
"properties": {
"command": {"type": "string"},
},
},
),
)
@pytest.mark.parametrize(
"tools",
[[_ANGLE_BRACKET_TOOL], None],
ids=["with_tools", "without_tools"],
)
def test_no_dsml_closing_tag_leak_in_streamed_args(tools):
"""Streaming must not leak </DSMLparameter> into argument values.
When a parameter value contains '>' (e.g. shell redirects like
'2>&1'), certain chunk boundaries cause the incremental lexer to
emit the closing delimiter text as part of the content token. The
partial regex then captures it as part of the value, violating the
prefix invariant and corrupting the streamed JSON.
"""
full_text = build_tool_call("run_command", {"command": "git --version 2>&1"})
expected = {"command": "git --version 2>&1"}
for chunk_size in range(1, len(full_text) + 1):
parser = make_parser(tools=tools)
deltas = stream(parser, full_text, chunk_size=chunk_size)
args_str = reconstruct_args(deltas)
assert args_str, f"No args emitted at chunk_size={chunk_size}"
assert "DSML" not in args_str, (
f"DSML marker leaked into args at chunk_size={chunk_size}: {args_str!r}"
)
parsed = json.loads(args_str)
assert parsed == expected, (
f"Args mismatch at chunk_size={chunk_size}: "
f"got {parsed!r}, expected {expected!r}"
)
def test_non_streaming_extract_with_angle_brackets():
"""Non-streaming extraction must correctly handle '>' in values."""
parser = make_parser()
full_text = build_tool_call("run_command", {"command": "git --version 2>&1"})
result = parser.extract_tool_calls(full_text, make_request())
assert result.tools_called
assert len(result.tool_calls) == 1
args = json.loads(result.tool_calls[0].function.arguments)
assert args == {"command": "git --version 2>&1"}
assert "DSML" not in result.tool_calls[0].function.arguments
def test_composed_schema_converts_object_and_array_params():
tool = ChatCompletionToolsParam(
type="function",
@@ -19,7 +19,7 @@ from vllm.tool_parsers import ToolParser, ToolParserManager
def default_tokenizer() -> TokenizerLike:
"""Override module-scoped default_tokenizer because gigachat tests
mutate the tokenizer via ``add_tokens``."""
return AutoTokenizer.from_pretrained("gpt2")
return AutoTokenizer.from_pretrained("openai-community/gpt2")
MSG_SEP_TOKEN = "<|message_sep|>\n\n"
@@ -16,9 +16,11 @@ from vllm.entrypoints.openai.chat_completion.protocol import (
from vllm.parser.abstract_parser import DelegatingParser
from vllm.tool_parsers.abstract_tool_parser import ToolParser
from vllm.tool_parsers.deepseekv3_tool_parser import DeepSeekV3ToolParser
from vllm.tool_parsers.deepseekv4_tool_parser import DeepSeekV4ToolParser
from vllm.tool_parsers.deepseekv4_engine_tool_parser import DeepSeekV4EngineToolParser
from vllm.tool_parsers.deepseekv31_tool_parser import DeepSeekV31ToolParser
from vllm.tool_parsers.deepseekv32_tool_parser import DeepSeekV32ToolParser
from vllm.tool_parsers.deepseekv32_engine_tool_parser import (
DeepSeekV32EngineToolParser,
)
from vllm.tool_parsers.glm47_moe_tool_parser import Glm47MoeModelToolParser
from vllm.tool_parsers.hermes_tool_parser import Hermes2ProToolParser
from vllm.tool_parsers.kimi_k2_tool_parser import KimiK2ToolParser
@@ -185,8 +187,8 @@ def test_get_model_structural_tag_supports_named_tool_choice(
[
(DeepSeekV3ToolParser, "deepseek_r1"),
(DeepSeekV31ToolParser, "deepseek_v3_1"),
(DeepSeekV32ToolParser, "deepseek_v3_2"),
(DeepSeekV4ToolParser, "deepseek_v4"),
(DeepSeekV32EngineToolParser, "deepseek_v3_2"),
(DeepSeekV4EngineToolParser, "deepseek_v4"),
(Glm47MoeModelToolParser, "glm_4_7"),
(Hermes2ProToolParser, "hermes"),
(KimiK2ToolParser, "kimi"),
+3
View File
@@ -272,6 +272,9 @@ def test_abort_request_when_structured_output_fsm_cannot_advance():
scheduler.connector = None
scheduler.structured_output_manager = Mock()
scheduler.structured_output_manager.should_advance.return_value = True
scheduler.structured_output_manager.trim_reasoning_for_advance.side_effect = (
lambda request, new_token_ids: new_token_ids
)
scheduler.requests = {request.request_id: request}
scheduler.running = [request]
scheduler.waiting = Mock()
+3
View File
@@ -2911,6 +2911,9 @@ def test_abort_request_when_structured_output_fsm_cannot_advance():
scheduler.connector = None
scheduler.structured_output_manager = Mock()
scheduler.structured_output_manager.should_advance.return_value = True
scheduler.structured_output_manager.trim_reasoning_for_advance.side_effect = (
lambda request, new_token_ids: new_token_ids
)
scheduler.requests = {request.request_id: request}
scheduler.running = [request]
scheduler.waiting = Mock()
+328
View File
@@ -0,0 +1,328 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
import torch
from vllm.config import (
CompilationConfig,
CUDAGraphMode,
ParallelConfig,
SchedulerConfig,
VllmConfig,
)
from vllm.v1.worker.gpu import cudagraph_utils as gpu_cudagraph_utils
pytestmark = pytest.mark.cpu_test
def _create_vllm_config_for_dsd(
max_num_seqs: int,
max_spec_tokens: int,
*,
cudagraph_mode: str = "FULL_AND_PIECEWISE",
use_dynamic_sd: bool = True,
num_spec_per_batch_size: list[tuple[int, int, int]] | None = None,
) -> MagicMock:
"""Create a minimal config that exercises DSD cudagraph dispatch.
The test uses an exact capture-size grid so that every valid uniform decode
shape has a directly matching FULL graph candidate.
``num_spec_per_batch_size`` lets a test supply an explicit DSD schedule of
``(range_start, range_end, num_speculative_tokens)`` tuples. When omitted,
a schedule covering every query length in ``[1, max_decode_query_len]`` is
generated.
"""
max_decode_query_len = max_spec_tokens + 1
max_capture_tokens = max_num_seqs * max_decode_query_len
compilation_config = CompilationConfig(
cudagraph_mode=cudagraph_mode,
cudagraph_capture_sizes=list(range(1, max_capture_tokens + 1)),
)
compilation_config.max_cudagraph_capture_size = max_capture_tokens
compilation_config.post_init_cudagraph_sizes()
vllm_config = MagicMock(spec=VllmConfig)
vllm_config.compilation_config = compilation_config
vllm_config.scheduler_config = SchedulerConfig.default_factory(
max_num_seqs=max_num_seqs,
)
vllm_config.parallel_config = ParallelConfig()
# num_speculative_tokens is the max K (num_speculative_steps). The manager
# recovers num_new_sampled_tokens_per_step as
# decode_query_len - num_speculative_tokens; with decode_query_len =
# max_spec_tokens + 1 this yields the normal per-step bonus of 1.
vllm_config.num_speculative_tokens = max_spec_tokens
speculative_config = MagicMock()
speculative_config.uses_dynamic_speculative_decoding.return_value = use_dynamic_sd
if use_dynamic_sd:
# DSD reads the per-batch-size schedule; a schedule entry with K
# speculative tokens maps to decode query length K + 1. By default
# provide every query length in [1, max_decode_query_len] (i.e. K in
# [0, max_spec_tokens]) so the manager captures a FULL decode graph for
# each uniform shape.
if num_spec_per_batch_size is None:
num_spec_per_batch_size = [
(qlen, qlen, qlen - 1) for qlen in range(1, max_decode_query_len + 1)
]
speculative_config.num_speculative_tokens_per_batch_size = (
num_spec_per_batch_size
)
else:
speculative_config.num_speculative_tokens_per_batch_size = None
vllm_config.speculative_config = speculative_config
return vllm_config
def test_dynamic_sd_full_cudagraph_covers_all_uniform_decode_shapes(monkeypatch):
"""Dynamic SD should create FULL decode candidates for every k in [1, K+1].
This validates the MRv2 CudaGraphManager path directly: once candidate
shapes have been built, dispatch() should pick a FULL graph for every
uniform decode batch shape produced by DSD up to max_num_seqs.
"""
max_num_seqs = 512
max_spec_tokens = 7
max_decode_query_len = max_spec_tokens + 1
# CudaGraphManager consults PP rank helpers during initialization even
# though this test only exercises CPU-side candidate generation.
monkeypatch.setattr(
gpu_cudagraph_utils,
"get_pp_group",
lambda: SimpleNamespace(is_first_rank=True, is_last_rank=True),
)
vllm_config = _create_vllm_config_for_dsd(
max_num_seqs=max_num_seqs,
max_spec_tokens=max_spec_tokens,
)
manager = gpu_cudagraph_utils.CudaGraphManager(
vllm_config=vllm_config,
device=torch.device("cpu"),
cudagraph_mode=CUDAGraphMode.FULL_AND_PIECEWISE,
decode_query_len=max_decode_query_len,
)
# dispatch() only uses the precomputed candidate table after graphs are
# considered captured. The actual graph objects are irrelevant here.
manager._graphs_captured = True
for num_reqs in range(1, max_num_seqs + 1):
for max_query_len in range(1, max_decode_query_len + 1):
# Uniform decode means every request contributes the same number of
# tokens, so the total token count is exactly num_reqs * query_len.
num_tokens = num_reqs * max_query_len
uniform_tok_count = gpu_cudagraph_utils.get_uniform_token_count(
num_reqs,
num_tokens,
max_query_len,
)
# The scheduler should mark every one of these shapes as a uniform
# decode batch, which is what enables FULL decode graph selection.
assert uniform_tok_count == max_query_len
desc = manager.dispatch(
num_reqs=num_reqs,
num_tokens=num_tokens,
uniform_token_count=uniform_tok_count,
num_active_loras=0,
)
# With DSD enabled, MRv2 should have captured a FULL candidate for
# every k in [1, K+1], so dispatch should stay on the FULL path.
assert desc.cg_mode == CUDAGraphMode.FULL
assert desc.uniform_token_count == max_query_len
assert desc.num_tokens == num_tokens
assert desc.num_reqs == num_reqs
assert desc.num_active_loras == 0
def test_dynamic_sd_non_uniform_batch_falls_back_to_piecewise(monkeypatch):
"""DSD should use PIECEWISE when the batch is not a uniform decode batch.
FULL DSD graphs are captured separately for each decode query length k.
When runtime tokens are not uniform, uniform_token_count is None and those
FULL candidates should be skipped in favor of the mixed-batch PIECEWISE
graph under FULL_AND_PIECEWISE mode.
"""
max_spec_tokens = 4
monkeypatch.setattr(
gpu_cudagraph_utils,
"get_pp_group",
lambda: SimpleNamespace(is_first_rank=True, is_last_rank=True),
)
vllm_config = _create_vllm_config_for_dsd(
max_num_seqs=512,
max_spec_tokens=max_spec_tokens,
cudagraph_mode="FULL_AND_PIECEWISE",
use_dynamic_sd=True,
)
manager = gpu_cudagraph_utils.CudaGraphManager(
vllm_config=vllm_config,
device=torch.device("cpu"),
cudagraph_mode=CUDAGraphMode.FULL_AND_PIECEWISE,
decode_query_len=max_spec_tokens + 1,
)
manager._graphs_captured = True
# This shape is intentionally non-uniform: 3 tokens across 2 requests
# cannot correspond to a single per-request query length.
desc = manager.dispatch(
num_reqs=2,
num_tokens=3,
uniform_token_count=None,
num_active_loras=0,
)
assert desc.cg_mode == CUDAGraphMode.PIECEWISE
assert desc.uniform_token_count is None
assert desc.num_reqs is None
assert desc.num_tokens == 3
assert desc.num_active_loras == 0
def test_basic_sd_does_not_capture_shorter_full_decode_shapes(monkeypatch):
"""Without DSD, only the max decode query length should get FULL graphs.
Basic SD captures FULL decode graphs only for decode_query_len = K + 1.
Uniform batches with smaller query lengths should therefore miss the FULL
path entirely when using FULL_AND_PIECEWISE.
"""
max_num_seqs = 512
max_spec_tokens = 7
max_decode_query_len = max_spec_tokens + 1
monkeypatch.setattr(
gpu_cudagraph_utils,
"get_pp_group",
lambda: SimpleNamespace(is_first_rank=True, is_last_rank=True),
)
vllm_config = _create_vllm_config_for_dsd(
max_num_seqs=max_num_seqs,
max_spec_tokens=max_spec_tokens,
cudagraph_mode="FULL_AND_PIECEWISE",
use_dynamic_sd=False,
)
manager = gpu_cudagraph_utils.CudaGraphManager(
vllm_config=vllm_config,
device=torch.device("cpu"),
cudagraph_mode=CUDAGraphMode.FULL_AND_PIECEWISE,
decode_query_len=max_decode_query_len,
)
manager._graphs_captured = True
for num_reqs in range(1, max_num_seqs + 1):
for max_query_len in range(1, max_decode_query_len):
# These are still uniform decode batches, but basic SD should only
# have FULL graphs for query_len == max_decode_query_len.
num_tokens = num_reqs * max_query_len
uniform_tok_count = gpu_cudagraph_utils.get_uniform_token_count(
num_reqs,
num_tokens,
max_query_len,
)
assert uniform_tok_count == max_query_len
desc = manager.dispatch(
num_reqs=num_reqs,
num_tokens=num_tokens,
uniform_token_count=uniform_tok_count,
num_active_loras=0,
)
assert desc.cg_mode == CUDAGraphMode.PIECEWISE
assert desc.uniform_token_count is None
assert desc.num_tokens == num_tokens
assert desc.num_reqs is None
assert desc.num_active_loras == 0
def test_dynamic_sd_only_captures_scheduled_query_lengths(monkeypatch):
"""DSD should only capture FULL graphs for query lengths in the schedule.
With a partial schedule of ``(1, 32, 4)`` and ``(32, 128, 3)``, only the
scheduled speculative-token counts (K = 4 and K = 3) become decode query
lengths (K + 1 = 5 and 4). Uniform batches at those query lengths should get
FULL graphs, while every other query length (e.g. the lower values 1, 2, 3)
must fall back to the mixed-batch PIECEWISE graph.
"""
max_num_seqs = 128
max_spec_tokens = 7
max_decode_query_len = max_spec_tokens + 1
# (range_start, range_end, num_speculative_tokens): K = 4 and K = 3 are
# scheduled, so FULL decode graphs should exist for query lengths K + 1,
# i.e. exactly {5, 4}.
num_spec_per_batch_size = [(1, 32, 4), (32, 128, 3)]
scheduled_query_lens = {entry[2] + 1 for entry in num_spec_per_batch_size}
monkeypatch.setattr(
gpu_cudagraph_utils,
"get_pp_group",
lambda: SimpleNamespace(is_first_rank=True, is_last_rank=True),
)
vllm_config = _create_vllm_config_for_dsd(
max_num_seqs=max_num_seqs,
max_spec_tokens=max_spec_tokens,
cudagraph_mode="FULL_AND_PIECEWISE",
use_dynamic_sd=True,
num_spec_per_batch_size=num_spec_per_batch_size,
)
manager = gpu_cudagraph_utils.CudaGraphManager(
vllm_config=vllm_config,
device=torch.device("cpu"),
cudagraph_mode=CUDAGraphMode.FULL_AND_PIECEWISE,
decode_query_len=max_decode_query_len,
)
manager._graphs_captured = True
for num_reqs in range(1, max_num_seqs + 1):
for max_query_len in range(1, max_decode_query_len + 1):
num_tokens = num_reqs * max_query_len
uniform_tok_count = gpu_cudagraph_utils.get_uniform_token_count(
num_reqs,
num_tokens,
max_query_len,
)
assert uniform_tok_count == max_query_len
desc = manager.dispatch(
num_reqs=num_reqs,
num_tokens=num_tokens,
uniform_token_count=uniform_tok_count,
num_active_loras=0,
)
if max_query_len in scheduled_query_lens:
# Scheduled query lengths get a dedicated FULL decode graph.
assert desc.cg_mode == CUDAGraphMode.FULL
assert desc.uniform_token_count == max_query_len
assert desc.num_tokens == num_tokens
assert desc.num_reqs == num_reqs
else:
# Unscheduled query lengths (including the lower values 1 and 2)
# have no FULL candidate and must fall back to PIECEWISE.
assert desc.cg_mode == CUDAGraphMode.PIECEWISE
assert desc.uniform_token_count is None
assert desc.num_tokens == num_tokens
assert desc.num_reqs is None
assert desc.num_active_loras == 0
@@ -0,0 +1,343 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""grammar_bitmask under spec-decode draft padding (#44006)."""
import pytest
from transformers import AutoTokenizer
from vllm.config import StructuredOutputsConfig, VllmConfig
from vllm.config.model import ModelConfig
from vllm.config.speculative import SpeculativeConfig
from vllm.sampling_params import SamplingParams, StructuredOutputsParams
from vllm.v1.request import Request
from vllm.v1.structured_output import StructuredOutputManager
TOKENIZER = "gpt2"
NUM_SPEC_TOKENS = 4
def _make_manager_and_request(backend: str, prompt_str: str = '{"a": "b"}'):
tokenizer = AutoTokenizer.from_pretrained(TOKENIZER)
prompt = tokenizer.encode(prompt_str)
vllm_config = VllmConfig(
model_config=ModelConfig(tokenizer=TOKENIZER),
structured_outputs_config=StructuredOutputsConfig(backend=backend),
speculative_config=SpeculativeConfig(
model="[ngram]", num_speculative_tokens=NUM_SPEC_TOKENS
),
)
manager = StructuredOutputManager(vllm_config)
sampling_params = SamplingParams(
structured_outputs=StructuredOutputsParams(json='{"type": "object"}'),
)
sampling_params.structured_outputs._backend = backend
sampling_params.update_from_generation_config({}, tokenizer.eos_token_id)
request = Request(
"mtp_req",
prompt_token_ids=prompt,
sampling_params=sampling_params,
pooling_params=None,
)
manager.grammar_init(request)
while not request.structured_output_request._check_grammar_completion():
continue
return tokenizer, manager, request, prompt
@pytest.mark.parametrize("backend", ["xgrammar", "guidance"])
def test_bitmask_with_padded_invalid_drafts(backend):
"""Bitmask handles -1 padded drafts and returns N+1 rows."""
tokenizer, manager, request, prompt = _make_manager_and_request(
backend, prompt_str='{"a"'
)
grammar = request.structured_output_request.grammar
assert grammar.accept_tokens(request.request_id, prompt)
valid_drafts = [tokenizer.encode(":")[0], tokenizer.encode(' "')[0]]
padded = valid_drafts + [-1, -1]
bitmask = manager.grammar_bitmask(
requests={request.request_id: request},
structured_output_request_ids=[request.request_id],
scheduled_spec_decode_tokens={request.request_id: padded},
)
assert bitmask is not None
assert bitmask.shape[0] == len(padded) + 1
assert not grammar.is_terminated()
@pytest.mark.parametrize("backend", ["xgrammar", "guidance"])
def test_bitmask_when_grammar_terminates_mid_window(backend):
"""Drafts following an EOS that terminates the grammar are a no-op."""
tokenizer, manager, request, prompt = _make_manager_and_request(backend)
grammar = request.structured_output_request.grammar
assert grammar.accept_tokens(request.request_id, prompt)
eos = tokenizer.eos_token_id
drafts = [eos] + [tokenizer.encode(" ")[0]] * (NUM_SPEC_TOKENS - 1)
bitmask = manager.grammar_bitmask(
requests={request.request_id: request},
structured_output_request_ids=[request.request_id],
scheduled_spec_decode_tokens={request.request_id: drafts},
)
assert bitmask is not None
assert bitmask.shape[0] == NUM_SPEC_TOKENS + 1
assert not grammar.is_terminated()
@pytest.mark.parametrize("backend", ["xgrammar", "guidance"])
def test_bitmask_idempotent_across_calls(backend):
"""Repeated calls with the same input return the same bitmask."""
tokenizer, manager, request, prompt = _make_manager_and_request(
backend, prompt_str='{"a"'
)
grammar = request.structured_output_request.grammar
assert grammar.accept_tokens(request.request_id, prompt)
drafts = [tokenizer.encode(":")[0], -1, -1, -1]
first = manager.grammar_bitmask(
requests={request.request_id: request},
structured_output_request_ids=[request.request_id],
scheduled_spec_decode_tokens={request.request_id: drafts},
)
second = manager.grammar_bitmask(
requests={request.request_id: request},
structured_output_request_ids=[request.request_id],
scheduled_spec_decode_tokens={request.request_id: drafts},
)
assert first is not None and second is not None
assert (first == second).all()
assert not grammar.is_terminated()
@pytest.mark.parametrize("backend", ["xgrammar", "guidance"])
def test_bonus_position_constrained_after_invalid_drafts(backend):
"""Regression for #44006: bonus row stays constrained after -1 padding."""
tokenizer, manager, request, prompt = _make_manager_and_request(
backend, prompt_str='{"a"'
)
grammar = request.structured_output_request.grammar
assert grammar.accept_tokens(request.request_id, prompt)
valid = tokenizer.encode(":")[0]
drafts = [valid, -1, -1, -1]
bitmask = manager.grammar_bitmask(
requests={request.request_id: request},
structured_output_request_ids=[request.request_id],
scheduled_spec_decode_tokens={request.request_id: drafts},
)
assert bitmask is not None
assert bitmask.shape[0] == len(drafts) + 1
assert not (bitmask[-1] == -1).all()
assert not grammar.is_terminated()
@pytest.mark.parametrize("backend", ["xgrammar", "guidance"])
def test_bitmask_constrained_when_reasoning_ends_midwindow(backend):
"""Drafts after a mid-window reasoning-end marker stay constrained."""
tokenizer, manager, request, prompt = _make_manager_and_request(backend)
grammar = request.structured_output_request.grammar
assert grammar.accept_tokens(request.request_id, prompt)
marker = tokenizer.encode("\n")[0]
class StubReasoner:
def __init__(self, *_, **__):
self.end_token_id = marker
def is_reasoning_end(self, input_ids):
return marker in list(input_ids)
def is_reasoning_end_streaming(self, input_ids, delta_ids):
return marker in list(delta_ids)
manager.reasoner_cls = StubReasoner
request.structured_output_request.reasoner = StubReasoner()
request.structured_output_request.reasoning_ended = False
pre = tokenizer.encode(" ")[0]
post = tokenizer.encode(",")[0]
drafts = [pre, marker, post]
bitmask = manager.grammar_bitmask(
requests={request.request_id: request},
structured_output_request_ids=[request.request_id],
scheduled_spec_decode_tokens={request.request_id: drafts},
)
assert bitmask is not None
assert bitmask.shape[0] == len(drafts) + 1
assert (bitmask[0] == -1).all()
assert (bitmask[1] == -1).all()
assert not (bitmask[2] == -1).all()
assert not (bitmask[-1] == -1).all()
assert not grammar.is_terminated()
@pytest.mark.parametrize("backend", ["xgrammar", "guidance"])
def test_bitmask_post_reasoning_end_drafts_skip_grammar_advance(backend):
"""Post-marker drafts predate the bitmask and may be grammar-invalid;
grammar_bitmask must skip the grammar advance instead of asserting.
"""
tokenizer, manager, request, prompt = _make_manager_and_request(
backend, prompt_str="{"
)
grammar = request.structured_output_request.grammar
assert grammar.accept_tokens(request.request_id, prompt)
assert not grammar.is_terminated()
marker = tokenizer.encode("\n")[0]
class StubReasoner:
def __init__(self, *_, **__):
self.end_token_id = marker
def is_reasoning_end(self, input_ids):
return marker in list(input_ids)
def is_reasoning_end_streaming(self, input_ids, delta_ids):
return marker in list(delta_ids)
manager.reasoner_cls = StubReasoner
request.structured_output_request.reasoner = StubReasoner()
request.structured_output_request.reasoning_ended = False
pre = tokenizer.encode(" ")[0]
# A token that the JSON grammar would reject as the first post-marker
# token; without the fix grammar.accept_tokens fires the assertion.
invalid_post = tokenizer.encode("z")[0]
drafts = [pre, marker, invalid_post]
bitmask = manager.grammar_bitmask(
requests={request.request_id: request},
structured_output_request_ids=[request.request_id],
scheduled_spec_decode_tokens={request.request_id: drafts},
)
assert bitmask is not None
assert bitmask.shape[0] == len(drafts) + 1
# Post-marker position is still bitmask-constrained.
assert not (bitmask[2] == -1).all()
# Grammar must not have advanced through the unvalidated draft.
assert not grammar.is_terminated()
@pytest.mark.parametrize("backend", ["xgrammar", "guidance"])
def test_validate_tokens_then_bitmask_round_trip(backend):
"""validate_tokens -> pad with -1 -> grammar_bitmask must not assert."""
tokenizer, manager, request, prompt = _make_manager_and_request(backend)
grammar = request.structured_output_request.grammar
assert grammar.accept_tokens(request.request_id, prompt)
raw_drafts = [tokenizer.encode(",")[0], 99999, 12345, 67890]
accepted = grammar.validate_tokens(raw_drafts)
assert len(accepted) <= len(raw_drafts)
padded = accepted + [-1] * (len(raw_drafts) - len(accepted))
assert len(padded) == len(raw_drafts)
bitmask = manager.grammar_bitmask(
requests={request.request_id: request},
structured_output_request_ids=[request.request_id],
scheduled_spec_decode_tokens={request.request_id: padded},
)
assert bitmask is not None
assert bitmask.shape[0] == len(padded) + 1
assert not grammar.is_terminated()
class _MarkerReasoner:
"""Stub reasoner whose reasoning-end marker is a single fixed token."""
def __init__(self, marker: int):
self.marker = marker
def is_reasoning_end(self, input_ids):
return self.marker in list(input_ids)
def is_reasoning_end_streaming(self, input_ids, delta_ids):
return self.marker in list(delta_ids)
def _setup_boundary_request(backend: str):
"""Request with a structural-tag key and reasoning not yet ended."""
from vllm.v1.structured_output.backend_types import StructuredOutputOptions
tokenizer, manager, request, prompt = _make_manager_and_request(backend)
marker = tokenizer.encode("\n")[0]
structured_req = request.structured_output_request
# The grammar itself is JSON (cheap to build); only the key kind matters
# for the should_advance structural-tag branch, so pre-seed the cached
# property.
structured_req.__dict__["structured_output_key"] = (
StructuredOutputOptions.STRUCTURAL_TAG,
"",
)
manager.reasoner_cls = _MarkerReasoner
structured_req.reasoner = _MarkerReasoner(marker)
structured_req.reasoning_ended = False
return tokenizer, manager, request, prompt, marker
def test_should_advance_records_reasoning_end_index():
"""Regression for #44006 on post-#42452 main: the boundary step must
record where reasoning ends so the scheduler can trim before advancing.
"""
tokenizer, manager, request, prompt, marker = _setup_boundary_request("xgrammar")
structured_req = request.structured_output_request
pre = tokenizer.encode(" ")[0]
post = tokenizer.encode("{")[0]
request.append_output_token_ids([pre, marker, post])
assert manager.should_advance(request)
assert structured_req.reasoning_ended
# Marker sits at absolute index len(prompt) + 1.
assert structured_req.reasoning_end_token_index == len(prompt) + 1
def test_trim_reasoning_for_advance():
"""trim drops the marker and everything before it; later steps and
requests without a recorded boundary pass through unchanged.
"""
tokenizer, manager, request, prompt, marker = _setup_boundary_request("xgrammar")
structured_req = request.structured_output_request
pre = tokenizer.encode(" ")[0]
post = tokenizer.encode("{")[0]
# No boundary recorded yet: pass-through.
assert manager.trim_reasoning_for_advance(request, [pre]) == [pre]
# Boundary step: marker mid-step keeps only the suffix.
step_tokens = [pre, marker, post]
request.append_output_token_ids(step_tokens)
assert manager.should_advance(request)
assert manager.trim_reasoning_for_advance(request, step_tokens) == [post]
# Boundary step variant: marker last (the #44006 crash shape
# [198, </think>]) trims to empty -> scheduler skips accept_tokens.
structured_req.reasoning_end_token_index = len(request.all_token_ids) - 1
assert manager.trim_reasoning_for_advance(request, step_tokens) == []
# Later steps: tokens are past the boundary, returned unchanged.
structured_req.reasoning_end_token_index = len(prompt) + 1
next_step = [post, post]
request.append_output_token_ids(next_step)
assert manager.trim_reasoning_for_advance(request, next_step) == next_step
@@ -17,7 +17,7 @@ from vllm.v1.structured_output import StructuredOutputManager
from vllm.v1.structured_output.backend_guidance import GuidanceBackend
from vllm.v1.structured_output.backend_types import StructuredOutputOptions
TOKENIZER = "gpt2"
TOKENIZER = "openai-community/gpt2"
@pytest.fixture(scope="module")
+6
View File
@@ -119,6 +119,11 @@ class CacheConfig:
mamba_page_size_padded: int | None = None
""" Optional override for mamba page size; used by hybrid mamba/attention
models to ensure exact alignment with attention page size."""
skip_page_size_padded: int | None = None
"""Optional override for the page size of layers skipped from KV cache
quantization (``--kv-cache-dtype-skip-layers``); set during block-size
alignment so unquantized skip layers pad up to the quantized primary's
page."""
mamba_block_size: int | None = Field(default=None, gt=0)
"""Size of a contiguous cache block in number of tokens for mamba cache.
Can be set only when prefix caching is enabled.
@@ -207,6 +212,7 @@ class CacheConfig:
# Prefix-caching implementation detail (doesn't affect compiled graph).
"hash_block_size",
"mamba_page_size_padded",
"skip_page_size_padded",
"user_specified_block_size",
"user_specified_mamba_block_size",
"_block_size_resolved",
+6 -2
View File
@@ -1321,6 +1321,7 @@ class CompilationConfig:
min_cg_support: "AttentionCGSupport",
min_cg_attn_backend: str | None,
uniform_decode_query_len: int = 1,
use_v2_model_runner: bool = False,
tensor_parallel_size: int = 1,
kv_cache_config: "KVCacheConfig | None" = None,
max_num_reqs: int | None = None,
@@ -1421,13 +1422,16 @@ class CompilationConfig:
"and make sure compilation mode is VLLM_COMPILE"
)
# Adjust cudagraph sizes to be a multiple of uniform_decode_query_len
# MRV1 adjusts cudagraph sizes to be a multiple of uniform_decode_query_len
# to avoid: https://github.com/vllm-project/vllm/issues/28207 and temp-fix:
# https://github.com/vllm-project/vllm/issues/28207#issuecomment-3504004536
# Will be removed in the near future when we have separate cudagraph capture
# sizes for decode and mixed prefill-decode.
# MRV2 handles cudagraph capture sizing in cudagraph_utils.py
# and doesn't need below: https://github.com/vllm-project/vllm/pull/45953
if (
cudagraph_mode.decode_mode() == CUDAGraphMode.FULL
not use_v2_model_runner
and cudagraph_mode.decode_mode() == CUDAGraphMode.FULL
and uniform_decode_query_len > 1
):
self.adjust_cudagraph_sizes_for_spec_decode(
+6 -1
View File
@@ -156,6 +156,8 @@ LinearBackend = Literal[
"conch",
"exllama",
"emulation",
"xpu",
"xpu_woq",
]
@@ -217,7 +219,10 @@ class KernelConfig:
- "fbgemm": Use FBGEMM kernels
- "conch": Use Conch mixed-precision kernels
- "exllama": Use Exllama mixed-precision kernels
- "emulation": Use slow dequant-to-BF16 emulation (for testing only)"""
- "emulation": Use slow dequant-to-BF16 emulation (for testing only)
- "xpu": Use XPU kernels
- "xpu_woq": Use XPU kernels for weight-only quantization (e.g. W8A16)
"""
@field_validator("moe_backend", mode="before")
@classmethod
+3 -4
View File
@@ -774,13 +774,15 @@ class VllmConfig:
speculative_config is None
or not speculative_config.uses_dynamic_speculative_decoding()
or not self.compilation_config.cudagraph_mode.has_full_cudagraphs()
or self.use_v2_model_runner
):
return
logger.warning_once(
"Dynamic speculative decoding changes the target verification "
"length at runtime. Overriding cudagraph_mode from %s to "
"PIECEWISE for reliability.",
"PIECEWISE for reliability. Use VLLM_USE_V2_MODEL_RUNNER=1 "
"if you want to use full CUDA graphs.",
self.compilation_config.cudagraph_mode.name,
)
self.compilation_config.cudagraph_mode = CUDAGraphMode.PIECEWISE
@@ -2062,9 +2064,6 @@ class VllmConfig:
):
unsupported.append(f"speculative method '{speculative_config.method}'")
if speculative_config.uses_dynamic_speculative_decoding():
unsupported.append("dynamic speculative decoding")
# V2 EagleSpeculator does not support parallel_drafting (for P-Eagle).
# DFlash and DSpark use parallel drafting natively in V2 via their
# own speculators.
+1 -1
View File
@@ -120,7 +120,7 @@ async def run_launch_fastapi(args: argparse.Namespace) -> None:
signal.signal(signal.SIGTERM, _interrupt_init)
# 1. Socket binding
listen_address, sock = setup_server(args)
listen_address, sock = setup_server(args, reuse_port=False)
# 2. Build and serve the API server
engine_args = AsyncEngineArgs.from_cli_args(args)
+1 -1
View File
@@ -281,7 +281,7 @@ def run_multi_api_server(args: argparse.Namespace):
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)
listen_address, sock = setup_server(args)
listen_address, sock = setup_server(args, reuse_port=num_api_servers > 1)
engine_args = vllm.AsyncEngineArgs.from_cli_args(args)
engine_args._api_process_count = num_api_servers
+10 -5
View File
@@ -508,14 +508,19 @@ async def init_render_app_state(
state.server_load_metrics = 0
def create_server_socket(addr: tuple[str, int]) -> socket.socket:
def create_server_socket(
addr: tuple[str, int],
*,
reuse_port: bool,
) -> socket.socket:
family = socket.AF_INET
if is_valid_ipv6_address(addr[0]):
family = socket.AF_INET6
sock = socket.socket(family=family, type=socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
if reuse_port:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
sock.bind(addr)
return sock
@@ -546,7 +551,7 @@ def validate_api_server_args(args):
@instrument(span_name="API server setup")
def setup_server(args):
def setup_server(args, *, reuse_port: bool):
"""Validate API server args and create the server socket."""
log_version_and_model(logger, VLLM_VERSION, args.model)
@@ -567,7 +572,7 @@ def setup_server(args):
sock = create_server_unix_socket(args.uds)
else:
sock_addr = (args.host or "", args.port)
sock = create_server_socket(sock_addr)
sock = create_server_socket(sock_addr, reuse_port=reuse_port)
# workaround to avoid footguns where uvicorn drops requests with too
# many concurrent requests active
@@ -688,7 +693,7 @@ async def run_server(args, **uvicorn_kwargs) -> None:
signal.signal(signal.SIGTERM, _interrupt_init)
listen_address, sock = setup_server(args)
listen_address, sock = setup_server(args, reuse_port=False)
await run_server_worker(listen_address, sock, args, **uvicorn_kwargs)
@@ -181,7 +181,7 @@ class OpenAIServingChatBatch(OpenAIServingChat):
sub_request_id,
lora_request=lora_request,
trace_headers=trace_headers,
priority=request.priority if hasattr(request, "priority") else 0,
priority=request.priority,
data_parallel_rank=data_parallel_rank,
reasoning_ended=None,
)
@@ -1000,8 +1000,8 @@ class BatchChatCompletionRequest(OpenAIBaseModel):
response_format: Any | None = None
seed: int | None = Field(None, ge=_INT64_MIN, le=_INT64_MAX)
stop: str | list[str] | None = Field(default_factory=list)
temperature: float | None = 0.7
top_p: float | None = 1.0
temperature: float | None = None
top_p: float | None = None
user: str | None = None
tool_choice: Literal["none"] | None = "none"
include_reasoning: bool = True
@@ -1010,8 +1010,8 @@ class BatchChatCompletionRequest(OpenAIBaseModel):
best_of: int | None = None
use_beam_search: bool = False
top_k: int | None = None
min_p: float | None = 0.0
repetition_penalty: float | None = 1.0
min_p: float | None = None
repetition_penalty: float | None = None
length_penalty: float | None = 1.0
early_stopping: bool = False
structured_outputs: StructuredOutputsParams | None = None
@@ -1020,6 +1020,10 @@ class BatchChatCompletionRequest(OpenAIBaseModel):
continue_final_message: bool = False
chat_template: str | None = None
chat_template_kwargs: dict[str, Any] | None = None
media_io_kwargs: dict[str, dict[str, Any]] | None = None
mm_processor_kwargs: dict[str, Any] | None = None
priority: int = Field(default=0, ge=_INT64_MIN, le=_INT64_MAX)
cache_salt: str | None = None
include_stop_str_in_output: bool = False
guided_decoding_backend: str | None = None
echo: bool = False
+6 -6
View File
@@ -18,7 +18,7 @@ from openai.types.responses.response_output_item import McpCall
from openai.types.responses.response_output_message import ResponseOutputMessage
from openai.types.responses.response_output_text import ResponseOutputText
from openai.types.responses.tool import Mcp
from openai_harmony import Author, HarmonyError, Message, Role, TextContent
from openai_harmony import Author, Message, Role, TextContent
from vllm import envs
from vllm.entrypoints.chat_utils import (
@@ -616,7 +616,7 @@ class HarmonyContext(ConversationContext):
self.num_tool_output_tokens = 0
self.last_append_segments: list[Segment] = []
self.last_append_flush_status: bool | HarmonyError = False
self.last_append_flush_status: bool = False
# Turn tracking - replaces multiple individual tracking variables
self.current_turn_metrics = TurnMetrics()
@@ -643,10 +643,10 @@ class HarmonyContext(ConversationContext):
if output.finished:
self.finish_reason = output.outputs[0].finish_reason
flushed = self.response_parser.flush()
if flushed is not None:
segments.append(flushed)
self.last_append_flush_status = flushed is not None
flushed_segments = self.response_parser.flush()
if flushed_segments:
segments.extend(flushed_segments)
self.last_append_flush_status = len(flushed_segments) > 0
self.all_turn_metrics.append(self.current_turn_metrics.copy())
self.current_turn_metrics.reset()
+15
View File
@@ -69,6 +69,21 @@ class PoolingBasicRequestMixin(OpenAIBaseModel):
)
# --8<-- [end:pooling-common-extra-params]
@model_validator(mode="before")
@classmethod
def check_cache_salt_support(cls, data):
if not isinstance(data, dict):
return data
if data.get("cache_salt") is not None and (
not isinstance(data["cache_salt"], str) or not data["cache_salt"]
):
raise VLLMValidationError(
"Parameter 'cache_salt' must be a non-empty string if provided.",
parameter="cache_salt",
)
return data
def _build_pooling_tok_params(
self,
model_config: ModelConfig,
@@ -10,6 +10,7 @@ from vllm import PoolingParams, PoolingRequestOutput, TokensPrompt
from vllm.inputs import EngineInput
from vllm.renderers import TokenizeParams
from vllm.renderers.hf import safe_apply_chat_template
from vllm.renderers.inputs.preprocess import extract_target_prompt
from vllm.tasks import PoolingTask
from vllm.utils.mistral import is_mistral_tokenizer
@@ -433,8 +434,16 @@ class CrossEncoderIOProcessor(ScoringIOProcessor):
max_tokens_per_doc: int = 0,
prompt_extras: dict[str, Any] | None = None,
) -> tuple[Sequence[EngineInput], list[PoolingParams]]:
# todo: support prompt_extras
arrival_time = time.time()
engine_prompt_extras = (
{
k: v
for k in ("mm_processor_kwargs", "cache_salt")
if (v := prompt_extras.get(k)) is not None
}
if prompt_extras
else None
)
data_1 = scoring_data.data_1
data_2 = scoring_data.data_2
@@ -463,12 +472,18 @@ class CrossEncoderIOProcessor(ScoringIOProcessor):
if token_type_ids := engine_prompt.pop("token_type_ids", None):
params = pooling_params.clone()
compressed = compress_token_type_ids(token_type_ids)
params.extra_kwargs = {"compressed_token_type_ids": compressed}
params.extra_kwargs = {
**(params.extra_kwargs or {}),
"compressed_token_type_ids": compressed,
}
pooling_params_list.append(params)
else:
pooling_params_list.append(pooling_params)
tok_params.apply_post_tokenization(self.tokenizer, engine_prompt)
if engine_prompt_extras:
target_prompt = extract_target_prompt(self.model_config, engine_prompt)
target_prompt.update(engine_prompt_extras)
engine_inputs.append(
self.renderer.process_for_engine(engine_prompt, arrival_time)
)
+12 -3
View File
@@ -171,7 +171,8 @@ from vllm.model_executor.kernels.linear.scaled_mm.triton import (
)
from vllm.model_executor.kernels.linear.scaled_mm.xpu import (
XPUFp8BlockScaledMMKernel,
XPUFP8ScaledMMLinearKernel,
XPUW8A8FP8LinearKernel,
XPUW8A16FP8LinearKernel,
)
from vllm.model_executor.kernels.linear.scaled_mm.zentorch import (
ZentorchInt8ScaledMMLinearKernel,
@@ -266,6 +267,13 @@ _LINEAR_BACKEND_KERNEL_MAP: dict[str, set[type]] = {
EmulationMxfp8LinearKernel,
EmulationNvFp4LinearKernel,
},
"xpu": {
XPUW8A8FP8LinearKernel,
XPUFp8BlockScaledMMKernel,
},
"xpu_woq": {
XPUW8A16FP8LinearKernel,
},
}
@@ -311,7 +319,8 @@ _POSSIBLE_FP8_KERNELS: dict[PlatformEnum, list[type[FP8ScaledMMLinearKernel]]] =
ChannelWiseTorchFP8ScaledMMLinearKernel,
],
PlatformEnum.XPU: [
XPUFP8ScaledMMLinearKernel,
XPUW8A16FP8LinearKernel,
XPUW8A8FP8LinearKernel,
],
}
@@ -351,7 +360,7 @@ _POSSIBLE_WFP8A16_KERNELS: dict[PlatformEnum, list[type[FP8ScaledMMLinearKernel]
# To be added
],
PlatformEnum.XPU: [
XPUFP8ScaledMMLinearKernel,
XPUW8A16FP8LinearKernel,
],
}
@@ -6,8 +6,11 @@ from collections.abc import Sequence
import torch
from vllm.model_executor.layers.quantization.utils.quant_utils import (
kFp8DynamicTensorSym,
kFp8DynamicTokenSym,
kFp8StaticChannelSym,
kFp8StaticTensorSym,
kFp8StaticTokenSym,
)
from vllm.model_executor.utils import replace_parameter
from vllm.platforms import current_platform
@@ -16,13 +19,112 @@ from .BlockScaledMMLinearKernel import Fp8BlockScaledMMLinearKernel
from .ScaledMMLinearKernel import FP8ScaledMMLinearKernel, FP8ScaledMMLinearLayerConfig
class XPUFP8ScaledMMLinearKernel(FP8ScaledMMLinearKernel):
class XPUW8A8FP8LinearKernel(FP8ScaledMMLinearKernel):
_SUPPORTED_ACT_QUANT_KEYS = {
kFp8DynamicTensorSym,
kFp8DynamicTokenSym,
kFp8StaticTensorSym,
kFp8StaticTokenSym,
}
_SUPPORTED_WEIGHT_QUANT_KEYS = {
kFp8StaticChannelSym,
kFp8StaticTensorSym,
}
@classmethod
def is_supported(
cls, compute_capability: int | None = None
) -> tuple[bool, str | None]:
if not current_platform.is_xpu():
return False, "XPUFP8ScaledMM only support on XPU"
return False, "XPUW8A8FP8Linear only support on XPU"
return True, None
@classmethod
def can_implement(cls, c: FP8ScaledMMLinearLayerConfig) -> tuple[bool, str | None]:
if c.weight_quant_key not in cls._SUPPORTED_WEIGHT_QUANT_KEYS:
return (
False,
"XPUW8A8FP8Linear only support per-channel and per-tensor quantization",
)
if c.activation_quant_key not in cls._SUPPORTED_ACT_QUANT_KEYS:
return (
False,
"XPUW8A8FP8Linear only support per-tensor and per-token activation "
"quantization",
)
if c.weight_quant_key.dtype not in {torch.float8_e5m2, torch.float8_e4m3fn}:
return False, "XPUW8A8FP8Linear only support FP8 weight dtype"
if c.activation_quant_key.dtype not in {
torch.float8_e5m2,
torch.float8_e4m3fn,
}:
return False, "XPUW8A8FP8Linear only support FP8 activation dtype"
return True, None
def __init__(
self, c: FP8ScaledMMLinearLayerConfig, layer_param_names: Sequence[str]
) -> None:
super().__init__(c, layer_param_names)
def process_weights_after_loading(self, layer: torch.nn.Module) -> None:
"""Ensure weight is stored as C-contiguous [K, N] (KN layout).
Checkpoints store weight as [N, K]; fp8_gemm requires [K, N],
C-contiguous. Three incoming layouts are possible:
[N, K] C-contiguous direct checkpoint .t().contiguous()
[K, N] Fortran-order fp8.py's weight.t() → .contiguous()
[K, N] C-contiguous already correct no-op
For square weights (K == N) the shape is ambiguous; contiguity is used
as a proxy: C-contiguous checkpoint [N, K] (needs transpose);
Fortran-order fp8.py already transposed (needs only contiguous).
"""
K = getattr(layer, "input_size_per_partition", self.config.weight_shape[1])
N = getattr(layer, "output_size_per_partition", self.config.weight_shape[0])
w = layer.weight
if w.shape not in {(K, N), (N, K)}:
raise ValueError(
f"XPUFP8ScaledMM expects weight shape ({K},{N}) or ({N},{K}), "
f"but got {tuple(w.shape)}"
)
needs_transpose = w.shape == (N, K) if K != N else w.is_contiguous()
layer_weight = w.t() if needs_transpose else w
replace_parameter(layer, "weight", layer_weight)
ws = layer.weight_scale
if ws.numel() == 1:
replace_parameter(layer, "weight_scale", ws.reshape(1))
def apply_scaled_mm(
self,
*,
A: torch.Tensor,
B: torch.Tensor,
out_dtype: torch.dtype,
As: torch.Tensor,
Bs: torch.Tensor,
bias: torch.Tensor | None,
output_shape: list,
) -> torch.Tensor:
# B is C-contiguous [K, N] from process_weights_after_loading.
# fp8_gemm routes on scale dtype (float32) and numel:
# As [1] → per-tensor (numel==1 branch)
# As [M,1] → per-token (group={1,K} branch, broadcast across K)
# Bs [1] → per-tensor
# Bs [N] → per-channel (mask=bit1 branch)
# No shape manipulation needed here.
output = torch.ops._xpu_C.fp8_gemm(A, B, out_dtype, As, Bs, bias)
return output.view(*output_shape)
class XPUW8A16FP8LinearKernel(FP8ScaledMMLinearKernel):
@classmethod
def is_supported(
cls, compute_capability: int | None = None
) -> tuple[bool, str | None]:
if not current_platform.is_xpu():
return False, "XPUW8A16FP8Linear only support on XPU"
return True, None
@classmethod
@@ -30,10 +132,11 @@ class XPUFP8ScaledMMLinearKernel(FP8ScaledMMLinearKernel):
if c.weight_quant_key not in {kFp8StaticChannelSym, kFp8StaticTensorSym}:
return (
False,
"XPUFP8ScaledMM only support per-channel and per-tensor quantization",
"XPUW8A16FP8Linear only support per-channel and per-tensor "
"quantization",
)
if c.weight_quant_key.dtype not in {torch.float8_e5m2, torch.float8_e4m3fn}:
return False, "XPUFP8ScaledMM only support FP8 weight dtype"
return False, "XPUW8A16FP8Linear only support FP8 weight dtype"
return True, None
def __init__(
@@ -92,6 +92,35 @@ def should_load_quant_weights(quant_method: QuantizeMethodBase | None) -> bool:
)
def _largest_kernel_block_within(
attn_backend: "type[AttentionBackend]",
per_token_bytes: int,
page_budget: int | None,
fallback: int,
) -> int:
"""Largest supported kernel block size whose page fits in ``page_budget``.
A padded spec (e.g. skip-quant layer) that pads its page up to a large shared page
wastes ``page_budget - block*per_token`` bytes per block. Picking the largest kernel
block whose natural page still fits under ``page_budget`` minimizes that waste.
Falls back to the smallest supported block when ``page_budget`` is None (no padding
the block is handled by ``unify``'s integer scaling instead) or nothing fits.
"""
from vllm.v1.attention.backend import MultipleOf
sizes = attn_backend.get_supported_kernel_block_sizes()
candidates = [s for s in sizes if isinstance(s, int)]
if not candidates:
candidates = [s.base for s in sizes if isinstance(s, MultipleOf)]
if not candidates:
return fallback
smallest = min(candidates)
if not page_budget or per_token_bytes <= 0:
return smallest
fitting = [b for b in candidates if b * per_token_bytes <= page_budget]
return max(fitting) if fitting else smallest
def set_default_quant_scales(layer: nn.Module, register_buffer: bool = False) -> None:
"""Sets default quantization scales for the layer."""
if register_buffer:
@@ -601,14 +630,35 @@ class Attention(nn.Module, AttentionLayerBase):
assert not vllm_config.model_config.use_mla, (
"MLA is not supported for slidingwindow"
)
return SlidingWindowSpec(
block_size=block_size,
# SW chooses its own block_size, decoupled from the user's
# ``--block-size`` (which only constrains primary attention).
# When this SW layer is a padded spec (skip-quant: its page is
# padded up to ``skip_page_size_padded``), pick the largest kernel
# block that still fits the shared page so we waste fewer padding
# bytes per block. Otherwise (page_size_padded is None) the smallest
# block is fine — ``unify`` scales it up by an integer ratio.
shared_page = vllm_config.cache_config.skip_page_size_padded
sw_per_token = SlidingWindowSpec(
block_size=1,
num_kv_heads=self.num_kv_heads,
head_size=self.head_size,
head_size_v=self.head_size_v,
dtype=self.kv_cache_torch_dtype,
kv_quant_mode=quant_mode,
sliding_window=self.sliding_window,
).real_page_size_bytes
sw_block_size = _largest_kernel_block_within(
self.attn_backend, sw_per_token, shared_page, block_size
)
return SlidingWindowSpec(
block_size=sw_block_size,
num_kv_heads=self.num_kv_heads,
head_size=self.head_size,
head_size_v=self.head_size_v,
dtype=self.kv_cache_torch_dtype,
kv_quant_mode=quant_mode,
sliding_window=self.sliding_window,
page_size_padded=shared_page,
)
elif self.kv_cache_dtype.startswith("turboquant_"):
from vllm.model_executor.layers.quantization.turboquant.config import (
@@ -17,6 +17,8 @@ from .op import exp, exp2
from .utils import FLA_CHUNK_SIZE, use_cuda_graph
NUM_WARPS = [2, 4, 8, 16]
# Triton's AMD backend fails to lower this kernel with num_stages=4.
_CHUNK_DELTA_H_NUM_STAGES = [2, 3] if torch.version.hip else [2, 3, 4]
@triton.heuristics(
@@ -33,7 +35,7 @@ NUM_WARPS = [2, 4, 8, 16]
configs=[
triton.Config({"BV": BV}, num_warps=num_warps, num_stages=num_stages)
for num_warps in [2, 4]
for num_stages in [2, 3, 4]
for num_stages in _CHUNK_DELTA_H_NUM_STAGES
for BV in [32, 64]
],
key=["H", "K", "V", "BT"],
@@ -1,6 +1,9 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
import functools
import inspect
import torch
import vllm.model_executor.layers.fused_moe.modular_kernel as mk
@@ -33,6 +36,40 @@ from vllm.utils.flashinfer import has_flashinfer_trtllm_fused_moe
logger = init_logger(__name__)
@functools.cache
def _flashinfer_moe_supports_swiglu_params(fn_name: str) -> bool:
try:
import flashinfer
fn = getattr(flashinfer.fused_moe, fn_name)
return "gemm1_alpha" in inspect.signature(fn).parameters
except (TypeError, ValueError):
# If FlashInfer stops exposing an inspectable Python signature, keep
# passing the parameters so an incompatible install fails explicitly.
return True
def _add_swiglu_params_if_supported(
kwargs: dict[str, object],
fn_name: str,
gemm1_alpha: torch.Tensor | None,
gemm1_beta: torch.Tensor | None,
gemm1_clamp_limit: torch.Tensor | None,
) -> None:
swiglu_params = {
"gemm1_alpha": gemm1_alpha,
"gemm1_beta": gemm1_beta,
"gemm1_clamp_limit": gemm1_clamp_limit,
}
if _flashinfer_moe_supports_swiglu_params(fn_name):
kwargs.update(swiglu_params)
elif any(param is not None for param in swiglu_params.values()):
raise RuntimeError(
"The installed FlashInfer TRTLLM FP8 MoE kernel does not support "
"per-expert SwiGLU parameters. Please upgrade FlashInfer."
)
class TrtLlmFp8ExpertsBase:
"""
Fp8 TRTLLM-Gen MoE kernels. Shared base for modular and monolithic
@@ -224,16 +261,13 @@ class TrtLlmFp8ExpertsModular(TrtLlmFp8ExpertsBase, mk.FusedMoEExpertsModular):
weight_layout = WeightLayout.BlockMajorK
hidden_states_scale = a1q_scale.t().contiguous()
flashinfer.fused_moe.trtllm_fp8_block_scale_routed_moe(
kwargs = dict(
topk_ids=packed_topk_ids,
routing_bias=None,
hidden_states=hidden_states,
hidden_states_scale=hidden_states_scale,
gemm1_weights=w1,
gemm1_weights_scale=self.quant_config.w1_scale,
gemm1_alpha=self.gemm1_alpha,
gemm1_beta=self.gemm1_beta,
gemm1_clamp_limit=self.gemm1_clamp_limit,
gemm2_weights=w2,
gemm2_weights_scale=self.quant_config.w2_scale,
num_experts=global_num_experts,
@@ -250,6 +284,14 @@ class TrtLlmFp8ExpertsModular(TrtLlmFp8ExpertsBase, mk.FusedMoEExpertsModular):
fp8_quantization_type=fp8_quant_type,
output=output,
)
_add_swiglu_params_if_supported(
kwargs,
"trtllm_fp8_block_scale_routed_moe",
self.gemm1_alpha,
self.gemm1_beta,
self.gemm1_clamp_limit,
)
flashinfer.fused_moe.trtllm_fp8_block_scale_routed_moe(**kwargs)
class TrtLlmFp8ExpertsMonolithic(TrtLlmFp8ExpertsBase, mk.FusedMoEExpertsMonolithic):
@@ -402,9 +444,6 @@ class TrtLlmFp8ExpertsMonolithic(TrtLlmFp8ExpertsBase, mk.FusedMoEExpertsMonolit
hidden_states_scale=hidden_states_scale,
gemm1_weights=w1,
gemm1_weights_scale=self.quant_config.w1_scale,
gemm1_alpha=self.gemm1_alpha,
gemm1_beta=self.gemm1_beta,
gemm1_clamp_limit=self.gemm1_clamp_limit,
gemm2_weights=w2,
gemm2_weights_scale=self.quant_config.w2_scale,
num_experts=global_num_experts,
@@ -422,6 +461,13 @@ class TrtLlmFp8ExpertsMonolithic(TrtLlmFp8ExpertsBase, mk.FusedMoEExpertsMonolit
)
if is_mxfp8 or activation == MoEActivation.RELU2_NO_MUL:
kwargs["activation_type"] = activation_type
_add_swiglu_params_if_supported(
kwargs,
"trtllm_fp8_block_scale_moe",
self.gemm1_alpha,
self.gemm1_beta,
self.gemm1_clamp_limit,
)
return flashinfer.fused_moe.trtllm_fp8_block_scale_moe(**kwargs)
def _apply_per_tensor(
@@ -13,6 +13,7 @@ from compressed_tensors.quantization import (
)
from compressed_tensors.transform import TransformConfig
from vllm.config import get_current_vllm_config_or_none
from vllm.distributed import (
get_tensor_model_parallel_rank,
get_tensor_model_parallel_world_size,
@@ -747,9 +748,18 @@ class CompressedTensorsConfig(QuantizationConfig):
act_quant_format = is_activation_quantization_format(format)
if act_quant_format:
if self._is_fp8_w8a8(weight_quant, input_quant):
is_fp8_w8a8_supported = self._check_scheme_supported(
CompressedTensorsW8A8Fp8.get_min_capability(), error=False
)
if current_platform.is_xpu():
# On XPU, --linear-backend xpu opts into W8A8 FP8
# linear kernel; otherwise default to W8A16.
config = get_current_vllm_config_or_none()
is_fp8_w8a8_supported = (
config is not None
and config.kernel_config.linear_backend == "xpu"
)
else:
is_fp8_w8a8_supported = self._check_scheme_supported(
CompressedTensorsW8A8Fp8.get_min_capability(), error=False
)
if is_fp8_w8a8_supported:
return CompressedTensorsW8A8Fp8(
weight_quant=weight_quant,
@@ -1322,6 +1322,11 @@ class QuarkOCP_MX_MoEMethod(QuarkMoEMethod):
gemm1_clamp_limit=getattr(layer, "swiglu_limit", None),
)
@property
def supports_eplb(self) -> bool:
# AITER shuffle keeps expert dim outermost, so EPLB row moves are layout-safe.
return True
@property
def is_monolithic(self) -> bool:
if self.moe_kernel is not None:
@@ -47,21 +47,29 @@ class RunaiModelStreamerLoader(BaseModelLoader):
# Validate every value before mutating os.environ, so a later
# invalid key cannot leave an earlier one partially applied.
env_updates: dict[str, str] = {}
for key, env_var in (
("concurrency", "RUNAI_STREAMER_CONCURRENCY"),
("memory_limit", "RUNAI_STREAMER_MEMORY_LIMIT"),
):
if key in extra_config:
value = extra_config[key]
if (
isinstance(value, bool)
or not isinstance(value, int)
or value <= 0
):
raise ValueError(
f"{key} must be a positive integer, got {value!r}"
)
env_updates[env_var] = str(value)
if "concurrency" in extra_config:
concurrency = extra_config["concurrency"]
if (
isinstance(concurrency, bool)
or not isinstance(concurrency, int)
or concurrency <= 0
):
raise ValueError(
f"concurrency must be a positive integer, got {concurrency!r}"
)
env_updates["RUNAI_STREAMER_CONCURRENCY"] = str(concurrency)
if "memory_limit" in extra_config:
memory_limit = extra_config["memory_limit"]
if (
isinstance(memory_limit, bool)
or not isinstance(memory_limit, int)
or memory_limit < -1
):
raise ValueError(
f"memory_limit must be an integer >= -1, got {memory_limit!r}"
)
env_updates["RUNAI_STREAMER_MEMORY_LIMIT"] = str(memory_limit)
os.environ.update(env_updates)
runai_streamer_s3_endpoint = os.getenv("RUNAI_STREAMER_S3_ENDPOINT")
+16 -20
View File
@@ -713,6 +713,7 @@ class Indexer(nn.Module):
)
self.is_inplace_rope = is_inplace_rope
self.n_head_scale = self.n_head**-0.5
self.use_fused_indexer_q = (
current_platform.is_cuda()
and self.quant_block_size == self.head_dim
@@ -761,15 +762,16 @@ class Indexer(nn.Module):
rotary_emb.cos_sin_cache,
weights,
self.softmax_scale,
self.n_head**-0.5,
self.n_head_scale,
rotary_emb.is_neox_style,
)
# rotate only the MQA K
q_dummy = torch.empty_like(k_pe.unsqueeze(1))
_, k_pe = rotary_emb(positions, q_dummy, k_pe.unsqueeze(1))
k_pe = k_pe.reshape(-1, 1, self.rope_dim)
k = torch.cat([k_pe.squeeze(-2), k_nope], dim=-1)
k_pe = k_pe.unsqueeze(1)
q_dummy = torch.empty_like(k_pe)
_, k_pe = rotary_emb(positions, q_dummy, k_pe)
k_pe = k_pe.reshape(-1, self.rope_dim)
k = torch.cat([k_pe, k_nope], dim=-1)
return self.indexer_op(hidden_states, q_fp8, k, weights)
else:
@@ -790,13 +792,13 @@ class Indexer(nn.Module):
# Note: RoPE (NeoX) can introduce extra leading dimensions during
# compilation so we need to reshape back to token-flattened shapes
q_pe = q_pe.reshape(-1, self.n_head, self.rope_dim)
k_pe = k_pe.reshape(-1, 1, self.rope_dim)
k_pe = k_pe.reshape(-1, self.rope_dim)
# `rotary_emb` is shape-preserving; `q_pe` is already
# [num_tokens, n_head, rope_dim].
q = torch.cat([q_pe, q_nope], dim=-1)
# `k_pe` is [num_tokens, 1, rope_dim] (MQA).
k = torch.cat([k_pe.squeeze(-2), k_nope], dim=-1)
# `k_pe` is [num_tokens, rope_dim] (MQA).
k = torch.cat([k_pe, k_nope], dim=-1)
# we only quant q here since k quant is fused with cache insertion
q = q.view(-1, self.head_dim)
@@ -807,12 +809,9 @@ class Indexer(nn.Module):
use_ue8m0=self.scale_fmt is not None,
)
q_fp8 = q_fp8.view(-1, self.n_head, self.head_dim)
q_scale = q_scale.view(-1, self.n_head, 1)
q_scale = q_scale.view(-1, self.n_head)
weights = (
weights.unsqueeze(-1) * q_scale * self.softmax_scale * self.n_head**-0.5
)
weights = weights.squeeze(-1)
weights = weights * q_scale * self.softmax_scale * self.n_head_scale
return self.indexer_op(hidden_states, q_fp8, k, weights)
@@ -1287,13 +1286,10 @@ class DeepseekV2DecoderLayer(nn.Module):
hidden_states = tensor_model_parallel_all_gather(hidden_states, 0)
hidden_states = hidden_states[:full_num_tokens]
attn_kwargs = {
"positions": positions,
"hidden_states": hidden_states,
}
if not self.use_mha:
attn_kwargs["llama_4_scaling"] = llama_4_scaling
hidden_states = self.self_attn(**attn_kwargs)
if self.use_mha:
hidden_states = self.self_attn(positions, hidden_states)
else:
hidden_states = self.self_attn(positions, hidden_states, llama_4_scaling)
if (
not isinstance(self.self_attn, DeepseekAttention)
+67 -8
View File
@@ -981,23 +981,50 @@ class Glm4vProcessingInfo(BaseProcessingInfo):
def get_video_processor(self, **kwargs: object) -> Glm4vVideoProcessor:
return self.get_hf_processor(**kwargs).video_processor
def _get_processor_class_name(self) -> str | None:
from vllm.transformers_utils.processor import (
get_processor_cls_name_from_config,
)
from vllm.transformers_utils.utils import convert_model_repo_to_path
return get_processor_cls_name_from_config(
convert_model_repo_to_path(self.ctx.model_config.model),
revision=self.ctx.model_config.revision,
)
@staticmethod
def _get_longest_edge(size: Any, config_name: str) -> int:
if isinstance(size, dict):
longest_edge = size.get("longest_edge")
else:
longest_edge = getattr(size, "longest_edge", None)
if longest_edge is None:
raise ValueError(f"{config_name} must define longest_edge")
return int(longest_edge)
def get_mm_max_tokens_per_item(
self,
seq_len: int,
mm_counts: Mapping[str, int],
) -> Mapping[str, int] | None:
processor = self.get_hf_processor()
if isinstance(processor, Glm4vProcessor):
processor_class_name = self._get_processor_class_name()
if processor_class_name == "Glm4vProcessor":
return None
if processor_class_name is None:
processor = self.get_hf_processor()
if isinstance(processor, Glm4vProcessor):
return None
result: dict[str, int] = {}
if mm_counts.get("image", 0) > 0:
result["image"] = self.get_max_image_tokens()
if mm_counts.get("video", 0) > 0:
video_processor = self.get_video_processor()
max_pixels = video_processor.size["longest_edge"]
max_pixels = self._get_video_max_pixels()
vision_config = self.get_hf_config().vision_config
temporal_patch_size = vision_config.temporal_patch_size
@@ -1076,7 +1103,40 @@ class Glm4vProcessingInfo(BaseProcessingInfo):
``smart_resize`` as the ``max_pixels`` argument, which constrains
``t_bar * h_bar * w_bar <= max_pixels``.
"""
return self.get_image_processor().size["longest_edge"]
mm_kwargs = self.ctx.get_merged_mm_kwargs({})
if (override_max_pixels := mm_kwargs.get("max_pixels")) is not None:
return int(override_max_pixels)
image_processor_config = self.ctx.get_hf_image_processor_config()
if not image_processor_config.get("size"):
from transformers.image_processing_base import ImageProcessingMixin
image_processor_config, _ = ImageProcessingMixin.get_image_processor_dict(
self.ctx.model_config.model,
revision=self.ctx.model_config.revision,
)
size = image_processor_config["size"]
if override_size := mm_kwargs.get("size"):
size = size | override_size
return self._get_longest_edge(size, "GLM4V image processor size")
def _get_video_max_pixels(self) -> int:
from transformers.video_processing_utils import BaseVideoProcessor
mm_kwargs = self.ctx.get_merged_mm_kwargs({})
if (override_max_pixels := mm_kwargs.get("max_pixels")) is not None:
return int(override_max_pixels)
video_processor_config, _ = BaseVideoProcessor.get_video_processor_dict(
self.ctx.model_config.model,
revision=self.ctx.model_config.revision,
)
size = video_processor_config["size"]
if override_size := mm_kwargs.get("size"):
size = size | override_size
return self._get_longest_edge(size, "GLM4V video processor size")
def get_image_size_with_most_features(self) -> ImageSize:
# Use num_frames=1 for single-image budget estimation.
@@ -1424,13 +1484,12 @@ class Glm4vDummyInputsBuilder(BaseDummyInputsBuilder[Glm4vProcessingInfo]):
num_videos = mm_counts.get("video", 0)
hf_config = self.info.get_hf_config()
hf_processor = self.info.get_hf_processor()
tokenizer = self.info.get_tokenizer()
image_token: str = hf_processor.image_token
image_token = tokenizer.decode([hf_config.image_token_id])
video_token_ids = [
hf_config.video_start_token_id,
hf_processor.video_token_id,
hf_config.video_token_id,
hf_config.video_end_token_id,
]
video_token = tokenizer.decode(video_token_ids)
@@ -8,19 +8,6 @@ from typing import TYPE_CHECKING
import torch
from vllm.logger import init_logger
from vllm.model_executor.layers.fla.ops.fused_gdn_prefill_post_conv import (
fused_post_conv_prep,
)
from vllm.model_executor.layers.fla.ops.fused_sigmoid_gating import (
fused_sigmoid_gating_delta_rule_update,
)
from vllm.model_executor.layers.mamba.mamba_utils import is_conv_state_dim_first
from vllm.model_executor.layers.mamba.ops.causal_conv1d import (
causal_conv1d_fn,
)
from vllm.v1.attention.backends.utils import NULL_BLOCK_ID, PAD_SLOT_ID
from vllm.v1.worker.block_table import BlockTable
from vllm.v1.worker.utils import _zero_kv_blocks_kernel
if TYPE_CHECKING:
from vllm.v1.worker.gpu_model_runner import GPUModelRunner
@@ -128,6 +115,10 @@ def _qwen_gdn_warmup_config(
continue
conv_cache, ssm_state = cache_tensors
from vllm.model_executor.layers.mamba.mamba_utils import (
is_conv_state_dim_first,
)
conv_state = (
conv_cache if is_conv_state_dim_first() else conv_cache.transpose(-1, -2)
)
@@ -191,6 +182,8 @@ def _warm_zero_kv_blocks_with_runner_zeroer(runner: object) -> bool:
def _warm_zero_kv_blocks_kernel(
device: torch.device, config: _ZeroKvWarmupConfig
) -> None:
from vllm.v1.worker.utils import _zero_kv_blocks_kernel
max_n_blocks = max(_ZERO_KV_N_BLOCKS)
scratch = torch.empty(
max_n_blocks * config.page_size_el,
@@ -219,6 +212,8 @@ def _warm_zero_kv_blocks_kernel(
def _warm_compute_slot_mapping_kernel(device: torch.device) -> None:
from vllm.v1.worker.block_table import BlockTable
# num_tokens/max_num_tokens are do_not_specialize; keep the launch tiny.
num_tokens = 1
query_start_loc = torch.tensor([0, num_tokens], dtype=torch.int32, device=device)
@@ -244,6 +239,11 @@ def _warm_compute_slot_mapping_kernel(device: torch.device) -> None:
def _warm_causal_conv1d_fwd_kernel(
device: torch.device, config: _QwenGDNWarmupConfig
) -> None:
from vllm.model_executor.layers.mamba.ops.causal_conv1d import (
causal_conv1d_fn,
)
from vllm.v1.attention.backends.utils import NULL_BLOCK_ID, PAD_SLOT_ID
x_storage = torch.empty(
(1, config.conv_dim), dtype=config.conv_dtype, device=device
)
@@ -276,6 +276,10 @@ def _warm_causal_conv1d_fwd_kernel(
def _warm_fused_post_conv_kernel(
device: torch.device, config: _QwenGDNWarmupConfig
) -> None:
from vllm.model_executor.layers.fla.ops.fused_gdn_prefill_post_conv import (
fused_post_conv_prep,
)
qkv_dim = 2 * config.h * config.k + config.hv * config.v
for length in _FLA_POST_CONV_WARMUP_LENGTHS:
conv_output = torch.empty(
@@ -302,6 +306,10 @@ def _warm_fused_sigmoid_gating_delta_rule_update_kernel(
device: torch.device,
config: _QwenGDNWarmupConfig,
) -> None:
from vllm.model_executor.layers.fla.ops.fused_sigmoid_gating import (
fused_sigmoid_gating_delta_rule_update,
)
q = torch.empty((1, 1, config.h, config.k), dtype=config.conv_dtype, device=device)
k = torch.empty_like(q)
v = torch.empty((1, 1, config.hv, config.v), dtype=config.conv_dtype, device=device)
+8
View File
@@ -24,6 +24,7 @@ from vllm.utils.collection_utils import is_list_of
from vllm.utils.import_utils import LazyLoader
from .audio import AudioResampler, AudioSpec, normalize_audio
from .image import convert_image_mode, normalize_image
from .inputs import (
AudioItem,
HfAudioItem,
@@ -621,6 +622,13 @@ class MultiModalDataParser:
else:
data_items = data
data_items = [
convert_image_mode(normalize_image(item), "RGB")
if isinstance(item, PILImage.Image)
else item
for item in data_items
]
return ImageProcessorItems(data_items)
def _parse_video_data(
+131
View File
@@ -0,0 +1,131 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""DeepSeek V3.2 parser: DSML tool calls with ``function_calls`` wrapper.
DeepSeek V3.2 output format::
<DSMLfunction_calls>
<DSMLinvoke name="func_name">
<DSMLparameter name="location" string="true">杭州</DSMLparameter>
<DSMLparameter name="count" string="false">5</DSMLparameter>
</DSMLinvoke>
</DSMLfunction_calls>
This is identical to DeepSeek V4 except for the outer wrapper
(``function_calls`` instead of ``tool_calls``) and the absence of
``<think>``/``</think>`` reasoning tags.
"""
from __future__ import annotations
import functools
from typing import TYPE_CHECKING
from vllm.parser.deepseek_v4 import (
DSML_INVOKE_END,
DSML_INVOKE_NAME_END,
DSML_INVOKE_PREFIX,
DSML_PARAM_CLOSE,
_dsml_arg_converter,
_unwrap_wrapper_args,
)
from vllm.parser.engine.events import EventType
from vllm.parser.engine.parser_engine import ParserEngine
from vllm.parser.engine.parser_engine_config import (
ParserEngineConfig,
ParserState,
Transition,
)
if TYPE_CHECKING:
from vllm.tokenizers import TokenizerLike
from vllm.tool_parsers.abstract_tool_parser import Tool
_DSML = "DSML"
DSML_FUNC_START = f"<{_DSML}function_calls>"
DSML_FUNC_END = f"</{_DSML}function_calls>"
@functools.cache
def deepseek_v32_config() -> ParserEngineConfig:
return ParserEngineConfig(
name="deepseek_v32",
initial_state=ParserState.CONTENT,
terminals={
"TOOL_START": DSML_FUNC_START,
"TOOL_END": DSML_FUNC_END,
"INVOKE_PREFIX": DSML_INVOKE_PREFIX,
"INVOKE_NAME_END": DSML_INVOKE_NAME_END,
"INVOKE_END": DSML_INVOKE_END,
"PARAM_CLOSE": DSML_PARAM_CLOSE,
},
token_id_terminals={
"TOOL_START": DSML_FUNC_START,
"TOOL_END": DSML_FUNC_END,
},
transitions={
(ParserState.CONTENT, "TOOL_START"): Transition(
ParserState.TOOL_PREAMBLE,
(),
),
(ParserState.TOOL_PREAMBLE, "INVOKE_PREFIX"): Transition(
ParserState.TOOL_NAME,
(EventType.TOOL_CALL_START,),
),
(ParserState.TOOL_NAME, "INVOKE_NAME_END"): Transition(
ParserState.TOOL_ARGS,
(),
),
(ParserState.TOOL_ARGS, "INVOKE_END"): Transition(
ParserState.TOOL_BETWEEN,
(EventType.TOOL_CALL_END,),
),
(ParserState.TOOL_ARGS, "TOOL_END"): Transition(
ParserState.CONTENT,
(EventType.TOOL_CALL_END,),
),
# Parallel tool calls
(ParserState.TOOL_BETWEEN, "INVOKE_PREFIX"): Transition(
ParserState.TOOL_NAME,
(EventType.TOOL_CALL_START,),
),
(ParserState.TOOL_BETWEEN, "TOOL_END"): Transition(
ParserState.CONTENT,
(),
),
},
content_events={
ParserState.CONTENT: EventType.TEXT_CHUNK,
ParserState.TOOL_NAME: EventType.TOOL_NAME,
ParserState.TOOL_ARGS: EventType.ARG_VALUE_CHUNK,
},
arg_converter=_dsml_arg_converter,
arg_structural_chars=frozenset(">"),
strip_content_whitespace_with_tools=False,
tool_args_json=False,
)
class DeepSeekV32Parser(ParserEngine):
def __init__(
self,
tokenizer: TokenizerLike,
tools: list[Tool] | None = None,
**kwargs,
) -> None:
kwargs.pop("chat_template_kwargs", None)
super().__init__(
tokenizer,
tools,
parser_engine_config=deepseek_v32_config(),
**kwargs,
)
self._arg_converter = self._convert_args
def _convert_args(self, raw_args: str, partial: bool) -> str:
result = _dsml_arg_converter(raw_args, partial)
if not self._tools:
return result
func_name = next((s.name for s in self._tool_slots if s.args == raw_args), None)
return _unwrap_wrapper_args(result, self._tools, func_name)
+237
View File
@@ -0,0 +1,237 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""DeepSeek V4 parser: ``<think>``/``</think>``
reasoning plus DSML tool calls in a single state machine.
DeepSeek V4 output format::
<think>
...reasoning...
</think>
<DSMLtool_calls>
<DSMLinvoke name="func_name">
<DSMLparameter name="location" string="true">杭州</DSMLparameter>
<DSMLparameter name="count" string="false">5</DSMLparameter>
</DSMLinvoke>
</DSMLtool_calls>
"""
from __future__ import annotations
import contextlib
import functools
import json
from typing import TYPE_CHECKING
import regex as re
from vllm.parser.engine.events import EventType
from vllm.parser.engine.parser_engine import ParserEngine
from vllm.parser.engine.parser_engine_config import (
ParserEngineConfig,
ParserState,
Transition,
)
from vllm.tool_parsers.utils import find_tool_properties
if TYPE_CHECKING:
from vllm.tokenizers import TokenizerLike
from vllm.tool_parsers.abstract_tool_parser import Tool
_DSML = "DSML"
DSML_THINK_START = "<think>"
DSML_THINK_END = "</think>"
DSML_TOOL_START = f"<{_DSML}tool_calls>"
DSML_TOOL_END = f"</{_DSML}tool_calls>"
DSML_INVOKE_PREFIX = f'<{_DSML}invoke name="'
DSML_INVOKE_NAME_END = '">'
DSML_INVOKE_END = f"</{_DSML}invoke>"
DSML_PARAM_CLOSE = f"</{_DSML}parameter>"
_ESCAPED_DSML = re.escape(_DSML)
_PARAM_RE = re.compile(
rf'<{_ESCAPED_DSML}parameter\s+name="([^"]+)"\s+string="(true|false)">'
rf"(.*?)</{_ESCAPED_DSML}parameter>",
re.DOTALL,
)
_PARTIAL_PARAM_RE = re.compile(
rf'<{_ESCAPED_DSML}parameter\s+name="([^"]+)"\s+string="(true|false)">'
rf"(.*)$",
re.DOTALL,
)
def _dsml_arg_converter(raw_args: str, partial: bool) -> str:
params: dict[str, object] = {}
last_end = 0
for m in _PARAM_RE.finditer(raw_args):
name, is_str, value = m.group(1), m.group(2), m.group(3)
if is_str == "true":
params[name] = value
else:
try:
params[name] = json.loads(value)
except (json.JSONDecodeError, ValueError):
params[name] = value
last_end = m.end()
if partial:
pm = _PARTIAL_PARAM_RE.search(raw_args, last_end)
if pm:
name, is_str, value = pm.group(1), pm.group(2), pm.group(3)
if is_str == "true":
params[name] = value
else:
with contextlib.suppress(json.JSONDecodeError, ValueError):
params[name] = json.loads(value)
return json.dumps(params, ensure_ascii=False)
def _unwrap_wrapper_args(
args_json: str,
tools: list[Tool] | None,
func_name: str | None,
) -> str:
if not tools or not func_name:
return args_json
try:
args = json.loads(args_json)
except (json.JSONDecodeError, ValueError):
return args_json
if not isinstance(args, dict):
return args_json
properties = find_tool_properties(tools, func_name)
if not properties:
return args_json
allowed = set(properties.keys())
for wrapper in ("arguments", "input"):
if set(args.keys()) != {wrapper} or wrapper in allowed:
continue
inner = args[wrapper]
if isinstance(inner, str):
try:
inner = json.loads(inner)
except json.JSONDecodeError:
return args_json
if isinstance(inner, dict) and set(inner.keys()).issubset(allowed):
return json.dumps(inner, ensure_ascii=False)
return args_json
@functools.cache
def deepseek_v4_config(thinking: bool = False) -> ParserEngineConfig:
return ParserEngineConfig(
name="deepseek_v4",
initial_state=ParserState.REASONING if thinking else ParserState.CONTENT,
terminals={
"THINK_START": DSML_THINK_START,
"THINK_END": DSML_THINK_END,
"TOOL_START": DSML_TOOL_START,
"TOOL_END": DSML_TOOL_END,
"INVOKE_PREFIX": DSML_INVOKE_PREFIX,
"INVOKE_NAME_END": DSML_INVOKE_NAME_END,
"INVOKE_END": DSML_INVOKE_END,
"PARAM_CLOSE": DSML_PARAM_CLOSE,
},
token_id_terminals={
"THINK_START": DSML_THINK_START,
"THINK_END": DSML_THINK_END,
"TOOL_START": DSML_TOOL_START,
"TOOL_END": DSML_TOOL_END,
},
transitions={
(ParserState.CONTENT, "THINK_START"): Transition(
ParserState.REASONING,
(EventType.REASONING_START,),
),
# Absorb a bare </think> with no prior <think>
(ParserState.CONTENT, "THINK_END"): Transition(
ParserState.CONTENT,
(),
),
# Absorb a duplicate <think> while already reasoning
(ParserState.REASONING, "THINK_START"): Transition(
ParserState.REASONING,
(),
),
(ParserState.REASONING, "THINK_END"): Transition(
ParserState.CONTENT,
(EventType.REASONING_END,),
),
# Tool call beginning while still inside <think>
(ParserState.REASONING, "TOOL_START"): Transition(
ParserState.TOOL_PREAMBLE,
(EventType.REASONING_END,),
),
(ParserState.CONTENT, "TOOL_START"): Transition(
ParserState.TOOL_PREAMBLE,
(),
),
(ParserState.TOOL_PREAMBLE, "INVOKE_PREFIX"): Transition(
ParserState.TOOL_NAME,
(EventType.TOOL_CALL_START,),
),
(ParserState.TOOL_NAME, "INVOKE_NAME_END"): Transition(
ParserState.TOOL_ARGS,
(),
),
(ParserState.TOOL_ARGS, "INVOKE_END"): Transition(
ParserState.TOOL_BETWEEN,
(EventType.TOOL_CALL_END,),
),
(ParserState.TOOL_ARGS, "TOOL_END"): Transition(
ParserState.CONTENT,
(EventType.TOOL_CALL_END,),
),
# Parallel tool calls
(ParserState.TOOL_BETWEEN, "INVOKE_PREFIX"): Transition(
ParserState.TOOL_NAME,
(EventType.TOOL_CALL_START,),
),
(ParserState.TOOL_BETWEEN, "TOOL_END"): Transition(
ParserState.CONTENT,
(),
),
},
content_events={
ParserState.CONTENT: EventType.TEXT_CHUNK,
ParserState.REASONING: EventType.REASONING_CHUNK,
ParserState.TOOL_NAME: EventType.TOOL_NAME,
ParserState.TOOL_ARGS: EventType.ARG_VALUE_CHUNK,
},
arg_converter=_dsml_arg_converter,
arg_structural_chars=frozenset(">"),
strip_content_whitespace_with_tools=False,
tool_args_json=False,
)
class DeepSeekV4Parser(ParserEngine):
def __init__(
self,
tokenizer: TokenizerLike,
tools: list[Tool] | None = None,
**kwargs,
) -> None:
chat_kwargs = kwargs.pop("chat_template_kwargs", None) or {}
thinking = (
bool(chat_kwargs.get("thinking") or chat_kwargs.get("enable_thinking"))
and chat_kwargs.get("reasoning_effort") != "none"
)
super().__init__(
tokenizer,
tools,
parser_engine_config=deepseek_v4_config(thinking=thinking),
**kwargs,
)
self._arg_converter = self._convert_args
def _convert_args(self, raw_args: str, partial: bool) -> str:
result = _dsml_arg_converter(raw_args, partial)
if not self._tools:
return result
func_name = next((s.name for s in self._tool_slots if s.args == raw_args), None)
return _unwrap_wrapper_args(result, self._tools, func_name)
+2 -1
View File
@@ -111,7 +111,8 @@ class ParserEngineReasoningAdapter(ReasoningParser):
return self._parser_engine.reasoning_ended
def finish_streaming(self) -> DeltaMessage | None:
return self._parser_engine.finish_streaming()
with self._skip_tool_parsing():
return self._parser_engine.finish_streaming()
def get_streaming_fallback_content(
self,
+7 -3
View File
@@ -108,7 +108,11 @@ class ParserEngine(Parser):
parser_engine_config, tokenizer, vocab=self.vocab
)
self._reasoning_ended: bool = False
self._has_reasoning = (
"THINK_END" in parser_engine_config.token_id_terminals
or parser_engine_config.initial_state == ParserState.REASONING
)
self._reasoning_ended: bool = not self._has_reasoning
self._streaming_initialized: bool = False
self._prompt_streaming_prepared: bool = False
@@ -188,7 +192,7 @@ class ParserEngine(Parser):
def _reset(self, initial_state: ParserState | None = None) -> None:
self._engine.reset(initial_state=initial_state)
self._reasoning_ended = False
self._reasoning_ended = not self._has_reasoning
self._tool_slots.clear()
self._deferred_content = ""
self._deferred_reasoning = ""
@@ -255,7 +259,7 @@ class ParserEngine(Parser):
types = extract_types_from_schema(schema)
as_str = json.dumps(value, ensure_ascii=False)
coerced = coerce_to_schema_type(as_str, types)
if coerced != value:
if type(coerced) is not type(value) or coerced != value:
return coerced, True
return value, False
+12
View File
@@ -7,6 +7,8 @@ names so that :class:`ReasoningParserManager` and
:class:`ToolParserManager` can load them lazily.
"""
from vllm.parser.deepseek_v4 import DeepSeekV4Parser
from vllm.parser.deepseek_v32 import DeepSeekV32Parser
from vllm.parser.engine.adapters import make_adapters
from vllm.parser.gemma4 import Gemma4Parser
from vllm.parser.glm47_moe import Glm47MoeParser
@@ -16,6 +18,16 @@ from vllm.parser.nemotron_v3 import NemotronV3Parser
from vllm.parser.qwen3 import Qwen3Parser
from vllm.parser.seed_oss import SeedOssParser
(
DeepSeekV32ParserReasoningAdapter,
DeepSeekV32ParserToolAdapter,
) = make_adapters(DeepSeekV32Parser)
(
DeepSeekV4ParserReasoningAdapter,
DeepSeekV4ParserToolAdapter,
) = make_adapters(DeepSeekV4Parser)
(
MinimaxM2ParserReasoningAdapter,
MinimaxM2ParserToolAdapter,
@@ -22,14 +22,13 @@ from vllm.parser.engine.parser_engine_config import (
Transition,
)
from vllm.parser.engine.token_id_scanner import (
DROP_TERMINAL,
LexerInput,
PreLexedTerminal,
TextChunk,
TokenIDScanner,
)
DROP_TERMINAL = "__DROP__"
@dataclass(slots=True)
class _DropInfo:
+11 -1
View File
@@ -8,6 +8,8 @@ from __future__ import annotations
from collections.abc import Sequence
from dataclasses import dataclass
DROP_TERMINAL = "__DROP__"
@dataclass(slots=True)
class TextChunk:
@@ -277,7 +279,15 @@ class TokenIDScanner:
consumed = pos + len(anchor.text)
else:
has_later_valid = any(p >= 0 for p in positions[i + 1 :])
if not has_later_valid and consumed < len(delta_text):
# DROP anchors (EOS, etc.) may have text that never
# arrives in delta_text (stripped by detokenizer).
# Don't defer remaining content waiting for text
# that will never come.
if (
not has_later_valid
and consumed < len(delta_text)
and anchor.terminal != DROP_TERMINAL
):
self._deferred_post_text += delta_text[consumed:]
consumed = len(delta_text)
self._deferred_terminals.append(anchor)
+46 -29
View File
@@ -9,7 +9,7 @@ from dataclasses import dataclass
from enum import Enum, auto
from typing import TYPE_CHECKING, NamedTuple
from openai_harmony import HarmonyError
from openai_harmony import HarmonyError, Message, Role
from vllm.entrypoints.chat_utils import make_tool_call_id
from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest
@@ -91,6 +91,9 @@ class HarmonyParser(DelegatingParser):
self._next_tool_call_index = 0
self._num_processed_messages = 0
# For error recovery
self._current_message_tokens: list[int] = []
@property
def _harmony_parser(self) -> StreamableParser:
"""Lazily initializes the Harmony parser."""
@@ -107,32 +110,48 @@ class HarmonyParser(DelegatingParser):
self._num_processed_messages += 1
return msg
def flush(self) -> Segment | None:
def flush(self) -> list[Segment]:
segments: list[Segment] = []
try:
self._harmony_parser.process_eos()
msg = self._poll_completed_message()
except HarmonyError:
logger.warning(
"Harmony parser ended in a non-terminal state; returning the "
"raw unparsed output. This usually indicates a malformed "
"assistant turn, e.g. a 'final' channel missing the "
"<|message|> delimiter."
"recovered raw output."
)
raise
finally:
# Reset to the initial assistant-parser state for the next turn.
self._parser = None
self._num_processed_messages = 0
final_channel = "final"
text = self.model_tokenizer.decode(self._current_message_tokens)
segments.append(
Segment(
channel=final_channel,
recipient=None,
delta=text,
completed_message=None,
)
)
msg = Message.from_role_and_content(Role.ASSISTANT, text).with_channel(
final_channel
)
# Reset to the initial assistant-parser state for the next turn.
self._parser = None
self._num_processed_messages = 0
self._current_message_tokens.clear()
if msg is None:
return None
return segments
return Segment(
channel=msg.channel,
recipient=msg.recipient,
delta="",
completed_message=msg,
segments.append(
Segment(
channel=msg.channel,
recipient=msg.recipient,
delta="",
completed_message=msg,
)
)
return segments
def parse(
self,
@@ -147,12 +166,9 @@ class HarmonyParser(DelegatingParser):
Callers must decide whether to surface them.
"""
result = self.process_chunk(model_output_token_ids)
try:
flushed_segment = self.flush()
except HarmonyError:
return None, model_output, None
if flushed_segment is not None:
result.segments.append(flushed_segment)
flushed_segments = self.flush()
if flushed_segments:
result.segments.extend(flushed_segments)
reasoning_parts: list[str] = []
content_parts: list[str] = []
@@ -209,13 +225,9 @@ class HarmonyParser(DelegatingParser):
)
result = self.process_chunk(delta_token_ids)
if finished:
try:
flushed_segment = self.flush()
except HarmonyError:
self._next_tool_call_index = 0
return DeltaMessage(content=delta_text)
if flushed_segment is not None:
result.segments.append(flushed_segment)
flushed_segments = self.flush()
if flushed_segments:
result.segments.extend(flushed_segments)
combined_content = ""
combined_reasoning = ""
tool_messages: list[DeltaToolCall] = []
@@ -298,6 +310,11 @@ class HarmonyParser(DelegatingParser):
delta = self._harmony_parser.last_content_delta or ""
completed_message = self._poll_completed_message()
if completed_message is not None:
self._current_message_tokens.clear()
else:
self._current_message_tokens.append(token_id)
if channel == "analysis" or (
channel == "commentary" and recipient is not None
):
+117
View File
@@ -629,6 +629,123 @@ class Platform:
if model_config.is_hybrid:
cls._align_hybrid_block_size(vllm_config, backend_cls)
# Phase 3: Align block/page sizes when multiple KV dtypes share the
# block pool (e.g. nvfp4 primary + unquantized skip layers).
# May override the user's --block-size.
if cache_config.kv_cache_dtype_skip_layers:
cls._align_heterogeneous_kv_block_size(vllm_config, backend_cls)
@classmethod
def _align_heterogeneous_kv_block_size(
cls,
vllm_config: "VllmConfig",
backend_cls: "type[AttentionBackend]",
) -> None:
"""Align block size when several KV dtypes share one block pool.
A quantized primary (e.g. nvfp4) shares the block pool with one or more
higher-precision "padded specs" (skip layers today; the first/last-N
sibling in the future). A padded spec's per-token page is larger than
the primary's *and not an integer multiple of it*, so the trivial
``unify_kv_cache_spec_page_size`` cannot reconcile them. We do it here
instead, before the specs are built:
1. Bump the primary ``block_size`` (kernel-aligned) until the primary
page is large enough to cover the largest padded-spec page.
2. Record that shared page in each padded spec's ``*_page_size_padded``
hint, so it pads up to the shared page.
``unify`` then sees equal pages and stays trivial.
To add a padded-spec type: append its per-token page to ``padded_pages``
and set its ``*_page_size_padded`` hint below.
"""
from vllm.config.vllm import set_current_vllm_config
from vllm.utils.math_utils import cdiv
from vllm.utils.torch_utils import STR_DTYPE_TO_TORCH_DTYPE
from vllm.v1.attention.backend import MultipleOf
from vllm.v1.kv_cache_interface import FullAttentionSpec, get_kv_quant_mode
cache_config = vllm_config.cache_config
model_config = vllm_config.model_config
parallel_config = vllm_config.parallel_config
if not model_config:
return
def per_token_page_bytes(dtype: "torch.dtype", cache_dtype: str) -> int:
"""Bytes one token occupies in one layer, for the given dtype."""
return FullAttentionSpec(
block_size=1,
num_kv_heads=model_config.get_num_kv_heads(parallel_config),
head_size=model_config.get_head_size(),
dtype=dtype,
kv_quant_mode=get_kv_quant_mode(cache_dtype),
).page_size_bytes
primary_dtype = (
STR_DTYPE_TO_TORCH_DTYPE[cache_config.cache_dtype]
if cache_config.cache_dtype != "auto"
else model_config.dtype
)
primary_page = per_token_page_bytes(primary_dtype, cache_config.cache_dtype)
# Per-token page of every higher-precision padded spec sharing the pool.
padded_pages: list[int] = []
if cache_config.kv_cache_dtype_skip_layers:
padded_pages.append(per_token_page_bytes(model_config.dtype, "auto"))
# To add the first/last-N sibling:
# padded_pages.append(per_token_page_bytes(<sibling_dtype>, "auto"))
if not padded_pages:
return
largest_padded_page = max(padded_pages)
assert largest_padded_page >= primary_page, (
f"padded-spec per-token page ({largest_padded_page}B) < primary "
f"({primary_page}B); a higher-precision padded spec must not be "
"smaller than the quantized primary."
)
if largest_padded_page == primary_page:
# Pages already match per token; ``unify`` reconciles the differing
# block sizes by integer scaling, so no bump or padding is needed.
return
# Smallest block the kernel supports, and the granularity the primary
# block is rounded up to (never below the already-chosen block_size).
with set_current_vllm_config(vllm_config):
supported = backend_cls.get_supported_kernel_block_sizes()
smallest_kernel_block = min(
s.base if isinstance(s, MultipleOf) else s for s in supported
)
block_alignment = max(smallest_kernel_block, cache_config.block_size)
# Bytes one padded-spec page spans at its own smallest kernel block;
# also cover any mamba page a hybrid model already padded.
required_page = max(
largest_padded_page * smallest_kernel_block,
cache_config.mamba_page_size_padded or 0,
)
# Smallest kernel-aligned primary block whose page covers required_page.
primary_block_size = block_alignment * cdiv(
required_page, block_alignment * primary_page
)
if cache_config.block_size < primary_block_size:
cache_config.block_size = primary_block_size
logger.info(
"Setting attention block size to %d tokens so the quantized "
"primary KV page covers the higher-precision padded-spec page.",
primary_block_size,
)
# The shared page that every padded spec (and mamba) pads up to.
shared_page = cache_config.block_size * primary_page
if cache_config.kv_cache_dtype_skip_layers:
cache_config.skip_page_size_padded = shared_page
# To add the first/last-N sibling:
# cache_config.sibling_page_size_padded = shared_page
if cache_config.mamba_page_size_padded is not None:
cache_config.mamba_page_size_padded = shared_page
@classmethod
def _align_hybrid_block_size(
cls,
+2 -2
View File
@@ -29,8 +29,8 @@ _REASONING_PARSERS_TO_REGISTER = {
"DeepSeekV3ReasoningParser",
),
"deepseek_v4": (
"deepseek_v3_reasoning_parser",
"DeepSeekV3ReasoningParser",
"deepseek_v4_engine_reasoning_parser",
"DeepSeekV4ParserReasoningAdapter",
),
"poolside_v1": (
"poolside_v1_reasoning_parser",
@@ -0,0 +1,6 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from vllm.parser.engine.registered_adapters import DeepSeekV4ParserReasoningAdapter
__all__ = ["DeepSeekV4ParserReasoningAdapter"]
+4 -4
View File
@@ -31,12 +31,12 @@ _TOOL_PARSERS_TO_REGISTER = {
"DeepSeekV31ToolParser",
),
"deepseek_v32": (
"deepseekv32_tool_parser",
"DeepSeekV32ToolParser",
"deepseekv32_engine_tool_parser",
"DeepSeekV32EngineToolParser",
),
"deepseek_v4": (
"deepseekv4_tool_parser",
"DeepSeekV4ToolParser",
"deepseekv4_engine_tool_parser",
"DeepSeekV4EngineToolParser",
),
"cohere_command3": (
"cohere_command_tool_parser",
@@ -0,0 +1,8 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from vllm.parser.engine.registered_adapters import DeepSeekV32ParserToolAdapter
class DeepSeekV32EngineToolParser(DeepSeekV32ParserToolAdapter): # type: ignore[valid-type, misc]
structural_tag_model = "deepseek_v3_2"
@@ -1,562 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
import json
import uuid
from collections.abc import Sequence
from typing import Any, Literal
import regex as re
from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionRequest,
)
from vllm.entrypoints.openai.engine.protocol import (
DeltaFunctionCall,
DeltaMessage,
DeltaToolCall,
ExtractedToolCallInformation,
FunctionCall,
ToolCall,
)
from vllm.entrypoints.openai.responses.protocol import ResponsesRequest
from vllm.logger import init_logger
from vllm.tokenizers import TokenizerLike
from vllm.tool_parsers.abstract_tool_parser import (
Tool,
ToolParser,
)
from vllm.tool_parsers.utils import (
coerce_to_schema_type,
extract_types_from_schema,
find_tool_properties,
partial_tag_overlap,
)
logger = init_logger(__name__)
class DeepSeekV32ToolParser(ToolParser):
"""
example tool call content:
<DSMLfunction_calls>
<DSMLinvoke name="get_weather">
<DSMLparameter name="location" string="true">杭州</DSMLparameter>
<DSMLparameter name="date" string="true">2024-01-16</DSMLparameter>
</DSMLinvoke>
<DSMLinvoke name="get_weather">
<DSMLparameter name="location" string="true">北京</DSMLparameter>
<DSMLparameter name="date" string="true">2024-01-16</DSMLparameter>
</DSMLinvoke>
</DSMLfunction_calls>
"""
tool_call_start_token: str = "<DSMLfunction_calls>"
tool_call_end_token: str = "</DSMLfunction_calls>"
structural_tag_model = "deepseek_v3_2"
def __init__(self, tokenizer: TokenizerLike, tools: list[Tool] | None = None):
super().__init__(tokenizer, tools)
self.prev_tool_call_arr: list[dict] = []
# Streaming state
self.current_tool_index: int = 0
self._sent_content_idx: int = 0
self._buffer: str = ""
self._in_tool_calls: bool = False
self._active_tool_index: int | None = None
self._active_tool_name: str | None = None
self._active_param_name: str | None = None
self._active_param_string_attr: str | None = None
self._active_param_mode: str | None = None
self._active_param_parts: list[str] = []
self._args_started: list[bool] = []
# Regex patterns for complete parsing
self.tool_call_complete_regex = re.compile(
re.escape(self.tool_call_start_token)
+ r"(.*?)"
+ re.escape(self.tool_call_end_token),
re.DOTALL,
)
self.invoke_complete_regex = re.compile(
r'<DSMLinvoke\s+name="([^"]+)"\s*>(.*?)</DSMLinvoke>', re.DOTALL
)
self.parameter_complete_regex = re.compile(
r'<DSMLparameter\s+name="([^"]+)"\s+string="(true|false)"\s*>(.*?)</DSMLparameter>',
re.DOTALL,
)
self.invoke_start_regex = re.compile(r'<DSMLinvoke\s+name="([^"]+)"\s*>')
self.parameter_start_regex = re.compile(
r'<DSMLparameter\s+name="([^"]+)"\s+string="(true|false)"\s*>'
)
if not self.model_tokenizer:
raise ValueError(
"The model tokenizer must be passed to the ToolParser "
"constructor during construction."
)
logger.debug(
"vLLM Successfully import tool parser %s !", self.__class__.__name__
)
def adjust_request(
self, request: ChatCompletionRequest | ResponsesRequest
) -> ChatCompletionRequest | ResponsesRequest:
request = super().adjust_request(request)
if request.tools and request.tool_choice != "none":
# Ensure tool call tokens
# (e.g. <DSMLfunction_calls>, </DSMLfunction_calls>)
# are not skippedduring decoding.
# Even though they are not marked as special tokens,
# setting skip_special_tokens=False ensures proper handling in
# transformers 5.x where decoding behavior may have changed.
request.skip_special_tokens = False
return request
def _generate_tool_call_id(self) -> str:
"""Generate a unique tool call ID."""
return f"call_{uuid.uuid4().hex[:24]}"
def _parse_invoke_params(self, invoke_str: str) -> dict[str, tuple[str, str]]:
param_dict: dict[str, tuple[str, str]] = {}
for param_name, string_attr, param_val in self.parameter_complete_regex.findall(
invoke_str
):
param_dict[param_name] = (param_val, string_attr)
return param_dict
@staticmethod
def _repair_param_dict(
param_dict: dict[str, Any],
param_config: dict[str, Any],
) -> dict[str, Any]:
"""Unwrap single 'arguments' / 'input' wrappers when the wrapper
is not part of the requested tool schema and the wrapped object
matches the schema fields."""
allowed = set(param_config.keys())
for wrapper in ("arguments", "input"):
if set(param_dict.keys()) != {wrapper} or wrapper in allowed:
continue
inner = param_dict[wrapper]
if isinstance(inner, str):
try:
inner = json.loads(inner)
except json.JSONDecodeError:
return param_dict
if isinstance(inner, dict) and set(inner.keys()).issubset(allowed):
return inner
return param_dict
def _convert_params_with_schema(
self,
function_name: str,
param_dict: dict[str, tuple[str, str]],
) -> dict[str, Any]:
"""Convert raw string param values using the tool schema types."""
param_config = find_tool_properties(self.tools, function_name)
converted: dict[str, Any] = {}
for name, (value, string_attr) in param_dict.items():
if string_attr == "true":
converted[name] = value
continue
param_types = extract_types_from_schema(param_config.get(name, {}))
converted[name] = coerce_to_schema_type(value, param_types)
return self._repair_param_dict(converted, param_config)
def _get_param_config(self, function_name: str | None) -> dict[str, Any]:
if not function_name or not self.tools:
return {}
return find_tool_properties(self.tools, function_name)
@staticmethod
def _json_escape_string_content(text: str) -> str:
return json.dumps(text, ensure_ascii=False)[1:-1]
def extract_tool_calls(
self,
model_output: str,
request: ChatCompletionRequest,
) -> ExtractedToolCallInformation:
"""Extract tool calls from complete model output (non-streaming)."""
# Quick check
if self.tool_call_start_token not in model_output:
return ExtractedToolCallInformation(
tools_called=False, tool_calls=[], content=model_output
)
try:
tool_calls = []
# Find all complete tool_call blocks
for tool_call_match in self.tool_call_complete_regex.findall(model_output):
# Find all invokes within this tool_call
for invoke_name, invoke_content in self.invoke_complete_regex.findall(
tool_call_match
):
param_dict = self._parse_invoke_params(invoke_content)
params = self._convert_params_with_schema(invoke_name, param_dict)
tool_calls.append(
ToolCall(
type="function",
function=FunctionCall(
name=invoke_name,
arguments=json.dumps(params, ensure_ascii=False),
),
)
)
if not tool_calls:
return ExtractedToolCallInformation(
tools_called=False, tool_calls=[], content=model_output
)
# Extract content before first tool call
first_tool_idx = model_output.find(self.tool_call_start_token)
content = model_output[:first_tool_idx] if first_tool_idx > 0 else None
return ExtractedToolCallInformation(
tools_called=True, tool_calls=tool_calls, content=content
)
except Exception:
logger.exception("Error extracting tool calls")
return ExtractedToolCallInformation(
tools_called=False, tool_calls=[], content=model_output
)
def _reset_streaming_state(self):
"""Reset all streaming state."""
self.current_tool_index = 0
self._sent_content_idx = 0
self._buffer = ""
self._in_tool_calls = False
self._active_tool_index = None
self._active_tool_name = None
self._active_param_name = None
self._active_param_string_attr = None
self._active_param_mode = None
self._active_param_parts.clear()
self.prev_tool_call_arr.clear()
self.streamed_args_for_tool.clear()
self._args_started.clear()
def _add_tool_call_delta(
self,
tool_call_deltas: dict[int, DeltaToolCall],
index: int,
*,
call_id: str | None = None,
call_type: Literal["function"] | None = None,
name: str | None = None,
arguments: str | None = None,
) -> None:
if arguments:
self.streamed_args_for_tool[index] += arguments
if index not in tool_call_deltas:
tool_call_deltas[index] = DeltaToolCall(
index=index,
id=call_id,
type=call_type,
function=DeltaFunctionCall(name=name, arguments=arguments),
)
return
delta = tool_call_deltas[index]
if call_id is not None:
delta.id = call_id
if call_type is not None:
delta.type = call_type
if delta.function is None:
delta.function = DeltaFunctionCall()
if name is not None:
delta.function.name = name
if arguments is not None:
delta.function.arguments = (delta.function.arguments or "") + arguments
def _begin_streaming_tool_call(
self,
name: str,
tool_call_deltas: dict[int, DeltaToolCall],
) -> None:
index = self.current_tool_index
self.current_tool_index += 1
self._active_tool_index = index
self._active_tool_name = name
self.prev_tool_call_arr.append({"name": name, "arguments": {}})
self.streamed_args_for_tool.append("")
self._args_started.append(False)
self._add_tool_call_delta(
tool_call_deltas,
index,
call_id=self._generate_tool_call_id(),
call_type="function",
name=name,
arguments="",
)
def _append_param_prefix(
self,
tool_call_deltas: dict[int, DeltaToolCall],
index: int,
key: str,
*,
as_string: bool,
) -> None:
prefix = "{" if not self._args_started[index] else ","
self._args_started[index] = True
arguments = prefix + json.dumps(key, ensure_ascii=False) + ":"
if as_string:
arguments += '"'
self._add_tool_call_delta(tool_call_deltas, index, arguments=arguments)
def _append_json_param_value(
self,
tool_call_deltas: dict[int, DeltaToolCall],
index: int,
key: str,
value: Any,
) -> None:
self._append_param_prefix(tool_call_deltas, index, key, as_string=False)
self._add_tool_call_delta(
tool_call_deltas,
index,
arguments=json.dumps(value, ensure_ascii=False),
)
def _param_types_for_name(self, name: str) -> list[str]:
param_config = self._get_param_config(self._active_tool_name)
if name in param_config and isinstance(param_config[name], dict):
return extract_types_from_schema(param_config[name])
return ["string"]
@staticmethod
def _can_stream_raw_param(param_types: list[str]) -> bool:
# Scalars and unions need the complete value so streaming and
# non-streaming share the same coercion fallback behavior.
return set(param_types).issubset({"object", "array"})
def _should_buffer_wrapper_param(self, name: str) -> bool:
if (
self._active_tool_index is None
or self._args_started[self._active_tool_index]
):
return False
param_config = self._get_param_config(self._active_tool_name)
return bool(
param_config and name in ("arguments", "input") and name not in param_config
)
def _finish_buffered_param(
self,
tool_call_deltas: dict[int, DeltaToolCall],
index: int,
) -> None:
assert self._active_param_name is not None
assert self._active_param_string_attr is not None
raw_value = "".join(self._active_param_parts)
converted = self._convert_params_with_schema(
self._active_tool_name or "",
{self._active_param_name: (raw_value, self._active_param_string_attr)},
)
for key, value in converted.items():
self._append_json_param_value(tool_call_deltas, index, key, value)
def _close_streaming_tool_call(
self,
tool_call_deltas: dict[int, DeltaToolCall],
) -> None:
index = self._active_tool_index
if index is None:
return
suffix = "}" if self._args_started[index] else "{}"
self._add_tool_call_delta(tool_call_deltas, index, arguments=suffix)
try:
self.prev_tool_call_arr[index] = {
"name": self._active_tool_name,
"arguments": json.loads(self.streamed_args_for_tool[index]),
}
except (json.JSONDecodeError, IndexError):
logger.exception("Failed to finalize DeepSeek DSML streaming tool call")
self._active_tool_index = None
self._active_tool_name = None
self._active_param_name = None
self._active_param_string_attr = None
self._active_param_mode = None
self._active_param_parts.clear()
def _process_streaming_buffer(
self,
content_parts: list[str],
tool_call_deltas: dict[int, DeltaToolCall],
) -> None:
parameter_end_token = "</DSMLparameter>"
invoke_end_token = "</DSMLinvoke>"
while True:
if not self._in_tool_calls:
start_idx = self._buffer.find(self.tool_call_start_token)
if start_idx == -1:
overlap = partial_tag_overlap(
self._buffer, self.tool_call_start_token
)
sendable_idx = len(self._buffer) - overlap
if sendable_idx > 0:
content_parts.append(self._buffer[:sendable_idx])
self._buffer = self._buffer[sendable_idx:]
return
if start_idx > 0:
content_parts.append(self._buffer[:start_idx])
self._buffer = self._buffer[start_idx:]
continue
self._buffer = self._buffer[len(self.tool_call_start_token) :]
self._in_tool_calls = True
continue
if self._active_tool_index is None:
stripped_len = len(self._buffer) - len(self._buffer.lstrip())
if stripped_len:
self._buffer = self._buffer[stripped_len:]
continue
if self._buffer.startswith(self.tool_call_end_token):
self._buffer = self._buffer[len(self.tool_call_end_token) :]
self._in_tool_calls = False
continue
match = self.invoke_start_regex.match(self._buffer)
if match is None:
return
self._buffer = self._buffer[match.end() :]
self._begin_streaming_tool_call(match.group(1), tool_call_deltas)
continue
index = self._active_tool_index
if self._active_param_mode is not None:
end_pos = self._buffer.find(parameter_end_token)
if end_pos != -1:
raw_content = self._buffer[:end_pos]
self._buffer = self._buffer[end_pos + len(parameter_end_token) :]
if self._active_param_mode in ("wrapper", "buffered"):
self._active_param_parts.append(raw_content)
self._finish_buffered_param(tool_call_deltas, index)
elif self._active_param_mode == "string":
arguments = self._json_escape_string_content(raw_content) + '"'
self._add_tool_call_delta(
tool_call_deltas, index, arguments=arguments
)
else:
self._add_tool_call_delta(
tool_call_deltas, index, arguments=raw_content
)
self._active_param_name = None
self._active_param_string_attr = None
self._active_param_mode = None
self._active_param_parts.clear()
continue
overlap = partial_tag_overlap(self._buffer, parameter_end_token)
safe_len = len(self._buffer) - overlap
if safe_len > 0:
raw_content = self._buffer[:safe_len]
self._buffer = self._buffer[safe_len:]
if self._active_param_mode in ("wrapper", "buffered"):
self._active_param_parts.append(raw_content)
elif self._active_param_mode == "string":
self._add_tool_call_delta(
tool_call_deltas,
index,
arguments=self._json_escape_string_content(raw_content),
)
else:
self._add_tool_call_delta(
tool_call_deltas, index, arguments=raw_content
)
return
stripped_len = len(self._buffer) - len(self._buffer.lstrip())
if stripped_len:
self._buffer = self._buffer[stripped_len:]
continue
if self._buffer.startswith(invoke_end_token):
self._buffer = self._buffer[len(invoke_end_token) :]
self._close_streaming_tool_call(tool_call_deltas)
continue
match = self.parameter_start_regex.match(self._buffer)
if match is None:
return
self._buffer = self._buffer[match.end() :]
name = match.group(1)
string_attr = match.group(2)
self._active_param_name = name
self._active_param_string_attr = string_attr
if self._should_buffer_wrapper_param(name):
self._active_param_mode = "wrapper"
continue
if string_attr == "true":
self._append_param_prefix(tool_call_deltas, index, name, as_string=True)
self._active_param_mode = "string"
continue
param_types = self._param_types_for_name(name)
if not self._can_stream_raw_param(param_types):
self._active_param_mode = "buffered"
continue
self._append_param_prefix(tool_call_deltas, index, name, as_string=False)
self._active_param_mode = "raw"
def extract_tool_calls_streaming(
self,
previous_text: str,
current_text: str,
delta_text: str,
previous_token_ids: Sequence[int], # pylint: disable=unused-argument
current_token_ids: Sequence[int], # pylint: disable=unused-argument
delta_token_ids: Sequence[int],
request: ChatCompletionRequest,
) -> DeltaMessage | None:
"""Extract tool calls from streaming model output.
Buffers DSML markup while streaming tool-call metadata and argument
JSON fragments as soon as they are complete enough to be valid deltas.
"""
# First chunk of a new stream — reset state from prior request.
if not previous_text:
self._reset_streaming_state()
self._buffer += delta_text
content_parts: list[str] = []
tool_call_deltas: dict[int, DeltaToolCall] = {}
self._process_streaming_buffer(content_parts, tool_call_deltas)
if content_parts or tool_call_deltas:
content = "".join(content_parts) or None
return DeltaMessage(
content=content, tool_calls=list(tool_call_deltas.values())
)
# Empty delta with token ids means EOS or closing tag; return
# non-None so the serving framework can finalize finish_reason.
if not delta_text and delta_token_ids and self.prev_tool_call_arr:
return DeltaMessage(content="")
return None
@@ -0,0 +1,8 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from vllm.parser.engine.registered_adapters import DeepSeekV4ParserToolAdapter
class DeepSeekV4EngineToolParser(DeepSeekV4ParserToolAdapter): # type: ignore[valid-type, misc]
structural_tag_model = "deepseek_v4"
@@ -1,17 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from vllm.tool_parsers.deepseekv32_tool_parser import DeepSeekV32ToolParser
class DeepSeekV4ToolParser(DeepSeekV32ToolParser):
"""
DeepSeek V4 DSML tool parser.
V4 keeps the V3.2 DSML invoke/parameter grammar, but wraps tool calls in
``<DSMLtool_calls>`` instead of ``<DSMLfunction_calls>``.
"""
tool_call_start_token: str = "<DSMLtool_calls>"
tool_call_end_token: str = "</DSMLtool_calls>"
structural_tag_model = "deepseek_v4"
+5 -4
View File
@@ -229,10 +229,11 @@ def get_model_path(model: str | Path, revision: str | None = None):
if os.path.exists(model):
return model
assert huggingface_hub.constants.HF_HUB_OFFLINE
common_kwargs = {
"local_files_only": huggingface_hub.constants.HF_HUB_OFFLINE,
"revision": revision,
}
common_kwargs = dict(
local_files_only=huggingface_hub.constants.HF_HUB_OFFLINE,
ignore_patterns="*",
revision=revision,
)
if envs.VLLM_USE_MODELSCOPE:
from modelscope.hub.snapshot_download import snapshot_download
+1 -1
View File
@@ -141,7 +141,7 @@ class CPUAttentionMetadataBuilder(AttentionMetadataBuilder[CPUAttentionMetadata]
self.vllm_config = vllm_config
parallel_config = vllm_config.parallel_config
self.num_kv_heads = vllm_config.model_config.get_num_kv_heads(parallel_config)
self.num_kv_heads = kv_cache_spec.num_kv_heads
self.num_heads = vllm_config.model_config.get_num_attention_heads(
parallel_config
)
+88 -48
View File
@@ -892,21 +892,28 @@ class FlashAttentionImpl(AttentionImpl):
and self.vllm_flash_attn_version == 4
):
max_ranges = mm_prefix_ranges.shape[1]
# Gemma4: clamp the bidirectional block to the sliding
# window (HF (causal OR blockwise) AND sliding_window on
# local layers). Other mm-prefix models pass sliding_window=0
# -> unclamped, behavior unchanged. sliding_window_size[0]+1
# is the window (self.sliding_window = (sw-1, 0) on sliding
# layers); causal mm_prefix never hits the symmetric path.
# Sliding window value in Triton convention
# (1 + window_size[0]). Global-attention layers
# store (-1, -1) → sw stays None / 0.
sw_val = (
1 + sliding_window_size[0]
if sliding_window_size is not None
and sliding_window_size[0] >= 0
else None
)
# Gemma4: also clamp the bidirectional block to the
# sliding window when the layer opts in
# (mm_prefix_clamp_sliding_window flag from PR #47217).
mm_clamp_sw = 0
if (
getattr(layer, "mm_prefix_clamp_sliding_window", False)
and sliding_window_size is not None
and sliding_window_size[0] >= 0
and sw_val is not None
):
mm_clamp_sw = sliding_window_size[0] + 1
mm_clamp_sw = sw_val
mm_mask_mod = _make_mm_prefix_mask_mod(
max_ranges, sliding_window=mm_clamp_sw
max_ranges,
sliding_window=mm_clamp_sw,
sliding_window_left=sw_val,
)
mm_aux = [mm_prefix_ranges]
@@ -1204,23 +1211,26 @@ class FlashAttentionImpl(AttentionImpl):
return output
def _make_mm_prefix_mask_mod(max_ranges: int, sliding_window: int = 0):
"""Build a CuTE-DSL mask_mod implementing (causal OR mm_prefix).
def _make_mm_prefix_mask_mod(
max_ranges: int,
sliding_window: int = 0,
sliding_window_left: int | None = None,
):
"""Build a CuTE-DSL mask_mod implementing
``(causal AND sliding_window) OR mm_prefix``.
Returns a @cute.jit callable that evaluates:
keep = (kv_idx <= q_idx) OR
(q_idx in [r_start,r_end] AND kv_idx in [r_start,r_end]
[AND (q_idx - kv_idx) < sliding_window])
for each mm_prefix range stored in aux_tensors[0].
The FA4 kernel passes *local* ``q_idx`` (0-based within the current
prefill chunk) while ``kv_idx`` is absolute (0-based over the full
KV cache). We recover the absolute Q position via
``q_abs = q_idx + seqlen_k - seqlen_q`` (the context-length offset)
so that causal, sliding-window, and mm_prefix range comparisons all
use consistent absolute positions. This matches the Triton
reference path (``compute_kv_seq_mask``).
When sliding_window > 0 (Gemma4 sliding layers), the bidirectional block is
clamped to the sliding window, matching HF's
(causal OR blockwise) AND sliding_window (== sliding_window_overlay
kv > q - sw; future kv passes since q - kv < 0). sliding_window <= 0
(default) leaves the block unclamped, so other mm-prefix models are
unchanged. q_idx/kv_idx are local offsets, but their difference is
frame-invariant and image bidirectional attention only matters during
prefill (where local == absolute).
``sliding_window_left`` enforces the sliding window on the causal
term (None = full causal, no window). ``sliding_window`` clamps the
bidirectional block to the window (0 = unclamped; >0 = Gemma4 local
layers via ``mm_prefix_clamp_sliding_window``).
"""
import cutlass
import cutlass.cute as cute
@@ -1230,29 +1240,59 @@ def _make_mm_prefix_mask_mod(max_ranges: int, sliding_window: int = 0):
scalar_to_ssa,
)
@cute.jit
def mm_prefix_mask_mod(
batch_idx: cute.TensorSSA,
head_idx: cute.TensorSSA,
q_idx: cute.TensorSSA,
kv_idx: cute.TensorSSA,
seqlen_info,
aux_tensors,
):
keep = kv_idx <= q_idx
ranges = aux_tensors[0]
b = batch_idx[0]
for i in cutlass.range_constexpr(max_ranges): # type: ignore[attr-defined]
r_start = scalar_to_ssa(ranges[b, i, 0], Int32)
r_end = scalar_to_ssa(ranges[b, i, 1], Int32)
valid = r_start < r_end
q_in = (q_idx >= r_start) & (q_idx <= r_end) & valid
k_in = (kv_idx >= r_start) & (kv_idx <= r_end) & valid
mm = q_in & k_in
if sliding_window > 0:
mm = mm & ((q_idx - kv_idx) < sliding_window)
keep = keep | mm
return keep
if sliding_window_left is not None:
@cute.jit
def mm_prefix_mask_mod(
batch_idx: cute.TensorSSA,
head_idx: cute.TensorSSA,
q_idx: cute.TensorSSA,
kv_idx: cute.TensorSSA,
seqlen_info,
aux_tensors,
):
ctx_off = scalar_to_ssa(seqlen_info.seqlen_k - seqlen_info.seqlen_q, Int32)
q_abs = q_idx + ctx_off
sw = scalar_to_ssa(Int32(sliding_window_left), Int32)
keep = (kv_idx <= q_abs) & ((q_abs - kv_idx) < sw)
ranges = aux_tensors[0]
b = batch_idx[0]
for i in cutlass.range_constexpr(max_ranges): # type: ignore[attr-defined]
r_start = scalar_to_ssa(ranges[b, i, 0], Int32)
r_end = scalar_to_ssa(ranges[b, i, 1], Int32)
valid = r_start < r_end
q_in = (q_abs >= r_start) & (q_abs <= r_end) & valid
k_in = (kv_idx >= r_start) & (kv_idx <= r_end) & valid
mm = q_in & k_in
if sliding_window > 0:
mm = mm & ((q_abs - kv_idx) < sw)
keep = keep | mm
return keep
else:
@cute.jit
def mm_prefix_mask_mod(
batch_idx: cute.TensorSSA,
head_idx: cute.TensorSSA,
q_idx: cute.TensorSSA,
kv_idx: cute.TensorSSA,
seqlen_info,
aux_tensors,
):
ctx_off = scalar_to_ssa(seqlen_info.seqlen_k - seqlen_info.seqlen_q, Int32)
q_abs = q_idx + ctx_off
keep = kv_idx <= q_abs
ranges = aux_tensors[0]
b = batch_idx[0]
for i in cutlass.range_constexpr(max_ranges): # type: ignore[attr-defined]
r_start = scalar_to_ssa(ranges[b, i, 0], Int32)
r_end = scalar_to_ssa(ranges[b, i, 1], Int32)
valid = r_start < r_end
q_in = (q_abs >= r_start) & (q_abs <= r_end) & valid
k_in = (kv_idx >= r_start) & (kv_idx <= r_end) & valid
keep = keep | (q_in & k_in)
return keep
mm_prefix_mask_mod.use_fast_sampling = True
return mm_prefix_mask_mod
@@ -536,7 +536,17 @@ class ROCMAiterMLASparseMetadataBuilder(
self._num_attention_heads,
clamped_seq_lens.tobytes(),
)
if metadata_key != self._prev_metadata_key:
# The persistent MLA kernel is numerically wrong for multi-token prefill
# batches; errors compound across chunked prefill and break long-context
# decode (vllm#47042). Use it only for decode and single-chunk prefills,
# not chunked-prefill continuations (>1 query token, seq_len > query_len).
step_query_lens = seg_lengths
total_seq_lens = common_attn_metadata.seq_lens_cpu[:num_reqs].numpy()
is_chunked_continuation = (step_query_lens > 1) & (
total_seq_lens > step_query_lens
)
use_persistent = not is_chunked_continuation.any()
if use_persistent and metadata_key != self._prev_metadata_key:
from aiter import get_mla_metadata_v1
get_mla_metadata_v1(
@@ -576,7 +586,7 @@ class ROCMAiterMLASparseMetadataBuilder(
paged_kv_last_page_len=paged_kv_last_page_len,
paged_kv_indices=paged_kv_indices,
paged_kv_indptr=paged_kv_indptr,
work_meta_data=self._mla_work_meta_data,
work_meta_data=self._mla_work_meta_data if use_persistent else None,
work_indptr=self._mla_work_indptr,
work_info_set=self._mla_work_info_set,
reduce_indptr=self._mla_reduce_indptr,
@@ -148,26 +148,10 @@ class RocmAiterUnifiedAttentionImpl(RocmAttentionImpl):
def _split_kv_cache(
self, kv_cache: torch.Tensor
) -> tuple[torch.Tensor, torch.Tensor]:
if self.attn_type != AttentionType.ENCODER_DECODER:
return kv_cache.unbind(1)
# NOTE: Encoder-decoder layers can share the same raw KV allocation with
# ROCM_ATTN decoder layers, whose physical layout is K/V first. Keep
# this cross-attention path on that physical layout so block IDs do not
# alias different bytes across the shared allocation.
num_blocks, _, block_size, num_kv_heads, head_size = kv_cache.shape
block_stride = block_size * num_kv_heads * head_size
kv_cache = kv_cache.as_strided(
(2, num_blocks, block_size, num_kv_heads, head_size),
(
num_blocks * block_stride,
block_stride,
num_kv_heads * head_size,
head_size,
1,
),
)
return kv_cache.unbind(0)
# Blocks-first ``(num_blocks, 2, ...)``. The model runner normalizes any
# shared decoder/cross-attention allocation to this layout, so no
# per-backend restriding is needed here.
return kv_cache.unbind(1)
def forward(
self,
@@ -242,27 +226,58 @@ class RocmAiterUnifiedAttentionImpl(RocmAttentionImpl):
max_seqlen_k = attn_metadata.max_seq_len
block_table = attn_metadata.block_table
self.unified_attention(
q=query[:num_actual_tokens],
k=key_cache,
v=value_cache,
out=output[:num_actual_tokens],
cu_seqlens_q=cu_seqlens_q,
max_seqlen_q=max_seqlen_q,
seqused_k=seqused_k,
max_seqlen_k=max_seqlen_k,
softmax_scale=softmax_scale,
causal=True,
alibi_slopes=self.alibi_slopes,
window_size=self.sliding_window,
block_table=block_table,
softcap=self.logits_soft_cap,
q_descale=layer._q_scale if query.dtype == self.fp8_dtype else None,
k_descale=layer._k_scale,
v_descale=layer._v_scale,
sinks=self.sinks,
output_scale=output_scale,
)
if attn_metadata.causal:
self.unified_attention(
q=query[:num_actual_tokens],
k=key_cache,
v=value_cache,
out=output[:num_actual_tokens],
cu_seqlens_q=cu_seqlens_q,
max_seqlen_q=max_seqlen_q,
seqused_k=seqused_k,
max_seqlen_k=max_seqlen_k,
softmax_scale=softmax_scale,
causal=True,
alibi_slopes=self.alibi_slopes,
window_size=self.sliding_window,
block_table=block_table,
softcap=self.logits_soft_cap,
q_descale=layer._q_scale if query.dtype == self.fp8_dtype else None,
k_descale=layer._k_scale,
v_descale=layer._v_scale,
sinks=self.sinks,
output_scale=output_scale,
)
else:
# The aiter kernel is causal-only. Non-causal cross-attention
# (ENCODER_DECODER, e.g. Whisper) falls back to the vLLM Triton
# unified kernel, which shares this layout and honors the flag.
from vllm.v1.attention.ops.triton_unified_attention import (
unified_attention as triton_unified_attention,
)
descale_shape = (cu_seqlens_q.shape[0] - 1, key_cache.shape[2])
triton_unified_attention(
q=query[:num_actual_tokens],
k=key_cache,
v=value_cache,
out=output[:num_actual_tokens],
cu_seqlens_q=cu_seqlens_q,
max_seqlen_q=max_seqlen_q,
seqused_k=seqused_k,
max_seqlen_k=max_seqlen_k,
softmax_scale=softmax_scale,
causal=attn_metadata.causal,
alibi_slopes=self.alibi_slopes,
window_size=self.sliding_window,
block_table=block_table,
softcap=self.logits_soft_cap,
q_descale=layer._q_scale if query.dtype == self.fp8_dtype else None,
k_descale=layer._k_scale.expand(descale_shape),
v_descale=layer._v_scale.expand(descale_shape),
sinks=self.sinks,
output_scale=output_scale,
)
return output
+13 -4
View File
@@ -1636,14 +1636,23 @@ class Scheduler(SchedulerInterface):
if new_token_ids and self.structured_output_manager.should_advance(request):
struct_output_request = request.structured_output_request
assert struct_output_request is not None
assert struct_output_request.grammar is not None
if not struct_output_request.grammar.accept_tokens( # type: ignore[union-attr]
req_id, new_token_ids
grammar = struct_output_request.grammar
assert grammar is not None
# new_token_ids can be a mixed block of reasoning content, then
# the reasoning end marker, then the start of the grammar content.
# Trim the reasoning content so the grammar only sees grammar content.
advance_token_ids = (
self.structured_output_manager.trim_reasoning_for_advance(
request, new_token_ids
)
)
if advance_token_ids and not grammar.accept_tokens(
req_id, advance_token_ids
):
logger.error(
"Unexpected: grammar rejected tokens %s for request %s. "
"Terminating request.",
new_token_ids,
advance_token_ids,
req_id,
)
request.status = RequestStatus.FINISHED_ERROR
+112 -12
View File
@@ -2,7 +2,7 @@
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
import itertools
import multiprocessing
from collections.abc import Iterable
from collections.abc import Iterable, Sequence
from concurrent.futures import Future, ThreadPoolExecutor
from typing import TYPE_CHECKING
@@ -272,23 +272,69 @@ class StructuredOutputManager:
grammar = structured_output_request.grammar
apply_bitmask = self.should_fill_bitmask(request)
reasoner = self._get_reasoner(request)
detect_reasoning_end = (
not apply_bitmask
and reasoner is not None
and not self.enable_in_reasoning
)
simulated_buf: list[int] | None = None
history_len = 0
state_advancements = 0
post_reasoning_end_in_window = False
req_tokens = scheduled_spec_decode_tokens.get(req_id, ())
if self.vllm_config.model_config.is_diffusion and req_tokens:
# Diffusion LLMs don't sample a bonus token after the
# scheduled positions, so don't append the -1 placeholder.
token_iter: Iterable[int] = req_tokens
else:
token_iter = itertools.chain(req_tokens, (-1,))
for token in token_iter:
for i, token in enumerate(req_tokens):
self._fill_bitmasks(((grammar, cumulative_index, apply_bitmask),))
advance_grammar = apply_bitmask
if token == -1:
# Stop advancing the grammar once we hit a padding token.
apply_bitmask = False
if apply_bitmask and not grammar.is_terminated():
advance_grammar = False
elif (
detect_reasoning_end
and reasoner is not None
and not apply_bitmask
):
if simulated_buf is None:
history = list(request.all_token_ids)
history_len = len(history)
simulated_buf = history + list(req_tokens)
simulated = simulated_buf[: history_len + i + 1]
if reasoner.is_reasoning_end_streaming(simulated, [token]):
# Reasoning ended mid-window. Constrain the rest
# of the window via bitmask. Skip grammar advance
# through the marker (it is reasoning content);
# try to advance through subsequent drafts so the
# next bitmask row reflects the post-advance state,
# but tolerate rejection since those drafts predate
# the bitmask and are not guaranteed valid.
apply_bitmask = True
advance_grammar = False
post_reasoning_end_in_window = True
if advance_grammar and not grammar.is_terminated():
accepted = grammar.accept_tokens(req_id, [token])
assert accepted, (token, req_id, scheduled_spec_decode_tokens)
state_advancements += 1
if accepted:
state_advancements += 1
elif not post_reasoning_end_in_window:
raise AssertionError(
(token, req_id, scheduled_spec_decode_tokens)
)
cumulative_index += 1
# Diffusion LLMs don't sample a bonus token after the
# scheduled positions, so skip its bitmask in that case.
if not (self.vllm_config.model_config.is_diffusion and req_tokens):
# bonus_apply must be True when the bonus-row position
# should be grammar-constrained. Two triggers:
# - should_fill_bitmask(request): reasoning was already
# over at step start (or no reasoner /
# enable_in_reasoning).
# - apply_bitmask: reasoning ended mid-window in this
# call and was flipped True after the marker;
# should_fill_bitmask still returns False here because
# reasoning_ended is only persisted later by
# should_advance.
bonus_apply = self.should_fill_bitmask(request) or apply_bitmask
self._fill_bitmasks(((grammar, cumulative_index, bonus_apply),))
cumulative_index += 1
if state_advancements > 0:
grammar.rollback(state_advancements)
@@ -368,10 +414,64 @@ class StructuredOutputManager:
and structured_req.structured_output_key[0]
== StructuredOutputOptions.STRUCTURAL_TAG
):
# The scheduler will advance the grammar with this step's
# tokens right away, but the step still contains reasoning
# content up to and including the end marker. Record where
# it ends so trim_reasoning_for_advance() can drop it.
structured_req.reasoning_end_token_index = (
self._find_reasoning_end_index(reasoner, all_token_ids, start)
)
return True
return False
@staticmethod
def _find_reasoning_end_index(
reasoner: "ReasoningParser", all_token_ids: Sequence[int], start: int
) -> int:
"""Locates the last reasoning token within ``all_token_ids[start:]``.
Returns:
The absolute index of the token at which
``is_reasoning_end_streaming`` first fires. Falls back to the
final index when no single token triggers the detection (e.g.
a multi-token marker only recognized on the full delta), which
conservatively treats the whole step as reasoning content.
"""
prefix = list(itertools.islice(all_token_ids, start))
for idx in range(start, len(all_token_ids)):
token = all_token_ids[idx]
prefix.append(token)
if reasoner.is_reasoning_end_streaming(prefix, [token]):
return idx
return len(all_token_ids) - 1
def trim_reasoning_for_advance(
self, request: "Request", new_token_ids: list[int]
) -> list[int]:
"""Drops reasoning content from tokens about to advance the grammar.
When reasoning ends mid-step (see should_advance), the step's output
still contains reasoning tokens up to and including the end marker.
Those are not grammar content: feeding them to accept_tokens makes
the grammar reject the marker and kills the request (#44006).
Returns:
The suffix of ``new_token_ids`` that follows the reasoning-end
marker. Steps fully after the boundary are returned unchanged.
"""
structured_req = request.structured_output_request
if structured_req is None:
return new_token_ids
end_idx = structured_req.reasoning_end_token_index
if end_idx is None:
return new_token_ids
first_idx = len(request.all_token_ids) - len(new_token_ids)
num_reasoning = end_idx + 1 - first_idx
if num_reasoning <= 0:
return new_token_ids
return new_token_ids[num_reasoning:]
def clear_backend(self) -> None:
if self.backend is not None:
self.backend.destroy()

Some files were not shown because too many files have changed in this diff Show More