5 Commits

Author SHA1 Message Date
df07aecb0c actually invoke vllm this time 2026-04-23 00:44:59 +00:00
510ef70f77 fix path 2026-04-23 00:38:38 +00:00
7899827a03 fix loop 2026-04-23 00:36:30 +00:00
66e37bf3fb handle custom weight downloading 2026-04-22 23:46:22 +00:00
82330a31b1 add await to redis write 2026-04-17 07:19:15 +00:00
13 changed files with 1885 additions and 3529 deletions

View File

@@ -1,8 +1,8 @@
#FROM vllm/vllm-openai:v0.19.0-cu130
#vllm says version 0.20.2rc1.dev9+g01d4d1ad3
FROM vllm/vllm-openai:glm52
#FROM vllm/vllm-openai:nightly
#FROM vllm/vllm-openai:glm51-cu130
FROM vllm/vllm-openai:cu130-nightly-x86_64
# Fix the broken ass nightly build that forgot to include pandas
RUN pip install --no-cache-dir pandas
# Install LMCache for KV cache offloading / sharing across nodes
# Build with system CUDA 13.0 for Blackwell (B200)
@@ -15,27 +15,40 @@ RUN apt-get update && apt-get install -y git \
libnvjitlink-dev-13-0 && \
git clone https://github.com/biondizzle/LMCache.git /tmp/lmcache && \
cd /tmp/lmcache && \
git checkout dream-build && \
git checkout feat/redis-ttl && \
CUDA_HOME=/usr/local/cuda \
TORCH_CUDA_ARCH_LIST="10.0" \
pip install --no-cache-dir --no-build-isolation . && \
rm -rf /tmp/lmcache && export CACHE_BUSTER=5
rm -rf /tmp/lmcache && export CACHE_BUSTER=1
# Copy over nemotron reasonong parser
COPY ./super_v3_reasoning_parser.py /opt/super_v3_reasoning_parser.py
# Make sure we have patch to make MTP work on GLM (I did this based on vllm-project/vllm#40989)
COPY indexer.py /usr/local/lib/python3.12/dist-packages/vllm/v1/attention/backends/mla/indexer.py
COPY deep_gemm.py /usr/local/lib/python3.12/dist-packages/vllm/utils/deep_gemm.py
# Copy over deepseek tool call parser with MTP fixes
COPY deepseekv32_tool_parser.py /usr/local/lib/python3.12/dist-packages/vllm/tool_parsers/deepseekv32_tool_parser.py
# These were from https://github.com/vllm-project/vllm/pull/41357/changes#diff-75b8ca6d854db6a47e75db6507afd20c15624f35229e2fd0d71642bffd70b11c
COPY shm_broadcast.py /usr/local/lib/python3.12/dist-packages/vllm/distributed/device_communicators/shm_broadcast.py
COPY multiproc_executor.py /usr/local/lib/python3.12/dist-packages/vllm/v1/executor/multiproc_executor.py
# Copy over minimax tool call parser with kwargs fixes
COPY minimax_tool_parser.py /usr/local/lib/python3.12/dist-packages/vllm/tool_parsers/minimax_tool_parser.py
# Make sure we have the latest up to date chat template
#COPY glm_5.1_chat_template.jinja /opt/chat_template.jinja
# Copy over minimax parsers with kwargs fixes
COPY minimax_tool_parser.py /usr/local/lib/python3.12/dist-packages/vllm/tool_parsers/minimax_tool_parser.py
COPY minimax_m2_parser.py /usr/local/lib/python3.12/dist-packages/vllm/parser/minimax_m2_parser.py
# GLM 5.1 LMCache config
COPY lmcache-config-glm-52.yaml /opt/lmcache-config-glm-52.yaml
# Copy vLLM shim that intercepts --model to download custom weights from URLs
COPY vllm_shim_module.py /opt/vllm-shim/vllm_shim_module.py
# DEEPSEEK v4 LMCache config
#COPY lmcache-config-dsv4.yaml /opt/lmcache-config-dsv4.yaml
# Shadow `python -m vllm.*` invocations via PYTHONPATH
# The shim masquerades as the vllm package so python -m vllm/entrypoints/openai/api_server
# hits our interceptor first, which downloads weights then execs the real vLLM
RUN mkdir -p /opt/vllm-shim/vllm/entrypoints/openai \
/opt/vllm-shim/vllm/entrypoints/cli && \
cp /opt/vllm-shim/vllm_shim_module.py /opt/vllm-shim/vllm/__main__.py && \
cp /opt/vllm-shim/vllm_shim_module.py /opt/vllm-shim/vllm/entrypoints/openai/api_server.py && \
cp /opt/vllm-shim/vllm_shim_module.py /opt/vllm-shim/vllm/entrypoints/cli/main.py && \
touch /opt/vllm-shim/vllm/__init__.py \
/opt/vllm-shim/vllm/entrypoints/__init__.py \
/opt/vllm-shim/vllm/entrypoints/openai/__init__.py \
/opt/vllm-shim/vllm/entrypoints/cli/__init__.py
ENV PYTHONPATH=/opt/vllm-shim
ENV PYTHONUNBUFFERED=1

View File

@@ -1,587 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Compatibility wrapper for DeepGEMM API changes.
Users of vLLM should always import **only** these wrappers.
"""
import functools
import importlib
import os
from collections.abc import Callable
from enum import Enum
from typing import Any, NoReturn
import torch
import vllm.envs as envs
from vllm.logger import logger
from vllm.model_executor.layers.quantization.utils.quant_utils import (
get_fp8_min_max,
)
from vllm.platforms import current_platform
from vllm.utils.import_utils import has_deep_gemm
from vllm.utils.math_utils import cdiv
_DEEPGEMM_BLACKWELL_EXCLUDED_MODEL_TYPES: set[str] = {
"qwen3_5_text",
"qwen3_5_moe_text",
}
def should_auto_disable_deep_gemm(model_type: str | None) -> bool:
"""Check if DeepGemm should be auto-disabled for this model on Blackwell.
Returns True if the model is known to have accuracy degradation with
DeepGemm's E8M0 scale format on Blackwell GPUs (SM100+).
"""
if model_type is None:
return False
if not current_platform.is_device_capability_family(100):
return False
return model_type in _DEEPGEMM_BLACKWELL_EXCLUDED_MODEL_TYPES
class DeepGemmQuantScaleFMT(Enum):
# Float32 scales in Float32 tensor
FLOAT32 = 0
# Compute float32 scales and ceil the scales to UE8M0.
# Keep the scales in Float32 tensor.
FLOAT32_CEIL_UE8M0 = 1
# Compute float32 scales and ceil the scales to UE8M0.
# Pack the scales into a int32 tensor where each int32
# element contains 4 scale values.
UE8M0 = 2
@classmethod
def init_oracle_cache(cls) -> None:
"""Initialize the oracle decision and store it in the class cache"""
cached = getattr(cls, "_oracle_cache", None)
if cached is not None:
return
use_e8m0 = (
envs.VLLM_USE_DEEP_GEMM_E8M0
and is_deep_gemm_supported()
and (_fp8_gemm_nt_impl is not None)
)
if not use_e8m0:
cls._oracle_cache = cls.FLOAT32 # type: ignore
return
cls._oracle_cache = ( # type: ignore
cls.UE8M0
if current_platform.is_device_capability_family(100)
else cls.FLOAT32_CEIL_UE8M0
)
@classmethod
def from_oracle(cls) -> "DeepGemmQuantScaleFMT":
"""Return the pre-initialized oracle decision"""
cached = getattr(cls, "_oracle_cache", None)
assert cached is not None, "DeepGemmQuantScaleFMT oracle cache not initialized"
return cached
@functools.cache
def is_deep_gemm_supported() -> bool:
"""Return `True` if DeepGEMM is supported on the current platform.
Currently, only Hopper and Blackwell GPUs are supported.
"""
is_supported_arch = current_platform.support_deep_gemm()
return envs.VLLM_USE_DEEP_GEMM and has_deep_gemm() and is_supported_arch
@functools.cache
def is_deep_gemm_e8m0_used() -> bool:
"""Return `True` if vLLM is configured to use DeepGEMM "
"E8M0 scale on a Hopper or Blackwell-class GPU.
"""
if not is_deep_gemm_supported():
logger.debug_once(
"DeepGEMM E8M0 disabled: DeepGEMM not supported on this system."
)
return False
_lazy_init()
if _fp8_gemm_nt_impl is None:
logger.info_once("DeepGEMM E8M0 disabled: _fp8_gemm_nt_impl not found")
return False
if envs.VLLM_USE_DEEP_GEMM_E8M0:
logger.info_once("DeepGEMM E8M0 enabled on current platform.")
return True
logger.info_once("DeepGEMM E8M0 disabled on current configuration.")
return False
def _missing(*_: Any, **__: Any) -> NoReturn:
"""Placeholder for unavailable DeepGEMM backend."""
raise RuntimeError(
"DeepGEMM backend is not available or outdated. Please install or "
"update the `deep_gemm` to a newer version to enable FP8 kernels."
)
_cublaslt_gemm_nt_impl: Callable[..., Any] | None = None
_fp8_gemm_nt_impl: Callable[..., Any] | None = None
_fp8_einsum_impl: Callable[..., Any] | None = None
_grouped_impl: Callable[..., Any] | None = None
_grouped_masked_impl: Callable[..., Any] | None = None
_grouped_fp4_impl: Callable[..., Any] | None = None
_fp8_fp4_mqa_logits_impl: Callable[..., Any] | None = None
_fp8_fp4_paged_mqa_logits_impl: Callable[..., Any] | None = None
_get_paged_mqa_logits_metadata_impl: Callable[..., Any] | None = None
_tf32_hc_prenorm_gemm_impl: Callable[..., Any] | None = None
_get_mn_major_tma_aligned_tensor_impl: Callable[..., Any] | None = None
_get_mk_alignment_for_contiguous_layout_impl: Callable[..., Any] | None = None
_transform_sf_into_required_layout_impl: Callable[..., Any] | None = None
@functools.cache
def _import_deep_gemm():
"""Import the deep_gemm module.
Prefers an externally installed ``deep_gemm`` package (so users can
pin a specific version), then falls back to the vendored copy bundled
in the vLLM wheel.
Returns ``None`` when neither source is usable.
"""
# 1. Try the external (pip-installed) package first.
try:
module = importlib.import_module("deep_gemm")
logger.debug_once("Imported deep_gemm module from site-packages")
return module
except ImportError:
logger.debug_once(
"deep_gemm not found in site-packages, "
"trying vendored vllm.third_party.deep_gemm"
)
# 2. Fall back to the vendored copy bundled in the vLLM wheel.
try:
module = importlib.import_module("vllm.third_party.deep_gemm")
logger.debug_once("Imported deep_gemm module from vllm.third_party.deep_gemm")
return module
except ImportError:
logger.debug_once("Vendored deep_gemm not found either")
except Exception as e:
# The vendored module may raise RuntimeError during _C.init()
# if JIT include files are missing (e.g. incomplete wheel).
logger.warning_once("Failed to import vendored deep_gemm: %s", e)
return None
def _lazy_init() -> None:
"""Import deep_gemm and resolve symbols on first use."""
global _cublaslt_gemm_nt_impl
global _fp8_gemm_nt_impl, _fp8_einsum_impl
global _grouped_impl, _grouped_masked_impl, _grouped_fp4_impl
global _fp8_fp4_mqa_logits_impl, _fp8_fp4_paged_mqa_logits_impl
global _get_paged_mqa_logits_metadata_impl
global _tf32_hc_prenorm_gemm_impl
global _get_mn_major_tma_aligned_tensor_impl
global _get_mk_alignment_for_contiguous_layout_impl
global _transform_sf_into_required_layout_impl
# fast path
if (
_cublaslt_gemm_nt_impl is not None
or _fp8_gemm_nt_impl is not None
or _fp8_einsum_impl is not None
or _grouped_impl is not None
or _grouped_masked_impl is not None
or _grouped_fp4_impl is not None
or _fp8_fp4_mqa_logits_impl is not None
or _fp8_fp4_paged_mqa_logits_impl is not None
or _get_paged_mqa_logits_metadata_impl is not None
or _tf32_hc_prenorm_gemm_impl is not None
or _get_mk_alignment_for_contiguous_layout_impl is not None
or _transform_sf_into_required_layout_impl is not None
):
return
if not has_deep_gemm():
return
# Set up deep_gemm cache path
DEEP_GEMM_JIT_CACHE_ENV_NAME = "DG_JIT_CACHE_DIR"
if not os.environ.get(DEEP_GEMM_JIT_CACHE_ENV_NAME, None):
os.environ[DEEP_GEMM_JIT_CACHE_ENV_NAME] = os.path.join(
envs.VLLM_CACHE_ROOT, "deep_gemm"
)
_dg = _import_deep_gemm()
if _dg is None:
return
_cublaslt_gemm_nt_impl = getattr(_dg, "cublaslt_gemm_nt", None)
_fp8_gemm_nt_impl = getattr(_dg, "fp8_gemm_nt", None)
_fp8_einsum_impl = getattr(_dg, "fp8_einsum", None)
_grouped_impl = getattr(_dg, "m_grouped_fp8_gemm_nt_contiguous", None)
_grouped_masked_impl = getattr(_dg, "fp8_m_grouped_gemm_nt_masked", None)
_grouped_fp4_impl = getattr(_dg, "m_grouped_fp8_fp4_gemm_nt_contiguous", None)
# DeepGEMM exposes fp8_fp4_*_mqa_logits as the canonical symbols that
# handle both the FP8 and FP4 Q/K paths via a tuple-typed `q`.
_fp8_fp4_mqa_logits_impl = getattr(_dg, "fp8_fp4_mqa_logits", None)
_fp8_fp4_paged_mqa_logits_impl = getattr(_dg, "fp8_fp4_paged_mqa_logits", None)
_get_paged_mqa_logits_metadata_impl = getattr(
_dg, "get_paged_mqa_logits_metadata", None
)
_tf32_hc_prenorm_gemm_impl = getattr(_dg, "tf32_hc_prenorm_gemm", None)
_get_mn_major_tma_aligned_tensor_impl = getattr(
_dg, "get_mn_major_tma_aligned_tensor", None
)
_get_mk_alignment_for_contiguous_layout_impl = getattr(
_dg, "get_mk_alignment_for_contiguous_layout", None
)
_transform_sf_into_required_layout_impl = getattr(
_dg, "transform_sf_into_required_layout", None
)
DeepGemmQuantScaleFMT.init_oracle_cache()
def get_num_sms() -> int:
_lazy_init()
dg = _import_deep_gemm()
if dg is None:
raise RuntimeError("DeepGEMM is not available")
return int(dg.get_num_sms())
def set_num_sms(num_sms: int) -> None:
_lazy_init()
dg = _import_deep_gemm()
if dg is None:
raise RuntimeError("DeepGEMM is not available")
dg.set_num_sms(num_sms)
@functools.cache
def get_mk_alignment_for_contiguous_layout() -> list[int]:
_lazy_init()
if _get_mk_alignment_for_contiguous_layout_impl is None:
return _missing()
mk_align_size = _get_mk_alignment_for_contiguous_layout_impl()
return [mk_align_size, mk_align_size]
def get_col_major_tma_aligned_tensor(x: torch.Tensor) -> torch.Tensor:
"""Wrapper for DeepGEMM's get_mn_major_tma_aligned_tensor"""
_lazy_init()
if _get_mn_major_tma_aligned_tensor_impl is None:
return _missing()
return _get_mn_major_tma_aligned_tensor_impl(x)
def cublaslt_gemm_nt(*args, **kwargs):
_lazy_init()
if _cublaslt_gemm_nt_impl is None:
return _missing(*args, **kwargs)
return _cublaslt_gemm_nt_impl(*args, **kwargs)
def fp8_gemm_nt(*args, **kwargs):
_lazy_init()
if _fp8_gemm_nt_impl is None:
return _missing(*args, **kwargs)
if "is_deep_gemm_e8m0_used" in kwargs:
use_ue8m0 = kwargs["is_deep_gemm_e8m0_used"]
del kwargs["is_deep_gemm_e8m0_used"]
else:
use_ue8m0 = is_deep_gemm_e8m0_used()
return _fp8_gemm_nt_impl(*args, disable_ue8m0_cast=not use_ue8m0, **kwargs)
def fp8_einsum(*args, **kwargs):
_lazy_init()
if _fp8_einsum_impl is None:
return _missing(*args, **kwargs)
return _fp8_einsum_impl(*args, **kwargs)
def m_grouped_fp8_gemm_nt_contiguous(*args, **kwargs):
_lazy_init()
if _grouped_impl is None:
return _missing(*args, **kwargs)
return _grouped_impl(
*args, disable_ue8m0_cast=not is_deep_gemm_e8m0_used(), **kwargs
)
def m_grouped_fp8_fp4_gemm_nt_contiguous(*args, **kwargs):
_lazy_init()
if _grouped_fp4_impl is None:
return _missing(*args, **kwargs)
return _grouped_fp4_impl(
*args, disable_ue8m0_cast=not is_deep_gemm_e8m0_used(), **kwargs
)
def fp8_m_grouped_gemm_nt_masked(*args, **kwargs):
_lazy_init()
if _grouped_masked_impl is None:
return _missing(*args, **kwargs)
return _grouped_masked_impl(
*args, disable_ue8m0_cast=not is_deep_gemm_e8m0_used(), **kwargs
)
def transform_sf_into_required_layout(*args, **kwargs):
_lazy_init()
if _transform_sf_into_required_layout_impl is None:
return _missing(*args, **kwargs)
return _transform_sf_into_required_layout_impl(
*args, disable_ue8m0_cast=not is_deep_gemm_e8m0_used(), **kwargs
)
def fp8_fp4_mqa_logits(
q: tuple[torch.Tensor, torch.Tensor | None],
kv: tuple[torch.Tensor, torch.Tensor],
weights: torch.Tensor,
cu_seqlen_ks: torch.Tensor,
cu_seqlen_ke: torch.Tensor,
clean_logits: bool,
) -> torch.Tensor:
"""Compute MQA logits for a single sequence without KV paging.
Unified FP8/FP4 dispatch — the underlying DeepGEMM kernel takes
``q = (values, scales_or_None)`` where ``scales`` is None for FP8 Q
(per-token scale is folded into ``weights``) and a packed block-scale
tensor for MXFP4 Q.
Args:
q: Tuple ``(q_values, q_scale)``. FP8 path: q_values is [M, H, D]
float8_e4m3fn and q_scale is None (per-token scale is folded
into ``weights``). FP4 path: q_values is packed uint8 and
q_scale is the companion block-scale tensor.
kv: Tuple `(k_packed, k_scales)` — FP8 layout is [N, D]
float8_e4m3fn plus fp32 scales [N]; FP4 layout is packed uint8.
weights: weights of shape [M, H], dtype `torch.float32`.
cu_seqlen_ks: Start indices (inclusive) for valid K per query
position, shape [M], dtype int32.
cu_seqlen_ke: End indices (exclusive) for valid K per query
position, shape [M], dtype int32.
clean_logits: Whether to clean the unfilled logits into `-inf`.
Returns:
Logits tensor of shape [M, N], dtype `torch.float32`.
"""
_lazy_init()
if _fp8_fp4_mqa_logits_impl is None:
return _missing()
return _fp8_fp4_mqa_logits_impl(
q,
kv,
weights,
cu_seqlen_ks,
cu_seqlen_ke,
clean_logits=clean_logits,
)
def get_paged_mqa_logits_metadata(
context_lens: torch.Tensor, block_size: int, num_sms: int
) -> torch.Tensor:
"""Build scheduling metadata for paged MQA logits.
Args:
context_lens: Tensor of shape [B] or [B, 1], dtype int32; effective
context length per batch element.
block_size: KV-cache block size in tokens (e.g., 64).
num_sms: Number of SMs available. 132 for Hopper
Returns:
Backend-specific tensor consumed by `fp8_fp4_paged_mqa_logits` to
schedule work across SMs.
"""
_lazy_init()
if _get_paged_mqa_logits_metadata_impl is None:
return _missing()
if context_lens.dim() == 1:
context_lens = context_lens.unsqueeze(-1)
context_lens = context_lens.contiguous()
return _get_paged_mqa_logits_metadata_impl(context_lens, block_size, num_sms)
def fp8_fp4_paged_mqa_logits(
q: tuple[torch.Tensor, torch.Tensor | None],
kv_cache: torch.Tensor,
weights: torch.Tensor,
context_lens: torch.Tensor,
block_tables: torch.Tensor,
schedule_metadata: torch.Tensor,
max_model_len: int,
clean_logits: bool,
) -> torch.Tensor:
"""Compute MQA logits using a paged KV-cache.
Unified FP8/FP4 dispatch — the underlying DeepGEMM kernel takes
``q = (values, scales_or_None)``; pass ``(q_tensor, None)`` for the FP8
path and ``(q_values, q_scale)`` for MXFP4.
Args:
q: Tuple ``(q_values, q_scale)``. FP8 path: q_values is
[B, next_n, H, D] float8_e4m3fn and q_scale is None. FP4 path:
q_values is packed uint8 and q_scale is the companion
block-scale tensor.
kv_cache: Paged KV-cache. FP8 layout is [num_blocks, block_size, 1,
D+4], dtype `torch.uint8`, with the last 4 bytes per (block, pos)
storing the float dequant scale.
weights: Tensor of shape [B * next_n, H], dtype `torch.float32`.
context_lens: Tensor of shape [B], dtype int32; effective context length
for each batch element.
block_tables: Tensor of shape [B, max_blocks], dtype int32; maps logical
block indices to physical blocks in the paged cache.
schedule_metadata: Returned by `get_paged_mqa_logits_metadata`;
used to distribute work across SMs.
max_model_len: Maximum sequence length used to size the logits output.
clean_logits: Whether to clean the unfilled logits into `-inf`.
Returns:
Logits tensor of shape [B * next_n, max_model_len], dtype
`torch.float32`.
"""
_lazy_init()
if _fp8_fp4_paged_mqa_logits_impl is None:
return _missing()
return _fp8_fp4_paged_mqa_logits_impl(
q,
kv_cache,
weights,
context_lens,
block_tables,
schedule_metadata,
max_model_len,
clean_logits=clean_logits,
)
def tf32_hc_prenorm_gemm(
x: torch.Tensor,
fn: torch.Tensor,
out: torch.Tensor,
sqrsum: torch.Tensor,
num_split: int,
) -> torch.Tensor:
"""
Perform the following computation:
out = x.float() @ fn.T
sqrsum = x.float().square().sum(-1)
See the caller function for shape requirement
"""
_lazy_init()
if _tf32_hc_prenorm_gemm_impl is None:
return _missing()
return _tf32_hc_prenorm_gemm_impl(
x,
fn,
out,
sqrsum,
num_split,
)
def _ceil_to_ue8m0(x: torch.Tensor):
return torch.pow(2.0, torch.ceil(torch.log2(x.abs())))
def _align(x: int, y: int) -> int:
return cdiv(x, y) * y
# Taken from https://github.com/deepseek-ai/DeepGEMM/blob/v2.1.1/csrc/utils/math.hpp#L19
def get_tma_aligned_size(x: int, element_size: int) -> int:
return _align(x, 16 // element_size)
DEFAULT_BLOCK_SIZE = [128, 128]
# Taken from https://github.com/deepseek-ai/DeepGEMM/blob/dd6ed14acbc7445dcef224248a77ab4d22b5f240/deep_gemm/utils/math.py#L38
@torch.compile(dynamic=True, backend=current_platform.simple_compile_backend)
def per_block_cast_to_fp8(
x: torch.Tensor, block_size: list[int] = DEFAULT_BLOCK_SIZE, use_ue8m0: bool = False
) -> tuple[torch.Tensor, torch.Tensor]:
fp8_dtype = current_platform.fp8_dtype()
assert x.dim() == 2
m, n = x.shape
block_m, block_n = block_size
x_padded = torch.zeros(
(_align(m, block_m), _align(n, block_n)), dtype=x.dtype, device=x.device
)
x_padded[:m, :n] = x
x_view = x_padded.view(-1, block_m, x_padded.size(1) // block_n, block_n)
x_amax = x_view.abs().float().amax(dim=(1, 3), keepdim=True).clamp(1e-4)
_, fp8_max = get_fp8_min_max()
sf = x_amax / fp8_max
sf = _ceil_to_ue8m0(sf) if use_ue8m0 else sf
x_scaled = (x_view * (1.0 / sf)).to(fp8_dtype)
return x_scaled.view_as(x_padded)[:m, :n].contiguous(), sf.view(
x_view.size(0), x_view.size(2)
)
def calc_diff(x: torch.Tensor, y: torch.Tensor):
"""Return a global difference metric for unit tests.
DeepGEMM kernels on Blackwell/B200 currently exhibit noticeable per-element
error, causing `torch.testing.assert_close` to fail. Instead of checking
every element, we compute a cosine-style similarity over the whole tensor
and report `1 - sim`. Once kernel accuracy improves this helper can be
removed.
"""
x, y = x.double(), y.double()
denominator = (x * x + y * y).sum()
sim = 2 * (x * y).sum() / denominator
return 1 - sim
def should_use_deepgemm_for_fp8_linear(
output_dtype: torch.dtype,
weight_shape: tuple[int, int],
supports_deep_gemm: bool | None = None,
):
if supports_deep_gemm is None:
supports_deep_gemm = is_deep_gemm_supported()
# Verify DeepGEMM N/K dims requirements
# NOTE: Also synchronized with test_w8a8_block_fp8_deep_gemm_matmul
# test inside kernels/quantization/test_block_fp8.py
N_MULTIPLE = 64
K_MULTIPLE = 128
return (
supports_deep_gemm
and output_dtype == torch.bfloat16
and weight_shape[0] % N_MULTIPLE == 0
and weight_shape[1] % K_MULTIPLE == 0
)
__all__ = [
"calc_diff",
"DeepGemmQuantScaleFMT",
"fp8_gemm_nt",
"fp8_einsum",
"m_grouped_fp8_gemm_nt_contiguous",
"m_grouped_fp8_fp4_gemm_nt_contiguous",
"fp8_m_grouped_gemm_nt_masked",
"fp8_fp4_mqa_logits",
"fp8_fp4_paged_mqa_logits",
"get_paged_mqa_logits_metadata",
"per_block_cast_to_fp8",
"is_deep_gemm_e8m0_used",
"is_deep_gemm_supported",
"get_num_sms",
"set_num_sms",
"should_use_deepgemm_for_fp8_linear",
"get_col_major_tma_aligned_tensor",
"get_mk_alignment_for_contiguous_layout",
]

616
deepseekv32_tool_parser.py Normal file
View File

@@ -0,0 +1,616 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""
DeepSeek-V3.2 Tool Call Parser — re-parse-and-diff version.
Adapted from the GLM-4 streaming fix to make the streaming path robust
against multi-token deltas produced by MTP speculative decoding.
Instead of maintaining incremental state that advances one token at a
time, the streaming path re-parses the *entire* current_text on every
call, finds all <DSMLinvoke> regions (complete and in-progress),
builds a JSON arguments string for each, and diffs against what was
previously sent. This makes the parser agnostic to how many tokens
arrive per step.
Key changes vs. the upstream buffer-until-complete parser:
1. _extract_content() handles partial tag overlaps so content text
is never swallowed or duplicated when a tag boundary lands inside
a multi-token chunk.
2. _extract_invoke_regions() finds both complete and incomplete
invoke blocks, enabling streaming of partial arguments.
3. _build_args_json_so_far() constructs the JSON arguments string
incrementally from complete + partial <DSMLparameter> tags.
4. _compute_args_diff() emits only the newly-added characters.
Drop-in replacement: same class name, same interface.
"""
import json
import uuid
from collections.abc import Sequence
from typing import Any
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,
)
logger = init_logger(__name__)
def partial_tag_overlap(text: str, tag: str) -> int:
"""Length of the longest prefix of *tag* that matches a suffix of *text*.
E.g. text ending in ``"<tool_"`` returns 6 when tag is ``"<tool_call>"``.
Returns 0 when there is no overlap.
"""
max_check = min(len(tag) - 1, len(text))
for k in range(max_check, 0, -1):
if text.endswith(tag[:k]):
return k
return 0
class DeepSeekV32ToolParser(ToolParser):
"""
Re-parse-and-diff tool parser for DeepSeek-V3.2 DSML format.
On every streaming call the parser re-parses ``current_text`` to
find ``<DSMLinvoke>`` regions, builds the JSON arguments string
for each tool call, and diffs against what was previously sent to
emit only new content. This is robust against multi-token deltas
from MTP / EAGLE speculative decoding.
Example tool call format::
<DSMLfunction_calls>
<DSMLinvoke name="get_weather">
<DSMLparameter name="location" string="true">杭州</DSMLparameter>
<DSMLparameter name="date" string="true">2024-01-16</DSMLparameter>
</DSMLinvoke>
</DSMLfunction_calls>
"""
def __init__(self, tokenizer: TokenizerLike, tools: list[Tool] | None = None):
super().__init__(tokenizer, tools)
# ----- Tag constants -----
self.tool_call_start_token: str = "<DSMLfunction_calls>"
self.tool_call_end_token: str = "</DSMLfunction_calls>"
self.invoke_end_token: str = "</DSMLinvoke>"
self.param_end_token: str = "</DSMLparameter>"
# Alias expected by ToolParser base / adjust_request
self.tool_calls_start_token = self.tool_call_start_token
# ----- Compiled regexes -----
# Matches a complete <DSMLfunction_calls>…</DSMLfunction_calls>
self.tool_call_complete_regex = re.compile(
r"<DSMLfunction_calls>(.*?)</DSMLfunction_calls>", re.DOTALL
)
# Opening tag of an invoke block — captures the function name.
self.invoke_start_regex = re.compile(
r'<DSMLinvoke\s+name="([^"]+)"\s*>', re.DOTALL
)
# Complete invoke block.
self.invoke_complete_regex = re.compile(
r'<DSMLinvoke\s+name="([^"]+)"\s*>(.*?)</DSMLinvoke>',
re.DOTALL,
)
# Complete parameter tag — captures (name, string_attr, value).
self.parameter_complete_regex = re.compile(
r'<DSMLparameter\s+name="([^"]+)"\s+string="(true|false)"\s*>'
r"(.*?)"
r"</DSMLparameter>",
re.DOTALL,
)
# Just the opening header of a parameter tag (for partial params).
self.parameter_header_regex = re.compile(
r'<DSMLparameter\s+name="([^"]+)"\s+string="(true|false)"\s*>',
re.DOTALL,
)
# ----- Streaming state (reset per request) -----
self._sent_content_idx: int = 0
self._tool_call_ids: list[str] = []
self.streamed_args_for_tool: list[str] = []
self.prev_tool_call_arr: list[dict[str, Any]] = []
self.current_tool_id: int = -1
if not self.model_tokenizer:
raise ValueError(
"The model tokenizer must be passed to the ToolParser "
"constructor during construction."
)
logger.debug(
"Successfully initialized %s", self.__class__.__name__
)
# ------------------------------------------------------------------
# Request adjustment
# ------------------------------------------------------------------
def adjust_request(
self, request: ChatCompletionRequest | ResponsesRequest
) -> ChatCompletionRequest | ResponsesRequest:
request = super().adjust_request(request)
if request.tools and request.tool_choice != "none":
# Ensure DSML tokens are not stripped during decoding.
request.skip_special_tokens = False
return request
# ------------------------------------------------------------------
# Static / utility helpers
# ------------------------------------------------------------------
@staticmethod
def _tools_enabled(request: ChatCompletionRequest) -> bool:
"""Check whether tool calling is active for this request."""
try:
tools = getattr(request, "tools", None)
tool_choice = getattr(request, "tool_choice", None)
return bool(tools) and tool_choice != "none"
except Exception:
logger.exception("Failed to determine if tools are enabled.")
return False
def _generate_tool_call_id(self) -> str:
return f"call_{uuid.uuid4().hex[:24]}"
@staticmethod
def _json_escape_string_content(s: str) -> str:
"""JSON-escape a string value (without surrounding quotes)."""
if not s:
return ""
return json.dumps(s, ensure_ascii=False)[1:-1]
# ------------------------------------------------------------------
# Type conversion helpers
# ------------------------------------------------------------------
def _convert_param_value_checked(self, value: str, param_type: str) -> Any:
"""Convert a raw string value to the type indicated by *param_type*.
Raises on failure so the caller can try the next candidate type.
"""
if value.lower() == "null":
return None
param_type = param_type.lower()
if param_type in ("string", "str", "text"):
return value
elif param_type in ("integer", "int"):
return int(value)
elif param_type in ("number", "float"):
val = float(value)
return val if val != int(val) else int(val)
elif param_type in ("boolean", "bool"):
normed = value.strip().lower()
if normed not in ("false", "0", "true", "1"):
raise ValueError(f"Invalid boolean value: {value!r}")
return normed in ("true", "1")
elif param_type in ("object", "array"):
return json.loads(value)
else:
return json.loads(value)
def _convert_param_value(self, value: str, param_type: str | list[str]) -> Any:
"""Try each candidate type in turn; fall back to the raw string."""
if not isinstance(param_type, list):
param_type = [param_type]
for current_type in param_type:
try:
return self._convert_param_value_checked(value, current_type)
except Exception:
continue
return value
def _get_param_schema_type(
self, func_name: str, param_name: str
) -> str | list[str]:
"""Look up the JSON-schema type for a parameter, defaulting to
``"string"``."""
if self.tools:
for tool in self.tools:
if (
hasattr(tool, "function")
and tool.function.name == func_name
and hasattr(tool.function, "parameters")
):
schema = tool.function.parameters
if isinstance(schema, dict) and "properties" in schema:
prop = schema["properties"].get(param_name, {})
if isinstance(prop, dict):
return prop.get("type", "string")
break
return "string"
def _convert_with_schema(
self, func_name: str, param_name: str, value: str
) -> Any:
"""Convert *value* using the tool schema for *func_name*.*param_name*."""
param_type = self._get_param_schema_type(func_name, param_name)
return self._convert_param_value(value, param_type)
def _is_string_type(self, func_name: str, param_name: str) -> bool:
"""Return True if the schema says this parameter is a string."""
ptype = self._get_param_schema_type(func_name, param_name)
if isinstance(ptype, list):
return "string" in ptype
return ptype in ("string", "str", "text")
# ------------------------------------------------------------------
# Non-streaming extraction (unchanged logic, shared helpers)
# ------------------------------------------------------------------
def extract_tool_calls(
self,
model_output: str,
request: ChatCompletionRequest,
) -> ExtractedToolCallInformation:
"""Extract tool calls from complete model output (non-streaming)."""
if self.tool_call_start_token not in model_output:
return ExtractedToolCallInformation(
tools_called=False, tool_calls=[], content=model_output
)
try:
tool_calls: list[ToolCall] = []
for fc_block in self.tool_call_complete_regex.findall(model_output):
for invoke_name, invoke_body in self.invoke_complete_regex.findall(
fc_block
):
# Parse all parameters in this invoke.
raw_params: dict[str, str] = {}
for pname, _str_attr, pval in (
self.parameter_complete_regex.findall(invoke_body)
):
raw_params[pname] = pval
# Convert types via schema.
converted: dict[str, Any] = {}
for pname, pval in raw_params.items():
converted[pname] = self._convert_with_schema(
invoke_name, pname, pval
)
tool_calls.append(
ToolCall(
type="function",
function=FunctionCall(
name=invoke_name,
arguments=json.dumps(
converted, ensure_ascii=False
),
),
)
)
if not tool_calls:
return ExtractedToolCallInformation(
tools_called=False, tool_calls=[], content=model_output
)
first_idx = model_output.find(self.tool_call_start_token)
content = model_output[:first_idx] if first_idx > 0 else None
return ExtractedToolCallInformation(
tools_called=True, tool_calls=tool_calls, content=content
)
except Exception:
logger.exception("Error extracting tool calls from complete output")
return ExtractedToolCallInformation(
tools_called=False, tool_calls=[], content=model_output
)
# ------------------------------------------------------------------
# Streaming helpers — re-parse-and-diff
# ------------------------------------------------------------------
def _reset_streaming_state(self) -> None:
self._sent_content_idx = 0
self._tool_call_ids.clear()
self.streamed_args_for_tool.clear()
self.prev_tool_call_arr.clear()
self.current_tool_id = -1
def _extract_content(self, current_text: str) -> str | None:
"""Return any non-tool-call text that hasn't been sent yet.
Walks *current_text* from ``_sent_content_idx``, collecting text
outside ``<DSMLfunction_calls>`` regions. Uses
``partial_tag_overlap`` to avoid emitting bytes that might turn
out to be the start of the function-calls tag once the next
chunk arrives.
"""
content_segments: list[str] = []
pos = self._sent_content_idx
while pos < len(current_text):
start = current_text.find(self.tool_call_start_token, pos)
if start == -1:
# No (more) tool-call regions — send the tail, minus
# any suffix that could be the beginning of the tag.
tail = current_text[pos:]
overlap = partial_tag_overlap(tail, self.tool_call_start_token)
sendable = tail[: len(tail) - overlap] if overlap else tail
if sendable:
content_segments.append(sendable)
pos = len(current_text) - overlap
break
# Text between previous position and the tag start is content.
if start > pos:
content_segments.append(current_text[pos:start])
# Skip past the tool-call region.
end = current_text.find(self.tool_call_end_token, start)
if end != -1:
pos = end + len(self.tool_call_end_token)
else:
# Region still open — park cursor at start, stop.
pos = start
break
if content_segments:
self._sent_content_idx = pos
return "".join(content_segments)
if pos > self._sent_content_idx:
self._sent_content_idx = pos
return None
def _extract_invoke_regions(
self, text: str
) -> list[tuple[str, str, bool]]:
"""Find all invoke blocks inside the function_calls region.
Returns a list of ``(func_name, inner_text, is_complete)``
tuples. *inner_text* is everything between the invoke open
tag and the close tag (or the end of available text for the
last, potentially incomplete, invoke).
"""
results: list[tuple[str, str, bool]] = []
fc_start = text.find(self.tool_call_start_token)
if fc_start == -1:
return results
region_start = fc_start + len(self.tool_call_start_token)
fc_end = text.find(self.tool_call_end_token, region_start)
region = text[region_start:fc_end] if fc_end != -1 else text[region_start:]
pos = 0
while pos < len(region):
inv_match = self.invoke_start_regex.search(region, pos)
if not inv_match:
break
func_name = inv_match.group(1)
body_start = inv_match.end()
inv_end_pos = region.find(self.invoke_end_token, body_start)
if inv_end_pos != -1:
# Complete invoke block.
body = region[body_start:inv_end_pos]
results.append((func_name, body, True))
pos = inv_end_pos + len(self.invoke_end_token)
else:
# Incomplete — still being generated.
body = region[body_start:]
overlap = partial_tag_overlap(body, self.invoke_end_token)
if overlap:
body = body[:-overlap]
results.append((func_name, body, False))
break
return results
def _build_args_json_so_far(
self,
func_name: str,
inner_text: str,
is_complete: bool,
) -> str:
"""Build a JSON arguments string from the parameters found so far.
Handles both fully-closed ``<DSMLparameter>`` tags and the
single trailing partial parameter whose value is still being
streamed.
"""
# ---- Collect all fully-closed parameters ----
complete_params = self.parameter_complete_regex.findall(inner_text)
parts: list[str] = []
for param_name, string_attr, param_value in complete_params:
key_json = json.dumps(param_name, ensure_ascii=False)
if string_attr == "true":
val_json = json.dumps(param_value, ensure_ascii=False)
else:
converted = self._convert_with_schema(
func_name, param_name, param_value
)
val_json = json.dumps(converted, ensure_ascii=False)
parts.append(f"{key_json}: {val_json}")
# ---- Handle a trailing partial parameter ----
last_param_open = inner_text.rfind("<DSMLparameter")
last_param_close = inner_text.rfind(self.param_end_token)
has_partial = last_param_open != -1 and (
last_param_close == -1 or last_param_close < last_param_open
)
if has_partial:
partial_text = inner_text[last_param_open:]
header_match = self.parameter_header_regex.search(partial_text)
if header_match:
param_name = header_match.group(1)
string_attr = header_match.group(2)
partial_value = partial_text[header_match.end():]
# Strip any bytes that might be the beginning of the
# closing </DSMLparameter> tag.
overlap = partial_tag_overlap(
partial_value, self.param_end_token
)
if overlap:
partial_value = partial_value[:-overlap]
key_json = json.dumps(param_name, ensure_ascii=False)
if is_complete:
# Invoke is closed — treat whatever we have as final.
if string_attr == "true":
val_json = json.dumps(
partial_value, ensure_ascii=False
)
else:
converted = self._convert_with_schema(
func_name, param_name, partial_value
)
val_json = json.dumps(converted, ensure_ascii=False)
parts.append(f"{key_json}: {val_json}")
elif string_attr == "true" or self._is_string_type(
func_name, param_name
):
# Stream as an open JSON string (no closing quote).
escaped = self._json_escape_string_content(partial_value)
parts.append(f'{key_json}: "{escaped}')
else:
# Non-string — emit raw partial value.
parts.append(f"{key_json}: {partial_value}")
# ---- Assemble ----
if not parts:
return "{}" if is_complete else ""
joined = "{" + ", ".join(parts)
if is_complete:
joined += "}"
return joined
def _compute_args_diff(self, index: int, args_so_far: str) -> str | None:
"""Return only the characters in *args_so_far* that haven't been
sent yet, or ``None`` if there's nothing new."""
prev = self.streamed_args_for_tool[index]
if not args_so_far or len(args_so_far) <= len(prev):
return None
diff = args_so_far[len(prev):]
self.streamed_args_for_tool[index] = args_so_far
self.prev_tool_call_arr[index]["arguments"] = args_so_far
return diff
def _ensure_tool_state_for(self, index: int) -> None:
"""Grow the streaming-state arrays so *index* is valid."""
while len(self._tool_call_ids) <= index:
self._tool_call_ids.append(self._generate_tool_call_id())
while len(self.streamed_args_for_tool) <= index:
self.streamed_args_for_tool.append("")
while len(self.prev_tool_call_arr) <= index:
self.prev_tool_call_arr.append({})
# ------------------------------------------------------------------
# Main streaming entry point
# ------------------------------------------------------------------
def extract_tool_calls_streaming(
self,
previous_text: str,
current_text: str,
delta_text: str,
previous_token_ids: Sequence[int],
current_token_ids: Sequence[int],
delta_token_ids: Sequence[int],
request: ChatCompletionRequest,
) -> DeltaMessage | None:
"""Extract tool calls from streaming output using re-parse-and-diff.
On every call we:
1. Re-scan *current_text* for content outside tool-call regions.
2. Find all ``<DSMLinvoke>`` regions (complete + partial).
3. Build JSON args for each, diff against previous, emit deltas.
Because the entire text is re-parsed each time, the result is
correct regardless of how many tokens arrived in this step.
"""
# First chunk of a new stream — reset state.
if not previous_text:
self._reset_streaming_state()
# If tools aren't enabled, just forward content.
if not self._tools_enabled(request):
return DeltaMessage(content=delta_text) if delta_text else None
# 1. Extract any content outside tool-call regions.
content = self._extract_content(current_text)
# 2. Find all invoke regions.
regions = self._extract_invoke_regions(current_text)
tool_call_deltas: list[DeltaToolCall] = []
for i, (func_name, inner_text, is_complete) in enumerate(regions):
self._ensure_tool_state_for(i)
# Emit the tool name (once per tool call).
if "name" not in self.prev_tool_call_arr[i]:
self.prev_tool_call_arr[i]["name"] = func_name
tool_call_deltas.append(
DeltaToolCall(
index=i,
id=self._tool_call_ids[i],
type="function",
function=DeltaFunctionCall(
name=func_name,
arguments="",
),
)
)
# Build the JSON args so far and emit the diff.
args_so_far = self._build_args_json_so_far(
func_name, inner_text, is_complete
)
diff = self._compute_args_diff(i, args_so_far)
if diff:
tool_call_deltas.append(
DeltaToolCall(
index=i,
function=DeltaFunctionCall(arguments=diff),
)
)
if regions:
self.current_tool_id = len(regions) - 1
# 3. Return a delta if we have content or tool-call updates.
if content or tool_call_deltas:
return DeltaMessage(
content=content,
tool_calls=tool_call_deltas,
)
# 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

View File

@@ -1,119 +0,0 @@
[gMASK]<sop>
{%- if tools -%}
{%- macro tool_to_json(tool) -%}
{%- set ns_tool = namespace(first=true) -%}
{{ '{' -}}
{%- for k, v in tool.items() -%}
{%- if k != 'defer_loading' and k != 'strict' -%}
{%- if not ns_tool.first -%}{{- ', ' -}}{%- endif -%}
{%- set ns_tool.first = false -%}
"{{ k }}": {{ v | tojson(ensure_ascii=False) }}
{%- endif -%}
{%- endfor -%}
{{- '}' -}}
{%- endmacro -%}
<|system|>
# Tools
You may call one or more functions to assist with the user query.
You are provided with function signatures within <tools></tools> XML tags:
<tools>
{% for tool in tools %}
{%- if 'function' in tool -%}
{%- set tool = tool['function'] -%}
{%- endif -%}
{% if tool.defer_loading is not defined or not tool.defer_loading %}
{{ tool_to_json(tool) }}
{% endif %}
{% endfor %}
</tools>
For each function call, output the function name and arguments within the following XML format:
<tool_call>{function-name}<arg_key>{arg-key-1}</arg_key><arg_value>{arg-value-1}</arg_value><arg_key>{arg-key-2}</arg_key><arg_value>{arg-value-2}</arg_value>...</tool_call>{%- endif -%}
{%- macro visible_text(content) -%}
{%- if content is string -%}
{{- content }}
{%- elif content is iterable and content is not mapping -%}
{%- for item in content -%}
{%- if item is mapping and item.type == 'text' -%}
{{- item.text }}
{%- elif item is string -%}
{{- item }}
{%- endif -%}
{%- endfor -%}
{%- else -%}
{{- content }}
{%- endif -%}
{%- endmacro -%}
{%- set ns = namespace(last_user_index=-1, thinking_indices='') -%}
{%- for m in messages %}
{%- if m.role == 'user' %}
{%- set ns.last_user_index = loop.index0 -%}
{%- elif m.role == 'assistant' %}
{%- if m.reasoning_content is string %}
{%- set ns.thinking_indices = ns.thinking_indices ~ ',' ~ ns.last_user_index ~ ',' -%}
{%- endif %}
{%- endif %}
{%- endfor %}
{%- set ns.has_thinking = false -%}
{%- for m in messages -%}
{%- if m.role == 'user' -%}<|user|>{{ visible_text(m.content) }}{% set ns.has_thinking = (',' ~ loop.index0 ~ ',') in ns.thinking_indices -%}
{%- elif m.role == 'assistant' -%}
<|assistant|>
{%- set content = visible_text(m.content) %}
{%- if m.reasoning_content is string %}
{%- set reasoning_content = m.reasoning_content %}
{%- elif '</think>' in content %}
{%- set reasoning_content = content.split('</think>')[0].split('<think>')[-1] %}
{%- set content = content.split('</think>')[-1] %}
{%- elif loop.index0 > ns.last_user_index and not (enable_thinking is defined and not enable_thinking) %}
{%- set reasoning_content = '' %}
{%- elif loop.index0 < ns.last_user_index and ns.has_thinking %}
{%- set reasoning_content = '' %}
{%- endif %}
{%- if ((clear_thinking is defined and not clear_thinking) or loop.index0 > ns.last_user_index) and reasoning_content is defined -%}
{{ '<think>' + reasoning_content + '</think>'}}
{%- else -%}
{{ '</think>' }}
{%- endif -%}
{%- if content.strip() -%}
{{ content.strip() }}
{%- endif -%}
{% if m.tool_calls %}
{% for tc in m.tool_calls %}
{%- if tc.function %}
{%- set tc = tc.function %}
{%- endif %}
{{- '<tool_call>' + tc.name -}}
{% set _args = tc.arguments %}{% for k, v in _args.items() %}<arg_key>{{ k }}</arg_key><arg_value>{{ v | tojson(ensure_ascii=False) if v is not string else v }}</arg_value>{% endfor %}</tool_call>{% endfor %}
{% endif %}
{%- elif m.role == 'tool' -%}
{%- if loop.first or (messages[loop.index0 - 1].role != "tool") %}
{{- '<|observation|>' -}}
{%- endif %}
{%- if m.content is string -%}
{{- '<tool_response>' + m.content + '</tool_response>' -}}
{%- elif m.content is iterable and m.content is not mapping and m.content and m.content.0.type == "tool_reference" -%}
{{- '<tool_response><tools>\n' -}}
{% for tr in m.content %}
{%- for tool in tools -%}
{%- if 'function' in tool -%}
{%- set tool = tool['function'] -%}
{%- endif -%}
{%- if tool.name == tr.name -%}
{{- tool_to_json(tool) + '\n' -}}
{%- endif -%}
{%- endfor -%}
{%- endfor -%}
{{- '</tools></tool_response>' -}}
{%- else -%}
{{- '<tool_response>' + visible_text(m.content) + '</tool_response>' -}}
{% endif -%}
{%- elif m.role == 'system' -%}
<|system|>{{ visible_text(m.content) }}
{%- endif -%}
{%- endfor -%}
{%- if add_generation_prompt -%}
<|assistant|>{{- '</think>' if (enable_thinking is defined and not enable_thinking) else '<think>' -}}
{%- endif -%}

View File

@@ -1,779 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from dataclasses import dataclass
import torch
import vllm.envs as envs
from vllm.config import VllmConfig
from vllm.logger import init_logger
from vllm.platforms import current_platform
from vllm.triton_utils import tl, triton
from vllm.utils.deep_gemm import (
get_paged_mqa_logits_metadata,
has_deep_gemm,
)
from vllm.utils.math_utils import cdiv
from vllm.utils.platform_utils import num_compute_units
from vllm.v1.attention.backend import (
AttentionBackend,
AttentionCGSupport,
AttentionMetadataBuilder,
CommonAttentionMetadata,
MultipleOf,
)
from vllm.v1.attention.backends.mla.compressor_utils import get_compressed_slot_mapping
from vllm.v1.attention.backends.utils import (
split_decodes_and_prefills,
)
from vllm.v1.kv_cache_interface import AttentionSpec, MLAAttentionSpec
from vllm.v1.worker.cp_utils import get_total_cp_world_size
logger = init_logger(__name__)
@triton.jit
def _prepare_uniform_decode_kernel(
seq_lens_ptr,
decode_seq_lens_ptr,
block_table_ptr,
block_table_stride,
expanded_block_table_ptr,
expanded_bt_stride,
decode_lens_ptr,
max_decode_len,
BLOCK_SIZE: tl.constexpr,
):
idx = tl.program_id(0)
req_id = idx // max_decode_len
local_idx = idx % max_decode_len
# Compute number of KVs attended to by this token.
seq_len = tl.load(seq_lens_ptr + req_id)
per_token_seq_len = seq_len - max_decode_len + local_idx + 1
tl.store(decode_seq_lens_ptr + idx, per_token_seq_len)
# Copy block table row.
src = block_table_ptr + req_id * block_table_stride
dst = expanded_block_table_ptr + idx * expanded_bt_stride
for i in tl.range(0, expanded_bt_stride, BLOCK_SIZE):
off = i + tl.arange(0, BLOCK_SIZE)
mask = off < expanded_bt_stride
src_block = tl.load(src + off, mask=mask)
tl.store(dst + off, src_block, mask=mask)
# All reqs now have decode_len = 1.
tl.store(decode_lens_ptr + idx, 1)
def split_indexer_prefill_chunks(
seq_lens_cpu: torch.Tensor,
query_lens_cpu: torch.Tensor,
workspace_size: int,
max_logits_bytes: int,
request_offset: int = 0,
) -> list[tuple[slice, slice]]:
"""
Split prefill requests into chunks for the sparse indexer, respecting:
- N constraint: total_seq_lens <= workspace_size (existing O(N) workspace)
- Logits constraint: M * N * 4 <= max_logits_bytes
When a single request-level chunk still exceeds the logits budget,
sub-chunks on the query dimension (M) to bound peak memory.
Returns list of (req_slice, query_slice) tuples.
"""
chunks: list[tuple[slice, slice]] = []
n = len(seq_lens_cpu)
max_logits_elems = max_logits_bytes // 4
end = 0
while end < n:
start, chunk_m, chunk_n = end, 0, 0
while end < n:
q, s = query_lens_cpu[end].item(), seq_lens_cpu[end].item()
new_m, new_n = chunk_m + q, chunk_n + s
if new_n <= workspace_size and new_m * new_n <= max_logits_elems:
chunk_m, chunk_n = new_m, new_n
end += 1
else:
break
# A single request can exceed the budget, requiring sub-chunking
# on the query dimension.
if end == start:
chunk_m, chunk_n = query_lens_cpu[end].item(), seq_lens_cpu[end].item()
end += 1
req_slice = slice(start + request_offset, end + request_offset)
max_q = max(1, max_logits_elems // chunk_n) if chunk_n > 0 else chunk_m
for q_off in range(0, chunk_m, max_q):
sub_m = min(max_q, chunk_m - q_off)
chunks.append((req_slice, slice(q_off, q_off + sub_m)))
return chunks
class DeepseekV32IndexerBackend(AttentionBackend):
@staticmethod
def get_name() -> str:
return "DEEPSEEK_V32_INDEXER"
@staticmethod
def get_supported_kernel_block_sizes() -> list[int | MultipleOf]:
return [1, 64] if current_platform.is_rocm() else [64]
@classmethod
def get_supported_head_sizes(cls) -> list[int]:
return [32, 64, 128]
@staticmethod
def get_builder_cls() -> type["DeepseekV32IndexerMetadataBuilder"]:
return DeepseekV32IndexerMetadataBuilder
@staticmethod
def get_kv_cache_shape(
num_blocks: int,
block_size: int,
num_kv_heads: int,
head_size: int,
cache_dtype_str: str = "auto",
) -> tuple[int, ...]:
assert num_kv_heads == 1
return (num_blocks, block_size, head_size)
@staticmethod
def get_kv_cache_stride_order(
include_num_layers_dimension: bool = False,
) -> tuple[int, ...]:
if include_num_layers_dimension:
# DeepseekV32Indexer kernels do not support cross-layer
# KV cache layout. Identity permutation keeps num_layers
# first, signaling incompatibility.
return (0, 1, 2, 3)
return (0, 1, 2)
class DeepseekV4IndexerBackend(DeepseekV32IndexerBackend):
@staticmethod
def get_name() -> str:
return "DEEPSEEK_V4_INDEXER"
@staticmethod
def get_supported_kernel_block_sizes() -> list[int | MultipleOf]:
return [256]
@dataclass
class DeepseekV32IndexerPrefillChunkMetadata:
block_table: torch.Tensor
cu_seqlen_ks: torch.Tensor
cu_seqlen_ke: torch.Tensor
cu_seq_lens: torch.Tensor
token_to_seq: torch.Tensor
total_seq_lens: int
token_start: int
token_end: int
num_reqs: int
skip_kv_gather: bool = False
@dataclass
class DeepseekV32IndexerPrefillMetadata:
chunks: list[DeepseekV32IndexerPrefillChunkMetadata]
@dataclass
class DeepSeekV32IndexerDecodeMetadata:
block_table: torch.Tensor
# seq_lens: per-token effective context lengths.
# - flatten path / plain decode: 1D (batch_size,)
# - native MTP path: 2D (B, next_n) where [b,j] = L_b - next_n + j + 1
# Both fp8_fp4_paged_mqa_logits and the topk kernels accept both shapes.
seq_lens: torch.Tensor
decode_lens: torch.Tensor
requires_padding: bool
schedule_metadata: torch.Tensor
@dataclass
class DeepseekV32IndexerMetadata:
# FIXME (zyongye)
# hacky way to access the data now, need to be in chunked meta
seq_lens: torch.Tensor
max_seq_len: int
slot_mapping: torch.Tensor
# New for MLA (compared to FlashAttention)
# For handling prefill decode split
num_decodes: int
num_decode_tokens: int
num_prefills: int
num_prefill_tokens: int
decode: DeepSeekV32IndexerDecodeMetadata | None = None
prefill: DeepseekV32IndexerPrefillMetadata | None = None
def get_max_prefill_buffer_size(vllm_config: VllmConfig):
max_model_len = vllm_config.model_config.max_model_len
# NOTE(Chen): 40 is a magic number for controlling the prefill buffer size.
# Each entry is 128 fp8 bytes and 4 scale bytes for a total of 132 bytes.
# The flashmla_sparse backend uses a workspace size of 5 * max_model_len.
# The memory usage of the workspace there is 576 * 2 bytes; so we size this as
# (576 * 2 // 132) * 5 = 40 to maximize this workspace size while still fitting
# within the flashmla_sparse workspace.
# For DeepSeek-V3.2, the max_model_len is 163840.
# 40 * 163840 * 132 = 865075200 bytes = 825 MB
return max_model_len * 40
class DeepseekV32IndexerMetadataBuilder(AttentionMetadataBuilder):
reorder_batch_threshold: int = 1
@classmethod
def get_cudagraph_support(
cls,
vllm_config: VllmConfig,
kv_cache_spec: AttentionSpec,
) -> AttentionCGSupport:
return AttentionCGSupport.UNIFORM_BATCH
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
scheduler_config = self.vllm_config.scheduler_config
# NOTE(Chen):an estimated max size of flattened_kv. Need to double check.
self.max_prefill_buffer_size = get_max_prefill_buffer_size(self.vllm_config)
self.num_speculative_tokens = (
self.vllm_config.speculative_config.num_speculative_tokens
if self.vllm_config.speculative_config
else 0
)
self.use_fp4_indexer_cache = (
self.vllm_config.attention_config.use_fp4_indexer_cache
)
assert (
current_platform.is_device_capability_family(100)
or not self.use_fp4_indexer_cache
), (
"use_fp4_indexer_cache requires Blackwell datacenter GPUs "
"(sm_10x, e.g. B200/GB200); sm_120 (consumer Blackwell) and "
"earlier architectures are not supported."
)
next_n = self.num_speculative_tokens + 1
self.reorder_batch_threshold += self.num_speculative_tokens
# NOTE: SM100 datacenter GPUs support any next_n natively via the
# multi-atom paged MQA logits kernels (FP8 and FP4 indexer
# caches). Outside the SM100 family the FP8
# paged MQA logits kernel only supports next_n in (1, 2)
# (deepgemm smxx_fp8_fp4_paged_mqa_logits.hpp:233), so flatten there.
self.use_flattening = not current_platform.is_device_capability_family(
100
) and next_n not in (1, 2)
logger.info_once(
"DSA indexer decode path: use_flattening=%s "
"(next_n=%d, use_fp4_indexer_cache=%s)",
self.use_flattening,
next_n,
self.use_fp4_indexer_cache,
)
sm_count = num_compute_units(self.device.index)
self.num_sms = sm_count
self.offsets_buffer = torch.arange(
next_n, device=self.device, dtype=torch.int32
)
self.decode_lens_buffer = torch.zeros(
(scheduler_config.max_num_batched_tokens,),
dtype=torch.int32,
device=self.device,
)
# Shared workspace for decode seq_lens. Native MTP views this as
# (B, max_decode_len) at runtime, keeping context_lens contiguous even
# when max_decode_len is smaller than next_n.
self.decode_seq_lens_buffer = torch.zeros(
(scheduler_config.max_num_batched_tokens,),
dtype=torch.int32,
device=self.device,
)
self.arange_buffer = torch.arange(
max(
scheduler_config.max_num_seqs * next_n,
scheduler_config.max_num_batched_tokens,
),
dtype=torch.int32,
device=self.device,
)
max_num_blocks_per_req = cdiv(
self.vllm_config.model_config.max_model_len,
self.kv_cache_spec.block_size * get_total_cp_world_size(),
)
self.expanded_block_table_buffer = torch.zeros(
(
scheduler_config.max_num_batched_tokens,
max_num_blocks_per_req,
),
dtype=torch.int32,
device=self.device,
)
# See: DeepGMM/csrc/apis/attention.hpp
self.scheduler_metadata_buffer = torch.empty(
(self.num_sms + 1, 2), dtype=torch.int32, device=self.device
)
# KV compression. Default to 1 for no compression.
self.compress_ratio = 1
# Get compress_ratio for DeepseekV4 support
if isinstance(self.kv_cache_spec, MLAAttentionSpec):
self.compress_ratio = self.kv_cache_spec.compress_ratio
# Pre-allocate buffers for CUDA graph compatibility when
if self.compress_ratio > 1:
# compress_ratio > 1 (DeepseekV4)
# Compressed slot mapping output buffer
self.compressed_slot_mapping_buffer = torch.zeros(
(scheduler_config.max_num_batched_tokens,),
dtype=torch.int64,
device=self.device,
)
# Buffer for compressed seq_lens in decode path
self.expanded_seq_lens_buffer = torch.zeros(
(scheduler_config.max_num_batched_tokens,),
dtype=torch.int32,
device=self.device,
)
def _prepare_decode_tensors(
self,
seq_lens: torch.Tensor,
block_table: torch.Tensor,
decode_lens: torch.Tensor,
decode_lens_cpu: torch.Tensor,
query_start_loc: torch.Tensor,
num_decodes: int,
num_decode_tokens: int,
use_native: bool,
next_n: int,
max_decode_len: int,
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, int, bool]:
"""Expand seq_lens/block_table/decode_lens for the decode kernels.
Flatten path (not use_native, max_decode_len > 1):
Each multi-token decode request is expanded into individual
single-token entries so the kernel always sees next_n=1.
Native path (use_native or max_decode_len == 1):
Plain decode or spec-decode with 2D per-token context lengths.
Returns (seq_lens, block_table, decode_lens, batch_size, requires_padding).
seq_lens is 1D (batch_size,) for flatten/plain, 2D (B, max_decode_len)
for native MTP.
"""
min_decode_len = int(decode_lens_cpu.min().item())
if not use_native and max_decode_len > 1:
assert self.decode_seq_lens_buffer.dim() == 1
if min_decode_len == max_decode_len:
# Uniform decode lengths.
num_decode_tokens = num_decodes * max_decode_len
_prepare_uniform_decode_kernel[(num_decode_tokens,)](
seq_lens,
self.decode_seq_lens_buffer,
block_table,
block_table.stride(0),
self.expanded_block_table_buffer,
self.expanded_block_table_buffer.stride(0),
self.decode_lens_buffer,
max_decode_len,
BLOCK_SIZE=1024,
)
self.decode_seq_lens_buffer[num_decode_tokens:] = 0
seq_lens = self.decode_seq_lens_buffer[:num_decode_tokens]
block_table = self.expanded_block_table_buffer[:num_decode_tokens]
decode_lens = self.decode_lens_buffer[:num_decode_tokens]
return seq_lens, block_table, decode_lens, num_decode_tokens, False
else:
# Variable decode lengths.
# Assume 4 requests with seq_lens [10, 7, 12, 0] (the final req is
# padding) and decode_lens [3, 1, 4, 0] in the below example comments.
# The context lengths are therefore
# [10-3, 7-1, 12-4, 0-0] = [7, 6, 8, 0].
# 3 + 1 + 4 + 0 = 8
actual_expanded = int(decode_lens_cpu.sum().item())
# Fuse expanded_base and expanded_starts into a single
# repeat_interleave:
# seq_len_i = (context_start[b] - query_start_loc[b]) + arange[i] + 1
# where context_start[b] = seq_lens[b] - decode_lens[b].
# Example: offsets = [7-0, 6-3, 8-4, 0-8] = [7, 3, 4, -8]
# expanded_offsets = [7, 7, 7, 3, 4, 4, 4, 4]
# result = [8, 9, 10, 7, 9, 10, 11, 12]
expanded_offsets = torch.repeat_interleave(
seq_lens - decode_lens - query_start_loc,
decode_lens,
output_size=actual_expanded,
)
# [8, 9, 10, 7, 9, 10, 11, 12, ...] where ... is unused buffer space
self.decode_seq_lens_buffer[:actual_expanded] = (
expanded_offsets + self.arange_buffer[:actual_expanded] + 1
)
self.decode_seq_lens_buffer[actual_expanded:] = 0
seq_lens = self.decode_seq_lens_buffer[:num_decode_tokens]
# Give each of the flattened entries the same block table row as the
# original request.
self.expanded_block_table_buffer[:actual_expanded] = (
torch.repeat_interleave(
block_table, decode_lens, dim=0, output_size=actual_expanded
)
)
if actual_expanded < num_decode_tokens:
self.expanded_block_table_buffer[
actual_expanded:num_decode_tokens, 0
] = 0
block_table = self.expanded_block_table_buffer[:num_decode_tokens]
# All reqs now have decode_len=1
self.decode_lens_buffer[:num_decode_tokens] = 1
decode_lens = self.decode_lens_buffer[:num_decode_tokens]
return seq_lens, block_table, decode_lens, num_decode_tokens, False
else:
# Native path: plain decode (next_n==1) or spec decode
# with 2D per-token context lengths (next_n > 1).
#
# When decode_lens are not truly uniform (e.g. some requests have
# decode_len < next_n due to padding or short prefills), the simple
# reshape in sparse_attn_indexer won't work. Use pack_seq_triton
# (requires_padding) instead.
requires_padding = min_decode_len != max_decode_len
if use_native and next_n > 1:
assert self.decode_seq_lens_buffer.dim() == 1
# (B, max_decode_len): token j attends to
# L - max_decode_len + j + 1 KV tokens.
seq_lens_buffer = self.decode_seq_lens_buffer[
: num_decodes * max_decode_len
].view(num_decodes, max_decode_len)
seq_lens_buffer[:] = (
seq_lens.unsqueeze(1)
- max_decode_len
+ 1
+ self.offsets_buffer[:max_decode_len]
)
seq_lens = seq_lens_buffer
return seq_lens, block_table, decode_lens, num_decodes, requires_padding
def build(
self,
common_prefix_len: int,
common_attn_metadata: CommonAttentionMetadata,
fast_build: bool = False,
) -> DeepseekV32IndexerMetadata:
num_reqs = common_attn_metadata.num_reqs
num_tokens = common_attn_metadata.num_actual_tokens
query_start_loc = common_attn_metadata.query_start_loc
query_start_loc_cpu = common_attn_metadata.query_start_loc_cpu
seq_lens = common_attn_metadata.seq_lens
slot_mapping = common_attn_metadata.slot_mapping
block_table = common_attn_metadata.block_table_tensor
num_decodes, num_prefills, num_decode_tokens, num_prefill_tokens = (
split_decodes_and_prefills(
common_attn_metadata,
decode_threshold=self.reorder_batch_threshold,
require_uniform=not self.use_flattening,
)
)
assert num_decodes + num_prefills == num_reqs
assert num_decode_tokens + num_prefill_tokens == num_tokens
compressed_slot_mapping = slot_mapping
compressed_seq_lens = seq_lens
if self.compress_ratio > 1:
compressed_slot_mapping = get_compressed_slot_mapping(
num_tokens,
query_start_loc,
seq_lens,
block_table,
self.kv_cache_spec.storage_block_size,
self.compress_ratio,
out=self.compressed_slot_mapping_buffer,
)
compressed_seq_lens = seq_lens // self.compress_ratio
prefill_metadata = None
if num_prefills > 0:
# This CPU value is an upper bound for async-spec extend rows. It
# is safe for chunking/allocation because CUDA metadata below is
# built from exact device seq_lens and gather ignores the tail.
assert common_attn_metadata.seq_lens_cpu_upper_bound is not None
seq_lens_cpu = common_attn_metadata.seq_lens_cpu_upper_bound
compressed_seq_lens_cpu = (
seq_lens_cpu // self.compress_ratio
if self.compress_ratio > 1
else seq_lens_cpu
)
prefill_query_lens_cpu = torch.diff(
query_start_loc_cpu[num_decodes : num_decodes + num_prefills + 1]
)
max_logits_bytes = envs.VLLM_SPARSE_INDEXER_MAX_LOGITS_MB * 1024 * 1024
# Upper bound is exact for prefill rows (the `[num_decodes:]`
# slice below).
assert common_attn_metadata.seq_lens_cpu_upper_bound is not None
seq_lens_cpu = common_attn_metadata.seq_lens_cpu_upper_bound
chunk_specs = split_indexer_prefill_chunks(
compressed_seq_lens_cpu[num_decodes:],
prefill_query_lens_cpu,
self.max_prefill_buffer_size,
max_logits_bytes,
request_offset=num_decodes,
)
chunks = []
for req_slice, query_slice in chunk_specs:
metadata = build_prefill_chunk_metadata(
req_slice.start,
req_slice.stop,
query_start_loc,
query_start_loc_cpu,
seq_lens,
compressed_seq_lens,
compressed_seq_lens_cpu,
common_attn_metadata.block_table_tensor,
self.compress_ratio,
query_slice=query_slice,
skip_kv_gather=query_slice.start > 0,
)
# Skip when total_seq_lens is 0 (i.e., no compressed token).
if metadata is not None:
chunks.append(metadata)
prefill_metadata = DeepseekV32IndexerPrefillMetadata(chunks)
decode_metadata = None
if num_decodes > 0:
torch.diff(
common_attn_metadata.query_start_loc[: num_decodes + 1],
out=self.decode_lens_buffer[:num_decodes],
)
decode_lens = self.decode_lens_buffer[:num_decodes]
decode_lens_cpu = torch.diff(
common_attn_metadata.query_start_loc_cpu[: num_decodes + 1]
)
seq_lens = common_attn_metadata.seq_lens[:num_decodes]
block_table = common_attn_metadata.block_table_tensor[:num_decodes, ...]
max_decode_len = int(decode_lens_cpu.max().item())
next_n = 1 + self.num_speculative_tokens
use_native = not self.use_flattening and max_decode_len <= next_n
seq_lens, block_table, decode_lens, batch_size, requires_padding = (
self._prepare_decode_tensors(
seq_lens=seq_lens,
block_table=block_table,
decode_lens=decode_lens,
decode_lens_cpu=decode_lens_cpu,
query_start_loc=common_attn_metadata.query_start_loc[:num_decodes],
num_decodes=num_decodes,
num_decode_tokens=num_decode_tokens,
use_native=use_native,
next_n=next_n,
max_decode_len=max_decode_len,
)
)
# For DeepseekV4 (compress_ratio > 1), the indexer KV cache stores
# compressed tokens. Convert uncompressed seq_lens to compressed.
if self.compress_ratio > 1:
# True iff seq_lens aliases decode_seq_lens_buffer (flatten or
# native wrote it); False iff it aliases common_attn_metadata.
seq_lens_is_local_view = (use_native and next_n > 1) or (
not use_native and max_decode_len > 1
)
if seq_lens_is_local_view:
seq_lens //= self.compress_ratio
else:
# Copy to avoid mutating shared state; keeps CG address stable.
self.expanded_seq_lens_buffer[:num_decodes] = (
seq_lens // self.compress_ratio
)
self.expanded_seq_lens_buffer[num_decodes:num_decode_tokens] = 0
seq_lens = self.expanded_seq_lens_buffer[:num_decode_tokens]
# Non-MTP: deep_gemm paged MQA logits requires 2D context_lens
# (csrc/apis/attention.hpp). Unsqueeze to (B, 1) so downstream
# kernels see the same (B, next_n) layout as the MTP path.
if seq_lens.dim() == 1:
seq_lens = seq_lens.unsqueeze(-1)
seq_lens = seq_lens.contiguous()
# DeepGEMM is required for the paged MQA logits on CUDA devices
if current_platform.is_cuda() and has_deep_gemm():
self.scheduler_metadata_buffer[:] = get_paged_mqa_logits_metadata(
seq_lens,
self.kv_cache_spec.storage_block_size,
self.num_sms,
)
decode_metadata = DeepSeekV32IndexerDecodeMetadata(
block_table=block_table,
seq_lens=seq_lens,
decode_lens=decode_lens,
requires_padding=requires_padding,
schedule_metadata=self.scheduler_metadata_buffer,
)
attn_metadata = DeepseekV32IndexerMetadata(
seq_lens=common_attn_metadata.seq_lens,
max_seq_len=common_attn_metadata.max_seq_len,
slot_mapping=compressed_slot_mapping,
num_decodes=num_decodes,
num_decode_tokens=num_decode_tokens,
num_prefills=num_prefills,
num_prefill_tokens=num_prefill_tokens,
prefill=prefill_metadata,
decode=decode_metadata,
)
return attn_metadata
def build_prefill_chunk_metadata(
start_idx: int,
end_idx: int,
query_start_loc: torch.Tensor,
query_start_loc_cpu: torch.Tensor,
uncompressed_seq_lens: torch.Tensor,
compressed_seq_lens: torch.Tensor,
compressed_seq_lens_cpu: torch.Tensor,
block_table: torch.Tensor,
compress_ratio: int,
query_slice: slice | None = None,
skip_kv_gather: bool = False,
) -> DeepseekV32IndexerPrefillChunkMetadata | None:
total_seq_lens = compressed_seq_lens_cpu[start_idx:end_idx].sum().item()
if total_seq_lens == 0:
return None
num_reqs = end_idx - start_idx
device = block_table.device
token_to_seq = torch.empty(total_seq_lens, dtype=torch.int32, device=device)
cu_seq_lens = torch.empty(num_reqs + 1, dtype=torch.int32, device=device)
# Assigning to slice avoids cpu sync.
cu_seq_lens[:1] = 0
torch.cumsum(compressed_seq_lens[start_idx:end_idx], dim=0, out=cu_seq_lens[1:])
query_start_loc = (
query_start_loc[start_idx : end_idx + 1] - query_start_loc[start_idx]
)
total_query_len = int(
(query_start_loc_cpu[end_idx] - query_start_loc_cpu[start_idx]).item()
)
if query_slice is not None:
qs_start = query_slice.start
qs_stop = query_slice.stop
else:
qs_start = 0
qs_stop = total_query_len
output_query_len = qs_stop - qs_start
cu_seq_len_ks = torch.empty(output_query_len, dtype=torch.int32, device=device)
cu_seq_len_ke = torch.empty(output_query_len, dtype=torch.int32, device=device)
_build_prefill_chunk_metadata_kernel[(num_reqs,)](
query_start_loc,
uncompressed_seq_lens[start_idx:end_idx],
cu_seq_lens,
token_to_seq,
cu_seq_len_ks,
cu_seq_len_ke,
qs_start,
qs_stop,
BLOCK_SIZE=1024,
COMPRESS_RATIO=compress_ratio,
)
token_start = query_start_loc_cpu[start_idx].item()
if query_slice is not None:
token_end = token_start + qs_stop
token_start = token_start + qs_start
skip_kv_gather = skip_kv_gather or qs_start > 0
else:
token_end = query_start_loc_cpu[end_idx].item()
return DeepseekV32IndexerPrefillChunkMetadata(
cu_seqlen_ks=cu_seq_len_ks,
cu_seqlen_ke=cu_seq_len_ke,
cu_seq_lens=cu_seq_lens,
token_to_seq=token_to_seq,
total_seq_lens=total_seq_lens,
block_table=block_table[start_idx:end_idx],
token_start=token_start,
token_end=token_end,
num_reqs=num_reqs,
skip_kv_gather=skip_kv_gather,
)
@triton.jit
def _build_prefill_chunk_metadata_kernel(
# Inputs
query_start_loc_ptr,
uncompressed_seq_lens_ptr,
cu_compressed_seq_lens_ptr,
# Outputs
token_to_seq_ptr,
cu_compressed_seq_len_ks_ptr,
cu_compressed_seq_len_ke_ptr,
query_slice_start,
query_slice_stop,
BLOCK_SIZE: tl.constexpr,
COMPRESS_RATIO: tl.constexpr,
):
batch_idx = tl.program_id(0)
query_start = tl.load(query_start_loc_ptr + batch_idx)
query_end = tl.load(query_start_loc_ptr + batch_idx + 1)
query_len = query_end - query_start
seq_start = tl.load(cu_compressed_seq_lens_ptr + batch_idx)
seq_end = tl.load(cu_compressed_seq_lens_ptr + batch_idx + 1)
compressed_seq_len = seq_end - seq_start
uncompressed_seq_len = tl.load(uncompressed_seq_lens_ptr + batch_idx)
start_pos = uncompressed_seq_len - query_len
for i in range(0, query_len, BLOCK_SIZE):
offset = i + tl.arange(0, BLOCK_SIZE)
abs_pos = query_start + offset
mask = (
(offset < query_len)
& (abs_pos >= query_slice_start)
& (abs_pos < query_slice_stop)
)
out_pos = abs_pos - query_slice_start
# Compute cu_seq_len_ks
tl.store(cu_compressed_seq_len_ks_ptr + out_pos, seq_start, mask=mask)
# Compute cu_seq_len_ke
seq_len_per_token = (start_pos + 1 + offset) // COMPRESS_RATIO
tl.store(
cu_compressed_seq_len_ke_ptr + out_pos,
seq_start + seq_len_per_token,
mask=mask,
)
# Compute token_to_seq
for i in range(0, compressed_seq_len, BLOCK_SIZE):
offset = i + tl.arange(0, BLOCK_SIZE)
mask = offset < compressed_seq_len
tl.store(token_to_seq_ptr + seq_start + offset, batch_idx, mask=mask)

View File

@@ -1,13 +0,0 @@
chunk_size: 256
local_cpu: true
max_local_cpu_size: 512.0
enable_lazy_memory_allocator: true
lazy_memory_initial_ratio: 0.2
remote_url: "redis://10.66.0.100:6379"
remote_serde: naive
remote_ttl: 1800
save_decode_cache: true
use_layerwise: true
save_unfull_chunk: true
blocking_timeout_secs: 60
cache_policy: LRU

View File

@@ -1,10 +0,0 @@
chunk_size: 256
local_cpu: true
max_local_cpu_size: 384.0
save_decode_cache: true
enable_lazy_memory_allocator: true
lazy_memory_initial_ratio: 1.0
use_gpu_connector_v3: true
remote_url: "redis://10.66.0.100:6379"
remote_serde: naive
remote_ttl: 3600

61
minimax_m2_parser.py Normal file
View File

@@ -0,0 +1,61 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""
MiniMax M2 Parser - A unified parser for MiniMax M2 models.
This parser combines the existing MiniMaxM2ReasoningParser and
MinimaxM2ToolParser into a single unified interface by delegating
to those implementations.
"""
from vllm.logger import init_logger
from vllm.parser.abstract_parser import DelegatingParser
from vllm.reasoning.minimax_m2_reasoning_parser import MiniMaxM2ReasoningParser
from vllm.tokenizers import TokenizerLike
from vllm.tool_parsers.abstract_tool_parser import (
Tool,
)
from vllm.tool_parsers.minimax_m2_tool_parser import MinimaxM2ToolParser
logger = init_logger(__name__)
class MiniMaxM2Parser(DelegatingParser):
"""
Unified parser for MiniMax M2 models that handles both reasoning
extraction and tool call parsing.
This parser delegates to the existing implementations:
- MiniMaxM2ReasoningParser for reasoning extraction
- MinimaxM2ToolParser for tool call parsing
MiniMax M2 models have two special behaviors:
1. Reasoning: They don't generate <think> start token, only </think> end
token. All content before </think> is reasoning, content after is the
actual response.
2. Tool Calls: They use <minimax:tool_call>...</minimax:tool_call> tags
with <invoke name="...">...</invoke> and <parameter name="...">...</parameter>
syntax.
"""
# Class-level parser classes for compatibility
reasoning_parser_cls = MiniMaxM2ReasoningParser
tool_parser_cls = MinimaxM2ToolParser
def __init__(
self,
tokenizer: TokenizerLike,
tools: list[Tool] | None = None,
*args,
**kwargs,
):
super().__init__(tokenizer, *args, **kwargs)
# Initialize the underlying parsers
self._reasoning_parser = MiniMaxM2ReasoningParser(tokenizer, *args, **kwargs)
self._tool_parser = MinimaxM2ToolParser(tokenizer, tools)
logger.debug(
"vLLM Successfully initialized parser %s!", self.__class__.__name__
)

852
minimax_tool_parser.py Normal file
View File

@@ -0,0 +1,852 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
import json
from collections.abc import Sequence
from typing import Any
import regex as re
from vllm.entrypoints.chat_utils import make_tool_call_id
from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionRequest,
)
from vllm.entrypoints.openai.engine.protocol import (
DeltaFunctionCall,
DeltaMessage,
DeltaToolCall,
ExtractedToolCallInformation,
FunctionCall,
ToolCall,
)
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 extract_intermediate_diff
logger = init_logger(__name__)
class MinimaxToolParser(ToolParser):
def __init__(self, tokenizer: TokenizerLike, tools: list[Tool] | None = None, **kwargs):
super().__init__(tokenizer, tools)
# Initialize streaming state for tracking tool call progress
self.streaming_state: dict[str, Any] = {
"current_tool_index": -1, # Index of current tool being processed
"tool_ids": [], # List of tool call IDs
"sent_tools": [], # List of tools that have been sent
}
# Define tool call tokens and patterns
self.tool_call_start_token = "<tool_calls>"
self.tool_call_end_token = "</tool_calls>"
self.tool_call_regex = re.compile(
r"<tool_calls>(.*?)</tool_calls>|<tool_calls>(.*)", re.DOTALL
)
self.thinking_tag_pattern = r"<think>(.*?)</think>"
self.tool_name_pattern = re.compile(r'"name":\s*"([^"]+)"')
self.tool_args_pattern = re.compile(r'"arguments":\s*')
# Buffer for handling partial tool calls during streaming
self.pending_buffer = ""
self.in_thinking_tag = False
if not self.model_tokenizer:
raise ValueError(
"The model tokenizer must be passed to the ToolParser "
"constructor during construction."
)
# Get token IDs for tool call start/end tokens
self.tool_call_start_token_id = self.vocab.get(self.tool_call_start_token)
self.tool_call_end_token_id = self.vocab.get(self.tool_call_end_token)
if self.tool_call_start_token_id is None or self.tool_call_end_token_id is None:
logger.warning(
"Minimax Tool parser could not locate tool call start/end "
"tokens in the tokenizer. Falling back to string matching."
)
def preprocess_model_output(self, model_output: str) -> str:
"""
Preprocess model output by removing tool calls from thinking tags.
Args:
model_output: Raw model output string
Returns:
Preprocessed model output with tool calls removed from thinking tags
"""
def remove_tool_calls_from_think(match):
think_content = match.group(1)
cleaned_content = re.sub(
r"<tool_calls>.*?</tool_calls>", "", think_content, flags=re.DOTALL
)
return f"<think>{cleaned_content}</think>"
return re.sub(
self.thinking_tag_pattern,
remove_tool_calls_from_think,
model_output,
flags=re.DOTALL,
)
def _clean_duplicate_braces(self, args_text: str) -> str:
"""
Clean duplicate closing braces from arguments text.
Args:
args_text: Raw arguments text
Returns:
Cleaned arguments text with proper JSON formatting
"""
args_text = args_text.strip()
if not args_text:
return args_text
try:
json.loads(args_text)
return args_text
except json.JSONDecodeError:
pass
while args_text.endswith("}}"):
candidate = args_text[:-1]
try:
json.loads(candidate)
return candidate
except json.JSONDecodeError:
args_text = candidate
return args_text
def _clean_delta_braces(self, delta_text: str) -> str:
"""
Clean delta text by removing excessive closing braces.
Args:
delta_text: Delta text to clean
Returns:
Cleaned delta text
"""
if not delta_text:
return delta_text
delta_stripped = delta_text.strip()
if delta_stripped and all(c in "}\n\r\t " for c in delta_stripped):
brace_count = delta_stripped.count("}")
if brace_count > 1:
return "}\n" if delta_text.endswith("\n") else "}"
return delta_text
def extract_tool_calls(
self,
model_output: str,
request: ChatCompletionRequest,
) -> ExtractedToolCallInformation:
"""
Extract tool calls from model output for non-streaming mode.
Args:
model_output: Complete model output
request: Chat completion request
Returns:
ExtractedToolCallInformation containing tool calls and content
"""
processed_output = self.preprocess_model_output(model_output)
if self.tool_call_start_token not in processed_output:
return ExtractedToolCallInformation(
tools_called=False, tool_calls=[], content=model_output
)
try:
function_call_tuples = self.tool_call_regex.findall(processed_output)
raw_function_calls = []
for match in function_call_tuples:
tool_call_content = match[0] if match[0] else match[1]
if tool_call_content.strip():
lines = tool_call_content.strip().split("\n")
for line in lines:
line = line.strip()
if line and line.startswith("{") and line.endswith("}"):
try:
parsed_call = json.loads(line)
raw_function_calls.append(parsed_call)
except json.JSONDecodeError:
continue
tool_calls = []
for function_call in raw_function_calls:
if "name" in function_call and "arguments" in function_call:
tool_calls.append(
ToolCall(
type="function",
function=FunctionCall(
name=function_call["name"],
arguments=json.dumps(
function_call["arguments"], ensure_ascii=False
),
),
)
)
processed_pos = processed_output.find(self.tool_call_start_token)
if processed_pos != -1:
processed_content = processed_output[:processed_pos].strip()
if processed_content:
lines = processed_content.split("\n")
for line in reversed(lines):
line = line.strip()
if line:
pos = model_output.find(line)
if pos != -1:
content = model_output[: pos + len(line)]
break
else:
content = ""
else:
content = ""
else:
content = model_output
return ExtractedToolCallInformation(
tools_called=len(tool_calls) > 0,
tool_calls=tool_calls,
content=content.strip() if content.strip() else None,
)
except Exception:
logger.exception(
"An unexpected error occurred during tool call extraction."
)
return ExtractedToolCallInformation(
tools_called=False, tool_calls=[], content=model_output
)
def _update_thinking_state(self, text: str) -> None:
"""
Update the thinking tag state based on text content.
Args:
text: Text to analyze for thinking tags
"""
open_count = text.count("<think>")
close_count = text.count("</think>")
self.in_thinking_tag = open_count > close_count or (
open_count == close_count and text.endswith("</think>")
)
def _is_potential_tag_start(self, text: str) -> bool:
"""
Check if text might be the start of a tool call tag.
Args:
text: Text to check
Returns:
True if text could be the start of a tool call tag
"""
for tag in [self.tool_call_start_token, self.tool_call_end_token]:
if any(
tag.startswith(text[-i:])
for i in range(1, min(len(text) + 1, len(tag)))
):
return True
return False
def _should_buffer_content(self, delta_text: str) -> bool:
"""
Determine if content should be buffered for later processing.
Args:
delta_text: Delta text to check
Returns:
True if content should be buffered
"""
if self.in_thinking_tag:
return False
return bool(
self.pending_buffer
or self.tool_call_start_token in delta_text
or self.tool_call_end_token in delta_text
or delta_text.startswith("<")
)
def _split_content_for_buffering(self, delta_text: str) -> tuple[str, str]:
"""
Split delta text into safe content and potential tag content.
Args:
delta_text: Delta text to split
Returns:
Tuple of (safe_content, potential_tag_content)
"""
if self.in_thinking_tag:
return delta_text, ""
for tag in [self.tool_call_start_token, self.tool_call_end_token]:
for i in range(1, len(tag)):
tag_prefix = tag[:i]
pos = delta_text.rfind(tag_prefix)
if pos != -1 and tag.startswith(delta_text[pos:]):
return delta_text[:pos], delta_text[pos:]
return delta_text, ""
def _process_buffer(self, new_content: str) -> str:
"""
Process buffered content and return output content.
Args:
new_content: New content to add to buffer
Returns:
Processed output content
"""
self.pending_buffer += new_content
output_content = ""
if self.in_thinking_tag:
output_content = self.pending_buffer
self.pending_buffer = ""
return output_content
while self.pending_buffer:
start_pos = self.pending_buffer.find(self.tool_call_start_token)
end_pos = self.pending_buffer.find(self.tool_call_end_token)
if start_pos != -1 and (end_pos == -1 or start_pos < end_pos):
tag_pos, tag_len = start_pos, len(self.tool_call_start_token)
elif end_pos != -1:
tag_pos, tag_len = end_pos, len(self.tool_call_end_token)
else:
if self._is_potential_tag_start(self.pending_buffer):
break
output_content += self.pending_buffer
self.pending_buffer = ""
break
output_content += self.pending_buffer[:tag_pos]
self.pending_buffer = self.pending_buffer[tag_pos + tag_len :]
return output_content
def _reset_streaming_state(self) -> None:
"""Reset the streaming state to initial values."""
self.streaming_state = {
"current_tool_index": -1,
"tool_ids": [],
"sent_tools": [],
}
def _advance_to_next_tool(self) -> None:
"""Advance to the next tool in the streaming sequence."""
self.streaming_state["current_tool_index"] = (
int(self.streaming_state["current_tool_index"]) + 1
)
def _set_current_tool_index(self, index: int) -> None:
"""
Set the current tool index.
Args:
index: Tool index to set
"""
self.streaming_state["current_tool_index"] = index
def _get_current_tool_index(self) -> int:
"""
Get the current tool index.
Returns:
Current tool index
"""
return int(self.streaming_state["current_tool_index"])
def _get_next_unsent_tool_index(self, tool_count: int) -> int:
"""
Get the index of the next unsent tool.
Args:
tool_count: Total number of tools
Returns:
Index of next unsent tool, or -1 if all tools sent
"""
sent_tools = list(self.streaming_state["sent_tools"])
for i in range(tool_count):
if i < len(sent_tools):
if not sent_tools[i]["sent_name"]:
return i
else:
return i
return -1
def _ensure_state_arrays(self, tool_count: int) -> None:
"""
Ensure state arrays have sufficient capacity for tool_count tools.
Args:
tool_count: Number of tools to prepare for
"""
sent_tools = list(self.streaming_state["sent_tools"])
tool_ids = list(self.streaming_state["tool_ids"])
while len(sent_tools) < tool_count:
sent_tools.append(
{
"sent_name": False,
"sent_arguments": "",
"id": make_tool_call_id(),
}
)
while len(tool_ids) < tool_count:
tool_ids.append(None)
self.streaming_state["sent_tools"] = sent_tools
self.streaming_state["tool_ids"] = tool_ids
def _detect_tools_in_text(self, text: str) -> int:
"""
Detect the number of tools in text by counting name patterns.
Args:
text: Text to analyze
Returns:
Number of tools detected
"""
matches = self.tool_name_pattern.findall(text)
return len(matches)
def _find_tool_boundaries(self, text: str) -> list[tuple[int, int]]:
"""
Find the boundaries of tool calls in text.
Args:
text: Text to analyze
Returns:
List of (start, end) positions for tool calls
"""
boundaries = []
i = 0
while i < len(text):
if text[i] == "{":
start = i
depth = 0
has_name = False
has_arguments = False
while i < len(text):
if text[i] == "{":
depth += 1
elif text[i] == "}":
depth -= 1
if depth == 0:
end = i + 1
segment = text[start:end]
if '"name"' in segment and '"arguments"' in segment:
boundaries.append((start, end))
break
if not has_name and '"name"' in text[start : i + 1]:
has_name = True
if not has_arguments and '"arguments"' in text[start : i + 1]:
has_arguments = True
i += 1
if depth > 0 and has_name:
boundaries.append((start, i))
else:
i += 1
return boundaries
def _extract_tool_args(self, tool_content: str, args_match: re.Match[str]) -> str:
"""
Extract tool arguments from tool content.
Args:
tool_content: Tool call content
args_match: Regex match for arguments pattern
Returns:
Extracted arguments as string
"""
args_start_pos = args_match.end()
remaining_content = tool_content[args_start_pos:]
if remaining_content.strip().startswith("{"):
depth = 0
for i, char in enumerate(remaining_content):
if char == "{":
depth += 1
elif char == "}":
depth -= 1
if depth == 0:
return remaining_content[: i + 1]
else:
args_end = remaining_content.find("}")
if args_end > 0:
return remaining_content[:args_end].strip()
return remaining_content.rstrip("}").strip()
def _get_current_tool_content(
self, text: str, tool_index: int
) -> tuple[str | None, str | None]:
"""
Get the content of a specific tool by index.
Args:
text: Text containing tool calls
tool_index: Index of tool to extract
Returns:
Tuple of (tool_name, tool_arguments) or (None, None) if not found
"""
boundaries = self._find_tool_boundaries(text)
if tool_index >= len(boundaries):
return None, None
start, end = boundaries[tool_index]
tool_content = text[start:end]
name_match = self.tool_name_pattern.search(tool_content)
name = name_match.group(1) if name_match else None
args_match = self.tool_args_pattern.search(tool_content)
if args_match:
try:
args_text = self._extract_tool_args(tool_content, args_match)
return name, args_text
except Exception:
remaining_content = tool_content[args_match.end() :]
args_text = remaining_content.rstrip("}").strip()
return name, args_text
return name, None
def _handle_tool_name_streaming(
self, tool_content: str, tool_count: int
) -> DeltaMessage | None:
"""
Handle streaming of tool names.
Args:
tool_content: Content containing tool calls
tool_count: Total number of tools
Returns:
DeltaMessage with tool name or None if no tool to stream
"""
next_idx = self._get_next_unsent_tool_index(tool_count)
if next_idx == -1:
return None
boundaries = self._find_tool_boundaries(tool_content)
if next_idx >= len(boundaries):
return None
tool_name, _ = self._get_current_tool_content(tool_content, next_idx)
if not tool_name:
return None
self._set_current_tool_index(next_idx)
sent_tools = list(self.streaming_state["sent_tools"])
tool_ids = list(self.streaming_state["tool_ids"])
tool_id = sent_tools[next_idx]["id"]
tool_ids[next_idx] = tool_id
sent_tools[next_idx]["sent_name"] = True
self.streaming_state["sent_tools"] = sent_tools
self.streaming_state["tool_ids"] = tool_ids
return DeltaMessage(
tool_calls=[
DeltaToolCall(
index=next_idx,
type="function",
id=tool_id,
function=DeltaFunctionCall(name=tool_name).model_dump(
exclude_none=True
),
)
]
)
def _handle_tool_args_streaming(
self, tool_content: str, tool_count: int
) -> DeltaMessage | None:
"""
Handle streaming of tool arguments.
Args:
tool_content: Content containing tool calls
tool_count: Total number of tools
Returns:
DeltaMessage with tool arguments or None if no arguments to stream
"""
current_idx = self._get_current_tool_index()
if current_idx < 0 or current_idx >= tool_count:
return None
tool_name, tool_args = self._get_current_tool_content(tool_content, current_idx)
if not tool_name or tool_args is None:
return None
sent_tools = list(self.streaming_state["sent_tools"])
if not sent_tools[current_idx]["sent_name"]:
return None
clean_args = self._clean_duplicate_braces(tool_args)
sent_args = sent_tools[current_idx]["sent_arguments"]
if clean_args != sent_args:
if sent_args and clean_args.startswith(sent_args):
args_delta = extract_intermediate_diff(clean_args, sent_args)
if args_delta:
args_delta = self._clean_delta_braces(args_delta)
sent_tools[current_idx]["sent_arguments"] = clean_args
self.streaming_state["sent_tools"] = sent_tools
if clean_args.endswith("}"):
self._advance_to_next_tool()
return DeltaMessage(
tool_calls=[
DeltaToolCall(
index=current_idx,
function=DeltaFunctionCall(
arguments=args_delta
).model_dump(exclude_none=True),
)
]
)
elif not sent_args and clean_args:
clean_args_delta = self._clean_delta_braces(clean_args)
sent_tools[current_idx]["sent_arguments"] = clean_args
self.streaming_state["sent_tools"] = sent_tools
if clean_args.endswith("}"):
self._advance_to_next_tool()
return DeltaMessage(
tool_calls=[
DeltaToolCall(
index=current_idx,
function=DeltaFunctionCall(
arguments=clean_args_delta
).model_dump(exclude_none=True),
)
]
)
return None
def _is_end_tool_calls(self, current_text: str) -> bool:
if self.tool_call_end_token not in current_text:
return False
end_token_positions = []
search_start = 0
while True:
pos = current_text.find(self.tool_call_end_token, search_start)
if pos == -1:
break
end_token_positions.append(pos)
search_start = pos + 1
think_regions = []
for match in re.finditer(
self.thinking_tag_pattern, current_text, flags=re.DOTALL
):
think_regions.append((match.start(), match.end()))
for pos in end_token_positions:
in_think = any(
pos >= t_start and pos < t_end for t_start, t_end in think_regions
)
if not in_think:
return True
return False
def extract_tool_calls_streaming(
self,
previous_text: str,
current_text: str,
delta_text: str,
previous_token_ids: Sequence[int],
current_token_ids: Sequence[int],
delta_token_ids: Sequence[int],
request: ChatCompletionRequest,
) -> DeltaMessage | None:
self._update_thinking_state(current_text)
if self.in_thinking_tag:
return DeltaMessage(content=delta_text)
if self._should_buffer_content(delta_text):
buffered_output = self._process_buffer(delta_text)
return DeltaMessage(content=buffered_output) if buffered_output else None
if self._is_end_tool_calls(current_text):
return DeltaMessage(content=delta_text)
safe_content, potential_tag = self._split_content_for_buffering(delta_text)
if potential_tag:
self.pending_buffer += potential_tag
return DeltaMessage(content=safe_content) if safe_content else None
processed_current_text = self.preprocess_model_output(current_text)
if self.tool_call_start_token not in processed_current_text:
if (
self.tool_call_end_token in delta_text
and self.tool_call_start_token in current_text
):
return None
if delta_text.strip() == "" and self.tool_call_start_token in current_text:
return None
if (
self._get_current_tool_index() != -1
and self.tool_call_end_token in current_text
):
self._reset_streaming_state()
return DeltaMessage(content=delta_text)
if (
self.tool_call_start_token_id is not None
and self.tool_call_start_token_id in delta_token_ids
and len(delta_token_ids) == 1
):
return None
original_tool_start = self._find_tool_start_outside_thinking(current_text)
if original_tool_start is None:
return None
content_before_tools = self._extract_content_before_tools(
current_text, delta_text, original_tool_start
)
if content_before_tools:
return DeltaMessage(content=content_before_tools)
try:
tool_content = self._extract_tool_content(current_text, original_tool_start)
current_tools_count = self._detect_tools_in_text(tool_content)
if current_tools_count == 0:
return None
if self._get_current_tool_index() == -1:
self._reset_streaming_state()
self._ensure_state_arrays(current_tools_count)
return self._handle_tool_name_streaming(
tool_content, current_tools_count
) or self._handle_tool_args_streaming(tool_content, current_tools_count)
except Exception:
logger.exception(
"An unexpected error occurred ", "during streaming tool call handling."
)
return None
def _find_tool_start_outside_thinking(self, current_text: str) -> int | None:
"""
Find the start position of tool calls outside of thinking tags.
Args:
current_text: Current text to search
Returns:
Position of tool call start or None if not found
"""
search_start = 0
while True:
pos = current_text.find(self.tool_call_start_token, search_start)
if pos == -1:
return None
think_regions = [
(m.start(), m.end())
for m in re.finditer(
r"<think>(.*?)</think>", current_text, flags=re.DOTALL
)
]
in_think = any(
pos >= t_start and pos < t_end for t_start, t_end in think_regions
)
if not in_think:
return pos
search_start = pos + 1
def _extract_content_before_tools(
self, current_text: str, delta_text: str, tool_start: int
) -> str | None:
"""
Extract content that appears before tool calls.
Args:
current_text: Current text
delta_text: Delta text
tool_start: Start position of tools
Returns:
Content before tools or None
"""
if tool_start > 0:
delta_start_pos = len(current_text) - len(delta_text)
if delta_start_pos < tool_start:
content_part = delta_text
if delta_start_pos + len(delta_text) > tool_start:
content_part = delta_text[: tool_start - delta_start_pos]
return content_part if content_part else None
return None
def _extract_tool_content(self, current_text: str, tool_start: int) -> str:
"""
Extract tool content from current text starting at tool_start.
Args:
current_text: Current text
tool_start: Start position of tool calls
Returns:
Extracted tool content
"""
tool_content_start = tool_start + len(self.tool_call_start_token)
tool_content = current_text[tool_content_start:]
end_pos = tool_content.find(self.tool_call_end_token)
if end_pos != -1:
tool_content = tool_content[:end_pos]
return tool_content

File diff suppressed because it is too large Load Diff

View File

@@ -1,942 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
import functools
import pickle
import sys
import threading
import time
from contextlib import contextmanager
from dataclasses import dataclass, field
from multiprocessing import shared_memory
from pickle import PickleBuffer
from typing import TYPE_CHECKING, Any, cast
from unittest.mock import patch
import torch
import torch.distributed as dist
import zmq
from torch.distributed import ProcessGroup
from zmq import ( # type: ignore
IPV6, # type: ignore
PUB,
SUB,
SUBSCRIBE,
XPUB,
XPUB_VERBOSE,
Context,
)
import vllm.envs as envs
from vllm.distributed.utils import StatelessProcessGroup, sched_yield
from vllm.logger import init_logger
from vllm.platforms import current_platform
from vllm.utils.network_utils import (
get_ip,
get_open_port,
get_open_zmq_inproc_path,
get_open_zmq_ipc_path,
is_valid_ipv6_address,
)
logger = init_logger(__name__)
SPINLOOP_EXT_ENABLED = False
if envs.VLLM_USE_SPINLOOP_EXT:
try:
from vllm.spinloop import spinloop
SPINLOOP_EXT_ENABLED = True
except ImportError:
logger.warning(
"spinloop extension could not be loaded, disabling VLLM_USE_SPINLOOP_EXT!"
)
SPINLOOP_TIMEOUT_SECONDS = 0.1
if TYPE_CHECKING:
from _typeshed import SizedBuffer
VLLM_RINGBUFFER_WARNING_INTERVAL = envs.VLLM_RINGBUFFER_WARNING_INTERVAL
from_bytes_big = functools.partial(int.from_bytes, byteorder="big")
# Memory fence for cross-process shared memory visibility.
# Required for correct producer-consumer synchronization when using
# shared memory without locks.
_memory_fence_lock = threading.Lock()
def memory_fence():
"""
Full memory barrier for shared memory synchronization.
Ensures all prior memory writes are visible to other processes before
any subsequent reads. This is critical for lock-free producer-consumer
patterns using shared memory.
Implementation acquires and immediately releases a lock. Python's
threading.Lock provides sequentially consistent memory barrier semantics
across all major platforms (POSIX, Windows). This is a lightweight
operation (~20ns) that guarantees:
- All stores before the barrier are visible to other threads/processes
- All loads after the barrier see the latest values
"""
# Lock acquire/release provides full memory barrier semantics.
# Using context manager ensures lock release even on exceptions.
with _memory_fence_lock:
pass
def to_bytes_big(value: int, size: int) -> bytes:
return value.to_bytes(size, byteorder="big")
LONG_WAIT_TIME_LOG_MSG = (
"No available shared memory broadcast block found "
"in %d seconds. This typically happens "
"when some processes are hanging or doing some "
"time-consuming work (e.g. compilation, "
"weight/kv cache quantization)."
)
class SpinCondition:
"""
This class implements an interface similar to a threading.Condition. It
allows a writer to notify readers to wake up and read from the shared memory
buffer. This notification is done over a zmq socket.
For optimal performance under load we don't want the readers to need to poll
the zmq socket for every read. So the `wait` method here will return
immediately when reads are frequent, and will only enter "idle mode" and
await a notification on the zmq socket after a period of inactivity. This
allows the readers to spin quickly, hence "SpinCondition".
To support clean shutdown, a separate thread in the reader's process must be
able to wake the reader so that it can exit. A separate cancel() method is
implemented with an in-process socket to allow this interruption.
"""
def __init__(
self,
is_reader: bool,
context: zmq.Context,
notify_address: str,
busy_loop_s: float = 1,
):
self.is_reader = is_reader
if is_reader:
# Time of last shm buffer read
self.last_read = time.monotonic()
# Time to keep busy-looping on the shm buffer before going idle
self.busy_loop_s = busy_loop_s
# Readers subscribe to write notifications
self.local_notify_socket: zmq.Socket = context.socket(SUB)
# Set zmq.CONFLATE to only keep the last message that the socket
# receives. This prevents us from piling up notification messages
# under high load when we aren't polling the socket.
self.local_notify_socket.setsockopt(zmq.CONFLATE, 1)
# Subscribe to all messages on the socket
self.local_notify_socket.setsockopt_string(SUBSCRIBE, "")
self.local_notify_socket.connect(notify_address)
# Readers require a process-local socket to poll for cancellation
cancel_path = get_open_zmq_inproc_path()
self.write_cancel_socket: zmq.Socket = context.socket(zmq.PAIR)
self.write_cancel_socket.bind(cancel_path)
self.read_cancel_socket: zmq.Socket = context.socket(zmq.PAIR)
self.read_cancel_socket.connect(cancel_path)
# Poller allows waiting on either `.notify()` or `.cancel()`
self.poller = zmq.Poller()
self.poller.register(self.read_cancel_socket, zmq.POLLIN)
self.poller.register(self.local_notify_socket, zmq.POLLIN)
else:
# Writer side publishes write notifications
self.local_notify_socket: zmq.Socket = context.socket(PUB) # type: ignore
# Set high water mark to 1 - we don't need to send a massive amount of
# pings during busy operation. PUB sockets will silently drop subsequent
# messages after the high water mark is reached.
self.local_notify_socket.setsockopt(zmq.SNDHWM, 1)
self.local_notify_socket.bind(notify_address)
self.last_read = 0
self.busy_loop_s = 0
self.read_cancel_socket = None
self.write_cancel_socket = None
self.poller = None
def record_read(self):
self.last_read = time.monotonic()
def cancel(self):
# Sends cancellation ping that will cause the reader to wake up.
# This is done from a monitor thread in the same process as the reader.
if self.is_reader:
logger.debug("Canceling waiting reads on SHM Buffer")
self.write_cancel_socket.send(b"\x00")
def wait(self, timeout_ms: int | None = None) -> None:
"""Wait for data on the shared memory buffer.
Yields the scheduler then returns immediately if it has been less than
self.busy_loop_s since the last read.
Otherwise, enters idle mode and awaits a socket ping for at most
`timeout_ms` milliseconds, or indefinitely if timeout_ms is None.
"""
assert self.is_reader, "Only readers can wait"
current_time = time.monotonic()
if current_time <= self.last_read + self.busy_loop_s:
sched_yield()
else:
events = dict(self.poller.poll(timeout=timeout_ms))
if self.read_cancel_socket in events:
logger.debug("Poller received cancel event")
elif self.local_notify_socket in events:
logger.debug("Poller received notify event")
# Since zmq.CONFLATE is set, there will only be one notification
# to read from the socket
self.local_notify_socket.recv(flags=zmq.NOBLOCK, copy=False)
else:
logger.debug("Poller timed out")
def notify(self):
"""Notifies all readers to wake up"""
assert not self.is_reader, "Only writers can notify"
self.local_notify_socket.send(b"\x00")
class ShmRingBuffer:
def __init__(
self,
n_reader: int,
max_chunk_bytes: int,
max_chunks: int,
name: str | None = None,
):
"""
A shared memory ring buffer implementation for broadcast communication.
Essentially, it is a queue where only one will `enqueue` and multiple
will `dequeue`. The max size of each item, together with the max number
of items that can be stored in the buffer are known in advance.
In this case, we don't need to synchronize the access to
the buffer.
Buffer memory layout:
data metadata
| |
| (current_idx) | (current_idx)
v v
+-------------------------------+----------------------------------------+
| chunk0 | chunk1 | ... | chunk | metadata0 | metadata1 | ... | metadata |
+-------------------------------+----------------------------------------+
| max_chunks x max_chunk_bytes | max_chunks x (1 + n_reader) bytes |
metadata memory layout: each byte is a flag, the first byte is the written
flag, and the rest are reader flags. The flags are set to 0 by default.
+--------------+--------------+--------------+-----+--------------+
| written_flag | reader0_flag | reader1_flag | ... | readerN_flag |
+--------------+--------------+--------------+-----+--------------+
The state of metadata is as follows:
(case 1) 0???...???: the block is not written yet, cannot read, can write
(case 2) 1000...000: the block is just written, can read, cannot write
(case 3) 1???...???: the block is written and read by some readers, can read if not read, cannot write
(case 4) 1111...111: the block is written and read by all readers, cannot read, can write
State transition for readers:
When a reader finds a block that it can read (case 2 or 3), it can yield the block for caller to read.
Only after the caller finishes reading the block, the reader can mark the block as read.
Readers only mark the block as read (from 0 to 1), the writer marks the block as ready to read (from 1 to 0).
State transition for writer:
When the writer writes to a block (case 1 or 4), it first resets the written flag to 0, converting either case
to case 1. Then it can yield the block for caller to write. After the caller finishes writing the block, the writer
can reset the reader flags to 0, and mark the block as written (from 0 to 1).
NOTE: the order is important here, first reset the reader flags (so that we are still in case 1), then mark the block as written. The state transition is atomic. If we do it in the reverse order, it will go through case 3 and then back to case 2, and readers might read the intermediate case 3, which is not correct.
During creation, `name` is None and the buffer is created. We can pass the
created object to other processes by pickling it. The other processes will
get the name of the shared memory and open it, so that they can access the
same shared memory buffer.
""" # noqa
self.n_reader = n_reader
self.metadata_size = 1 + n_reader
self.max_chunk_bytes = max_chunk_bytes
self.max_chunks = max_chunks
self.total_bytes_of_buffer = (
self.max_chunk_bytes + self.metadata_size
) * self.max_chunks
self.data_offset = 0
self.metadata_offset = self.max_chunk_bytes * self.max_chunks
if name is None:
# we are creating a buffer
self.is_creator = True
self.shared_memory = shared_memory.SharedMemory(
create=True, size=self.total_bytes_of_buffer
)
assert self.shared_memory.buf is not None, "Buffer was not created"
# initialize the metadata section to 0
with self.shared_memory.buf[self.metadata_offset :] as metadata_buffer:
torch.frombuffer(metadata_buffer, dtype=torch.uint8).fill_(0)
else:
# we are opening an existing buffer
self.is_creator = False
# fix to https://stackoverflow.com/q/62748654/9191338
# Python incorrectly tracks shared memory even if it is not
# created by the process. The following patch is a workaround.
with patch(
"multiprocessing.resource_tracker.register",
lambda *args, **kwargs: None,
):
try:
self.shared_memory = shared_memory.SharedMemory(name=name)
# See https://docs.python.org/3/library/multiprocessing.shared_memory.html # noqa
# Some platforms allocate memory based on page size,
# so the shared memory block size may be larger or equal
# to the requested size. The size parameter is ignored
# when attaching to an existing block.
assert self.shared_memory.size >= self.total_bytes_of_buffer
except FileNotFoundError:
# we might deserialize the object in a different node
# in this case, this object is not used,
# and we should suppress the error
pass
def handle(self):
return (
self.n_reader,
self.max_chunk_bytes,
self.max_chunks,
self.shared_memory.name,
)
def __reduce__(self):
return (
self.__class__,
self.handle(),
)
def __del__(self):
if hasattr(self, "shared_memory"):
self.shared_memory.close()
if self.is_creator:
self.shared_memory.unlink()
@contextmanager
def get_data(self, current_idx: int):
start = self.data_offset + current_idx * self.max_chunk_bytes
end = start + self.max_chunk_bytes
assert self.shared_memory.buf is not None, "Buffer has been closed"
with self.shared_memory.buf[start:end] as buf:
yield buf
@contextmanager
def get_metadata(self, current_idx: int):
start = self.metadata_offset + current_idx * self.metadata_size
end = start + self.metadata_size
assert self.shared_memory.buf is not None, "Buffer has been closed"
with self.shared_memory.buf[start:end] as buf:
yield buf
@dataclass
class Handle:
local_reader_ranks: list[int] = field(default_factory=list)
buffer_handle: tuple[int, int, int, str] | None = None
local_subscribe_addr: str | None = None
local_notify_addr: str | None = None
remote_subscribe_addr: str | None = None
remote_addr_ipv6: bool = False
class MessageQueue:
def __init__(
self,
n_reader, # number of all readers
n_local_reader, # number of local readers through shared memory
local_reader_ranks: list[int] | None = None,
# Default of 24MiB chosen to be large enough to accommodate grammar
# bitmask tensors for large batches (1024 requests).
max_chunk_bytes: int = 1024 * 1024 * 24,
max_chunks: int = 10,
connect_ip: str | None = None,
):
if local_reader_ranks is None:
local_reader_ranks = list(range(n_local_reader))
else:
assert len(local_reader_ranks) == n_local_reader
self.n_local_reader = n_local_reader
n_remote_reader = n_reader - n_local_reader
self.n_remote_reader = n_remote_reader
self.shutting_down = False
context = Context()
if n_local_reader > 0:
# for local readers, we will:
# 1. create a shared memory ring buffer to communicate small data
# 2. create a publish-subscribe socket to communicate large data
self.buffer = ShmRingBuffer(n_local_reader, max_chunk_bytes, max_chunks)
# XPUB is very similar to PUB,
# except that it can receive subscription messages
# to confirm the number of subscribers
self.local_socket = context.socket(XPUB)
# set the verbose option so that we can receive every subscription
# message. otherwise, we will only receive the first subscription
# see http://api.zeromq.org/3-3:zmq-setsockopt for more details
self.local_socket.setsockopt(XPUB_VERBOSE, True)
local_subscribe_addr = get_open_zmq_ipc_path()
logger.debug("Binding to %s", local_subscribe_addr)
self.local_socket.bind(local_subscribe_addr)
self.current_idx = 0
# Create the notification side of the SpinCondition
local_notify_addr = get_open_zmq_ipc_path()
self._spin_condition = SpinCondition(
is_reader=False, context=context, notify_address=local_notify_addr
)
else:
self.buffer = None # type: ignore
local_subscribe_addr = None
self.local_socket = None
self.current_idx = -1
local_notify_addr = None
self._spin_condition = None # type: ignore
remote_addr_ipv6 = False
if n_remote_reader > 0:
# for remote readers, we will:
# create a publish-subscribe socket to communicate large data
if not connect_ip:
connect_ip = get_ip()
self.remote_socket = context.socket(XPUB)
self.remote_socket.setsockopt(XPUB_VERBOSE, True)
remote_subscribe_port = get_open_port()
if is_valid_ipv6_address(connect_ip):
self.remote_socket.setsockopt(IPV6, 1)
remote_addr_ipv6 = True
connect_ip = f"[{connect_ip}]"
socket_addr = f"tcp://{connect_ip}:{remote_subscribe_port}"
self.remote_socket.bind(socket_addr)
remote_subscribe_addr = f"tcp://{connect_ip}:{remote_subscribe_port}"
else:
remote_subscribe_addr = None
self.remote_socket = None
self._is_writer = True
self._is_local_reader = False
self.local_reader_rank = -1
# rank does not matter for remote readers
self._is_remote_reader = False
self.handle = Handle(
local_reader_ranks=local_reader_ranks,
buffer_handle=self.buffer.handle() if self.buffer is not None else None,
local_subscribe_addr=local_subscribe_addr,
local_notify_addr=local_notify_addr,
remote_subscribe_addr=remote_subscribe_addr,
remote_addr_ipv6=remote_addr_ipv6,
)
logger.debug("vLLM message queue communication handle: %s", self.handle)
def export_handle(self) -> Handle:
return self.handle
@staticmethod
def create_from_handle(handle: Handle, rank) -> "MessageQueue":
self = MessageQueue.__new__(MessageQueue)
self.handle = handle
self._is_writer = False
context = Context()
if rank in handle.local_reader_ranks:
assert handle.buffer_handle is not None
self.buffer = ShmRingBuffer(*handle.buffer_handle)
self.current_idx = 0
self.local_reader_rank = handle.local_reader_ranks.index(rank)
self._is_local_reader = True
self._is_remote_reader = False
self.local_socket = context.socket(SUB)
self.local_socket.setsockopt_string(SUBSCRIBE, "")
socket_addr = handle.local_subscribe_addr
logger.debug("Connecting to %s", socket_addr)
self.local_socket.connect(socket_addr)
self.remote_socket = None
assert isinstance(handle.local_notify_addr, str)
self._spin_condition = SpinCondition(
is_reader=True, context=context, notify_address=handle.local_notify_addr
)
else:
self.buffer = None # type: ignore
self.current_idx = -1
self.local_reader_rank = -1
self._is_local_reader = False
self._is_remote_reader = True
self.local_socket = None
self.remote_socket = context.socket(SUB)
self.remote_socket.setsockopt_string(SUBSCRIBE, "")
if handle.remote_addr_ipv6:
self.remote_socket.setsockopt(IPV6, 1)
socket_addr = handle.remote_subscribe_addr
logger.debug("Connecting to %s", socket_addr)
self.remote_socket.connect(socket_addr)
self._spin_condition = None # type: ignore
self.shutting_down = False
return self
def wait_until_ready(self):
"""This is a collective operation. All processes (including the
readers and the writer) should call this function.
"""
if self._is_writer:
# wait for all readers to connect
# local readers
for i in range(self.n_local_reader):
# wait for subscription messages from all local readers
self.local_socket.recv()
if self.n_local_reader > 0:
# send a message to all local readers
# to make sure the publish channel is working
self.local_socket.send(b"READY")
# remote readers
for i in range(self.n_remote_reader):
# wait for subscription messages from all remote readers
self.remote_socket.recv()
if self.n_remote_reader > 0:
# send a message to all remote readers
# to make sure the publish channel is working
self.remote_socket.send(b"READY")
elif self._is_local_reader:
# wait for the writer to send a message
recv = self.local_socket.recv()
assert recv == b"READY"
elif self._is_remote_reader:
# wait for the writer to send a message
recv = self.remote_socket.recv()
assert recv == b"READY"
def shutdown(self):
"""If this is an idle reader, wakes it up so it can clean up and shut
down"""
self.shutting_down = True
if self._spin_condition is not None:
self._spin_condition.cancel()
@contextmanager
def acquire_write(self, timeout: float | None = None):
assert self._is_writer, "Only writers can acquire write"
start_time = time.monotonic()
n_warning = 1
while True:
with self.buffer.get_metadata(self.current_idx) as metadata_buffer:
def check():
memory_fence()
read_count = sum(metadata_buffer[1:])
written_flag = metadata_buffer[0]
return not (written_flag and read_count != self.buffer.n_reader)
if SPINLOOP_EXT_ENABLED and not check():
spinloop(metadata_buffer, check, timeout=SPINLOOP_TIMEOUT_SECONDS)
if not check():
# this block is written and not read by all readers
# for writers, `self.current_idx` is the next block to write
# if this block is not ready to write,
# we need to wait until it is read by all readers
# Release the processor to other threads
sched_yield()
# if we time out, raise an exception
elapsed = time.monotonic() - start_time
if timeout is not None and elapsed > timeout:
raise TimeoutError
# if we wait for a long time, log a message
if elapsed > VLLM_RINGBUFFER_WARNING_INTERVAL * n_warning:
logger.info(
LONG_WAIT_TIME_LOG_MSG, VLLM_RINGBUFFER_WARNING_INTERVAL
)
n_warning += 1
continue
# found a block that is either
# (1) not written
# (2) read by all readers
# mark the block as not written
metadata_buffer[0] = 0
# let caller write to the buffer
with self.buffer.get_data(self.current_idx) as buf:
yield buf
# caller has written to the buffer
# NOTE: order is important here
# first set the read flags to 0
# then set the written flag to 1
# otherwise, the readers may think they already read the block
for i in range(1, self.buffer.n_reader + 1):
# set read flag to 0, meaning it is not read yet
metadata_buffer[i] = 0
# Memory fence here ensures the order of the buffer and flag
# writes. This guarantees that when `metadata_buffer[0] = 1` is
# visible to readers, `buf` can be completely ready. Without
# this, some CPU architectures with weak ordering may incur
# memory inconsistency.
memory_fence()
# mark the block as written
metadata_buffer[0] = 1
# Memory fence ensures the write is visible to readers on other cores
# before we proceed. Without this, readers may spin indefinitely
# waiting for a write that's stuck in our CPU's store buffer.
memory_fence()
self.current_idx = (self.current_idx + 1) % self.buffer.max_chunks
break
class ReadTimeoutWithWarnings:
def __init__(self, timeout: float | None, should_warn: bool) -> None:
self.started = time.monotonic()
self.deadline = sys.maxsize if timeout is None else self.started + timeout
# if should_warn, we need to wake up periodically to log
self.warning_wait_time_ms: int | None = (
VLLM_RINGBUFFER_WARNING_INTERVAL * 1000 if should_warn else None
)
self._should_warn = should_warn
self.n_warning = 1
self.timeout = timeout
def timeout_ms(self) -> int | None:
"""Returns a timeout that is:
- min(time to deadline, time to next warning) if we're logging warnings
- time to deadline, if we're not logging warnings
- None if the timeout is None and we're not logging warnings
- raise TimeoutError if we are past the deadline
"""
warning_wait_time = self.warning_wait_time_ms
if self.timeout is None:
return warning_wait_time
time_left_ms = int((self.deadline - time.monotonic()) * 1000)
if time_left_ms <= 0:
raise TimeoutError
if warning_wait_time and warning_wait_time < time_left_ms:
return warning_wait_time
return time_left_ms
def should_warn(self) -> bool:
"""Returns true if it's time to log a warning for a timeout that is not
indefinite"""
if self._should_warn:
elapsed = time.monotonic() - self.started
if elapsed >= VLLM_RINGBUFFER_WARNING_INTERVAL * self.n_warning:
self.n_warning += 1
return True
return False
@contextmanager
def acquire_read(
self,
timeout: float | None = None,
indefinite: bool = False,
):
assert self._is_local_reader, "Only readers can acquire read"
read_timeout = self.ReadTimeoutWithWarnings(
timeout=timeout, should_warn=not indefinite
)
with self.buffer.get_metadata(self.current_idx) as metadata_buffer:
while True:
def check():
memory_fence()
read_flag = metadata_buffer[self.local_reader_rank + 1]
written_flag = metadata_buffer[0]
return not (not written_flag or read_flag)
if SPINLOOP_EXT_ENABLED and not check():
spinloop(
metadata_buffer[0 : self.local_reader_rank + 1],
check,
timeout=SPINLOOP_TIMEOUT_SECONDS,
)
if not check():
# this block is either
# (1) not written
# (2) already read by this reader
# for readers, `self.current_idx` is the next block to read
# if this block is not ready,
# we need to wait until it is written
self._spin_condition.wait(timeout_ms=read_timeout.timeout_ms())
if self.shutting_down:
raise RuntimeError("cancelled")
# if we wait for a long time, log a message
if read_timeout.should_warn():
logger.info(
LONG_WAIT_TIME_LOG_MSG, VLLM_RINGBUFFER_WARNING_INTERVAL
)
continue
# found a block that is not read by this reader
# let caller read from the buffer
with self.buffer.get_data(self.current_idx) as buf:
yield buf
# caller has read from the buffer
# set the read flag
metadata_buffer[self.local_reader_rank + 1] = 1
# Memory fence ensures the read flag is visible to the writer.
# Without this, writer may not see our read completion and
# could wait indefinitely for all readers to finish.
memory_fence()
self.current_idx = (self.current_idx + 1) % self.buffer.max_chunks
self._spin_condition.record_read()
break
def enqueue(self, obj, timeout: float | None = None):
"""Write to message queue with optional timeout (in seconds)"""
assert self._is_writer, "Only writers can enqueue"
all_buffers: list[SizedBuffer] = [b""]
total_bytes = 6 # 2 bytes for oob buffer count, 4 for main buffer size
def oob_callback(buf: PickleBuffer) -> bool:
raw_buf = buf.raw()
if len(raw_buf) < 1024 * 1024:
# In-line buffers smaller than 1MiB.
return True
all_buffers.append(raw_buf)
nonlocal total_bytes
total_bytes += len(raw_buf) + 4
return False
all_buffers[0] = pickle.dumps(
obj, protocol=pickle.HIGHEST_PROTOCOL, buffer_callback=oob_callback
)
if self.n_local_reader > 0:
if total_bytes + len(all_buffers[0]) >= self.buffer.max_chunk_bytes:
with self.acquire_write(timeout) as buf:
buf[0] = 1 # overflow
self.local_socket.send_multipart(all_buffers, copy=False)
else:
# Byte 0: 0
# Bytes 1-2: Count of buffers
# Then each buffer follows, preceded by 4 bytes containing its length:
# [4 byte int L][L bytes of buffer content] ...
with self.acquire_write(timeout) as buf:
buf[0] = 0 # not overflow
offset = 3
buf[1:offset] = to_bytes_big(len(all_buffers), 2) # oob buf count
for buffer in all_buffers:
buf_len = len(buffer)
# prepend each buffer with 4 bytes containing its size.
buf_offset = offset + 4
buf[offset:buf_offset] = to_bytes_big(buf_len, 4)
buf[buf_offset : (offset := buf_offset + buf_len)] = buffer
self._spin_condition.notify()
if self.n_remote_reader > 0:
self.remote_socket.send_multipart(all_buffers, copy=False)
def dequeue(
self,
timeout: float | None = None,
indefinite: bool = False,
):
"""Read from message queue with optional timeout (in seconds)"""
if self._is_local_reader:
with self.acquire_read(timeout, indefinite) as buf:
overflow = buf[0] == 1
if not overflow:
offset = 3
buf_count = from_bytes_big(buf[1:offset])
all_buffers = []
for i in range(buf_count):
buf_offset = offset + 4
buf_len = from_bytes_big(buf[offset:buf_offset])
offset = buf_offset + buf_len
all_buffers.append(buf[buf_offset:offset])
obj = pickle.loads(all_buffers[0], buffers=all_buffers[1:])
if overflow:
obj = MessageQueue.recv(self.local_socket, timeout)
elif self._is_remote_reader:
obj = MessageQueue.recv(self.remote_socket, timeout)
else:
raise RuntimeError("Only readers can dequeue")
return obj
@staticmethod
def recv(socket: zmq.Socket, timeout: float | None) -> Any:
# Ensure non-negative timeout passed to zmq poll.
timeout_ms = None if timeout is None else max(0, int(timeout * 1000))
if not socket.poll(timeout=timeout_ms):
raise TimeoutError
recv, *recv_oob = socket.recv_multipart(copy=False)
return pickle.loads(recv, buffers=recv_oob)
def broadcast_object(self, obj=None):
if self._is_writer:
self.enqueue(obj)
return obj
return self.dequeue()
@staticmethod
def create_from_process_group_single_reader(
pg: ProcessGroup,
max_chunk_bytes,
max_chunks,
reader_rank: int = 0,
blocking: bool = False,
) -> tuple["MessageQueue", list[Handle]]:
"""
Creates a MessageQueue for a process group with a single reader.
This method is designed for scenarios where only one process (the reader)
will consume messages, and all other processes are writers. It sets up
the shared memory buffer and communication handles accordingly, and
gathers the handles from all processes to the reader.
Args:
pg (ProcessGroup): The torch distributed process group.
max_chunk_bytes (int): Maximum size in bytes for each chunk in the buffer.
max_chunks (int): Maximum number of chunks in the buffer.
reader_rank (int, optional): The global rank that will act as the reader.
Defaults to 0.
blocking (bool, optional): If True, blocks until all processes are ready.
Defaults to False.
Returns:
tuple[MessageQueue, list[Handle]]:
The MessageQueue instance for the calling process,
and a list of handles (only non-empty for the reader process).
"""
local_size = current_platform.device_count()
rank = dist.get_rank()
same_node = rank // local_size == reader_rank // local_size
buffer_io = MessageQueue(
n_reader=1,
n_local_reader=1 if same_node else 0,
max_chunk_bytes=max_chunk_bytes,
max_chunks=max_chunks,
)
handle = buffer_io.export_handle()
handles = [None] * dist.get_world_size(pg) if rank == reader_rank else None
dist.gather_object(handle, handles, dst=reader_rank, group=pg)
if blocking:
buffer_io.wait_until_ready()
return buffer_io, cast(list[Handle], handles or [])
@staticmethod
def create_from_process_group(
pg: ProcessGroup | StatelessProcessGroup,
max_chunk_bytes,
max_chunks,
writer_rank: int = 0,
external_writer_handle=None,
blocking: bool = True,
) -> "MessageQueue":
"""
Creates a MessageQueue for a distributed process group with one writer and
multiple readers.
This method is designed for scenarios where one process (the writer) sends
messages, and all other processes (the readers) receive messages. It sets up
the shared memory buffer and socket communication handles accordingly, and
broadcasts the handle from the writer to all readers.
Args:
pg (ProcessGroup | StatelessProcessGroup): The torch distributed process
group.
max_chunk_bytes (int): Maximum size in bytes for each chunk in the buffer.
max_chunks (int): Maximum number of chunks in the buffer.
writer_rank (int, optional): The global rank that will act as the writer.
Defaults to 0.
external_writer_handle (Handle, optional): Used when there is a handle
from an external Message Queue. If provided, use this handle to init
PG writer message queue instead of creating a new one. Defaults to None.
blocking (bool, optional): If True, blocks until all processes are ready.
Defaults to True.
Returns:
MessageQueue: The MessageQueue instance for the calling process.
"""
if isinstance(pg, ProcessGroup):
group_rank = dist.get_rank(pg)
group_world_size = dist.get_world_size(pg)
global_ranks = dist.get_process_group_ranks(pg)
else:
group_rank = pg.rank
group_world_size = pg.world_size
global_ranks = list(range(pg.world_size))
from vllm.distributed.parallel_state import in_the_same_node_as
status = in_the_same_node_as(pg, source_rank=writer_rank)
if group_rank == writer_rank:
if external_writer_handle is not None:
buffer_io = MessageQueue.create_from_handle(
external_writer_handle, group_rank
)
else:
same_node_ranks = [i for i, s in enumerate(status) if s]
n_reader = group_world_size - 1
n_local_reader = len(same_node_ranks) - 1
local_reader_ranks = [i for i in same_node_ranks if i != writer_rank]
buffer_io = MessageQueue(
n_reader=n_reader,
n_local_reader=n_local_reader,
local_reader_ranks=local_reader_ranks,
max_chunk_bytes=max_chunk_bytes,
max_chunks=max_chunks,
)
handle = buffer_io.export_handle()
if isinstance(pg, ProcessGroup):
dist.broadcast_object_list(
[handle], src=global_ranks[writer_rank], group=pg
)
else:
pg.broadcast_obj(handle, writer_rank)
else:
if isinstance(pg, ProcessGroup):
recv = [None]
dist.broadcast_object_list(
recv, src=global_ranks[writer_rank], group=pg
)
handle = recv[0] # type: ignore
else:
handle = pg.broadcast_obj(None, writer_rank)
buffer_io = MessageQueue.create_from_handle(handle, group_rank)
if blocking:
buffer_io.wait_until_ready()
return buffer_io

View File

@@ -0,0 +1,28 @@
from vllm.reasoning.abs_reasoning_parsers import ReasoningParserManager
from vllm.reasoning.deepseek_r1_reasoning_parser import DeepSeekR1ReasoningParser
@ReasoningParserManager.register_module("super_v3")
class SuperV3ReasoningParser(DeepSeekR1ReasoningParser):
def extract_reasoning(self, model_output, request):
reasoning_content, final_content = super().extract_reasoning(
model_output, request
)
if (
hasattr(request, "chat_template_kwargs")
and request.chat_template_kwargs
and (
request.chat_template_kwargs.get("enable_thinking") is False
or request.chat_template_kwargs.get("force_nonempty_content") is True
)
and final_content is None
):
"""
The original `deepseek_r1` reasoning parser this inherits from will automatically put everything in the reasoning content when it cannot parse out reasoning. This was fine for the DeepSeek R1 model that was not intended to be used without reasoning.
1. Since the Nemotron 3 Nano and Super both have thinking off modes modulated by "enable_thinking=false" in the chat template kwargs, this change instead which will properly place the content in cases where there is no thinking enabled via config.
2. There are rare cases where the model will output only reasoning without an end-think token `</think>` (e.g. reasoning exceeds max length), which results in empty content returned. End users may want to unilaterally avoid such cases and always have a content response even if the model does not finish its reasoning.
"""
# Put all nonempty content into the content, rather than return content
reasoning_content, final_content = None, reasoning_content
return reasoning_content, final_content

297
vllm_shim_module.py Normal file
View File

@@ -0,0 +1,297 @@
#!/usr/bin/env python3
"""
vLLM shim with custom weights download.
Intercepts `python -m vllm.entrypoints.openai.api_server` so that
if --model or the positional model arg (after "serve") points to a URL,
we download + extract it to a local cache dir, then replace it with
the local path before handing off to the real vLLM server.
Supported archive formats (detected from URL extension):
.tar, .tar.gz, .tgz, .tar.bz2, .tar.xz, .zip
"""
import os
import sys
import subprocess
import datetime
import shutil
import time
import urllib.parse
import urllib.request
# Where to cache downloaded+extracted weights
# Production stack mounts the PVC at /data — use a subdir so it persists across pod restarts
CACHE_DIR = os.environ.get("VLLM_WEIGHTS_CACHE", "/data/weights")
# The shim dir that shadows the vllm package — must be stripped from PYTHONPATH
# before exec'ing the real vLLM, otherwise we loop forever.
SHIM_DIR = "/opt/vllm-shim"
def log(msg: str):
"""Write to both stdout and the shim log file."""
log_path = os.environ.get("VLLM_SHIM_LOG", "/tmp/vllm-shim.log")
ts = datetime.datetime.now().isoformat()
line = f"[{ts}] {msg}"
print(line, flush=True)
try:
with open(log_path, "a") as f:
f.write(line + "\n")
except Exception:
pass
def is_url(value: str) -> bool:
return value.startswith("http://") or value.startswith("https://")
def detect_archive_type(url: str) -> str:
"""
Detect archive type from URL path extension.
Returns one of: 'tar', 'tar.gz', 'tar.bz2', 'tar.xz', 'zip', or '' (unknown).
"""
path = urllib.parse.urlparse(url).path
for ext in (".tar.gz", ".tar.bz2", ".tar.xz"):
if path.endswith(ext):
return ext.lstrip(".")
_, ext = os.path.splitext(path)
mapping = {
".tar": "tar",
".tgz": "tar.gz",
".zip": "zip",
}
return mapping.get(ext.lower(), "")
MAX_DOWNLOAD_RETRIES = int(os.environ.get("VLLM_SHIM_MAX_RETRIES", "5"))
RETRY_DELAY_SECONDS = 5
def download_file(url: str, dest: str):
"""Download url to dest with retries and a progress indicator."""
for attempt in range(1, MAX_DOWNLOAD_RETRIES + 1):
try:
log(f"Downloading {url} -> {dest} (attempt {attempt}/{MAX_DOWNLOAD_RETRIES})")
urllib.request.urlretrieve(url, dest, reporthook=_download_progress)
log(f"Download complete: {dest}")
return
except Exception as e:
log(f"Download attempt {attempt} failed: {e}")
if os.path.exists(dest):
os.remove(dest)
if attempt < MAX_DOWNLOAD_RETRIES:
wait = RETRY_DELAY_SECONDS * attempt
log(f"Retrying in {wait}s...")
time.sleep(wait)
else:
log(f"All {MAX_DOWNLOAD_RETRIES} download attempts failed")
raise
def _download_progress(block_num, block_size, total_size):
"""Simple download progress callback."""
if total_size <= 0:
return
downloaded = block_num * block_size
pct = min(downloaded * 100 // total_size, 100)
if pct % 10 == 0 and pct > 0:
mb_down = downloaded / (1024 * 1024)
mb_total = total_size / (1024 * 1024)
sys.stdout.write(f"\r {pct}% ({mb_down:.0f}/{mb_total:.0f} MB)")
sys.stdout.flush()
def extract_archive(archive_path: str, dest_dir: str, archive_type: str):
"""Extract archive to dest_dir based on archive_type."""
log(f"Extracting {archive_path} ({archive_type}) -> {dest_dir}")
if archive_type == "tar.gz" or archive_type == "tgz":
shutil.unpack_archive(archive_path, dest_dir, "gztar")
elif archive_type == "tar.bz2":
shutil.unpack_archive(archive_path, dest_dir, "bztar")
elif archive_type == "tar.xz":
subprocess.run(
["tar", "-xJf", archive_path, "-C", dest_dir],
check=True,
)
elif archive_type == "tar":
shutil.unpack_archive(archive_path, dest_dir, "tar")
elif archive_type == "zip":
shutil.unpack_archive(archive_path, dest_dir, "zip")
else:
raise ValueError(f"Unsupported archive type: {archive_type}")
log(f"Extraction complete: {dest_dir}")
def find_model_dir(extract_dir: str) -> str:
"""
After extraction, find the directory containing the actual model weights.
Walks the tree looking for .safetensors files and returns the directory
that contains one. This handles archives with extra parent dirs,
nested structures, or flat extractions.
"""
for root, dirs, files in os.walk(extract_dir):
if any(f.endswith(".safetensors") for f in files):
return root
log("WARNING: No .safetensors files found in extracted archive, falling back to single-dir heuristic")
entries = [e for e in os.listdir(extract_dir)
if not e.startswith(".") and e != "__MACOSX"]
if len(entries) == 1 and os.path.isdir(os.path.join(extract_dir, entries[0])):
return os.path.join(extract_dir, entries[0])
return extract_dir
def download_and_extract_model(url: str) -> str:
"""
Download a model from URL, extract it, and return the local path.
Uses a cache keyed by URL filename to avoid re-downloading.
"""
url_filename = os.path.basename(urllib.parse.urlparse(url).path)
cache_key = os.path.splitext(url_filename)[0]
local_dir = os.path.join(CACHE_DIR, cache_key)
if os.path.isdir(local_dir) and os.listdir(local_dir):
model_path = find_model_dir(local_dir)
log(f"Using cached weights: {model_path}")
return model_path
os.makedirs(local_dir, exist_ok=True)
archive_type = detect_archive_type(url)
if not archive_type:
raise ValueError(
f"Cannot determine archive type from URL: {url}\n"
f"Supported extensions: .tar, .tar.gz, .tgz, .tar.bz2, .tar.xz, .zip"
)
tmp_archive = os.path.join(CACHE_DIR, url_filename + ".tmp")
try:
download_file(url, tmp_archive)
extract_archive(tmp_archive, local_dir, archive_type)
finally:
if os.path.exists(tmp_archive):
os.remove(tmp_archive)
return find_model_dir(local_dir)
def parse_args(args):
"""
Parse argv, intercepting --model and positional model args.
Production stack invokes: python -m vllm.entrypoints.openai.api_server serve <model-url> ...
The model can appear as:
- --model <url>
- --model=<url>
- A positional arg after "serve" subcommand
If the value is a URL, download+extract and replace with local path.
Returns the modified argv list.
"""
result = []
i = 0
model_replaced = False
saw_serve = False
while i < len(args):
arg = args[i]
# --model=<value>
if arg.startswith("--model="):
value = arg.split("=", 1)[1]
if is_url(value):
local_path = download_and_extract_model(value)
result.append(f"--model={local_path}")
model_replaced = True
else:
result.append(arg)
i += 1
continue
# --model <value>
if arg == "--model":
result.append(arg)
i += 1
if i < len(args):
value = args[i]
if is_url(value):
local_path = download_and_extract_model(value)
result.append(local_path)
model_replaced = True
else:
result.append(value)
i += 1
continue
# "serve" subcommand — next positional is the model
if arg == "serve":
result.append(arg)
saw_serve = True
i += 1
# The next non-flag argument is the model
if i < len(args) and not args[i].startswith("-") and is_url(args[i]):
local_path = download_and_extract_model(args[i])
result.append(local_path)
model_replaced = True
i += 1
continue
# Positional model arg when there's no "serve" subcommand
# (first non-flag arg if no serve seen)
if not arg.startswith("-") and not saw_serve and not model_replaced:
if is_url(arg):
local_path = download_and_extract_model(arg)
result.append(local_path)
model_replaced = True
i += 1
continue
result.append(arg)
i += 1
if model_replaced:
log("Model URL was replaced with local path")
return result
def strip_shim_from_pythonpath():
"""
Remove the shim directory from PYTHONPATH so that when we exec the
real vLLM, Python doesn't find our shadow package again (infinite loop).
"""
pp = os.environ.get("PYTHONPATH", "")
parts = [p for p in pp.split(":") if p != SHIM_DIR]
new_pp = ":".join(parts)
if new_pp != pp:
os.environ["PYTHONPATH"] = new_pp
log(f"Stripped {SHIM_DIR} from PYTHONPATH (was: {pp!r}, now: {new_pp!r})")
def main():
args = sys.argv[1:]
# Determine which vllm module was actually invoked so we exec the real one
# (could be vllm.entrypoints.cli.main, vllm.entrypoints.openai.api_server, etc.)
invoked_module = __name__ # e.g. "vllm.entrypoints.cli.main" or "vllm.entrypoints.openai.api_server"
log("=" * 50)
log("vLLM Custom Weights Shim")
log(f" Invoked as: python -m {invoked_module} {' '.join(args)}")
log("=" * 50)
# Intercept --model / positional model if it's a URL
modified_args = parse_args(args)
# Strip our shim from PYTHONPATH so the real vLLM resolves correctly
strip_shim_from_pythonpath()
# Build the real vLLM command using the same module that was invoked
vllm_cmd = [sys.executable, "-m", invoked_module] + modified_args
log(f"Launching vLLM: {' '.join(vllm_cmd)}")
# Exec into vLLM — replace this process so signals flow through cleanly
os.execvp(vllm_cmd[0], vllm_cmd)
if __name__ == "__main__":
main()
# Also run if imported as a module (some invocation paths just import the file)
main()